loop/staticaddr/loopin/sql_store.go

package loopin

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

	"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/version"
	"github.com/lightningnetwork/lnd/clock"
	"github.com/lightningnetwork/lnd/keychain"
	"github.com/lightningnetwork/lnd/lntypes"
	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
)

const outpointSeparator = ";"

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

// 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

	// 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 an input for the IN primitive in
	// sqlite, hence the format needs to be '{State1,State2,...}'.
	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)
}

// 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,
	}
}

// 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 {
		updates, err = s.baseDB.GetLoopInSwapUpdates(
			ctx, row.SwapHash,
		)
		if err != nil {
			return nil, err
		}

		loopIn, err = toStaticAddressLoopIn(
			ctx, s.network, sqlc.GetStaticAddressLoopInSwapRow(row),
			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 {

	swapArgs := sqlc.InsertSwapParams{
		SwapHash:         loopIn.SwapHash[:],
		Preimage:         loopIn.SwapPreimage[:],
		InitiationTime:   loopIn.InitiationTime,
		AmountRequested:  int64(loopIn.TotalDepositAmount()),
		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,
		PaymentTimeoutSeconds:   int32(loopIn.PaymentTimeoutSeconds),
	}

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

	return 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
			}

			return q.InsertStaticAddressMetaUpdate(ctx, updateArgs)
		})
}

// 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 != "",
		},
	}

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

	return 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)
		},
	)
}

// 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[:])
}

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

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

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

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

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

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

	depositOutpoints := strings.Split(
		row.DepositOutpoints, outpointSeparator,
	)

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

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

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

	return loopIn, nil
}