[NONEVM-916] logpoller log processing & decoding

.tool-versions

@@ -2,7 +2,7 @@ nodejs 18.20.2
 yarn 1.22.19
 rust 1.59.0
 golang 1.23.3
-golangci-lint 1.60.1
+golangci-lint 1.61.0
 actionlint 1.6.22
 shellcheck 0.8.0
 helm 3.9.4

go.mod

@@ -14,7 +14,6 @@ require (
 	github.com/hashicorp/go-plugin v1.6.2
 	github.com/jackc/pgx/v4 v4.18.3
 	github.com/jpillora/backoff v1.0.0
-	github.com/lib/pq v1.10.9
 	github.com/pelletier/go-toml/v2 v2.2.0
 	github.com/prometheus/client_golang v1.17.0
 	github.com/smartcontractkit/chainlink-common v0.4.2-0.20250116214855-f49c5c27db51
@@ -76,6 +75,7 @@ require (
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/leodido/go-urn v1.2.0 // indirect
+	github.com/lib/pq v1.10.9 // indirect
 	github.com/linkedin/goavro/v2 v2.12.0 // indirect
 	github.com/logrusorgru/aurora v2.0.3+incompatible // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect

integration-tests/smoke/event_loader_test.go

@@ -25,6 +25,7 @@ import (
 
 	contract "github.com/smartcontractkit/chainlink-solana/contracts/generated/log_read_test"
 	"github.com/smartcontractkit/chainlink-solana/pkg/solana/client"
+	"github.com/smartcontractkit/chainlink-solana/pkg/solana/config"
 	"github.com/smartcontractkit/chainlink-solana/pkg/solana/logpoller"
 
 	"github.com/smartcontractkit/chainlink-solana/integration-tests/solclient"
@@ -49,7 +50,8 @@ func TestEventLoader(t *testing.T) {
 	require.NoError(t, err)
 
 	rpcURL, wsURL := setupTestValidator(t, privateKey.PublicKey().String())
-	rpcClient := rpc.New(rpcURL)
+	cl, rpcClient, err := client.NewTestClient(rpcURL, config.NewDefault(), 1*time.Second, logger.Nop())
+	require.NoError(t, err)
 	wsClient, err := ws.Connect(ctx, wsURL)
 	require.NoError(t, err)
 
@@ -62,7 +64,7 @@ func TestEventLoader(t *testing.T) {
 	parser := &printParser{t: t}
 	sender := newLogSender(t, rpcClient, wsClient)
 	collector := logpoller.NewEncodedLogCollector(
-		rpcClient,
+		cl,
 		parser,
 		logger.Nop(),
 	)

pkg/solana/chain.go

@@ -28,16 +28,25 @@ import (
 	mn "github.com/smartcontractkit/chainlink-solana/pkg/solana/client/multinode"
 	"github.com/smartcontractkit/chainlink-solana/pkg/solana/config"
 	"github.com/smartcontractkit/chainlink-solana/pkg/solana/internal"
+	"github.com/smartcontractkit/chainlink-solana/pkg/solana/logpoller"
 	"github.com/smartcontractkit/chainlink-solana/pkg/solana/monitor"
 	"github.com/smartcontractkit/chainlink-solana/pkg/solana/txm"
 	txmutils "github.com/smartcontractkit/chainlink-solana/pkg/solana/txm/utils"
 )
 
+type LogPoller interface {
+	Start(context.Context) error
+	Close() error
+	RegisterFilter(ctx context.Context, filter logpoller.Filter) error
+	UnregisterFilter(ctx context.Context, name string) error
+}
+
 type Chain interface {
 	types.ChainService
 
 	ID() string
 	Config() config.Config
+	LogPoller() LogPoller
 	TxManager() TxManager
 	// Reader returns a new Reader from the available list of nodes (if there are multiple, it will randomly select one)
 	Reader() (client.Reader, error)
@@ -90,6 +99,7 @@ type chain struct {
 	services.StateMachine
 	id             string
 	cfg            *config.TOMLConfig
+	lp             LogPoller
 	txm            *txm.Txm
 	balanceMonitor services.Service
 	lggr           logger.Logger
@@ -312,6 +322,8 @@ func newChain(id string, cfg *config.TOMLConfig, ks core.Keystore, lggr logger.L
 		bc = internal.NewLoader[monitor.BalanceClient](func() (monitor.BalanceClient, error) { return ch.multiNode.SelectRPC() })
 	}
 
+	// TODO: import typeProvider function from codec package and pass to constructor
+	ch.lp = logpoller.New(logger.Sugared(logger.Named(lggr, "LogPoller")), logpoller.NewORM(ch.ID(), ds, lggr), ch.multiClient)
 	ch.txm = txm.NewTxm(ch.id, tc, sendTx, cfg, ks, lggr)
 	ch.balanceMonitor = monitor.NewBalanceMonitor(ch.id, cfg, lggr, ks, bc)
 	return &ch, nil
@@ -401,6 +413,10 @@ func (c *chain) Config() config.Config {
 	return c.cfg
 }
 
+func (c *chain) LogPoller() LogPoller {
+	return c.lp
+}
+
 func (c *chain) TxManager() TxManager {
 	return c.txm
 }

pkg/solana/client/client.go

@@ -41,6 +41,7 @@ type Reader interface {
 	GetTransaction(ctx context.Context, txHash solana.Signature, opts *rpc.GetTransactionOpts) (*rpc.GetTransactionResult, error)
 	GetBlocks(ctx context.Context, startSlot uint64, endSlot *uint64) (rpc.BlocksResult, error)
 	GetBlocksWithLimit(ctx context.Context, startSlot uint64, limit uint64) (*rpc.BlocksResult, error)
+	GetBlockWithOpts(context.Context, uint64, *rpc.GetBlockOpts) (*rpc.GetBlockResult, error)
 	GetBlock(ctx context.Context, slot uint64) (*rpc.GetBlockResult, error)
 	GetSignaturesForAddressWithOpts(ctx context.Context, addr solana.PublicKey, opts *rpc.GetSignaturesForAddressOpts) ([]*rpc.TransactionSignature, error)
 }
@@ -72,18 +73,25 @@ type Client struct {
 	requestGroup *singleflight.Group
 }
 
-func NewClient(endpoint string, cfg config.Config, requestTimeout time.Duration, log logger.Logger) (*Client, error) {
-	return &Client{
+// Return both the client and the underlying rpc client for testing
+func NewTestClient(endpoint string, cfg config.Config, requestTimeout time.Duration, log logger.Logger) (*Client, *rpc.Client, error) {
+	rpcClient := Client{
 		url:             endpoint,
-		rpc:             rpc.New(endpoint),
 		skipPreflight:   cfg.SkipPreflight(),
 		commitment:      cfg.Commitment(),
 		maxRetries:      cfg.MaxRetries(),
 		txTimeout:       cfg.TxTimeout(),
 		contextDuration: requestTimeout,
 		log:             log,
 		requestGroup:    &singleflight.Group{},
-	}, nil
+	}
+	rpcClient.rpc = rpc.New(endpoint)
+	return &rpcClient, rpcClient.rpc, nil
+}
+
+func NewClient(endpoint string, cfg config.Config, requestTimeout time.Duration, log logger.Logger) (*Client, error) {
+	rpcClient, _, err := NewTestClient(endpoint, cfg, requestTimeout, log)
+	return rpcClient, err
 }
 
 func (c *Client) latency(name string) func() {
@@ -346,6 +354,15 @@ func (c *Client) GetLatestBlockHeight(ctx context.Context) (uint64, error) {
 	return v.(uint64), err
 }
 
+func (c *Client) GetBlockWithOpts(ctx context.Context, slot uint64, opts *rpc.GetBlockOpts) (*rpc.GetBlockResult, error) {
+	// get block based on slot with custom options set
+	done := c.latency("get_block_with_opts")
+	defer done()
+	ctx, cancel := context.WithTimeout(ctx, c.txTimeout)
+	defer cancel()
+	return c.rpc.GetBlockWithOpts(ctx, slot, opts)
+}
+
 func (c *Client) GetBlock(ctx context.Context, slot uint64) (*rpc.GetBlockResult, error) {
 	// get block based on slot
 	done := c.latency("get_block")

pkg/solana/client/multi_client.go

@@ -166,3 +166,12 @@ func (m *MultiClient) GetSignaturesForAddressWithOpts(ctx context.Context, addr
 
 	return r.GetSignaturesForAddressWithOpts(ctx, addr, opts)
 }
+
+func (m *MultiClient) GetBlockWithOpts(ctx context.Context, slot uint64, opts *rpc.GetBlockOpts) (*rpc.GetBlockResult, error) {
+	r, err := m.getClient()
+	if err != nil {
+		return nil, err
+	}
+
+	return r.GetBlockWithOpts(ctx, slot, opts)
+}

pkg/solana/codec/solana.go

@@ -491,3 +491,36 @@ func saveDependency(refs *codecRefs, parent, child string) {
 
 	refs.dependencies[parent] = append(deps, child)
 }
+func NewIDLEventCodec(idl IDL, builder commonencodings.Builder) (commontypes.RemoteCodec, error) {
+	typeCodecs := make(commonencodings.LenientCodecFromTypeCodec)
+	refs := &codecRefs{
+		builder:      builder,
+		codecs:       make(map[string]commonencodings.TypeCodec),
+		typeDefs:     idl.Types,
+		dependencies: make(map[string][]string),
+	}
+
+	for _, event := range idl.Events {
+		name, instCodec, err := asStruct(eventFieldsAsStandardFields(event.Fields), refs, event.Name, false, false)
+		if err != nil {
+			return nil, err
+		}
+
+		typeCodecs[name] = instCodec
+	}
+
+	return typeCodecs, nil
+}
+
+func eventFieldsAsStandardFields(event []IdlEventField) []IdlField {
+	output := make([]IdlField, len(event))
+
+	for idx := range output {
+		output[idx] = IdlField{
+			Name: event[idx].Name,
+			Type: event[idx].Type,
+		}
+	}
+
+	return output
+}

pkg/solana/logpoller/discriminator.go

@@ -0,0 +1,14 @@
+package logpoller
+
+import (
+	"crypto/sha256"
+	"fmt"
+)
+
+const DiscriminatorLength = 8
+
+func Discriminator(namespace, name string) [DiscriminatorLength]byte {
+	h := sha256.New()
+	h.Write([]byte(fmt.Sprintf("%s:%s", namespace, name)))
+	return [DiscriminatorLength]byte(h.Sum(nil)[:DiscriminatorLength])
+}