loop/sweepbatcher/store.go

package sweepbatcher

import (
	"context"
	"database/sql"
	"fmt"

	"github.com/btcsuite/btcd/btcutil"
	"github.com/btcsuite/btcd/chaincfg"
	"github.com/btcsuite/btcd/chaincfg/chainhash"
	"github.com/btcsuite/btcd/wire"
	"github.com/lightninglabs/loop/loopdb"
	"github.com/lightninglabs/loop/loopdb/sqlc"
	"github.com/lightningnetwork/lnd/clock"
	"github.com/lightningnetwork/lnd/lntypes"
)

type BaseDB interface {
	// ConfirmBatch confirms a batch by setting the state to confirmed.
	ConfirmBatch(ctx context.Context, id int32) error

	// GetBatchSweeps fetches all the sweeps that are part a batch.
	GetBatchSweeps(ctx context.Context, batchID int32) (
		[]sqlc.GetBatchSweepsRow, error)

	// GetBatchSweptAmount returns the total amount of sats swept by a
	// (confirmed) batch.
	GetBatchSweptAmount(ctx context.Context, batchID int32) (int64, error)

	// GetSweepStatus returns true if the sweep has been completed.
	GetSweepStatus(ctx context.Context, swapHash []byte) (bool, error)

	// GetParentBatch fetches the parent batch of a completed sweep.
	GetParentBatch(ctx context.Context, swapHash []byte) (sqlc.SweepBatch,
		error)

	// GetSwapUpdates fetches all the updates for a swap.
	GetSwapUpdates(ctx context.Context, swapHash []byte) (
		[]sqlc.SwapUpdate, error)

	// GetUnconfirmedBatches fetches all the batches from the
	// database that are not in a confirmed state.
	GetUnconfirmedBatches(ctx context.Context) ([]sqlc.SweepBatch, error)

	// InsertBatch inserts a batch into the database, returning the id of
	// the inserted batch.
	InsertBatch(ctx context.Context, arg sqlc.InsertBatchParams) (
		int32, error)

	// DropBatch drops a batch from the database.
	DropBatch(ctx context.Context, id int32) error

	// UpdateBatch updates a batch in the database.
	UpdateBatch(ctx context.Context, arg sqlc.UpdateBatchParams) error

	// UpsertSweep inserts a sweep into the database, or updates an existing
	// sweep if it already exists.
	UpsertSweep(ctx context.Context, arg sqlc.UpsertSweepParams) error

	// ExecTx allows for executing a function in the context of a database
	// transaction.
	ExecTx(ctx context.Context, txOptions loopdb.TxOptions,
		txBody func(*sqlc.Queries) error) error
}

// SQLStore manages the reservations in the database.
type SQLStore struct {
	baseDb BaseDB

	network *chaincfg.Params
	clock   clock.Clock
}

// NewSQLStore creates a new SQLStore.
func NewSQLStore(db BaseDB, network *chaincfg.Params) *SQLStore {
	return &SQLStore{
		baseDb:  db,
		network: network,
		clock:   clock.NewDefaultClock(),
	}
}

// FetchUnconfirmedSweepBatches fetches all the batches from the database that
// are not in a confirmed state.
func (s *SQLStore) FetchUnconfirmedSweepBatches(ctx context.Context) ([]*dbBatch,
	error) {

	var batches []*dbBatch

	dbBatches, err := s.baseDb.GetUnconfirmedBatches(ctx)
	if err != nil {
		return nil, err
	}

	for _, dbBatch := range dbBatches {
		batch := convertBatchRow(dbBatch)
		if err != nil {
			return nil, err
		}

		batches = append(batches, batch)
	}

	return batches, err
}

// InsertSweepBatch inserts a batch into the database, returning the id of the
// inserted batch.
func (s *SQLStore) InsertSweepBatch(ctx context.Context, batch *dbBatch) (int32,
	error) {

	return s.baseDb.InsertBatch(ctx, batchToInsertArgs(*batch))
}

// DropBatch drops a batch from the database. Note that we only use this call
// for batches that have no sweeps and so we'd not be able to resume.
func (s *SQLStore) DropBatch(ctx context.Context, id int32) error {
	readOpts := loopdb.NewSqlReadOpts()
	return s.baseDb.ExecTx(ctx, readOpts, func(tx *sqlc.Queries) error {
		dbSweeps, err := tx.GetBatchSweeps(ctx, id)
		if err != nil {
			return err
		}

		if len(dbSweeps) != 0 {
			return fmt.Errorf("cannot drop a non-empty batch")
		}

		return tx.DropBatch(ctx, id)
	})
}

// UpdateSweepBatch updates a batch in the database.
func (s *SQLStore) UpdateSweepBatch(ctx context.Context, batch *dbBatch) error {
	return s.baseDb.UpdateBatch(ctx, batchToUpdateArgs(*batch))
}

// ConfirmBatch confirms a batch by setting the state to confirmed.
func (s *SQLStore) ConfirmBatch(ctx context.Context, id int32) error {
	return s.baseDb.ConfirmBatch(ctx, id)
}

// FetchBatchSweeps fetches all the sweeps that are part a batch.
func (s *SQLStore) FetchBatchSweeps(ctx context.Context, id int32) (
	[]*dbSweep, error) {

	readOpts := loopdb.NewSqlReadOpts()
	var sweeps []*dbSweep

	err := s.baseDb.ExecTx(ctx, readOpts, func(tx *sqlc.Queries) error {
		dbSweeps, err := tx.GetBatchSweeps(ctx, id)
		if err != nil {
			return err
		}

		for _, dbSweep := range dbSweeps {
			updates, err := s.baseDb.GetSwapUpdates(
				ctx, dbSweep.SwapHash,
			)
			if err != nil {
				return err
			}

			sweep, err := s.convertSweepRow(dbSweep, updates)
			if err != nil {
				return err
			}

			sweeps = append(sweeps, &sweep)
		}

		return nil
	})
	if err != nil {
		return nil, err
	}

	return sweeps, nil
}

// TotalSweptAmount returns the total amount swept by a (confirmed) batch.
func (s *SQLStore) TotalSweptAmount(ctx context.Context, id int32) (
	btcutil.Amount, error) {

	amt, err := s.baseDb.GetBatchSweptAmount(ctx, id)
	if err != nil {
		return 0, err
	}

	return btcutil.Amount(amt), nil
}

// GetParentBatch fetches the parent batch of a completed sweep.
func (s *SQLStore) GetParentBatch(ctx context.Context, swapHash lntypes.Hash) (
	*dbBatch, error) {

	batch, err := s.baseDb.GetParentBatch(ctx, swapHash[:])
	if err != nil {
		return nil, err
	}

	if err != nil {
		return nil, err
	}

	return convertBatchRow(batch), nil
}

// UpsertSweep inserts a sweep into the database, or updates an existing sweep
// if it already exists.
func (s *SQLStore) UpsertSweep(ctx context.Context, sweep *dbSweep) error {
	return s.baseDb.UpsertSweep(ctx, sweepToUpsertArgs(*sweep))
}

// GetSweepStatus returns true if the sweep has been completed.
func (s *SQLStore) GetSweepStatus(ctx context.Context, swapHash lntypes.Hash) (
	bool, error) {

	return s.baseDb.GetSweepStatus(ctx, swapHash[:])
}

type dbBatch struct {
	// ID is the unique identifier of the batch.
	ID int32

	// State is the current state of the batch.
	State string

	// BatchTxid is the txid of the batch transaction.
	BatchTxid chainhash.Hash

	// BatchPkScript is the pkscript of the batch transaction.
	BatchPkScript []byte

	// LastRbfHeight is the height at which the last RBF attempt was made.
	LastRbfHeight int32

	// LastRbfSatPerKw is the sat per kw of the last RBF attempt.
	LastRbfSatPerKw int32

	// MaxTimeoutDistance is the maximum timeout distance of the batch.
	MaxTimeoutDistance int32
}

type dbSweep struct {
	// ID is the unique identifier of the sweep.
	ID int32

	// BatchID is the ID of the batch that the sweep belongs to.
	BatchID int32

	// SwapHash is the hash of the swap that the sweep belongs to.
	SwapHash lntypes.Hash

	// Outpoint is the outpoint of the sweep.
	Outpoint wire.OutPoint

	// Amount is the amount of the sweep.
	Amount btcutil.Amount

	// Completed indicates whether this sweep is completed.
	Completed bool

	// LoopOut is the loop out that the sweep belongs to.
	LoopOut *loopdb.LoopOut
}

// convertBatchRow converts a batch row from db to a sweepbatcher.Batch struct.
func convertBatchRow(row sqlc.SweepBatch) *dbBatch {
	batch := dbBatch{
		ID: row.ID,
	}

	if row.Confirmed {
		batch.State = batchOpen
	}

	if row.BatchTxID.Valid {
		err := chainhash.Decode(&batch.BatchTxid, row.BatchTxID.String)
		if err != nil {
			return nil
		}
	}

	batch.BatchPkScript = row.BatchPkScript

	if row.LastRbfHeight.Valid {
		batch.LastRbfHeight = row.LastRbfHeight.Int32
	}

	if row.LastRbfSatPerKw.Valid {
		batch.LastRbfSatPerKw = row.LastRbfSatPerKw.Int32
	}

	batch.MaxTimeoutDistance = row.MaxTimeoutDistance

	return &batch
}

// BatchToUpsertArgs converts a Batch struct to the arguments needed to insert
// it into the database.
func batchToInsertArgs(batch dbBatch) sqlc.InsertBatchParams {
	args := sqlc.InsertBatchParams{
		Confirmed: false,
		BatchTxID: sql.NullString{
			Valid:  true,
			String: batch.BatchTxid.String(),
		},
		BatchPkScript: batch.BatchPkScript,
		LastRbfHeight: sql.NullInt32{
			Valid: true,
			Int32: batch.LastRbfHeight,
		},
		LastRbfSatPerKw: sql.NullInt32{
			Valid: true,
			Int32: batch.LastRbfSatPerKw,
		},
		MaxTimeoutDistance: batch.MaxTimeoutDistance,
	}

	if batch.State == batchConfirmed {
		args.Confirmed = true
	}

	return args
}

// BatchToUpsertArgs converts a Batch struct to the arguments needed to insert
// it into the database.
func batchToUpdateArgs(batch dbBatch) sqlc.UpdateBatchParams {
	args := sqlc.UpdateBatchParams{
		ID:        batch.ID,
		Confirmed: false,
		BatchTxID: sql.NullString{
			Valid:  true,
			String: batch.BatchTxid.String(),
		},
		BatchPkScript: batch.BatchPkScript,
		LastRbfHeight: sql.NullInt32{
			Valid: true,
			Int32: batch.LastRbfHeight,
		},
		LastRbfSatPerKw: sql.NullInt32{
			Valid: true,
			Int32: batch.LastRbfSatPerKw,
		},
	}

	if batch.State == batchConfirmed {
		args.Confirmed = true
	}

	return args
}

// convertSweepRow converts a sweep row from db to a sweep struct.
func (s *SQLStore) convertSweepRow(row sqlc.GetBatchSweepsRow,
	updates []sqlc.SwapUpdate) (dbSweep, error) {

	sweep := dbSweep{
		ID:      row.ID,
		BatchID: row.BatchID,
		Amount:  btcutil.Amount(row.Amt),
	}

	swapHash, err := lntypes.MakeHash(row.SwapHash)
	if err != nil {
		return sweep, err
	}

	sweep.SwapHash = swapHash

	hash, err := chainhash.NewHash(row.OutpointTxid)
	if err != nil {
		return sweep, err
	}

	sweep.Outpoint = wire.OutPoint{
		Hash:  *hash,
		Index: uint32(row.OutpointIndex),
	}

	sweep.LoopOut, err = loopdb.ConvertLoopOutRow(
		s.network,
		sqlc.GetLoopOutSwapRow{
			ID:                     row.ID,
			SwapHash:               row.SwapHash,
			Preimage:               row.Preimage,
			InitiationTime:         row.InitiationTime,
			AmountRequested:        row.AmountRequested,
			CltvExpiry:             row.CltvExpiry,
			MaxMinerFee:            row.MaxMinerFee,
			MaxSwapFee:             row.MaxSwapFee,
			InitiationHeight:       row.InitiationHeight,
			ProtocolVersion:        row.ProtocolVersion,
			Label:                  row.Label,
			DestAddress:            row.DestAddress,
			SwapInvoice:            row.SwapInvoice,
			MaxSwapRoutingFee:      row.MaxSwapRoutingFee,
			SweepConfTarget:        row.SweepConfTarget,
			HtlcConfirmations:      row.HtlcConfirmations,
			OutgoingChanSet:        row.OutgoingChanSet,
			PrepayInvoice:          row.PrepayInvoice,
			MaxPrepayRoutingFee:    row.MaxPrepayRoutingFee,
			PublicationDeadline:    row.PublicationDeadline,
			SingleSweep:            row.SingleSweep,
			SenderScriptPubkey:     row.SenderScriptPubkey,
			ReceiverScriptPubkey:   row.ReceiverScriptPubkey,
			SenderInternalPubkey:   row.SenderInternalPubkey,
			ReceiverInternalPubkey: row.ReceiverInternalPubkey,
			ClientKeyFamily:        row.ClientKeyFamily,
			ClientKeyIndex:         row.ClientKeyIndex,
		}, updates,
	)

	return sweep, err
}

// sweepToUpsertArgs converts a Sweep struct to the arguments needed to insert.
func sweepToUpsertArgs(sweep dbSweep) sqlc.UpsertSweepParams {
	return sqlc.UpsertSweepParams{
		SwapHash:      sweep.SwapHash[:],
		BatchID:       sweep.BatchID,
		OutpointTxid:  sweep.Outpoint.Hash[:],
		OutpointIndex: int32(sweep.Outpoint.Index),
		Amt:           int64(sweep.Amount),
		Completed:     sweep.Completed,
	}
}