staticaddr: normalize swapped deposits

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.MigrateDepositNormalization(
+		d.mainCtx, swapDb, depositStore, staticAddressLoopInStore,
+	)
+	if err != nil {
+		errorf("Deposit normalization migration failed: %v", err)
+
+		return err
+	}
+
 	staticLoopInManager = loopin.NewManager(&loopin.Config{
 		Server:                               staticAddressClient,
 		QuoteGetter:                          swapClient.Server,

loopdb/sqlc/migrations/000017_deposit_to_swap.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS deposit_to_swap;

loopdb/sqlc/migrations/000017_deposit_to_swap.up.sql

@@ -0,0 +1,8 @@
+-- deposit_to_swap table maps deposits to swaps.
+CREATE TABLE IF NOT EXISTS deposit_to_swap (
+    -- deposit_id is a unique identifier from the deposits table.
+    deposit_id BLOB NOT NULL UNIQUE REFERENCES deposits(deposit_id),
+
+    -- swap_hash is the identifier from static_address_swaps.
+    swap_hash BLOB NOT NULL REFERENCES static_address_swaps(swap_hash)
+);

loopdb/sqlc/models.go

@@ -93,3 +93,31 @@ SELECT EXISTS (
     FROM static_address_swaps
     WHERE swap_hash = $1
 );
+
+-- name: MapDepositToSwap :exec
+INSERT INTO deposit_to_swap (
+    deposit_id,
+    swap_hash
+) VALUES (
+       $1,
+       $2
+);
+
+-- name: SwapHashForDepositID :one
+SELECT
+    swap_hash
+FROM
+    deposit_to_swap
+WHERE
+    deposit_id = $1;
+
+
+-- name: DepositIDsForSwapHash :many
+SELECT
+    deposit_id
+FROM
+    deposit_to_swap
+WHERE
+    swap_hash = $1;
+
+

loopdb/sqlc/querier.go

@@ -0,0 +1,84 @@
+package loopin
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/lightninglabs/loop/loopdb"
+	"github.com/lightninglabs/loop/staticaddr/deposit"
+	"github.com/lightningnetwork/lnd/lntypes"
+)
+
+const (
+	// depositNormalizationMigrationID is the identifier for the deposit
+	// normalization migration.
+	depositNormalizationMigrationID = "deposit_normalization"
+)
+
+// MigrateDepositNormalization will retrieve the comma separated deposit list of
+// past and pending swaps and map them to the swap hash in table
+// deposit_to_swap.
+func MigrateDepositNormalization(ctx context.Context, db loopdb.SwapStore,
+	depositStore *deposit.SqlStore, swapStore *SqlStore) error {
+
+	migrationDone, err := db.HasMigration(
+		ctx, depositNormalizationMigrationID,
+	)
+	if err != nil {
+		return fmt.Errorf("unable to check migration status: %w", err)
+	}
+	if migrationDone {
+		log.Infof("Deposit normalization migration already done, " +
+			"skipping")
+
+		return nil
+	}
+
+	log.Infof("Starting deposit normalization migration")
+	startTs := time.Now()
+	defer func() {
+		log.Infof("Finished deposit normalization migration in %v",
+			time.Since(startTs))
+	}()
+
+	// First we'll fetch all past and pending static loop in swaps from the
+	// database.
+	swaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(
+		ctx, append(PendingStates, FinalStates...),
+	)
+	if err != nil {
+		return err
+	}
+
+	// Now we'll map each deposit of a swap to its respective swap hash.
+	depositsToSwapHashes := make(map[deposit.ID]lntypes.Hash)
+	for _, swap := range swaps {
+		for _, outpoint := range swap.DepositOutpoints {
+			deposit, err := depositStore.DepositForOutpoint(
+				ctx, outpoint,
+			)
+			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)
+			}
+
+			depositsToSwapHashes[deposit.ID] = swap.SwapHash
+		}
+	}
+
+	log.Infof("Batch-mapping %d deposits to swap hashes",
+		len(depositsToSwapHashes))
+
+	err = swapStore.BatchMapDepositsToSwapHashes(ctx, depositsToSwapHashes)
+	if err != nil {
+		return err
+	}
+
+	// Finally mark the migration as done.
+	return db.SetMigration(ctx, depositNormalizationMigrationID)
+}