storage

package
v0.2.3-beta Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 31, 2025 License: Apache-2.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var DEFAULT_MAX_ROWS_PER_INSERT = 100000
View Source 
var DEFAULT_REDIS_POOL_SIZE = 20

Functions

func IsInRange 

func IsInRange(num *big.Int, rangeStart *big.Int, rangeEnd *big.Int) bool

func NewConnector 

func NewConnector[T any](cfg *config.StorageConnectionConfig) (T, error)

Types

type BigInt 

type BigInt struct {
	big.Int
}

func (BigInt) MarshalJSON 

func (b BigInt) MarshalJSON() ([]byte, error)

type ClickHouseConnector 

type ClickHouseConnector struct {
	// contains filtered or unexported fields
}

func NewClickHouseConnector 

func NewClickHouseConnector(cfg *config.ClickhouseConfig) (*ClickHouseConnector, error)

func (*ClickHouseConnector) DeleteBlockData 

func (c *ClickHouseConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) error

func (*ClickHouseConnector) DeleteBlockFailures 

func (c *ClickHouseConnector) DeleteBlockFailures(failures []common.BlockFailure) error

func (*ClickHouseConnector) DeleteStagingData 

func (c *ClickHouseConnector) DeleteStagingData(data *[]common.BlockData) error

func (*ClickHouseConnector) GetAggregations 

func (c *ClickHouseConnector) GetAggregations(table string, qf QueryFilter) (QueryResult[interface{}], error)

func (*ClickHouseConnector) GetBlockFailures 

func (c *ClickHouseConnector) GetBlockFailures(qf QueryFilter) ([]common.BlockFailure, error)

func (*ClickHouseConnector) GetBlockHeadersDescending 

func (c *ClickHouseConnector) GetBlockHeadersDescending(chainId *big.Int, from *big.Int, to *big.Int) (blockHeaders []common.BlockHeader, err error)

func (*ClickHouseConnector) GetBlocks 

func (c *ClickHouseConnector) GetBlocks(qf QueryFilter) (QueryResult[common.Block], error)

func (*ClickHouseConnector) GetLastReorgCheckedBlockNumber 

func (c *ClickHouseConnector) GetLastReorgCheckedBlockNumber(chainId *big.Int) (*big.Int, error)

func (*ClickHouseConnector) GetLastStagedBlockNumber 

func (c *ClickHouseConnector) GetLastStagedBlockNumber(chainId *big.Int, rangeStart *big.Int, rangeEnd *big.Int) (maxBlockNumber *big.Int, err error)

func (*ClickHouseConnector) GetLogs 

func (c *ClickHouseConnector) GetLogs(qf QueryFilter) (QueryResult[common.Log], error)

func (*ClickHouseConnector) GetMaxBlockNumber 

func (c *ClickHouseConnector) GetMaxBlockNumber(chainId *big.Int) (maxBlockNumber *big.Int, err error)

func (*ClickHouseConnector) GetStagingData 

func (c *ClickHouseConnector) GetStagingData(qf QueryFilter) (*[]common.BlockData, error)

func (*ClickHouseConnector) GetTraces 

func (c *ClickHouseConnector) GetTraces(qf QueryFilter) (traces []common.Trace, err error)

func (*ClickHouseConnector) GetTransactions 

func (c *ClickHouseConnector) GetTransactions(qf QueryFilter) (QueryResult[common.Transaction], error)

func (*ClickHouseConnector) InsertBlockData 

func (c *ClickHouseConnector) InsertBlockData(data *[]common.BlockData) error

TODO make this atomic

func (*ClickHouseConnector) InsertStagingData 

func (c *ClickHouseConnector) InsertStagingData(data []common.BlockData) error

func (*ClickHouseConnector) SetLastReorgCheckedBlockNumber 

func (c *ClickHouseConnector) SetLastReorgCheckedBlockNumber(chainId *big.Int, blockNumber *big.Int) error

func (*ClickHouseConnector) StoreBlockFailures 

func (c *ClickHouseConnector) StoreBlockFailures(failures []common.BlockFailure) error

type IMainStorage 

type IMainStorage interface {
	InsertBlockData(data *[]common.BlockData) error

	GetBlocks(qf QueryFilter) (blocks QueryResult[common.Block], err error)
	GetTransactions(qf QueryFilter) (transactions QueryResult[common.Transaction], err error)
	GetLogs(qf QueryFilter) (logs QueryResult[common.Log], err error)
	GetAggregations(table string, qf QueryFilter) (QueryResult[interface{}], error)
	GetTraces(qf QueryFilter) (traces []common.Trace, err error)
	GetMaxBlockNumber(chainId *big.Int) (maxBlockNumber *big.Int, err error)
	/**
	 * Get block headers ordered from latest to oldest.
	 */
	GetBlockHeadersDescending(chainId *big.Int, from *big.Int, to *big.Int) (blockHeaders []common.BlockHeader, err error)
	DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) error
}

type IOrchestratorStorage 

type IOrchestratorStorage interface {
	GetBlockFailures(qf QueryFilter) ([]common.BlockFailure, error)
	StoreBlockFailures(failures []common.BlockFailure) error
	DeleteBlockFailures(failures []common.BlockFailure) error
	GetLastReorgCheckedBlockNumber(chainId *big.Int) (*big.Int, error)
	SetLastReorgCheckedBlockNumber(chainId *big.Int, blockNumber *big.Int) error
}

type IStagingStorage 

type IStagingStorage interface {
	InsertStagingData(data []common.BlockData) error
	GetStagingData(qf QueryFilter) (data *[]common.BlockData, err error)
	DeleteStagingData(data *[]common.BlockData) error
	GetLastStagedBlockNumber(chainId *big.Int, rangeStart *big.Int, rangeEnd *big.Int) (maxBlockNumber *big.Int, err error)
}

type IStorage 

type IStorage struct {
	OrchestratorStorage IOrchestratorStorage
	MainStorage         IMainStorage
	StagingStorage      IStagingStorage
}

func NewStorageConnector 

func NewStorageConnector(cfg *config.StorageConfig) (IStorage, error)

type MemoryConnector 

type MemoryConnector struct {
	// contains filtered or unexported fields
}

func NewMemoryConnector 

func NewMemoryConnector(cfg *config.MemoryConfig) (*MemoryConnector, error)

func (*MemoryConnector) DeleteBlockData 

func (m *MemoryConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) error

func (*MemoryConnector) DeleteBlockFailures 

func (m *MemoryConnector) DeleteBlockFailures(failures []common.BlockFailure) error

func (*MemoryConnector) DeleteStagingData 

func (m *MemoryConnector) DeleteStagingData(data *[]common.BlockData) error

func (*MemoryConnector) GetBlockFailures 

func (m *MemoryConnector) GetBlockFailures(qf QueryFilter) ([]common.BlockFailure, error)

func (*MemoryConnector) GetBlockHeadersDescending 

func (m *MemoryConnector) GetBlockHeadersDescending(chainId *big.Int, from *big.Int, to *big.Int) ([]common.BlockHeader, error)

func (*MemoryConnector) GetBlocks 

func (m *MemoryConnector) GetBlocks(qf QueryFilter) (QueryResult[common.Block], error)

func (*MemoryConnector) GetLastReorgCheckedBlockNumber 

func (m *MemoryConnector) GetLastReorgCheckedBlockNumber(chainId *big.Int) (*big.Int, error)

func (*MemoryConnector) GetLastStagedBlockNumber 

func (m *MemoryConnector) GetLastStagedBlockNumber(chainId *big.Int, rangeStart *big.Int, rangeEnd *big.Int) (*big.Int, error)

func (*MemoryConnector) GetLogs 

func (m *MemoryConnector) GetLogs(qf QueryFilter) (QueryResult[common.Log], error)

func (*MemoryConnector) GetMaxBlockNumber 

func (m *MemoryConnector) GetMaxBlockNumber(chainId *big.Int) (*big.Int, error)

func (*MemoryConnector) GetStagingData 

func (m *MemoryConnector) GetStagingData(qf QueryFilter) (*[]common.BlockData, error)

func (*MemoryConnector) GetTraces 

func (m *MemoryConnector) GetTraces(qf QueryFilter) (QueryResult[common.Trace], error)

func (*MemoryConnector) GetTransactions 

func (m *MemoryConnector) GetTransactions(qf QueryFilter) (QueryResult[common.Transaction], error)

func (*MemoryConnector) InsertBlockData 

func (m *MemoryConnector) InsertBlockData(data *[]common.BlockData) error

func (*MemoryConnector) InsertStagingData 

func (m *MemoryConnector) InsertStagingData(data []common.BlockData) error

func (*MemoryConnector) SetLastReorgCheckedBlockNumber 

func (m *MemoryConnector) SetLastReorgCheckedBlockNumber(chainId *big.Int, blockNumber *big.Int) error

func (*MemoryConnector) StoreBlockFailures 

func (m *MemoryConnector) StoreBlockFailures(failures []common.BlockFailure) error

type QueryFilter 

type QueryFilter struct {
	ChainId         *big.Int
	BlockNumbers    []*big.Int
	FilterParams    map[string]string
	GroupBy         []string
	SortBy          string
	SortOrder       string
	Page            int
	Limit           int
	Offset          int
	Aggregates      []string // e.g., ["COUNT(*) AS count", "SUM(amount) AS total_amount"]
	FromAddress     string
	ContractAddress string
	Signature       string
}

type QueryResult 

type QueryResult[T any] struct {
	// TODO: findout how to only allow Log/transaction arrays or split the result
	Data       []T                      `json:"data"`
	Aggregates []map[string]interface{} `json:"aggregates"`
}

type RedisConnector 

type RedisConnector struct {
	// contains filtered or unexported fields
}

func NewRedisConnector 

func NewRedisConnector(cfg *config.RedisConfig) (*RedisConnector, error)

func (*RedisConnector) DeleteBlockFailures 

func (r *RedisConnector) DeleteBlockFailures(failures []common.BlockFailure) error

func (*RedisConnector) GetBlockFailures 

func (r *RedisConnector) GetBlockFailures(qf QueryFilter) ([]common.BlockFailure, error)

func (*RedisConnector) GetLastReorgCheckedBlockNumber 

func (r *RedisConnector) GetLastReorgCheckedBlockNumber(chainId *big.Int) (*big.Int, error)

func (*RedisConnector) SetLastReorgCheckedBlockNumber 

func (r *RedisConnector) SetLastReorgCheckedBlockNumber(chainId *big.Int, blockNumber *big.Int) error

func (*RedisConnector) StoreBlockFailures 

func (r *RedisConnector) StoreBlockFailures(failures []common.BlockFailure) error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.