chanevents: implement GetChannelEvents rpc

chanevents/monitor.go

@@ -333,7 +333,7 @@ func (m *Monitor) addChannel(ctx context.Context, pubKeyBytes route.Vertex,
 
 	// Check if the channel already exists.
 	channel, err := m.store.GetChannel(ctx, channelPoint)
-	if err != nil && !errors.Is(err, errUnknownChannel) {
+	if err != nil && !errors.Is(err, ErrUnknownChannel) {
 		return fmt.Errorf("error getting channel %s: %w",
 			channelPoint, err)
 	}

chanevents/store.go

@@ -15,8 +15,11 @@ import (
 )
 
 var (
-	errUnknownPeer    = errors.New("unknown peer")
-	errUnknownChannel = errors.New("unknown channel")
+	errUnknownPeer = errors.New("unknown peer")
+
+	// ErrUnknownChannel signals that the requested channel is not
+	// present in the store.
+	ErrUnknownChannel = errors.New("unknown channel")
 )
 
 // Queries is a subset of the sqlc.Queries interface that can be used to
@@ -167,7 +170,7 @@ func (s *Store) GetChannel(ctx context.Context, channelPoint string) (*Channel,
 	dbChannel, err := s.db.GetChannelByChanPoint(ctx, channelPoint)
 	if err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
-			return nil, errUnknownChannel
+			return nil, ErrUnknownChannel
 		}
 
 		return nil, fmt.Errorf("failed to get channel: %w", err)
@@ -185,6 +188,8 @@ func (s *Store) GetChannel(ctx context.Context, channelPoint string) (*Channel,
 func (s *Store) AddChannelEvent(ctx context.Context,
 	event *ChannelEvent) error {
 
+	log.Tracef("Adding channel event: %+v", event)
+
 	var localBalance sql.NullInt64
 	event.LocalBalance.WhenSome(
 		func(b btcutil.Amount) {
@@ -223,18 +228,21 @@ func (s *Store) AddChannelEvent(ctx context.Context,
 	return nil
 }
 
-// GetChannelEvents retrieves all events for a channel within a given time
-// range.
-// TODO: Add pagination support (LIMIT/OFFSET) to prevent OOM on high-traffic
-// channels.
-func (s *Store) GetChannelEvents(ctx context.Context, channelID int64,
-	startTime, endTime time.Time) ([]*ChannelEvent, error) {
+// GetChannelEvents returns up to limit events for a channel where
+// id > afterID AND startTime <= timestamp < endTime, ordered by id ASC.
+// Pass afterID = 0 on the first call; for subsequent calls pass the
+// previous page's last event id. The (startTime, endTime) bounds are
+// independent filters and do not need to advance between pages.
+func (s *Store) GetChannelEvents(ctx context.Context, channelID, afterID int64,
+	startTime, endTime time.Time, limit int32) ([]*ChannelEvent, error) {
 
 	dbEvents, err := s.db.GetChannelEvents(
 		ctx, sqlc.GetChannelEventsParams{
 			ChannelID:   channelID,
+			ID:          afterID,
 			Timestamp:   startTime.UTC(),
 			Timestamp_2: endTime.UTC(),
+			Limit:       limit,
 		},
 	)
 	if err != nil {

chanevents/store_test.go

@@ -83,7 +83,7 @@ func TestStore(t *testing.T) {
 
 	// Get a non-existent channel and assert an error is returned.
 	dbChannel, err := store.GetChannel(ctx, "non-existent-chan-point")
-	require.ErrorIs(t, err, errUnknownChannel)
+	require.ErrorIs(t, err, ErrUnknownChannel)
 	require.Nil(t, dbChannel)
 
 	// Get the channel and assert it is the same.
@@ -125,7 +125,7 @@ func TestStore(t *testing.T) {
 
 	// Get the channel events and assert they are correct.
 	events, err := store.GetChannelEvents(
-		ctx, channelID, time.Unix(0, 0), time.Unix(3, 0),
+		ctx, channelID, 0, time.Unix(0, 0), time.Unix(3, 0), 100,
 	)
 	require.NoError(t, err)
 	require.Len(t, events, 2)
@@ -148,7 +148,7 @@ func TestStore(t *testing.T) {
 	require.NoError(t, err)
 
 	events, err = store.GetChannelEvents(
-		ctx, channelID, time.Unix(0, 0), time.Unix(4, 0),
+		ctx, channelID, 0, time.Unix(0, 0), time.Unix(4, 0), 100,
 	)
 	require.NoError(t, err)
 	require.Len(t, events, 3)
@@ -157,3 +157,50 @@ func TestStore(t *testing.T) {
 		t, syncEvent, testTime.Add(2*time.Second), events[2],
 	)
 }
+
+// TestPagination verifies that the keyset cursor advances correctly across
+// events sharing one second-resolution timestamp.
+func TestPagination(t *testing.T) {
+	t.Parallel()
+
+	clock := clock.NewTestClock(testTime)
+	store := NewTestDB(t, clock)
+	ctx := context.Background()
+
+	peerID, err := store.AddPeer(ctx, testPubKey)
+	require.NoError(t, err)
+
+	channelID, err := store.AddChannel(
+		ctx, testChanPoint1, testShortChanID1, peerID,
+	)
+	require.NoError(t, err)
+
+	sameTime := testTime.Add(10 * time.Second)
+	for i := 0; i < 5; i++ {
+		err = store.AddChannelEvent(ctx, &ChannelEvent{
+			ChannelID:    channelID,
+			EventType:    EventTypeUpdate,
+			Timestamp:    sameTime,
+			LocalBalance: fn.Some(btcutil.Amount(i)),
+		})
+		require.NoError(t, err)
+	}
+
+	endTime := sameTime.Add(time.Hour)
+
+	page1, err := store.GetChannelEvents(
+		ctx, channelID, 0, time.Unix(0, 0), endTime, 3,
+	)
+	require.NoError(t, err)
+	require.Len(t, page1, 3)
+
+	page2, err := store.GetChannelEvents(
+		ctx, channelID, page1[len(page1)-1].ID,
+		time.Unix(0, 0), endTime, 3,
+	)
+	require.NoError(t, err)
+	require.Len(t, page2, 2)
+
+	require.Equal(t, btcutil.Amount(3), page2[0].LocalBalance.UnwrapOr(0))
+	require.Equal(t, btcutil.Amount(4), page2[1].LocalBalance.UnwrapOr(0))
+}

cmd/frcli/chan_events.go

@@ -0,0 +1,97 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/lightninglabs/faraday/frdrpc"
+	"github.com/urfave/cli"
+)
+
+var chanEventsCommand = cli.Command{
+	Name:     "chanevents",
+	Category: "reporting",
+	Usage:    "Get a report of channel events.",
+	Description: `
+	Get a report for a channel which provides a detailed account of its
+	lifecycle events. The server caps each response; if has_more is true,
+	fetch the next page by re-running with --last_id set to the previous
+	response's last_id. Stop when has_more is false. --start_time and
+	--end_time are independent filters and do not need to advance between
+	paginated calls.`,
+	ArgsUsage: "funding_txid [output_index]",
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "funding_txid",
+			Usage: "the txid of the channel's funding transaction",
+		},
+		cli.IntFlag{
+			Name: "output_index",
+			Usage: "the output index for the funding output of " +
+				"the funding transaction",
+		},
+		cli.Int64Flag{
+			Name:  "start_time",
+			Usage: "start time of the query range as a unix timestamp",
+		},
+		cli.Int64Flag{
+			Name: "end_time",
+			Usage: "end time of the query range as a unix " +
+				"timestamp; zero defaults to the server's " +
+				"current time",
+		},
+		cli.UintFlag{
+			Name: "max_events",
+			Usage: "maximum number of events to return; zero " +
+				"uses the server default (capped server-side)",
+		},
+		cli.Int64Flag{
+			Name: "last_id",
+			Usage: "pagination cursor; pass the previous " +
+				"response's last_id to continue, or zero " +
+				"for the first page",
+		},
+	},
+	Action: queryChanEvents,
+}
+
+func queryChanEvents(ctx *cli.Context) error {
+	client, cleanup := getClient(ctx)
+	defer cleanup()
+
+	// Show command help if the channel point was not provided.
+	if ctx.NArg() == 0 && ctx.String("funding_txid") == "" {
+		return cli.ShowCommandHelp(ctx, "chanevents")
+	}
+
+	outpoint, err := parseChannelPoint(ctx)
+	if err != nil {
+		return err
+	}
+
+	startTime := ctx.Int64("start_time")
+	endTime := ctx.Int64("end_time")
+	if startTime < 0 || endTime < 0 {
+		return fmt.Errorf("start_time and end_time must be >= 0")
+	}
+	if endTime != 0 && startTime > endTime {
+		return fmt.Errorf("start_time must be <= end_time")
+	}
+
+	req := &frdrpc.ChannelEventsRequest{
+		ChanPoint: outpoint.String(),
+		StartTime: startTime,
+		EndTime:   endTime,
+		MaxEvents: uint32(ctx.Uint("max_events")),
+		LastId:    ctx.Int64("last_id"),
+	}
+
+	rpcCtx := context.Background()
+	report, err := client.GetChannelEvents(rpcCtx, req)
+	if err != nil {
+		return err
+	}
+
+	printRespJSON(report)
+	return nil
+}

cmd/frcli/main.go

@@ -57,6 +57,7 @@ func main() {
 		fiatEstimateCommand,
 		onChainReportCommand,
 		closeReportCommand,
+		chanEventsCommand,
 	}
 
 	if err := app.Run(os.Args); err != nil {

db/sqlc/migrations/000001_chanevents.down.sql

@@ -1,3 +1,4 @@
+DROP INDEX IF EXISTS channel_events_chan_id_id_idx;
 DROP INDEX IF EXISTS channel_events_chan_id_ts_idx;
 DROP TABLE IF EXISTS channel_events;
 DROP INDEX IF EXISTS channel_peer_idx;

db/sqlc/migrations/000001_chanevents.up.sql

@@ -41,8 +41,13 @@ CREATE TABLE IF NOT EXISTS channel_events (
     is_sync BOOLEAN NOT NULL DEFAULT FALSE
 );
 
--- This composite index is crucial for efficiently querying the event history
--- of a specific channel. It allows the database to quickly locate relevant rows
--- for a given channel, sorted by time. This is useful for fetching events
--- within a time range, and for finding the latest event before a certain time.
+-- This composite index supports the chronological access patterns
+-- (GetChannelEventsIter, GetLatestChannelEventBefore): events for a given
+-- channel sorted by time, with a per-channel time-range scan.
 CREATE INDEX IF NOT EXISTS channel_events_chan_id_ts_idx ON channel_events (channel_id, timestamp);
+
+-- This composite index supports the public GetChannelEvents query, which
+-- walks events for a given channel by id-keyset cursor (ORDER BY id ASC,
+-- WHERE id > $cursor). Without it, the planner would scan every event with
+-- id > $cursor across all channels and filter by channel_id afterwards.
+CREATE INDEX IF NOT EXISTS channel_events_chan_id_id_idx ON channel_events (channel_id, id);

db/sqlc/queries/chanevents.sql

@@ -21,5 +21,9 @@ INSERT INTO channel_events (
 
 -- name: GetChannelEvents :many
 SELECT * FROM channel_events
-WHERE channel_id = $1 AND timestamp >= $2 AND timestamp < $3
-ORDER BY timestamp ASC, id ASC;
+WHERE channel_id = $1
+  AND id > $2
+  AND timestamp >= $3
+  AND timestamp < $4
+ORDER BY id ASC
+LIMIT $5;

faraday.go

@@ -179,6 +179,7 @@ func (f *Faraday) Start() error {
 
 	cfg := &frdrpcserver.Config{
 		Lnd:           f.lnd.LndServices,
+		ChanEvents:    f.stores.ChanEventsStore,
 		BitcoinClient: f.bitcoinClient,
 	}
 
@@ -401,6 +402,7 @@ func (f *Faraday) StartAsSubserver(lndGrpc *lndclient.GrpcLndServices,
 
 	cfg := &frdrpcserver.Config{
 		Lnd:           lndGrpc.LndServices,
+		ChanEvents:    f.stores.ChanEventsStore,
 		BitcoinClient: f.bitcoinClient,
 	}