cake_wallet/cw_bitcoin/lib/payjoin/manager.dart

import 'dart:async';
import 'dart:isolate';
import 'dart:math';
import 'dart:typed_data';

import 'package:bitcoin_base/bitcoin_base.dart';
import 'package:cw_bitcoin/bitcoin_wallet.dart';
import 'package:cw_bitcoin/payjoin/payjoin_receive_worker.dart';
import 'package:cw_bitcoin/payjoin/payjoin_send_worker.dart';
import 'package:cw_bitcoin/payjoin/payjoin_session_errors.dart';
import 'package:cw_bitcoin/payjoin/storage.dart';
import 'package:cw_bitcoin/psbt_signer.dart';
import 'package:cw_core/utils/print_verbose.dart';
import 'package:payjoin_flutter/common.dart';
import 'package:payjoin_flutter/receive.dart';
import 'package:payjoin_flutter/send.dart';
import 'package:payjoin_flutter/uri.dart' as PayjoinUri;

class PayjoinManager {
  PayjoinManager(this._payjoinStorage, this._wallet);

  final PayjoinStorage _payjoinStorage;
  final BitcoinWalletBase _wallet;
  final Map<String, PayjoinPollerSession> _activePollers = {};

  static const List<String> ohttpRelayUrls = [
    'https://pj.bobspacebkk.com',
    'https://ohttp.achow101.com',
  ];

  static Future<PayjoinUri.Url> randomOhttpRelayUrl() => PayjoinUri.Url.fromStr(
      ohttpRelayUrls[Random.secure().nextInt(ohttpRelayUrls.length)]);

  static const payjoinDirectoryUrl = 'https://payjo.in';

  Future<void> resumeSessions() async {
    final allSessions = _payjoinStorage.readAllOpenSessions(_wallet.id);

    final spawnedSessions = allSessions.map((session) {
      if (session.isSenderSession) {
        return spawnSender(
          sender: Sender.fromJson(session.sender!),
          pjUri: session.pjUri!,
        );
      }
      return spawnReceiver(
        receiver: Receiver.fromJson(session.receiver!),
      );
    });

    await Future.wait(spawnedSessions);
  }

  Future<Sender> initSender(
      String pjUriString, String originalPsbt, int networkFeesSatPerVb) async {
    try {
      final pjUri =
          (await PayjoinUri.Uri.fromStr(pjUriString)).checkPjSupported();
      final minFeeRateSatPerKwu = BigInt.from(networkFeesSatPerVb * 250);
      final senderBuilder = await SenderBuilder.fromPsbtAndUri(
        psbtBase64: originalPsbt,
        pjUri: pjUri,
      );
      return senderBuilder.buildRecommended(minFeeRate: minFeeRateSatPerKwu);
    } catch (e) {
      throw Exception('Error initializing Payjoin Sender: $e');
    }
  }

  Future<void> spawnNewSender({
    required Sender sender,
    required String pjUrl,
    bool isTestnet = false,
  }) async {
    await _payjoinStorage.insertSenderSession(
      sender,
      pjUrl,
      _wallet.id,
    );

    return spawnSender(
      isTestnet: isTestnet,
      sender: sender,
      pjUri: pjUrl,
    );
  }

  Future<void> spawnSender({
    required Sender sender,
    required String pjUri,
    bool isTestnet = false,
  }) async {
    final completer = Completer();
    final receivePort = ReceivePort();

    receivePort.listen((message) async {
      print('Sender isolate: $message');
      if (message is Map<String, dynamic>) {
        try {
          switch (message['type']) {
            case PayjoinSenderRequestTypes.requestPosted:
              //ToDo: Update frontend
              return;
            case PayjoinSenderRequestTypes.psbtToSign:
              final proposalPsbt = message['psbt'] as String;
              final utxos = _wallet.getUtxoWithPrivateKeys();
              final finalizedPsbt = await _wallet.signPsbt(proposalPsbt, utxos);
              _wallet.commitPsbt(finalizedPsbt);

              //ToDo: Update frontend
              _cleanupSession(pjUri);
              await _payjoinStorage.markSenderSessionComplete(pjUri);
              completer.complete();
          }
        } catch (e) {
          _cleanupSession(pjUri);
          await _payjoinStorage.markReceiverSessionUnrecoverable(pjUri);
          completer.completeError(e);
        }
      } else if (message is PayjoinSessionError) {
        _cleanupSession(pjUri);
        if (message is UnrecoverableError) {
          printV(message.message);
          await _payjoinStorage.markReceiverSessionUnrecoverable(pjUri);
        }
        completer.completeError(message);
      }
    });

    final args = [
      receivePort.sendPort,
      sender.toJson(),
    ];

    final isolate = await Isolate.spawn(
      PayjoinSenderWorker.run,
      args,
    );

    _activePollers[pjUri] = PayjoinPollerSession(isolate, receivePort);

    return completer.future;
  }

  Future<Receiver> initReceiver(String address,
      [bool isTestnet = false]) async {
    try {
      final payjoinDirectory =
          await PayjoinUri.Url.fromStr(payjoinDirectoryUrl);

      final ohttpKeys = await PayjoinUri.fetchOhttpKeys(
        ohttpRelay: await randomOhttpRelayUrl(),
        payjoinDirectory: payjoinDirectory,
      );

      return Receiver.create(
        address: address,
        network: isTestnet ? Network.testnet : Network.bitcoin,
        directory: payjoinDirectory,
        ohttpKeys: ohttpKeys,
        ohttpRelay: await randomOhttpRelayUrl(),
      );
    } catch (e) {
      throw Exception('Error initializing Payjoin Receiver: $e');
    }
  }

  Future<void> spawnNewReceiver({
    required Receiver receiver,
    bool isTestnet = false,
  }) async {
    await _payjoinStorage.insertReceiverSession(
      receiver,
      _wallet.id,
    );
    return spawnReceiver(
      isTestnet: isTestnet,
      receiver: receiver,
    );
  }

  Future<void> spawnReceiver({
    required Receiver receiver,
    bool isTestnet = false,
  }) async {
    final completer = Completer();
    final receivePort = ReceivePort();

    SendPort? mainToIsolateSendPort;
    List<UtxoWithPrivateKey> utxos = [];

    receivePort.listen((message) async {
      print('Receiver isolate: $message');
      if (message is Map<String, dynamic>) {
        try {
          switch (message['type']) {
            case PayjoinReceiverRequestTypes.checkIsOwned:
              final inputScript = message['input_script'] as Uint8List;
              final isOwned =
                  _wallet.isMine(Script.fromRaw(byteData: inputScript));
              mainToIsolateSendPort?.send({
                'requestId': message['requestId'],
                'result': isOwned,
              });
              break;

            case PayjoinReceiverRequestTypes.checkIsReceiverOutput:
              final outputScript = message['output_script'] as Uint8List;
              final isReceiverOutput =
                  _wallet.isMine(Script.fromRaw(byteData: outputScript));
              mainToIsolateSendPort?.send({
                'requestId': message['requestId'],
                'result': isReceiverOutput,
              });
              break;

            case PayjoinReceiverRequestTypes.getCandidateInputs:
              utxos = _wallet.getUtxoWithPrivateKeys();
              mainToIsolateSendPort?.send({
                'requestId': message['requestId'],
                'result': utxos,
              });
              break;

            case PayjoinReceiverRequestTypes.processPsbt:
              final psbt = message['psbt'] as String;
              final signedPsbt = await _wallet.signPsbt(psbt, utxos);
              mainToIsolateSendPort?.send({
                'requestId': message['requestId'],
                'result': signedPsbt,
              });
              break;

            case PayjoinReceiverRequestTypes.proposalSent:
              _cleanupSession(receiver.id());
              await _payjoinStorage.markReceiverSessionComplete(receiver.id());
              completer.complete();
          }
        } catch (e) {
          _cleanupSession(receiver.id());
          await _payjoinStorage.markReceiverSessionUnrecoverable(receiver.id());
          completer.completeError(e);
        }
      } else if (message is PayjoinSessionError) {
        _cleanupSession(receiver.id());
        if (message is UnrecoverableError) {
          printV(message.message);
          await _payjoinStorage.markReceiverSessionUnrecoverable(receiver.id());
        }
        completer.completeError(message);
      } else if (message is SendPort) {
        mainToIsolateSendPort = message;
      }
    });

    final args = [
      receivePort.sendPort,
      receiver.toJson(),
    ];

    final isolate = await Isolate.spawn(
      PayjoinReceiverWorker.run,
      args,
    );

    _activePollers[receiver.id()] = PayjoinPollerSession(isolate, receivePort);

    return completer.future;
  }

  void cleanupSessions() {
    for (final sessionId in _activePollers.keys) {
      _cleanupSession(sessionId);
    }
  }

  void _cleanupSession(String sessionId) {
    _activePollers[sessionId]?.close();
    _activePollers.remove(sessionId);
  }
}

class PayjoinPollerSession {
  final Isolate isolate;
  final ReceivePort port;

  PayjoinPollerSession(this.isolate, this.port);

  void close() {
    isolate.kill();
    port.close();
  }
}