multi: read List(ACTIVITY) from the canonical store + pagination

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 <next_cursor>\n" +
 			"  darepocli activity inspect <id>",
 		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)

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 {

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())
 }

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 {

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

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)

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.

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