chanevents: implement store

Author: bitromortac

chanevents/chanevents.go

@@ -0,0 +1,109 @@
+// Package chanevents contains functions for monitoring and storing channel
+// events such as online/offline and balance updates.
+package chanevents
+
+import (
+	"time"
+
+	"github.com/btcsuite/btcd/btcutil"
+	"github.com/lightningnetwork/lnd/fn/v2"
+)
+
+// EventType is an enum for the different types of channel events.
+type EventType int16
+
+const (
+	// EventTypeUnknown is the unknown event type.
+	EventTypeUnknown = 0
+
+	// EventTypeOnline is the online event type.
+	EventTypeOnline = 1
+
+	// EventTypeOffline is the offline event type.
+	EventTypeOffline = 2
+
+	// EventTypeUpdate is the balance update event type.
+	EventTypeUpdate = 3
+)
+
+// String returns the string representation of the event type.
+func (e EventType) String() string {
+	switch e {
+	case EventTypeOnline:
+		return "online"
+
+	case EventTypeOffline:
+		return "offline"
+
+	case EventTypeUpdate:
+		return "update"
+
+	default:
+		return "unknown"
+	}
+}
+
+// EventTypeFromString returns the event type from a string.
+func EventTypeFromString(s string) EventType {
+	switch s {
+	case "online":
+		return EventTypeOnline
+
+	case "offline":
+		return EventTypeOffline
+
+	case "update":
+		return EventTypeUpdate
+
+	default:
+		return EventTypeUnknown
+	}
+}
+
+// Peer is the application-level representation of a peer.
+type Peer struct {
+	// ID is the database ID of the peer.
+	ID int64
+
+	// PubKey is the public key of the peer.
+	PubKey string
+}
+
+// Channel is the application-level representation of a channel.
+type Channel struct {
+	// ID is the database ID of the channel.
+	ID int64
+
+	// ChannelPoint is the channel point of the channel.
+	ChannelPoint string
+
+	// ShortChannelID is the short channel ID of the channel.
+	ShortChannelID uint64
+
+	// PeerID is the database ID of the peer that this channel is with.
+	PeerID int64
+}
+
+// ChannelEvent is the application-level representation of a channel event.
+type ChannelEvent struct {
+	// ID is the database ID of the event.
+	ID int64
+
+	// ChannelID is the database ID of the channel that this event is
+	// associated with.
+	ChannelID int64
+
+	// EventType is the type of the event.
+	EventType EventType
+
+	// Timestamp is the time that the event occurred.
+	Timestamp time.Time
+
+	// LocalBalance is the local balance of the channel at the time of the
+	// event. This is only populated for balance update events.
+	LocalBalance fn.Option[btcutil.Amount]
+
+	// RemoteBalance is the remote balance of the channel at the time of the
+	// event. This is only populated for balance update events.
+	RemoteBalance fn.Option[btcutil.Amount]
+}

chanevents/store.go

@@ -0,0 +1,268 @@
+package chanevents
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/btcsuite/btcd/btcutil"
+	"github.com/lightninglabs/faraday/db/sqlc"
+	"github.com/lightningnetwork/lnd/clock"
+	"github.com/lightningnetwork/lnd/fn/v2"
+	"github.com/lightningnetwork/lnd/sqldb/v2"
+)
+
+var (
+	errUnknownPeer    = errors.New("unknown peer")
+	errUnknownChannel = errors.New("unknown channel")
+)
+
+// Queries is a subset of the sqlc.Queries interface that can be used to
+// interact with the peers, channels and channel_events tables.
+type Queries interface {
+	InsertPeer(ctx context.Context, pubkey string) (int64, error)
+
+	GetPeerByPubKey(ctx context.Context, pubkey string) (sqlc.Peer, error)
+
+	InsertChannel(ctx context.Context,
+		arg sqlc.InsertChannelParams) (int64, error)
+
+	GetChannelByChanPoint(ctx context.Context,
+		channelPoint string) (sqlc.Channel, error)
+
+	GetChannelByShortChanID(ctx context.Context,
+		shortChannelID int64) (sqlc.Channel, error)
+
+	InsertChannelEvent(ctx context.Context,
+		arg sqlc.InsertChannelEventParams) error
+
+	GetChannelEvents(ctx context.Context,
+		arg sqlc.GetChannelEventsParams) ([]sqlc.ChannelEvent, error)
+}
+
+// Store provides access to the db for channel events.
+type Store struct {
+	// db is all the higher level queries that the SQLStore has access to in
+	// order to implement all its CRUD logic.
+	db BatchedSQLQueries
+
+	// BaseDB represents the underlying database connection.
+	*sqldb.BaseDB
+
+	clock clock.Clock
+}
+
+// BatchedSQLQueries combines the SQLQueries interface with the BatchedTx
+// interface, allowing for multiple queries to be executed in single SQL
+// transaction.
+type BatchedSQLQueries interface {
+	SQLQueries
+
+	sqldb.BatchedTx[SQLQueries]
+}
+
+// SQLQueries is a subset of the sqlc.Queries interface that can be used to
+// interact with various chanevents tables.
+type SQLQueries interface {
+	sqldb.BaseQuerier
+
+	Queries
+}
+
+type SQLQueriesExecutor[T sqldb.BaseQuerier] struct {
+	*sqldb.TransactionExecutor[T]
+
+	SQLQueries
+}
+
+// NewStore creates a new SQLStore instance given an open SQLQueries storage
+// backend.
+func NewStore(sqlDB *sqldb.BaseDB, queries *sqlc.Queries,
+	clock clock.Clock) *Store {
+
+	txExecutor := sqldb.NewTransactionExecutor(
+		sqlDB,
+		func(tx *sql.Tx) SQLQueries {
+			return queries.WithTx(tx)
+		},
+	)
+
+	executor := &SQLQueriesExecutor[SQLQueries]{
+		TransactionExecutor: txExecutor,
+		SQLQueries:          queries,
+	}
+
+	return &Store{
+		db:     executor,
+		BaseDB: sqlDB,
+		clock:  clock,
+	}
+}
+
+// AddPeer adds a new peer to the database.
+func (s *Store) AddPeer(ctx context.Context, pubkey string) (int64, error) {
+	id, err := s.db.InsertPeer(ctx, pubkey)
+	if err != nil {
+		return 0, fmt.Errorf("failed to insert peer: %w", err)
+	}
+
+	return id, nil
+}
+
+// GetPeer retrieves a peer by their public key.
+func (s *Store) GetPeer(ctx context.Context, pubkey string) (*Peer, error) {
+	dbPeer, err := s.db.GetPeerByPubKey(ctx, pubkey)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, errUnknownPeer
+		}
+
+		return nil, fmt.Errorf("failed to get peer: %w", err)
+	}
+
+	return &Peer{
+		ID:     dbPeer.ID,
+		PubKey: dbPeer.Pubkey,
+	}, nil
+}
+
+// int64ToSCID converts an int64 to a uint64 ShortChannelID. The BOLT spec
+// encodes SCIDs as uint64, but SQL only supports signed int64. We preserve the
+// bits, which means SCIDs with the high bit set will appear negative in the
+// database. Direct SQL queries (e.g. ORDER BY short_channel_id) will not sort
+// these correctly, but round-tripping through Go preserves the value.
+func int64ToSCID(i int64) uint64 {
+	return uint64(i)
+}
+
+// scidToInt64 converts a uint64 ShortChannelID to an int64 for SQL storage.
+func scidToInt64(u uint64) int64 {
+	return int64(u)
+}
+
+// AddChannel adds a new channel for a peer.
+func (s *Store) AddChannel(ctx context.Context, channelPoint string,
+	shortChannelID uint64, peerID int64) (int64, error) {
+
+	id, err := s.db.InsertChannel(
+		ctx, sqlc.InsertChannelParams{
+			ChannelPoint:   channelPoint,
+			ShortChannelID: scidToInt64(shortChannelID),
+			PeerID:         peerID,
+		},
+	)
+	if err != nil {
+		return 0, fmt.Errorf("failed to insert channel: %w", err)
+	}
+
+	return id, nil
+}
+
+// GetChannel retrieves a channel by its channel point.
+func (s *Store) GetChannel(ctx context.Context, channelPoint string) (*Channel,
+	error) {
+
+	dbChannel, err := s.db.GetChannelByChanPoint(ctx, channelPoint)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, errUnknownChannel
+		}
+
+		return nil, fmt.Errorf("failed to get channel: %w", err)
+	}
+
+	return &Channel{
+		ID:             dbChannel.ID,
+		ChannelPoint:   dbChannel.ChannelPoint,
+		ShortChannelID: int64ToSCID(dbChannel.ShortChannelID),
+		PeerID:         dbChannel.PeerID,
+	}, nil
+}
+
+// AddChannelEvent adds a new channel event.
+func (s *Store) AddChannelEvent(ctx context.Context,
+	event *ChannelEvent) error {
+
+	var localBalance sql.NullInt64
+	event.LocalBalance.WhenSome(
+		func(b btcutil.Amount) {
+			localBalance.Int64 = int64(b)
+			localBalance.Valid = true
+		},
+	)
+
+	var remoteBalance sql.NullInt64
+	event.RemoteBalance.WhenSome(
+		func(b btcutil.Amount) {
+			remoteBalance.Int64 = int64(b)
+			remoteBalance.Valid = true
+		},
+	)
+
+	timestamp := event.Timestamp.UTC()
+	if timestamp.IsZero() {
+		timestamp = s.clock.Now().UTC()
+	}
+
+	err := s.db.InsertChannelEvent(
+		ctx, sqlc.InsertChannelEventParams{
+			ChannelID:        event.ChannelID,
+			EventType:        int16(event.EventType),
+			Timestamp:        timestamp,
+			LocalBalanceSat:  localBalance,
+			RemoteBalanceSat: remoteBalance,
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("failed to insert channel event: %w", err)
+	}
+
+	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) {
+
+	dbEvents, err := s.db.GetChannelEvents(
+		ctx, sqlc.GetChannelEventsParams{
+			ChannelID:   channelID,
+			Timestamp:   startTime.UTC(),
+			Timestamp_2: endTime.UTC(),
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get channel events: %w", err)
+	}
+
+	events := make([]*ChannelEvent, len(dbEvents))
+	for i, dbEvent := range dbEvents {
+		var localBalance fn.Option[btcutil.Amount]
+		if dbEvent.LocalBalanceSat.Valid {
+			amt := btcutil.Amount(dbEvent.LocalBalanceSat.Int64)
+			localBalance = fn.Some(amt)
+		}
+
+		var remoteBalance fn.Option[btcutil.Amount]
+		if dbEvent.RemoteBalanceSat.Valid {
+			amt := btcutil.Amount(dbEvent.RemoteBalanceSat.Int64)
+			remoteBalance = fn.Some(amt)
+		}
+
+		events[i] = &ChannelEvent{
+			ID:            dbEvent.ID,
+			ChannelID:     dbEvent.ChannelID,
+			EventType:     EventType(dbEvent.EventType),
+			Timestamp:     dbEvent.Timestamp.UTC(),
+			LocalBalance:  localBalance,
+			RemoteBalance: remoteBalance,
+		}
+	}
+
+	return events, nil
+}