Recover stranded payjoin stack (pj/01-06)

FEATURES.md

@@ -45,13 +45,16 @@ graph TB
     PAY[Pay]
     BUY[Buy]
     COINS[Coins / UTXOs]
+    ANNOUNCEMENTS[Announcements]
     CONSOLIDATION[Consolidation]
 
     %% Dependencies to Core (all features depend on Core, but showing it explicitly would clutter the diagram)
     %% Instead, we note this in the documentation below
 
     %% Feature-to-feature dependencies (extracted from draw.io diagram)
     ADDRESS_MGMT --> LABELS
+    ANNOUNCEMENTS --> SETTINGS
+    ANNOUNCEMENTS --> TX_HISTORY
     APP_STARTUP --> WALLETS
     AUTOSWAPS --> TRANSFER
     BIP85 --> SECRETS
@@ -73,9 +76,12 @@ graph TB
     LABELS --> CORE
     PAY --> RECIPIENTS
     PAYJOIN --> UTXO_MGMT
+    PAYJOIN --> LABELS
     PIN_CODE --> CORE
     RECEIVE --> PAYJOIN
+    RECEIVE --> SETTINGS
     RECEIVE --> SWAPS
+    RECEIVE --> TX_HISTORY
     RECIPIENTS --> EXCHANGE
     SECRETS --> CORE
     SELL --> EXCHANGE
@@ -84,6 +90,7 @@ graph TB
     SEND --> NETWORK
     SEND --> PAYJOIN
     SEND --> SWAPS
+    SEND --> TX_HISTORY
     SEND --> UTXO_MGMT
     SEND --> WALLETS
     SETTINGS --> CORE
@@ -109,7 +116,7 @@ graph TB
     classDef featureStyle fill:#1a202c,stroke:#2d3748,stroke-width:2px,color:#e2e8f0
 
     class CORE coreStyle
-    class SETTINGS,TOR,PIN_CODE,LABELS,SECRETS,HW_WALLETS,BTC_PRICE,NETWORK,BIP85,FEES,WALLETS,EXCHANGE,APP_STARTUP,UTXO_MGMT,ADDRESS_MGMT,RECIPIENTS,FUNDING,BACKUPS,SWAPS,PAYJOIN,WITHDRAWAL,STATUS,SEND,RECEIVE,TRANSFER,TX_HISTORY,BG_TASKS,AUTOSWAPS,DCA,SELL,PAY,BUY,COINS,CONSOLIDATION featureStyle
+    class SETTINGS,TOR,PIN_CODE,LABELS,SECRETS,HW_WALLETS,BTC_PRICE,NETWORK,BIP85,FEES,WALLETS,EXCHANGE,APP_STARTUP,UTXO_MGMT,ADDRESS_MGMT,RECIPIENTS,FUNDING,BACKUPS,SWAPS,PAYJOIN,WITHDRAWAL,STATUS,SEND,RECEIVE,TRANSFER,TX_HISTORY,BG_TASKS,AUTOSWAPS,DCA,SELL,PAY,BUY,COINS,ANNOUNCEMENTS,CONSOLIDATION featureStyle
 ```
 
 ## About Package Dependency Diagrams

integration_test/payjoin_test.dart

@@ -18,7 +18,6 @@ import 'package:bb_mobile/core/wallet/domain/repositories/wallet_utxo_repository
 import 'package:bb_mobile/core/wallet/domain/usecases/prepare_bitcoin_send_usecase.dart';
 import 'package:bb_mobile/features/send/domain/usecases/sign_bitcoin_tx_usecase.dart';
 import 'package:bb_mobile/features/settings/domain/usecases/set_environment_usecase.dart';
-import 'package:bb_mobile/features/settings/domain/usecases/set_payjoin_enabled_usecase.dart';
 import 'package:bb_mobile/locator.dart';
 import 'package:bb_mobile/main.dart';
 
@@ -109,10 +108,6 @@ Future<void> main({bool isInitialized = false}) async {
 
   setUpAll(() async {
     await locator<SetEnvironmentUsecase>().execute(Environment.testnet);
-    // Payjoin is disabled by default (opt-in) — this suite exercises the
-    // receive-with-payjoin usecase directly, so it must explicitly opt in,
-    // same as a real user would from the payjoin settings screen.
-    await locator<SetPayjoinEnabledUsecase>().execute(true);
 
     // Drain any persisted payjoin state so the test starts clean. Ongoing
     // payjoins left behind by a previous (possibly crashed) run keep their
@@ -249,8 +244,7 @@ Future<void> main({bool isInitialized = false}) async {
             walletId: receiverWallet.id,
             address: address.address,
           );
-          expect(payjoin, isNotNull, reason: 'payjoin is enabled in setUpAll');
-          debugPrint('Payjoin receiver created: ${payjoin!.id}');
+          debugPrint('Payjoin receiver created: ${payjoin.id}');
 
           expect(payjoin.status, PayjoinStatus.started);
           // Check that the payjoin uri is correct
@@ -336,8 +330,7 @@ Future<void> main({bool isInitialized = false}) async {
           address: address.address,
           expireAfterSec: expireAfterSec,
         );
-        expect(payjoin, isNotNull, reason: 'payjoin is enabled in setUpAll');
-        debugPrint('Payjoin receiver created: ${payjoin!.id}');
+        debugPrint('Payjoin receiver created: ${payjoin.id}');
 
         final didReceiverExpire = await Future.any([
           payjoinReceiverExpiredEvent.future,

lib/core/payjoin/data/datasources/local_payjoin_datasource.dart

@@ -58,8 +58,15 @@ class LocalPayjoinDatasource {
       Expression<bool> expr = const Constant(true); // identity
 
       if (onlyUnfinished) {
+        // isAborted is a terminal outcome too (we already broadcast the
+        // original in its place) — excluded here for the same reason
+        // isCompleted/isExpired are, otherwise an aborted session would
+        // keep being "resumed" on every app start.
         expr =
-            expr & row.isExpired.equals(false) & row.isCompleted.equals(false);
+            expr &
+            row.isExpired.equals(false) &
+            row.isCompleted.equals(false) &
+            row.isAborted.equals(false);
       }
 
       if (walletId != null) {
@@ -78,7 +85,10 @@ class LocalPayjoinDatasource {
 
       if (onlyUnfinished) {
         expr =
-            expr & row.isExpired.equals(false) & row.isCompleted.equals(false);
+            expr &
+            row.isExpired.equals(false) &
+            row.isCompleted.equals(false) &
+            row.isAborted.equals(false);
       }
 
       if (walletId != null) {
@@ -103,10 +113,24 @@ class LocalPayjoinDatasource {
     ];
   }
 
+  /// Fetches the payjoin session(s) a transaction id belongs to, matching
+  /// BOTH the payjoin transaction id and the original transaction id. The
+  /// original matters as much as the payjoin one: an aborted session (we
+  /// broadcast the original instead of completing a real payjoin — see
+  /// PayjoinStatus.aborted) has no [txId] at all, so the transaction that
+  /// actually hit the chain IS the original — matching only [txId] made
+  /// that transaction's details lose its payjoin context entirely, hiding
+  /// the very "aborted" outcome the status exists to communicate. The
+  /// transactions LIST already joins on both ids
+  /// (GetTransactionsUsecase); this keeps the details path consistent.
   Future<List<PayjoinModel>> fetchByTxId(String txId) async {
     final (receivers, senders) = await (
-      _db.managers.payjoinReceivers.filter((f) => f.txId(txId)).get(),
-      _db.managers.payjoinSenders.filter((f) => f.txId(txId)).get(),
+      _db.managers.payjoinReceivers
+          .filter((f) => f.txId(txId) | f.originalTxId(txId))
+          .get(),
+      _db.managers.payjoinSenders
+          .filter((f) => f.txId(txId) | f.originalTxId(txId))
+          .get(),
     ).wait;
 
     return [
@@ -124,6 +148,7 @@ class LocalPayjoinDatasource {
       receivers = await receiversTable
           .filter((f) => f.isExpired(false))
           .filter((f) => f.isCompleted(false))
+          .filter((f) => f.isAborted(false))
           .get();
     } else {
       receivers = await receiversTable.get();
@@ -147,6 +172,7 @@ class LocalPayjoinDatasource {
       senders = await sendersTable
           .filter((f) => f.isExpired(false))
           .filter((f) => f.isCompleted(false))
+          .filter((f) => f.isAborted(false))
           .get();
     } else {
       senders = await sendersTable.get();

lib/core/payjoin/data/datasources/pdk_payjoin_datasource.dart

@@ -5,6 +5,7 @@ import 'dart:developer';
 import 'package:bb_mobile/core/errors/bull_exception.dart';
 import 'package:bb_mobile/core/payjoin/data/models/payjoin_input_pair_model.dart';
 import 'package:bb_mobile/core/payjoin/data/models/payjoin_model.dart';
+import 'package:bb_mobile/core/payjoin/domain/entity/payjoin.dart' show Payjoin;
 import 'package:bb_mobile/core/utils/bitcoin_tx.dart';
 import 'package:bb_mobile/core/utils/constants.dart';
 import 'package:bb_mobile/core/utils/logger.dart' as logger;
@@ -43,6 +44,8 @@ class PdkPayjoinDatasource {
   final Set<String> _receiverPollsInFlight = {};
   final Set<String> _senderPollsInFlight = {};
 
+  bool _disposed = false;
+
   PdkPayjoinDatasource({
     this._payjoinDirectoryUrl = PayjoinConstants.directoryUrl,
     required this._dio,
@@ -60,6 +63,46 @@ class PdkPayjoinDatasource {
 
   Stream<PayjoinModel> get expiredPayjoins => _expiredController.stream;
 
+  /// Stops the directory polling of one session — both the receiver
+  /// request poll and the sender proposal poll, whichever exists for
+  /// [payjoinId]. Called by the repository the moment a session resolves
+  /// through a path the poll itself can't see (the plain-broadcast fallback
+  /// landing on-chain): the poll only self-cancels on request/proposal
+  /// found or expiry, so without this it kept firing until expiry and then
+  /// raised a stale expired event for an already-completed session
+  /// (observed live: a redundant second broadcast of the original
+  /// transaction a minute after the session had already resolved).
+  void stopPolling(String payjoinId) {
+    _receiverTimers.remove(payjoinId)?.cancel();
+    _senderTimers.remove(payjoinId)?.cancel();
+  }
+
+  /// Cancels every polling timer and closes the event streams. Individual
+  /// poll timers self-cancel on success/expiry, but a session that never
+  /// resolves (a relay permanently down) would otherwise leave a
+  /// [Timer.periodic] firing forever plus three unclosed broadcast
+  /// controllers. The production singleton lives for the whole app session,
+  /// but tests (and any future teardown) need a clean exit; the repository's
+  /// own dispose delegates here. Idempotent: a second call is a no-op (closing
+  /// an already-closed controller would otherwise throw).
+  Future<void> dispose() async {
+    if (_disposed) return;
+    _disposed = true;
+    for (final timer in _receiverTimers.values) {
+      timer.cancel();
+    }
+    _receiverTimers.clear();
+    for (final timer in _senderTimers.values) {
+      timer.cancel();
+    }
+    _senderTimers.clear();
+    _receiverPollsInFlight.clear();
+    _senderPollsInFlight.clear();
+    await _payjoinRequestedController.close();
+    await _proposalSentController.close();
+    await _expiredController.close();
+  }
+
   Future<(OhttpKeys?, String?)> fetchOhttpKeyAndRelay({
     required String payjoinDirectory,
   }) async {
@@ -274,6 +317,57 @@ class PdkPayjoinDatasource {
     return updatedModel;
   }
 
+  /// Formally cancels a receiver session that was declined below the
+  /// configured minimum-receive-amount threshold (see
+  /// PayjoinRepositoryImpl._processPayjoinRequest), and closes the
+  /// underlying PDK session so it persists a terminal event.
+  ///
+  /// This replaces silently abandoning the session after broadcasting the
+  /// original transaction out of band: without this, the PDK's own
+  /// typestate machine never learns the session ended, so only our local
+  /// DB flag (isAborted) stood between it and being replayed/resumed as if
+  /// still pending. `cancel()` is available on every receive typestate that
+  /// carries a fallback transaction (verified against the installed
+  /// `payjoin` package's Dart bindings — `MaybeInputsOwned.cancel()` is one
+  /// of them); calling it here transitions to `ReceiverPendingFallback`,
+  /// whose `close()` persists the closing `SessionEvent` via the
+  /// persister. The original transaction itself is still broadcast by the
+  /// caller from the already-captured, already-validated
+  /// [PayjoinReceiverModel.originalTxBytes] — this method only concludes
+  /// the PDK-side state machine to match that outcome.
+  ///
+  /// Always called right after `_pollReceiverOnce` has persisted a session
+  /// at exactly the `MaybeInputsOwned` typestate (where
+  /// `originalTxBytes`/`amountSat` first become available) — any other
+  /// state means the session already progressed past the point a
+  /// below-minimum decline is possible, or is already resolved.
+  String declineReceiverSession(PayjoinReceiverModel receiverModel) {
+    final persister = InMemoryJsonReceiverSessionPersister.fromJson(
+      receiverModel.receiver,
+    );
+    final state = replayReceiverEventLog(persister: persister).state();
+    if (state is! MaybeInputsOwnedReceiveSession) {
+      throw StateError(
+        'Cannot decline payjoin receiver ${receiverModel.id}: expected a '
+        'MaybeInputsOwned session, got $state',
+      );
+    }
+
+    final pendingFallback = state.inner.cancel().save(persister: persister);
+    if (pendingFallback == null) {
+      // The session was already terminal (e.g. a race with another decline
+      // path) — nothing further to persist, but not an error either.
+      logger.log.info(
+        'Payjoin receiver ${receiverModel.id} was already resolved when '
+        'declining below minimum',
+      );
+      return persister.toJson();
+    }
+
+    pendingFallback.close().save(persister: persister);
+    return persister.toJson();
+  }
+
   Future<({Monitor monitor, String psbt})> processReceiveSession({
     required ReceiveSession state,
     required InMemoryJsonReceiverSessionPersister persister,
@@ -688,14 +782,18 @@ class PdkPayjoinDatasource {
     PayjoinSenderModel senderModel,
     Timer timer,
   ) async {
+    // logRef, never the raw id in log lines/exception messages: a sender id
+    //  is the full BIP21 URI (address+amount+endpoint). The raw id is still
+    //  used as the internal map key below, which never reaches a log.
+    final senderLogRef = Payjoin.logRefForId(senderModel.id);
     if (!_senderPollsInFlight.add(senderModel.id)) return;
-    log('[sender poll] checking for proposal for ${senderModel.id}');
+    log('[sender poll] checking for proposal for $senderLogRef');
     try {
       // Local expiry backstop: don't rely solely on the PDK surfacing an
       // "expired" error — bound polling by the session's own expiry time.
       if (senderModel.isExpiryTimePassed) {
         throw PayjoinExpiredException(
-          'Payjoin sender ${senderModel.id} expiry time passed',
+          'Payjoin sender $senderLogRef expiry time passed',
         );
       }
       final persister = InMemoryJsonSenderSessionPersister.fromJson(
@@ -715,7 +813,7 @@ class PdkPayjoinDatasource {
       final proposalPsbt = await _getProposalPsbt(state.inner, persister);
       if (proposalPsbt == null) return;
 
-      log('[sender poll] proposal found for ${senderModel.id}');
+      log('[sender poll] proposal found for $senderLogRef');
       final txId = (await BitcoinTx.fromPsbt(proposalPsbt)).txid;
       final updatedModel = senderModel.copyWith(
         sender: persister.toJson(),
@@ -730,13 +828,13 @@ class PdkPayjoinDatasource {
       _senderTimers.remove(senderModel.id);
       _proposalSentController.add(updatedModel);
     } on PayjoinExpiredException catch (e) {
-      logger.log.info('[sender poll] expired for ${senderModel.id}: $e');
+      logger.log.info('[sender poll] expired for $senderLogRef: $e');
       if (!timer.isActive) return;
       timer.cancel();
       _senderTimers.remove(senderModel.id);
       _expiredController.add(senderModel.copyWith(isExpired: true));
     } catch (e) {
-      logger.log.info('[sender poll] ${senderModel.id}: $e');
+      logger.log.info('[sender poll] $senderLogRef: $e');
     } finally {
       _senderPollsInFlight.remove(senderModel.id);
     }