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
	ProcessLatestAvailableNetworkInferencesOutlierResistant(ctx context.Context, topicID uint64) error
	ProcessLatestNetworkInferencesOutlierResistant(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)
	GetLatestAvailableNetworkInferencesOutlierResistant(ctx context.Context, topicID uint64) (*QueryResult, error)
	GetLatestNetworkInferencesOutlierResistant(ctx context.Context, topicID uint64) (*QueryResult, 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
}