searchForTransactions function

Implementation

Future<(Set<ElectrumTransactionInfo>, List<NodeWithAddress>)>
    searchForTransactions({
  required bip32.BIP32 masterNode,
  required int chainIndex,
  required Iterable<AddressType> addressTypes,
  required HDWalletPurpose? walletPurpose,
  required UTXONetworkType networkType,
  required List<(String, int)> endpoints,
  required List<NodeWithAddress> cachedNodes,
  required IsolateManager isolateManager,
  int emptyLimit = kEmptyLimit,
}) async {
  final watch = Stopwatch()..start();

  int emptyCount = 0;

  final txs0 = <ElectrumTransactionInfo>{};

  final clients = await createClients(
    endpoints: endpoints,
    token: networkType.coin,
  );
  final batchSize = clients.length;
  final nodes = <NodeWithAddress>[];

  Logger.logFetch(
    "Fetching transactions from $batchSize ElectrumX Nodes",
  );

  if (batchSize == 0) {
    Logger.logWarning("No ElectrumX Nodes available for $networkType");
    return (txs0, nodes);
  }

  for (var index = 0; index * batchSize < kAddressesUpperLimit; index++) {
    final indexes = List.generate(batchSize, (i) => index * batchSize + i);
    final newIndexes = indexes.where(
      (i) => !cachedNodes.any(
        (cNode) => cNode.index == i && cNode.chainIndex == chainIndex,
      ),
    );

    final List<NodeWithAddress> newNodes;
    if (newIndexes.isEmpty) {
      newNodes = [];
    } else {
      newNodes = await isolateManager.executeTask(
        IsolateTask(
          task: (arg) => [
            for (var index in indexes)
              deriveChildNode(
                masterNode: masterNode,
                chainIndex: chainIndex,
                index: index,
                networkType: networkType,
                addressTypes: addressTypes,
                walletPurpose: walletPurpose,
              ),
          ],
          argument: null,
        ),
      );
    }

    final batchNodes = [
      ...newNodes,
      ...cachedNodes.where(
        (cNode) =>
            indexes.contains(cNode.index) && cNode.chainIndex == chainIndex,
      ),
    ];

    final futures = [
      for (int i = 0; i < batchSize; i++)
        fetchFromRandomElectrumXNode(
          (client) async {
            final txsInfos = [
              for (final type in addressTypes)
                if (batchNodes[i].addresses.containsKey(type))
                  await client.getHistory(
                    P2Hash(batchNodes[i].addresses[type]!).publicKeyScriptHash,
                  )
            ];

            return txsInfos
                .expand((e) => e ?? <ElectrumTransactionInfo>{})
                .whereType<ElectrumTransactionInfo>()
                .toSet();
          },
          endpoints: networkType.endpoints,
          client: clients[i],
          token: networkType.coin,
        ),
    ];

    final results = await Future.wait(futures);
    final batchTxs = <ElectrumTransactionInfo>{};

    for (int i = 0; i < batchSize; i++) {
      final (txs, client, error) = results[i];
      if (error != null) continue;
      if (txs == null || txs.isEmpty) continue;
      if (client != null) clients[i] = client;

      batchTxs.addAll(txs);
    }

    nodes.addAll(batchNodes);

    if (batchTxs.isEmpty) {
      emptyCount += batchSize;
      if (emptyCount >= kEmptyLimit) {
        Logger.log("Abort Search after ${index + batchSize} addresses");
        break;
      }
      continue;
    }

    txs0.addAll(batchTxs);
    emptyCount = 0;
  }

  ///
  /// Disconnect Clients
  ///
  await Future.wait([
    for (final client in clients) client.disconnect(),
  ]);

  watch.stop();
  Logger.logFetch(
    "Fetched ${txs0.length} TXs in ${watch.elapsed} $chainIndex",
  );

  return (txs0, nodes);
}