/*
 This file is part of GNU Taler
 (C) 2022 GNUnet e.V.

 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,
  ConfirmPeerPullDebitRequest,
  AcceptPeerPullPaymentResponse,
  ConfirmPeerPushCreditRequest,
  AcceptPeerPushPaymentResponse,
  AgeCommitmentProof,
  AmountJson,
  Amounts,
  AmountString,
  buildCodecForObject,
  PreparePeerPullDebitRequest,
  PreparePeerPullDebitResponse,
  PreparePeerPushCredit,
  PreparePeerPushCreditResponse,
  Codec,
  codecForAmountString,
  codecForAny,
  codecForExchangeGetContractResponse,
  codecForPeerContractTerms,
  CoinStatus,
  constructPayPullUri,
  constructPayPushUri,
  ContractTermsUtil,
  decodeCrock,
  eddsaGetPublic,
  encodeCrock,
  ExchangePurseDeposits,
  ExchangePurseMergeRequest,
  ExchangeReservePurseRequest,
  getRandomBytes,
  InitiatePeerPullCreditRequest,
  InitiatePeerPullCreditResponse,
  InitiatePeerPushPaymentRequest,
  InitiatePeerPushPaymentResponse,
  j2s,
  Logger,
  parsePayPullUri,
  parsePayPushUri,
  PayPeerInsufficientBalanceDetails,
  PeerContractTerms,
  CheckPeerPullCreditRequest,
  CheckPeerPullCreditResponse,
  CheckPeerPushDebitRequest,
  CheckPeerPushDebitResponse,
  RefreshReason,
  strcmp,
  TalerErrorCode,
  TalerProtocolTimestamp,
  TransactionType,
  UnblindedSignature,
  WalletAccountMergeFlags,
  codecOptional,
  codecForTimestamp,
  CancellationToken,
  NotificationType,
  HttpStatusCode,
  codecForWalletKycUuid,
  WalletKycUuid,
} from "@gnu-taler/taler-util";
import { SpendCoinDetails } from "../crypto/cryptoImplementation.js";
import {
  DenominationRecord,
  KycPendingInfo,
  KycUserType,
  OperationStatus,
  PeerPullPaymentIncomingStatus,
  PeerPullPaymentInitiationRecord,
  PeerPullPaymentInitiationStatus,
  PeerPushPaymentCoinSelection,
  PeerPushPaymentIncomingRecord,
  PeerPushPaymentIncomingStatus,
  PeerPushPaymentInitiationStatus,
  ReserveRecord,
  WithdrawalGroupStatus,
  WithdrawalRecordType,
} from "../db.js";
import { TalerError } from "@gnu-taler/taler-util";
import { InternalWalletState } from "../internal-wallet-state.js";
import {
  LongpollResult,
  makeTransactionId,
  resetOperationTimeout,
  runLongpollAsync,
  runOperationWithErrorReporting,
  spendCoins,
} from "../operations/common.js";
import {
  readSuccessResponseJsonOrErrorCode,
  readSuccessResponseJsonOrThrow,
  throwUnexpectedRequestError,
} from "@gnu-taler/taler-util/http";
import { checkDbInvariant } from "../util/invariants.js";
import {
  constructTaskIdentifier,
  OperationAttemptResult,
  OperationAttemptResultType,
  TaskIdentifiers,
} from "../util/retries.js";
import { getPeerPaymentBalanceDetailsInTx } from "./balance.js";
import { updateExchangeFromUrl } from "./exchanges.js";
import { getTotalRefreshCost } from "./refresh.js";
import {
  checkWithdrawalKycStatus,
  getExchangeWithdrawalInfo,
  internalCreateWithdrawalGroup,
  processWithdrawalGroup,
} from "./withdraw.js";
import { PendingTaskType } from "../pending-types.js";
import {
  constructTransactionIdentifier,
  stopLongpolling,
} from "./transactions.js";

const logger = new Logger("operations/peer-to-peer.ts");

interface SelectedPeerCoin {
  coinPub: string;
  coinPriv: string;
  contribution: AmountString;
  denomPubHash: string;
  denomSig: UnblindedSignature;
  ageCommitmentProof: AgeCommitmentProof | undefined;
}

interface PeerCoinSelectionDetails {
  exchangeBaseUrl: string;

  /**
   * Info of Coins that were selected.
   */
  coins: SelectedPeerCoin[];

  /**
   * How much of the deposit fees is the customer paying?
   */
  depositFees: AmountJson;
}

/**
 * Information about a selected coin for peer to peer payments.
 */
interface CoinInfo {
  /**
   * Public key of the coin.
   */
  coinPub: string;

  coinPriv: string;

  /**
   * Deposit fee for the coin.
   */
  feeDeposit: AmountJson;

  value: AmountJson;

  denomPubHash: string;

  denomSig: UnblindedSignature;

  maxAge: number;

  ageCommitmentProof?: AgeCommitmentProof;
}

export type SelectPeerCoinsResult =
  | { type: "success"; result: PeerCoinSelectionDetails }
  | {
      type: "failure";
      insufficientBalanceDetails: PayPeerInsufficientBalanceDetails;
    };

export async function queryCoinInfosForSelection(
  ws: InternalWalletState,
  csel: PeerPushPaymentCoinSelection,
): Promise<SpendCoinDetails[]> {
  let infos: SpendCoinDetails[] = [];
  await ws.db
    .mktx((x) => [x.coins, x.denominations])
    .runReadOnly(async (tx) => {
      for (let i = 0; i < csel.coinPubs.length; i++) {
        const coin = await tx.coins.get(csel.coinPubs[i]);
        if (!coin) {
          throw Error("coin not found anymore");
        }
        const denom = await ws.getDenomInfo(
          ws,
          tx,
          coin.exchangeBaseUrl,
          coin.denomPubHash,
        );
        if (!denom) {
          throw Error("denom for coin not found anymore");
        }
        infos.push({
          coinPriv: coin.coinPriv,
          coinPub: coin.coinPub,
          denomPubHash: coin.denomPubHash,
          denomSig: coin.denomSig,
          ageCommitmentProof: coin.ageCommitmentProof,
          contribution: csel.contributions[i],
        });
      }
    });
  return infos;
}

export async function selectPeerCoins(
  ws: InternalWalletState,
  instructedAmount: AmountJson,
): Promise<SelectPeerCoinsResult> {
  if (Amounts.isZero(instructedAmount)) {
    // Other parts of the code assume that we have at least
    // one coin to spend.
    throw new Error("amount of zero not allowed");
  }
  return await ws.db
    .mktx((x) => [
      x.exchanges,
      x.contractTerms,
      x.coins,
      x.coinAvailability,
      x.denominations,
      x.refreshGroups,
      x.peerPushPaymentInitiations,
    ])
    .runReadWrite(async (tx) => {
      const exchanges = await tx.exchanges.iter().toArray();
      const exchangeFeeGap: { [url: string]: AmountJson } = {};
      const currency = Amounts.currencyOf(instructedAmount);
      for (const exch of exchanges) {
        if (exch.detailsPointer?.currency !== currency) {
          continue;
        }
        const coins = (
          await tx.coins.indexes.byBaseUrl.getAll(exch.baseUrl)
        ).filter((x) => x.status === CoinStatus.Fresh);
        const coinInfos: CoinInfo[] = [];
        for (const coin of coins) {
          const denom = await ws.getDenomInfo(
            ws,
            tx,
            coin.exchangeBaseUrl,
            coin.denomPubHash,
          );
          if (!denom) {
            throw Error("denom not found");
          }
          coinInfos.push({
            coinPub: coin.coinPub,
            feeDeposit: Amounts.parseOrThrow(denom.feeDeposit),
            value: Amounts.parseOrThrow(denom.value),
            denomPubHash: denom.denomPubHash,
            coinPriv: coin.coinPriv,
            denomSig: coin.denomSig,
            maxAge: coin.maxAge,
            ageCommitmentProof: coin.ageCommitmentProof,
          });
        }
        if (coinInfos.length === 0) {
          continue;
        }
        coinInfos.sort(
          (o1, o2) =>
            -Amounts.cmp(o1.value, o2.value) ||
            strcmp(o1.denomPubHash, o2.denomPubHash),
        );
        let amountAcc = Amounts.zeroOfCurrency(currency);
        let depositFeesAcc = Amounts.zeroOfCurrency(currency);
        const resCoins: {
          coinPub: string;
          coinPriv: string;
          contribution: AmountString;
          denomPubHash: string;
          denomSig: UnblindedSignature;
          ageCommitmentProof: AgeCommitmentProof | undefined;
        }[] = [];
        let lastDepositFee = Amounts.zeroOfCurrency(currency);
        for (const coin of coinInfos) {
          if (Amounts.cmp(amountAcc, instructedAmount) >= 0) {
            break;
          }
          const gap = Amounts.add(
            coin.feeDeposit,
            Amounts.sub(instructedAmount, amountAcc).amount,
          ).amount;
          const contrib = Amounts.min(gap, coin.value);
          amountAcc = Amounts.add(
            amountAcc,
            Amounts.sub(contrib, coin.feeDeposit).amount,
          ).amount;
          depositFeesAcc = Amounts.add(depositFeesAcc, coin.feeDeposit).amount;
          resCoins.push({
            coinPriv: coin.coinPriv,
            coinPub: coin.coinPub,
            contribution: Amounts.stringify(contrib),
            denomPubHash: coin.denomPubHash,
            denomSig: coin.denomSig,
            ageCommitmentProof: coin.ageCommitmentProof,
          });
          lastDepositFee = coin.feeDeposit;
        }
        if (Amounts.cmp(amountAcc, instructedAmount) >= 0) {
          const res: PeerCoinSelectionDetails = {
            exchangeBaseUrl: exch.baseUrl,
            coins: resCoins,
            depositFees: depositFeesAcc,
          };
          return { type: "success", result: res };
        }
        const diff = Amounts.sub(instructedAmount, amountAcc).amount;
        exchangeFeeGap[exch.baseUrl] = Amounts.add(lastDepositFee, diff).amount;

        continue;
      }
      // We were unable to select coins.
      // Now we need to produce error details.

      const infoGeneral = await getPeerPaymentBalanceDetailsInTx(ws, tx, {
        currency,
      });

      const perExchange: PayPeerInsufficientBalanceDetails["perExchange"] = {};

      let maxFeeGapEstimate = Amounts.zeroOfCurrency(currency);

      for (const exch of exchanges) {
        if (exch.detailsPointer?.currency !== currency) {
          continue;
        }
        const infoExchange = await getPeerPaymentBalanceDetailsInTx(ws, tx, {
          currency,
          restrictExchangeTo: exch.baseUrl,
        });
        let gap =
          exchangeFeeGap[exch.baseUrl] ?? Amounts.zeroOfCurrency(currency);
        if (Amounts.cmp(infoExchange.balanceMaterial, instructedAmount) < 0) {
          // Show fee gap only if we should've been able to pay with the material amount
          gap = Amounts.zeroOfCurrency(currency);
        }
        perExchange[exch.baseUrl] = {
          balanceAvailable: Amounts.stringify(infoExchange.balanceAvailable),
          balanceMaterial: Amounts.stringify(infoExchange.balanceMaterial),
          feeGapEstimate: Amounts.stringify(gap),
        };

        maxFeeGapEstimate = Amounts.max(maxFeeGapEstimate, gap);
      }

      const errDetails: PayPeerInsufficientBalanceDetails = {
        amountRequested: Amounts.stringify(instructedAmount),
        balanceAvailable: Amounts.stringify(infoGeneral.balanceAvailable),
        balanceMaterial: Amounts.stringify(infoGeneral.balanceMaterial),
        feeGapEstimate: Amounts.stringify(maxFeeGapEstimate),
        perExchange,
      };

      return { type: "failure", insufficientBalanceDetails: errDetails };
    });
}

export async function getTotalPeerPaymentCost(
  ws: InternalWalletState,
  pcs: SelectedPeerCoin[],
): Promise<AmountJson> {
  return ws.db
    .mktx((x) => [x.coins, x.denominations])
    .runReadOnly(async (tx) => {
      const costs: AmountJson[] = [];
      for (let i = 0; i < pcs.length; i++) {
        const coin = await tx.coins.get(pcs[i].coinPub);
        if (!coin) {
          throw Error("can't calculate payment cost, coin not found");
        }
        const denom = await tx.denominations.get([
          coin.exchangeBaseUrl,
          coin.denomPubHash,
        ]);
        if (!denom) {
          throw Error(
            "can't calculate payment cost, denomination for coin not found",
          );
        }
        const allDenoms = await tx.denominations.indexes.byExchangeBaseUrl
          .iter(coin.exchangeBaseUrl)
          .filter((x) =>
            Amounts.isSameCurrency(
              DenominationRecord.getValue(x),
              pcs[i].contribution,
            ),
          );
        const amountLeft = Amounts.sub(
          DenominationRecord.getValue(denom),
          pcs[i].contribution,
        ).amount;
        const refreshCost = getTotalRefreshCost(
          allDenoms,
          DenominationRecord.toDenomInfo(denom),
          amountLeft,
          ws.config.testing.denomselAllowLate,
        );
        costs.push(Amounts.parseOrThrow(pcs[i].contribution));
        costs.push(refreshCost);
      }
      const zero = Amounts.zeroOfAmount(pcs[0].contribution);
      return Amounts.sum([zero, ...costs]).amount;
    });
}

export async function checkPeerPushDebit(
  ws: InternalWalletState,
  req: CheckPeerPushDebitRequest,
): Promise<CheckPeerPushDebitResponse> {
  const instructedAmount = Amounts.parseOrThrow(req.amount);
  const coinSelRes = await selectPeerCoins(ws, instructedAmount);
  if (coinSelRes.type === "failure") {
    throw TalerError.fromDetail(
      TalerErrorCode.WALLET_PEER_PUSH_PAYMENT_INSUFFICIENT_BALANCE,
      {
        insufficientBalanceDetails: coinSelRes.insufficientBalanceDetails,
      },
    );
  }
  const totalAmount = await getTotalPeerPaymentCost(
    ws,
    coinSelRes.result.coins,
  );
  return {
    amountEffective: Amounts.stringify(totalAmount),
    amountRaw: req.amount,
  };
}

export async function processPeerPushInitiation(
  ws: InternalWalletState,
  pursePub: string,
): Promise<OperationAttemptResult> {
  const peerPushInitiation = await ws.db
    .mktx((x) => [x.peerPushPaymentInitiations])
    .runReadOnly(async (tx) => {
      return tx.peerPushPaymentInitiations.get(pursePub);
    });
  if (!peerPushInitiation) {
    throw Error("peer push payment not found");
  }

  const purseExpiration = peerPushInitiation.purseExpiration;
  const hContractTerms = peerPushInitiation.contractTermsHash;

  const purseSigResp = await ws.cryptoApi.signPurseCreation({
    hContractTerms,
    mergePub: peerPushInitiation.mergePub,
    minAge: 0,
    purseAmount: peerPushInitiation.amount,
    purseExpiration,
    pursePriv: peerPushInitiation.pursePriv,
  });

  const coins = await queryCoinInfosForSelection(
    ws,
    peerPushInitiation.coinSel,
  );

  const depositSigsResp = await ws.cryptoApi.signPurseDeposits({
    exchangeBaseUrl: peerPushInitiation.exchangeBaseUrl,
    pursePub: peerPushInitiation.pursePub,
    coins,
  });

  const econtractResp = await ws.cryptoApi.encryptContractForMerge({
    contractTerms: peerPushInitiation.contractTerms,
    mergePriv: peerPushInitiation.mergePriv,
    pursePriv: peerPushInitiation.pursePriv,
    pursePub: peerPushInitiation.pursePub,
    contractPriv: peerPushInitiation.contractPriv,
    contractPub: peerPushInitiation.contractPub,
  });

  const createPurseUrl = new URL(
    `purses/${peerPushInitiation.pursePub}/create`,
    peerPushInitiation.exchangeBaseUrl,
  );

  const httpResp = await ws.http.postJson(createPurseUrl.href, {
    amount: peerPushInitiation.amount,
    merge_pub: peerPushInitiation.mergePub,
    purse_sig: purseSigResp.sig,
    h_contract_terms: hContractTerms,
    purse_expiration: purseExpiration,
    deposits: depositSigsResp.deposits,
    min_age: 0,
    econtract: econtractResp.econtract,
  });

  const resp = await httpResp.json();

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

  if (httpResp.status !== 200) {
    throw Error("got error response from exchange");
  }

  await ws.db
    .mktx((x) => [x.peerPushPaymentInitiations])
    .runReadWrite(async (tx) => {
      const ppi = await tx.peerPushPaymentInitiations.get(pursePub);
      if (!ppi) {
        return;
      }
      ppi.status = PeerPushPaymentInitiationStatus.PurseCreated;
      await tx.peerPushPaymentInitiations.put(ppi);
    });

  return {
    type: OperationAttemptResultType.Finished,
    result: undefined,
  };
}

/**
 * Initiate sending a peer-to-peer push payment.
 */
export async function initiatePeerPushPayment(
  ws: InternalWalletState,
  req: InitiatePeerPushPaymentRequest,
): Promise<InitiatePeerPushPaymentResponse> {
  const instructedAmount = Amounts.parseOrThrow(
    req.partialContractTerms.amount,
  );
  const purseExpiration = req.partialContractTerms.purse_expiration;
  const contractTerms = req.partialContractTerms;

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

  const hContractTerms = ContractTermsUtil.hashContractTerms(contractTerms);

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

  const coinSelRes = await selectPeerCoins(ws, instructedAmount);

  if (coinSelRes.type !== "success") {
    throw TalerError.fromDetail(
      TalerErrorCode.WALLET_PEER_PUSH_PAYMENT_INSUFFICIENT_BALANCE,
      {
        insufficientBalanceDetails: coinSelRes.insufficientBalanceDetails,
      },
    );
  }

  const sel = coinSelRes.result;

  logger.info(`selected p2p coins (push): ${j2s(coinSelRes)}`);

  const totalAmount = await getTotalPeerPaymentCost(
    ws,
    coinSelRes.result.coins,
  );

  await ws.db
    .mktx((x) => [
      x.exchanges,
      x.contractTerms,
      x.coins,
      x.coinAvailability,
      x.denominations,
      x.refreshGroups,
      x.peerPushPaymentInitiations,
    ])
    .runReadWrite(async (tx) => {
      // FIXME: Instead of directly doing a spendCoin here,
      // we might want to mark the coins as used and spend them
      // after we've been able to create the purse.
      await spendCoins(ws, tx, {
        allocationId: `txn:peer-push-debit:${pursePair.pub}`,
        coinPubs: sel.coins.map((x) => x.coinPub),
        contributions: sel.coins.map((x) =>
          Amounts.parseOrThrow(x.contribution),
        ),
        refreshReason: RefreshReason.PayPeerPush,
      });

      await tx.peerPushPaymentInitiations.add({
        amount: Amounts.stringify(instructedAmount),
        contractPriv: contractKeyPair.priv,
        contractPub: contractKeyPair.pub,
        contractTermsHash: hContractTerms,
        exchangeBaseUrl: sel.exchangeBaseUrl,
        mergePriv: mergePair.priv,
        mergePub: mergePair.pub,
        purseExpiration: purseExpiration,
        pursePriv: pursePair.priv,
        pursePub: pursePair.pub,
        timestampCreated: TalerProtocolTimestamp.now(),
        status: PeerPushPaymentInitiationStatus.Initiated,
        contractTerms: contractTerms,
        coinSel: {
          coinPubs: sel.coins.map((x) => x.coinPub),
          contributions: sel.coins.map((x) => x.contribution),
        },
        totalCost: Amounts.stringify(totalAmount),
      });

      await tx.contractTerms.put({
        h: hContractTerms,
        contractTermsRaw: contractTerms,
      });
    });

  const taskId = constructTaskIdentifier({
    tag: PendingTaskType.PeerPushInitiation,
    pursePub: pursePair.pub,
  });

  await runOperationWithErrorReporting(ws, taskId, async () => {
    return await processPeerPushInitiation(ws, pursePair.pub);
  });

  return {
    contractPriv: contractKeyPair.priv,
    mergePriv: mergePair.priv,
    pursePub: pursePair.pub,
    exchangeBaseUrl: coinSelRes.result.exchangeBaseUrl,
    talerUri: constructPayPushUri({
      exchangeBaseUrl: coinSelRes.result.exchangeBaseUrl,
      contractPriv: contractKeyPair.priv,
    }),
    transactionId: makeTransactionId(
      TransactionType.PeerPushDebit,
      pursePair.pub,
    ),
  };
}

interface ExchangePurseStatus {
  balance: AmountString;
  deposit_timestamp?: TalerProtocolTimestamp;
}

export const codecForExchangePurseStatus = (): Codec<ExchangePurseStatus> =>
  buildCodecForObject<ExchangePurseStatus>()
    .property("balance", codecForAmountString())
    .property("deposit_timestamp", codecOptional(codecForTimestamp))
    .build("ExchangePurseStatus");

export async function preparePeerPushCredit(
  ws: InternalWalletState,
  req: PreparePeerPushCredit,
): Promise<PreparePeerPushCreditResponse> {
  const uri = parsePayPushUri(req.talerUri);

  if (!uri) {
    throw Error("got invalid taler://pay-push URI");
  }

  const existing = await ws.db
    .mktx((x) => [x.contractTerms, x.peerPushPaymentIncoming])
    .runReadOnly(async (tx) => {
      const existingPushInc =
        await tx.peerPushPaymentIncoming.indexes.byExchangeAndContractPriv.get([
          uri.exchangeBaseUrl,
          uri.contractPriv,
        ]);
      if (!existingPushInc) {
        return;
      }
      const existingContractTermsRec = await tx.contractTerms.get(
        existingPushInc.contractTermsHash,
      );
      if (!existingContractTermsRec) {
        throw Error(
          "contract terms for peer push payment credit not found in database",
        );
      }
      const existingContractTerms = codecForPeerContractTerms().decode(
        existingContractTermsRec.contractTermsRaw,
      );
      return { existingPushInc, existingContractTerms };
    });

  if (existing) {
    return {
      amount: existing.existingContractTerms.amount,
      amountEffective: existing.existingPushInc.estimatedAmountEffective,
      amountRaw: existing.existingContractTerms.amount,
      contractTerms: existing.existingContractTerms,
      peerPushPaymentIncomingId:
        existing.existingPushInc.peerPushPaymentIncomingId,
    };
  }

  const exchangeBaseUrl = uri.exchangeBaseUrl;

  await updateExchangeFromUrl(ws, exchangeBaseUrl);

  const contractPriv = uri.contractPriv;
  const contractPub = encodeCrock(eddsaGetPublic(decodeCrock(contractPriv)));

  const getContractUrl = new URL(`contracts/${contractPub}`, exchangeBaseUrl);

  const contractHttpResp = await ws.http.get(getContractUrl.href);

  const contractResp = await readSuccessResponseJsonOrThrow(
    contractHttpResp,
    codecForExchangeGetContractResponse(),
  );

  const pursePub = contractResp.purse_pub;

  const dec = await ws.cryptoApi.decryptContractForMerge({
    ciphertext: contractResp.econtract,
    contractPriv: contractPriv,
    pursePub: pursePub,
  });

  const getPurseUrl = new URL(`purses/${pursePub}/deposit`, exchangeBaseUrl);

  const purseHttpResp = await ws.http.get(getPurseUrl.href);

  const purseStatus = await readSuccessResponseJsonOrThrow(
    purseHttpResp,
    codecForExchangePurseStatus(),
  );

  const peerPushPaymentIncomingId = encodeCrock(getRandomBytes(32));

  const contractTermsHash = ContractTermsUtil.hashContractTerms(
    dec.contractTerms,
  );

  const withdrawalGroupId = encodeCrock(getRandomBytes(32));

  const wi = await getExchangeWithdrawalInfo(
    ws,
    exchangeBaseUrl,
    Amounts.parseOrThrow(purseStatus.balance),
    undefined,
  );

  await ws.db
    .mktx((x) => [x.contractTerms, x.peerPushPaymentIncoming])
    .runReadWrite(async (tx) => {
      await tx.peerPushPaymentIncoming.add({
        peerPushPaymentIncomingId,
        contractPriv: contractPriv,
        exchangeBaseUrl: exchangeBaseUrl,
        mergePriv: dec.mergePriv,
        pursePub: pursePub,
        timestamp: TalerProtocolTimestamp.now(),
        contractTermsHash,
        status: PeerPushPaymentIncomingStatus.Proposed,
        withdrawalGroupId,
        currency: Amounts.currencyOf(purseStatus.balance),
        estimatedAmountEffective: Amounts.stringify(
          wi.withdrawalAmountEffective,
        ),
      });

      await tx.contractTerms.put({
        h: contractTermsHash,
        contractTermsRaw: dec.contractTerms,
      });
    });

  return {
    amount: purseStatus.balance,
    amountEffective: wi.withdrawalAmountEffective,
    amountRaw: purseStatus.balance,
    contractTerms: dec.contractTerms,
    peerPushPaymentIncomingId,
  };
}

export function talerPaytoFromExchangeReserve(
  exchangeBaseUrl: string,
  reservePub: string,
): string {
  const url = new URL(exchangeBaseUrl);
  let proto: string;
  if (url.protocol === "http:") {
    proto = "taler-reserve-http";
  } else if (url.protocol === "https:") {
    proto = "taler-reserve";
  } else {
    throw Error(`unsupported exchange base URL protocol (${url.protocol})`);
  }

  let path = url.pathname;
  if (!path.endsWith("/")) {
    path = path + "/";
  }

  return `payto://${proto}/${url.host}${url.pathname}${reservePub}`;
}

async function getMergeReserveInfo(
  ws: InternalWalletState,
  req: {
    exchangeBaseUrl: string;
  },
): Promise<ReserveRecord> {
  // We have to eagerly create the key pair outside of the transaction,
  // due to the async crypto API.
  const newReservePair = await ws.cryptoApi.createEddsaKeypair({});

  const mergeReserveRecord: ReserveRecord = await ws.db
    .mktx((x) => [x.exchanges, x.reserves, x.withdrawalGroups])
    .runReadWrite(async (tx) => {
      const ex = await tx.exchanges.get(req.exchangeBaseUrl);
      checkDbInvariant(!!ex);
      if (ex.currentMergeReserveRowId != null) {
        const reserve = await tx.reserves.get(ex.currentMergeReserveRowId);
        checkDbInvariant(!!reserve);
        return reserve;
      }
      const reserve: ReserveRecord = {
        reservePriv: newReservePair.priv,
        reservePub: newReservePair.pub,
      };
      const insertResp = await tx.reserves.put(reserve);
      checkDbInvariant(typeof insertResp.key === "number");
      reserve.rowId = insertResp.key;
      ex.currentMergeReserveRowId = reserve.rowId;
      await tx.exchanges.put(ex);
      return reserve;
    });

  return mergeReserveRecord;
}

export async function processPeerPushCredit(
  ws: InternalWalletState,
  peerPushPaymentIncomingId: string,
): Promise<OperationAttemptResult> {
  let peerInc: PeerPushPaymentIncomingRecord | undefined;
  let contractTerms: PeerContractTerms | undefined;
  await ws.db
    .mktx((x) => [x.contractTerms, x.peerPushPaymentIncoming])
    .runReadWrite(async (tx) => {
      peerInc = await tx.peerPushPaymentIncoming.get(peerPushPaymentIncomingId);
      if (!peerInc) {
        return;
      }
      const ctRec = await tx.contractTerms.get(peerInc.contractTermsHash);
      if (ctRec) {
        contractTerms = ctRec.contractTermsRaw;
      }
      await tx.peerPushPaymentIncoming.put(peerInc);
    });

  if (!peerInc) {
    throw Error(
      `can't accept unknown incoming p2p push payment (${peerPushPaymentIncomingId})`,
    );
  }

  checkDbInvariant(!!contractTerms);

  const amount = Amounts.parseOrThrow(contractTerms.amount);

  if (
    peerInc.status === PeerPushPaymentIncomingStatus.KycRequired &&
    peerInc.kycInfo
  ) {
    const txId = makeTransactionId(
      TransactionType.PeerPushCredit,
      peerInc.peerPushPaymentIncomingId,
    );
    await checkWithdrawalKycStatus(
      ws,
      peerInc.exchangeBaseUrl,
      txId,
      peerInc.kycInfo,
      "individual",
    );
  }

  const mergeReserveInfo = await getMergeReserveInfo(ws, {
    exchangeBaseUrl: peerInc.exchangeBaseUrl,
  });

  const mergeTimestamp = TalerProtocolTimestamp.now();

  const reservePayto = talerPaytoFromExchangeReserve(
    peerInc.exchangeBaseUrl,
    mergeReserveInfo.reservePub,
  );

  const sigRes = await ws.cryptoApi.signPurseMerge({
    contractTermsHash: ContractTermsUtil.hashContractTerms(contractTerms),
    flags: WalletAccountMergeFlags.MergeFullyPaidPurse,
    mergePriv: peerInc.mergePriv,
    mergeTimestamp: mergeTimestamp,
    purseAmount: Amounts.stringify(amount),
    purseExpiration: contractTerms.purse_expiration,
    purseFee: Amounts.stringify(Amounts.zeroOfCurrency(amount.currency)),
    pursePub: peerInc.pursePub,
    reservePayto,
    reservePriv: mergeReserveInfo.reservePriv,
  });

  const mergePurseUrl = new URL(
    `purses/${peerInc.pursePub}/merge`,
    peerInc.exchangeBaseUrl,
  );

  const mergeReq: ExchangePurseMergeRequest = {
    payto_uri: reservePayto,
    merge_timestamp: mergeTimestamp,
    merge_sig: sigRes.mergeSig,
    reserve_sig: sigRes.accountSig,
  };

  const mergeHttpResp = await ws.http.postJson(mergePurseUrl.href, mergeReq);

  if (mergeHttpResp.status === HttpStatusCode.UnavailableForLegalReasons) {
    const respJson = await mergeHttpResp.json();
    const kycPending = codecForWalletKycUuid().decode(respJson);
    logger.info(`kyc uuid response: ${j2s(kycPending)}`);

    await ws.db
      .mktx((x) => [x.peerPushPaymentIncoming])
      .runReadWrite(async (tx) => {
        const peerInc = await tx.peerPushPaymentIncoming.get(
          peerPushPaymentIncomingId,
        );
        if (!peerInc) {
          return;
        }
        peerInc.kycInfo = {
          paytoHash: kycPending.h_payto,
          requirementRow: kycPending.requirement_row,
        };
        peerInc.status = PeerPushPaymentIncomingStatus.KycRequired;
        await tx.peerPushPaymentIncoming.put(peerInc);
      });
    return {
      type: OperationAttemptResultType.Pending,
      result: undefined,
    };
  }

  logger.trace(`merge request: ${j2s(mergeReq)}`);
  const res = await readSuccessResponseJsonOrThrow(
    mergeHttpResp,
    codecForAny(),
  );
  logger.trace(`merge response: ${j2s(res)}`);

  await internalCreateWithdrawalGroup(ws, {
    amount,
    wgInfo: {
      withdrawalType: WithdrawalRecordType.PeerPushCredit,
      contractTerms,
    },
    forcedWithdrawalGroupId: peerInc.withdrawalGroupId,
    exchangeBaseUrl: peerInc.exchangeBaseUrl,
    reserveStatus: WithdrawalGroupStatus.QueryingStatus,
    reserveKeyPair: {
      priv: mergeReserveInfo.reservePriv,
      pub: mergeReserveInfo.reservePub,
    },
  });

  await ws.db
    .mktx((x) => [x.contractTerms, x.peerPushPaymentIncoming])
    .runReadWrite(async (tx) => {
      const peerInc = await tx.peerPushPaymentIncoming.get(
        peerPushPaymentIncomingId,
      );
      if (!peerInc) {
        return;
      }
      if (
        peerInc.status === PeerPushPaymentIncomingStatus.Accepted ||
        peerInc.status === PeerPushPaymentIncomingStatus.KycRequired
      ) {
        peerInc.status = PeerPushPaymentIncomingStatus.WithdrawalCreated;
      }
      await tx.peerPushPaymentIncoming.put(peerInc);
    });

  return {
    type: OperationAttemptResultType.Finished,
    result: undefined,
  };
}

export async function confirmPeerPushCredit(
  ws: InternalWalletState,
  req: ConfirmPeerPushCreditRequest,
): Promise<AcceptPeerPushPaymentResponse> {
  let peerInc: PeerPushPaymentIncomingRecord | undefined;
  let contractTerms: PeerContractTerms | undefined;
  await ws.db
    .mktx((x) => [x.contractTerms, x.peerPushPaymentIncoming])
    .runReadWrite(async (tx) => {
      peerInc = await tx.peerPushPaymentIncoming.get(
        req.peerPushPaymentIncomingId,
      );
      if (!peerInc) {
        return;
      }
      const ctRec = await tx.contractTerms.get(peerInc.contractTermsHash);
      if (ctRec) {
        contractTerms = ctRec.contractTermsRaw;
      }
      if (peerInc.status === PeerPushPaymentIncomingStatus.Proposed) {
        peerInc.status = PeerPushPaymentIncomingStatus.Accepted;
      }
      await tx.peerPushPaymentIncoming.put(peerInc);
    });

  if (!peerInc) {
    throw Error(
      `can't accept unknown incoming p2p push payment (${req.peerPushPaymentIncomingId})`,
    );
  }

  checkDbInvariant(!!contractTerms);

  await updateExchangeFromUrl(ws, peerInc.exchangeBaseUrl);

  const retryTag = TaskIdentifiers.forPeerPushCredit(peerInc);

  await runOperationWithErrorReporting(ws, retryTag, () =>
    processPeerPushCredit(ws, req.peerPushPaymentIncomingId),
  );

  return {
    transactionId: makeTransactionId(
      TransactionType.PeerPushCredit,
      req.peerPushPaymentIncomingId,
    ),
  };
}

export async function processPeerPullDebit(
  ws: InternalWalletState,
  peerPullPaymentIncomingId: string,
): Promise<OperationAttemptResult> {
  const peerPullInc = await ws.db
    .mktx((x) => [x.peerPullPaymentIncoming])
    .runReadOnly(async (tx) => {
      return tx.peerPullPaymentIncoming.get(peerPullPaymentIncomingId);
    });
  if (!peerPullInc) {
    throw Error("peer pull debit not found");
  }
  if (peerPullInc.status === PeerPullPaymentIncomingStatus.Accepted) {
    const pursePub = peerPullInc.pursePub;

    const coinSel = peerPullInc.coinSel;
    if (!coinSel) {
      throw Error("invalid state, no coins selected");
    }

    const coins = await queryCoinInfosForSelection(ws, coinSel);

    const depositSigsResp = await ws.cryptoApi.signPurseDeposits({
      exchangeBaseUrl: peerPullInc.exchangeBaseUrl,
      pursePub: peerPullInc.pursePub,
      coins,
    });

    const purseDepositUrl = new URL(
      `purses/${pursePub}/deposit`,
      peerPullInc.exchangeBaseUrl,
    );

    const depositPayload: ExchangePurseDeposits = {
      deposits: depositSigsResp.deposits,
    };

    if (logger.shouldLogTrace()) {
      logger.trace(`purse deposit payload: ${j2s(depositPayload)}`);
    }

    const httpResp = await ws.http.postJson(
      purseDepositUrl.href,
      depositPayload,
    );
    const resp = await readSuccessResponseJsonOrThrow(httpResp, codecForAny());
    logger.trace(`purse deposit response: ${j2s(resp)}`);
  }

  await ws.db
    .mktx((x) => [x.peerPullPaymentIncoming])
    .runReadWrite(async (tx) => {
      const pi = await tx.peerPullPaymentIncoming.get(
        peerPullPaymentIncomingId,
      );
      if (!pi) {
        throw Error("peer pull payment not found anymore");
      }
      if (pi.status === PeerPullPaymentIncomingStatus.Accepted) {
        pi.status = PeerPullPaymentIncomingStatus.Paid;
      }
      await tx.peerPullPaymentIncoming.put(pi);
    });

  return {
    type: OperationAttemptResultType.Finished,
    result: undefined,
  };
}

export async function confirmPeerPullDebit(
  ws: InternalWalletState,
  req: ConfirmPeerPullDebitRequest,
): Promise<AcceptPeerPullPaymentResponse> {
  const peerPullInc = await ws.db
    .mktx((x) => [x.peerPullPaymentIncoming])
    .runReadOnly(async (tx) => {
      return tx.peerPullPaymentIncoming.get(req.peerPullPaymentIncomingId);
    });

  if (!peerPullInc) {
    throw Error(
      `can't accept unknown incoming p2p pull payment (${req.peerPullPaymentIncomingId})`,
    );
  }

  const instructedAmount = Amounts.parseOrThrow(
    peerPullInc.contractTerms.amount,
  );

  const coinSelRes = await selectPeerCoins(ws, instructedAmount);
  logger.info(`selected p2p coins (pull): ${j2s(coinSelRes)}`);

  if (coinSelRes.type !== "success") {
    throw TalerError.fromDetail(
      TalerErrorCode.WALLET_PEER_PUSH_PAYMENT_INSUFFICIENT_BALANCE,
      {
        insufficientBalanceDetails: coinSelRes.insufficientBalanceDetails,
      },
    );
  }

  const sel = coinSelRes.result;

  const totalAmount = await getTotalPeerPaymentCost(
    ws,
    coinSelRes.result.coins,
  );

  const ppi = await ws.db
    .mktx((x) => [
      x.exchanges,
      x.coins,
      x.denominations,
      x.refreshGroups,
      x.peerPullPaymentIncoming,
      x.coinAvailability,
    ])
    .runReadWrite(async (tx) => {
      await spendCoins(ws, tx, {
        allocationId: `txn:peer-pull-debit:${req.peerPullPaymentIncomingId}`,
        coinPubs: sel.coins.map((x) => x.coinPub),
        contributions: sel.coins.map((x) =>
          Amounts.parseOrThrow(x.contribution),
        ),
        refreshReason: RefreshReason.PayPeerPull,
      });

      const pi = await tx.peerPullPaymentIncoming.get(
        req.peerPullPaymentIncomingId,
      );
      if (!pi) {
        throw Error();
      }
      if (pi.status === PeerPullPaymentIncomingStatus.Proposed) {
        pi.status = PeerPullPaymentIncomingStatus.Accepted;
        pi.coinSel = {
          coinPubs: sel.coins.map((x) => x.coinPub),
          contributions: sel.coins.map((x) => x.contribution),
          totalCost: Amounts.stringify(totalAmount),
        };
      }
      await tx.peerPullPaymentIncoming.put(pi);
      return pi;
    });

  await runOperationWithErrorReporting(
    ws,
    TaskIdentifiers.forPeerPullPaymentDebit(ppi),
    async () => {
      return processPeerPullDebit(ws, ppi.peerPullPaymentIncomingId);
    },
  );

  return {
    transactionId: makeTransactionId(
      TransactionType.PeerPullDebit,
      req.peerPullPaymentIncomingId,
    ),
  };
}

/**
 * Look up information about an incoming peer pull payment.
 * Store the results in the wallet DB.
 */
export async function preparePeerPullDebit(
  ws: InternalWalletState,
  req: PreparePeerPullDebitRequest,
): Promise<PreparePeerPullDebitResponse> {
  const uri = parsePayPullUri(req.talerUri);

  if (!uri) {
    throw Error("got invalid taler://pay-pull URI");
  }

  const existingPullIncomingRecord = await ws.db
    .mktx((x) => [x.peerPullPaymentIncoming])
    .runReadOnly(async (tx) => {
      return tx.peerPullPaymentIncoming.indexes.byExchangeAndContractPriv.get([
        uri.exchangeBaseUrl,
        uri.contractPriv,
      ]);
    });

  if (existingPullIncomingRecord) {
    return {
      amount: existingPullIncomingRecord.contractTerms.amount,
      amountRaw: existingPullIncomingRecord.contractTerms.amount,
      amountEffective: existingPullIncomingRecord.totalCostEstimated,
      contractTerms: existingPullIncomingRecord.contractTerms,
      peerPullPaymentIncomingId:
        existingPullIncomingRecord.peerPullPaymentIncomingId,
    };
  }

  const exchangeBaseUrl = uri.exchangeBaseUrl;
  const contractPriv = uri.contractPriv;
  const contractPub = encodeCrock(eddsaGetPublic(decodeCrock(contractPriv)));

  const getContractUrl = new URL(`contracts/${contractPub}`, exchangeBaseUrl);

  const contractHttpResp = await ws.http.get(getContractUrl.href);

  const contractResp = await readSuccessResponseJsonOrThrow(
    contractHttpResp,
    codecForExchangeGetContractResponse(),
  );

  const pursePub = contractResp.purse_pub;

  const dec = await ws.cryptoApi.decryptContractForDeposit({
    ciphertext: contractResp.econtract,
    contractPriv: contractPriv,
    pursePub: pursePub,
  });

  const getPurseUrl = new URL(`purses/${pursePub}/merge`, exchangeBaseUrl);

  const purseHttpResp = await ws.http.get(getPurseUrl.href);

  const purseStatus = await readSuccessResponseJsonOrThrow(
    purseHttpResp,
    codecForExchangePurseStatus(),
  );

  const peerPullPaymentIncomingId = encodeCrock(getRandomBytes(32));

  let contractTerms: PeerContractTerms;

  if (dec.contractTerms) {
    contractTerms = codecForPeerContractTerms().decode(dec.contractTerms);
    // FIXME: Check that the purseStatus balance matches contract terms amount
  } else {
    // FIXME: In this case, where do we get the purse expiration from?!
    // https://bugs.gnunet.org/view.php?id=7706
    throw Error("pull payments without contract terms not supported yet");
  }

  // FIXME: Why don't we compute the totalCost here?!

  const instructedAmount = Amounts.parseOrThrow(contractTerms.amount);

  const coinSelRes = await selectPeerCoins(ws, instructedAmount);
  logger.info(`selected p2p coins (pull): ${j2s(coinSelRes)}`);

  if (coinSelRes.type !== "success") {
    throw TalerError.fromDetail(
      TalerErrorCode.WALLET_PEER_PUSH_PAYMENT_INSUFFICIENT_BALANCE,
      {
        insufficientBalanceDetails: coinSelRes.insufficientBalanceDetails,
      },
    );
  }

  const totalAmount = await getTotalPeerPaymentCost(
    ws,
    coinSelRes.result.coins,
  );

  await ws.db
    .mktx((x) => [x.peerPullPaymentIncoming])
    .runReadWrite(async (tx) => {
      await tx.peerPullPaymentIncoming.add({
        peerPullPaymentIncomingId,
        contractPriv: contractPriv,
        exchangeBaseUrl: exchangeBaseUrl,
        pursePub: pursePub,
        timestampCreated: TalerProtocolTimestamp.now(),
        contractTerms,
        status: PeerPullPaymentIncomingStatus.Proposed,
        totalCostEstimated: Amounts.stringify(totalAmount),
      });
    });

  return {
    amount: contractTerms.amount,
    amountEffective: Amounts.stringify(totalAmount),
    amountRaw: contractTerms.amount,
    contractTerms: contractTerms,
    peerPullPaymentIncomingId,
  };
}

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.QueryingStatus,
    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.PurseCreated) {
        finPi.status = PeerPullPaymentInitiationStatus.PurseDeposited;
      }
      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.PeerPullInitiation,
    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.PurseDeposited: {
      // 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.PurseCreated:
      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.KycRequired: {
      if (pullIni.kycInfo) {
        const txId = makeTransactionId(
          TransactionType.PeerPullCredit,
          pullIni.pursePub,
        );
        await checkWithdrawalKycStatus(
          ws,
          pullIni.exchangeBaseUrl,
          txId,
          pullIni.kycInfo,
          "individual",
        );
      }
      break;
    }
    case PeerPullPaymentInitiationStatus.Initial:
      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: 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: 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.KycRequired;
        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.PurseCreated;
      await tx.peerPullPaymentInitiations.put(pi2);
    });

  ws.notify({
    type: NotificationType.PeerPullCreditReady,
    transactionId: constructTransactionIdentifier({
      tag: TransactionType.PeerPullCredit,
      pursePub: pullIni.pursePub,
    }),
  });

  return {
    type: OperationAttemptResultType.Finished,
    result: undefined,
  };
}

/**
 * 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.fromTimestamp(e.lastWithdrawal),
              AbsoluteTime.fromTimestamp(candidate.lastWithdrawal),
            ) > 0
          ) {
            candidate = e;
          }
        }
      }
      if (candidate) {
        return candidate.baseUrl;
      }
      return undefined;
    });
  return url;
}

/**
 * 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,
  };
}

/**
 * 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 = TalerProtocolTimestamp.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.Initial,
        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.PeerPullInitiation,
    pursePub: pursePair.pub,
  });

  await runOperationWithErrorReporting(ws, taskId, async () => {
    return processPeerPullCredit(ws, pursePair.pub);
  });

  return {
    talerUri: constructPayPullUri({
      exchangeBaseUrl: exchangeBaseUrl,
      contractPriv: contractKeyPair.priv,
    }),
    transactionId: makeTransactionId(
      TransactionType.PeerPullCredit,
      pursePair.pub,
    ),
  };
}