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.Deposit{ 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 }