feat(api): transfer history + outgoing/expired endpoints (#331)

* feat(api): transfer to external party via party id

Enable sending a token to a Canton party that is not registered in the
middleware (e.g. a party on an external participant node), addressed by
its party id rather than an EVM address.

- Non-custodial: PrepareRequest accepts an optional to_party_id; when set,
  Prepare uses it directly and skips the userstore recipient lookup.
  Reuses the existing /prepare + /execute flow.
- Custodial: new POST /api/v2/transfer/custodial does a single server-signed
  transfer via TransferByPartyID (the middleware holds the user's key).
- Shared validatePartyID helper (hint::fingerprint) and self-transfer guard.

Authorization/whitelist policy and outgoing-offer tracking are out of scope
(tracked separately).

* fix(api): reject ambiguous recipient in Prepare at the service layer

Address review feedback: Prepare now explicitly rejects a request that
sets both (or neither) of `to` and `to_party_id` instead of silently
preferring `to_party_id`. The HTTP handler already enforced this; this
adds the same guard at the service boundary for direct callers.

* feat(indexer): move storage to generalized indexer_transfers table

Introduce a single indexer_transfers table holding both direct (atomic
CIP-56) and offer-based (2-step, e.g. USDCx) transfers with a mutable status
lifecycle, and retire indexer_pending_offers. The indexer's external API is
unchanged: the existing pending-offers endpoints are served by a thin
compatibility shim over the new table, so the api-server and custodial accept
worker are untouched (API modernization is a follow-up PR).

- Migration 7: create indexer_transfers, backfill history from the offers
  table + the TRANSFER event log, then drop indexer_pending_offers. No data
  lost; the two sources are disjoint so nothing is double-counted.
- Processor: offer CREATE -> insert pending transfer; ARCHIVE -> complete;
  CIP-56 TRANSFER event -> upsert a direct (completed) transfer. Balances are
  untouched (still maintained from the event/holding paths).
- Store: new transfer write methods; ListPendingOffersForParty /
  ListAllPendingOffers reimplemented over indexer_transfers, returning the
  existing PendingOffer shape.

* refactor(indexer): address review on transfers PR

- Remove the unused Transfer.Archived field (the offer decoder uses
  PendingOffer.IsArchived in this PR).
- Document why the migration loads rows in memory (the per-participant
  transfer history is small) and note the INSERT…SELECT alternative if it
  ever grows.
- Add composite (from_party_id, created_at) / (to_party_id, created_at)
  indexes so the list-by-party-newest-first queries avoid a filesort.

* refactor(indexer): single InsertTransfer; review fixes on transfers PR

Address PR review:
- Consolidate to one write method: InsertTransfer now records a transfer with
  the caller-provided Kind/Status (offer/pending or direct/completed) instead
  of two methods that each forced them. Removes UpsertDirectTransfer. The
  ON CONFLICT (contract_id) DO NOTHING makes it idempotent for ledger-stream
  replays (the indexer re-reads from the last saved offset on restart).
- migration 5: clearer comment on legacyOffer — PendingOfferDao was removed,
  the struct stays here for backward compat so the historical migrations can
  manage the legacy table with bun.
- migration 7: order the backfill by ledger_offset (canonical monotonic
  sequence) instead of created_at (an effective-time that can tie).

* feat(api): transfer history + outgoing/expired endpoints on indexer_transfers

Re-synced on the updated storage PR (#330): inherits the single InsertTransfer
write method (no UpsertDirectTransfer) and the migration fixes.

- Indexer: /parties/{id}/pending-offers -> /parties/{id}/transfers?role=&status=
  + /pending-transfers; ListTransfers is a single-table query with native
  pagination and a derived "expired" status. Removes the PendingOffer compat
  shim; decoder/processor now produce Transfer.
- api-server: /api/v2/transfer/incoming (unchanged shape) + /outgoing and
  /completed.
- Accept worker reads GetPendingTransfers, oldest-pending first (FIFO).

* docs(indexer): fix stale ListPendingTransfers order comment

The doc comment still said 'created_at DESC' from before the FIFO change; the
query is created_at ASC (oldest first). Also reword the inline comment to be
precise — it's FIFO (longest-pending first), which correlates with nearest
expiry rather than strictly ordering by it.

* test(indexer): mirror migration 7 composite indexes in pg_test setup

Author: sadiq1971

pkg/app/api/server.go

@@ -115,7 +115,7 @@ func (s *Server) Run() error {
 	logger.Info("Connected to Canton")
 
 	// The indexer is now a hard dependency: the token provider (when in
-	// indexer mode), the accept worker, and the pending-offers list endpoint
+	// indexer mode), the accept worker, and the transfers list endpoints
 	// all read from it. Build the HTTP client once and share — separate
 	// instances would just open redundant idle connections to the same host.
 	indexerClient, err := buildIndexerClient(cfg, reg)
@@ -185,7 +185,7 @@ func (s *Server) Run() error {
 // the dual configuration exists for historical reasons and is consolidated
 // here so the rest of the code never has to pick between them.
 //
-// An indexer is now required (the pending-offers endpoint backs onto it), so
+// An indexer is now required (the transfer list/history endpoints back onto it), so
 // startup fails fast if neither location is configured. The returned client is
 // wrapped with metrics so all outbound indexer calls are observed.
 func buildIndexerClient(cfg *config.APIServer, reg sharedmetrics.NamespacedRegisterer) (indexerclient.Client, error) {

pkg/custodial/accept_worker.go

@@ -115,13 +115,13 @@ func (w *AcceptWorker) acceptPending(ctx context.Context) {
 
 	page := 1
 	for {
-		result, err := w.indexer.GetAllPendingOffers(ctx, indexer.Pagination{
+		result, err := w.indexer.GetPendingTransfers(ctx, indexer.Pagination{
 			Page:  page,
 			Limit: acceptWorkerPageLimit,
 		})
 		if err != nil {
 			w.metrics.ErrorsTotal.WithLabelValues("fetch_offers").Inc()
-			w.logger.Warn("accept worker: failed to fetch pending offers", zap.Error(err))
+			w.logger.Warn("accept worker: failed to fetch pending transfers", zap.Error(err))
 			return
 		}
 
@@ -134,35 +134,36 @@ func (w *AcceptWorker) acceptPending(ctx context.Context) {
 		w.metrics.OffersFetchedTotal.Add(float64(len(result.Items)))
 
 		for i := range result.Items {
-			offer := result.Items[i]
-			if !custodialParties[offer.ReceiverPartyID] {
+			transfer := result.Items[i]
+			// ToPartyID is the offer receiver — the custodial party that must accept.
+			if !custodialParties[transfer.ToPartyID] {
 				continue
 			}
 			acceptStart := time.Now()
 			err := w.cantonToken.AcceptTransferInstruction(
-				ctx, offer.ReceiverPartyID, offer.ContractID, offer.InstrumentAdmin,
+				ctx, transfer.ToPartyID, transfer.ContractID, transfer.InstrumentAdmin,
 			)
 			w.metrics.OfferAcceptDuration.Observe(time.Since(acceptStart).Seconds())
 			if err != nil {
 				w.metrics.OffersAcceptedTotal.WithLabelValues("error").Inc()
 				w.logger.Warn("accept worker: failed to accept offer",
-					zap.String("party_id", offer.ReceiverPartyID),
-					zap.String("contract_id", offer.ContractID),
+					zap.String("party_id", transfer.ToPartyID),
+					zap.String("contract_id", transfer.ContractID),
 					zap.Error(err),
 				)
 				continue
 			}
 			w.metrics.OffersAcceptedTotal.WithLabelValues("success").Inc()
 			w.logger.Info("accept worker: accepted transfer offer",
-				zap.String("party_id", offer.ReceiverPartyID),
-				zap.String("contract_id", offer.ContractID),
-				zap.String("sender", offer.SenderPartyID),
-				zap.String("amount", offer.Amount),
+				zap.String("party_id", transfer.ToPartyID),
+				zap.String("contract_id", transfer.ContractID),
+				zap.String("sender", transfer.FromPartyID),
+				zap.String("amount", transfer.Amount),
 			)
 		}
 
 		if int64(page*acceptWorkerPageLimit) >= result.Total {
-			// Reached the end of the pending-offers stream — cycle completed
+			// Reached the end of the pending-transfers stream — cycle completed
 			// without an unrecoverable error. Per-offer failures don't count
 			// against the cycle's success: those are tracked in
 			// OffersAcceptedTotal{result="error"}.

pkg/custodial/accept_worker_test.go

@@ -28,21 +28,22 @@ func custodialUser() *user.User {
 
 const testInstrumentAdmin = "admin-party::zzz"
 
-func pendingOffer(contractID string) indexer.PendingOffer {
-	return indexer.PendingOffer{
+func pendingOffer(contractID string) indexer.Transfer {
+	return indexer.Transfer{
 		ContractID:      contractID,
-		Status:          indexer.OfferStatusPending,
-		ReceiverPartyID: "custodial-party::abc",
-		SenderPartyID:   "sender-party::xyz",
+		Kind:            indexer.TransferKindOffer,
+		Status:          indexer.TransferStatusPending,
+		ToPartyID:       "custodial-party::abc",
+		FromPartyID:     "sender-party::xyz",
 		InstrumentAdmin: testInstrumentAdmin,
 		InstrumentID:    "USDCX",
 		Amount:          "100.0",
 		LedgerOffset:    42,
 	}
 }
 
-func allOffersPage(offers ...indexer.PendingOffer) *indexer.Page[indexer.PendingOffer] {
-	return &indexer.Page[indexer.PendingOffer]{
+func allOffersPage(offers ...indexer.Transfer) *indexer.Page[indexer.Transfer] {
+	return &indexer.Page[indexer.Transfer]{
 		Items: offers,
 		Total: int64(len(offers)),
 		Page:  1,
@@ -71,7 +72,7 @@ func TestAcceptWorker_AcceptsSingleOffer(t *testing.T) {
 	page := allOffersPage(offer)
 
 	lister.EXPECT().ListCustodialUsers(mock.Anything).Return([]*user.User{custodialUser()}, nil)
-	ic.EXPECT().GetAllPendingOffers(mock.Anything, indexer.Pagination{Page: 1, Limit: acceptWorkerPageLimit}).
+	ic.EXPECT().GetPendingTransfers(mock.Anything, indexer.Pagination{Page: 1, Limit: acceptWorkerPageLimit}).
 		Return(page, nil)
 	tok.EXPECT().AcceptTransferInstruction(mock.Anything, "custodial-party::abc", "contract-1", testInstrumentAdmin).
 		Return(nil)
@@ -86,19 +87,19 @@ func TestAcceptWorker_SkipsOffersForUnregisteredParties(t *testing.T) {
 	ic := indexermocks.NewClient(t)
 
 	// Offer for a party not in our custodialParties map
-	unregisteredOffer := indexer.PendingOffer{
+	unregisteredOffer := indexer.Transfer{
 		ContractID:      "contract-unknown",
-		ReceiverPartyID: "unknown-party::xyz",
+		ToPartyID:       "unknown-party::xyz",
 		InstrumentAdmin: testInstrumentAdmin,
 	}
 	custodialOffer := pendingOffer("contract-custodial")
-	page := &indexer.Page[indexer.PendingOffer]{
-		Items: []indexer.PendingOffer{unregisteredOffer, custodialOffer},
+	page := &indexer.Page[indexer.Transfer]{
+		Items: []indexer.Transfer{unregisteredOffer, custodialOffer},
 		Total: 2, Page: 1, Limit: acceptWorkerPageLimit,
 	}
 
 	lister.EXPECT().ListCustodialUsers(mock.Anything).Return([]*user.User{custodialUser()}, nil)
-	ic.EXPECT().GetAllPendingOffers(mock.Anything, mock.Anything).Return(page, nil)
+	ic.EXPECT().GetPendingTransfers(mock.Anything, mock.Anything).Return(page, nil)
 	// Only the registered custodial offer should be accepted
 	tok.EXPECT().AcceptTransferInstruction(mock.Anything, "custodial-party::abc", "contract-custodial", testInstrumentAdmin).
 		Return(nil)
@@ -114,13 +115,13 @@ func TestAcceptWorker_LogsAndContinuesOnAcceptError(t *testing.T) {
 
 	offer1 := pendingOffer("contract-1")
 	offer2 := pendingOffer("contract-2")
-	page := &indexer.Page[indexer.PendingOffer]{
-		Items: []indexer.PendingOffer{offer1, offer2},
+	page := &indexer.Page[indexer.Transfer]{
+		Items: []indexer.Transfer{offer1, offer2},
 		Total: 2, Page: 1, Limit: acceptWorkerPageLimit,
 	}
 
 	lister.EXPECT().ListCustodialUsers(mock.Anything).Return([]*user.User{custodialUser()}, nil)
-	ic.EXPECT().GetAllPendingOffers(mock.Anything, mock.Anything).Return(page, nil)
+	ic.EXPECT().GetPendingTransfers(mock.Anything, mock.Anything).Return(page, nil)
 	tok.EXPECT().AcceptTransferInstruction(mock.Anything, mock.Anything, "contract-1", mock.Anything).
 		Return(errors.New("ALREADY_EXISTS"))
 	tok.EXPECT().AcceptTransferInstruction(mock.Anything, mock.Anything, "contract-2", mock.Anything).
@@ -136,7 +137,7 @@ func TestAcceptWorker_LogsAndContinuesOnIndexerError(t *testing.T) {
 	ic := indexermocks.NewClient(t)
 
 	lister.EXPECT().ListCustodialUsers(mock.Anything).Return([]*user.User{custodialUser()}, nil)
-	ic.EXPECT().GetAllPendingOffers(mock.Anything, mock.Anything).
+	ic.EXPECT().GetPendingTransfers(mock.Anything, mock.Anything).
 		Return(nil, errors.New("indexer unavailable"))
 
 	worker := NewAcceptWorker(tok, lister, ic, time.Hour, NewNopMetrics(), zap.NewNop())
@@ -176,13 +177,13 @@ func TestAcceptWorker_PaginatesAllOffers(t *testing.T) {
 	offer2 := pendingOffer("contract-2")
 
 	// Total is 400, limit 200 → 2 pages
-	page1 := &indexer.Page[indexer.PendingOffer]{Items: []indexer.PendingOffer{offer1}, Total: 400, Page: 1, Limit: acceptWorkerPageLimit}
-	page2 := &indexer.Page[indexer.PendingOffer]{Items: []indexer.PendingOffer{offer2}, Total: 400, Page: 2, Limit: acceptWorkerPageLimit}
+	page1 := &indexer.Page[indexer.Transfer]{Items: []indexer.Transfer{offer1}, Total: 400, Page: 1, Limit: acceptWorkerPageLimit}
+	page2 := &indexer.Page[indexer.Transfer]{Items: []indexer.Transfer{offer2}, Total: 400, Page: 2, Limit: acceptWorkerPageLimit}
 
 	lister.EXPECT().ListCustodialUsers(mock.Anything).Return([]*user.User{custodialUser()}, nil)
-	ic.EXPECT().GetAllPendingOffers(mock.Anything, indexer.Pagination{Page: 1, Limit: acceptWorkerPageLimit}).
+	ic.EXPECT().GetPendingTransfers(mock.Anything, indexer.Pagination{Page: 1, Limit: acceptWorkerPageLimit}).
 		Return(page1, nil)
-	ic.EXPECT().GetAllPendingOffers(mock.Anything, indexer.Pagination{Page: 2, Limit: acceptWorkerPageLimit}).
+	ic.EXPECT().GetPendingTransfers(mock.Anything, indexer.Pagination{Page: 2, Limit: acceptWorkerPageLimit}).
 		Return(page2, nil)
 	tok.EXPECT().AcceptTransferInstruction(mock.Anything, mock.Anything, "contract-1", mock.Anything).Return(nil)
 	tok.EXPECT().AcceptTransferInstruction(mock.Anything, mock.Anything, "contract-2", mock.Anything).Return(nil)

pkg/custodial/metrics.go

@@ -33,7 +33,7 @@ type Metrics struct {
 
 	// ErrorsTotal counts cycles that aborted early.
 	//   phase=list_users   – ListCustodialUsers failed
-	//   phase=fetch_offers – GetAllPendingOffers failed on some page
+	//   phase=fetch_offers – GetPendingTransfers failed on some page
 	// Per-offer accept failures are *not* counted here (see OffersAccepted).
 	ErrorsTotal *prometheus.CounterVec
 

pkg/indexer/client/http.go

@@ -50,9 +50,11 @@ type Client interface {
 		p indexer.Pagination,
 	) (*indexer.Page[*indexer.ParsedEvent], error)
 
-	// Pending inbound transfers
-	GetPendingOffersForParty(ctx context.Context, partyID string, p indexer.Pagination) (*indexer.Page[indexer.PendingOffer], error)
-	GetAllPendingOffers(ctx context.Context, p indexer.Pagination) (*indexer.Page[indexer.PendingOffer], error)
+	// Transfers
+	GetTransfers(
+		ctx context.Context, partyID string, query indexer.TransferQuery, p indexer.Pagination,
+	) (*indexer.Page[indexer.Transfer], error)
+	GetPendingTransfers(ctx context.Context, p indexer.Pagination) (*indexer.Page[indexer.Transfer], error)
 }
 
 // HTTP implements Client by calling the indexer's unauthenticated admin HTTP API.
@@ -193,26 +195,35 @@ func (c *HTTP) ListPartyEvents(
 	return &page, nil
 }
 
-// GetPendingOffersForParty calls GET /indexer/v1/admin/parties/{partyID}/pending-offers.
-func (c *HTTP) GetPendingOffersForParty(
-	ctx context.Context, partyID string, p indexer.Pagination,
-) (*indexer.Page[indexer.PendingOffer], error) {
-	u := c.partyBase(partyID) + "/pending-offers?" + pageQuery(p).Encode()
-	var page indexer.Page[indexer.PendingOffer]
+// GetTransfers calls GET /indexer/v1/admin/parties/{partyID}/transfers,
+// filtering by role (receiver/sender/any) and status (pending/expired/completed;
+// "" = all).
+func (c *HTTP) GetTransfers(
+	ctx context.Context, partyID string, query indexer.TransferQuery, p indexer.Pagination,
+) (*indexer.Page[indexer.Transfer], error) {
+	q := pageQuery(p)
+	if query.Role != "" {
+		q.Set("role", string(query.Role))
+	}
+	if query.Status != "" {
+		q.Set("status", query.Status)
+	}
+	u := c.partyBase(partyID) + "/transfers?" + q.Encode()
+	var page indexer.Page[indexer.Transfer]
 	if err := c.getJSON(ctx, u, &page); err != nil {
-		return nil, fmt.Errorf("pending offers for party %s: %w", partyID, err)
+		return nil, fmt.Errorf("transfers for party %s: %w", partyID, err)
 	}
 	return &page, nil
 }
 
-// GetAllPendingOffers calls GET /indexer/v1/admin/pending-offers.
-func (c *HTTP) GetAllPendingOffers(
+// GetPendingTransfers calls GET /indexer/v1/admin/pending-transfers.
+func (c *HTTP) GetPendingTransfers(
 	ctx context.Context, p indexer.Pagination,
-) (*indexer.Page[indexer.PendingOffer], error) {
-	u := c.baseURL + "/indexer/v1/admin/pending-offers?" + pageQuery(p).Encode()
-	var page indexer.Page[indexer.PendingOffer]
+) (*indexer.Page[indexer.Transfer], error) {
+	u := c.baseURL + "/indexer/v1/admin/pending-transfers?" + pageQuery(p).Encode()
+	var page indexer.Page[indexer.Transfer]
 	if err := c.getJSON(ctx, u, &page); err != nil {
-		return nil, fmt.Errorf("get all pending offers: %w", err)
+		return nil, fmt.Errorf("get pending transfers: %w", err)
 	}
 	return &page, nil
 }

pkg/indexer/client/instrumented.go

@@ -111,20 +111,20 @@ func (c *InstrumentedClient) ListPartyEvents(
 	return page, err
 }
 
-func (c *InstrumentedClient) GetPendingOffersForParty(
-	ctx context.Context, partyID string, p indexer.Pagination,
-) (*indexer.Page[indexer.PendingOffer], error) {
-	defer c.observe(OpGetPendingOffersForPty)()
-	page, err := c.inner.GetPendingOffersForParty(ctx, partyID, p)
-	c.incErr(OpGetPendingOffersForPty, err)
+func (c *InstrumentedClient) GetTransfers(
+	ctx context.Context, partyID string, query indexer.TransferQuery, p indexer.Pagination,
+) (*indexer.Page[indexer.Transfer], error) {
+	defer c.observe(OpGetTransfers)()
+	page, err := c.inner.GetTransfers(ctx, partyID, query, p)
+	c.incErr(OpGetTransfers, err)
 	return page, err
 }
 
-func (c *InstrumentedClient) GetAllPendingOffers(
+func (c *InstrumentedClient) GetPendingTransfers(
 	ctx context.Context, p indexer.Pagination,
-) (*indexer.Page[indexer.PendingOffer], error) {
-	defer c.observe(OpGetAllPendingOffers)()
-	page, err := c.inner.GetAllPendingOffers(ctx, p)
-	c.incErr(OpGetAllPendingOffers, err)
+) (*indexer.Page[indexer.Transfer], error) {
+	defer c.observe(OpGetPendingTransfers)()
+	page, err := c.inner.GetPendingTransfers(ctx, p)
+	c.incErr(OpGetPendingTransfers, err)
 	return page, err
 }

pkg/indexer/client/metrics.go

@@ -11,17 +11,17 @@ import (
 type ClientOperation string
 
 const (
-	OpGetToken               ClientOperation = "get_token"
-	OpListTokens             ClientOperation = "list_tokens"
-	OpTotalSupply            ClientOperation = "total_supply"
-	OpGetBalance             ClientOperation = "get_balance"
-	OpListBalancesForParty   ClientOperation = "list_balances_for_party"
-	OpListBalancesForToken   ClientOperation = "list_balances_for_token"
-	OpGetEvent               ClientOperation = "get_event"
-	OpListTokenEvents        ClientOperation = "list_token_events"
-	OpListPartyEvents        ClientOperation = "list_party_events"
-	OpGetPendingOffersForPty ClientOperation = "get_pending_offers_for_party"
-	OpGetAllPendingOffers    ClientOperation = "get_all_pending_offers"
+	OpGetToken             ClientOperation = "get_token"
+	OpListTokens           ClientOperation = "list_tokens"
+	OpTotalSupply          ClientOperation = "total_supply"
+	OpGetBalance           ClientOperation = "get_balance"
+	OpListBalancesForParty ClientOperation = "list_balances_for_party"
+	OpListBalancesForToken ClientOperation = "list_balances_for_token"
+	OpGetEvent             ClientOperation = "get_event"
+	OpListTokenEvents      ClientOperation = "list_token_events"
+	OpListPartyEvents      ClientOperation = "list_party_events"
+	OpGetTransfers         ClientOperation = "get_transfers"
+	OpGetPendingTransfers  ClientOperation = "get_pending_transfers"
 )
 
 // Metrics holds Prometheus collectors for the indexer HTTP client.