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 Payload
  • func NewPayload(metadata Metadata, data json.RawMessage) Payload
• type ProcessedBlock
• type ProcessedBlockEvent
• type ProcessedBlockRepositoryInterface
• type ProcessorService
• 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)
}

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 an event
	ParseEvent(event *abcitypes.Event) (proto.Message, error)
	// MarshalProtoJSON marshals a protobuf message to JSON
	MarshalProtoJSON(event proto.Message) (json.RawMessage, error)
}

DecodeTxMsg decodes a transaction message

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 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 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
}