db: add chanevents tables and queries

db/migrations.go

@@ -6,5 +6,5 @@ const (
 	// daemon.
 	//
 	// NOTE: This MUST be updated when a new migration is added.
-	LatestMigrationVersion = 0
+	LatestMigrationVersion = 1
 )

db/migrations_test.go

@@ -0,0 +1,41 @@
+package db
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestLatestMigrationVersion ensures that LatestMigrationVersion stays in sync
+// with the highest-numbered .up.sql file in the migrations directory. Each
+// migration — whether pure SQL or programmatic (with a dummy SQL file) — gets
+// its own numbered file pair, so the max file number must equal the constant.
+func TestLatestMigrationVersion(t *testing.T) {
+	entries, err := sqlSchemas.ReadDir("sqlc/migrations")
+	require.NoError(t, err)
+
+	var maxVersion uint
+	for _, entry := range entries {
+		if !strings.HasSuffix(entry.Name(), ".up.sql") {
+			continue
+		}
+
+		parts := strings.SplitN(entry.Name(), "_", 2)
+		require.NotEmpty(t, parts)
+
+		v, err := strconv.ParseUint(parts[0], 10, 64)
+		require.NoError(t, err)
+
+		if uint(v) > maxVersion {
+			maxVersion = uint(v)
+		}
+	}
+
+	require.EqualValues(
+		t, maxVersion, LatestMigrationVersion,
+		"LatestMigrationVersion is out of date, update "+
+			"db/migrations.go",
+	)
+}

db/schemas.go

@@ -5,4 +5,5 @@ import (
 	_ "embed"
 )
 
+//go:embed sqlc/migrations/*.*.sql
 var sqlSchemas embed.FS

db/sqlc/chanevents.sql.go

@@ -0,0 +1,150 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.25.0
+// source: chanevents.sql
+
+package sqlc
+
+import (
+	"context"
+	"database/sql"
+	"time"
+)
+
+const getChannelByChanPoint = `-- name: GetChannelByChanPoint :one
+SELECT id, channel_point, short_channel_id, peer_id FROM channels WHERE channel_point = $1
+`
+
+func (q *Queries) GetChannelByChanPoint(ctx context.Context, channelPoint string) (Channel, error) {
+	row := q.db.QueryRowContext(ctx, getChannelByChanPoint, channelPoint)
+	var i Channel
+	err := row.Scan(
+		&i.ID,
+		&i.ChannelPoint,
+		&i.ShortChannelID,
+		&i.PeerID,
+	)
+	return i, err
+}
+
+const getChannelByShortChanID = `-- name: GetChannelByShortChanID :one
+SELECT id, channel_point, short_channel_id, peer_id FROM channels WHERE short_channel_id = $1
+`
+
+func (q *Queries) GetChannelByShortChanID(ctx context.Context, shortChannelID int64) (Channel, error) {
+	row := q.db.QueryRowContext(ctx, getChannelByShortChanID, shortChannelID)
+	var i Channel
+	err := row.Scan(
+		&i.ID,
+		&i.ChannelPoint,
+		&i.ShortChannelID,
+		&i.PeerID,
+	)
+	return i, err
+}
+
+const getChannelEvents = `-- name: GetChannelEvents :many
+SELECT id, channel_id, event_type, timestamp, local_balance_sat, remote_balance_sat FROM channel_events
+WHERE channel_id = $1 AND timestamp >= $2 AND timestamp < $3
+ORDER BY timestamp ASC, id ASC
+`
+
+type GetChannelEventsParams struct {
+	ChannelID   int64
+	Timestamp   time.Time
+	Timestamp_2 time.Time
+}
+
+func (q *Queries) GetChannelEvents(ctx context.Context, arg GetChannelEventsParams) ([]ChannelEvent, error) {
+	rows, err := q.db.QueryContext(ctx, getChannelEvents, arg.ChannelID, arg.Timestamp, arg.Timestamp_2)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []ChannelEvent
+	for rows.Next() {
+		var i ChannelEvent
+		if err := rows.Scan(
+			&i.ID,
+			&i.ChannelID,
+			&i.EventType,
+			&i.Timestamp,
+			&i.LocalBalanceSat,
+			&i.RemoteBalanceSat,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
+const getPeerByPubKey = `-- name: GetPeerByPubKey :one
+SELECT id, pubkey FROM peers WHERE pubkey = $1
+`
+
+func (q *Queries) GetPeerByPubKey(ctx context.Context, pubkey string) (Peer, error) {
+	row := q.db.QueryRowContext(ctx, getPeerByPubKey, pubkey)
+	var i Peer
+	err := row.Scan(&i.ID, &i.Pubkey)
+	return i, err
+}
+
+const insertChannel = `-- name: InsertChannel :one
+INSERT INTO channels (channel_point, short_channel_id, peer_id) VALUES ($1, $2, $3) RETURNING id
+`
+
+type InsertChannelParams struct {
+	ChannelPoint   string
+	ShortChannelID int64
+	PeerID         int64
+}
+
+func (q *Queries) InsertChannel(ctx context.Context, arg InsertChannelParams) (int64, error) {
+	row := q.db.QueryRowContext(ctx, insertChannel, arg.ChannelPoint, arg.ShortChannelID, arg.PeerID)
+	var id int64
+	err := row.Scan(&id)
+	return id, err
+}
+
+const insertChannelEvent = `-- name: InsertChannelEvent :exec
+INSERT INTO channel_events (
+    channel_id, event_type, timestamp, local_balance_sat, remote_balance_sat
+) VALUES ($1, $2, $3, $4, $5)
+`
+
+type InsertChannelEventParams struct {
+	ChannelID        int64
+	EventType        int16
+	Timestamp        time.Time
+	LocalBalanceSat  sql.NullInt64
+	RemoteBalanceSat sql.NullInt64
+}
+
+func (q *Queries) InsertChannelEvent(ctx context.Context, arg InsertChannelEventParams) error {
+	_, err := q.db.ExecContext(ctx, insertChannelEvent,
+		arg.ChannelID,
+		arg.EventType,
+		arg.Timestamp,
+		arg.LocalBalanceSat,
+		arg.RemoteBalanceSat,
+	)
+	return err
+}
+
+const insertPeer = `-- name: InsertPeer :one
+INSERT INTO peers (pubkey) VALUES ($1) RETURNING id
+`
+
+func (q *Queries) InsertPeer(ctx context.Context, pubkey string) (int64, error) {
+	row := q.db.QueryRowContext(ctx, insertPeer, pubkey)
+	var id int64
+	err := row.Scan(&id)
+	return id, err
+}

db/sqlc/db_custom.go

@@ -0,0 +1,34 @@
+// Package sqlc provides a set of custom database queries and utilities
+// for interacting with the SQL database used in the application. It includes
+// generated code from sqlc as well as custom wrappers to handle different
+// database backends.
+package sqlc
+
+import (
+	"github.com/lightningnetwork/lnd/sqldb/v2"
+)
+
+// wrappedTX is a wrapper around a DBTX that also stores the database backend
+// type.
+type wrappedTX struct {
+	DBTX
+
+	backendType sqldb.BackendType
+}
+
+// Backend returns the type of database backend we're using.
+func (q *Queries) Backend() sqldb.BackendType {
+	wtx, ok := q.db.(*wrappedTX)
+	if !ok {
+		// Shouldn't happen unless a new database backend type is added
+		// but not initialized correctly.
+		return sqldb.BackendTypeUnknown
+	}
+
+	return wtx.backendType
+}
+
+// NewForType creates a new Queries instance for the given database type.
+func NewForType(db DBTX, typ sqldb.BackendType) *Queries {
+	return &Queries{db: &wrappedTX{db, typ}}
+}

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

@@ -0,0 +1,5 @@
+DROP INDEX IF EXISTS channel_events_chan_id_ts_idx;
+DROP TABLE IF EXISTS channel_events;
+DROP INDEX IF EXISTS channel_peer_idx;
+DROP TABLE IF EXISTS channels;
+DROP TABLE IF EXISTS peers;

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

@@ -0,0 +1,45 @@
+-- The peers table stores all the peers that we have channels with.
+CREATE TABLE IF NOT EXISTS peers (
+    -- The auto incrementing primary key.
+    id INTEGER PRIMARY KEY,
+    -- The public key of the peer.
+    pubkey TEXT NOT NULL UNIQUE
+);
+
+-- The channels table stores all the channels that we have with our peers.
+CREATE TABLE IF NOT EXISTS channels (
+    -- The auto incrementing primary key.
+    id INTEGER PRIMARY KEY,
+    -- The channel point, as a 'txid:output_index' string.
+    channel_point TEXT NOT NULL UNIQUE,
+    -- The short channel ID.
+    short_channel_id BIGINT NOT NULL UNIQUE,
+    -- The peer that this channel is with.
+    peer_id BIGINT NOT NULL REFERENCES peers(id) ON DELETE CASCADE
+);
+CREATE INDEX IF NOT EXISTS channel_peer_idx ON channels (peer_id);
+
+-- The channel_events table stores all the events that are associated with a
+-- particular channel.
+CREATE TABLE IF NOT EXISTS channel_events (
+    -- The auto incrementing primary key.
+    id INTEGER PRIMARY KEY,
+    -- The channel that this event is associated with.
+    channel_id BIGINT NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
+    -- The type of event.
+    event_type SMALLINT NOT NULL,
+    -- The time the event occurred.
+    timestamp TIMESTAMP NOT NULL,
+    -- The local balance of the channel at the time of the event.
+    -- This is only populated for balance update events.
+    local_balance_sat BIGINT CHECK (local_balance_sat >= 0),
+    -- The remote balance of the channel at the time of the event.
+    -- This is only populated for balance update events.
+    remote_balance_sat BIGINT CHECK (remote_balance_sat >= 0)
+);
+
+-- 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.
+CREATE INDEX IF NOT EXISTS channel_events_chan_id_ts_idx ON channel_events (channel_id, timestamp);

db/sqlc/models.go

@@ -0,0 +1,31 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.25.0
+
+package sqlc
+
+import (
+	"database/sql"
+	"time"
+)
+
+type Channel struct {
+	ID             int64
+	ChannelPoint   string
+	ShortChannelID int64
+	PeerID         int64
+}
+
+type ChannelEvent struct {
+	ID               int64
+	ChannelID        int64
+	EventType        int16
+	Timestamp        time.Time
+	LocalBalanceSat  sql.NullInt64
+	RemoteBalanceSat sql.NullInt64
+}
+
+type Peer struct {
+	ID     int64
+	Pubkey string
+}

db/sqlc/querier.go

@@ -0,0 +1,21 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.25.0
+
+package sqlc
+
+import (
+	"context"
+)
+
+type Querier interface {
+	GetChannelByChanPoint(ctx context.Context, channelPoint string) (Channel, error)
+	GetChannelByShortChanID(ctx context.Context, shortChannelID int64) (Channel, error)
+	GetChannelEvents(ctx context.Context, arg GetChannelEventsParams) ([]ChannelEvent, error)
+	GetPeerByPubKey(ctx context.Context, pubkey string) (Peer, error)
+	InsertChannel(ctx context.Context, arg InsertChannelParams) (int64, error)
+	InsertChannelEvent(ctx context.Context, arg InsertChannelEventParams) error
+	InsertPeer(ctx context.Context, pubkey string) (int64, error)
+}
+
+var _ Querier = (*Queries)(nil)

db/sqlc/queries/chanevents.sql

@@ -0,0 +1,24 @@
+-- name: InsertPeer :one
+INSERT INTO peers (pubkey) VALUES ($1) RETURNING id;
+
+-- name: GetPeerByPubKey :one
+SELECT * FROM peers WHERE pubkey = $1;
+
+-- name: InsertChannel :one
+INSERT INTO channels (channel_point, short_channel_id, peer_id) VALUES ($1, $2, $3) RETURNING id;
+
+-- name: GetChannelByChanPoint :one
+SELECT * FROM channels WHERE channel_point = $1;
+
+-- name: GetChannelByShortChanID :one
+SELECT * FROM channels WHERE short_channel_id = $1;
+
+-- name: InsertChannelEvent :exec
+INSERT INTO channel_events (
+    channel_id, event_type, timestamp, local_balance_sat, remote_balance_sat
+) VALUES ($1, $2, $3, $4, $5);
+
+-- name: GetChannelEvents :many
+SELECT * FROM channel_events
+WHERE channel_id = $1 AND timestamp >= $2 AND timestamp < $3
+ORDER BY timestamp ASC, id ASC;