assets+loopdb: publish cooperative deposit withdrawal transaction

This commit implements the full cooperative deposit withdrawal
flow. The client first fetches keys for any pending withdrawals,
then publishes sweep transactions using the revealed key to sign
the deposit sweep. Once the sweep confirms, the deposit’s state
is updated in the deposit store.

Author: bhandras

assets/deposit/manager.go

@@ -173,6 +173,11 @@ func (m *Manager) Run(ctx context.Context, bestBlock uint32) error {
 		return err
 	}
 
+	// Wake the manager up very 10 seconds to check if there're any pending
+	// chores to do.
+	const wakeupInterval = time.Duration(10) * time.Second
+	withdrawTicker := time.NewTicker(wakeupInterval)
+
 	for {
 		select {
 		case <-m.callEnter:
@@ -192,6 +197,15 @@ func (m *Manager) Run(ctx context.Context, bestBlock uint32) error {
 				return err
 			}
 
+		case <-withdrawTicker.C:
+			err := m.publishPendingWithdrawals(ctx)
+			if err != nil {
+				log.Errorf("Unable to publish pending "+
+					"withdrawals: %v", err)
+
+				return err
+			}
+
 		case err := <-blockErrChan:
 			log.Errorf("received error from block epoch "+
 				"notification: %v", err)
@@ -985,6 +999,8 @@ func (m *Manager) handleDepositSpend(ctx context.Context, d *Deposit,
 
 	switch d.State {
 	case StateTimeoutSweepPublished:
+		fallthrough
+	case StateCooperativeSweepPublished:
 		d.State = StateSwept
 
 		err := m.releaseDepositSweepInputs(ctx, d)
@@ -1126,3 +1142,73 @@ func (m *Manager) WithdrawDeposits(ctx context.Context,
 
 	return nil
 }
+
+// publishPendingWithdrawals publishes any pending deposit withdrawals.
+func (m *Manager) publishPendingWithdrawals(ctx context.Context) error {
+	for _, d := range m.deposits {
+		// TODO(bhandras): republish on StateCooperativeSweepPublished.
+		if d.State != StateWithdrawn {
+			continue
+		}
+
+		serverKey, err := m.store.GetAssetDepositServerKey(
+			ctx, d.ID,
+		)
+		if err != nil {
+			return err
+		}
+
+		lockID, err := d.lockID()
+		if err != nil {
+			return err
+		}
+
+		// TODO(bhandras): conf target should be dynamic/configrable.
+		const confTarget = 2
+		feeRateSatPerKw, err := m.walletKit.EstimateFeeRate(
+			ctx, confTarget,
+		)
+		if err != nil {
+			return err
+		}
+
+		funder := true
+		sendResp, err := m.sweeper.PublishDepositSweepMuSig2(
+			ctx, d.Kit, funder, d.Proof, serverKey,
+			asset.NewScriptKey(d.SweepScriptKey),
+			d.SweepInternalKey, d.withdrawLabel(),
+			feeRateSatPerKw.FeePerVByte(), lockID, lockExpiration,
+		)
+		if err != nil {
+			log.Errorf("Unable to publish deposit sweep for %v: %v",
+				d.ID, err)
+		} else {
+			log.Infof("Published sweep for deposit %v: %v", d.ID,
+				sendResp.Transfer.AnchorTxHash)
+
+			d.State = StateCooperativeSweepPublished
+			err = m.handleDepositStateUpdate(ctx, d)
+			if err != nil {
+				log.Errorf("Unable to update deposit %v "+
+					"state: %v", d.ID, err)
+
+				return err
+			}
+		}
+
+		// Start monitoring the sweep unless we're already doing so.
+		if _, ok := m.pendingSweeps[d.ID]; !ok {
+			err := m.waitForDepositSweep(ctx, d, d.withdrawLabel())
+			if err != nil {
+				log.Errorf("Unable to wait for deposit %v "+
+					"spend: %v", d.ID, err)
+
+				return err
+			}
+
+			m.pendingSweeps[d.ID] = struct{}{}
+		}
+	}
+
+	return nil
+}

assets/deposit/manager_test.go

@@ -54,6 +54,16 @@ func (s *mockStore) SetAssetDepositServerKey(context.Context, string,
 	return nil
 }
 
+// GetAssetDepositServerKey is a mock implementation of the
+// GetAssetDepositServerKey method.
+func (s *mockStore) GetAssetDepositServerKey(context.Context, string) (
+	*btcec.PrivateKey, error) {
+
+	// Return a dummy private key for testing
+	privKey, _ := btcec.NewPrivateKey()
+	return privKey, nil
+}
+
 // testAddDeposit is a helper function that (intrusively) adds a deposit to the
 // manager.
 func testAddDeposit(t *testing.T, m *Manager, d *Deposit) {

assets/deposit/sql_store.go

@@ -38,6 +38,9 @@ type Querier interface {
 
 	SetAssetDepositServerInternalKey(ctx context.Context,
 		arg sqlc.SetAssetDepositServerInternalKeyParams) error
+
+	GetAssetDepositServerInternalKey(ctx context.Context,
+		depositID string) ([]byte, error)
 }
 
 // DepositBaseDB is the interface that contains all the queries generated
@@ -329,3 +332,19 @@ func (s *SQLStore) SetAssetDepositServerKey(ctx context.Context,
 		},
 	)
 }
+
+func (s *SQLStore) GetAssetDepositServerKey(ctx context.Context,
+	depositID string) (*btcec.PrivateKey, error) {
+
+	keyBytes, err := s.db.GetAssetDepositServerInternalKey(ctx, depositID)
+	if err != nil {
+		return nil, err
+	}
+
+	key, _ := btcec.PrivKeyFromBytes(keyBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	return key, nil
+}

assets/deposit/store.go

@@ -26,4 +26,9 @@ type Store interface {
 	// asset deposit.
 	SetAssetDepositServerKey(ctx context.Context, depositID string,
 		key *btcec.PrivateKey) error
+
+	// GetAssetDepositServerKey gets the server's internal key for the given
+	// asset deposit.
+	GetAssetDepositServerKey(ctx context.Context, depositID string) (
+		*btcec.PrivateKey, error)
 }

loopdb/sqlc/asset_deposits.sql.go

@@ -64,3 +64,8 @@ UPDATE asset_deposits
 SET server_internal_key = $2
 WHERE deposit_id = $1
 AND server_internal_key IS NULL;
+
+-- name: GetAssetDepositServerInternalKey :one
+SELECT server_internal_key
+FROM asset_deposits
+WHERE deposit_id = $1;