refactor(transactions): speed up transfer progress and transaction details

lib/features/swap/presentation/transfer_bloc.dart

@@ -1520,20 +1520,22 @@ class TransferBloc extends Bloc<TransferEvent, TransferState>
           state.copyWith(
             orderSwap: broadcast,
             swap: displaySwap.copyWith(status: SwapStatus.paid, sendTxid: txId),
+            txId: txId,
           ),
         );
-        await _getWalletUsecase.execute(state.fromWallet!.id, sync: true);
+        unawaited(_syncWalletAfterBroadcast(state.fromWallet!.id));
+        return;
       } else if (state.isSameChainTransfer) {
         txId = await _broadcastBitcoinTxUsecase.execute(
           signedPsbt,
           isPsbt: true,
         );
-        if (state.fromWallet != null) {
-          await _getWalletUsecase.execute(state.fromWallet!.id, sync: true);
-        }
-        if (state.toWallet != null) {
-          await _getWalletUsecase.execute(state.toWallet!.id, sync: true);
-        }
+        unawaited(
+          _syncWalletsAfterBroadcast([
+            if (state.fromWallet != null) state.fromWallet!.id,
+            if (state.toWallet != null) state.toWallet!.id,
+          ]),
+        );
       } else {
         return;
       }
@@ -1550,6 +1552,20 @@ class TransferBloc extends Bloc<TransferEvent, TransferState>
     }
   }
 
+  Future<void> _syncWalletAfterBroadcast(String walletId) async {
+    try {
+      await _getWalletUsecase.execute(walletId, sync: true);
+    } catch (_) {
+      log.warning('Failed to sync wallet after transfer broadcast');
+    }
+  }
+
+  Future<void> _syncWalletsAfterBroadcast(List<String> walletIds) async {
+    for (final walletId in walletIds) {
+      await _syncWalletAfterBroadcast(walletId);
+    }
+  }
+
   SwapFailure _swapFailureFromException(Object exception) {
     if (exception is DioException) {
       return switch (exception.type) {
@@ -1647,10 +1663,10 @@ class TransferBloc extends Bloc<TransferEvent, TransferState>
       ),
     );
     if (orderSwap.localStatus.isTerminal) {
-      unawaited(_getWalletUsecase.execute(state.fromWallet!.id, sync: true));
+      unawaited(_syncWalletAfterBroadcast(state.fromWallet!.id));
       final destinationWalletId = orderSwap.destinationWalletId;
       if (destinationWalletId != null) {
-        unawaited(_getWalletUsecase.execute(destinationWalletId, sync: true));
+        unawaited(_syncWalletAfterBroadcast(destinationWalletId));
       }
     }
   }

lib/features/transactions/domain/entities/transaction.dart

@@ -53,6 +53,10 @@ sealed class Transaction with _$Transaction {
       order?.isLiquid ??
       false;
   String? get toAddress => walletTransaction?.toAddress ?? order?.toAddress;
+  String? get orderSwapDestinationAddress =>
+      orderSwap?.outNetwork == OrderSwapNetwork.lightning
+      ? null
+      : orderSwap?.destination;
 
   bool get isBroadcasted => walletTransaction != null;
   bool get isSwap => swap != null || orderSwap != null;
@@ -175,6 +179,19 @@ sealed class Transaction with _$Transaction {
       swap?.type == SwapType.liquidToBitcoin ||
       (orderSwap?.inNetwork == OrderSwapNetwork.liquid &&
           orderSwap?.outNetwork == OrderSwapNetwork.bitcoin);
+  // Internal swaps are outgoing from one wallet and incoming to the other.
+  bool isReceivingWallet(String? walletId, {bool isCounterpart = false}) {
+    final orderSwap = this.orderSwap;
+    if (walletId != null && orderSwap != null) {
+      return orderSwap.destinationWalletId == walletId &&
+          orderSwap.sourceWalletId != walletId;
+    }
+    final swap = this.swap;
+    if (walletId != null && swap is ChainSwap) {
+      return swap.receiveWalletId == walletId && swap.sendWalletId != walletId;
+    }
+    return isCounterpart ? isOutgoing : isIncoming;
+  }
 
   DateTime? get timestamp =>
       // Completed swaps are displayed (and should sort) by when they finished,

lib/features/transactions/presentation/blocs/transaction_details/transaction_details_cubit.dart

@@ -11,6 +11,7 @@ import 'package:bb_mobile/core/wallet/domain/entities/wallet_transaction.dart';
 import 'package:bb_mobile/core/wallet/domain/usecases/get_wallet_transaction_usecase.dart';
 import 'package:bb_mobile/core/wallet/domain/usecases/get_wallet_usecase.dart';
 import 'package:bb_mobile/core/wallet/domain/usecases/watch_wallet_transaction_by_tx_id_usecase.dart';
+import 'package:bb_mobile/core/wallet/domain/wallet_failure.dart';
 import 'package:bb_mobile/features/labels/labels_facade.dart';
 import 'package:bb_mobile/features/swap/public/swap_facade.dart';
 import 'package:bb_mobile/features/transactions/domain/entities/transaction.dart';
@@ -75,6 +76,9 @@ class TransactionDetailsCubit extends Cubit<TransactionDetailsState> {
   StreamSubscription? _payjoinTxSubscription;
   StreamSubscription? _payjoinOriginalTxSubscription;
   StreamSubscription? _orderSwapSubscription;
+  String? _watchedOrderSwapTransactionId;
+  WalletTransaction? _pendingOrderSwapWalletTransaction;
+  int _orderSwapTransactionWatchGeneration = 0;
 
   // The payjoin id _payjoinSubscription is currently listening to on the
   // by-wallet-tx path, so reloads triggered by its own events don't
@@ -142,13 +146,13 @@ class TransactionDetailsCubit extends Cubit<TransactionDetailsState> {
   }
 
   Future<void> initByOrderSwapLocalId(String localId) async {
+    _reload = () => _loadDetailsByOrderSwapLocalId(localId);
     await _loadDetailsByOrderSwapLocalId(localId);
     _orderSwapSubscription = _watchTransactionOrderSwapUsecase
         .execute(localId)
         .listen(
           (orderSwap) {
-            if (isClosed) return;
-            _loadOrderSwapDetails(orderSwap);
+            unawaited(_handleOrderSwapUpdate(orderSwap));
           },
           onError: (Object error) {
             if (isClosed) return;
@@ -157,10 +161,88 @@ class TransactionDetailsCubit extends Cubit<TransactionDetailsState> {
         );
   }
 
+  Future<void> _handleOrderSwapUpdate(OrderSwapRecord orderSwap) async {
+    if (isClosed) return;
+    final transaction = state.transaction;
+    if (transaction?.orderSwap?.localId != orderSwap.localId) {
+      await _loadDetailsByOrderSwapLocalId(orderSwap.localId);
+      await _watchOrderSwapWalletTransaction(state.transaction?.orderSwap);
+      return;
+    }
+    final walletTransaction = transaction?.walletTransaction;
+    final canonicalTransactionChanged =
+        walletTransaction != null &&
+        walletTransaction.txId != orderSwap.canonicalWalletTransactionId;
+    emit(
+      state.copyWith(
+        transaction: transaction!.copyWith(
+          walletTransaction: canonicalTransactionChanged
+              ? null
+              : walletTransaction,
+          orderSwap: orderSwap,
+        ),
+        swapCounterpartTxId: orderSwap.counterpartTransactionId,
+      ),
+    );
+    await _watchOrderSwapWalletTransaction(orderSwap);
+  }
+
+  Future<void> _watchOrderSwapWalletTransaction(
+    OrderSwapRecord? orderSwap,
+  ) async {
+    final transactionId = orderSwap?.canonicalWalletTransactionId;
+    final walletId = orderSwap?.canonicalWalletId;
+    if (transactionId == null || walletId == null) return;
+    if (_watchedOrderSwapTransactionId == transactionId) return;
+    final generation = ++_orderSwapTransactionWatchGeneration;
+    _pendingOrderSwapWalletTransaction = null;
+    await _walletTransactionSubscription?.cancel();
+    if (isClosed || generation != _orderSwapTransactionWatchGeneration) return;
+    try {
+      _walletTransactionSubscription = _watchWalletTransactionByTxIdUsecase
+          .execute(txId: transactionId, walletId: walletId)
+          .listen(
+            (walletTransaction) {
+              if (isClosed) return;
+              final latestOrderSwap = state.transaction?.orderSwap;
+              if (latestOrderSwap == null) {
+                _pendingOrderSwapWalletTransaction = walletTransaction;
+                return;
+              }
+              if (latestOrderSwap.canonicalWalletTransactionId !=
+                  transactionId) {
+                return;
+              }
+              emit(
+                state.copyWith(
+                  transaction: state.transaction?.copyWith(
+                    walletTransaction: walletTransaction,
+                  ),
+                ),
+              );
+            },
+            onError: (_) {
+              if (_watchedOrderSwapTransactionId == transactionId) {
+                _watchedOrderSwapTransactionId = null;
+              }
+              log.warning('Order swap wallet transaction watcher failed');
+            },
+          );
+      _watchedOrderSwapTransactionId = transactionId;
+    } catch (_) {
+      _watchedOrderSwapTransactionId = null;
+      log.warning('Order swap wallet transaction watcher failed');
+    }
+  }
+
   Future<void> _loadDetailsByOrderSwapLocalId(String localId) async {
     try {
       final orderSwap = await _getTransactionOrderSwapUsecase.execute(localId);
+      await _watchOrderSwapWalletTransaction(orderSwap);
       await _loadOrderSwapDetails(orderSwap);
+    } on ParallelWaitError catch (error) {
+      if (isClosed) return;
+      emit(state.copyWith(err: _firstParallelError(error)));
     } on TransactionNotFoundError catch (error) {
       if (isClosed) return;
       emit(state.copyWith(notFoundError: error));
@@ -177,39 +259,63 @@ class TransactionDetailsCubit extends Cubit<TransactionDetailsState> {
       emit(state.copyWith(err: TransactionNotFoundError()));
       return;
     }
-    final wallet = await _getWalletUsecase.execute(walletId);
     final counterpartWalletId = orderSwap.sourceWalletId == walletId
         ? orderSwap.destinationWalletId
         : orderSwap.sourceWalletId;
-    final counterpartWallet = counterpartWalletId == null
-        ? null
-        : await _getWalletUsecase.execute(counterpartWalletId);
-    var transaction = Transaction(orderSwap: orderSwap);
-    final walletTransactionId = orderSwap.canonicalWalletTransactionId;
-    if (walletTransactionId != null) {
-      try {
-        final transactions = await _getTransactionsByTxIdUsecase.execute(
-          walletTransactionId,
-        );
-        transaction = transactions.firstWhere(
-          (candidate) => candidate.walletId == walletId,
-          orElse: () => transaction,
-        );
-      } on TransactionNotFoundError {
-        // The order remains displayable while the wallet sync catches up.
-      }
-    }
+    final transactionId = orderSwap.canonicalWalletTransactionId;
+    final (wallet, counterpartWallet, walletTransactionResult) = await (
+      _getWalletUsecase.execute(walletId),
+      counterpartWalletId == null
+          ? Future<Wallet?>.value()
+          : _getWalletUsecase.execute(counterpartWalletId),
+      transactionId == null
+          ? Future.value(
+              const Ok<WalletTransaction?, WalletTransactionLookupFailure>(
+                null,
+              ),
+            )
+          : _getWalletTransactionUsecase.execute(
+              txId: transactionId,
+              walletId: walletId,
+            ),
+    ).wait;
+    final loadedWalletTransaction = switch (walletTransactionResult) {
+      Ok(:final value) => value,
+      Err() => () {
+        log.warning('Order swap wallet transaction lookup failed');
+        return null;
+      }(),
+    };
+    final pendingWalletTransaction = _pendingOrderSwapWalletTransaction;
+    _pendingOrderSwapWalletTransaction = null;
+    final matchingPendingWalletTransaction =
+        pendingWalletTransaction?.txId == transactionId &&
+            pendingWalletTransaction?.walletId == walletId
+        ? pendingWalletTransaction
+        : null;
+    final walletTransaction =
+        matchingPendingWalletTransaction ?? loadedWalletTransaction;
     if (isClosed) return;
     emit(
       state.copyWith(
-        transaction: transaction,
+        transaction: Transaction(
+          walletTransaction: walletTransaction,
+          orderSwap: orderSwap,
+        ),
         wallet: wallet,
         counterpartWallet: counterpartWallet,
         swapCounterpartTxId: orderSwap.counterpartTransactionId,
+        err: null,
+        notFoundError: null,
       ),
     );
   }
 
+  Object _firstParallelError(ParallelWaitError error) {
+    final errors = error.errors as (AsyncError?, AsyncError?, AsyncError?);
+    return (errors.$1 ?? errors.$2 ?? errors.$3)?.error ?? error;
+  }
+
   Future<void> _loadDetailsByWalletTxId(
     String txId, {
     required String walletId,

lib/features/transactions/ui/widgets/transaction_details_table.dart

@@ -54,7 +54,10 @@ class TransactionDetailsTable extends StatelessWidget {
 
     final swap = transaction?.swap;
     final orderSwap = transaction?.orderSwap;
-    final toAddress = swap?.receiveAddress ?? transaction?.toAddress;
+    final toAddress =
+        swap?.receiveAddress ??
+        transaction?.orderSwapDestinationAddress ??
+        transaction?.toAddress;
     final payjoin = transaction?.payjoin;
     final order = transaction?.order;
     final txFee = transaction?.payjoinSenderFeeSat ?? walletTransaction?.feeSat;
@@ -115,14 +118,19 @@ class TransactionDetailsTable extends StatelessWidget {
           ),
         if (walletLabel.isNotEmpty)
           DetailsTableItem(
-            label: transaction?.isIncoming == true
+            label: transaction?.isReceivingWallet(wallet?.id) == true
                 ? context.loc.transactionDetailLabelToWallet
                 : context.loc.transactionDetailLabelFromWallet,
             displayValue: walletLabel,
           ),
         if (counterpartWalletLabel.isNotEmpty && !recovered)
           DetailsTableItem(
-            label: transaction?.isOutgoing == true
+            label:
+                transaction?.isReceivingWallet(
+                      counterpartWallet?.id,
+                      isCounterpart: true,
+                    ) ==
+                    true
                 ? context.loc.transactionDetailLabelToWallet
                 : context.loc.transactionDetailLabelFromWallet,
             displayValue: counterpartWalletLabel,

test/features/swap/presentation/transfer_bloc_test.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:bb_mobile/core/blockchain/domain/usecases/broadcast_bitcoin_transaction_usecase.dart';
 import 'package:bb_mobile/core/blockchain/domain/usecases/broadcast_liquid_transaction_usecase.dart';
 import 'package:bb_mobile/core/exchange/domain/usecases/convert_sats_to_currency_amount_usecase.dart';
@@ -232,6 +234,44 @@ void main() {
     },
   );
 
+  test('emits the transaction id before wallet sync completes', () async {
+    final syncCompleter = Completer<Wallet>();
+    final broadcasting = _prepared(
+      status: OrderSwapLocalStatus.broadcastUnknown,
+    );
+    final broadcasted = _prepared(
+      status: OrderSwapLocalStatus.payinBroadcast,
+      transactionId: 'txid-1',
+    );
+    when(
+      () => getWallet.execute('wallet-1', sync: true),
+    ).thenAnswer((_) => syncCompleter.future);
+    when(
+      () => markUnknown.execute('local-1'),
+    ).thenAnswer((_) async => Ok(broadcasting));
+    when(
+      () => broadcastBitcoin.execute('signed-psbt', isPsbt: true),
+    ).thenAnswer((_) async => 'txid-1');
+    when(
+      () => markBroadcast.execute(localId: 'local-1', transactionId: 'txid-1'),
+    ).thenAnswer((_) async => Ok(broadcasted));
+    bloc.emit(
+      TransferState(
+        orderSwap: prepared,
+        signedPsbt: 'signed-psbt',
+        fromWallet: _wallet(),
+        swap: _swap(),
+      ),
+    );
+
+    bloc.add(const TransferEvent.confirmed());
+    await bloc.stream.firstWhere((state) => state.txId == 'txid-1');
+
+    expect(syncCompleter.isCompleted, isFalse);
+    expect(bloc.state.orderSwap, broadcasted);
+    syncCompleter.complete(_wallet());
+  });
+
   test('resumes a stored prepared transfer on start', () async {
     final settings = SettingsEntity(
       environment: Environment.testnet,