log reader without running full lp service

pkg/logpoller/backend/loader/account/loader.go

@@ -63,7 +63,7 @@ func (l *accountTxLoader) LoadTxsForAddresses(ctx context.Context, blockRange *t
 	for _, addr := range srcAddrs {
 		currAddr := addr
 		eg.Go(func() error {
-			txs, err := l.fetchTxsForAddress(egCtx, currAddr, blockRange)
+			txs, err := l.FetchTxsForAddress(egCtx, currAddr, blockRange)
 			if err != nil {
 				return fmt.Errorf("failed to fetch for %s: %w", currAddr.String(), err)
 			}
@@ -93,7 +93,7 @@ func (l *accountTxLoader) LoadTxsForAddresses(ctx context.Context, blockRange *t
 //
 // Note: Block range (prevBlock, toBlock] is exclusive of prevBlock, inclusive of toBlock
 // TODO: stream tx back to log poller to avoid memory overhead in production
-func (l *accountTxLoader) fetchTxsForAddress(ctx context.Context, addr *address.Address, blockRange *types.BlockRange) ([]types.TxWithBlock, error) {
+func (l *accountTxLoader) FetchTxsForAddress(ctx context.Context, addr *address.Address, blockRange *types.BlockRange) ([]types.TxWithBlock, error) {
 	if blockRange.Prev != nil && blockRange.Prev.SeqNo >= blockRange.To.SeqNo {
 		return nil, fmt.Errorf("prevBlock %d is not before toBlock %d", blockRange.Prev.SeqNo, blockRange.To.SeqNo)
 	}

pkg/logpoller/backend/txparser/parser.go

@@ -4,10 +4,11 @@ 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"
 	"github.com/smartcontractkit/chainlink-ton/pkg/logpoller/types"
 )
@@ -101,3 +102,15 @@ func (p *txParser) parseTx(ctx context.Context, tx types.TxWithBlock) ([]types.L
 	}
 	return allLogs, nil
 }
+
+// messageParserWithoutFilters provides a way to parse messages without using log poller filters
+type messageParserWithoutFilters struct{}
+
+func NewMessageParser() logpoller.MessageParser {
+	return &messageParserWithoutFilters{}
+}
+
+func (p *messageParserWithoutFilters) ParseExtMsgOut(msg *tlb.ExternalMessageOut) (uint32, *cell.Cell, error) {
+	// eventSig=0 to bypass filtering
+	return ParseExtMsgOut(msg, 0)
+}

pkg/logpoller/backend/txparser/utils.go

@@ -20,7 +20,8 @@ func ParseExtMsgOut(msg *tlb.ExternalMessageOut, eventSig uint32) (sig uint32, b
 		return 0, nil, errors.New("failed to decode event topic")
 	}
 
-	if topic != eventSig {
+	// eventSig == 0 bypasses filtering
+	if eventSig != 0 && topic != eventSig {
 		return 0, nil, nil // topic doesn't match this event sig
 	}
 
@@ -41,7 +42,8 @@ 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 {
+	// eventSig == 0 bypasses filtering
+	if eventSig != 0 && opcode != eventSig {
 		return 0, nil, nil // opcode doesn't match this event sig
 	}
 

pkg/logpoller/interfaces.go

@@ -5,6 +5,8 @@ import (
 
 	"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-common/pkg/services"
 
@@ -38,9 +40,12 @@ 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 all transactions from single source address
+	// within the given block range (prevBlock, toBlock] - exclusive of prevBlock, inclusive of toBlock.
+	FetchTxsForAddress(ctx context.Context, addr *address.Address, blockRange *types.BlockRange) ([]types.TxWithBlock, error)
 }
 
 // TxParser defines the interface for parsing raw blockchain transactions into structured logs.
@@ -52,6 +57,12 @@ type TxParser interface {
 	ParseTransactions(ctx context.Context, txs []types.TxWithBlock) ([]types.Log, error)
 }
 
+// MessageParser defines an interface for parsing TON messages
+type MessageParser interface {
+	// ParseExtMsgOut extracts event signature and payload from an external message output.
+	ParseExtMsgOut(msg *tlb.ExternalMessageOut) (eventSig uint32, body *cell.Cell, err error)
+}
+
 // LogStore defines the interface for storing and retrieving logs.
 type LogStore interface {
 	SaveLog(log types.Log)
@@ -102,3 +113,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)
 }
+
+// LogReader provides a way to retrieve logs without running the full LogPoller service.
+type LogReader interface {
+	// GetLogs retrieves all external message outputs for an address between fromBlockSeqNo (exclusive) and toBlock (inclusive).
+	GetLogs(ctx context.Context, addr *address.Address, fromBlockSeqNo uint32, toBlock *ton.BlockIDExt) ([]types.Log, error)
+}

pkg/logpoller/log_reader.go

@@ -0,0 +1,116 @@
+package logpoller
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/xssnick/tonutils-go/address"
+	"github.com/xssnick/tonutils-go/tlb"
+	"github.com/xssnick/tonutils-go/ton"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+
+	"github.com/smartcontractkit/chainlink-ton/pkg/logpoller/types"
+)
+
+var _ LogReader = (*logReader)(nil)
+
+type logReader struct {
+	client    ton.APIClientWrapped
+	lggr      logger.Logger
+	loader    TxLoader
+	msgParser MessageParser
+}
+
+// NewLogReader creates a new LogReader instance.
+func NewLogReader(client ton.APIClientWrapped, lggr logger.Logger, loader TxLoader, msgParser MessageParser) LogReader {
+	return &logReader{
+		client:    client,
+		lggr:      lggr,
+		loader:    loader,
+		msgParser: msgParser,
+	}
+}
+
+// GetLogs retrieves all ExternalMsgOutLogs for an address between fromBlockSeqNo (exclusive) and toBlock (inclusive).
+func (lr *logReader) GetLogs(ctx context.Context, addr *address.Address, fromBlockSeqNo uint32, toBlock *ton.BlockIDExt) ([]types.Log, error) {
+	// No new logs to fetch
+	if toBlock.SeqNo <= fromBlockSeqNo {
+		return nil, nil
+	}
+
+	// Resolve previous block if exists
+	var prevBlock *ton.BlockIDExt
+	var err error
+	if fromBlockSeqNo == 0 {
+		prevBlock = nil // genesis has no prevBlock
+	} else {
+		prevBlock, err = lr.client.LookupBlock(ctx, toBlock.Workchain, toBlock.Shard, fromBlockSeqNo)
+		if err != nil {
+			return nil, fmt.Errorf("failed to lookup block for address=%s, fromSeqNo=%d: %w", addr.String(), fromBlockSeqNo, err)
+		}
+	}
+
+	// Fetch tx for address on given blockRange
+	blockRange := &types.BlockRange{Prev: prevBlock, To: toBlock}
+	txs, err := lr.loader.FetchTxsForAddress(ctx, addr, blockRange)
+	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, toBlock.SeqNo, err)
+	}
+
+	// Extract only externalMsgOut logs that we found in all these txes.
+	logs, err := lr.extractExternalMsgOutLogs(txs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract logs for address=%s: %w", addr.String(), err)
+	}
+
+	return logs, nil
+}
+
+func (lr *logReader) extractExternalMsgOutLogs(txs []types.TxWithBlock) ([]types.Log, error) {
+	var allLogs []types.Log
+
+	for _, tx := range txs {
+		msgs, _ := tx.Tx.IO.Out.ToSlice()
+		for _, msg := range msgs {
+			// Skip any message that's not an external out message
+			if msg.MsgType != tlb.MsgTypeExternalOut {
+				continue
+			}
+
+			srcAddr := msg.Msg.SenderAddr()
+			extMsg := msg.AsExternalOut()
+
+			// Fail hard so we don't skip events. We want at-least-once delivery guarantees on events
+			eventSig, body, err := lr.msgParser.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.Log{
+					EventSig:    eventSig,
+					Address:     srcAddr,
+					Data:        body,
+					TxHash:      types.TxHash(tx.Tx.Hash),
+					TxLT:        tx.Tx.LT,
+					TxTimestamp: time.Unix(int64(tx.Tx.Now), 0).UTC(),
+					Block:       tx.Block,
+				}
+
+				allLogs = append(allLogs, log)
+			}
+		}
+	}
+
+	return allLogs, nil
+}