engine

type EventFetcher interface {
	// Start begins streaming from offset in a background goroutine.
	// Must be called exactly once before Events is used.
	Start(ctx context.Context, offset int64)

	// Events returns the read-only channel of decoded batches.
	// The channel is closed when the stream terminates.
	Events() <-chan *streaming.Batch[any]
}

type Fetcher struct {
	// contains filtered or unexported fields
}

type Metrics struct {

	// EventsProcessedTotal counts events successfully indexed, partitioned by
	// event type (MINT, BURN, TRANSFER).
	EventsProcessedTotal *prometheus.CounterVec

	// BatchProcessingErrors counts batch processing failures that triggered a
	// retry. Each increment represents one failed attempt, not one lost batch.
	BatchProcessingErrors prometheus.Counter

	// SyncLagSeconds reports how far behind real-time the indexer is, measured
	// as time.Since(lastEvent.EffectiveTime). Updated after each non-empty batch.
	SyncLagSeconds prometheus.Gauge

	// LastOffset is the ledger offset of the most recently committed batch.
	// Updated after every successful SaveOffset.
	LastOffset prometheus.Gauge
}

type Processor struct {
	// contains filtered or unexported fields
}

type Store interface {
	// LatestOffset returns the last successfully persisted ledger offset.
	// Returns 0 and no error when no offset has been stored yet (fresh start).
	// Called once at startup, outside any transaction.
	LatestOffset(ctx context.Context) (int64, error)

	// RunInTx executes fn inside a single database transaction.
	// On success fn's return value is nil and the transaction is committed.
	// On any error the transaction is rolled back and the error is returned.
	// The Store passed to fn is scoped to the transaction — all methods on it
	// participate in the same underlying DB transaction.
	RunInTx(ctx context.Context, fn func(ctx context.Context, tx Store) error) error

	// InsertEvent persists one ParsedEvent by ContractID.
	// Returns inserted=false when the event already exists and should therefore not
	// mutate any derived state a second time.
	InsertEvent(ctx context.Context, event *indexer.ParsedEvent) (inserted bool, err error)

	// SaveOffset advances the stored ledger offset after all newly inserted events in
	// the transaction have updated derived state. It must be safe to call even when the
	// batch was empty or every event was already present.
	SaveOffset(ctx context.Context, offset int64) error

	// UpsertToken records a token deployment on first observation.
	// Subsequent calls for the same {InstrumentAdmin, InstrumentID} are no-ops
	// (ON CONFLICT DO NOTHING).
	UpsertToken(ctx context.Context, token *indexer.Token) error

	// ApplyBalanceDelta adjusts a party's token balance by delta (signed decimal string).
	// The balance row is created at zero if it does not yet exist, then delta is added.
	// The store must also update Token.HolderCount atomically:
	//   - increment when a party's balance transitions from zero to positive
	//   - decrement when a party's balance transitions from positive to zero
	ApplyBalanceDelta(ctx context.Context, partyID, instrumentAdmin, instrumentID, delta string) error

	// ApplySupplyDelta adjusts a token's TotalSupply by delta (signed decimal string).
	// Called once per mint (+amount) or burn (-amount). Transfer events must not call this.
	ApplySupplyDelta(ctx context.Context, instrumentAdmin, instrumentID, delta string) error

	// InsertPendingOffer records a new TransferOffer (idempotent by ContractID).
	// Status is set to PENDING on insert.
	InsertPendingOffer(ctx context.Context, offer *indexer.PendingOffer) error

	// MarkOfferAccepted transitions a TransferOffer to ACCEPTED status when the Canton
	// ledger emits an ARCHIVED event for the contract (receiver exercised Accept, or the
	// offer was rejected/expired). The row is kept for audit history; no-op when not found.
	MarkOfferAccepted(ctx context.Context, contractID string) error

	// InsertHolding records an active Utility.Registry.Holding contract so its amount
	// and owner can be recovered when the contract is later archived (archive events
	// carry only the contract ID). Idempotent on ContractID.
	InsertHolding(ctx context.Context, h *indexer.HoldingChange) error

	// TakeHolding deletes the row for contractID and returns the stored owner/
	// instrument/amount needed to decrement the matching balance on archive.
	// Returns ok=false on missing rows so replayed ARCHIVED events become no-ops
	// instead of errors.
	TakeHolding(ctx context.Context, contractID string) (h indexer.HoldingChange, ok bool, err error)
}