domain

Documentation

Index

• Constants
• type AlloraClientInterface
• type BlockMetadata
• type CodecInterface
• type EventsProducer
• type FilterInterface
• type Message
  • func NewMessage(msgType, name string, payload Payload) (Message, error)
• type Metadata
  • func NewMetadata(blockHeight int64, chainID string, blockHash string, time time.Time, ...) Metadata
• type OHLCData
• type OHLCResponse
• type Payload
  • func NewPayload(metadata Metadata, data json.RawMessage) Payload
• type ProcessedBlock
• type ProcessedBlockEvent
• type ProcessedBlockRepositoryInterface
• type ProcessorService
• type QueryProcessor
• type QueryProducer
• type QueryResult
• type QueryResultRepositoryInterface
• type QueryService
• type ReputerStake
• type StreamingClient
• type TopicRouter
• type TransactionMetadata
• type TransactionsProducer

Constants

const (
	MessageTypeEvent       string = "event"
	MessageTypeTransaction string = "transaction"
)

const (
	StatusPending   string = "pending"
	StatusCompleted string = "completed"
	StatusFailed    string = "failed"
)

Types

type AlloraClientInterface

type AlloraClientInterface interface {
	// GetLatestBlockHeight returns the latest block height
	GetLatestBlockHeight(ctx context.Context) (int64, error)
	// GetBlockByHeight returns a block by height
	GetBlockByHeight(ctx context.Context, height int64) (*ctypes.ResultBlock, error)
	// GetBlockResults returns the results of a block by height
	GetBlockResults(ctx context.Context, height int64) (*ctypes.ResultBlockResults, error)
	// GetHeader returns the header of a block by height
	GetHeader(ctx context.Context, height int64) (*ctypes.ResultHeader, error)
	// Query queries the ABCI
	Query(ctx context.Context, path string, data bytes.HexBytes) (*ctypes.ResultABCIQuery, error)
}

type BlockMetadata

type BlockMetadata struct {
	Height  int64     `json:"height"`   // block height
	ChainID string    `json:"chain_id"` // chain id
	Hash    string    `json:"hash"`     // block hash
	Time    time.Time `json:"time"`     // block time
}

type CodecInterface

type CodecInterface interface {
	// ParseTx parses a transaction
	ParseTx(txBytes []byte) (*tx.Tx, error)
	// ParseTxMessage parses a transaction message
	ParseTxMessage(message *codectypes.Any) (proto.Message, error)
	// ParseTxMessages parses transaction messages
	ParseTxMessages(txMessages []*codectypes.Any) ([]proto.Message, error)
	// ParseEvent parses a typed event
	ParseEvent(event *abcitypes.Event) (proto.Message, error)
	// MarshalProtoJSON marshals a protobuf message to JSON
	MarshalProtoJSON(event proto.Message) (json.RawMessage, error)
	// ParseUntypedEvent parses an untyped event
	ParseUntypedEvent(event *abcitypes.Event) (json.RawMessage, error)
	// IsTypedEvent checks if an event is typed
	IsTypedEvent(event *abcitypes.Event) bool
}

CodecInterface defines the interface for encoding and decoding messages

type EventsProducer

type EventsProducer interface {
	Execute(ctx context.Context) error
}

type FilterInterface

type FilterInterface[T any] interface {
	ShouldProcess(typeValue *T) bool
}

FilterInterface defines the interface for filters.

type Message

type Message struct {
	ID        string    `json:"id"`        // hash of the entire payload
	Type      string    `json:"type"`      // event or transaction
	Name      string    `json:"name"`      // type of the payload (full qualified name, e.g. emisions.v3.EventNetworkLossSet)
	Timestamp time.Time `json:"timestamp"` // timestamp of this message (created at, in UTC)
	Payload   Payload   `json:"payload"`   // payload of the message
}

These are the messages that are sent to Kafka

func NewMessage

func NewMessage(msgType, name string, payload Payload) (Message, error)

type Metadata

type Metadata struct {
	BlockMetadata       BlockMetadata       `json:"block_metadata"`       // metadata of the block
	TransactionMetadata TransactionMetadata `json:"transaction_metadata"` // metadata of the transaction
}

func NewMetadata

func NewMetadata(blockHeight int64, chainID string, blockHash string, time time.Time, txIndex int, txHash string, typeURL string) Metadata

type OHLCData

type OHLCData struct {
	Ticker         string `json:"ticker"`
	ExchangeCode   string `json:"exchange_code"`
	Date           string `json:"date"`
	Open           string `json:"open"`
	High           string `json:"high"`
	Low            string `json:"low"`
	Close          string `json:"close"`
	TradesDone     int    `json:"trades_done"`
	Volume         string `json:"volume"`
	VolumeNotional string `json:"volume_notional"`
}

type OHLCResponse

type OHLCResponse struct {
	RequestID string `json:"request_id"`
	Status    bool   `json:"status"`
	Data      struct {
		Data []OHLCData `json:"data"`
	} `json:"data"`
}

type Payload

type Payload struct {
	Metadata Metadata        `json:"metadata"` // metadata of the block and transaction (when available)
	Data     json.RawMessage `json:"data"`     // json encoded payload (transaction or event)
}

func NewPayload

func NewPayload(metadata Metadata, data json.RawMessage) Payload

type ProcessedBlock

type ProcessedBlock struct {
	ID          int64
	Height      int64
	ProcessedAt time.Time
	Status      string
}

ProcessedBlock is the model used to keep track of the processed blocks

type ProcessedBlockEvent

type ProcessedBlockEvent struct {
	ID          int64
	Height      int64
	ProcessedAt time.Time
	Status      string
}

ProcessedBlockEvent is the model used to keep track of the processed block events

type ProcessedBlockRepositoryInterface

type ProcessedBlockRepositoryInterface interface {
	GetLastProcessedBlock(ctx context.Context) (ProcessedBlock, error)
	SaveProcessedBlock(ctx context.Context, block ProcessedBlock) error

	GetLastProcessedBlockEvent(ctx context.Context) (ProcessedBlockEvent, error)
	SaveProcessedBlockEvent(ctx context.Context, blockEvent ProcessedBlockEvent) error
}

type ProcessorService

type ProcessorService interface {
	ProcessBlock(ctx context.Context, block *ctypes.ResultBlock) error
	ProcessBlockResults(ctx context.Context, blockResults *ctypes.ResultBlockResults, header *types.Header) error
}

type QueryProcessor

type QueryProcessor interface {
	GetMaxTopicID(ctx context.Context) (uint64, error)
	ProcessNetworkInferences(ctx context.Context, topicID uint64) error
	ProcessNaiveInfererNetworkRegret(ctx context.Context, topicID uint64) error
	ProcessInflation(ctx context.Context) error
	ProcessEmissionInfo(ctx context.Context) error
	ProcessGroundTruthData(ctx context.Context, topicID uint64) error
	ProcessReputerStakes(ctx context.Context, topicID uint64) error
}

type QueryProducer

type QueryProducer interface {
	Execute(ctx context.Context) error
}

type QueryResult

type QueryResult struct {
	Height int64
	Data   json.RawMessage
}

type QueryResultRepositoryInterface

type QueryResultRepositoryInterface interface {
	SaveQueryResult(ctx context.Context, key string, queryType string, metadata json.RawMessage, value json.RawMessage) error
}

type QueryService

type QueryService interface {
	GetLatestNetworkInferences(ctx context.Context, topicID uint64) (*QueryResult, error)
	GetNextTopicID(ctx context.Context) (uint64, error)
	GetNaiveInfererNetworkRegret(ctx context.Context, topicID uint64) (*QueryResult, error)
	GetInflation(ctx context.Context) (*QueryResult, error)
	GetEmissionInfo(ctx context.Context) (*QueryResult, error)
	GetOHLCData(ctx context.Context, ticker string, fromDate string) (OHLCResponse, error)
	GetActiveReputers(ctx context.Context, topicID uint64) (emissionTypes.GetActiveReputersForTopicResponse, int64, error)
	GetReputerStake(ctx context.Context, reputerAddr string, topicID uint64) (math.Int, error)
}

type ReputerStake

type ReputerStake struct {
	Address string   `json:"address"`
	Stake   math.Int `json:"stake"`
}

type StreamingClient

type StreamingClient interface {
	// PublishAsync publishes a message to a topic asynchronously
	PublishAsync(ctx context.Context, msgType string, message []byte, blockHeight int64) error
	// Close closes the client
	Close() error
	// Flush flushes the client
	Flush(ctx context.Context) error
}

type TopicRouter

type TopicRouter interface {
	// GetTopic returns the Kafka topic for a given message type
	GetTopic(msgType string) (string, error)
}

TopicRouter defines the interface for determining Kafka topics based on message types.

type TransactionMetadata

type TransactionMetadata struct {
	TxIndex int    `json:"tx_index,omitempty"` // transaction index
	TxHash  string `json:"tx_hash,omitempty"`  // transaction hash
	Type    string `json:"type"`               // protobuf type (full qualified name, e.g. emisions.v3.EventNetworkLossSet)
}

type TransactionsProducer

type TransactionsProducer interface {
	Execute(ctx context.Context) error
}