diff --git a/cmd/darepocli/darepoclicommands/cmd_list.go b/cmd/darepocli/darepoclicommands/cmd_list.go index ea8ee62ef..29dcdfb23 100644 --- a/cmd/darepocli/darepoclicommands/cmd_list.go +++ b/cmd/darepocli/darepoclicommands/cmd_list.go @@ -20,6 +20,7 @@ func newActivityCmd() *cobra.Command { " darepocli activity\n" + " darepocli activity --pending --kind send,recv\n" + " darepocli activity --format json\n" + + " darepocli activity --cursor \n" + " darepocli activity inspect ", Args: cobra.NoArgs, RunE: walletActivity, @@ -31,7 +32,8 @@ func newActivityCmd() *cobra.Command { "filter by kind (send,recv,deposit,exit); repeatable") cmd.Flags().Uint32("limit", 0, "page size; 0 uses the daemon default") - cmd.Flags().Uint32("offset", 0, "pagination offset") + cmd.Flags().String("cursor", "", + "activity page token from a prior page's next_cursor") cmd.Flags().String("format", "table", "output format (table|expanded|x|json)") @@ -45,7 +47,7 @@ func walletActivity(cmd *cobra.Command, _ []string) error { pending, _ := cmd.Flags().GetBool("pending") kinds, _ := cmd.Flags().GetStringSlice("kind") limit, _ := cmd.Flags().GetUint32("limit") - offset, _ := cmd.Flags().GetUint32("offset") + cursor, _ := cmd.Flags().GetString("cursor") format, _ := cmd.Flags().GetString("format") if err := validateListFormat( @@ -58,7 +60,7 @@ func walletActivity(cmd *cobra.Command, _ []string) error { View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, PendingOnly: pending, Limit: limit, - Offset: offset, + Cursor: cursor, } for _, k := range kinds { parsed, err := parseEntryKind(k) @@ -77,10 +79,22 @@ func walletActivity(cmd *cobra.Command, _ []string) error { switch format { case "", "table": - return printWalletActivityTable(resp) + if err := printWalletActivityTable( + resp, + ); err != nil { + return err + } + + return printWalletActivityNextPage(resp) case "expanded", "x": - return printWalletActivityExpanded(resp) + if err := printWalletActivityExpanded( + resp, + ); err != nil { + return err + } + + return printWalletActivityNextPage(resp) case "json": return printWalletProto(resp) diff --git a/cmd/darepocli/darepoclicommands/mcp_wallet.go b/cmd/darepocli/darepoclicommands/mcp_wallet.go index deafb458f..95c26a98d 100644 --- a/cmd/darepocli/darepoclicommands/mcp_wallet.go +++ b/cmd/darepocli/darepoclicommands/mcp_wallet.go @@ -54,10 +54,10 @@ func registerMCPWalletQueryTools(s *mcp.Server, }) type activityArgs struct { - PendingOnly bool `json:"pending_only,omitempty" jsonschema:"filter to in-flight entries"` //nolint:ll - Kinds []string `json:"kinds,omitempty" jsonschema:"kind filter (send, recv, deposit, exit)"` //nolint:ll - Limit uint32 `json:"limit,omitempty" jsonschema:"page size; zero uses daemon default"` //nolint:ll - Offset uint32 `json:"offset,omitempty" jsonschema:"pagination offset"` //nolint:ll + PendingOnly bool `json:"pending_only,omitempty" jsonschema:"filter to in-flight entries"` //nolint:ll + Kinds []string `json:"kinds,omitempty" jsonschema:"kind filter (send, recv, deposit, exit)"` //nolint:ll + Limit uint32 `json:"limit,omitempty" jsonschema:"page size; zero uses daemon default"` //nolint:ll + Cursor string `json:"cursor,omitempty" jsonschema:"activity page token; pass next_cursor to page"` //nolint:ll } mcp.AddTool(s, &mcp.Tool{ Name: "activity", @@ -66,7 +66,7 @@ func registerMCPWalletQueryTools(s *mcp.Server, args activityArgs) (*mcp.CallToolResult, any, error) { req, err := buildWalletActivityRequest( - args.PendingOnly, args.Kinds, args.Limit, args.Offset, + args.PendingOnly, args.Kinds, args.Limit, args.Cursor, ) if err != nil { return nil, nil, err @@ -267,14 +267,14 @@ func registerMCPWalletMutateTools(s *mcp.Server, // buildWalletActivityRequest translates MCP activityArgs into a ListRequest, // applying the same activity filter parsing the CLI uses so the two // surfaces stay in lockstep. -func buildWalletActivityRequest(pendingOnly bool, kinds []string, limit, - offset uint32) (*walletdkrpc.ListRequest, error) { +func buildWalletActivityRequest(pendingOnly bool, kinds []string, limit uint32, + cursor string) (*walletdkrpc.ListRequest, error) { req := &walletdkrpc.ListRequest{ View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, PendingOnly: pendingOnly, Limit: limit, - Offset: offset, + Cursor: cursor, } for _, k := range kinds { diff --git a/cmd/darepocli/darepoclicommands/mcp_wallet_test.go b/cmd/darepocli/darepoclicommands/mcp_wallet_test.go index 82d31a526..e231cbe27 100644 --- a/cmd/darepocli/darepoclicommands/mcp_wallet_test.go +++ b/cmd/darepocli/darepoclicommands/mcp_wallet_test.go @@ -125,7 +125,7 @@ func TestBuildWalletActivityRequestHappyPath(t *testing.T) { t.Parallel() req, err := buildWalletActivityRequest( - true, []string{"send", "recv"}, 50, 100, + true, []string{"send", "recv"}, 50, "cursor-token", ) require.NoError(t, err) require.Equal( @@ -134,5 +134,5 @@ func TestBuildWalletActivityRequestHappyPath(t *testing.T) { require.True(t, req.GetPendingOnly()) require.Len(t, req.GetKinds(), 2) require.Equal(t, uint32(50), req.GetLimit()) - require.Equal(t, uint32(100), req.GetOffset()) + require.Equal(t, "cursor-token", req.GetCursor()) } diff --git a/cmd/darepocli/darepoclicommands/wallet_table.go b/cmd/darepocli/darepoclicommands/wallet_table.go index 152940e4a..bd83fb2c5 100644 --- a/cmd/darepocli/darepoclicommands/wallet_table.go +++ b/cmd/darepocli/darepoclicommands/wallet_table.go @@ -43,6 +43,24 @@ func printWalletActivityExpanded(resp *walletdkrpc.ListResponse) error { return renderWalletActivityExpanded(os.Stdout, resp) } +// printWalletActivityNextPage writes a next-page hint to stdout when the +// activity feed has more entries. The human-facing table and expanded views +// omit the raw cursor otherwise, so without this line a caller has no way to +// discover the token needed to reach page two. +func printWalletActivityNextPage(resp *walletdkrpc.ListResponse) error { + activity := resp.GetActivity() + if !activity.GetHasMore() { + return nil + } + + _, err := fmt.Fprintf( + os.Stdout, "\nmore entries available; next page: --cursor %s\n", + activity.GetNextCursor(), + ) + + return err +} + // renderWalletActivityTable renders activity entries as a tabwriter table. func renderWalletActivityTable(out io.Writer, resp *walletdkrpc.ListResponse) error { diff --git a/darepod/config.go b/darepod/config.go index f95d0b4a3..22bc92bc2 100644 --- a/darepod/config.go +++ b/darepod/config.go @@ -16,6 +16,7 @@ import ( "github.com/lightninglabs/darepo-client/chainbackends" "github.com/lightninglabs/darepo-client/credit" "github.com/lightninglabs/darepo-client/db" + "github.com/lightninglabs/darepo-client/db/sqlc" mailboxpb "github.com/lightninglabs/darepo-client/mailbox/pb" "github.com/lightninglabs/darepo-client/metrics" "github.com/lightninglabs/darepo-client/oor" @@ -257,7 +258,7 @@ type Config struct { // top-level field (not under SwapWallet) because the subserver is // registered by build tag regardless of whether the operator supplied // a [swapwallet] config section. A nil value disables projection. - ActivityStore ActivityProjector `mapstructure:"-"` + ActivityStore ActivityStore `mapstructure:"-"` // MaxOperatorFeeSat caps the per-round operator fee the client // is willing to pay under the #270 seal-time fee handshake. @@ -643,15 +644,27 @@ type SwapBackend interface { ResumePending(ctx context.Context) } -// ActivityProjector persists wallet activity rows to the canonical activity -// log as their lifecycle advances. *db.ActivityPersistenceStore satisfies it; -// the walletdkrpc subserver's projector calls ProjectEntry from the existing -// emit sites and the startup backfill. The interface keeps the daemon-side -// store out of the swapwallet build-tag boundary and lets tests pass nil. -type ActivityProjector interface { +// ActivityStore is the walletdkrpc subserver's handle on the canonical +// activity log. *db.ActivityPersistenceStore satisfies it; the projector +// writes through ProjectEntry from the emit sites and the startup backfill, +// and the List read path pages current-state rows through ListEntries. The +// interface keeps the daemon-side store out of the swapwallet build-tag +// boundary and lets tests pass nil. +type ActivityStore interface { // ProjectEntry advances the activity row to the projected state and // records the transition, atomically. ProjectEntry(ctx context.Context, p db.ActivityProjection) error + + // ListEntries returns up to limit current-state rows newest-first, + // starting after the (cursorCreated, cursorID) keyset. A cursorCreated + // of 0 starts from the newest row. + ListEntries(ctx context.Context, cursorCreated int64, cursorID string, + limit int32) ([]sqlc.ActivityEntry, error) + + // CountByStatus returns the number of current-state rows in the given + // status. It backs the wallet status summary's full-feed pending count, + // which the paginated List path cannot report. + CountByStatus(ctx context.Context, status int64) (int64, error) } // SwapWalletConfig configures the optional walletdkrpc subserver. The struct diff --git a/db/activity_store.go b/db/activity_store.go index 1bdec905a..ea52edc1f 100644 --- a/db/activity_store.go +++ b/db/activity_store.go @@ -23,6 +23,9 @@ type ActivityStore interface { GetActivityEntry(ctx context.Context, canonicalID string) (sqlc.ActivityEntry, error) + CountActivityEntriesByStatus(ctx context.Context, + status int64) (int64, error) + ListActivityEntries(ctx context.Context, arg sqlc.ListActivityEntriesParams) ( []sqlc.ActivityEntry, @@ -196,6 +199,24 @@ func (s *ActivityPersistenceStore) GetEntry(ctx context.Context, return entry, err } +// CountByStatus returns the number of current-state rows in the given status. +// Unlike ListEntries it is not paginated, so it backs the wallet status +// summary's pending count with a true full-feed total. +func (s *ActivityPersistenceStore) CountByStatus(ctx context.Context, + status int64) (int64, error) { + + var count int64 + + err := s.db.ExecTx(ctx, ReadTxOption(), func(q ActivityStore) error { + var err error + count, err = q.CountActivityEntriesByStatus(ctx, status) + + return err + }) + + return count, err +} + // ListEntries returns up to limit current-state rows newest-first, starting // after the (cursorCreated, cursorID) keyset. A cursorCreated of 0 starts from // the newest row. The cursor is the immutable (created_at_unix, canonical_id) diff --git a/db/activity_store_test.go b/db/activity_store_test.go index 0ba772583..6349db1f9 100644 --- a/db/activity_store_test.go +++ b/db/activity_store_test.go @@ -76,6 +76,35 @@ func TestActivityStoreProjectInsertsEntryAndEvent(t *testing.T) { require.EqualValues(t, 1, events[0].Status) } +// TestActivityStoreCountByStatus verifies CountByStatus returns a full, +// unpaginated count of the rows in a given status — the primitive the wallet +// status summary's pending count relies on. +func TestActivityStoreCountByStatus(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newActivityStoreForTest(t) + + // Two pending rows (status 1) and one complete row (status 2). + complete := sampleProjection("c1") + complete.Status = 2 + require.NoError(t, store.ProjectEntry(ctx, sampleProjection("p1"))) + require.NoError(t, store.ProjectEntry(ctx, sampleProjection("p2"))) + require.NoError(t, store.ProjectEntry(ctx, complete)) + + pending, err := store.CountByStatus(ctx, 1) + require.NoError(t, err) + require.EqualValues(t, 2, pending) + + completed, err := store.CountByStatus(ctx, 2) + require.NoError(t, err) + require.EqualValues(t, 1, completed) + + failed, err := store.CountByStatus(ctx, 3) + require.NoError(t, err) + require.EqualValues(t, 0, failed) +} + // TestActivityStoreProjectSuppressesUnchanged verifies that re-projecting an // identical state appends no new event, so the backfill and the swap monitor's // replay do not accumulate duplicate transitions in the append-only log. diff --git a/db/sqlc/activity_log.sql.go b/db/sqlc/activity_log.sql.go index 0c0fc12cd..ba6a0b00b 100644 --- a/db/sqlc/activity_log.sql.go +++ b/db/sqlc/activity_log.sql.go @@ -39,6 +39,20 @@ func (q *Queries) AppendActivityEvent(ctx context.Context, arg AppendActivityEve return err } +const CountActivityEntriesByStatus = `-- name: CountActivityEntriesByStatus :one +SELECT COUNT(*) FROM activity_entries WHERE status = $1 +` + +// CountActivityEntriesByStatus returns the number of current-state rows in the +// given status. It backs the wallet status summary's pending count, which must +// reflect the whole feed rather than a single paginated page. +func (q *Queries) CountActivityEntriesByStatus(ctx context.Context, status int64) (int64, error) { + row := q.db.QueryRowContext(ctx, CountActivityEntriesByStatus, status) + var count int64 + err := row.Scan(&count) + return count, err +} + const GetActivityEntry = `-- name: GetActivityEntry :one SELECT canonical_id, kind, status, amount_sat, fee_sat, counterparty, note, phase, phase_label, failure_code, failure_reason, payment_hash, txid, confirmation_height, vtxo_outpoint, swap_session_id, ledger_txid, boarding_addr, request_json, created_at_unix, updated_at_unix FROM activity_entries WHERE canonical_id = $1 ` diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 97c2b6a25..074862bda 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -16,6 +16,10 @@ type Querier interface { CancelVHTLCRecoveryJob(ctx context.Context, arg CancelVHTLCRecoveryJobParams) (int64, error) ClearPendingIntentAnchorByOutpoint(ctx context.Context, arg ClearPendingIntentAnchorByOutpointParams) error CompleteVHTLCRecoveryJob(ctx context.Context, arg CompleteVHTLCRecoveryJobParams) (int64, error) + // CountActivityEntriesByStatus returns the number of current-state rows in the + // given status. It backs the wallet status summary's pending count, which must + // reflect the whole feed rather than a single paginated page. + CountActivityEntriesByStatus(ctx context.Context, status int64) (int64, error) CountBoardingIntentsByStatus(ctx context.Context, status string) (int64, error) CountClientLedgerEntries(ctx context.Context) (int64, error) CountUnresolvedBoardingSweepInputs(ctx context.Context, txid []byte) (int64, error) diff --git a/db/sqlc/queries/activity_log.sql b/db/sqlc/queries/activity_log.sql index 82baebae5..70231c501 100644 --- a/db/sqlc/queries/activity_log.sql +++ b/db/sqlc/queries/activity_log.sql @@ -56,6 +56,12 @@ INSERT INTO activity_events ( -- GetActivityEntry returns one entry by its canonical id. SELECT * FROM activity_entries WHERE canonical_id = $1; +-- name: CountActivityEntriesByStatus :one +-- CountActivityEntriesByStatus returns the number of current-state rows in the +-- given status. It backs the wallet status summary's pending count, which must +-- reflect the whole feed rather than a single paginated page. +SELECT COUNT(*) FROM activity_entries WHERE status = sqlc.arg(status); + -- name: ListActivityEntries :many -- ListActivityEntries returns entries newest-first, paged by the immutable -- (created_at_unix, canonical_id) cursor so a row that transitions in place diff --git a/rpc/walletdkrpc/wallet.pb.go b/rpc/walletdkrpc/wallet.pb.go index a1c6249f5..e417f8495 100644 --- a/rpc/walletdkrpc/wallet.pb.go +++ b/rpc/walletdkrpc/wallet.pb.go @@ -1607,8 +1607,16 @@ type ListRequest struct { Kinds []EntryKind `protobuf:"varint,3,rep,packed,name=kinds,proto3,enum=walletdkrpc.EntryKind" json:"kinds,omitempty"` // limit caps the response size. Zero means use the daemon default. Limit uint32 `protobuf:"varint,4,opt,name=limit,proto3" json:"limit,omitempty"` - // offset is the pagination offset within the chosen view. - Offset uint32 `protobuf:"varint,5,opt,name=offset,proto3" json:"offset,omitempty"` + // offset is the pagination offset within the chosen view. It applies + // to the VTXOS and ONCHAIN views; the ACTIVITY view paginates by the + // opaque cursor below instead and ignores offset. + Offset uint32 `protobuf:"varint,5,opt,name=offset,proto3" json:"offset,omitempty"` + // cursor is the opaque pagination token for the ACTIVITY view. Empty + // starts from the newest entry; otherwise it is the next_cursor + // returned by the previous ActivityList page. It is stable across + // concurrent inserts, so paging never skips or duplicates rows. + // Ignored for the VTXOS and ONCHAIN views. + Cursor string `protobuf:"bytes,6,opt,name=cursor,proto3" json:"cursor,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1678,6 +1686,13 @@ func (x *ListRequest) GetOffset() uint32 { return 0 } +func (x *ListRequest) GetCursor() string { + if x != nil { + return x.Cursor + } + return "" +} + type ListResponse struct { state protoimpl.MessageState `protogen:"open.v1"` // WIRE-BREAKING CHANGE: PR #440 had @@ -1791,9 +1806,15 @@ type ActivityList struct { state protoimpl.MessageState `protogen:"open.v1"` // entries are the unified, time-sorted wallet operations. Entries []*WalletEntry `protobuf:"bytes,1,rep,name=entries,proto3" json:"entries,omitempty"` - // total is the total number of entries matching the filter before - // limit and offset are applied. - Total uint32 `protobuf:"varint,2,opt,name=total,proto3" json:"total,omitempty"` + // total is the number of entries in this page. It is a page count, + // not a full-feed count: the feed is paged by an opaque cursor, so + // callers use has_more, not total, to decide whether to fetch again. + Total uint32 `protobuf:"varint,2,opt,name=total,proto3" json:"total,omitempty"` + // has_more reports whether more entries exist after this page. + HasMore bool `protobuf:"varint,3,opt,name=has_more,json=hasMore,proto3" json:"has_more,omitempty"` + // next_cursor is the opaque token to pass as ListRequest.cursor to + // fetch the next page. Empty when has_more is false. + NextCursor string `protobuf:"bytes,4,opt,name=next_cursor,json=nextCursor,proto3" json:"next_cursor,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1842,6 +1863,20 @@ func (x *ActivityList) GetTotal() uint32 { return 0 } +func (x *ActivityList) GetHasMore() bool { + if x != nil { + return x.HasMore + } + return false +} + +func (x *ActivityList) GetNextCursor() string { + if x != nil { + return x.NextCursor + } + return "" +} + type VTXOInventory struct { state protoimpl.MessageState `protogen:"open.v1"` // vtxos are the live spendable VTXOs in the wallet. Order is @@ -4664,21 +4699,25 @@ const file_wallet_proto_rawDesc = "" + "\foperation_id\x18\x01 \x01(\tR\voperationId\x12\x1d\n" + "\n" + "amount_sat\x18\x02 \x01(\x04R\tamountSat\x12!\n" + - "\fpayment_hash\x18\x03 \x01(\tR\vpaymentHash\"\xb7\x01\n" + + "\fpayment_hash\x18\x03 \x01(\tR\vpaymentHash\"\xcf\x01\n" + "\vListRequest\x12)\n" + "\x04view\x18\x01 \x01(\x0e2\x15.walletdkrpc.ListViewR\x04view\x12!\n" + "\fpending_only\x18\x02 \x01(\bR\vpendingOnly\x12,\n" + "\x05kinds\x18\x03 \x03(\x0e2\x16.walletdkrpc.EntryKindR\x05kinds\x12\x14\n" + "\x05limit\x18\x04 \x01(\rR\x05limit\x12\x16\n" + - "\x06offset\x18\x05 \x01(\rR\x06offset\"\xbc\x01\n" + + "\x06offset\x18\x05 \x01(\rR\x06offset\x12\x16\n" + + "\x06cursor\x18\x06 \x01(\tR\x06cursor\"\xbc\x01\n" + "\fListResponse\x127\n" + "\bactivity\x18\x01 \x01(\v2\x19.walletdkrpc.ActivityListH\x00R\bactivity\x122\n" + "\x05vtxos\x18\x02 \x01(\v2\x1a.walletdkrpc.VTXOInventoryH\x00R\x05vtxos\x127\n" + "\aonchain\x18\x03 \x01(\v2\x1b.walletdkrpc.OnchainHistoryH\x00R\aonchainB\x06\n" + - "\x04body\"X\n" + + "\x04body\"\x94\x01\n" + "\fActivityList\x122\n" + "\aentries\x18\x01 \x03(\v2\x18.walletdkrpc.WalletEntryR\aentries\x12\x14\n" + - "\x05total\x18\x02 \x01(\rR\x05total\"T\n" + + "\x05total\x18\x02 \x01(\rR\x05total\x12\x19\n" + + "\bhas_more\x18\x03 \x01(\bR\ahasMore\x12\x1f\n" + + "\vnext_cursor\x18\x04 \x01(\tR\n" + + "nextCursor\"T\n" + "\rVTXOInventory\x12-\n" + "\x05vtxos\x18\x01 \x03(\v2\x17.walletdkrpc.WalletVTXOR\x05vtxos\x12\x14\n" + "\x05total\x18\x02 \x01(\rR\x05total\"\xd4\x01\n" + diff --git a/rpc/walletdkrpc/wallet.proto b/rpc/walletdkrpc/wallet.proto index 1e332f2a6..793d95728 100644 --- a/rpc/walletdkrpc/wallet.proto +++ b/rpc/walletdkrpc/wallet.proto @@ -430,8 +430,17 @@ message ListRequest { // limit caps the response size. Zero means use the daemon default. uint32 limit = 4; - // offset is the pagination offset within the chosen view. + // offset is the pagination offset within the chosen view. It applies + // to the VTXOS and ONCHAIN views; the ACTIVITY view paginates by the + // opaque cursor below instead and ignores offset. uint32 offset = 5; + + // cursor is the opaque pagination token for the ACTIVITY view. Empty + // starts from the newest entry; otherwise it is the next_cursor + // returned by the previous ActivityList page. It is stable across + // concurrent inserts, so paging never skips or duplicates rows. + // Ignored for the VTXOS and ONCHAIN views. + string cursor = 6; } message ListResponse { @@ -456,9 +465,17 @@ message ActivityList { // entries are the unified, time-sorted wallet operations. repeated WalletEntry entries = 1; - // total is the total number of entries matching the filter before - // limit and offset are applied. + // total is the number of entries in this page. It is a page count, + // not a full-feed count: the feed is paged by an opaque cursor, so + // callers use has_more, not total, to decide whether to fetch again. uint32 total = 2; + + // has_more reports whether more entries exist after this page. + bool has_more = 3; + + // next_cursor is the opaque token to pass as ListRequest.cursor to + // fetch the next page. Empty when has_more is false. + string next_cursor = 4; } message VTXOInventory { diff --git a/sdk/walletdk/client.go b/sdk/walletdk/client.go index 267e8a716..261b6af45 100644 --- a/sdk/walletdk/client.go +++ b/sdk/walletdk/client.go @@ -419,6 +419,7 @@ func (c *Client) List(ctx context.Context, req ListRequest) (*ListResult, Kinds: kinds, Limit: req.Limit, Offset: req.Offset, + Cursor: req.Cursor, }) if err != nil { return nil, fmt.Errorf("list wallet entries: %w", err) diff --git a/sdk/walletdk/convert.go b/sdk/walletdk/convert.go index 3d78a8d4e..ded34b763 100644 --- a/sdk/walletdk/convert.go +++ b/sdk/walletdk/convert.go @@ -47,8 +47,10 @@ func listResultFromProto(view ListView, entries = append(entries, entryFromProto(e)) } out.Activity = &ActivityList{ - Entries: entries, - Total: activity.GetTotal(), + Entries: entries, + Total: activity.GetTotal(), + HasMore: activity.GetHasMore(), + NextCursor: activity.GetNextCursor(), } case ListViewVTXOs: diff --git a/sdk/walletdk/convert_test.go b/sdk/walletdk/convert_test.go index a3abea0a3..118576d6b 100644 --- a/sdk/walletdk/convert_test.go +++ b/sdk/walletdk/convert_test.go @@ -363,7 +363,9 @@ func TestListResultFromProtoActivity(t *testing.T) { Kind: recv, }, }, - Total: 42, + Total: 42, + HasMore: true, + NextCursor: "cursor-token", }, }, } @@ -373,6 +375,8 @@ func TestListResultFromProtoActivity(t *testing.T) { require.Nil(t, out.VTXOs) require.Nil(t, out.Onchain) require.Equal(t, uint32(42), out.Activity.Total) + require.True(t, out.Activity.HasMore) + require.Equal(t, "cursor-token", out.Activity.NextCursor) require.Len(t, out.Activity.Entries, 2) require.Equal(t, "hash1", out.Activity.Entries[0].ID) require.Equal(t, EntryKindSend, out.Activity.Entries[0].Kind) diff --git a/sdk/walletdk/types.go b/sdk/walletdk/types.go index a16b0455c..9a42c936e 100644 --- a/sdk/walletdk/types.go +++ b/sdk/walletdk/types.go @@ -286,8 +286,15 @@ type ListRequest struct { // Limit caps the page size; zero uses the daemon default. Limit uint32 - // Offset is the pagination offset within the chosen view. + // Offset is the pagination offset. It applies to the VTXOs and + // Onchain views; the Activity view paginates by Cursor and ignores + // Offset. Offset uint32 + + // Cursor is the opaque pagination token for the Activity view. Empty + // starts from the newest entry; otherwise pass the NextCursor returned + // by the previous ActivityList page. + Cursor string } // ListResult is a tagged union: exactly one of Activity, VTXOs, or @@ -312,7 +319,18 @@ type ListResult struct { // activity view. type ActivityList struct { Entries []Entry - Total uint32 + + // Total is the number of entries on this page, not a full-feed count: + // the feed is cursor-paged, so use HasMore to decide whether to fetch + // again. + Total uint32 + + // HasMore reports whether more entries exist after this page. + HasMore bool + + // NextCursor is the token to pass as ListRequest.Cursor to fetch the + // next page. Empty when HasMore is false. + NextCursor string } // VTXOInventory is the live VTXO inventory returned by the vtxos view. diff --git a/swapwallet/activity_dualwrite_test.go b/swapwallet/activity_dualwrite_test.go index 4af1bb3c6..05038f6dc 100644 --- a/swapwallet/activity_dualwrite_test.go +++ b/swapwallet/activity_dualwrite_test.go @@ -75,8 +75,13 @@ func TestBackfillMirrorsLegacyMerge(t *testing.T) { }, } - // Legacy merge result is the comparison oracle. - merged, err := h.listActivity(ctx, &walletdkrpc.ListRequest{Limit: 100}) + // The derive-on-read merge is the comparison oracle — it is what the + // backfill seeds the store from (listActivity now reads the store). + merged, err := h.deriveActivity( + ctx, &walletdkrpc.ListRequest{ + Limit: 100, + }, + ) require.NoError(t, err) wantStatus := make(map[string]int64) for _, e := range merged.GetEntries() { diff --git a/swapwallet/admin.go b/swapwallet/admin.go index 0188431dd..d8b0b0483 100644 --- a/swapwallet/admin.go +++ b/swapwallet/admin.go @@ -219,9 +219,15 @@ func (s *Service) forceUnroll(ctx context.Context, req *walletdkrpc.ExitRequest, return nil, status.Errorf(status.Code(err), "exit: %v", err) } + // Project the new EXIT row into the canonical store as well as fanning + // it out, mirroring the cooperative-leave and credit-pay paths. Without + // this the store-backed List would miss a user-initiated unilateral + // exit until the next startup backfill. Use a cancel-safe context so a + // client disconnect after the accepted Unroll cannot drop the store + // write. entry := unilateralExitEntryStub(req.GetOutpoint()) s.runtime.trackPendingEntryWithoutTimeout(entry) - s.runtime.emit(entry) + s.runtime.projectAndEmit(context.WithoutCancel(ctx), entry) return &walletdkrpc.ExitResponse{ Created: resp.GetCreated(), diff --git a/swapwallet/deps.go b/swapwallet/deps.go index 9eb99276e..cfb6e208b 100644 --- a/swapwallet/deps.go +++ b/swapwallet/deps.go @@ -195,7 +195,7 @@ type Deps struct { // writes each emitted WalletEntry through as state advances; the // startup backfill seeds it from the history collectors. Nil disables // projection. - ActivityStore darepod.ActivityProjector + ActivityStore darepod.ActivityStore } // resolveDeadline returns the effective wallet deadline, applying the diff --git a/swapwallet/doc.go b/swapwallet/doc.go index da68e78d4..8d13c4270 100644 --- a/swapwallet/doc.go +++ b/swapwallet/doc.go @@ -76,16 +76,33 @@ // injected via Deps.ActivityStore): activity_entries is the current-state // projection and activity_events is the append-only transition log. Writes // happen project-then-emit at the swap monitor loop, the cooperative-leave -// submit, and the deadline overlay, plus a one-time startup backfill from the -// collectors below. The read path is UNCHANGED for now: List and -// SubscribeWallet still derive from the live merge in history.go. A later -// change cuts List over to the store's keyset cursor and SubscribeWallet over -// to the activity_events event_seq cursor, then removes the merge. +// submit, the credit poll, the forced unilateral exit, and the deadline +// overlay, plus a one-time startup backfill from the collectors below. // -// Until that cutover, the canonical id stored for a cooperative-leave EXIT is -// still the consumed VTXO outpoint and a DEPOSIT is still keyed by txid:vout -// (or the synthetic boarding-unconfirmed row), so the same operation can hold -// different ids pending vs. confirmed — exactly the limitation above. Giving -// EXIT/DEPOSIT/on-chain-send a stable cross-lifecycle id needs the daemon-side -// hooks this V1 LIMITATIONS block describes and is tracked separately. +// The RPC read path now reads the store: List(ACTIVITY) pages activity_entries +// by the immutable (created_at_unix, canonical_id) keyset cursor, and +// SubscribeWallet's include_existing snapshot goes through the same +// store-backed List. deriveActivity (the live merge) is retained only for the +// store-less/test build and to seed the startup backfill. Because the store is +// ordered by the immutable created_at keyset, the feed is newest-by-creation, +// not newest-by-update. +// +// Consequences of the store-backed read that are tracked, not yet closed: +// - Producers without an ongoing projector — confirmed boarding DEPOSIT and +// daemon-side sweep/EXIT rows derived from ListTransactions — reach the +// store only via the startup backfill, so a newly-confirmed one appears in +// List after the next restart rather than immediately. +// - The synthetic boarding-unconfirmed DEPOSIT row is derive-path-only: it +// is ephemeral live state (recomputed from GetBalance, no durable id) and +// is deliberately NOT projected, so on a store build an unconfirmed +// boarding deposit surfaces via Balance rather than as an activity row +// until it confirms. +// +// The canonical id stored for a cooperative-leave EXIT is still the consumed +// VTXO outpoint and a DEPOSIT is still keyed by txid:vout, so the same +// operation can hold different ids pending vs. confirmed — exactly the +// limitation above. Giving EXIT/DEPOSIT/on-chain-send a stable cross-lifecycle +// id, and projecting the backfill-only producers on an ongoing basis, needs +// the daemon-side hooks this V1 LIMITATIONS block describes and is tracked +// separately. package swapwallet diff --git a/swapwallet/history.go b/swapwallet/history.go index bb7ca840b..e10b2fa60 100644 --- a/swapwallet/history.go +++ b/swapwallet/history.go @@ -4,9 +4,12 @@ package swapwallet import ( "context" + "encoding/base64" "encoding/hex" + "errors" "fmt" "sort" + "strconv" "strings" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -93,14 +96,244 @@ func (h *history) List(ctx context.Context, req *walletdkrpc.ListRequest) ( } } -// listActivity returns the merged WalletEntry stream — the v1 unified -// history. The page size is capped at the daemon-level maximum so a -// malformed request cannot fan out unbounded work; sources are queried -// with the request's own limit so per-source pagination remains the -// per-source contract. +// errInvalidActivityCursor is returned when the ACTIVITY cursor token cannot be +// decoded into a keyset position. +var errInvalidActivityCursor = errors.New("invalid activity cursor") + +// syntheticBoardingUnconfirmedID is the id of the derive-path-only row that +// represents an unconfirmed boarding deposit. It is recomputed live from +// GetBalance and has no durable identity, so it is deliberately kept out of the +// canonical store (see the projector's id guard): a delete-free upsert store +// could never clear it once the deposit confirms under its real txid:vout id. +const syntheticBoardingUnconfirmedID = "boarding-unconfirmed" + +// activityScanBudgetFactor bounds how many store rows a single filtered +// ACTIVITY page may scan (and protojson-decode), as a multiple of the page +// limit. Filters (pending_only/kinds) are applied in Go after decode, so a +// highly selective filter over a large table could otherwise scan the whole +// table for one page. When the budget is hit before the page fills, the call +// returns a short page plus a cursor so the caller resumes — bounding the work +// of any one request at the cost of an occasional extra round-trip. +const activityScanBudgetFactor = 8 + +// listActivity returns an ACTIVITY page read from the canonical activity store, +// newest-first and paged by the opaque cursor. Because the store orders by the +// immutable (created_at_unix, canonical_id) keyset, paging is stable across +// concurrent inserts: it neither skips nor duplicates rows. When no store is +// wired (tests without a database) it falls back to the derive-on-read merge. func (h *history) listActivity(ctx context.Context, req *walletdkrpc.ListRequest) (*walletdkrpc.ActivityList, error) { + if h.deps.ActivityStore == nil { + return h.deriveActivity(ctx, req) + } + + limit := h.deps.resolveListLimit(req.GetLimit()) + kindFilter, err := buildKindFilter(req.GetKinds()) + if err != nil { + return nil, err + } + + cursorCreated, cursorID, err := decodeActivityCursor(req.GetCursor()) + if err != nil { + return nil, err + } + + pendingOnly := req.GetPendingOnly() + + // Scan the keyset, applying filters in Go, until limit+1 rows match so + // has_more can be computed with the standard extra-row trick even when + // filters skip store rows. The keyset advances by the last SCANNED row; + // next_cursor points at the last RETURNED row so the next page resumes + // exactly after it. + matched := make([]*walletdkrpc.WalletEntry, 0, limit+1) + lastCreated, lastID := cursorCreated, cursorID + scanBudget := int(limit) * activityScanBudgetFactor + scanned := 0 + budgetExhausted := false + for uint32(len(matched)) <= limit { + batch, err := h.deps.ActivityStore.ListEntries( + ctx, lastCreated, lastID, int32(limit)+1, + ) + if err != nil { + return nil, fmt.Errorf("list activity entries: %w", err) + } + if len(batch) == 0 { + break + } + + for _, row := range batch { + lastCreated, lastID = row.CreatedAtUnix, row.CanonicalID + scanned++ + + entry, err := rowToWalletEntry(row) + if err != nil { + return nil, fmt.Errorf("decode activity row "+ + "%q: %w", row.CanonicalID, err) + } + if !matchesActivityFilter( + entry, pendingOnly, kindFilter, + ) { + + continue + } + + matched = append(matched, entry) + if uint32(len(matched)) > limit { + break + } + } + + // A short page means the store has no more rows to scan. + if uint32(len(batch)) < uint32(limit)+1 { + break + } + + // Stop once the scan budget is spent without filling the page. + // The last scanned row becomes the resume cursor so the caller + // continues rather than the server scanning the rest in one + // call. + if uint32(len(matched)) <= limit && scanned >= scanBudget { + budgetExhausted = true + + break + } + } + + hasMore := uint32(len(matched)) > limit + if hasMore { + matched = matched[:limit] + } + + var nextCursor string + switch { + case hasMore: + last := matched[len(matched)-1] + nextCursor = encodeActivityCursor( + last.GetCreatedAtUnix(), last.GetId(), + ) + + case budgetExhausted: + // The page did not fill but the store is not drained: resume + // strictly after the last scanned row. + hasMore = true + nextCursor = encodeActivityCursor(lastCreated, lastID) + } + + return &walletdkrpc.ActivityList{ + Entries: matched, + Total: uint32(len(matched)), + HasMore: hasMore, + NextCursor: nextCursor, + }, nil +} + +// countPending returns the total number of in-flight (PENDING) activity +// entries. When the canonical store is wired it counts rows directly, so the +// result is a true full-feed count instead of the single-page total the +// paginated listActivity path reports. Without a store it derives the merged +// pending set and counts that, matching the deadline-overlay semantics of the +// derive path. +func (h *history) countPending(ctx context.Context) (uint32, error) { + if h.deps.ActivityStore == nil { + list, err := h.deriveActivity(ctx, &walletdkrpc.ListRequest{ + View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, + PendingOnly: true, + Limit: h.deps.resolveMaxListLimit(), + }) + if err != nil { + return 0, err + } + + return list.GetTotal(), nil + } + + count, err := h.deps.ActivityStore.CountByStatus( + ctx, int64(walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING), + ) + if err != nil { + return 0, err + } + if count < 0 { + count = 0 + } + + return uint32(count), nil +} + +// matchesActivityFilter reports whether an entry passes the pending_only and +// kind filters. It is the single-entry form of filterEntries, applied per row +// during the store keyset scan. +func matchesActivityFilter(e *walletdkrpc.WalletEntry, pendingOnly bool, + kindFilter map[walletdkrpc.EntryKind]struct{}) bool { + + if pendingOnly && + e.GetStatus() != walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING { + return false + } + if len(kindFilter) > 0 { + if _, ok := kindFilter[e.GetKind()]; !ok { + return false + } + } + + return true +} + +// encodeActivityCursor encodes the immutable keyset position +// (created_at_unix, canonical_id) as an opaque base64 token. +func encodeActivityCursor(created int64, id string) string { + raw := strconv.FormatInt(created, 10) + ":" + id + + return base64.RawURLEncoding.EncodeToString([]byte(raw)) +} + +// decodeActivityCursor decodes a cursor token back into its keyset position. An +// empty cursor decodes to the newest-first start position (0, ""). +func decodeActivityCursor(cursor string) (int64, string, error) { + if cursor == "" { + return 0, "", nil + } + + raw, err := base64.RawURLEncoding.DecodeString(cursor) + if err != nil { + return 0, "", fmt.Errorf("%w: %v", errInvalidActivityCursor, + err) + } + + created, id, ok := strings.Cut(string(raw), ":") + if !ok { + return 0, "", errInvalidActivityCursor + } + + createdUnix, err := strconv.ParseInt(created, 10, 64) + if err != nil { + return 0, "", fmt.Errorf("%w: %v", errInvalidActivityCursor, + err) + } + + // A real row always has a positive created_at_unix (ProjectEntry + // substitutes the clock when a projection omits it), and the + // empty-cursor "start from newest" sentinel is handled above. So a + // non-positive decoded timestamp is a forged or corrupt token: reject + // it loudly rather than let created_at_unix == 0 collide with the + // return-all sentinel in the keyset query and silently restart paging + // from the newest row. + if createdUnix <= 0 { + return 0, "", errInvalidActivityCursor + } + + return createdUnix, id, nil +} + +// deriveActivity returns the merged WalletEntry stream by re-joining the live +// sources on read. It is the pre-canonical-log path, retained only to seed the +// canonical store during the startup backfill; the RPC read path +// (listActivity) reads the store instead. The page size is capped at the +// daemon-level maximum so a malformed request cannot fan out unbounded work. +func (h *history) deriveActivity(ctx context.Context, + req *walletdkrpc.ListRequest) (*walletdkrpc.ActivityList, error) { + limit := h.deps.resolveListLimit(req.GetLimit()) kindFilter, err := buildKindFilter(req.GetKinds()) if err != nil { @@ -352,7 +585,7 @@ func (h *history) collectPendingBoardingEntries(ctx context.Context) ( return []*walletdkrpc.WalletEntry{ { - Id: "boarding-unconfirmed", + Id: syntheticBoardingUnconfirmedID, Kind: walletdkrpc.EntryKind_ENTRY_KIND_DEPOSIT, Status: walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, AmountSat: resp.GetBoardingUnconfirmedSat(), diff --git a/swapwallet/list_store_test.go b/swapwallet/list_store_test.go new file mode 100644 index 000000000..64363c389 --- /dev/null +++ b/swapwallet/list_store_test.go @@ -0,0 +1,319 @@ +//go:build walletdkrpc && swapruntime + +package swapwallet + +import ( + "context" + "fmt" + "testing" + + "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/darepo-client/db" + "github.com/lightninglabs/darepo-client/rpc/walletdkrpc" + "github.com/lightningnetwork/lnd/clock" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +// newStoreListFixture wires a history reader over a real in-memory activity +// store so the store-backed List read path can be exercised end to end. +func newStoreListFixture(t *testing.T) (*history, + *db.ActivityPersistenceStore) { + + t.Helper() + + testDB := db.NewTestDB(t) + store := db.NewStore( + testDB.DB, testDB.Queries, testDB.Backend(), btclog.Disabled, + ).NewActivityStore(clock.NewDefaultClock()) + + deps := &Deps{ActivityStore: store} + runtime := newRuntime(t.Context(), deps) + t.Cleanup(runtime.stop) + + return newHistory(deps, runtime), store +} + +// seedActivity projects one entry into the store via the production mapping. +func seedActivity(t *testing.T, store *db.ActivityPersistenceStore, id string, + kind walletdkrpc.EntryKind, status walletdkrpc.EntryStatus, + created int64) { + + t.Helper() + + proj, err := entryToProjection(&walletdkrpc.WalletEntry{ + Id: id, + Kind: kind, + Status: status, + AmountSat: 1000, + CreatedAtUnix: created, + UpdatedAtUnix: created, + }) + require.NoError(t, err) + require.NoError(t, store.ProjectEntry(context.Background(), proj)) +} + +func activityIDs(list *walletdkrpc.ActivityList) []string { + out := make([]string, 0, len(list.GetEntries())) + for _, e := range list.GetEntries() { + out = append(out, e.GetId()) + } + + return out +} + +// TestListActivityReadsStore verifies List pages the store newest-first and +// resumes via next_cursor with a correct has_more. +func TestListActivityReadsStore(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + seedActivity( + t, store, "a", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 100, + ) + seedActivity( + t, store, "b", walletdkrpc.EntryKind_ENTRY_KIND_RECV, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 200, + ) + seedActivity( + t, store, "c", walletdkrpc.EntryKind_ENTRY_KIND_DEPOSIT, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 300, + ) + + page1, err := h.listActivity(ctx, &walletdkrpc.ListRequest{Limit: 2}) + require.NoError(t, err) + require.Equal(t, []string{"c", "b"}, activityIDs(page1)) + require.True(t, page1.GetHasMore()) + require.NotEmpty(t, page1.GetNextCursor()) + + page2, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 2, + Cursor: page1.GetNextCursor(), + }) + require.NoError(t, err) + require.Equal(t, []string{"a"}, activityIDs(page2)) + require.False(t, page2.GetHasMore()) + require.Empty(t, page2.GetNextCursor()) +} + +// TestCountPendingReflectsFullFeed verifies countPending returns the full +// number of pending rows rather than the single-page total the paginated read +// path reports. This is the store-backed count behind the wallet status +// summary's pending count. +func TestCountPendingReflectsFullFeed(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + // Three pending rows plus one terminal row. + seedActivity( + t, store, "p1", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 100, + ) + seedActivity( + t, store, "p2", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 200, + ) + seedActivity( + t, store, "p3", walletdkrpc.EntryKind_ENTRY_KIND_RECV, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 300, + ) + seedActivity( + t, store, "done", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 400, + ) + + // A single-page pending read caps its total at the page size, so it + // cannot stand in for the pending count. + page, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 1, + PendingOnly: true, + }) + require.NoError(t, err) + require.EqualValues(t, 1, page.GetTotal()) + require.True(t, page.GetHasMore()) + + // countPending reports every pending row regardless of page size. + count, err := h.countPending(ctx) + require.NoError(t, err) + require.EqualValues(t, 3, count) +} + +// TestListActivityStablePaginationUnderInsert verifies the #781 acceptance +// criterion: a row inserted between page fetches never causes an existing row +// to be skipped or duplicated, because the cursor is an immutable keyset. +func TestListActivityStablePaginationUnderInsert(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + seedActivity( + t, store, "a", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 100, + ) + seedActivity( + t, store, "b", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 200, + ) + seedActivity( + t, store, "c", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 300, + ) + + page1, err := h.listActivity(ctx, &walletdkrpc.ListRequest{Limit: 2}) + require.NoError(t, err) + require.Equal(t, []string{"c", "b"}, activityIDs(page1)) + + // A new op lands between page fetches, newer than the page-1 cursor. + seedActivity( + t, store, "d", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 250, + ) + + page2, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 2, + Cursor: page1.GetNextCursor(), + }) + require.NoError(t, err) + + // Page 2 continues strictly older than the cursor: "a" is returned + // once, "b"/"c" are not duplicated, and the newer "d" is simply above + // this pagination pass (a fresh read would surface it at the top). + require.Equal(t, []string{"a"}, activityIDs(page2)) +} + +// TestListActivityFilters verifies pending_only and kind filters apply over the +// store keyset scan. +func TestListActivityFilters(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + seedActivity( + t, store, "send", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 100, + ) + seedActivity( + t, store, "recv", walletdkrpc.EntryKind_ENTRY_KIND_RECV, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 200, + ) + seedActivity( + t, store, "exit", walletdkrpc.EntryKind_ENTRY_KIND_EXIT, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 300, + ) + + pending, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 10, + PendingOnly: true, + }) + require.NoError(t, err) + require.Equal(t, []string{"exit", "send"}, activityIDs(pending)) + + recvOnly, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 10, + Kinds: []walletdkrpc.EntryKind{ + walletdkrpc.EntryKind_ENTRY_KIND_RECV, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"recv"}, activityIDs(recvOnly)) +} + +// TestListActivityRejectsBadCursor verifies a malformed cursor is a clean +// error. +func TestListActivityRejectsBadCursor(t *testing.T) { + t.Parallel() + + h, _ := newStoreListFixture(t) + + _, err := h.listActivity(context.Background(), &walletdkrpc.ListRequest{ + Cursor: "!!!not-base64!!!", + }) + require.ErrorIs(t, err, errInvalidActivityCursor) +} + +// TestListActivityRejectsNonPositiveCursor verifies a cursor whose timestamp is +// zero or negative is rejected rather than silently colliding with the +// return-all sentinel and restarting paging from the newest row. +func TestListActivityRejectsNonPositiveCursor(t *testing.T) { + t.Parallel() + + h, _ := newStoreListFixture(t) + + for _, created := range []int64{0, -1} { + cursor := encodeActivityCursor(created, "x") + _, err := h.listActivity( + context.Background(), &walletdkrpc.ListRequest{ + Cursor: cursor, + }, + ) + require.ErrorIs(t, err, errInvalidActivityCursor) + } +} + +// TestListActivityBoundsFilteredScan verifies a selective filter over a large +// non-matching table does not scan the whole table in one request: the call +// stops at the scan budget and returns an empty page plus a cursor to resume, +// instead of decoding every row (the H-2 amplification cliff). +func TestListActivityBoundsFilteredScan(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + // Seed far more terminal rows than one page's scan budget + // (limit * activityScanBudgetFactor), none of which match --pending. + const rows = 60 + for i := 0; i < rows; i++ { + seedActivity( + t, store, fmt.Sprintf("c%02d", i), + walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, + int64(100+i), + ) + } + + page, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 2, + PendingOnly: true, + }) + require.NoError(t, err) + + // The budget-bounded scan returns no matches but signals more work with + // a resume cursor, rather than draining the table (which would report + // has_more=false). + require.Empty(t, page.GetEntries()) + require.True(t, page.GetHasMore()) + require.NotEmpty(t, page.GetNextCursor()) +} + +// TestRowToWalletEntryRoundTrip verifies a WalletEntry survives the +// project → store → row → rowToWalletEntry round trip unchanged. +func TestRowToWalletEntryRoundTrip(t *testing.T) { + t.Parallel() + + ctx := context.Background() + _, store := newStoreListFixture(t) + + entry := sampleWalletEntry() + + proj, err := entryToProjection(entry) + require.NoError(t, err) + require.NoError(t, store.ProjectEntry(ctx, proj)) + + row, err := store.GetEntry(ctx, entry.GetId()) + require.NoError(t, err) + + got, err := rowToWalletEntry(row) + require.NoError(t, err) + require.True( + t, proto.Equal(entry, got), + "reconstructed entry must equal the original", + ) +} diff --git a/swapwallet/projector.go b/swapwallet/projector.go index 66b3ef88f..7d7d856fb 100644 --- a/swapwallet/projector.go +++ b/swapwallet/projector.go @@ -9,6 +9,7 @@ import ( "log/slog" "github.com/lightninglabs/darepo-client/db" + "github.com/lightninglabs/darepo-client/db/sqlc" "github.com/lightninglabs/darepo-client/rpc/walletdkrpc" "google.golang.org/protobuf/encoding/protojson" ) @@ -26,8 +27,14 @@ func (r *Runtime) project(ctx context.Context, entry *walletdkrpc.WalletEntry) { } // A row with no canonical id cannot be keyed; skip it, matching the - // id-guard the pending tracker already applies. - if entry.GetId() == "" { + // id-guard the pending tracker already applies. The synthetic + // boarding-unconfirmed row is skipped for the same reason: it is + // ephemeral live state recomputed from GetBalance with no durable + // identity, so persisting it into a delete-free store would strand a + // PENDING row that never clears once the deposit confirms under its + // real txid:vout id. + if entry.GetId() == "" || + entry.GetId() == syntheticBoardingUnconfirmedID { return } @@ -79,7 +86,7 @@ func (r *Runtime) backfillActivity(ctx context.Context) { projected int ) for { - list, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + list, err := h.deriveActivity(ctx, &walletdkrpc.ListRequest{ Limit: limit, Offset: offset, }) @@ -195,3 +202,70 @@ func hexBytesOrNil(s string) []byte { return b } + +// rowToWalletEntry reconstructs a WalletEntry from a stored current-state row — +// the inverse of entryToProjection, used by the store-backed List read path. +// Every WalletEntry field has a backing column: BLOB handles are hex-encoded +// back, the confirmation height is widened, and the request oneof is decoded +// from its protojson form. The reconstruction is not byte-for-byte identical in +// one case: Progress is always materialized, so an entry projected with a nil +// Progress round-trips to a non-nil empty Progress (no current producer emits a +// nil-Progress row, so this is latent). +// +// request_json is decoded with DiscardUnknown so a row written by a newer +// daemon (carrying a WalletEntryRequest field this binary does not know) still +// decodes instead of failing the whole page. A genuinely malformed request +// still errors — a corrupt row is inconsistent state that should surface, not +// be silently skipped. +func rowToWalletEntry(row sqlc.ActivityEntry) (*walletdkrpc.WalletEntry, + error) { + + var request *walletdkrpc.WalletEntryRequest + if row.RequestJson != "" { + request = &walletdkrpc.WalletEntryRequest{} + opts := protojson.UnmarshalOptions{DiscardUnknown: true} + if err := opts.Unmarshal( + []byte(row.RequestJson), request, + ); err != nil { + return nil, fmt.Errorf("unmarshal request: %w", err) + } + } + + var confHeight int32 + if row.ConfirmationHeight.Valid { + confHeight = int32(row.ConfirmationHeight.Int64) + } + + entry := &walletdkrpc.WalletEntry{ + Id: row.CanonicalID, + Kind: walletdkrpc.EntryKind(row.Kind), + Status: walletdkrpc.EntryStatus(row.Status), + AmountSat: row.AmountSat, + FeeSat: row.FeeSat, + Counterparty: row.Counterparty, + Note: row.Note, + FailureReason: row.FailureReason, + Request: request, + Progress: &walletdkrpc.WalletEntryProgress{ + Phase: walletdkrpc.WalletEntryPhase( + row.Phase, + ), + PhaseLabel: row.PhaseLabel, + PaymentHash: hex.EncodeToString(row.PaymentHash), + Txid: hex.EncodeToString(row.Txid), + ConfirmationHeight: confHeight, + VtxoOutpoint: row.VtxoOutpoint, + }, + CreatedAtUnix: row.CreatedAtUnix, + UpdatedAtUnix: row.UpdatedAtUnix, + } + + // failure_code is presence-tracked on the wire: absent means "no + // failure", so only set it for a non-zero stored code. + if row.FailureCode != 0 { + code := walletdkrpc.EntryFailureCode(row.FailureCode) + entry.FailureCode = &code + } + + return entry, nil +} diff --git a/swapwallet/projector_test.go b/swapwallet/projector_test.go index 65aa5a847..d10037d22 100644 --- a/swapwallet/projector_test.go +++ b/swapwallet/projector_test.go @@ -9,6 +9,7 @@ import ( "time" "github.com/lightninglabs/darepo-client/db" + "github.com/lightninglabs/darepo-client/db/sqlc" "github.com/lightninglabs/darepo-client/rpc/walletdkrpc" "github.com/stretchr/testify/require" "google.golang.org/protobuf/encoding/protojson" @@ -54,6 +55,23 @@ func (f *fakeActivityProjector) count() int { return len(f.projected) } +// ListEntries satisfies darepod.ActivityStore. This fake exercises only the +// write path; the store-backed read path is tested against a real DB store, so +// this returns no rows. +func (f *fakeActivityProjector) ListEntries(_ context.Context, _ int64, + _ string, _ int32) ([]sqlc.ActivityEntry, error) { + + return nil, nil +} + +// CountByStatus satisfies darepod.ActivityStore. The count path is tested +// against a real DB store, so this fake reports nothing. +func (f *fakeActivityProjector) CountByStatus(_ context.Context, _ int64) ( + int64, error) { + + return 0, nil +} + // ids returns the set of canonical ids the fake has been asked to project. func (f *fakeActivityProjector) ids() map[string]bool { f.mu.Lock() @@ -250,3 +268,49 @@ func TestProjectAndEmitStoreErrorStillEmits(t *testing.T) { runtime.projectAndEmit(context.Background(), sampleWalletEntry()) require.Equal(t, "payment-hash", recvEntry(t, ch).GetId()) } + +// TestProjectAndEmitSkipsEphemeralBoardingRow verifies the synthetic +// boarding-unconfirmed row is emitted to subscribers but never persisted: it +// is ephemeral live state with no durable id, and a delete-free store could +// never clear it once the deposit confirms under its real txid:vout id. +func TestProjectAndEmitSkipsEphemeralBoardingRow(t *testing.T) { + t.Parallel() + + store := &fakeActivityProjector{} + runtime, ch := newProjectorRuntime(t, store) + + entry := sampleWalletEntry() + entry.Id = syntheticBoardingUnconfirmedID + runtime.projectAndEmit(context.Background(), entry) + + require.Equal( + t, syntheticBoardingUnconfirmedID, recvEntry(t, ch).GetId(), + ) + require.Equal(t, 0, store.count(), "ephemeral row must not be stored") +} + +// TestRowToWalletEntryDiscardsUnknownRequestFields verifies a stored request +// carrying a field this binary does not know (schema drift from a newer +// daemon) still decodes, while genuinely malformed JSON still fails loudly — +// a corrupt row is inconsistent state that must surface, not be skipped. +func TestRowToWalletEntryDiscardsUnknownRequestFields(t *testing.T) { + t.Parallel() + + forward := sqlc.ActivityEntry{ + CanonicalID: "a", + RequestJson: `{"lightningInvoice":{"invoice":"lnbc1"},` + + `"futureField":42}`, + } + got, err := rowToWalletEntry(forward) + require.NoError(t, err) + require.Equal( + t, "lnbc1", got.GetRequest().GetLightningInvoice().GetInvoice(), + ) + + corrupt := sqlc.ActivityEntry{ + CanonicalID: "b", + RequestJson: `{not valid json`, + } + _, err = rowToWalletEntry(corrupt) + require.Error(t, err, "a corrupt request row must fail loudly") +} diff --git a/swapwallet/service.go b/swapwallet/service.go index 852476e32..8128484ec 100644 --- a/swapwallet/service.go +++ b/swapwallet/service.go @@ -418,17 +418,14 @@ func (s *Service) fetchBalance(ctx context.Context) ( return resp, nil } -// countPendingEntries asks the history merger for a pending-only page and -// returns the size as the wallet-level pending count. +// countPendingEntries returns the wallet-level count of in-flight entries. It +// delegates to the history merger's full-feed pending count rather than a +// List page total, which is capped at one page under cursor pagination. func (s *Service) countPendingEntries(ctx context.Context) (uint32, error) { - resp, err := s.history.List(ctx, &walletdkrpc.ListRequest{ - View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, - PendingOnly: true, - Limit: s.deps.resolveMaxListLimit(), - }) + count, err := s.history.countPending(ctx) if err != nil { return 0, fmt.Errorf("count pending: %w", err) } - return resp.GetActivity().GetTotal(), nil + return count, nil }