From e7c3717886d75237ed40d74aa448ff02cb3e0854 Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Wed, 30 Oct 2024 13:20:52 +0100 Subject: [PATCH] staticaddr: sql_store --- staticaddr/loopin/sql_store.go | 370 +++++++++++++++++++++++++++++++++ 1 file changed, 370 insertions(+) create mode 100644 staticaddr/loopin/sql_store.go diff --git a/staticaddr/loopin/sql_store.go b/staticaddr/loopin/sql_store.go new file mode 100644 index 00000000..06caae63 --- /dev/null +++ b/staticaddr/loopin/sql_store.go @@ -0,0 +1,370 @@ +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 +}