lsps2: extend store for forwarding

lsps2/store.go

@@ -3,9 +3,12 @@ package lsps2
 import (
 	"context"
 	"errors"
+	"log"
+	"time"
 
 	"github.com/breez/lspd/basetypes"
 	"github.com/breez/lspd/shared"
+	"github.com/btcsuite/btcd/wire"
 )
 
 type RegisterBuy struct {
@@ -17,8 +20,45 @@ type RegisterBuy struct {
 	Mode             OpeningMode
 }
 
+type BuyRegistration struct {
+	Id               uint64
+	LspId            string
+	PeerId           string // TODO: Make peerId in the registration a byte array.
+	Scid             basetypes.ShortChannelID
+	OpeningFeeParams shared.OpeningFeeParams
+	PaymentSizeMsat  *uint64
+	Mode             OpeningMode
+	ChannelPoint     *wire.OutPoint
+	IsComplete       bool
+}
+
+func (b *BuyRegistration) IsExpired() bool {
+	t, err := time.Parse(basetypes.TIME_FORMAT, b.OpeningFeeParams.ValidUntil)
+	if err != nil {
+		log.Printf("BuyRegistration.IsExpired(): time.Parse(%v, %v) error: %v", basetypes.TIME_FORMAT, b.OpeningFeeParams.ValidUntil, err)
+		return true
+	}
+
+	if time.Now().UTC().After(t) {
+		return true
+	}
+
+	return false
+}
+
+type ChannelOpened struct {
+	RegistrationId  uint64
+	Outpoint        *wire.OutPoint
+	FeeMsat         uint64
+	PaymentSizeMsat uint64
+}
+
 var ErrScidExists = errors.New("scid exists")
+var ErrNotFound = errors.New("not found")
 
 type Lsps2Store interface {
 	RegisterBuy(ctx context.Context, req *RegisterBuy) error
+	GetBuyRegistration(ctx context.Context, scid basetypes.ShortChannelID) (*BuyRegistration, error)
+	SetChannelOpened(ctx context.Context, channelOpened *ChannelOpened) error
+	SetCompleted(ctx context.Context, registrationId uint64) error
 }

postgresql/lsps2_store.go

@@ -2,9 +2,14 @@ package postgresql
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
+	"github.com/breez/lspd/basetypes"
 	"github.com/breez/lspd/lsps2"
+	"github.com/breez/lspd/shared"
+	"github.com/btcsuite/btcd/wire"
+	"github.com/jackc/pgx/v4"
 	"github.com/jackc/pgx/v4/pgxpool"
 )
 
@@ -35,7 +40,7 @@ func (s *Lsps2Store) RegisterBuy(
 		 ,  params_max_client_to_self_delay
 		 ,  params_promise
 		 )
-		 VALUES ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?`,
+		 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`,
 		req.LspId,
 		req.PeerId,
 		int64(uint64(req.Scid)),
@@ -58,3 +63,138 @@ func (s *Lsps2Store) RegisterBuy(
 
 	return nil
 }
+
+func (s *Lsps2Store) GetBuyRegistration(ctx context.Context, scid basetypes.ShortChannelID) (*lsps2.BuyRegistration, error) {
+	row := s.pool.QueryRow(
+		ctx,
+		`SELECT r.id
+		,       r.lsp_id
+		,       r.peer_id
+		,       r.scid
+		,       r.mode
+		,       r.payment_size_msat
+		,       r.params_min_fee_msat
+		,       r.params_proportional
+		,       r.params_valid_until
+		,       r.params_min_lifetime
+		,       r.params_max_client_to_self_delay
+		,       r.params_promise
+		,       c.funding_tx_id
+		,       c.funding_tx_outnum
+		,       c.is_completed
+		 FROM lsps2.buy_registrations r
+		 LEFT JOIN lsps2.bought_channels c ON r.id = c.registration_id
+		 WHERE r.scid = $1`,
+		int64(uint64(scid)),
+	)
+	var db_id uint64
+	var db_lsp_id string
+	var db_peer_id string
+	var db_scid int64
+	var db_mode int
+	var db_payment_size_msat *int64
+	var db_params_min_fee_msat int64
+	var db_params_proportional uint32
+	var db_params_valid_until string
+	var db_params_min_lifetime uint32
+	var db_params_max_client_to_self_delay uint32
+	var db_params_promise string
+	var db_funding_tx_id []byte
+	var db_funding_tx_outnum uint32
+	var db_is_completed bool
+	err := row.Scan(
+		&db_id,
+		&db_lsp_id,
+		&db_peer_id,
+		&db_scid,
+		db_payment_size_msat,
+		&db_params_min_fee_msat,
+		&db_params_proportional,
+		&db_params_valid_until,
+		&db_params_min_lifetime,
+		&db_params_max_client_to_self_delay,
+		&db_params_promise,
+		&db_funding_tx_id,
+		&db_funding_tx_outnum,
+		&db_is_completed,
+	)
+	if err == pgx.ErrNoRows {
+		return nil, lsps2.ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	var paymentSizeMsat *uint64
+	if db_payment_size_msat != nil {
+		p := uint64(*db_payment_size_msat)
+		paymentSizeMsat = &p
+	}
+
+	var cp *wire.OutPoint
+	if db_funding_tx_id != nil {
+		cp, err = basetypes.NewOutPoint(db_funding_tx_id, db_funding_tx_outnum)
+		if err != nil {
+			return nil, fmt.Errorf("invalid funding tx id in db: %x", db_funding_tx_id)
+		}
+	}
+
+	return &lsps2.BuyRegistration{
+		Id:     db_id,
+		LspId:  db_lsp_id,
+		PeerId: db_peer_id,
+		Scid:   basetypes.ShortChannelID(uint64(db_scid)),
+		OpeningFeeParams: shared.OpeningFeeParams{
+			MinFeeMsat:           uint64(db_params_min_fee_msat),
+			Proportional:         db_params_proportional,
+			ValidUntil:           db_params_valid_until,
+			MinLifetime:          db_params_min_lifetime,
+			MaxClientToSelfDelay: db_params_max_client_to_self_delay,
+			Promise:              db_params_promise,
+		},
+		PaymentSizeMsat: paymentSizeMsat,
+		Mode:            lsps2.OpeningMode(db_mode),
+		ChannelPoint:    cp,
+		IsComplete:      db_is_completed,
+	}, nil
+}
+
+func (s *Lsps2Store) SetChannelOpened(ctx context.Context, channelOpened *lsps2.ChannelOpened) error {
+	_, err := s.pool.Exec(
+		ctx,
+		`INSERT INTO lsps2.bought_channels (
+			registration_id,
+			funding_tx_id,
+			funding_tx_outnum,
+			fee_msat,
+			payment_size_msat,
+			is_completed
+		) VALUES ($1, $2, $3, $4, $5, false)`,
+		channelOpened.RegistrationId,
+		channelOpened.Outpoint.Hash[:],
+		channelOpened.Outpoint.Index,
+		int64(channelOpened.FeeMsat),
+		int64(channelOpened.PaymentSizeMsat),
+	)
+
+	return err
+}
+
+func (s *Lsps2Store) SetCompleted(ctx context.Context, registrationId uint64) error {
+	rows, err := s.pool.Exec(
+		ctx,
+		`UPDATE lsps2.bought_channels
+		 SET is_completed = true
+		 WHERE registration_id = $1`,
+		registrationId,
+	)
+	if err != nil {
+		return err
+	}
+
+	if rows.RowsAffected() == 0 {
+		return fmt.Errorf("no rows were updated")
+	}
+
+	return nil
+}

postgresql/migrations/000014_lsps2_buy.down.sql

@@ -1,3 +1,5 @@
+DROP INDEX idx_lsps2_bought_channels_registration_id;
+DROP TABLE lsps2.bought_channels;
 DROP INDEX idx_lsps2_buy_registrations_valid_until;
 DROP INDEX idx_lsps2_buy_registrations_scid;
 DROP TABLE lsps2.buy_registrations;

postgresql/migrations/000014_lsps2_buy.up.sql

@@ -15,3 +15,18 @@ CREATE TABLE lsps2.buy_registrations (
 );
 CREATE UNIQUE INDEX idx_lsps2_buy_registrations_scid ON lsps2.buy_registrations (scid);
 CREATE INDEX idx_lsps2_buy_registrations_valid_until ON lsps2.buy_registrations (params_valid_until);
+
+CREATE TABLE lsps2.bought_channels (
+	id bigserial PRIMARY KEY,
+	registration_id bigint NOT NULL,
+	funding_tx_id bytea NOT NULL,
+	funding_tx_outnum bigint NOT NULL,
+	fee_msat bigint NOT NULL,
+	payment_size_msat bigint NOT NULL,
+	is_completed boolean NOT NULL,
+	CONSTRAINT fk_buy_registration
+	  FOREIGN KEY(registration_id)
+	  REFERENCES lsps2.buy_registrations(id)
+	  ON DELETE CASCADE
+);
+CREATE INDEX idx_lsps2_bought_channels_registration_id ON lsps2.bought_channels (registration_id);