| 1 | import 'dart:async'; |
| 2 | |
| 3 | import 'package:cw_core/utils/print_verbose.dart'; |
| 4 | import 'package:cw_zcash/src/zcash_wallet_service.dart'; |
| 5 | import 'package:zkool/src/rust/api/coin.dart' as zkool_coin; |
| 6 | import 'package:zkool/src/rust/api/mempool.dart' as zkool_mempool; |
| 7 | |
| 8 | typedef ZcashMempoolRefreshCallback = void Function(Set<int> accountIds); |
| 9 | |
| 10 | class ZcashMempoolService { |
| 11 | ZcashMempoolService._(); |
| 12 | |
| 13 | static final ZcashMempoolService instance = ZcashMempoolService._(); |
| 14 | |
| 15 | zkool_mempool.Mempool? _mempool; |
| 16 | final Map<String, zkool_mempool.MempoolTx> _txsById = {}; |
| 17 | |
| 18 | zkool_mempool.Mempool get _mempoolInstance => _mempool ??= zkool_mempool.Mempool(); |
| 19 | |
| 20 | ZcashMempoolRefreshCallback? onAccountsUpdated; |
| 21 | |
| 22 | bool _loopRunning = false; |
| 23 | String? _nodeUrl; |
| 24 | int? _lastBlockHeight; |
| 25 | StreamSubscription<zkool_mempool.MempoolMsg>? _subscription; |
| 26 | |
| 27 | void removeTx(final String txId) { |
| 28 | _txsById.remove(ZcashWalletService.normalizeTxId(txId)); |
| 29 | } |
| 30 | |
| 31 | void removeKnownTxs(final Iterable<String> txIds) { |
| 32 | for (final txId in txIds) { |
| 33 | removeTx(txId); |
| 34 | } |
| 35 | } |
| 36 | |
| 37 | List<zkool_mempool.MempoolTx> txsForAccount(final int accountId) => _txsById.values |
| 38 | .where( |
| 39 | (final tx) => |
| 40 | tx.notes.any((final n) => n.account == accountId) || |
| 41 | tx.amounts.any((final a) => a.account == accountId), |
| 42 | ) |
| 43 | .toList(); |
| 44 | |
| 45 | Future<void> ensureRunning(final zkool_coin.Coin c) async { |
| 46 | if (c.url.isEmpty) { |
| 47 | return; |
| 48 | } |
| 49 | if (_loopRunning && _nodeUrl == c.url) { |
| 50 | return; |
| 51 | } |
| 52 | await stop(); |
| 53 | _nodeUrl = c.url; |
| 54 | _loopRunning = true; |
| 55 | unawaited(_runLoop(c)); |
| 56 | } |
| 57 | |
| 58 | Future<void> stop() async { |
| 59 | _loopRunning = false; |
| 60 | await _subscription?.cancel(); |
| 61 | _subscription = null; |
| 62 | _txsById.clear(); |
| 63 | _lastBlockHeight = null; |
| 64 | try { |
| 65 | await _mempool?.cancel(); |
| 66 | } catch (e) { |
| 67 | printV("mempool cancel: $e"); |
| 68 | } |
| 69 | _mempool = null; |
| 70 | } |
| 71 | |
| 72 | Future<void> _runLoop(final zkool_coin.Coin c) async { |
| 73 | while (_loopRunning && _nodeUrl == c.url) { |
| 74 | try { |
| 75 | final completer = Completer<void>(); |
| 76 | _subscription = _mempoolInstance |
| 77 | .run(c: c) |
| 78 | .listen( |
| 79 | (final msg) { |
| 80 | if (!_loopRunning) { |
| 81 | return; |
| 82 | } |
| 83 | switch (msg) { |
| 84 | case zkool_mempool.MempoolMsg_BlockHeight(:final field0): |
| 85 | _onBlockHeight(field0); |
| 86 | case zkool_mempool.MempoolMsg_TxId(:final field0): |
| 87 | _onMempoolTx(field0); |
| 88 | } |
| 89 | }, |
| 90 | onDone: () { |
| 91 | if (!completer.isCompleted) { |
| 92 | completer.complete(); |
| 93 | } |
| 94 | }, |
| 95 | onError: (final e) { |
| 96 | printV("mempool stream error: $e"); |
| 97 | if (!completer.isCompleted) { |
| 98 | completer.complete(); |
| 99 | } |
| 100 | }, |
| 101 | ); |
| 102 | await completer.future; |
| 103 | } catch (e) { |
| 104 | printV("mempool loop error: $e"); |
| 105 | } |
| 106 | if (_loopRunning) { |
| 107 | await Future.delayed(const Duration(seconds: 5)); |
| 108 | } |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | void _onBlockHeight(final int height) { |
| 113 | final heightChanged = _lastBlockHeight != null && height > _lastBlockHeight!; |
| 114 | _lastBlockHeight = height; |
| 115 | if (!heightChanged) { |
| 116 | return; |
| 117 | } |
| 118 | final affected = _allAffectedAccounts(); |
| 119 | _txsById.clear(); |
| 120 | onAccountsUpdated?.call(affected); |
| 121 | } |
| 122 | |
| 123 | void _onMempoolTx(final zkool_mempool.MempoolTx tx) { |
| 124 | final normalized = ZcashWalletService.normalizeTxId(tx.txid); |
| 125 | _txsById[normalized] = tx; |
| 126 | onAccountsUpdated?.call(_affectedAccountsForTx(tx)); |
| 127 | } |
| 128 | |
| 129 | Set<int> _allAffectedAccounts() => _txsById.values.expand(_affectedAccountsForTx).toSet(); |
| 130 | |
| 131 | Set<int> _affectedAccountsForTx(final zkool_mempool.MempoolTx tx) => { |
| 132 | ...tx.notes.map((final n) => n.account), |
| 133 | ...tx.amounts.map((final a) => a.account), |
| 134 | }; |
| 135 | } |