staticaddr/loopin: add risk decision watcher

staticaddr/loopin/risk_watcher.go

@@ -0,0 +1,182 @@
+package loopin
+
+import (
+	"bytes"
+	"context"
+	"time"
+
+	"github.com/lightningnetwork/lnd/lntypes"
+)
+
+// confirmationRiskUpdate is the normalized result of a server confirmation-risk
+// notification.
+type confirmationRiskUpdate struct {
+	decision ConfirmationRiskDecision
+	reason   string
+}
+
+// confirmationRiskWatcher normalizes static loop-in confirmation risk
+// notifications and restores the durable decision timestamp.
+type confirmationRiskWatcher struct {
+	swapHash            lntypes.Hash
+	store               StaticAddressLoopInStore
+	notificationManager NotificationManager
+	logWarnf            func(string, ...any)
+}
+
+// newConfirmationRiskWatcher creates a helper that handles confirmation-risk
+// notification plumbing for a single static loop-in swap.
+func newConfirmationRiskWatcher(cfg *Config, swapHash lntypes.Hash,
+	warnf func(string, ...any)) *confirmationRiskWatcher {
+
+	return &confirmationRiskWatcher{
+		swapHash:            swapHash,
+		store:               cfg.Store,
+		notificationManager: cfg.NotificationManager,
+		logWarnf:            warnf,
+	}
+}
+
+// warnf logs through the FSM-scoped logger when one is available.
+func (w *confirmationRiskWatcher) warnf(format string, args ...any) {
+	if w.logWarnf != nil {
+		w.logWarnf(format, args...)
+
+		return
+	}
+
+	log.Warnf(format, args...)
+}
+
+// subscribe subscribes to accepted and rejected confirmation-risk notifications
+// and emits normalized updates for the watcher's swap hash.
+func (w *confirmationRiskWatcher) subscribe(ctx context.Context) (
+	<-chan confirmationRiskUpdate, func()) {
+
+	if w.notificationManager == nil {
+		return nil, func() {}
+	}
+
+	notificationCtx, cancel := context.WithCancel(ctx)
+	riskAcceptedChan := w.notificationManager.SubscribeStaticLoopInRiskAccepted(
+		notificationCtx, w.swapHash,
+	)
+	riskRejectedChan := w.notificationManager.SubscribeStaticLoopInRiskRejected(
+		notificationCtx, w.swapHash,
+	)
+
+	riskUpdates := make(chan confirmationRiskUpdate, 1)
+	go func() {
+		defer close(riskUpdates)
+
+		for {
+			select {
+			case riskAccepted, ok := <-riskAcceptedChan:
+				if !ok {
+					riskAcceptedChan = nil
+					continue
+				}
+
+				if riskAccepted == nil || !bytes.Equal(
+					riskAccepted.SwapHash, w.swapHash[:],
+				) {
+
+					continue
+				}
+
+				update := confirmationRiskUpdate{
+					decision: ConfirmationRiskDecisionAccepted,
+					reason:   "risk accepted notification",
+				}
+				select {
+				case riskUpdates <- update:
+				case <-notificationCtx.Done():
+					return
+				}
+
+			case riskRejected, ok := <-riskRejectedChan:
+				if !ok {
+					riskRejectedChan = nil
+					continue
+				}
+
+				if riskRejected == nil || !bytes.Equal(
+					riskRejected.SwapHash, w.swapHash[:],
+				) {
+
+					continue
+				}
+
+				update := confirmationRiskUpdate{
+					decision: ConfirmationRiskDecisionRejected,
+					reason:   "risk rejection",
+				}
+				select {
+				case riskUpdates <- update:
+				case <-notificationCtx.Done():
+					return
+				}
+
+			case <-notificationCtx.Done():
+				return
+			}
+		}
+	}()
+
+	return riskUpdates, cancel
+}
+
+// decisionTime returns the durable decision timestamp, recording the decision
+// first if the notification was replayed before it could be persisted.
+func (w *confirmationRiskWatcher) decisionTime(ctx context.Context,
+	decision ConfirmationRiskDecision) time.Time {
+
+	now := time.Now()
+	if w.store == nil {
+		return now
+	}
+
+	storedLoopIn, err := w.store.GetLoopInByHash(ctx, w.swapHash)
+	if err != nil {
+		w.warnf("unable to reload persisted risk decision for swap %v: %v",
+			w.swapHash, err)
+
+		return now
+	}
+
+	if storedLoopIn == nil {
+		return now
+	}
+
+	hasPersistedDecision :=
+		storedLoopIn.ConfirmationRiskDecision == decision &&
+			!storedLoopIn.ConfirmationRiskDecisionTime.IsZero()
+
+	if !hasPersistedDecision {
+		err = w.store.RecordStaticAddressRiskDecision(
+			ctx, w.swapHash, decision,
+		)
+		if err != nil {
+			w.warnf("unable to persist replayed risk decision for "+
+				"swap %v: %v", w.swapHash, err)
+
+			return now
+		}
+
+		storedLoopIn, err = w.store.GetLoopInByHash(ctx, w.swapHash)
+		if err != nil {
+			w.warnf("unable to reload persisted risk decision for "+
+				"swap %v: %v", w.swapHash, err)
+
+			return now
+		}
+		if storedLoopIn == nil ||
+			storedLoopIn.ConfirmationRiskDecision != decision ||
+			storedLoopIn.ConfirmationRiskDecisionTime.IsZero() {
+
+			return now
+		}
+	}
+
+	return storedLoopIn.ConfirmationRiskDecisionTime
+}

staticaddr/loopin/risk_watcher_test.go

@@ -0,0 +1,121 @@
+package loopin
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/stretchr/testify/require"
+)
+
+// TestConfirmationRiskWatcherSubscribeFiltersSwapHash verifies that the watcher
+// only emits normalized decisions for the swap it was created for.
+func TestConfirmationRiskWatcherSubscribeFiltersSwapHash(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+
+	swapHash := lntypes.Hash{1, 2, 3}
+	otherHash := lntypes.Hash{3, 2, 1}
+	notificationMgr := &mockNotificationManager{
+		riskAccepted: make(
+			chan *swapserverrpc.
+				ServerStaticLoopInRiskAcceptedNotification, 1,
+		),
+		riskRejected: make(
+			chan *swapserverrpc.
+				ServerStaticLoopInRiskRejectedNotification, 1,
+		),
+	}
+
+	watcher := newConfirmationRiskWatcher(
+		&Config{NotificationManager: notificationMgr}, swapHash,
+		t.Logf,
+	)
+	updates, stop := watcher.subscribe(ctx)
+	defer stop()
+
+	notificationMgr.riskAccepted <- &swapserverrpc.
+		ServerStaticLoopInRiskAcceptedNotification{
+		SwapHash: otherHash[:],
+	}
+
+	select {
+	case update := <-updates:
+		t.Fatalf("received wrong-hash risk update: %v", update)
+
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	notificationMgr.riskRejected <- &swapserverrpc.
+		ServerStaticLoopInRiskRejectedNotification{
+		SwapHash: swapHash[:],
+	}
+
+	select {
+	case update := <-updates:
+		require.Equal(t, ConfirmationRiskDecisionRejected,
+			update.decision)
+		require.Equal(t, "risk rejection", update.reason)
+
+	case <-ctx.Done():
+		t.Fatalf("risk update not received: %v", ctx.Err())
+	}
+}
+
+// TestConfirmationRiskWatcherDecisionTimeRestoration verifies that the watcher
+// preserves existing persisted decision timestamps and records missing ones.
+func TestConfirmationRiskWatcherDecisionTimeRestoration(t *testing.T) {
+	t.Parallel()
+
+	ctx := t.Context()
+	swapHash := lntypes.Hash{4, 5, 6}
+	decisionTime := time.Unix(123, 0).UTC()
+	store := &recordingRiskStore{
+		mockStore: &mockStore{
+			loopIns: map[lntypes.Hash]*StaticAddressLoopIn{
+				swapHash: {
+					ConfirmationRiskDecision:     ConfirmationRiskDecisionAccepted,
+					ConfirmationRiskDecisionTime: decisionTime,
+				},
+			},
+		},
+		decisions: make(chan ConfirmationRiskDecision, 1),
+	}
+
+	watcher := newConfirmationRiskWatcher(&Config{Store: store}, swapHash,
+		t.Logf)
+	restoredTime := watcher.decisionTime(
+		ctx, ConfirmationRiskDecisionAccepted,
+	)
+	require.True(t, restoredTime.Equal(decisionTime))
+
+	select {
+	case decision := <-store.decisions:
+		t.Fatalf("persisted already-recorded decision: %v", decision)
+
+	default:
+	}
+
+	store.loopIns[swapHash] = &StaticAddressLoopIn{}
+	recordedTime := watcher.decisionTime(
+		ctx, ConfirmationRiskDecisionRejected,
+	)
+	require.False(t, recordedTime.IsZero())
+
+	select {
+	case decision := <-store.decisions:
+		require.Equal(t, ConfirmationRiskDecisionRejected, decision)
+
+	case <-time.After(time.Second):
+		t.Fatal("missing risk decision was not persisted")
+	}
+	require.Equal(t, ConfirmationRiskDecisionRejected,
+		store.loopIns[swapHash].ConfirmationRiskDecision)
+	require.True(t, recordedTime.Equal(
+		store.loopIns[swapHash].ConfirmationRiskDecisionTime,
+	))
+}