/*
 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/>
 */

/**
 * Imports.
 */
import {
  AbsoluteTime,
  Amounts,
  CheckPeerPullCreditRequest,
  CheckPeerPullCreditResponse,
  ContractTermsUtil,
  ExchangeReservePurseRequest,
  ExchangeWalletKycStatus,
  HttpStatusCode,
  InitiatePeerPullCreditRequest,
  InitiatePeerPullCreditResponse,
  Logger,
  NotificationType,
  PeerContractTerms,
  TalerErrorCode,
  TalerPreciseTimestamp,
  TalerProtocolTimestamp,
  TalerUriAction,
  Transaction,
  TransactionAction,
  TransactionIdStr,
  TransactionMajorState,
  TransactionMinorState,
  TransactionState,
  TransactionType,
  WalletAccountMergeFlags,
  assertUnreachable,
  checkDbInvariant,
  codecForAccountKycStatus,
  codecForAny,
  codecForLegitimizationNeededResponse,
  encodeCrock,
  getRandomBytes,
  j2s,
  stringifyPayPullUri,
  stringifyTalerUri,
  talerPaytoFromExchangeReserve,
} from "@gnu-taler/taler-util";
import {
  readResponseJsonOrThrow,
  readSuccessResponseJsonOrThrow,
} from "@gnu-taler/taler-util/http";
import {
  PendingTaskType,
  TaskIdStr,
  TaskIdentifiers,
  TaskRunResult,
  TombstoneTag,
  TransactionContext,
  TransitionResult,
  TransitionResultType,
  constructTaskIdentifier,
  genericWaitForStateVal,
  requireExchangeTosAcceptedOrThrow,
  runWithClientCancellation,
} from "./common.js";
import {
  OperationRetryRecord,
  PeerPullCreditRecord,
  PeerPullPaymentCreditStatus,
  WalletDbAllStoresReadOnlyTransaction,
  WalletDbReadWriteTransaction,
  WalletDbStoresArr,
  WithdrawalGroupRecord,
  WithdrawalGroupStatus,
  WithdrawalRecordType,
  timestampOptionalPreciseFromDb,
  timestampPreciseFromDb,
  timestampPreciseToDb,
} from "./db.js";
import {
  BalanceThresholdCheckResult,
  checkIncomingAmountLegalUnderKycBalanceThreshold,
  fetchFreshExchange,
  getScopeForAllExchanges,
  handleStartExchangeWalletKyc,
} from "./exchanges.js";
import {
  codecForExchangePurseStatus,
  getMergeReserveInfo,
} from "./pay-peer-common.js";
import {
  TransitionInfo,
  constructTransactionIdentifier,
  isUnsuccessfulTransaction,
  notifyTransition,
} from "./transactions.js";
import { WalletExecutionContext } from "./wallet.js";
import {
  getExchangeWithdrawalInfo,
  internalCreateWithdrawalGroup,
  waitWithdrawalFinal,
} from "./withdraw.js";

const logger = new Logger("pay-peer-pull-credit.ts");

export class PeerPullCreditTransactionContext implements TransactionContext {
  readonly transactionId: TransactionIdStr;
  readonly taskId: TaskIdStr;

  constructor(
    public wex: WalletExecutionContext,
    public pursePub: string,
  ) {
    this.taskId = constructTaskIdentifier({
      tag: PendingTaskType.PeerPullCredit,
      pursePub,
    });
    this.transactionId = constructTransactionIdentifier({
      tag: TransactionType.PeerPullCredit,
      pursePub,
    });
  }

  /**
   * Transition a peer-pull-credit transaction.
   * Extra object stores may be accessed during the transition.
   */
  async transition<StoreNameArray extends WalletDbStoresArr = []>(
    opts: { extraStores?: StoreNameArray; transactionLabel?: string },
    f: (
      rec: PeerPullCreditRecord | undefined,
      tx: WalletDbReadWriteTransaction<
        [
          "peerPullCredit",
          "transactionsMeta",
          "operationRetries",
          "exchanges",
          "exchangeDetails",
          ...StoreNameArray,
        ]
      >,
    ) => Promise<TransitionResult<PeerPullCreditRecord>>,
  ): Promise<TransitionInfo | undefined> {
    const baseStores = [
      "peerPullCredit" as const,
      "transactionsMeta" as const,
      "operationRetries" as const,
      "exchanges" as const,
      "exchangeDetails" as const,
    ];
    const stores = opts.extraStores
      ? [...baseStores, ...opts.extraStores]
      : baseStores;

    let errorThrown: Error | undefined;
    const transitionInfo = await this.wex.db.runReadWriteTx(
      { storeNames: stores },
      async (tx) => {
        const rec = await tx.peerPullCredit.get(this.pursePub);
        let oldTxState: TransactionState;
        if (rec) {
          oldTxState = computePeerPullCreditTransactionState(rec);
        } else {
          oldTxState = {
            major: TransactionMajorState.None,
          };
        }
        let res: TransitionResult<PeerPullCreditRecord> | undefined;
        try {
          res = await f(rec, tx);
        } catch (error) {
          if (error instanceof Error) {
            errorThrown = error;
          }
          return undefined;
        }

        switch (res.type) {
          case TransitionResultType.Transition: {
            await tx.peerPullCredit.put(res.rec);
            await this.updateTransactionMeta(tx);
            const newTxState = computePeerPullCreditTransactionState(res.rec);
            return {
              oldTxState,
              newTxState,
            };
          }
          case TransitionResultType.Delete:
            await tx.peerPullCredit.delete(this.pursePub);
            await this.updateTransactionMeta(tx);
            return {
              oldTxState,
              newTxState: {
                major: TransactionMajorState.None,
              },
            };
          default:
            return undefined;
        }
      },
    );
    if (errorThrown) {
      throw errorThrown;
    }
    notifyTransition(this.wex, this.transactionId, transitionInfo);
    return transitionInfo;
  }

  async updateTransactionMeta(
    tx: WalletDbReadWriteTransaction<["peerPullCredit", "transactionsMeta"]>,
  ): Promise<void> {
    const ppcRec = await tx.peerPullCredit.get(this.pursePub);
    if (!ppcRec) {
      await tx.transactionsMeta.delete(this.transactionId);
      return;
    }
    await tx.transactionsMeta.put({
      transactionId: this.transactionId,
      status: ppcRec.status,
      timestamp: ppcRec.mergeTimestamp,
      currency: Amounts.currencyOf(ppcRec.amount),
      exchanges: [ppcRec.exchangeBaseUrl],
    });
  }

  /**
   * Get the full transaction details for the transaction.
   *
   * Returns undefined if the transaction is in a state where we do not have a
   * transaction item (e.g. if it was deleted).
   */
  async lookupFullTransaction(
    tx: WalletDbAllStoresReadOnlyTransaction,
  ): Promise<Transaction | undefined> {
    const pullCredit = await tx.peerPullCredit.get(this.pursePub);
    if (!pullCredit) {
      return undefined;
    }
    const ct = await tx.contractTerms.get(pullCredit.contractTermsHash);
    checkDbInvariant(!!ct, `no contract terms for p2p push ${this.pursePub}`);

    const peerContractTerms = ct.contractTermsRaw;

    let wsr: WithdrawalGroupRecord | undefined = undefined;
    let wsrOrt: OperationRetryRecord | undefined = undefined;
    if (pullCredit.withdrawalGroupId) {
      wsr = await tx.withdrawalGroups.get(pullCredit.withdrawalGroupId);
      if (wsr) {
        const withdrawalOpId = TaskIdentifiers.forWithdrawal(wsr);
        wsrOrt = await tx.operationRetries.get(withdrawalOpId);
      }
    }
    const pullCreditOpId =
      TaskIdentifiers.forPeerPullPaymentInitiation(pullCredit);
    let pullCreditOrt = await tx.operationRetries.get(pullCreditOpId);

    let kycUrl: string | undefined = undefined;
    if (pullCredit.kycPaytoHash) {
      kycUrl = new URL(
        `kyc-spa/${pullCredit.kycPaytoHash}`,
        pullCredit.exchangeBaseUrl,
      ).href;
    }

    if (wsr) {
      if (wsr.wgInfo.withdrawalType !== WithdrawalRecordType.PeerPullCredit) {
        throw Error(`Unexpected withdrawalType: ${wsr.wgInfo.withdrawalType}`);
      }
      /**
       * FIXME: this should be handled in the withdrawal process.
       * PeerPull withdrawal fails until reserve have funds but it is not
       * an error from the user perspective.
       */
      const silentWithdrawalErrorForInvoice =
        wsrOrt?.lastError &&
        wsrOrt.lastError.code ===
          TalerErrorCode.WALLET_WITHDRAWAL_GROUP_INCOMPLETE &&
        Object.values(wsrOrt.lastError.errorsPerCoin ?? {}).every((e) => {
          return (
            e.code === TalerErrorCode.WALLET_UNEXPECTED_REQUEST_ERROR &&
            e.httpStatusCode === 409
          );
        });
      const txState = computePeerPullCreditTransactionState(pullCredit);
      checkDbInvariant(wsr.instructedAmount !== undefined, "wg uninitialized");
      checkDbInvariant(wsr.denomsSel !== undefined, "wg uninitialized");
      checkDbInvariant(wsr.exchangeBaseUrl !== undefined, "wg uninitialized");
      return {
        type: TransactionType.PeerPullCredit,
        txState,
        scopes: await getScopeForAllExchanges(tx, [pullCredit.exchangeBaseUrl]),
        txActions: computePeerPullCreditTransactionActions(pullCredit),
        amountEffective: isUnsuccessfulTransaction(txState)
          ? Amounts.stringify(Amounts.zeroOfAmount(wsr.instructedAmount))
          : Amounts.stringify(wsr.denomsSel.totalCoinValue),
        amountRaw: Amounts.stringify(wsr.instructedAmount),
        exchangeBaseUrl: wsr.exchangeBaseUrl,
        timestamp: timestampPreciseFromDb(pullCredit.mergeTimestamp),
        info: {
          expiration: peerContractTerms.purse_expiration,
          summary: peerContractTerms.summary,
        },
        talerUri: stringifyPayPullUri({
          exchangeBaseUrl: wsr.exchangeBaseUrl,
          contractPriv: wsr.wgInfo.contractPriv,
        }),
        transactionId: constructTransactionIdentifier({
          tag: TransactionType.PeerPullCredit,
          pursePub: pullCredit.pursePub,
        }),
        // FIXME: Is this the KYC URL of the withdrawal group?!
        kycUrl: kycUrl,
        ...(wsrOrt?.lastError
          ? {
              error: silentWithdrawalErrorForInvoice
                ? undefined
                : wsrOrt.lastError,
            }
          : {}),
      };
    }

    const txState = computePeerPullCreditTransactionState(pullCredit);
    return {
      type: TransactionType.PeerPullCredit,
      txState,
      scopes: await getScopeForAllExchanges(tx, [pullCredit.exchangeBaseUrl]),
      txActions: computePeerPullCreditTransactionActions(pullCredit),
      amountEffective: isUnsuccessfulTransaction(txState)
        ? Amounts.stringify(Amounts.zeroOfAmount(peerContractTerms.amount))
        : Amounts.stringify(pullCredit.estimatedAmountEffective),
      amountRaw: Amounts.stringify(peerContractTerms.amount),
      exchangeBaseUrl: pullCredit.exchangeBaseUrl,
      timestamp: timestampPreciseFromDb(pullCredit.mergeTimestamp),
      info: {
        expiration: peerContractTerms.purse_expiration,
        summary: peerContractTerms.summary,
      },
      talerUri: stringifyPayPullUri({
        exchangeBaseUrl: pullCredit.exchangeBaseUrl,
        contractPriv: pullCredit.contractPriv,
      }),
      transactionId: constructTransactionIdentifier({
        tag: TransactionType.PeerPullCredit,
        pursePub: pullCredit.pursePub,
      }),
      kycUrl,
      kycAccessToken: pullCredit.kycAccessToken,
      kycPaytoHash: pullCredit.kycPaytoHash,
      ...(pullCreditOrt?.lastError ? { error: pullCreditOrt.lastError } : {}),
    };
  }

  async deleteTransaction(): Promise<void> {
    const { wex: ws, pursePub } = this;
    await ws.db.runReadWriteTx(
      {
        storeNames: [
          "withdrawalGroups",
          "peerPullCredit",
          "tombstones",
          "transactionsMeta",
        ],
      },
      async (tx) => {
        const pullIni = await tx.peerPullCredit.get(pursePub);
        if (!pullIni) {
          return;
        }
        if (pullIni.withdrawalGroupId) {
          const withdrawalGroupId = pullIni.withdrawalGroupId;
          const withdrawalGroupRecord =
            await tx.withdrawalGroups.get(withdrawalGroupId);
          if (withdrawalGroupRecord) {
            await tx.withdrawalGroups.delete(withdrawalGroupId);
            await tx.tombstones.put({
              id: TombstoneTag.DeleteWithdrawalGroup + ":" + withdrawalGroupId,
            });
          }
        }
        await tx.peerPullCredit.delete(pursePub);
        await this.updateTransactionMeta(tx);
        await tx.tombstones.put({
          id: TombstoneTag.DeletePeerPullCredit + ":" + pursePub,
        });
      },
    );

    return;
  }

  async suspendTransaction(): Promise<void> {
    const { wex, pursePub, taskId: retryTag, transactionId } = this;
    const transitionInfo = await wex.db.runReadWriteTx(
      { storeNames: ["peerPullCredit", "transactionsMeta"] },
      async (tx) => {
        const pullCreditRec = await tx.peerPullCredit.get(pursePub);
        if (!pullCreditRec) {
          logger.warn(`peer pull credit ${pursePub} not found`);
          return;
        }
        let newStatus: PeerPullPaymentCreditStatus | undefined = undefined;
        switch (pullCreditRec.status) {
          case PeerPullPaymentCreditStatus.PendingCreatePurse:
            newStatus = PeerPullPaymentCreditStatus.SuspendedCreatePurse;
            break;
          case PeerPullPaymentCreditStatus.PendingMergeKycRequired:
            newStatus = PeerPullPaymentCreditStatus.SuspendedMergeKycRequired;
            break;
          case PeerPullPaymentCreditStatus.PendingWithdrawing:
            newStatus = PeerPullPaymentCreditStatus.SuspendedWithdrawing;
            break;
          case PeerPullPaymentCreditStatus.PendingReady:
            newStatus = PeerPullPaymentCreditStatus.SuspendedReady;
            break;
          case PeerPullPaymentCreditStatus.AbortingDeletePurse:
            newStatus =
              PeerPullPaymentCreditStatus.SuspendedAbortingDeletePurse;
            break;
          case PeerPullPaymentCreditStatus.PendingBalanceKycRequired:
            newStatus = PeerPullPaymentCreditStatus.SuspendedBalanceKycRequired;
            break;
          case PeerPullPaymentCreditStatus.PendingBalanceKycInit:
            newStatus = PeerPullPaymentCreditStatus.SuspendedBalanceKycInit;
            break;
          case PeerPullPaymentCreditStatus.SuspendedBalanceKycRequired:
          case PeerPullPaymentCreditStatus.SuspendedCreatePurse:
          case PeerPullPaymentCreditStatus.SuspendedMergeKycRequired:
          case PeerPullPaymentCreditStatus.SuspendedReady:
          case PeerPullPaymentCreditStatus.SuspendedWithdrawing:
          case PeerPullPaymentCreditStatus.SuspendedBalanceKycInit:
          case PeerPullPaymentCreditStatus.SuspendedAbortingDeletePurse:
          case PeerPullPaymentCreditStatus.Done:
          case PeerPullPaymentCreditStatus.Aborted:
          case PeerPullPaymentCreditStatus.Failed:
          case PeerPullPaymentCreditStatus.Expired:
            break;
          default:
            assertUnreachable(pullCreditRec.status);
        }
        if (newStatus != null) {
          const oldTxState =
            computePeerPullCreditTransactionState(pullCreditRec);
          pullCreditRec.status = newStatus;
          const newTxState =
            computePeerPullCreditTransactionState(pullCreditRec);
          await tx.peerPullCredit.put(pullCreditRec);
          await this.updateTransactionMeta(tx);
          return {
            oldTxState,
            newTxState,
          };
        }
        return undefined;
      },
    );
    wex.taskScheduler.stopShepherdTask(retryTag);
    notifyTransition(wex, transactionId, transitionInfo);
  }

  async failTransaction(): Promise<void> {
    const { wex, pursePub, taskId: retryTag, transactionId } = this;
    const transitionInfo = await wex.db.runReadWriteTx(
      { storeNames: ["peerPullCredit", "transactionsMeta"] },
      async (tx) => {
        const pullCreditRec = await tx.peerPullCredit.get(pursePub);
        if (!pullCreditRec) {
          logger.warn(`peer pull credit ${pursePub} not found`);
          return;
        }
        let newStatus: PeerPullPaymentCreditStatus | undefined = undefined;
        switch (pullCreditRec.status) {
          case PeerPullPaymentCreditStatus.PendingCreatePurse:
          case PeerPullPaymentCreditStatus.PendingMergeKycRequired:
          case PeerPullPaymentCreditStatus.PendingWithdrawing:
          case PeerPullPaymentCreditStatus.PendingReady:
          case PeerPullPaymentCreditStatus.Done:
          case PeerPullPaymentCreditStatus.SuspendedCreatePurse:
          case PeerPullPaymentCreditStatus.SuspendedMergeKycRequired:
          case PeerPullPaymentCreditStatus.SuspendedReady:
          case PeerPullPaymentCreditStatus.SuspendedWithdrawing:
          case PeerPullPaymentCreditStatus.Aborted:
          case PeerPullPaymentCreditStatus.Failed:
          case PeerPullPaymentCreditStatus.Expired:
          case PeerPullPaymentCreditStatus.PendingBalanceKycRequired:
          case PeerPullPaymentCreditStatus.SuspendedBalanceKycRequired:
          case PeerPullPaymentCreditStatus.PendingBalanceKycInit:
          case PeerPullPaymentCreditStatus.SuspendedBalanceKycInit:
            break;
          case PeerPullPaymentCreditStatus.AbortingDeletePurse:
          case PeerPullPaymentCreditStatus.SuspendedAbortingDeletePurse:
            newStatus = PeerPullPaymentCreditStatus.Failed;
            break;
          default:
            assertUnreachable(pullCreditRec.status);
        }
        if (newStatus != null) {
          const oldTxState =
            computePeerPullCreditTransactionState(pullCreditRec);
          pullCreditRec.status = newStatus;
          const newTxState =
            computePeerPullCreditTransactionState(pullCreditRec);
          await tx.peerPullCredit.put(pullCreditRec);
          await this.updateTransactionMeta(tx);
          return {
            oldTxState,
            newTxState,
          };
        }
        return undefined;
      },
    );
    notifyTransition(wex, transactionId, transitionInfo);
    wex.taskScheduler.stopShepherdTask(retryTag);
  }

  async resumeTransaction(): Promise<void> {
    const { wex, pursePub, taskId: retryTag, transactionId } = this;
    const transitionInfo = await wex.db.runReadWriteTx(
      { storeNames: ["peerPullCredit", "transactionsMeta"] },
      async (tx) => {
        const pullCreditRec = await tx.peerPullCredit.get(pursePub);
        if (!pullCreditRec) {
          logger.warn(`peer pull credit ${pursePub} not found`);
          return;
        }
        let newStatus: PeerPullPaymentCreditStatus | undefined = undefined;
        switch (pullCreditRec.status) {
          case PeerPullPaymentCreditStatus.PendingCreatePurse:
          case PeerPullPaymentCreditStatus.PendingMergeKycRequired:
          case PeerPullPaymentCreditStatus.PendingWithdrawing:
          case PeerPullPaymentCreditStatus.PendingReady:
          case PeerPullPaymentCreditStatus.PendingBalanceKycRequired:
          case PeerPullPaymentCreditStatus.PendingBalanceKycInit:
          case PeerPullPaymentCreditStatus.AbortingDeletePurse:
          case PeerPullPaymentCreditStatus.Done:
          case PeerPullPaymentCreditStatus.Failed:
          case PeerPullPaymentCreditStatus.Expired:
          case PeerPullPaymentCreditStatus.Aborted:
            break;
          case PeerPullPaymentCreditStatus.SuspendedBalanceKycInit:
            newStatus = PeerPullPaymentCreditStatus.PendingBalanceKycInit;
            break;
          case PeerPullPaymentCreditStatus.SuspendedCreatePurse:
            newStatus = PeerPullPaymentCreditStatus.PendingCreatePurse;
            break;
          case PeerPullPaymentCreditStatus.SuspendedMergeKycRequired:
            newStatus = PeerPullPaymentCreditStatus.PendingMergeKycRequired;
            break;
          case PeerPullPaymentCreditStatus.SuspendedReady:
            newStatus = PeerPullPaymentCreditStatus.PendingReady;
            break;
          case PeerPullPaymentCreditStatus.SuspendedWithdrawing:
            newStatus = PeerPullPaymentCreditStatus.PendingWithdrawing;
            break;
          case PeerPullPaymentCreditStatus.SuspendedAbortingDeletePurse:
            newStatus = PeerPullPaymentCreditStatus.AbortingDeletePurse;
            break;
          case PeerPullPaymentCreditStatus.SuspendedBalanceKycRequired:
            newStatus = PeerPullPaymentCreditStatus.PendingBalanceKycRequired;
            break;
          default:
            assertUnreachable(pullCreditRec.status);
        }
        if (newStatus != null) {
          const oldTxState =
            computePeerPullCreditTransactionState(pullCreditRec);
          pullCreditRec.status = newStatus;
          const newTxState =
            computePeerPullCreditTransactionState(pullCreditRec);
          await tx.peerPullCredit.put(pullCreditRec);
          await this.updateTransactionMeta(tx);
          return {
            oldTxState,
            newTxState,
          };
        }
        return undefined;
      },
    );
    notifyTransition(wex, transactionId, transitionInfo);
    wex.taskScheduler.startShepherdTask(retryTag);
  }

  async abortTransaction(): Promise<void> {
    const { wex, pursePub, taskId: retryTag, transactionId } = this;
    const transitionInfo = await wex.db.runReadWriteTx(
      { storeNames: ["peerPullCredit", "transactionsMeta"] },
      async (tx) => {
        const pullCreditRec = await tx.peerPullCredit.get(pursePub);
        if (!pullCreditRec) {
          logger.warn(`peer pull credit ${pursePub} not found`);
          return;
        }
        let newStatus: PeerPullPaymentCreditStatus | undefined = undefined;
        switch (pullCreditRec.status) {
          case PeerPullPaymentCreditStatus.PendingBalanceKycRequired:
          case PeerPullPaymentCreditStatus.SuspendedBalanceKycRequired:
          case PeerPullPaymentCreditStatus.PendingBalanceKycInit:
          case PeerPullPaymentCreditStatus.SuspendedBalanceKycInit:
          case PeerPullPaymentCreditStatus.PendingCreatePurse:
          case PeerPullPaymentCreditStatus.PendingMergeKycRequired:
            newStatus = PeerPullPaymentCreditStatus.AbortingDeletePurse;
            break;
          case PeerPullPaymentCreditStatus.PendingWithdrawing:
            throw Error("can't abort anymore");
          case PeerPullPaymentCreditStatus.PendingReady:
            newStatus = PeerPullPaymentCreditStatus.AbortingDeletePurse;
            break;
          case PeerPullPaymentCreditStatus.Done:
          case PeerPullPaymentCreditStatus.SuspendedCreatePurse:
          case PeerPullPaymentCreditStatus.SuspendedMergeKycRequired:
          case PeerPullPaymentCreditStatus.SuspendedReady:
          case PeerPullPaymentCreditStatus.SuspendedWithdrawing:
          case PeerPullPaymentCreditStatus.Aborted:
          case PeerPullPaymentCreditStatus.AbortingDeletePurse:
          case PeerPullPaymentCreditStatus.Failed:
          case PeerPullPaymentCreditStatus.Expired:
          case PeerPullPaymentCreditStatus.SuspendedAbortingDeletePurse:
            break;
          default:
            assertUnreachable(pullCreditRec.status);
        }
        if (newStatus != null) {
          const oldTxState =
            computePeerPullCreditTransactionState(pullCreditRec);
          pullCreditRec.status = newStatus;
          const newTxState =
            computePeerPullCreditTransactionState(pullCreditRec);
          await tx.peerPullCredit.put(pullCreditRec);
          await this.updateTransactionMeta(tx);
          return {
            oldTxState,
            newTxState,
          };
        }
        return undefined;
      },
    );
    wex.taskScheduler.stopShepherdTask(retryTag);
    notifyTransition(wex, transactionId, transitionInfo);
    wex.taskScheduler.startShepherdTask(retryTag);
  }
}

async function queryPurseForPeerPullCredit(
  wex: WalletExecutionContext,
  pullIni: PeerPullCreditRecord,
): Promise<TaskRunResult> {
  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 wex.ws.runLongpollQueueing(
    wex,
    purseDepositUrl.hostname,
    async () => {
      return await wex.http.fetch(purseDepositUrl.href, {
        timeout: { d_ms: 60000 },
        cancellationToken: wex.cancellationToken,
      });
    },
  );
  const ctx = new PeerPullCreditTransactionContext(wex, pullIni.pursePub);

  logger.info(`purse status code: HTTP ${resp.status}`);

  switch (resp.status) {
    case HttpStatusCode.Gone: {
      // Exchange says that purse doesn't exist anymore => expired!
      const transitionInfo = await wex.db.runReadWriteTx(
        { storeNames: ["peerPullCredit", "transactionsMeta"] },
        async (tx) => {
          const finPi = await tx.peerPullCredit.get(pullIni.pursePub);
          if (!finPi) {
            logger.warn("peerPullCredit not found anymore");
            return;
          }
          const oldTxState = computePeerPullCreditTransactionState(finPi);
          if (finPi.status === PeerPullPaymentCreditStatus.PendingReady) {
            finPi.status = PeerPullPaymentCreditStatus.Expired;
          }
          await tx.peerPullCredit.put(finPi);
          await ctx.updateTransactionMeta(tx);
          const newTxState = computePeerPullCreditTransactionState(finPi);
          return { oldTxState, newTxState };
        },
      );
      notifyTransition(wex, ctx.transactionId, transitionInfo);
      return TaskRunResult.backoff();
    }
    case HttpStatusCode.NotFound:
      // FIXME: Maybe check error code? 404 could also mean something else.
      return TaskRunResult.longpollReturnedPending();
  }

  const result = await readSuccessResponseJsonOrThrow(
    resp,
    codecForExchangePurseStatus(),
  );

  logger.trace(`purse status: ${j2s(result)}`);

  const depositTimestamp = result.deposit_timestamp;

  if (!depositTimestamp || TalerProtocolTimestamp.isNever(depositTimestamp)) {
    logger.info("purse not ready yet (no deposit)");
    return TaskRunResult.backoff();
  }

  const reserve = await wex.db.runReadOnlyTx(
    { storeNames: ["reserves"] },
    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(wex, {
    amount: Amounts.parseOrThrow(pullIni.amount),
    wgInfo: {
      withdrawalType: WithdrawalRecordType.PeerPullCredit,
      contractPriv: pullIni.contractPriv,
    },
    forcedWithdrawalGroupId: pullIni.withdrawalGroupId,
    exchangeBaseUrl: pullIni.exchangeBaseUrl,
    reserveStatus: WithdrawalGroupStatus.PendingQueryingStatus,
    reserveKeyPair: {
      priv: reserve.reservePriv,
      pub: reserve.reservePub,
    },
  });
  const transitionInfo = await wex.db.runReadWriteTx(
    { storeNames: ["peerPullCredit", "transactionsMeta"] },
    async (tx) => {
      const finPi = await tx.peerPullCredit.get(pullIni.pursePub);
      if (!finPi) {
        logger.warn("peerPullCredit not found anymore");
        return;
      }
      const oldTxState = computePeerPullCreditTransactionState(finPi);
      if (finPi.status === PeerPullPaymentCreditStatus.PendingReady) {
        finPi.status = PeerPullPaymentCreditStatus.PendingWithdrawing;
      }
      await tx.peerPullCredit.put(finPi);
      await ctx.updateTransactionMeta(tx);
      const newTxState = computePeerPullCreditTransactionState(finPi);
      return { oldTxState, newTxState };
    },
  );
  notifyTransition(wex, ctx.transactionId, transitionInfo);
  return TaskRunResult.backoff();
}

async function longpollKycStatus(
  wex: WalletExecutionContext,
  pursePub: string,
  exchangeUrl: string,
  kycPaytoHash: string,
): Promise<TaskRunResult> {
  // FIXME: What if this changes? Should be part of the p2p record
  const mergeReserveInfo = await getMergeReserveInfo(wex, {
    exchangeBaseUrl: exchangeUrl,
  });

  const sigResp = await wex.cryptoApi.signWalletKycAuth({
    accountPriv: mergeReserveInfo.reservePriv,
    accountPub: mergeReserveInfo.reservePub,
  });

  const ctx = new PeerPullCreditTransactionContext(wex, pursePub);
  const url = new URL(`kyc-check/${kycPaytoHash}`, exchangeUrl);
  const kycStatusRes = await wex.ws.runLongpollQueueing(
    wex,
    url.hostname,
    async (timeoutMs) => {
      url.searchParams.set("timeout_ms", `${timeoutMs}`);
      logger.info(`kyc url ${url.href}`);
      return await wex.http.fetch(url.href, {
        method: "GET",
        headers: {
          ["Account-Owner-Signature"]: sigResp.sig,
        },
        cancellationToken: wex.cancellationToken,
      });
    },
  );
  if (
    kycStatusRes.status === HttpStatusCode.Ok ||
    // FIXME: NoContent is not expected https://docs.taler.net/core/api-exchange.html#post--purses-$PURSE_PUB-merge
    // remove after the exchange is fixed or clarified
    kycStatusRes.status === HttpStatusCode.NoContent
  ) {
    const transitionInfo = await wex.db.runReadWriteTx(
      { storeNames: ["peerPullCredit", "transactionsMeta"] },
      async (tx) => {
        const peerIni = await tx.peerPullCredit.get(pursePub);
        if (!peerIni) {
          return;
        }
        if (
          peerIni.status !== PeerPullPaymentCreditStatus.PendingMergeKycRequired
        ) {
          return;
        }
        const oldTxState = computePeerPullCreditTransactionState(peerIni);
        peerIni.status = PeerPullPaymentCreditStatus.PendingCreatePurse;
        const newTxState = computePeerPullCreditTransactionState(peerIni);
        await tx.peerPullCredit.put(peerIni);
        await ctx.updateTransactionMeta(tx);
        return { oldTxState, newTxState };
      },
    );
    notifyTransition(wex, ctx.transactionId, transitionInfo);
    return TaskRunResult.progress();
  } else if (kycStatusRes.status === HttpStatusCode.Accepted) {
    return TaskRunResult.longpollReturnedPending();
  } else {
    throw Error(`unexpected response from kyc-check (${kycStatusRes.status})`);
  }
}

async function processPeerPullCreditAbortingDeletePurse(
  wex: WalletExecutionContext,
  peerPullIni: PeerPullCreditRecord,
): Promise<TaskRunResult> {
  const { pursePub, pursePriv } = peerPullIni;
  const ctx = new PeerPullCreditTransactionContext(wex, peerPullIni.pursePub);

  const sigResp = await wex.cryptoApi.signDeletePurse({
    pursePriv,
  });
  const purseUrl = new URL(`purses/${pursePub}`, peerPullIni.exchangeBaseUrl);
  const resp = await wex.http.fetch(purseUrl.href, {
    method: "DELETE",
    headers: {
      "taler-purse-signature": sigResp.sig,
    },
    cancellationToken: wex.cancellationToken,
  });
  logger.info(`deleted purse with response status ${resp.status}`);

  const transitionInfo = await wex.db.runReadWriteTx(
    {
      storeNames: [
        "peerPullCredit",
        "refreshGroups",
        "denominations",
        "coinAvailability",
        "coins",
        "transactionsMeta",
      ],
    },
    async (tx) => {
      const ppiRec = await tx.peerPullCredit.get(pursePub);
      if (!ppiRec) {
        return undefined;
      }
      if (ppiRec.status !== PeerPullPaymentCreditStatus.AbortingDeletePurse) {
        return undefined;
      }
      const oldTxState = computePeerPullCreditTransactionState(ppiRec);
      ppiRec.status = PeerPullPaymentCreditStatus.Aborted;
      await tx.peerPullCredit.put(ppiRec);
      await ctx.updateTransactionMeta(tx);
      const newTxState = computePeerPullCreditTransactionState(ppiRec);
      return {
        oldTxState,
        newTxState,
      };
    },
  );
  notifyTransition(wex, ctx.transactionId, transitionInfo);

  return TaskRunResult.backoff();
}

async function handlePeerPullCreditWithdrawing(
  wex: WalletExecutionContext,
  pullIni: PeerPullCreditRecord,
): Promise<TaskRunResult> {
  if (!pullIni.withdrawalGroupId) {
    throw Error("invalid db state (withdrawing, but no withdrawal group ID");
  }
  await waitWithdrawalFinal(wex, pullIni.withdrawalGroupId);
  const ctx = new PeerPullCreditTransactionContext(wex, pullIni.pursePub);
  const wgId = pullIni.withdrawalGroupId;
  let finished: boolean = false;
  const transitionInfo = await wex.db.runReadWriteTx(
    { storeNames: ["peerPullCredit", "withdrawalGroups", "transactionsMeta"] },
    async (tx) => {
      const ppi = await tx.peerPullCredit.get(pullIni.pursePub);
      if (!ppi) {
        finished = true;
        return;
      }
      if (ppi.status !== PeerPullPaymentCreditStatus.PendingWithdrawing) {
        finished = true;
        return;
      }
      const oldTxState = computePeerPullCreditTransactionState(ppi);
      const wg = await tx.withdrawalGroups.get(wgId);
      if (!wg) {
        // FIXME: Fail the operation instead?
        return undefined;
      }
      switch (wg.status) {
        case WithdrawalGroupStatus.Done:
          finished = true;
          ppi.status = PeerPullPaymentCreditStatus.Done;
          break;
        // FIXME: Also handle other final states!
      }
      await tx.peerPullCredit.put(ppi);
      await ctx.updateTransactionMeta(tx);
      const newTxState = computePeerPullCreditTransactionState(ppi);
      return {
        oldTxState,
        newTxState,
      };
    },
  );
  notifyTransition(wex, ctx.transactionId, transitionInfo);
  if (finished) {
    return TaskRunResult.finished();
  } else {
    // FIXME: Return indicator that we depend on the other operation!
    return TaskRunResult.backoff();
  }
}

async function handlePeerPullCreditCreatePurse(
  wex: WalletExecutionContext,
  pullIni: PeerPullCreditRecord,
): Promise<TaskRunResult> {
  const ctx = new PeerPullCreditTransactionContext(wex, pullIni.pursePub);

  const kycCheckRes = await checkIncomingAmountLegalUnderKycBalanceThreshold(
    wex,
    pullIni.exchangeBaseUrl,
    pullIni.estimatedAmountEffective,
  );

  if (kycCheckRes.result === "violation") {
    // Do this before we transition so that the exchange is already in the right state.
    await handleStartExchangeWalletKyc(wex, {
      amount: kycCheckRes.nextThreshold,
      exchangeBaseUrl: pullIni.exchangeBaseUrl,
    });
    await ctx.transition({}, async (rec) => {
      if (!rec) {
        return TransitionResult.stay();
      }
      if (rec.status !== PeerPullPaymentCreditStatus.PendingCreatePurse) {
        return TransitionResult.stay();
      }
      rec.status = PeerPullPaymentCreditStatus.PendingBalanceKycInit;
      return TransitionResult.transition(rec);
    });
    return TaskRunResult.progress();
  }

  const purseFee = Amounts.stringify(Amounts.zeroOfAmount(pullIni.amount));
  const pursePub = pullIni.pursePub;
  const mergeReserve = await wex.db.runReadOnlyTx(
    { storeNames: ["reserves"] },
    async (tx) => {
      return tx.reserves.get(pullIni.mergeReserveRowId);
    },
  );

  if (!mergeReserve) {
    throw Error("merge reserve for peer pull payment not found in database");
  }

  const contractTermsRecord = await wex.db.runReadOnlyTx(
    { storeNames: ["contractTerms"] },
    async (tx) => {
      return tx.contractTerms.get(pullIni.contractTermsHash);
    },
  );

  if (!contractTermsRecord) {
    throw Error("contract terms for peer pull payment not found in database");
  }

  const contractTerms: PeerContractTerms = contractTermsRecord.contractTermsRaw;

  const reservePayto = talerPaytoFromExchangeReserve(
    pullIni.exchangeBaseUrl,
    mergeReserve.reservePub,
  );

  const econtractResp = await wex.cryptoApi.encryptContractForDeposit({
    contractPriv: pullIni.contractPriv,
    contractPub: pullIni.contractPub,
    contractTerms: contractTermsRecord.contractTermsRaw,
    pursePriv: pullIni.pursePriv,
    pursePub: pullIni.pursePub,
    nonce: pullIni.contractEncNonce,
  });

  const mergeTimestamp = timestampPreciseFromDb(pullIni.mergeTimestamp);

  const purseExpiration = contractTerms.purse_expiration;
  const sigRes = await wex.cryptoApi.signReservePurseCreate({
    contractTermsHash: pullIni.contractTermsHash,
    flags: WalletAccountMergeFlags.CreateWithPurseFee,
    mergePriv: pullIni.mergePriv,
    mergeTimestamp: TalerPreciseTimestamp.round(mergeTimestamp),
    purseAmount: pullIni.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(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.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 wex.http.fetch(reservePurseMergeUrl.href, {
    method: "POST",
    body: reservePurseReqBody,
    cancellationToken: wex.cancellationToken,
  });

  if (httpResp.status === HttpStatusCode.UnavailableForLegalReasons) {
    const respJson = await httpResp.json();
    const kycPending = codecForLegitimizationNeededResponse().decode(respJson);
    logger.info(`kyc uuid response: ${j2s(kycPending)}`);
    return processPeerPullCreditKycRequired(wex, pullIni, kycPending.h_payto);
  }

  const resp = await readSuccessResponseJsonOrThrow(httpResp, codecForAny());

  logger.info(`reserve merge response: ${j2s(resp)}`);

  const transitionInfo = await wex.db.runReadWriteTx(
    { storeNames: ["peerPullCredit", "transactionsMeta"] },
    async (tx) => {
      const pi2 = await tx.peerPullCredit.get(pursePub);
      if (!pi2) {
        return;
      }
      const oldTxState = computePeerPullCreditTransactionState(pi2);
      pi2.status = PeerPullPaymentCreditStatus.PendingReady;
      await tx.peerPullCredit.put(pi2);
      await ctx.updateTransactionMeta(tx);
      const newTxState = computePeerPullCreditTransactionState(pi2);
      return { oldTxState, newTxState };
    },
  );
  notifyTransition(wex, ctx.transactionId, transitionInfo);
  return TaskRunResult.backoff();
}

export async function processPeerPullCredit(
  wex: WalletExecutionContext,
  pursePub: string,
): Promise<TaskRunResult> {
  if (!wex.ws.networkAvailable) {
    return TaskRunResult.networkRequired();
  }

  const pullIni = await wex.db.runReadOnlyTx(
    { storeNames: ["peerPullCredit"] },
    async (tx) => {
      return tx.peerPullCredit.get(pursePub);
    },
  );
  if (!pullIni) {
    throw Error("peer pull payment initiation not found in database");
  }

  const retryTag = constructTaskIdentifier({
    tag: PendingTaskType.PeerPullCredit,
    pursePub,
  });

  logger.trace(`processing ${retryTag}, status=${pullIni.status}`);

  const ctx = new PeerPullCreditTransactionContext(wex, pullIni.pursePub);

  switch (pullIni.status) {
    case PeerPullPaymentCreditStatus.Done: {
      return TaskRunResult.finished();
    }
    case PeerPullPaymentCreditStatus.PendingReady:
      return queryPurseForPeerPullCredit(wex, pullIni);
    case PeerPullPaymentCreditStatus.PendingMergeKycRequired: {
      if (!pullIni.kycPaytoHash) {
        throw Error("invalid state, kycPaytoHash required");
      }
      return await longpollKycStatus(
        wex,
        pursePub,
        pullIni.exchangeBaseUrl,
        pullIni.kycPaytoHash,
      );
    }
    case PeerPullPaymentCreditStatus.PendingCreatePurse:
      return handlePeerPullCreditCreatePurse(wex, pullIni);
    case PeerPullPaymentCreditStatus.AbortingDeletePurse:
      return await processPeerPullCreditAbortingDeletePurse(wex, pullIni);
    case PeerPullPaymentCreditStatus.PendingWithdrawing:
      return handlePeerPullCreditWithdrawing(wex, pullIni);
    case PeerPullPaymentCreditStatus.PendingBalanceKycRequired:
    case PeerPullPaymentCreditStatus.PendingBalanceKycInit:
      return processPeerPullCreditBalanceKyc(ctx, pullIni);
    case PeerPullPaymentCreditStatus.Aborted:
    case PeerPullPaymentCreditStatus.Failed:
    case PeerPullPaymentCreditStatus.Expired:
    case PeerPullPaymentCreditStatus.SuspendedBalanceKycRequired:
    case PeerPullPaymentCreditStatus.SuspendedAbortingDeletePurse:
    case PeerPullPaymentCreditStatus.SuspendedCreatePurse:
    case PeerPullPaymentCreditStatus.SuspendedMergeKycRequired:
    case PeerPullPaymentCreditStatus.SuspendedReady:
    case PeerPullPaymentCreditStatus.SuspendedWithdrawing:
    case PeerPullPaymentCreditStatus.SuspendedBalanceKycInit:
      break;
    default:
      assertUnreachable(pullIni.status);
  }

  return TaskRunResult.finished();
}

async function processPeerPullCreditBalanceKyc(
  ctx: PeerPullCreditTransactionContext,
  peerInc: PeerPullCreditRecord,
): Promise<TaskRunResult> {
  const exchangeBaseUrl = peerInc.exchangeBaseUrl;
  const amount = peerInc.estimatedAmountEffective;

  const ret = await genericWaitForStateVal(ctx.wex, {
    async checkState(): Promise<BalanceThresholdCheckResult | undefined> {
      const checkRes = await checkIncomingAmountLegalUnderKycBalanceThreshold(
        ctx.wex,
        exchangeBaseUrl,
        amount,
      );
      logger.info(
        `balance check result for ${exchangeBaseUrl} +${amount}: ${j2s(
          checkRes,
        )}`,
      );
      if (checkRes.result === "ok") {
        return checkRes;
      }
      if (
        peerInc.status === PeerPullPaymentCreditStatus.PendingBalanceKycInit &&
        checkRes.walletKycStatus === ExchangeWalletKycStatus.Legi
      ) {
        return checkRes;
      }
      await handleStartExchangeWalletKyc(ctx.wex, {
        amount: checkRes.nextThreshold,
        exchangeBaseUrl,
      });
      return undefined;
    },
    filterNotification(notif) {
      return (
        (notif.type === NotificationType.ExchangeStateTransition &&
          notif.exchangeBaseUrl === exchangeBaseUrl) ||
        notif.type === NotificationType.BalanceChange
      );
    },
  });

  if (ret.result === "ok") {
    await ctx.transition({}, async (rec) => {
      if (!rec) {
        return TransitionResult.stay();
      }
      if (
        rec.status !== PeerPullPaymentCreditStatus.PendingBalanceKycRequired
      ) {
        return TransitionResult.stay();
      }
      rec.status = PeerPullPaymentCreditStatus.PendingCreatePurse;
      return TransitionResult.transition(rec);
    });
    return TaskRunResult.progress();
  } else if (
    peerInc.status === PeerPullPaymentCreditStatus.PendingBalanceKycInit &&
    ret.walletKycStatus === ExchangeWalletKycStatus.Legi
  ) {
    await ctx.transition({}, async (rec) => {
      if (!rec) {
        return TransitionResult.stay();
      }
      if (rec.status !== PeerPullPaymentCreditStatus.PendingBalanceKycInit) {
        return TransitionResult.stay();
      }
      rec.status = PeerPullPaymentCreditStatus.PendingBalanceKycRequired;
      rec.kycAccessToken = ret.walletKycAccessToken;
      return TransitionResult.transition(rec);
    });
    return TaskRunResult.progress();
  } else {
    throw Error("not reached");
  }
}

async function processPeerPullCreditKycRequired(
  wex: WalletExecutionContext,
  peerIni: PeerPullCreditRecord,
  kycPayoHash: string,
): Promise<TaskRunResult> {
  const ctx = new PeerPullCreditTransactionContext(wex, peerIni.pursePub);
  const { pursePub } = peerIni;

  // FIXME: What if this changes? Should be part of the p2p record
  const mergeReserveInfo = await getMergeReserveInfo(wex, {
    exchangeBaseUrl: peerIni.exchangeBaseUrl,
  });

  const sigResp = await wex.cryptoApi.signWalletKycAuth({
    accountPriv: mergeReserveInfo.reservePriv,
    accountPub: mergeReserveInfo.reservePub,
  });

  const url = new URL(`kyc-check/${kycPayoHash}`, peerIni.exchangeBaseUrl);

  logger.info(`kyc url ${url.href}`);
  const kycStatusRes = await wex.http.fetch(url.href, {
    method: "GET",
    headers: {
      ["Account-Owner-Signature"]: sigResp.sig,
    },
    cancellationToken: wex.cancellationToken,
  });

  if (
    kycStatusRes.status === HttpStatusCode.Ok ||
    kycStatusRes.status === HttpStatusCode.NoContent
  ) {
    logger.warn("kyc requested, but already fulfilled");
    return TaskRunResult.backoff();
  } else if (kycStatusRes.status === HttpStatusCode.Accepted) {
    const kycStatus = await readResponseJsonOrThrow(
      kycStatusRes,
      codecForAccountKycStatus(),
    );
    logger.info(`kyc status: ${j2s(kycStatus)}`);
    const { transitionInfo, result } = await wex.db.runReadWriteTx(
      { storeNames: ["peerPullCredit", "transactionsMeta"] },
      async (tx) => {
        const peerInc = await tx.peerPullCredit.get(pursePub);
        if (!peerInc) {
          return {
            transitionInfo: undefined,
            result: TaskRunResult.finished(),
          };
        }
        const oldTxState = computePeerPullCreditTransactionState(peerInc);
        peerInc.kycPaytoHash = kycPayoHash;
        logger.info(
          `setting peer-pull-credit kyc payto hash to ${kycPayoHash}`,
        );
        peerInc.kycAccessToken = kycStatus.access_token;
        peerInc.status = PeerPullPaymentCreditStatus.PendingMergeKycRequired;
        const newTxState = computePeerPullCreditTransactionState(peerInc);
        await tx.peerPullCredit.put(peerInc);
        await ctx.updateTransactionMeta(tx);
        return {
          transitionInfo: { oldTxState, newTxState },
          result: TaskRunResult.progress(),
        };
      },
    );
    notifyTransition(wex, ctx.transactionId, transitionInfo);
    return result;
  } else {
    throw Error(`unexpected response from kyc-check (${kycStatusRes.status})`);
  }
}

/**
 * Check fees and available exchanges for a peer push payment initiation.
 */
export async function checkPeerPullCredit(
  wex: WalletExecutionContext,
  req: CheckPeerPullCreditRequest,
): Promise<CheckPeerPullCreditResponse> {
  return runWithClientCancellation(
    wex,
    "checkPeerPullCredit",
    req.clientCancellationId,
    async () => internalCheckPeerPullCredit(wex, req),
  );
}

/**
 * Check fees and available exchanges for a peer push payment initiation.
 */
export async function internalCheckPeerPullCredit(
  wex: WalletExecutionContext,
  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(wex, 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(
    wex,
    exchangeUrl,
    Amounts.parseOrThrow(req.amount),
    undefined,
  );
  if (wi.selectedDenoms.selectedDenoms.length === 0) {
    throw Error(
      `unable to check pull payment from ${exchangeUrl}, can't select denominations for instructed amount (${req.amount}`,
    );
  }

  logger.trace(`got withdrawal info`);

  let numCoins = 0;
  for (let i = 0; i < wi.selectedDenoms.selectedDenoms.length; i++) {
    numCoins += wi.selectedDenoms.selectedDenoms[i].count;
  }

  return {
    exchangeBaseUrl: exchangeUrl,
    amountEffective: wi.withdrawalAmountEffective,
    amountRaw: req.amount,
    numCoins,
  };
}

/**
 * Find a preferred exchange based on when we withdrew last from this exchange.
 */
async function getPreferredExchangeForCurrency(
  wex: WalletExecutionContext,
  currency: string,
): Promise<string | undefined> {
  // Find an exchange with the matching currency.
  // Prefer exchanges with the most recent withdrawal.
  const url = await wex.db.runReadOnlyTx(
    { storeNames: ["exchanges"] },
    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;
        }
        const exchangeLastWithdrawal = timestampOptionalPreciseFromDb(
          e.lastWithdrawal,
        );
        const candidateLastWithdrawal = timestampOptionalPreciseFromDb(
          candidate.lastWithdrawal,
        );
        if (exchangeLastWithdrawal && candidateLastWithdrawal) {
          if (
            AbsoluteTime.cmp(
              AbsoluteTime.fromPreciseTimestamp(exchangeLastWithdrawal),
              AbsoluteTime.fromPreciseTimestamp(candidateLastWithdrawal),
            ) > 0
          ) {
            candidate = e;
          }
        }
      }
      if (candidate) {
        return candidate.baseUrl;
      }
      return undefined;
    },
  );
  return url;
}

/**
 * Initiate a peer pull payment.
 */
export async function initiatePeerPullPayment(
  wex: WalletExecutionContext,
  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(wex, currency);
  }

  if (!maybeExchangeBaseUrl) {
    throw Error("no exchange found for initiating a peer pull payment");
  }

  const exchangeBaseUrl = maybeExchangeBaseUrl;

  const exchange = await fetchFreshExchange(wex, exchangeBaseUrl);
  requireExchangeTosAcceptedOrThrow(exchange);

  const mergeReserveInfo = await getMergeReserveInfo(wex, {
    exchangeBaseUrl: exchangeBaseUrl,
  });

  const pursePair = await wex.cryptoApi.createEddsaKeypair({});
  const mergePair = await wex.cryptoApi.createEddsaKeypair({});

  const contractTerms = req.partialContractTerms;

  const hContractTerms = ContractTermsUtil.hashContractTerms(contractTerms);

  const contractKeyPair = await wex.cryptoApi.createEddsaKeypair({});

  const withdrawalGroupId = encodeCrock(getRandomBytes(32));

  const mergeReserveRowId = mergeReserveInfo.rowId;
  checkDbInvariant(
    !!mergeReserveRowId,
    `merge reserve for ${exchangeBaseUrl} without rowid`,
  );

  const contractEncNonce = encodeCrock(getRandomBytes(24));

  const wi = await getExchangeWithdrawalInfo(
    wex,
    exchangeBaseUrl,
    Amounts.parseOrThrow(req.partialContractTerms.amount),
    undefined,
  );
  if (wi.selectedDenoms.selectedDenoms.length === 0) {
    throw Error(
      `unable to initiate pull payment from ${exchangeBaseUrl}, can't select denominations for instructed amount (${req.partialContractTerms.amount}`,
    );
  }

  const mergeTimestamp = TalerPreciseTimestamp.now();

  const ctx = new PeerPullCreditTransactionContext(wex, pursePair.pub);

  const transitionInfo = await wex.db.runReadWriteTx(
    { storeNames: ["peerPullCredit", "contractTerms", "transactionsMeta"] },
    async (tx) => {
      const ppi: PeerPullCreditRecord = {
        amount: req.partialContractTerms.amount,
        contractTermsHash: hContractTerms,
        exchangeBaseUrl: exchangeBaseUrl,
        pursePriv: pursePair.priv,
        pursePub: pursePair.pub,
        mergePriv: mergePair.priv,
        mergePub: mergePair.pub,
        status: PeerPullPaymentCreditStatus.PendingCreatePurse,
        mergeTimestamp: timestampPreciseToDb(mergeTimestamp),
        contractEncNonce,
        mergeReserveRowId: mergeReserveRowId,
        contractPriv: contractKeyPair.priv,
        contractPub: contractKeyPair.pub,
        withdrawalGroupId,
        estimatedAmountEffective: wi.withdrawalAmountEffective,
      };
      await tx.peerPullCredit.put(ppi);
      await ctx.updateTransactionMeta(tx);
      const oldTxState: TransactionState = {
        major: TransactionMajorState.None,
      };
      const newTxState = computePeerPullCreditTransactionState(ppi);
      await tx.contractTerms.put({
        contractTermsRaw: contractTerms,
        h: hContractTerms,
      });
      return { oldTxState, newTxState };
    },
  );

  notifyTransition(wex, ctx.transactionId, transitionInfo);
  wex.taskScheduler.startShepherdTask(ctx.taskId);

  // The pending-incoming balance has changed.
  wex.ws.notify({
    type: NotificationType.BalanceChange,
    hintTransactionId: ctx.transactionId,
  });

  return {
    talerUri: stringifyTalerUri({
      type: TalerUriAction.PayPull,
      exchangeBaseUrl: exchangeBaseUrl,
      contractPriv: contractKeyPair.priv,
    }),
    transactionId: ctx.transactionId,
  };
}

export function computePeerPullCreditTransactionState(
  pullCreditRecord: PeerPullCreditRecord,
): TransactionState {
  switch (pullCreditRecord.status) {
    case PeerPullPaymentCreditStatus.PendingCreatePurse:
      return {
        major: TransactionMajorState.Pending,
        minor: TransactionMinorState.CreatePurse,
      };
    case PeerPullPaymentCreditStatus.PendingMergeKycRequired:
      return {
        major: TransactionMajorState.Pending,
        minor: TransactionMinorState.MergeKycRequired,
      };
    case PeerPullPaymentCreditStatus.PendingReady:
      return {
        major: TransactionMajorState.Pending,
        minor: TransactionMinorState.Ready,
      };
    case PeerPullPaymentCreditStatus.Done:
      return {
        major: TransactionMajorState.Done,
      };
    case PeerPullPaymentCreditStatus.PendingWithdrawing:
      return {
        major: TransactionMajorState.Pending,
        minor: TransactionMinorState.Withdraw,
      };
    case PeerPullPaymentCreditStatus.SuspendedCreatePurse:
      return {
        major: TransactionMajorState.Suspended,
        minor: TransactionMinorState.CreatePurse,
      };
    case PeerPullPaymentCreditStatus.SuspendedReady:
      return {
        major: TransactionMajorState.Suspended,
        minor: TransactionMinorState.Ready,
      };
    case PeerPullPaymentCreditStatus.SuspendedWithdrawing:
      return {
        major: TransactionMajorState.Pending,
        minor: TransactionMinorState.Withdraw,
      };
    case PeerPullPaymentCreditStatus.SuspendedMergeKycRequired:
      return {
        major: TransactionMajorState.Suspended,
        minor: TransactionMinorState.MergeKycRequired,
      };
    case PeerPullPaymentCreditStatus.Aborted:
      return {
        major: TransactionMajorState.Aborted,
      };
    case PeerPullPaymentCreditStatus.AbortingDeletePurse:
      return {
        major: TransactionMajorState.Aborting,
        minor: TransactionMinorState.DeletePurse,
      };
    case PeerPullPaymentCreditStatus.Failed:
      return {
        major: TransactionMajorState.Failed,
      };
    case PeerPullPaymentCreditStatus.Expired:
      return {
        major: TransactionMajorState.Expired,
      };
    case PeerPullPaymentCreditStatus.SuspendedAbortingDeletePurse:
      return {
        major: TransactionMajorState.Aborting,
        minor: TransactionMinorState.DeletePurse,
      };
    case PeerPullPaymentCreditStatus.PendingBalanceKycRequired:
      return {
        major: TransactionMajorState.Pending,
        minor: TransactionMinorState.BalanceKycRequired,
      };
    case PeerPullPaymentCreditStatus.SuspendedBalanceKycRequired:
      return {
        major: TransactionMajorState.Suspended,
        minor: TransactionMinorState.BalanceKycRequired,
      };
    case PeerPullPaymentCreditStatus.PendingBalanceKycInit:
      return {
        major: TransactionMajorState.Pending,
        minor: TransactionMinorState.BalanceKycInit,
      };
    case PeerPullPaymentCreditStatus.SuspendedBalanceKycInit:
      return {
        major: TransactionMajorState.Suspended,
        minor: TransactionMinorState.BalanceKycInit,
      };
  }
}

export function computePeerPullCreditTransactionActions(
  pullCreditRecord: PeerPullCreditRecord,
): TransactionAction[] {
  switch (pullCreditRecord.status) {
    case PeerPullPaymentCreditStatus.PendingCreatePurse:
      return [
        TransactionAction.Retry,
        TransactionAction.Abort,
        TransactionAction.Suspend,
      ];
    case PeerPullPaymentCreditStatus.PendingMergeKycRequired:
      return [
        TransactionAction.Retry,
        TransactionAction.Abort,
        TransactionAction.Suspend,
      ];
    case PeerPullPaymentCreditStatus.PendingReady:
      return [
        TransactionAction.Retry,
        TransactionAction.Abort,
        TransactionAction.Suspend,
      ];
    case PeerPullPaymentCreditStatus.Done:
      return [TransactionAction.Delete];
    case PeerPullPaymentCreditStatus.PendingWithdrawing:
      return [
        TransactionAction.Retry,
        TransactionAction.Abort,
        TransactionAction.Suspend,
      ];
    case PeerPullPaymentCreditStatus.SuspendedCreatePurse:
      return [TransactionAction.Resume, TransactionAction.Abort];
    case PeerPullPaymentCreditStatus.SuspendedReady:
      return [TransactionAction.Abort, TransactionAction.Resume];
    case PeerPullPaymentCreditStatus.SuspendedWithdrawing:
      return [TransactionAction.Resume, TransactionAction.Fail];
    case PeerPullPaymentCreditStatus.SuspendedMergeKycRequired:
      return [TransactionAction.Resume, TransactionAction.Fail];
    case PeerPullPaymentCreditStatus.Aborted:
      return [TransactionAction.Delete];
    case PeerPullPaymentCreditStatus.AbortingDeletePurse:
      return [
        TransactionAction.Retry,
        TransactionAction.Suspend,
        TransactionAction.Fail,
      ];
    case PeerPullPaymentCreditStatus.Failed:
      return [TransactionAction.Delete];
    case PeerPullPaymentCreditStatus.Expired:
      return [TransactionAction.Delete];
    case PeerPullPaymentCreditStatus.SuspendedAbortingDeletePurse:
      return [TransactionAction.Resume, TransactionAction.Fail];
    case PeerPullPaymentCreditStatus.PendingBalanceKycRequired:
      return [TransactionAction.Suspend, TransactionAction.Abort];
    case PeerPullPaymentCreditStatus.SuspendedBalanceKycRequired:
      return [TransactionAction.Resume, TransactionAction.Abort];
    case PeerPullPaymentCreditStatus.PendingBalanceKycInit:
      return [TransactionAction.Suspend, TransactionAction.Abort];
    case PeerPullPaymentCreditStatus.SuspendedBalanceKycInit:
      return [TransactionAction.Resume, TransactionAction.Abort];
  }
}