diff --git a/db/migrations.go b/db/migrations.go index b98fc6d..bd882c0 100644 --- a/db/migrations.go +++ b/db/migrations.go @@ -6,5 +6,5 @@ const ( // daemon. // // NOTE: This MUST be updated when a new migration is added. - LatestMigrationVersion = 0 + LatestMigrationVersion = 1 ) diff --git a/db/migrations_test.go b/db/migrations_test.go new file mode 100644 index 0000000..d87826e --- /dev/null +++ b/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", + ) +} diff --git a/db/schemas.go b/db/schemas.go index d65258b..1a7a209 100644 --- a/db/schemas.go +++ b/db/schemas.go @@ -5,4 +5,5 @@ import ( _ "embed" ) +//go:embed sqlc/migrations/*.*.sql var sqlSchemas embed.FS diff --git a/db/sqlc/chanevents.sql.go b/db/sqlc/chanevents.sql.go new file mode 100644 index 0000000..9720995 --- /dev/null +++ b/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 +} diff --git a/db/sqlc/db_custom.go b/db/sqlc/db_custom.go new file mode 100644 index 0000000..00b8afb --- /dev/null +++ b/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}} +} diff --git a/db/sqlc/migrations/000001_chanevents.down.sql b/db/sqlc/migrations/000001_chanevents.down.sql new file mode 100644 index 0000000..df2c627 --- /dev/null +++ b/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; diff --git a/db/sqlc/migrations/000001_chanevents.up.sql b/db/sqlc/migrations/000001_chanevents.up.sql new file mode 100644 index 0000000..f733b59 --- /dev/null +++ b/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); diff --git a/db/sqlc/models.go b/db/sqlc/models.go new file mode 100644 index 0000000..1094533 --- /dev/null +++ b/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 +} diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go new file mode 100644 index 0000000..f56f962 --- /dev/null +++ b/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) diff --git a/db/sqlc/queries/chanevents.sql b/db/sqlc/queries/chanevents.sql new file mode 100644 index 0000000..5370e4e --- /dev/null +++ b/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;