staticaddr: selected swap amount migration

Author: hieblmi

loopd/daemon.go

@@ -626,6 +626,16 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		clock.NewDefaultClock(), d.lnd.ChainParams,
 	)
 
+	// Run the selected amount migration.
+	err = loopin.MigrateSelectedSwapAmount(
+		d.mainCtx, swapDb, depositStore, staticAddressLoopInStore,
+	)
+	if err != nil {
+		errorf("Selected amount migration failed: %v", err)
+
+		return err
+	}
+
 	staticLoopInManager = loopin.NewManager(&loopin.Config{
 		Server:                               staticAddressClient,
 		QuoteGetter:                          swapClient.Server,

loopdb/sqlc/querier.go

@@ -95,3 +95,9 @@ SELECT EXISTS (
     FROM static_address_swaps
     WHERE swap_hash = $1
 );
+
+-- name: OverrideSelectedSwapAmount :exec
+UPDATE static_address_swaps
+SET
+    selected_amount = $2
+WHERE swap_hash = $1;

loopdb/sqlc/queries/static_address_loopin.sql

@@ -0,0 +1,99 @@
+package loopin
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/btcsuite/btcd/btcutil"
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/lightninglabs/loop/loopdb"
+	"github.com/lightninglabs/loop/staticaddr/deposit"
+	"github.com/lightningnetwork/lnd/lntypes"
+)
+
+const (
+	// selectedAmountMigrationID is the identifier for the selected swap
+	// amount migration.
+	selectedAmountMigrationID = "selected_amount"
+)
+
+// MigrateSelectedSwapAmount will update the selected swap amount of past swaps
+// with the sum of the values of the deposits they swapped.
+func MigrateSelectedSwapAmount(ctx context.Context, db loopdb.SwapStore,
+	depositStore *deposit.SqlStore, swapStore *SqlStore) error {
+
+	migrationDone, err := db.HasMigration(ctx, selectedAmountMigrationID)
+	if err != nil {
+		return fmt.Errorf("unable to check migration status: %w", err)
+	}
+	if migrationDone {
+		log.Infof("Selected swap amount migration already done, " +
+			"skipping")
+
+		return nil
+	}
+
+	log.Infof("Starting swap amount migration")
+	startTs := time.Now()
+	defer func() {
+		log.Infof("Finished swap amount migration in %v",
+			time.Since(startTs))
+	}()
+
+	// First we'll fetch all loop out swaps from the database.
+	swaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(
+		ctx, FinalStates,
+	)
+	if err != nil {
+		return err
+	}
+
+	// Now we'll calculate the cost for each swap and finally update the
+	// costs in the database.
+	// TODO(hieblmi): normalize swap hash and deposit ids.
+	updateAmounts := make(map[lntypes.Hash]btcutil.Amount)
+	for _, swap := range swaps {
+		for _, outpoint := range swap.DepositOutpoints {
+			op := strings.Split(outpoint, ":")
+			hash, err := chainhash.NewHashFromStr(op[0])
+			if err != nil {
+				return fmt.Errorf("unable to parse outpoint "+
+					"hash: %w", err)
+			}
+
+			idx, err := strconv.ParseUint(op[1], 10, 32)
+			if err != nil {
+				return err
+			}
+
+			deposit, err := depositStore.DepositForOutpoint(
+				ctx, *hash, uint32(idx),
+			)
+			if err != nil {
+				return fmt.Errorf("unable to fetch deposit "+
+					"for outpoint %s: %w", outpoint, err)
+			}
+			if deposit == nil {
+				return fmt.Errorf("deposit for outpoint %s "+
+					"not found", outpoint)
+			}
+
+			// Set the selected amount to the value of the deposit.
+			updateAmounts[swap.SwapHash] += deposit.Value
+		}
+	}
+
+	log.Infof("Updating selected swap amounts for %d loop out swaps",
+		len(updateAmounts))
+
+	err = swapStore.BatchUpdateSelectedSwapAmounts(ctx, updateAmounts)
+	if err != nil {
+		return err
+	}
+
+	// Finally mark the migration as done.
+	return db.SetMigration(ctx, selectedAmountMigrationID)
+}

loopdb/sqlc/static_address_loopin.sql.go

@@ -0,0 +1,106 @@
+package loopin
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"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/staticaddr/deposit"
+	"github.com/lightninglabs/loop/test"
+	"github.com/lightningnetwork/lnd/clock"
+	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/stretchr/testify/require"
+)
+
+// TestMigrateSelectedSwapAmount tests the selected amount migration.
+func TestMigrateSelectedSwapAmount(t *testing.T) {
+	// Set up test context objects.
+	ctxb := context.Background()
+	testDb := loopdb.NewTestDB(t)
+	testClock := clock.NewTestClock(time.Now())
+	defer testDb.Close()
+
+	db := loopdb.NewStoreMock(t)
+	depositStore := deposit.NewSqlStore(testDb.BaseDB)
+	swapStore := NewSqlStore(
+		loopdb.NewTypedStore[Querier](testDb), testClock,
+		&chaincfg.MainNetParams,
+	)
+
+	newID := func() deposit.ID {
+		did, err := deposit.GetRandomDepositID()
+		require.NoError(t, err)
+
+		return did
+	}
+
+	d1, d2 := &deposit.Deposit{
+		ID: newID(),
+		OutPoint: wire.OutPoint{
+			Hash:  chainhash.Hash{0x1a, 0x2b, 0x3c, 0x4d},
+			Index: 0,
+		},
+		Value: btcutil.Amount(100_000),
+		TimeOutSweepPkScript: []byte{
+			0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x41,
+		},
+	},
+		&deposit.Deposit{
+			ID: newID(),
+			OutPoint: wire.OutPoint{
+				Hash:  chainhash.Hash{0x2a, 0x2b, 0x3c, 0x4e},
+				Index: 1,
+			},
+			Value: btcutil.Amount(200_000),
+			TimeOutSweepPkScript: []byte{
+				0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4d,
+			},
+		}
+
+	err := depositStore.CreateDeposit(ctxb, d1)
+	require.NoError(t, err)
+	err = depositStore.CreateDeposit(ctxb, d2)
+	require.NoError(t, err)
+
+	outpoints := []string{
+		d1.OutPoint.String(),
+		d2.OutPoint.String(),
+	}
+	_, clientPubKey := test.CreateKey(1)
+	_, serverPubKey := test.CreateKey(2)
+	p2wkhAddr := "bcrt1qq68r6ff4k4pjx39efs44gcyccf7unqnu5qtjjz"
+	addr, err := btcutil.DecodeAddress(p2wkhAddr, nil)
+	require.NoError(t, err)
+
+	swap := StaticAddressLoopIn{
+		SwapHash:                lntypes.Hash{0x1, 0x2, 0x3, 0x4},
+		DepositOutpoints:        outpoints,
+		ClientPubkey:            clientPubKey,
+		ServerPubkey:            serverPubKey,
+		HtlcTimeoutSweepAddress: addr,
+	}
+	swap.SetState(Succeeded)
+
+	err = swapStore.CreateLoopIn(ctxb, &swap)
+	require.NoError(t, err)
+
+	storedSwaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(
+		ctxb, FinalStates,
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, 0, storedSwaps[0].SelectedAmount)
+
+	err = MigrateSelectedSwapAmount(ctxb, db, depositStore, swapStore)
+	require.NoError(t, err)
+
+	storedSwaps, err = swapStore.GetStaticAddressLoopInSwapsByStates(
+		ctxb, FinalStates,
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, d1.Value+d2.Value, storedSwaps[0].SelectedAmount)
+}

staticaddr/loopin/selected_amount_migration.go

@@ -20,7 +20,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
 )
 
-const outpointSeparator = ";"
+const OutpointSeparator = ";"
 
 var (
 	// ErrInvalidOutpoint is returned when an outpoint contains the outpoint
@@ -68,6 +68,11 @@ type Querier interface {
 	// 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)
+
+	// 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
@@ -218,13 +223,13 @@ func (s *SqlStore) CreateLoopIn(ctx context.Context,
 	// If so, we reject the loop-in to prevent potential issues with
 	// parsing.
 	for _, outpoint := range loopIn.DepositOutpoints {
-		if strings.Contains(outpoint, outpointSeparator) {
+		if strings.Contains(outpoint, OutpointSeparator) {
 			return ErrInvalidOutpoint
 		}
 	}
 
 	joinedOutpoints := strings.Join(
-		loopIn.DepositOutpoints, outpointSeparator,
+		loopIn.DepositOutpoints, OutpointSeparator,
 	)
 	staticAddressLoopInParams := sqlc.InsertStaticAddressLoopInParams{
 		SwapHash:                loopIn.SwapHash[:],
@@ -303,6 +308,27 @@ func (s *SqlStore) UpdateLoopIn(ctx context.Context,
 	)
 }
 
+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,
@@ -347,7 +373,7 @@ func toStaticAddressLoopIn(_ context.Context, network *chaincfg.Params,
 	}
 
 	depositOutpoints := strings.Split(
-		row.DepositOutpoints, outpointSeparator,
+		row.DepositOutpoints, OutpointSeparator,
 	)
 
 	timeoutAddressString := row.HtlcTimeoutSweepAddress