loop/staticaddr/loopin/sql_store.go

package loopin

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"time"

	"github.com/btcsuite/btcd/btcec/v2"
	"github.com/btcsuite/btcd/btcutil"
	"github.com/btcsuite/btcd/chaincfg"
	"github.com/btcsuite/btcd/chaincfg/chainhash"
	"github.com/lightninglabs/loop/fsm"
	"github.com/lightninglabs/loop/loopdb"
	"github.com/lightninglabs/loop/loopdb/sqlc"
	"github.com/lightninglabs/loop/staticaddr/deposit"
	"github.com/lightninglabs/loop/staticaddr/version"
	"github.com/lightningnetwork/lnd/clock"
	"github.com/lightningnetwork/lnd/keychain"
	"github.com/lightningnetwork/lnd/lntypes"
	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
)

const OutpointSeparator = ";"

// sqlStoreUpdateTime returns the PostgreSQL-compatible timestamp precision used
// for persisted loop-in update metadata.
func sqlStoreUpdateTime(clock clock.Clock) time.Time {
	return clock.Now().Truncate(time.Microsecond)
}

var (
	// ErrInvalidOutpoint is returned when an outpoint contains the outpoint
	// separator.
	ErrInvalidOutpoint = errors.New("outpoint contains outpoint separator")

	// ErrLoopInNotFound is returned when a loop-in swap is not stored.
	ErrLoopInNotFound = errors.New("static address loop-in not found")
)

// Querier is the interface that contains all the queries generated by sqlc for
// the static_address_swaps table.
type Querier interface {
	// InsertSwap inserts a new base swap.
	InsertSwap(ctx context.Context, arg sqlc.InsertSwapParams) error

	// InsertHtlcKeys inserts the htlc keys for a swap.
	InsertHtlcKeys(ctx context.Context, arg sqlc.InsertHtlcKeysParams) error

	// InsertStaticAddressLoopIn inserts a new static address loop-in swap.
	InsertStaticAddressLoopIn(ctx context.Context,
		arg sqlc.InsertStaticAddressLoopInParams) error

	// InsertStaticAddressMetaUpdate inserts metadata about loop-in
	// updates.
	InsertStaticAddressMetaUpdate(ctx context.Context,
		arg sqlc.InsertStaticAddressMetaUpdateParams) error

	// UpdateStaticAddressLoopIn updates a loop-in swap.
	UpdateStaticAddressLoopIn(ctx context.Context,
		arg sqlc.UpdateStaticAddressLoopInParams) error

	// RecordStaticAddressRiskDecision stores the server's confirmation-risk
	// decision for a loop-in swap.
	RecordStaticAddressRiskDecision(ctx context.Context,
		arg sqlc.RecordStaticAddressRiskDecisionParams) error

	// GetStaticAddressLoopInSwap retrieves a loop-in swap by its swap hash.
	GetStaticAddressLoopInSwap(ctx context.Context,
		swapHash []byte) (sqlc.GetStaticAddressLoopInSwapRow, error)

	// GetStaticAddressLoopInSwapsByStates retrieves all swaps with the
	// given states. The states string is comma-separated so the query can
	// match complete state names by wrapping it with comma sentinels.
	GetStaticAddressLoopInSwapsByStates(ctx context.Context,
		states sql.NullString) ([]sqlc.GetStaticAddressLoopInSwapsByStatesRow,
		error)

	// GetLoopInSwapUpdates retrieves all updates for a loop-in swap.
	GetLoopInSwapUpdates(ctx context.Context,
		swapHash []byte) ([]sqlc.StaticAddressSwapUpdate, error)

	// IsStored returns true if a swap with the given hash is stored in the
	// database, false otherwise.
	IsStored(ctx context.Context, swapHash []byte) (bool, error)

	// MapDepositToSwap maps a deposit to a swap in the database.
	MapDepositToSwap(ctx context.Context,
		arg sqlc.MapDepositToSwapParams) error

	// SwapHashForDepositID retrieves the swap hash for the given deposit
	// ID.
	SwapHashForDepositID(ctx context.Context,
		depositID []byte) ([]byte, error)

	// DepositIDsForSwapHash retrieves all deposit IDs for a given swap
	// hash.
	DepositIDsForSwapHash(ctx context.Context,
		swapHash []byte) ([][]byte, error)

	// DepositsForSwapHash retrieves all deposits for a given swap hash.
	DepositsForSwapHash(ctx context.Context,
		swapHash []byte) ([]sqlc.DepositsForSwapHashRow, error)

	// OverrideSelectedSwapAmount updates the selected swap amount for
	// a given swap hash.
	OverrideSelectedSwapAmount(ctx context.Context,
		params sqlc.OverrideSelectedSwapAmountParams) error
}

// BaseDB is the interface that contains all the queries generated by sqlc for
// the static_address_swaps table and transaction functionality.
type BaseDB interface {
	Querier

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

// SqlStore is the backing store for static address loop-ins.
type SqlStore struct {
	baseDB  BaseDB
	clock   clock.Clock
	network *chaincfg.Params
}

// NewSqlStore constructs a new SQLStore from a BaseDB. The BaseDB is agnostic
// to the underlying driver which can be postgres or sqlite.
func NewSqlStore(db BaseDB, clock clock.Clock,
	network *chaincfg.Params) *SqlStore {

	return &SqlStore{
		baseDB:  db,
		clock:   clock,
		network: network,
	}
}

// GetLoopInByHash returns the loop-in swap with the given hash.
func (s *SqlStore) GetLoopInByHash(ctx context.Context,
	swapHash lntypes.Hash) (*StaticAddressLoopIn, error) {

	var (
		err     error
		swap    sqlc.GetStaticAddressLoopInSwapRow
		updates []sqlc.StaticAddressSwapUpdate
	)
	swap, err = s.baseDB.GetStaticAddressLoopInSwap(ctx, swapHash[:])
	if err != nil {
		return nil, err
	}

	deposits, err := s.baseDB.DepositsForSwapHash(ctx, swapHash[:])
	if err != nil {
		return nil, err
	}

	updates, err = s.baseDB.GetLoopInSwapUpdates(ctx, swapHash[:])
	if err != nil {
		return nil, err
	}

	return toStaticAddressLoopIn(ctx, s.network, swap, deposits, updates)
}

// GetStaticAddressLoopInSwapsByStates returns all static address loop-ins from
// the db that are in the given states.
func (s *SqlStore) GetStaticAddressLoopInSwapsByStates(ctx context.Context,
	states []fsm.StateType) ([]*StaticAddressLoopIn, error) {

	var (
		err     error
		rows    []sqlc.GetStaticAddressLoopInSwapsByStatesRow
		updates []sqlc.StaticAddressSwapUpdate
		loopIn  *StaticAddressLoopIn
	)
	joinedStates := toJointStringStates(states)
	joinedNullStringStates := sql.NullString{
		String: joinedStates,
		Valid:  joinedStates != "",
	}
	rows, err = s.baseDB.GetStaticAddressLoopInSwapsByStates(
		ctx, joinedNullStringStates,
	)
	if err != nil {
		return nil, err
	}

	loopIns := make([]*StaticAddressLoopIn, 0, len(rows))
	for _, row := range rows {
		deposits, err := s.baseDB.DepositsForSwapHash(ctx, row.SwapHash)
		if err != nil {
			return nil, err
		}

		updates, err = s.baseDB.GetLoopInSwapUpdates(
			ctx, row.SwapHash,
		)
		if err != nil {
			return nil, err
		}

		loopIn, err = toStaticAddressLoopIn(
			ctx, s.network, sqlc.GetStaticAddressLoopInSwapRow(row),
			deposits, updates,
		)
		if err != nil {
			return nil, err
		}

		loopIns = append(loopIns, loopIn)
	}

	return loopIns, nil
}

func toJointStringStates(states []fsm.StateType) string {
	return strings.Join(toStrings(states), ",")
}

func toStrings(states []fsm.StateType) []string {
	stringStates := make([]string, len(states))
	for i, state := range states {
		stringStates[i] = string(state)
	}

	return stringStates
}

// CreateLoopIn inserts a new loop-in swap into the database. Basic loop-in
// parameters are stored in the swaps table, htlc key information is stored in
// the htlc_keys table, and loop-in specific information is stored in the
// static_address_swaps table.
func (s *SqlStore) CreateLoopIn(ctx context.Context,
	loopIn *StaticAddressLoopIn) error {

	if loopIn == nil {
		return errors.New("loop-in cannot be nil")
	}

	if len(loopIn.Deposits) == 0 {
		return errors.New("loop-in must have at least one deposit")
	}

	amountRequested := int64(loopIn.TotalDepositAmount())
	if loopIn.SelectedAmount > 0 {
		amountRequested = int64(loopIn.SelectedAmount)
	}

	swapArgs := sqlc.InsertSwapParams{
		SwapHash:         loopIn.SwapHash[:],
		Preimage:         loopIn.SwapPreimage[:],
		InitiationTime:   loopIn.InitiationTime,
		AmountRequested:  amountRequested,
		CltvExpiry:       loopIn.HtlcCltvExpiry,
		MaxSwapFee:       int64(loopIn.MaxSwapFee),
		InitiationHeight: int32(loopIn.InitiationHeight),
		ProtocolVersion:  int32(loopIn.ProtocolVersion),
		Label:            loopIn.Label,
	}

	htlcKeyArgs := sqlc.InsertHtlcKeysParams{
		SwapHash:             loopIn.SwapHash[:],
		SenderScriptPubkey:   loopIn.ClientPubkey.SerializeCompressed(),
		ReceiverScriptPubkey: loopIn.ServerPubkey.SerializeCompressed(),
		ClientKeyFamily:      int32(loopIn.HtlcKeyLocator.Family),
		ClientKeyIndex:       int32(loopIn.HtlcKeyLocator.Index),
	}

	// Sanity check, if any of the outpoints contain the outpoint separator.
	// If so, we reject the loop-in to prevent potential issues with
	// parsing.
	for _, outpoint := range loopIn.DepositOutpoints {
		if strings.Contains(outpoint, OutpointSeparator) {
			return ErrInvalidOutpoint
		}
	}

	joinedOutpoints := strings.Join(
		loopIn.DepositOutpoints, OutpointSeparator,
	)
	staticAddressLoopInParams := sqlc.InsertStaticAddressLoopInParams{
		SwapHash:                loopIn.SwapHash[:],
		SwapInvoice:             loopIn.SwapInvoice,
		LastHop:                 loopIn.LastHop,
		QuotedSwapFeeSatoshis:   int64(loopIn.QuotedSwapFee),
		HtlcTimeoutSweepAddress: loopIn.HtlcTimeoutSweepAddress.String(),
		HtlcTxFeeRateSatKw:      int64(loopIn.HtlcTxFeeRate),
		DepositOutpoints:        joinedOutpoints,
		SelectedAmount:          int64(loopIn.SelectedAmount),
		PaymentTimeoutSeconds:   int32(loopIn.PaymentTimeoutSeconds),
		Fast:                    loopIn.Fast,
	}

	updateTime := sqlStoreUpdateTime(s.clock)
	updateArgs := sqlc.InsertStaticAddressMetaUpdateParams{
		SwapHash:        loopIn.SwapHash[:],
		UpdateTimestamp: updateTime,
		UpdateState:     string(loopIn.GetState()),
	}

	err := s.baseDB.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
		func(q Querier) error {
			err := q.InsertSwap(ctx, swapArgs)
			if err != nil {
				return err
			}

			err = q.InsertHtlcKeys(ctx, htlcKeyArgs)
			if err != nil {
				return err
			}

			err = q.InsertStaticAddressLoopIn(
				ctx, staticAddressLoopInParams,
			)
			if err != nil {
				return err
			}

			// Map each deposit to the swap hash in the
			// deposit_to_swap table. This allows us to track which
			// deposits are used for which swaps.
			for _, d := range loopIn.Deposits {
				err = q.MapDepositToSwap(
					ctx, sqlc.MapDepositToSwapParams{
						DepositID: d.ID[:],
						SwapHash:  loopIn.SwapHash[:],
					},
				)
				if err != nil {
					return err
				}
			}

			return q.InsertStaticAddressMetaUpdate(ctx, updateArgs)
		},
	)
	if err != nil {
		return err
	}

	loopIn.LastUpdateTime = updateTime

	return nil
}

// UpdateLoopIn updates the loop-in in the database.
func (s *SqlStore) UpdateLoopIn(ctx context.Context,
	loopIn *StaticAddressLoopIn) error {

	var htlcTimeoutSweepTxID string
	if loopIn.HtlcTimeoutSweepTxHash != nil {
		htlcTimeoutSweepTxID = loopIn.HtlcTimeoutSweepTxHash.String()
	}

	updateParams := sqlc.UpdateStaticAddressLoopInParams{
		SwapHash:           loopIn.SwapHash[:],
		HtlcTxFeeRateSatKw: int64(loopIn.HtlcTxFeeRate),
		HtlcTimeoutSweepTxID: sql.NullString{
			String: htlcTimeoutSweepTxID,
			Valid:  htlcTimeoutSweepTxID != "",
		},
	}

	updateTime := sqlStoreUpdateTime(s.clock)
	updateArgs := sqlc.InsertStaticAddressMetaUpdateParams{
		SwapHash:        loopIn.SwapHash[:],
		UpdateState:     string(loopIn.GetState()),
		UpdateTimestamp: updateTime,
	}

	err := s.baseDB.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
		func(q Querier) error {
			err := q.UpdateStaticAddressLoopIn(ctx, updateParams)
			if err != nil {
				return err
			}

			return q.InsertStaticAddressMetaUpdate(ctx, updateArgs)
		},
	)
	if err != nil {
		return err
	}

	loopIn.LastUpdateTime = updateTime

	return nil
}

// RecordStaticAddressRiskDecision stores the server's confirmation-risk
// decision for a static address loop-in. The timestamp is written by the store
// so recovery can reconstruct the remaining payment deadline from one durable
// clock source.
func (s *SqlStore) RecordStaticAddressRiskDecision(ctx context.Context,
	swapHash lntypes.Hash, decision ConfirmationRiskDecision) error {

	if decision != ConfirmationRiskDecisionAccepted &&
		decision != ConfirmationRiskDecisionRejected {

		return errors.New("unknown confirmation risk decision")
	}

	params := sqlc.RecordStaticAddressRiskDecisionParams{
		SwapHash:                 swapHash[:],
		ConfirmationRiskDecision: string(decision),
		ConfirmationRiskDecisionTime: sql.NullTime{
			Time:  s.clock.Now(),
			Valid: true,
		},
	}

	return s.baseDB.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
		func(q Querier) error {
			stored, err := q.IsStored(ctx, swapHash[:])
			if err != nil {
				return err
			}
			if !stored {
				return ErrLoopInNotFound
			}

			return q.RecordStaticAddressRiskDecision(ctx, params)
		},
	)
}

func (s *SqlStore) BatchUpdateSelectedSwapAmounts(ctx context.Context,
	updateAmounts map[lntypes.Hash]btcutil.Amount) error {

	return s.baseDB.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
		func(q Querier) error {
			for swapHash, amount := range updateAmounts {
				err := q.OverrideSelectedSwapAmount(
					ctx, sqlc.OverrideSelectedSwapAmountParams{
						SwapHash:       swapHash[:],
						SelectedAmount: int64(amount),
					},
				)
				if err != nil {
					return err
				}
			}

			return nil
		})
}

// IsStored returns true if a swap with the given hash is stored in the
// database, false otherwise.
func (s *SqlStore) IsStored(ctx context.Context, swapHash lntypes.Hash) (bool,
	error) {

	return s.baseDB.IsStored(ctx, swapHash[:])
}

// BatchMapDepositsToSwapHashes maps multiple deposits to their respective swap
// hashes in a single transaction.
func (s *SqlStore) BatchMapDepositsToSwapHashes(ctx context.Context,
	depositsToHashes map[deposit.ID]lntypes.Hash) error {

	return s.baseDB.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
		func(q Querier) error {
			for deposit, swapHash := range depositsToHashes {
				err := q.MapDepositToSwap(
					ctx, sqlc.MapDepositToSwapParams{
						DepositID: deposit[:],
						SwapHash:  swapHash[:],
					},
				)
				if err != nil {
					return err
				}
			}

			return nil
		})
}

// SwapHashesForDepositIDs retrieves the swap hashes for the given deposit IDs.
func (s *SqlStore) SwapHashesForDepositIDs(ctx context.Context,
	depositIDs []deposit.ID) (map[lntypes.Hash][]deposit.ID, error) {

	swapHashes := make(map[lntypes.Hash][]deposit.ID)
	for _, id := range depositIDs {
		swapHash, err := s.baseDB.SwapHashForDepositID(ctx, id[:])
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return nil, nil
			}

			return nil, err
		}

		if swapHash == nil {
			return nil, nil
		}

		if len(swapHash) != lntypes.HashSize {
			return nil, errors.New("invalid swap hash length")
		}

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

		// Place the deposit ID in the map under the
		// corresponding swap hash.
		swapHashes[swapHashParsed] = append(
			swapHashes[swapHashParsed], id,
		)
	}

	return swapHashes, nil
}

// DepositIDsForSwapHash retrieves all deposit IDs for a given swap hash.
func (s *SqlStore) DepositIDsForSwapHash(ctx context.Context,
	swapHash lntypes.Hash) ([]deposit.ID, error) {

	byteIDs, err := s.baseDB.DepositIDsForSwapHash(ctx, swapHash[:])
	if err != nil {
		return nil, err
	}

	depositIDs := make([]deposit.ID, len(byteIDs))
	for i, id := range byteIDs {
		if len(id) != deposit.IdLength {
			return nil, errors.New("invalid deposit ID length")
		}
		copy(depositIDs[i][:], id)
	}

	return depositIDs, nil
}

// toStaticAddressLoopIn converts sql rows to an instant out struct.
func toStaticAddressLoopIn(_ context.Context, network *chaincfg.Params,
	swap sqlc.GetStaticAddressLoopInSwapRow,
	deposits []sqlc.DepositsForSwapHashRow,
	updates []sqlc.StaticAddressSwapUpdate) (*StaticAddressLoopIn, error) {

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

	swapPreImage, err := lntypes.MakePreimage(swap.Preimage)
	if err != nil {
		return nil, err
	}

	clientKey, err := btcec.ParsePubKey(swap.SenderScriptPubkey)
	if err != nil {
		return nil, err
	}

	serverKey, err := btcec.ParsePubKey(swap.ReceiverScriptPubkey)
	if err != nil {
		return nil, err
	}

	var htlcTimeoutSweepTxHash *chainhash.Hash
	if swap.HtlcTimeoutSweepTxID.Valid {
		htlcTimeoutSweepTxHash, err = chainhash.NewHashFromStr(
			swap.HtlcTimeoutSweepTxID.String,
		)
		if err != nil {
			return nil, err
		}
	}

	var depositOutpoints []string
	if swap.DepositOutpoints != "" {
		depositOutpoints = strings.Split(
			swap.DepositOutpoints, OutpointSeparator,
		)
	}

	timeoutAddressString := swap.HtlcTimeoutSweepAddress
	var timeoutAddress btcutil.Address
	if timeoutAddressString != "" {
		timeoutAddress, err = btcutil.DecodeAddress(
			timeoutAddressString, network,
		)
		if err != nil {
			return nil, err
		}
	}

	depositList := make([]*deposit.Deposit, 0, len(deposits))
	for _, d := range deposits {
		id := deposit.ID{}
		err = id.FromByteSlice(d.DepositID)
		if err != nil {
			return nil, err
		}

		sqlcDeposit := sqlc.AllDepositsRow{
			DepositID:             id[:],
			TxHash:                d.TxHash,
			Amount:                d.Amount,
			OutIndex:              d.OutIndex,
			ConfirmationHeight:    d.ConfirmationHeight,
			TimeoutSweepPkScript:  d.TimeoutSweepPkScript,
			ExpirySweepTxid:       d.ExpirySweepTxid,
			FinalizedWithdrawalTx: d.FinalizedWithdrawalTx,
		}

		sqlcDepositUpdate := sqlc.DepositUpdate{
			DepositID:       id[:],
			UpdateState:     d.UpdateState.String,
			UpdateTimestamp: d.UpdateTimestamp.Time,
		}
		deposit, err := deposit.ToDeposit(
			sqlcDeposit, sqlcDepositUpdate,
		)
		if err != nil {
			return nil, err
		}

		depositList = append(depositList, deposit)
	}
	depositList = orderDepositsBySnapshot(depositList, depositOutpoints)

	loopIn := &StaticAddressLoopIn{
		SwapHash:         swapHash,
		SwapPreimage:     swapPreImage,
		HtlcCltvExpiry:   swap.CltvExpiry,
		MaxSwapFee:       btcutil.Amount(swap.MaxSwapFee),
		InitiationHeight: uint32(swap.InitiationHeight),
		InitiationTime:   swap.InitiationTime,
		ProtocolVersion: version.AddressProtocolVersion(
			swap.ProtocolVersion,
		),
		Label:        swap.Label,
		ClientPubkey: clientKey,
		ServerPubkey: serverKey,
		HtlcKeyLocator: keychain.KeyLocator{
			Family: keychain.KeyFamily(swap.ClientKeyFamily),
			Index:  uint32(swap.ClientKeyIndex),
		},
		SwapInvoice:           swap.SwapInvoice,
		PaymentTimeoutSeconds: uint32(swap.PaymentTimeoutSeconds),
		LastHop:               swap.LastHop,
		QuotedSwapFee:         btcutil.Amount(swap.QuotedSwapFeeSatoshis),
		DepositOutpoints:      depositOutpoints,
		SelectedAmount:        btcutil.Amount(swap.SelectedAmount),
		Fast:                  swap.Fast,
		ConfirmationRiskDecision: ConfirmationRiskDecision(
			swap.ConfirmationRiskDecision,
		),
		HtlcTxFeeRate: chainfee.SatPerKWeight(
			swap.HtlcTxFeeRateSatKw,
		),
		HtlcTimeoutSweepAddress: timeoutAddress,
		HtlcTimeoutSweepTxHash:  htlcTimeoutSweepTxHash,
		Deposits:                depositList,
	}
	if swap.ConfirmationRiskDecisionTime.Valid {
		loopIn.ConfirmationRiskDecisionTime =
			swap.ConfirmationRiskDecisionTime.Time
	}

	if len(updates) > 0 {
		lastUpdate := updates[len(updates)-1]
		loopIn.SetState(fsm.StateType(lastUpdate.UpdateState))
		loopIn.LastUpdateTime = lastUpdate.UpdateTimestamp
	}

	return loopIn, nil
}

// orderDepositsBySnapshot returns deposits ordered by the stored outpoint
// snapshot when the current deposit rows still match that snapshot. If any
// snapshot outpoint no longer maps to a current deposit row, recovery keeps the
// store reconstruction untouched so callers can handle the divergence.
func orderDepositsBySnapshot(deposits []*deposit.Deposit,
	depositOutpoints []string) []*deposit.Deposit {

	if len(deposits) != len(depositOutpoints) {
		return deposits
	}

	byOutpoint := make(map[string]*deposit.Deposit, len(deposits))
	for _, d := range deposits {
		outpoint := d.OutPoint.String()
		if _, ok := byOutpoint[outpoint]; ok {
			return deposits
		}

		byOutpoint[outpoint] = d
	}

	orderedDeposits := make([]*deposit.Deposit, len(depositOutpoints))
	for i, outpoint := range depositOutpoints {
		d, ok := byOutpoint[outpoint]
		if !ok {
			return deposits
		}

		orderedDeposits[i] = d
	}

	return orderedDeposits
}