feat: [Indexer] implemented event fetching and processing

pkg/cantonsdk/streaming/builder.go

@@ -0,0 +1,68 @@
+package streaming
+
+import (
+	"time"
+
+	lapiv2 "github.com/chainsafe/canton-middleware/pkg/cantonsdk/lapi/v2"
+)
+
+// FieldValue is an opaque DAML value used to construct LedgerEvents.
+// Use the Make* functions below to create values of each DAML type.
+// This keeps callers free of any direct lapiv2 dependency.
+type FieldValue struct{ v *lapiv2.Value }
+
+// MakeTextField wraps a Go string as a DAML Text value.
+func MakeTextField(s string) FieldValue {
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_Text{Text: s}}}
+}
+
+// MakePartyField wraps a party ID string as a DAML Party value.
+func MakePartyField(s string) FieldValue {
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_Party{Party: s}}}
+}
+
+// MakeNumericField wraps a decimal string as a DAML Numeric value.
+func MakeNumericField(s string) FieldValue {
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_Numeric{Numeric: s}}}
+}
+
+// MakeTimestampField wraps a time.Time as a DAML Timestamp value.
+func MakeTimestampField(t time.Time) FieldValue {
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_Timestamp{Timestamp: t.UnixMicro()}}}
+}
+
+// MakeNoneField returns a DAML Optional None value.
+func MakeNoneField() FieldValue {
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_Optional{Optional: &lapiv2.Optional{}}}}
+}
+
+// MakeSomePartyField returns a DAML Optional(Party) Some value.
+func MakeSomePartyField(party string) FieldValue {
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_Optional{Optional: &lapiv2.Optional{
+		Value: &lapiv2.Value{Sum: &lapiv2.Value_Party{Party: party}},
+	}}}}
+}
+
+// MakeRecordField builds a DAML Record value from a map of sub-fields.
+func MakeRecordField(fields map[string]FieldValue) FieldValue {
+	rf := make([]*lapiv2.RecordField, 0, len(fields))
+	for k, v := range fields {
+		rf = append(rf, &lapiv2.RecordField{Label: k, Value: v.v})
+	}
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_Record{Record: &lapiv2.Record{Fields: rf}}}}
+}
+
+// MakeSomeRecordField wraps a record in a DAML Optional(Record) Some value.
+func MakeSomeRecordField(fields map[string]FieldValue) FieldValue {
+	inner := MakeRecordField(fields)
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_Optional{Optional: &lapiv2.Optional{Value: inner.v}}}}
+}
+
+// MakeTextMapField builds a DAML TextMap value from a Go string map.
+func MakeTextMapField(entries map[string]string) FieldValue {
+	es := make([]*lapiv2.TextMap_Entry, 0, len(entries))
+	for k, v := range entries {
+		es = append(es, &lapiv2.TextMap_Entry{Key: k, Value: &lapiv2.Value{Sum: &lapiv2.Value_Text{Text: v}}})
+	}
+	return FieldValue{&lapiv2.Value{Sum: &lapiv2.Value_TextMap{TextMap: &lapiv2.TextMap{Entries: es}}}}
+}

pkg/cantonsdk/streaming/client.go

@@ -0,0 +1,262 @@
+package streaming
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync/atomic"
+	"time"
+
+	lapiv2 "github.com/chainsafe/canton-middleware/pkg/cantonsdk/lapi/v2"
+	"github.com/chainsafe/canton-middleware/pkg/cantonsdk/ledger"
+	"github.com/chainsafe/canton-middleware/pkg/cantonsdk/values"
+
+	"go.uber.org/zap"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+const (
+	reconnectBaseDelay = 5 * time.Second
+	reconnectMaxDelay  = 60 * time.Second
+	txChannelCap       = 100
+)
+
+// Streamer is the interface for opening a live Canton ledger stream.
+// *Client satisfies this interface.
+type Streamer interface {
+	Subscribe(ctx context.Context, req SubscribeRequest, lastOffset *int64) <-chan *LedgerTransaction
+}
+
+// Client wraps UpdateService.GetUpdates with automatic reconnection and auth handling.
+// It mirrors the streaming pattern established in pkg/cantonsdk/bridge/client.go.
+type Client struct {
+	ledger ledger.Ledger
+	party  string
+	logger *zap.Logger
+}
+
+// New creates a new streaming Client for the given ledger and party.
+func New(l ledger.Ledger, party string, opts ...Option) *Client {
+	s := applyOptions(opts)
+	return &Client{
+		ledger: l,
+		party:  party,
+		logger: s.logger,
+	}
+}
+
+// Subscribe opens a live stream against the Canton ledger and returns a read-only channel
+// of decoded transactions. It reconnects automatically with exponential backoff (5s → 60s)
+// on transient errors, and invalidates the auth token on 401/403.
+//
+// lastOffset is updated atomically after each received transaction so that reconnects
+// resume from the last safely received point. The caller is responsible for persisting
+// lastOffset to the database (the processor does this atomically with event writes).
+//
+// The returned channel is closed when ctx is canceled or a terminal error occurs
+// (io.EOF, context cancellation).
+func (c *Client) Subscribe(
+	ctx context.Context,
+	req SubscribeRequest,
+	lastOffset *int64,
+) <-chan *LedgerTransaction {
+	out := make(chan *LedgerTransaction, txChannelCap)
+
+	go func() {
+		defer close(out)
+
+		reconnectDelay := reconnectBaseDelay
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+
+			err := c.runStream(ctx, req.FromOffset, req.TemplateIDs, lastOffset, out)
+			if err == nil || errors.Is(err, io.EOF) || ctx.Err() != nil {
+				return
+			}
+
+			if isAuthError(err) {
+				c.ledger.InvalidateToken()
+				reconnectDelay = reconnectBaseDelay
+			}
+
+			// Advance FromOffset to where the stream last successfully delivered a
+			// transaction so the next connection resumes from the correct position.
+			req.FromOffset = atomic.LoadInt64(lastOffset)
+
+			c.logger.Warn("canton stream disconnected, reconnecting",
+				zap.Error(err),
+				zap.Int64("resume_offset", req.FromOffset),
+				zap.Duration("backoff", reconnectDelay),
+			)
+
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(reconnectDelay):
+			}
+
+			reconnectDelay = min(reconnectDelay*2, reconnectMaxDelay)
+		}
+	}()
+
+	return out
+}
+
+// runStream opens a single GetUpdates stream and forwards transactions to out until
+// the stream ends or ctx is canceled. It updates lastOffset atomically on each
+// received transaction.
+func (c *Client) runStream(
+	ctx context.Context,
+	fromOffset int64,
+	templateIDs []TemplateID,
+	lastOffset *int64,
+	out chan<- *LedgerTransaction,
+) error {
+	authCtx := c.ledger.AuthContext(ctx)
+
+	stream, err := c.ledger.Update().GetUpdates(authCtx, &lapiv2.GetUpdatesRequest{
+		BeginExclusive: fromOffset,
+		UpdateFormat: &lapiv2.UpdateFormat{
+			IncludeTransactions: &lapiv2.TransactionFormat{
+				EventFormat: &lapiv2.EventFormat{
+					FiltersByParty: map[string]*lapiv2.Filters{
+						c.party: buildTemplateFilters(templateIDs),
+					},
+					Verbose: true,
+				},
+				TransactionShape: lapiv2.TransactionShape_TRANSACTION_SHAPE_ACS_DELTA,
+			},
+		},
+	})
+	if err != nil {
+		if isAuthError(err) {
+			c.ledger.InvalidateToken()
+		}
+		return fmt.Errorf("open canton stream: %w", err)
+	}
+
+	for {
+		resp, err := stream.Recv()
+		if err != nil {
+			if isAuthError(err) {
+				c.ledger.InvalidateToken()
+			}
+			return err
+		}
+
+		tx := resp.GetTransaction()
+		if tx == nil {
+			// Checkpoint or topology update — nothing to index.
+			continue
+		}
+
+		lt := decodeLedgerTransaction(tx)
+		atomic.StoreInt64(lastOffset, lt.Offset)
+
+		select {
+		case out <- lt:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+// buildTemplateFilters constructs the Filters value for a set of TemplateIDs.
+//
+// This is the gRPC-level (template-level) filter. It controls which contract types
+// are delivered by the Canton Ledger API — reducing bandwidth to only the requested
+// templates. It is NOT an instrument filter: it cannot filter by contract field values
+// such as instrumentId. Instrument filtering (by InstrumentKey{Admin, ID}) happens
+// downstream in the parser after events are received.
+//
+// When TemplateID.PackageID is empty the filter matches the template across all
+// deployed package versions. Setting PackageID="" for CIP56.Events.TokenTransferEvent
+// enables indexing of any third-party CIP56-compliant token regardless of which
+// package version it was deployed with.
+func buildTemplateFilters(templateIDs []TemplateID) *lapiv2.Filters {
+	cumulative := make([]*lapiv2.CumulativeFilter, 0, len(templateIDs))
+	for _, tid := range templateIDs {
+		cumulative = append(cumulative, &lapiv2.CumulativeFilter{
+			IdentifierFilter: &lapiv2.CumulativeFilter_TemplateFilter{
+				TemplateFilter: &lapiv2.TemplateFilter{
+					TemplateId: &lapiv2.Identifier{
+						PackageId:  tid.PackageID, // empty = match all package versions
+						ModuleName: tid.ModuleName,
+						EntityName: tid.EntityName,
+					},
+				},
+			},
+		})
+	}
+	return &lapiv2.Filters{Cumulative: cumulative}
+}
+
+// decodeLedgerTransaction converts a proto Transaction into a LedgerTransaction.
+func decodeLedgerTransaction(tx *lapiv2.Transaction) *LedgerTransaction {
+	lt := &LedgerTransaction{
+		UpdateID:      tx.GetUpdateId(),
+		Offset:        tx.GetOffset(),
+		EffectiveTime: tx.GetEffectiveAt().AsTime(),
+		Events:        make([]*LedgerEvent, 0, len(tx.Events)),
+	}
+	for _, ev := range tx.Events {
+		if le := decodeLedgerEvent(ev); le != nil {
+			lt.Events = append(lt.Events, le)
+		}
+	}
+	return lt
+}
+
+// decodeLedgerEvent converts a proto Event to a LedgerEvent.
+// For created events the DAML CreateArguments are pre-decoded into LedgerEvent.fields
+// so that callers never need to import lapiv2 directly.
+// Returns nil for event kinds the indexer does not process (e.g. exercised events).
+func decodeLedgerEvent(ev *lapiv2.Event) *LedgerEvent {
+	if created := ev.GetCreated(); created != nil {
+		le := &LedgerEvent{
+			ContractID: created.GetContractId(),
+			IsCreated:  true,
+			fields:     values.RecordToMap(created.GetCreateArguments()),
+		}
+		if tid := created.GetTemplateId(); tid != nil {
+			le.PackageID = tid.GetPackageId()
+			le.ModuleName = tid.GetModuleName()
+			le.TemplateName = tid.GetEntityName()
+		}
+		return le
+	}
+
+	if archived := ev.GetArchived(); archived != nil {
+		le := &LedgerEvent{
+			ContractID: archived.GetContractId(),
+			IsCreated:  false,
+		}
+		if tid := archived.GetTemplateId(); tid != nil {
+			le.PackageID = tid.GetPackageId()
+			le.ModuleName = tid.GetModuleName()
+			le.TemplateName = tid.GetEntityName()
+		}
+		return le
+	}
+
+	return nil
+}
+
+// isAuthError returns true if err signals authentication or authorisation failure.
+func isAuthError(err error) bool {
+	if err == nil {
+		return false
+	}
+	st, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+	return st.Code() == codes.Unauthenticated || st.Code() == codes.PermissionDenied
+}

pkg/cantonsdk/streaming/options.go

@@ -0,0 +1,25 @@
+package streaming
+
+import "go.uber.org/zap"
+
+// Option configures a streaming Client.
+type Option func(*settings)
+
+type settings struct {
+	logger *zap.Logger
+}
+
+// WithLogger sets a custom logger on the streaming Client.
+func WithLogger(l *zap.Logger) Option {
+	return func(s *settings) { s.logger = l }
+}
+
+func applyOptions(opts []Option) settings {
+	s := settings{logger: zap.NewNop()}
+	for _, opt := range opts {
+		if opt != nil {
+			opt(&s)
+		}
+	}
+	return s
+}