1 files changed, 910 insertions, 0 deletions
diff --git a/packages/taler-wallet-core/src/operations/pay-peer-pull-credit.ts b/packages/taler-wallet-core/src/operations/pay-peer-pull-credit.ts
new file mode 100644
index 000000000..b9c9728a1
--- /dev/null
+++ b/packages/taler-wallet-core/src/operations/pay-peer-pull-credit.ts
@@ -0,0 +1,910 @@
+/*
+ This file is part of GNU Taler
+ (C) 2022-2023 Taler Systems S.A.
+
+ GNU Taler is free software; you can redistribute it and/or modify it under the
+ terms of the GNU General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ GNU Taler; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+
+import {
+  AbsoluteTime,
+  Amounts,
+  CancellationToken,
+  CheckPeerPullCreditRequest,
+  CheckPeerPullCreditResponse,
+  ContractTermsUtil,
+  ExchangeReservePurseRequest,
+  HttpStatusCode,
+  InitiatePeerPullCreditRequest,
+  InitiatePeerPullCreditResponse,
+  Logger,
+  TalerPreciseTimestamp,
+  TransactionAction,
+  TransactionMajorState,
+  TransactionMinorState,
+  TransactionState,
+  TransactionType,
+  WalletAccountMergeFlags,
+  codecForAny,
+  codecForWalletKycUuid,
+  constructPayPullUri,
+  encodeCrock,
+  getRandomBytes,
+  j2s,
+} from "@gnu-taler/taler-util";
+import {
+  readSuccessResponseJsonOrErrorCode,
+  readSuccessResponseJsonOrThrow,
+  throwUnexpectedRequestError,
+} from "@gnu-taler/taler-util/http";
+import {
+  PeerPullPaymentInitiationRecord,
+  PeerPullPaymentInitiationStatus,
+  WithdrawalGroupStatus,
+  WithdrawalRecordType,
+  updateExchangeFromUrl,
+} from "../index.js";
+import { InternalWalletState } from "../internal-wallet-state.js";
+import { PendingTaskType } from "../pending-types.js";
+import { assertUnreachable } from "../util/assertUnreachable.js";
+import { checkDbInvariant } from "../util/invariants.js";
+import {
+  OperationAttemptResult,
+  OperationAttemptResultType,
+  constructTaskIdentifier,
+} from "../util/retries.js";
+import {
+  LongpollResult,
+  resetOperationTimeout,
+  runLongpollAsync,
+  runOperationWithErrorReporting,
+} from "./common.js";
+import {
+  codecForExchangePurseStatus,
+  getMergeReserveInfo,
+  talerPaytoFromExchangeReserve,
+} from "./pay-peer-common.js";
+import {
+  constructTransactionIdentifier,
+  notifyTransition,
+  stopLongpolling,
+} from "./transactions.js";
+import {
+  checkWithdrawalKycStatus,
+  getExchangeWithdrawalInfo,
+  internalCreateWithdrawalGroup,
+  processWithdrawalGroup,
+} from "./withdraw.js";
+
+const logger = new Logger("pay-peer-pull-credit.ts");
+
+export async function queryPurseForPeerPullCredit(
+  ws: InternalWalletState,
+  pullIni: PeerPullPaymentInitiationRecord,
+  cancellationToken: CancellationToken,
+): Promise<LongpollResult> {
+  const purseDepositUrl = new URL(
+    `purses/${pullIni.pursePub}/deposit`,
+    pullIni.exchangeBaseUrl,
+  );
+  purseDepositUrl.searchParams.set("timeout_ms", "30000");
+  logger.info(`querying purse status via ${purseDepositUrl.href}`);
+  const resp = await ws.http.get(purseDepositUrl.href, {
+    timeout: { d_ms: 60000 },
+    cancellationToken,
+  });
+
+  logger.info(`purse status code: HTTP ${resp.status}`);
+
+  const result = await readSuccessResponseJsonOrErrorCode(
+    resp,
+    codecForExchangePurseStatus(),
+  );
+
+  if (result.isError) {
+    logger.info(`got purse status error, EC=${result.talerErrorResponse.code}`);
+    if (resp.status === 404) {
+      return { ready: false };
+    } else {
+      throwUnexpectedRequestError(resp, result.talerErrorResponse);
+    }
+  }
+
+  if (!result.response.deposit_timestamp) {
+    logger.info("purse not ready yet (no deposit)");
+    return { ready: false };
+  }
+
+  const reserve = await ws.db
+    .mktx((x) => [x.reserves])
+    .runReadOnly(async (tx) => {
+      return await tx.reserves.get(pullIni.mergeReserveRowId);
+    });
+
+  if (!reserve) {
+    throw Error("reserve for peer pull credit not found in wallet DB");
+  }
+
+  await internalCreateWithdrawalGroup(ws, {
+    amount: Amounts.parseOrThrow(pullIni.amount),
+    wgInfo: {
+      withdrawalType: WithdrawalRecordType.PeerPullCredit,
+      contractTerms: pullIni.contractTerms,
+      contractPriv: pullIni.contractPriv,
+    },
+    forcedWithdrawalGroupId: pullIni.withdrawalGroupId,
+    exchangeBaseUrl: pullIni.exchangeBaseUrl,
+    reserveStatus: WithdrawalGroupStatus.PendingQueryingStatus,
+    reserveKeyPair: {
+      priv: reserve.reservePriv,
+      pub: reserve.reservePub,
+    },
+  });
+
+  await ws.db
+    .mktx((x) => [x.peerPullPaymentInitiations])
+    .runReadWrite(async (tx) => {
+      const finPi = await tx.peerPullPaymentInitiations.get(pullIni.pursePub);
+      if (!finPi) {
+        logger.warn("peerPullPaymentInitiation not found anymore");
+        return;
+      }
+      if (finPi.status === PeerPullPaymentInitiationStatus.PendingReady) {
+        finPi.status = PeerPullPaymentInitiationStatus.DonePurseDeposited;
+      }
+      await tx.peerPullPaymentInitiations.put(finPi);
+    });
+  return {
+    ready: true,
+  };
+}
+
+export async function processPeerPullCredit(
+  ws: InternalWalletState,
+  pursePub: string,
+): Promise<OperationAttemptResult> {
+  const pullIni = await ws.db
+    .mktx((x) => [x.peerPullPaymentInitiations])
+    .runReadOnly(async (tx) => {
+      return tx.peerPullPaymentInitiations.get(pursePub);
+    });
+  if (!pullIni) {
+    throw Error("peer pull payment initiation not found in database");
+  }
+
+  const retryTag = constructTaskIdentifier({
+    tag: PendingTaskType.PeerPullCredit,
+    pursePub,
+  });
+
+  // We're already running!
+  if (ws.activeLongpoll[retryTag]) {
+    logger.info("peer-pull-credit already in long-polling, returning!");
+    return {
+      type: OperationAttemptResultType.Longpoll,
+    };
+  }
+
+  logger.trace(`processing ${retryTag}, status=${pullIni.status}`);
+
+  switch (pullIni.status) {
+    case PeerPullPaymentInitiationStatus.DonePurseDeposited: {
+      // We implement this case so that the "retry" action on a peer-pull-credit transaction
+      // also retries the withdrawal task.
+
+      logger.warn(
+        "peer pull payment initiation is already finished, retrying withdrawal",
+      );
+
+      const withdrawalGroupId = pullIni.withdrawalGroupId;
+
+      if (withdrawalGroupId) {
+        const taskId = constructTaskIdentifier({
+          tag: PendingTaskType.Withdraw,
+          withdrawalGroupId,
+        });
+        stopLongpolling(ws, taskId);
+        await resetOperationTimeout(ws, taskId);
+        await runOperationWithErrorReporting(ws, taskId, () =>
+          processWithdrawalGroup(ws, withdrawalGroupId),
+        );
+      }
+      return {
+        type: OperationAttemptResultType.Finished,
+        result: undefined,
+      };
+    }
+    case PeerPullPaymentInitiationStatus.PendingReady:
+      runLongpollAsync(ws, retryTag, async (cancellationToken) =>
+        queryPurseForPeerPullCredit(ws, pullIni, cancellationToken),
+      );
+      logger.trace(
+        "returning early from processPeerPullCredit for long-polling in background",
+      );
+      return {
+        type: OperationAttemptResultType.Longpoll,
+      };
+    case PeerPullPaymentInitiationStatus.PendingMergeKycRequired: {
+      const transactionId = constructTransactionIdentifier({
+        tag: TransactionType.PeerPullCredit,
+        pursePub: pullIni.pursePub,
+      });
+      if (pullIni.kycInfo) {
+        await checkWithdrawalKycStatus(
+          ws,
+          pullIni.exchangeBaseUrl,
+          transactionId,
+          pullIni.kycInfo,
+          "individual",
+        );
+      }
+      break;
+    }
+    case PeerPullPaymentInitiationStatus.PendingCreatePurse:
+      break;
+    default:
+      throw Error(`unknown PeerPullPaymentInitiationStatus ${pullIni.status}`);
+  }
+
+  const mergeReserve = await ws.db
+    .mktx((x) => [x.reserves])
+    .runReadOnly(async (tx) => {
+      return tx.reserves.get(pullIni.mergeReserveRowId);
+    });
+
+  if (!mergeReserve) {
+    throw Error("merge reserve for peer pull payment not found in database");
+  }
+
+  const purseFee = Amounts.stringify(Amounts.zeroOfAmount(pullIni.amount));
+
+  const reservePayto = talerPaytoFromExchangeReserve(
+    pullIni.exchangeBaseUrl,
+    mergeReserve.reservePub,
+  );
+
+  const econtractResp = await ws.cryptoApi.encryptContractForDeposit({
+    contractPriv: pullIni.contractPriv,
+    contractPub: pullIni.contractPub,
+    contractTerms: pullIni.contractTerms,
+    pursePriv: pullIni.pursePriv,
+    pursePub: pullIni.pursePub,
+  });
+
+  const purseExpiration = pullIni.contractTerms.purse_expiration;
+  const sigRes = await ws.cryptoApi.signReservePurseCreate({
+    contractTermsHash: pullIni.contractTermsHash,
+    flags: WalletAccountMergeFlags.CreateWithPurseFee,
+    mergePriv: pullIni.mergePriv,
+    mergeTimestamp: TalerPreciseTimestamp.round(pullIni.mergeTimestamp),
+    purseAmount: pullIni.contractTerms.amount,
+    purseExpiration: purseExpiration,
+    purseFee: purseFee,
+    pursePriv: pullIni.pursePriv,
+    pursePub: pullIni.pursePub,
+    reservePayto,
+    reservePriv: mergeReserve.reservePriv,
+  });
+
+  const reservePurseReqBody: ExchangeReservePurseRequest = {
+    merge_sig: sigRes.mergeSig,
+    merge_timestamp: TalerPreciseTimestamp.round(pullIni.mergeTimestamp),
+    h_contract_terms: pullIni.contractTermsHash,
+    merge_pub: pullIni.mergePub,
+    min_age: 0,
+    purse_expiration: purseExpiration,
+    purse_fee: purseFee,
+    purse_pub: pullIni.pursePub,
+    purse_sig: sigRes.purseSig,
+    purse_value: pullIni.contractTerms.amount,
+    reserve_sig: sigRes.accountSig,
+    econtract: econtractResp.econtract,
+  };
+
+  logger.info(`reserve purse request: ${j2s(reservePurseReqBody)}`);
+
+  const reservePurseMergeUrl = new URL(
+    `reserves/${mergeReserve.reservePub}/purse`,
+    pullIni.exchangeBaseUrl,
+  );
+
+  const httpResp = await ws.http.postJson(
+    reservePurseMergeUrl.href,
+    reservePurseReqBody,
+  );
+
+  if (httpResp.status === HttpStatusCode.UnavailableForLegalReasons) {
+    const respJson = await httpResp.json();
+    const kycPending = codecForWalletKycUuid().decode(respJson);
+    logger.info(`kyc uuid response: ${j2s(kycPending)}`);
+
+    await ws.db
+      .mktx((x) => [x.peerPullPaymentInitiations])
+      .runReadWrite(async (tx) => {
+        const peerIni = await tx.peerPullPaymentInitiations.get(pursePub);
+        if (!peerIni) {
+          return;
+        }
+        peerIni.kycInfo = {
+          paytoHash: kycPending.h_payto,
+          requirementRow: kycPending.requirement_row,
+        };
+        peerIni.status =
+          PeerPullPaymentInitiationStatus.PendingMergeKycRequired;
+        await tx.peerPullPaymentInitiations.put(peerIni);
+      });
+    return {
+      type: OperationAttemptResultType.Pending,
+      result: undefined,
+    };
+  }
+
+  const resp = await readSuccessResponseJsonOrThrow(httpResp, codecForAny());
+
+  logger.info(`reserve merge response: ${j2s(resp)}`);
+
+  await ws.db
+    .mktx((x) => [x.peerPullPaymentInitiations])
+    .runReadWrite(async (tx) => {
+      const pi2 = await tx.peerPullPaymentInitiations.get(pursePub);
+      if (!pi2) {
+        return;
+      }
+      pi2.status = PeerPullPaymentInitiationStatus.PendingReady;
+      await tx.peerPullPaymentInitiations.put(pi2);
+    });
+
+  return {
+    type: OperationAttemptResultType.Finished,
+    result: undefined,
+  };
+}
+
+/**
+ * Check fees and available exchanges for a peer push payment initiation.
+ */
+export async function checkPeerPullPaymentInitiation(
+  ws: InternalWalletState,
+  req: CheckPeerPullCreditRequest,
+): Promise<CheckPeerPullCreditResponse> {
+  // FIXME: We don't support exchanges with purse fees yet.
+  // Select an exchange where we have money in the specified currency
+  // FIXME: How do we handle regional currency scopes here? Is it an additional input?
+
+  logger.trace("checking peer-pull-credit fees");
+
+  const currency = Amounts.currencyOf(req.amount);
+  let exchangeUrl;
+  if (req.exchangeBaseUrl) {
+    exchangeUrl = req.exchangeBaseUrl;
+  } else {
+    exchangeUrl = await getPreferredExchangeForCurrency(ws, currency);
+  }
+
+  if (!exchangeUrl) {
+    throw Error("no exchange found for initiating a peer pull payment");
+  }
+
+  logger.trace(`found ${exchangeUrl} as preferred exchange`);
+
+  const wi = await getExchangeWithdrawalInfo(
+    ws,
+    exchangeUrl,
+    Amounts.parseOrThrow(req.amount),
+    undefined,
+  );
+
+  logger.trace(`got withdrawal info`);
+
+  return {
+    exchangeBaseUrl: exchangeUrl,
+    amountEffective: wi.withdrawalAmountEffective,
+    amountRaw: req.amount,
+  };
+}
+
+/**
+ * Find a preferred exchange based on when we withdrew last from this exchange.
+ */
+async function getPreferredExchangeForCurrency(
+  ws: InternalWalletState,
+  currency: string,
+): Promise<string | undefined> {
+  // Find an exchange with the matching currency.
+  // Prefer exchanges with the most recent withdrawal.
+  const url = await ws.db
+    .mktx((x) => [x.exchanges])
+    .runReadOnly(async (tx) => {
+      const exchanges = await tx.exchanges.iter().toArray();
+      let candidate = undefined;
+      for (const e of exchanges) {
+        if (e.detailsPointer?.currency !== currency) {
+          continue;
+        }
+        if (!candidate) {
+          candidate = e;
+          continue;
+        }
+        if (candidate.lastWithdrawal && !e.lastWithdrawal) {
+          continue;
+        }
+        if (candidate.lastWithdrawal && e.lastWithdrawal) {
+          if (
+            AbsoluteTime.cmp(
+              AbsoluteTime.fromPreciseTimestamp(e.lastWithdrawal),
+              AbsoluteTime.fromPreciseTimestamp(candidate.lastWithdrawal),
+            ) > 0
+          ) {
+            candidate = e;
+          }
+        }
+      }
+      if (candidate) {
+        return candidate.baseUrl;
+      }
+      return undefined;
+    });
+  return url;
+}
+
+/**
+ * Initiate a peer pull payment.
+ */
+export async function initiatePeerPullPayment(
+  ws: InternalWalletState,
+  req: InitiatePeerPullCreditRequest,
+): Promise<InitiatePeerPullCreditResponse> {
+  const currency = Amounts.currencyOf(req.partialContractTerms.amount);
+  let maybeExchangeBaseUrl: string | undefined;
+  if (req.exchangeBaseUrl) {
+    maybeExchangeBaseUrl = req.exchangeBaseUrl;
+  } else {
+    maybeExchangeBaseUrl = await getPreferredExchangeForCurrency(ws, currency);
+  }
+
+  if (!maybeExchangeBaseUrl) {
+    throw Error("no exchange found for initiating a peer pull payment");
+  }
+
+  const exchangeBaseUrl = maybeExchangeBaseUrl;
+
+  await updateExchangeFromUrl(ws, exchangeBaseUrl);
+
+  const mergeReserveInfo = await getMergeReserveInfo(ws, {
+    exchangeBaseUrl: exchangeBaseUrl,
+  });
+
+  const mergeTimestamp = TalerPreciseTimestamp.now();
+
+  const pursePair = await ws.cryptoApi.createEddsaKeypair({});
+  const mergePair = await ws.cryptoApi.createEddsaKeypair({});
+
+  const contractTerms = req.partialContractTerms;
+
+  const hContractTerms = ContractTermsUtil.hashContractTerms(contractTerms);
+
+  const contractKeyPair = await ws.cryptoApi.createEddsaKeypair({});
+
+  const withdrawalGroupId = encodeCrock(getRandomBytes(32));
+
+  const mergeReserveRowId = mergeReserveInfo.rowId;
+  checkDbInvariant(!!mergeReserveRowId);
+
+  const wi = await getExchangeWithdrawalInfo(
+    ws,
+    exchangeBaseUrl,
+    Amounts.parseOrThrow(req.partialContractTerms.amount),
+    undefined,
+  );
+
+  await ws.db
+    .mktx((x) => [x.peerPullPaymentInitiations, x.contractTerms])
+    .runReadWrite(async (tx) => {
+      await tx.peerPullPaymentInitiations.put({
+        amount: req.partialContractTerms.amount,
+        contractTermsHash: hContractTerms,
+        exchangeBaseUrl: exchangeBaseUrl,
+        pursePriv: pursePair.priv,
+        pursePub: pursePair.pub,
+        mergePriv: mergePair.priv,
+        mergePub: mergePair.pub,
+        status: PeerPullPaymentInitiationStatus.PendingCreatePurse,
+        contractTerms: contractTerms,
+        mergeTimestamp,
+        mergeReserveRowId: mergeReserveRowId,
+        contractPriv: contractKeyPair.priv,
+        contractPub: contractKeyPair.pub,
+        withdrawalGroupId,
+        estimatedAmountEffective: wi.withdrawalAmountEffective,
+      });
+      await tx.contractTerms.put({
+        contractTermsRaw: contractTerms,
+        h: hContractTerms,
+      });
+    });
+
+  // FIXME: Should we somehow signal to the client
+  // whether purse creation has failed, or does the client/
+  // check this asynchronously from the transaction status?
+
+  const taskId = constructTaskIdentifier({
+    tag: PendingTaskType.PeerPullCredit,
+    pursePub: pursePair.pub,
+  });
+
+  await runOperationWithErrorReporting(ws, taskId, async () => {
+    return processPeerPullCredit(ws, pursePair.pub);
+  });
+
+  const transactionId = constructTransactionIdentifier({
+    tag: TransactionType.PeerPullCredit,
+    pursePub: pursePair.pub,
+  });
+
+  return {
+    talerUri: constructPayPullUri({
+      exchangeBaseUrl: exchangeBaseUrl,
+      contractPriv: contractKeyPair.priv,
+    }),
+    transactionId,
+  };
+}
+
+export async function suspendPeerPullCreditTransaction(
+  ws: InternalWalletState,
+  pursePub: string,
+) {
+  const taskId = constructTaskIdentifier({
+    tag: PendingTaskType.PeerPullCredit,
+    pursePub,
+  });
+  const transactionId = constructTransactionIdentifier({
+    tag: TransactionType.PeerPullCredit,
+    pursePub,
+  });
+  stopLongpolling(ws, taskId);
+  const transitionInfo = await ws.db
+    .mktx((x) => [x.peerPullPaymentInitiations])
+    .runReadWrite(async (tx) => {
+      const pullCreditRec = await tx.peerPullPaymentInitiations.get(pursePub);
+      if (!pullCreditRec) {
+        logger.warn(`peer pull credit ${pursePub} not found`);
+        return;
+      }
+      let newStatus: PeerPullPaymentInitiationStatus | undefined = undefined;
+      switch (pullCreditRec.status) {
+        case PeerPullPaymentInitiationStatus.PendingCreatePurse:
+          newStatus = PeerPullPaymentInitiationStatus.SuspendedCreatePurse;
+          break;
+        case PeerPullPaymentInitiationStatus.PendingMergeKycRequired:
+          newStatus = PeerPullPaymentInitiationStatus.SuspendedMergeKycRequired;
+          break;
+        case PeerPullPaymentInitiationStatus.PendingWithdrawing:
+          newStatus = PeerPullPaymentInitiationStatus.SuspendedWithdrawing;
+          break;
+        case PeerPullPaymentInitiationStatus.PendingReady:
+          newStatus = PeerPullPaymentInitiationStatus.SuspendedReady;
+          break;
+        case PeerPullPaymentInitiationStatus.AbortingDeletePurse:
+          newStatus =
+            PeerPullPaymentInitiationStatus.SuspendedAbortingDeletePurse;
+          break;
+        case PeerPullPaymentInitiationStatus.DonePurseDeposited:
+        case PeerPullPaymentInitiationStatus.SuspendedCreatePurse:
+        case PeerPullPaymentInitiationStatus.SuspendedMergeKycRequired:
+        case PeerPullPaymentInitiationStatus.SuspendedReady:
+        case PeerPullPaymentInitiationStatus.SuspendedWithdrawing:
+        case PeerPullPaymentInitiationStatus.Aborted:
+        case PeerPullPaymentInitiationStatus.Failed:
+        case PeerPullPaymentInitiationStatus.SuspendedAbortingDeletePurse:
+          break;
+        default:
+          assertUnreachable(pullCreditRec.status);
+      }
+      if (newStatus != null) {
+        const oldTxState = computePeerPullCreditTransactionState(pullCreditRec);
+        pullCreditRec.status = newStatus;
+        const newTxState = computePeerPullCreditTransactionState(pullCreditRec);
+        await tx.peerPullPaymentInitiations.put(pullCreditRec);
+        return {
+          oldTxState,
+          newTxState,
+        };
+      }
+      return undefined;
+    });
+  notifyTransition(ws, transactionId, transitionInfo);
+}
+
+export async function abortPeerPullCreditTransaction(
+  ws: InternalWalletState,
+  pursePub: string,
+) {
+  const taskId = constructTaskIdentifier({
+    tag: PendingTaskType.PeerPullCredit,
+    pursePub,
+  });
+  const transactionId = constructTransactionIdentifier({
+    tag: TransactionType.PeerPullCredit,
+    pursePub,
+  });
+  stopLongpolling(ws, taskId);
+  const transitionInfo = await ws.db
+    .mktx((x) => [x.peerPullPaymentInitiations])
+    .runReadWrite(async (tx) => {
+      const pullCreditRec = await tx.peerPullPaymentInitiations.get(pursePub);
+      if (!pullCreditRec) {
+        logger.warn(`peer pull credit ${pursePub} not found`);
+        return;
+      }
+      let newStatus: PeerPullPaymentInitiationStatus | undefined = undefined;
+      switch (pullCreditRec.status) {
+        case PeerPullPaymentInitiationStatus.PendingCreatePurse:
+        case PeerPullPaymentInitiationStatus.PendingMergeKycRequired:
+          newStatus = PeerPullPaymentInitiationStatus.AbortingDeletePurse;
+          break;
+        case PeerPullPaymentInitiationStatus.PendingWithdrawing:
+          throw Error("can't abort anymore");
+        case PeerPullPaymentInitiationStatus.PendingReady:
+          newStatus = PeerPullPaymentInitiationStatus.AbortingDeletePurse;
+          break;
+        case PeerPullPaymentInitiationStatus.DonePurseDeposited:
+        case PeerPullPaymentInitiationStatus.SuspendedCreatePurse:
+        case PeerPullPaymentInitiationStatus.SuspendedMergeKycRequired:
+        case PeerPullPaymentInitiationStatus.SuspendedReady:
+        case PeerPullPaymentInitiationStatus.SuspendedWithdrawing:
+        case PeerPullPaymentInitiationStatus.Aborted:
+        case PeerPullPaymentInitiationStatus.AbortingDeletePurse:
+        case PeerPullPaymentInitiationStatus.Failed:
+        case PeerPullPaymentInitiationStatus.SuspendedAbortingDeletePurse:
+          break;
+        default:
+          assertUnreachable(pullCreditRec.status);
+      }
+      if (newStatus != null) {
+        const oldTxState = computePeerPullCreditTransactionState(pullCreditRec);
+        pullCreditRec.status = newStatus;
+        const newTxState = computePeerPullCreditTransactionState(pullCreditRec);
+        await tx.peerPullPaymentInitiations.put(pullCreditRec);
+        return {
+          oldTxState,
+          newTxState,
+        };
+      }
+      return undefined;
+    });
+  notifyTransition(ws, transactionId, transitionInfo);
+}
+
+export async function failPeerPullCreditTransaction(
+  ws: InternalWalletState,
+  pursePub: string,
+) {
+  const taskId = constructTaskIdentifier({
+    tag: PendingTaskType.PeerPullCredit,
+    pursePub,
+  });
+  const transactionId = constructTransactionIdentifier({
+    tag: TransactionType.PeerPullCredit,
+    pursePub,
+  });
+  stopLongpolling(ws, taskId);
+  const transitionInfo = await ws.db
+    .mktx((x) => [x.peerPullPaymentInitiations])
+    .runReadWrite(async (tx) => {
+      const pullCreditRec = await tx.peerPullPaymentInitiations.get(pursePub);
+      if (!pullCreditRec) {
+        logger.warn(`peer pull credit ${pursePub} not found`);
+        return;
+      }
+      let newStatus: PeerPullPaymentInitiationStatus | undefined = undefined;
+      switch (pullCreditRec.status) {
+        case PeerPullPaymentInitiationStatus.PendingCreatePurse:
+        case PeerPullPaymentInitiationStatus.PendingMergeKycRequired:
+        case PeerPullPaymentInitiationStatus.PendingWithdrawing:
+        case PeerPullPaymentInitiationStatus.PendingReady:
+        case PeerPullPaymentInitiationStatus.DonePurseDeposited:
+        case PeerPullPaymentInitiationStatus.SuspendedCreatePurse:
+        case PeerPullPaymentInitiationStatus.SuspendedMergeKycRequired:
+        case PeerPullPaymentInitiationStatus.SuspendedReady:
+        case PeerPullPaymentInitiationStatus.SuspendedWithdrawing:
+        case PeerPullPaymentInitiationStatus.Aborted:
+        case PeerPullPaymentInitiationStatus.Failed:
+          break;
+        case PeerPullPaymentInitiationStatus.AbortingDeletePurse:
+        case PeerPullPaymentInitiationStatus.SuspendedAbortingDeletePurse:
+          newStatus = PeerPullPaymentInitiationStatus.Failed;
+          break;
+        default:
+          assertUnreachable(pullCreditRec.status);
+      }
+      if (newStatus != null) {
+        const oldTxState = computePeerPullCreditTransactionState(pullCreditRec);
+        pullCreditRec.status = newStatus;
+        const newTxState = computePeerPullCreditTransactionState(pullCreditRec);
+        await tx.peerPullPaymentInitiations.put(pullCreditRec);
+        return {
+          oldTxState,
+          newTxState,
+        };
+      }
+      return undefined;
+    });
+  notifyTransition(ws, transactionId, transitionInfo);
+}
+
+export async function resumePeerPullCreditTransaction(
+  ws: InternalWalletState,
+  pursePub: string,
+) {
+  const taskId = constructTaskIdentifier({
+    tag: PendingTaskType.PeerPullCredit,
+    pursePub,
+  });
+  const transactionId = constructTransactionIdentifier({
+    tag: TransactionType.PeerPullCredit,
+    pursePub,
+  });
+  stopLongpolling(ws, taskId);
+  const transitionInfo = await ws.db
+    .mktx((x) => [x.peerPullPaymentInitiations])
+    .runReadWrite(async (tx) => {
+      const pullCreditRec = await tx.peerPullPaymentInitiations.get(pursePub);
+      if (!pullCreditRec) {
+        logger.warn(`peer pull credit ${pursePub} not found`);
+        return;
+      }
+      let newStatus: PeerPullPaymentInitiationStatus | undefined = undefined;
+      switch (pullCreditRec.status) {
+        case PeerPullPaymentInitiationStatus.PendingCreatePurse:
+        case PeerPullPaymentInitiationStatus.PendingMergeKycRequired:
+        case PeerPullPaymentInitiationStatus.PendingWithdrawing:
+        case PeerPullPaymentInitiationStatus.PendingReady:
+        case PeerPullPaymentInitiationStatus.AbortingDeletePurse:
+        case PeerPullPaymentInitiationStatus.DonePurseDeposited:
+        case PeerPullPaymentInitiationStatus.Failed:
+        case PeerPullPaymentInitiationStatus.Aborted:
+          break;
+        case PeerPullPaymentInitiationStatus.SuspendedCreatePurse:
+          newStatus = PeerPullPaymentInitiationStatus.PendingCreatePurse;
+          break;
+        case PeerPullPaymentInitiationStatus.SuspendedMergeKycRequired:
+          newStatus = PeerPullPaymentInitiationStatus.PendingMergeKycRequired;
+          break;
+        case PeerPullPaymentInitiationStatus.SuspendedReady:
+          newStatus = PeerPullPaymentInitiationStatus.PendingReady;
+          break;
+        case PeerPullPaymentInitiationStatus.SuspendedWithdrawing:
+          newStatus = PeerPullPaymentInitiationStatus.PendingWithdrawing;
+          break;
+        case PeerPullPaymentInitiationStatus.SuspendedAbortingDeletePurse:
+          newStatus = PeerPullPaymentInitiationStatus.AbortingDeletePurse;
+          break;
+        default:
+          assertUnreachable(pullCreditRec.status);
+      }
+      if (newStatus != null) {
+        const oldTxState = computePeerPullCreditTransactionState(pullCreditRec);
+        pullCreditRec.status = newStatus;
+        const newTxState = computePeerPullCreditTransactionState(pullCreditRec);
+        await tx.peerPullPaymentInitiations.put(pullCreditRec);
+        return {
+          oldTxState,
+          newTxState,
+        };
+      }
+      return undefined;
+    });
+  ws.workAvailable.trigger();
+  notifyTransition(ws, transactionId, transitionInfo);
+}
+
+export function computePeerPullCreditTransactionState(
+  pullCreditRecord: PeerPullPaymentInitiationRecord,
+): TransactionState {
+  switch (pullCreditRecord.status) {
+    case PeerPullPaymentInitiationStatus.PendingCreatePurse:
+      return {
+        major: TransactionMajorState.Pending,
+        minor: TransactionMinorState.CreatePurse,
+      };
+    case PeerPullPaymentInitiationStatus.PendingMergeKycRequired:
+      return {
+        major: TransactionMajorState.Pending,
+        minor: TransactionMinorState.MergeKycRequired,
+      };
+    case PeerPullPaymentInitiationStatus.PendingReady:
+      return {
+        major: TransactionMajorState.Pending,
+        minor: TransactionMinorState.Ready,
+      };
+    case PeerPullPaymentInitiationStatus.DonePurseDeposited:
+      return {
+        major: TransactionMajorState.Done,
+      };
+    case PeerPullPaymentInitiationStatus.PendingWithdrawing:
+      return {
+        major: TransactionMajorState.Pending,
+        minor: TransactionMinorState.Withdraw,
+      };
+    case PeerPullPaymentInitiationStatus.SuspendedCreatePurse:
+      return {
+        major: TransactionMajorState.Suspended,
+        minor: TransactionMinorState.CreatePurse,
+      };
+    case PeerPullPaymentInitiationStatus.SuspendedReady:
+      return {
+        major: TransactionMajorState.Suspended,
+        minor: TransactionMinorState.Ready,
+      };
+    case PeerPullPaymentInitiationStatus.SuspendedWithdrawing:
+      return {
+        major: TransactionMajorState.Pending,
+        minor: TransactionMinorState.Withdraw,
+      };
+    case PeerPullPaymentInitiationStatus.SuspendedMergeKycRequired:
+      return {
+        major: TransactionMajorState.Suspended,
+        minor: TransactionMinorState.MergeKycRequired,
+      };
+    case PeerPullPaymentInitiationStatus.Aborted:
+      return {
+        major: TransactionMajorState.Aborted,
+      };
+    case PeerPullPaymentInitiationStatus.AbortingDeletePurse:
+      return {
+        major: TransactionMajorState.Aborting,
+        minor: TransactionMinorState.DeletePurse,
+      };
+    case PeerPullPaymentInitiationStatus.Failed:
+      return {
+        major: TransactionMajorState.Failed,
+      };
+    case PeerPullPaymentInitiationStatus.SuspendedAbortingDeletePurse:
+      return {
+        major: TransactionMajorState.Aborting,
+        minor: TransactionMinorState.DeletePurse,
+      };
+  }
+}
+
+export function computePeerPullCreditTransactionActions(
+  pullCreditRecord: PeerPullPaymentInitiationRecord,
+): TransactionAction[] {
+  switch (pullCreditRecord.status) {
+    case PeerPullPaymentInitiationStatus.PendingCreatePurse:
+      return [TransactionAction.Abort, TransactionAction.Suspend];
+    case PeerPullPaymentInitiationStatus.PendingMergeKycRequired:
+      return [TransactionAction.Abort, TransactionAction.Suspend];
+    case PeerPullPaymentInitiationStatus.PendingReady:
+      return [TransactionAction.Abort, TransactionAction.Suspend];
+    case PeerPullPaymentInitiationStatus.DonePurseDeposited:
+      return [TransactionAction.Delete];
+    case PeerPullPaymentInitiationStatus.PendingWithdrawing:
+      return [TransactionAction.Abort, TransactionAction.Suspend];
+    case PeerPullPaymentInitiationStatus.SuspendedCreatePurse:
+      return [TransactionAction.Resume, TransactionAction.Abort];
+    case PeerPullPaymentInitiationStatus.SuspendedReady:
+      return [TransactionAction.Abort, TransactionAction.Resume];
+    case PeerPullPaymentInitiationStatus.SuspendedWithdrawing:
+      return [TransactionAction.Resume, TransactionAction.Fail];
+    case PeerPullPaymentInitiationStatus.SuspendedMergeKycRequired:
+      return [TransactionAction.Resume, TransactionAction.Fail];
+    case PeerPullPaymentInitiationStatus.Aborted:
+      return [TransactionAction.Delete];
+    case PeerPullPaymentInitiationStatus.AbortingDeletePurse:
+      return [TransactionAction.Suspend, TransactionAction.Fail];
+    case PeerPullPaymentInitiationStatus.Failed:
+      return [TransactionAction.Delete];
+    case PeerPullPaymentInitiationStatus.SuspendedAbortingDeletePurse:
+      return [TransactionAction.Resume, TransactionAction.Fail];
+  }
+}