log reader without running full lp service

pkg/ccip/chainaccessor/config.go

@@ -5,13 +5,14 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/smartcontractkit/chainlink-ccip/pkg/consts"
-	"github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
 	"github.com/xssnick/tonutils-go/address"
 	"github.com/xssnick/tonutils-go/tlb"
 	"github.com/xssnick/tonutils-go/ton"
 	"github.com/xssnick/tonutils-go/tvm/cell"
 
+	"github.com/smartcontractkit/chainlink-ccip/pkg/consts"
+	"github.com/smartcontractkit/chainlink-common/pkg/types/ccipocr3"
+
 	"github.com/smartcontractkit/chainlink-ton/pkg/ccip/bindings/feequoter"
 	"github.com/smartcontractkit/chainlink-ton/pkg/ccip/bindings/offramp"
 	"github.com/smartcontractkit/chainlink-ton/pkg/ccip/bindings/onramp"

pkg/logpoller/backend/txparser/parser.go

@@ -4,11 +4,13 @@ import (
 	"context"
 	"time"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/xssnick/tonutils-go/tlb"
 	"github.com/xssnick/tonutils-go/tvm/cell"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+
 	"github.com/smartcontractkit/chainlink-ton/pkg/logpoller"
+	txparserutils "github.com/smartcontractkit/chainlink-ton/pkg/logpoller/backend/txparser/utils"
 	"github.com/smartcontractkit/chainlink-ton/pkg/logpoller/types"
 )
 
@@ -71,9 +73,9 @@ func (p *txParser) parseTx(ctx context.Context, tx types.TxWithBlock) ([]types.L
 
 			switch msg.MsgType {
 			case tlb.MsgTypeExternalOut:
-				eventSig, body, err = ParseExtMsgOut(msg.AsExternalOut(), filter.EventSig)
+				eventSig, body, err = txparserutils.ParseExtMsgOut(msg.AsExternalOut())
 			case tlb.MsgTypeInternal:
-				eventSig, body, err = ParseInternalMsg(msg.AsInternal(), filter.EventSig)
+				eventSig, body, err = txparserutils.ParseInternalMsg(msg.AsInternal())
 			case tlb.MsgTypeExternalIn:
 				continue // not supported
 			}

pkg/logpoller/backend/txparser/utils.go → pkg/logpoller/backend/txparser/utils/utils.go

@@ -1,4 +1,4 @@
-package txparser
+package txparserutils
 
 import (
 	"errors"
@@ -11,7 +11,7 @@ import (
 )
 
 // ParseExtMsgOut returns body and event signature(topic) for an external out message.
-func ParseExtMsgOut(msg *tlb.ExternalMessageOut, eventSig uint32) (sig uint32, body *cell.Cell, err error) {
+func ParseExtMsgOut(msg *tlb.ExternalMessageOut) (sig uint32, body *cell.Cell, err error) {
 	// for ExtMsgOut we use topic for event sig
 	bucket := event.NewExtOutLogBucket(msg.DestAddr())
 	topic, err := bucket.DecodeEventTopic()
@@ -20,16 +20,12 @@ func ParseExtMsgOut(msg *tlb.ExternalMessageOut, eventSig uint32) (sig uint32, b
 		return 0, nil, errors.New("failed to decode event topic")
 	}
 
-	if topic != eventSig {
-		return 0, nil, nil // topic doesn't match this event sig
-	}
-
 	return topic, msg.Payload(), nil
 }
 
 // ParseInternalMsg returns body and event signature(opcode) for an internal message.
 // this function extracts opcode, and returns remaining body slice as a cell
-func ParseInternalMsg(msg *tlb.InternalMessage, eventSig uint32) (sig uint32, body *cell.Cell, err error) {
+func ParseInternalMsg(msg *tlb.InternalMessage) (sig uint32, body *cell.Cell, err error) {
 	payload := msg.Payload()
 	if payload == nil {
 		return 0, nil, nil // no payload
@@ -41,10 +37,6 @@ func ParseInternalMsg(msg *tlb.InternalMessage, eventSig uint32) (sig uint32, bo
 		return 0, nil, fmt.Errorf("failed to extract opcode and body: %w", err)
 	}
 
-	if opcode != eventSig {
-		return 0, nil, nil // opcode doesn't match this event sig
-	}
-
 	return opcode, remainingBody, nil
 }
 

pkg/logpoller/interfaces.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/xssnick/tonutils-go/address"
 	"github.com/xssnick/tonutils-go/tlb"
+	"github.com/xssnick/tonutils-go/ton"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 
@@ -38,10 +39,11 @@ type FilterStore interface {
 
 // TxLoader defines the interface for loading transactions from the TON blockchain.
 type TxLoader interface {
-	// LoadTxsForAddresses retrieves all transactions from the specified source addresses
+	// LoadTxsForAddresses retrieves all transactions from multiple source addresses concurrently
 	// within the given block range (prevBlock, toBlock] - exclusive of prevBlock, inclusive of toBlock.
 	LoadTxsForAddresses(ctx context.Context, blockRange *types.BlockRange, srcAddrs []*address.Address) ([]types.TxWithBlock, error)
-	// FetchTxsForAddress retrieves transactions for a specific address within a block range.
+	// FetchTxsForAddress retrieves all transactions from single source address
+	// within the given block range (prevBlock, toBlock] - exclusive of prevBlock, inclusive of toBlock.
 	FetchTxsForAddress(ctx context.Context, blockRange *types.BlockRange, addr *address.Address) ([]types.TxWithBlock, error)
 }
 
@@ -110,3 +112,9 @@ type QueryBuilder[T any] interface {
 	// Execute runs the constructed query and returns the results.
 	Execute(ctx context.Context, store LogStore) (query.Result[T], error)
 }
+
+// RawLogProvider provides raw logs leveraging LogPoller libs without running the full service (o11y use case)
+type RawLogProvider interface {
+	// GetLogs retrieves all external message outputs for an address between fromBlockSeqNo (exclusive) and toBlock (inclusive).
+	GetLogs(ctx context.Context, addr *address.Address, from uint32, to *ton.BlockIDExt) ([]types.RawLog, error)
+}

pkg/logpoller/o11y_log_provider.go

@@ -0,0 +1,111 @@
+package logpoller
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+
+	"github.com/xssnick/tonutils-go/address"
+	"github.com/xssnick/tonutils-go/tlb"
+	"github.com/xssnick/tonutils-go/ton"
+
+	txparserUtils "github.com/smartcontractkit/chainlink-ton/pkg/logpoller/backend/txparser/utils"
+	"github.com/smartcontractkit/chainlink-ton/pkg/logpoller/types"
+)
+
+var _ RawLogProvider = (*tonO11yLogProvider)(nil)
+
+type tonO11yLogProvider struct {
+	client ton.APIClientWrapped
+	loader TxLoader
+}
+
+// NewLogReader creates a new LogReader instance.
+func NewTonO11yLogProvider(client ton.APIClientWrapped, loader TxLoader) RawLogProvider {
+	return &tonO11yLogProvider{
+		client: client,
+		loader: loader,
+	}
+}
+
+// GetLogs retrieves all ExternalMsgOutLogs for an address between fromBlockSeqNo (exclusive) and toBlock (inclusive).
+func (tlp *tonO11yLogProvider) GetLogs(ctx context.Context, addr *address.Address, from uint32, to *ton.BlockIDExt) ([]types.RawLog, error) {
+	// No new logs to fetch
+	if to.SeqNo <= from {
+		return nil, nil
+	}
+
+	// Resolve previous block if exists
+	var prevBlock *ton.BlockIDExt
+	var err error
+	if from == 0 {
+		prevBlock = nil // genesis has no prevBlock
+	} else {
+		prevBlock, err = tlp.client.LookupBlock(ctx, to.Workchain, to.Shard, from)
+		if err != nil {
+			return nil, fmt.Errorf("failed to lookup block for address=%s, fromSeqNo=%d: %w", addr.String(), from, err)
+		}
+	}
+
+	// Fetch tx for address on given blockRange
+	blockRange := &types.BlockRange{Prev: prevBlock, To: to}
+	txs, err := tlp.loader.FetchTxsForAddress(ctx, blockRange, addr)
+	if err != nil {
+		// display "genesis" if nil and don't panic
+		fromSeqNoStr := "genesis"
+		if prevBlock != nil {
+			fromSeqNoStr = strconv.FormatUint(uint64(prevBlock.SeqNo), 10)
+		}
+
+		return nil, fmt.Errorf("failed to fetch transactions fromSeqNo=%s, toSeqNo=%d: %w", fromSeqNoStr, to.SeqNo, err)
+	}
+
+	// Extract only externalMsgOut logs that we found in all these txes.
+	logs, err := tlp.extractExternalMsgOutLogs(ctx, txs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract logs for address=%s: %w", addr.String(), err)
+	}
+
+	return logs, nil
+}
+
+func (tlp *tonO11yLogProvider) extractExternalMsgOutLogs(ctx context.Context, txs []types.TxWithBlock) ([]types.RawLog, error) {
+	var allLogs []types.RawLog
+
+	for _, tx := range txs {
+		msgs, _ := tx.Tx.IO.Out.ToSlice()
+
+		blockData, err := tlp.client.GetBlockData(ctx, tx.Block)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, msg := range msgs {
+			// Skip any message that's not an external out message
+			if msg.MsgType != tlb.MsgTypeExternalOut {
+				continue
+			}
+
+			extMsg := msg.AsExternalOut()
+
+			// Fail hard so we don't skip events. We want at-least-once delivery guarantees on events
+			eventSig, body, err := txparserUtils.ParseExtMsgOut(extMsg)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse external message out for txHash=%v, LT=%d: %w", tx.Tx.Hash, tx.Tx.LT, err)
+			}
+
+			// If we got a valid event and body
+			if body != nil && eventSig != 0 {
+				log := types.RawLog{
+					Tx:    tx.Tx,
+					Block: blockData,
+					Data:  body,
+					Topic: eventSig,
+				}
+				allLogs = append(allLogs, log)
+			}
+		}
+	}
+
+	return allLogs, nil
+}

pkg/logpoller/types/types.go

@@ -77,3 +77,11 @@ func (l Log) String() string {
 
 	return sb.String()
 }
+
+// RawLog contains raw log data + metadata that can be transformed by consumers as needed (eg. o11y)
+type RawLog struct {
+	Tx    *tlb.Transaction
+	Block *tlb.Block
+	Data  *cell.Cell
+	Topic uint32
+}