wallet-core: various p2p payment fixes

13 files changed, 537 insertions, 74 deletions

diff --git a/packages/taler-harness/src/integrationtests/test-peer-to-peer-pull.ts b/packages/taler-harness/src/integrationtests/test-peer-to-peer-pull.ts
index 80978e726..15b274e6b 100644
--- a/packages/taler-harness/src/integrationtests/test-peer-to-peer-pull.ts
+++ b/packages/taler-harness/src/integrationtests/test-peer-to-peer-pull.ts
@@ -31,9 +31,7 @@ import {
 export async function runPeerToPeerPullTest(t: GlobalTestState) {
   // Set up test environment
 
-  const { bank, exchange, merchant } = await createSimpleTestkudosEnvironment(
-    t,
-  );
+  const { bank, exchange } = await createSimpleTestkudosEnvironment(t);
 
   // Withdraw digital cash into the wallet.
   const wallet1 = new WalletCli(t, "w1");
diff --git a/packages/taler-util/src/wallet-types.ts b/packages/taler-util/src/wallet-types.ts
index 06d76a6d4..5841b316e 100644
--- a/packages/taler-util/src/wallet-types.ts
+++ b/packages/taler-util/src/wallet-types.ts
@@ -2038,7 +2038,7 @@ export interface PreparePeerPushPaymentRequest {
 
   /**
    * Instructed amount.
-   * 
+   *
    * FIXME: Allow specifying the instructed amount type.
    */
   amount: AmountString;
@@ -2092,7 +2092,14 @@ export interface CheckPeerPushPaymentResponse {
 
 export interface CheckPeerPullPaymentResponse {
   contractTerms: PeerContractTerms;
+  /**
+   * @deprecated Redundant field with bad name, will be removed soon.
+   */
   amount: AmountString;
+
+  amountRaw: AmountString;
+  amountEffective: AmountString;
+
   peerPullPaymentIncomingId: string;
 }
 
@@ -2161,25 +2168,23 @@ export const codecForAcceptPeerPullPaymentRequest =
       .build("AcceptPeerPllPaymentRequest");
 
 export interface PreparePeerPullPaymentRequest {
-  exchangeBaseUrl: string;
+  exchangeBaseUrl?: string;
   amount: AmountString;
 }
 export const codecForPreparePeerPullPaymentRequest =
   (): Codec<PreparePeerPullPaymentRequest> =>
     buildCodecForObject<PreparePeerPullPaymentRequest>()
       .property("amount", codecForAmountString())
-      .property("exchangeBaseUrl", codecForString())
+      .property("exchangeBaseUrl", codecOptional(codecForString()))
       .build("PreparePeerPullPaymentRequest");
 
 export interface PreparePeerPullPaymentResponse {
+  exchangeBaseUrl: string;
   amountRaw: AmountString;
   amountEffective: AmountString;
 }
 export interface InitiatePeerPullPaymentRequest {
-  /**
-   * FIXME: Make this optional?
-   */
-  exchangeBaseUrl: string;
+  exchangeBaseUrl?: string;
   partialContractTerms: PeerContractTerms;
 }
 
@@ -2187,7 +2192,7 @@ export const codecForInitiatePeerPullPaymentRequest =
   (): Codec<InitiatePeerPullPaymentRequest> =>
     buildCodecForObject<InitiatePeerPullPaymentRequest>()
       .property("partialContractTerms", codecForPeerContractTerms())
-      .property("exchangeBaseUrl", codecForString())
+      .property("exchangeBaseUrl", codecOptional(codecForString()))
       .build("InitiatePeerPullPaymentRequest");
 
 export interface InitiatePeerPullPaymentResponse {
diff --git a/packages/taler-wallet-cli/src/index.ts b/packages/taler-wallet-cli/src/index.ts
index aed9a24c0..dbd5ce956 100644
--- a/packages/taler-wallet-cli/src/index.ts
+++ b/packages/taler-wallet-cli/src/index.ts
@@ -18,12 +18,14 @@
  * Imports.
  */
 import {
+  AbsoluteTime,
   addPaytoQueryParams,
   AgeRestriction,
   classifyTalerUri,
   codecForList,
   codecForString,
   CoreApiResponse,
+  Duration,
   encodeCrock,
   getErrorDetailFromException,
   getRandomBytes,
@@ -35,6 +37,7 @@ import {
   setDangerousTimetravel,
   setGlobalLogLevelFromString,
   summarizeTalerErrorDetail,
+  TalerProtocolTimestamp,
   TalerUriType,
   WalletNotification,
 } from "@gnu-taler/taler-util";
@@ -43,6 +46,7 @@ import {
   getenv,
   pathHomedir,
   processExit,
+  readlinePrompt,
   setUnhandledRejectionHandler,
 } from "@gnu-taler/taler-util/compat";
 import { createPlatformHttpLib } from "@gnu-taler/taler-util/http";
@@ -416,7 +420,7 @@ transactionsCli
   });
 
 transactionsCli
-  .subcommand("abortTransaction", "delete", {
+  .subcommand("abortTransaction", "abort", {
     help: "Abort a transaction.",
   })
   .requiredArgument("transactionId", clk.STRING, {
@@ -552,11 +556,16 @@ walletCli
   .subcommand("handleUri", "handle-uri", {
     help: "Handle a taler:// URI.",
   })
-  .requiredArgument("uri", clk.STRING)
+  .maybeArgument("uri", clk.STRING)
   .flag("autoYes", ["-y", "--yes"])
   .action(async (args) => {
     await withWallet(args, async (wallet) => {
-      const uri: string = args.handleUri.uri;
+      let uri;
+      if (args.handleUri.uri) {
+        uri = args.handleUri.uri;
+      } else {
+        uri = await readlinePrompt("Taler URI: ");
+      }
       const uriType = classifyTalerUri(uri);
       switch (uriType) {
         case TalerUriType.TalerPay:
@@ -921,6 +930,141 @@ const advancedCli = walletCli.subcommand("advancedArgs", "advanced", {
 });
 
 advancedCli
+  .subcommand("checkPayPull", "check-pay-pull", {
+    help: "Check fees for a peer-pull payment initiation.",
+  })
+  .requiredArgument("amount", clk.STRING, {
+    help: "Amount to request",
+  })
+  .action(async (args) => {
+    await withWallet(args, async (wallet) => {
+      const resp = await wallet.client.call(
+        WalletApiOperation.PreparePeerPullPayment,
+        {
+          amount: args.checkPayPull.amount,
+        },
+      );
+      console.log(JSON.stringify(resp, undefined, 2));
+    });
+  });
+
+advancedCli
+  .subcommand("prepareIncomingPayPull", "prepare-incoming-pay-pull")
+  .requiredArgument("talerUri", clk.STRING)
+  .action(async (args) => {
+    await withWallet(args, async (wallet) => {
+      const resp = await wallet.client.call(
+        WalletApiOperation.CheckPeerPullPayment,
+        {
+          talerUri: args.prepareIncomingPayPull.talerUri,
+        },
+      );
+      console.log(JSON.stringify(resp, undefined, 2));
+    });
+  });
+
+advancedCli
+  .subcommand("confirmIncomingPayPull", "confirm-incoming-pay-pull")
+  .requiredArgument("peerPullPaymentIncomingId", clk.STRING)
+  .action(async (args) => {
+    await withWallet(args, async (wallet) => {
+      const resp = await wallet.client.call(
+        WalletApiOperation.AcceptPeerPullPayment,
+        {
+          peerPullPaymentIncomingId:
+            args.confirmIncomingPayPull.peerPullPaymentIncomingId,
+        },
+      );
+      console.log(JSON.stringify(resp, undefined, 2));
+    });
+  });
+
+advancedCli
+  .subcommand("initiatePayPull", "initiate-pay-pull", {
+    help: "Initiate a peer-pull payment.",
+  })
+  .requiredArgument("amount", clk.STRING, {
+    help: "Amount to request",
+  })
+  .maybeOption("summary", ["--summary"], clk.STRING, {
+    help: "Summary to use in the contract terms.",
+  })
+  .maybeOption("exchangeBaseUrl", ["--exchange"], clk.STRING)
+  .action(async (args) => {
+    await withWallet(args, async (wallet) => {
+      const resp = await wallet.client.call(
+        WalletApiOperation.InitiatePeerPullPayment,
+        {
+          exchangeBaseUrl: args.initiatePayPull.exchangeBaseUrl,
+          partialContractTerms: {
+            amount: args.initiatePayPull.amount,
+            summary: args.initiatePayPull.summary ?? "Invoice",
+            // FIXME: Make the expiration configurable
+            purse_expiration: AbsoluteTime.toTimestamp(
+              AbsoluteTime.addDuration(
+                AbsoluteTime.now(),
+                Duration.fromSpec({ hours: 1 }),
+              ),
+            ),
+          },
+        },
+      );
+      console.log(JSON.stringify(resp, undefined, 2));
+    });
+  });
+
+advancedCli
+  .subcommand("checkPayPush", "check-pay-push", {
+    help: "Check fees for a peer-push payment.",
+  })
+  .requiredArgument("amount", clk.STRING, {
+    help: "Amount to pay",
+  })
+  .action(async (args) => {
+    await withWallet(args, async (wallet) => {
+      const resp = await wallet.client.call(
+        WalletApiOperation.PreparePeerPushPayment,
+        {
+          amount: args.checkPayPush.amount,
+        },
+      );
+      console.log(JSON.stringify(resp, undefined, 2));
+    });
+  });
+
+advancedCli
+  .subcommand("payPush", "initiate-pay-push", {
+    help: "Initiate a peer-push payment.",
+  })
+  .requiredArgument("amount", clk.STRING, {
+    help: "Amount to pay",
+  })
+  .maybeOption("summary", ["--summary"], clk.STRING, {
+    help: "Summary to use in the contract terms.",
+  })
+  .action(async (args) => {
+    await withWallet(args, async (wallet) => {
+      const resp = await wallet.client.call(
+        WalletApiOperation.InitiatePeerPushPayment,
+        {
+          partialContractTerms: {
+            amount: args.payPush.amount,
+            summary: args.payPush.summary ?? "Payment",
+            // FIXME: Make the expiration configurable
+            purse_expiration: AbsoluteTime.toTimestamp(
+              AbsoluteTime.addDuration(
+                AbsoluteTime.now(),
+                Duration.fromSpec({ hours: 1 }),
+              ),
+            ),
+          },
+        },
+      );
+      console.log(JSON.stringify(resp, undefined, 2));
+    });
+  });
+
+advancedCli
   .subcommand("serve", "serve", {
     help: "Serve the wallet API via a unix domain socket.",
   })
diff --git a/packages/taler-wallet-core/src/db.ts b/packages/taler-wallet-core/src/db.ts
index 75e6408f7..f8fbe2f07 100644
--- a/packages/taler-wallet-core/src/db.ts
+++ b/packages/taler-wallet-core/src/db.ts
@@ -54,9 +54,7 @@ import {
   WireInfo,
   HashCodeString,
   Amounts,
-  AttentionPriority,
   AttentionInfo,
-  AbsoluteTime,
   Logger,
   CoinPublicKeyString,
 } from "@gnu-taler/taler-util";
@@ -72,7 +70,6 @@ import {
   StoreWithIndexes,
 } from "./util/query.js";
 import { RetryInfo, RetryTags } from "./util/retries.js";
-import { Wallet } from "./wallet.js";
 
 /**
  * This file contains the database schema of the Taler wallet together
@@ -121,7 +118,7 @@ export const CURRENT_DB_CONFIG_KEY = "currentMainDbName";
  * backwards-compatible way or object stores and indices
  * are added.
  */
-export const WALLET_DB_MINOR_VERSION = 3;
+export const WALLET_DB_MINOR_VERSION = 4;
 
 /**
  * Ranges for operation status fields.
@@ -539,6 +536,13 @@ export interface ExchangeRecord {
   baseUrl: string;
 
   /**
+   * When did we confirm the last withdrawal from this exchange?
+   *
+   * Used mostly in the UI to suggest exchanges.
+   */
+  lastWithdrawal?: TalerProtocolTimestamp;
+
+  /**
    * Pointer to the current exchange details.
    *
    * Should usually not change.  Only changes when the
@@ -1852,6 +1856,20 @@ export enum PeerPullPaymentIncomingStatus {
   Paid = 50 /* DORMANT_START */,
 }
 
+export interface PeerPullPaymentCoinSelection {
+  contributions: AmountString[];
+  coinPubs: CoinPublicKeyString[];
+
+  /**
+   * Total cost based on the coin selection.
+   * Non undefined after status === "Accepted"
+   */
+  totalCost: AmountString | undefined;
+}
+
+/**
+ * AKA PeerPullDebit.
+ */
 export interface PeerPullPaymentIncomingRecord {
   peerPullPaymentIncomingId: string;
 
@@ -1863,6 +1881,9 @@ export interface PeerPullPaymentIncomingRecord {
 
   timestampCreated: TalerProtocolTimestamp;
 
+  /**
+   * Contract priv that we got from the other party.
+   */
   contractPriv: string;
 
   /**
@@ -1871,10 +1892,11 @@ export interface PeerPullPaymentIncomingRecord {
   status: PeerPullPaymentIncomingStatus;
 
   /**
-   * Total cost based on the coin selection.
-   * Non undefined after status === "Accepted"
+   * Estimated total cost when the record was created.
    */
-  totalCost: AmountString | undefined;
+  totalCostEstimated: AmountString;
+
+  coinSel?: PeerPullPaymentCoinSelection;
 }
 
 /**
@@ -2251,6 +2273,14 @@ export const WalletStoresV1 = {
         "exchangeBaseUrl",
         "pursePub",
       ]),
+      byExchangeAndContractPriv: describeIndex(
+        "byExchangeAndContractPriv",
+        ["exchangeBaseUrl", "contractPriv"],
+        {
+          versionAdded: 4,
+          unique: true,
+        },
+      ),
       byStatus: describeIndex("byStatus", "status"),
     },
   ),
@@ -2484,6 +2514,20 @@ export const walletDbFixups: FixupDescription[] = [
       });
     },
   },
+  {
+    name: "PeerPullPaymentIncomingRecord_totalCostEstimated_add",
+    async fn(tx): Promise<void> {
+      await tx.peerPullPaymentIncoming.iter().forEachAsync(async (pi) => {
+        if (pi.totalCostEstimated) {
+          return;
+        }
+        // Not really the cost, but a good substitute for older transactions
+        // that don't sture the effective cost of the transaction.
+        pi.totalCostEstimated = pi.contractTerms.amount;
+        await tx.peerPullPaymentIncoming.put(pi);
+      });
+    },
+  },
 ];
 
 const logger = new Logger("db.ts");
diff --git a/packages/taler-wallet-core/src/operations/common.ts b/packages/taler-wallet-core/src/operations/common.ts
index e61a6fe95..2db5cd7b4 100644
--- a/packages/taler-wallet-core/src/operations/common.ts
+++ b/packages/taler-wallet-core/src/operations/common.ts
@@ -51,6 +51,7 @@ import {
   OperationAttemptResultType,
   RetryInfo,
 } from "../util/retries.js";
+import { CryptoApiStoppedError } from "../crypto/workers/crypto-dispatcher.js";
 
 const logger = new Logger("operations/common.ts");
 
@@ -260,6 +261,19 @@ export async function runOperationWithErrorReporting<T1, T2>(
         return resp;
     }
   } catch (e) {
+    if (e instanceof CryptoApiStoppedError) {
+      if (ws.stopped) {
+        logger.warn("crypto API stopped during shutdown, ignoring error");
+        return {
+          type: OperationAttemptResultType.Error,
+          errorDetail: makeErrorDetail(
+            TalerErrorCode.WALLET_UNEXPECTED_EXCEPTION,
+            {},
+            "Crypto API stopped during shutdown",
+          ),
+        };
+      }
+    }
     if (e instanceof TalerError) {
       logger.warn("operation processed resulted in error");
       logger.warn(`error was: ${j2s(e.errorDetail)}`);
diff --git a/packages/taler-wallet-core/src/operations/pay-peer.ts b/packages/taler-wallet-core/src/operations/pay-peer.ts
index c1cacead9..eda107bea 100644
--- a/packages/taler-wallet-core/src/operations/pay-peer.ts
+++ b/packages/taler-wallet-core/src/operations/pay-peer.ts
@@ -18,6 +18,7 @@
  * Imports.
  */
 import {
+  AbsoluteTime,
   AcceptPeerPullPaymentRequest,
   AcceptPeerPullPaymentResponse,
   AcceptPeerPushPaymentRequest,
@@ -35,6 +36,7 @@ import {
   codecForAmountString,
   codecForAny,
   codecForExchangeGetContractResponse,
+  codecForPeerContractTerms,
   CoinStatus,
   constructPayPullUri,
   constructPayPushUri,
@@ -545,6 +547,9 @@ export async function initiatePeerPushPayment(
       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),
@@ -846,7 +851,77 @@ export async function acceptPeerPushPayment(
   };
 }
 
-export async function acceptPeerPullPayment(
+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 acceptIncomingPeerPullPayment(
   ws: InternalWalletState,
   req: AcceptPeerPullPaymentRequest,
 ): Promise<AcceptPeerPullPaymentResponse> {
@@ -885,7 +960,7 @@ export async function acceptPeerPullPayment(
     coinSelRes.result.coins,
   );
 
-  await ws.db
+  const ppi = await ws.db
     .mktx((x) => [
       x.exchanges,
       x.coins,
@@ -910,34 +985,26 @@ export async function acceptPeerPullPayment(
       if (!pi) {
         throw Error();
       }
-      pi.status = PeerPullPaymentIncomingStatus.Accepted;
-      pi.totalCost = Amounts.stringify(totalAmount);
+      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;
     });
 
-  const pursePub = peerPullInc.pursePub;
-
-  const coinSel = coinSelRes.result;
-
-  const depositSigsResp = await ws.cryptoApi.signPurseDeposits({
-    exchangeBaseUrl: coinSel.exchangeBaseUrl,
-    pursePub,
-    coins: coinSel.coins,
-  });
-
-  const purseDepositUrl = new URL(
-    `purses/${pursePub}/deposit`,
-    coinSel.exchangeBaseUrl,
+  await runOperationWithErrorReporting(
+    ws,
+    RetryTags.forPeerPullPaymentDebit(ppi),
+    async () => {
+      return processPeerPullDebit(ws, ppi.peerPullPaymentIncomingId);
+    },
   );
 
-  const depositPayload: ExchangePurseDeposits = {
-    deposits: depositSigsResp.deposits,
-  };
-
-  const httpResp = await ws.http.postJson(purseDepositUrl.href, depositPayload);
-  const resp = await readSuccessResponseJsonOrThrow(httpResp, codecForAny());
-  logger.trace(`purse deposit response: ${j2s(resp)}`);
-
   return {
     transactionId: makeTransactionId(
       TransactionType.PeerPullDebit,
@@ -946,14 +1013,38 @@ export async function acceptPeerPullPayment(
   };
 }
 
-export async function checkPeerPullPayment(
+/**
+ * Look up information about an incoming peer pull payment.
+ * Store the results in the wallet DB.
+ */
+export async function prepareIncomingPeerPullPayment(
   ws: InternalWalletState,
   req: CheckPeerPullPaymentRequest,
 ): Promise<CheckPeerPullPaymentResponse> {
   const uri = parsePayPullUri(req.talerUri);
 
   if (!uri) {
-    throw Error("got invalid taler://pay-push 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;
@@ -988,6 +1079,38 @@ export async function checkPeerPullPayment(
 
   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) => {
@@ -997,15 +1120,17 @@ export async function checkPeerPullPayment(
         exchangeBaseUrl: exchangeBaseUrl,
         pursePub: pursePub,
         timestampCreated: TalerProtocolTimestamp.now(),
-        contractTerms: dec.contractTerms,
+        contractTerms,
         status: PeerPullPaymentIncomingStatus.Proposed,
-        totalCost: undefined,
+        totalCostEstimated: Amounts.stringify(totalAmount),
       });
     });
 
   return {
-    amount: purseStatus.balance,
-    contractTerms: dec.contractTerms,
+    amount: contractTerms.amount,
+    amountEffective: Amounts.stringify(totalAmount),
+    amountRaw: contractTerms.amount,
+    contractTerms: contractTerms,
     peerPullPaymentIncomingId,
   };
 }
@@ -1119,12 +1244,75 @@ export async function processPeerPullInitiation(
   };
 }
 
-export async function preparePeerPullPayment(
+/**
+ * Find a prefered 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: PreparePeerPullPaymentRequest,
 ): Promise<PreparePeerPullPaymentResponse> {
-  //FIXME: look up for exchange details and use purse fee
+  // 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?
+
+  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");
+  }
+
   return {
+    exchangeBaseUrl: exchangeUrl,
     amountEffective: req.amount,
     amountRaw: req.amount,
   };
@@ -1137,10 +1325,24 @@ export async function initiatePeerPullPayment(
   ws: InternalWalletState,
   req: InitiatePeerPullPaymentRequest,
 ): Promise<InitiatePeerPullPaymentResponse> {
-  await updateExchangeFromUrl(ws, req.exchangeBaseUrl);
+  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: req.exchangeBaseUrl,
+    exchangeBaseUrl: exchangeBaseUrl,
   });
 
   const mergeTimestamp = TalerProtocolTimestamp.now();
@@ -1166,7 +1368,7 @@ export async function initiatePeerPullPayment(
       await tx.peerPullPaymentInitiations.put({
         amount: req.partialContractTerms.amount,
         contractTermsHash: hContractTerms,
-        exchangeBaseUrl: req.exchangeBaseUrl,
+        exchangeBaseUrl: exchangeBaseUrl,
         pursePriv: pursePair.priv,
         pursePub: pursePair.pub,
         mergePriv: mergePair.priv,
@@ -1196,6 +1398,9 @@ export async function initiatePeerPullPayment(
     },
   );
 
+  // FIXME: Why do we create this only here?
+  // What if the previous operation didn't succeed?
+
   const wg = await internalCreateWithdrawalGroup(ws, {
     amount: instructedAmount,
     wgInfo: {
@@ -1203,7 +1408,7 @@ export async function initiatePeerPullPayment(
       contractTerms,
       contractPriv: contractKeyPair.priv,
     },
-    exchangeBaseUrl: req.exchangeBaseUrl,
+    exchangeBaseUrl: exchangeBaseUrl,
     reserveStatus: WithdrawalGroupStatus.QueryingStatus,
     reserveKeyPair: {
       priv: mergeReserveInfo.reservePriv,
@@ -1213,7 +1418,7 @@ export async function initiatePeerPullPayment(
 
   return {
     talerUri: constructPayPullUri({
-      exchangeBaseUrl: req.exchangeBaseUrl,
+      exchangeBaseUrl: exchangeBaseUrl,
       contractPriv: contractKeyPair.priv,
     }),
     transactionId: makeTransactionId(
@@ -1222,5 +1427,3 @@ export async function initiatePeerPullPayment(
     ),
   };
 }
-
-
diff --git a/packages/taler-wallet-core/src/operations/pending.ts b/packages/taler-wallet-core/src/operations/pending.ts
index a73af528c..d1d1bb03a 100644
--- a/packages/taler-wallet-core/src/operations/pending.ts
+++ b/packages/taler-wallet-core/src/operations/pending.ts
@@ -29,6 +29,7 @@ import {
   OperationStatus,
   OperationStatusRange,
   PeerPushPaymentInitiationStatus,
+  PeerPullPaymentIncomingStatus,
 } from "../db.js";
 import {
   PendingOperationsResponse,
@@ -377,6 +378,32 @@ async function gatherPeerPullInitiationPending(
   });
 }
 
+async function gatherPeerPullDebitPending(
+  ws: InternalWalletState,
+  tx: GetReadOnlyAccess<{
+    peerPullPaymentIncoming: typeof WalletStoresV1.peerPullPaymentIncoming;
+    operationRetries: typeof WalletStoresV1.operationRetries;
+  }>,
+  now: AbsoluteTime,
+  resp: PendingOperationsResponse,
+): Promise<void> {
+  await tx.peerPullPaymentIncoming.iter().forEachAsync(async (pi) => {
+    if (pi.status === PeerPullPaymentIncomingStatus.Paid) {
+      return;
+    }
+    const opId = RetryTags.forPeerPullPaymentDebit(pi);
+    const retryRecord = await tx.operationRetries.get(opId);
+    const timestampDue = retryRecord?.retryInfo.nextRetry ?? AbsoluteTime.now();
+    resp.pendingOperations.push({
+      type: PendingTaskType.PeerPullDebit,
+      ...getPendingCommon(ws, opId, timestampDue),
+      givesLifeness: true,
+      retryInfo: retryRecord?.retryInfo,
+      peerPullPaymentIncomingId: pi.peerPullPaymentIncomingId,
+    });
+  });
+}
+
 async function gatherPeerPushInitiationPending(
   ws: InternalWalletState,
   tx: GetReadOnlyAccess<{
@@ -423,6 +450,7 @@ export async function getPendingOperations(
       x.operationRetries,
       x.peerPullPaymentInitiations,
       x.peerPushPaymentInitiations,
+      x.peerPullPaymentIncoming,
     ])
     .runReadWrite(async (tx) => {
       const resp: PendingOperationsResponse = {
@@ -438,6 +466,7 @@ export async function getPendingOperations(
       await gatherBackupPending(ws, tx, now, resp);
       await gatherPeerPushInitiationPending(ws, tx, now, resp);
       await gatherPeerPullInitiationPending(ws, tx, now, resp);
+      await gatherPeerPullDebitPending(ws, tx, now, resp);
       return resp;
     });
 }
diff --git a/packages/taler-wallet-core/src/operations/transactions.ts b/packages/taler-wallet-core/src/operations/transactions.ts
index d2a7e9d41..1864a0b50 100644
--- a/packages/taler-wallet-core/src/operations/transactions.ts
+++ b/packages/taler-wallet-core/src/operations/transactions.ts
@@ -24,7 +24,6 @@ import {
   constructPayPullUri,
   constructPayPushUri,
   ExtendedStatus,
-  j2s,
   Logger,
   OrderShortInfo,
   PaymentStatus,
@@ -402,8 +401,8 @@ function buildTransactionForPullPaymentDebit(
 ): Transaction {
   return {
     type: TransactionType.PeerPullDebit,
-    amountEffective: pi.totalCost
-      ? pi.totalCost
+    amountEffective: pi.coinSel?.totalCost
+      ? pi.coinSel?.totalCost
       : Amounts.stringify(pi.contractTerms.amount),
     amountRaw: Amounts.stringify(pi.contractTerms.amount),
     exchangeBaseUrl: pi.exchangeBaseUrl,
diff --git a/packages/taler-wallet-core/src/operations/withdraw.ts b/packages/taler-wallet-core/src/operations/withdraw.ts
index f6d79b229..e6c233e2b 100644
--- a/packages/taler-wallet-core/src/operations/withdraw.ts
+++ b/packages/taler-wallet-core/src/operations/withdraw.ts
@@ -1914,6 +1914,12 @@ export async function internalCreateWithdrawalGroup(
         reservePriv: withdrawalGroup.reservePriv,
       });
 
+      const exchange = await tx.exchanges.get(withdrawalGroup.exchangeBaseUrl);
+      if (exchange) {
+        exchange.lastWithdrawal = TalerProtocolTimestamp.now();
+        await tx.exchanges.put(exchange);
+      }
+
       if (!isAudited && !isTrusted) {
         await tx.exchangeTrust.put({
           currency: amount.currency,
diff --git a/packages/taler-wallet-core/src/pending-types.ts b/packages/taler-wallet-core/src/pending-types.ts
index 809fa52d4..fd742250c 100644
--- a/packages/taler-wallet-core/src/pending-types.ts
+++ b/packages/taler-wallet-core/src/pending-types.ts
@@ -39,6 +39,7 @@ export enum PendingTaskType {
   Backup = "backup",
   PeerPushInitiation = "peer-push-initiation",
   PeerPullInitiation = "peer-pull-initiation",
+  PeerPullDebit = "peer-pull-debit",
 }
 
 /**
@@ -57,6 +58,7 @@ export type PendingTaskInfo = PendingTaskInfoCommon &
     | PendingBackupTask
     | PendingPeerPushInitiationTask
     | PendingPeerPullInitiationTask
+    | PendingPeerPullDebitTask
   );
 
 export interface PendingBackupTask {
@@ -91,6 +93,14 @@ export interface PendingPeerPullInitiationTask {
 }
 
 /**
+ * The wallet wants to send a peer pull payment.
+ */
+export interface PendingPeerPullDebitTask {
+  type: PendingTaskType.PeerPullDebit;
+  peerPullPaymentIncomingId: string;
+}
+
+/**
  * The wallet should check whether coins from this exchange
  * need to be auto-refreshed.
  */
diff --git a/packages/taler-wallet-core/src/util/retries.ts b/packages/taler-wallet-core/src/util/retries.ts
index 742381f7b..6485a6b79 100644
--- a/packages/taler-wallet-core/src/util/retries.ts
+++ b/packages/taler-wallet-core/src/util/retries.ts
@@ -31,6 +31,7 @@ import {
   BackupProviderRecord,
   DepositGroupRecord,
   ExchangeRecord,
+  PeerPullPaymentIncomingRecord,
   PeerPullPaymentInitiationRecord,
   PeerPushPaymentInitiationRecord,
   PurchaseRecord,
@@ -215,6 +216,11 @@ export namespace RetryTags {
   ): string {
     return `${PendingTaskType.PeerPullInitiation}:${ppi.pursePub}`;
   }
+  export function forPeerPullPaymentDebit(
+    ppi: PeerPullPaymentIncomingRecord,
+  ): string {
+    return `${PendingTaskType.PeerPullDebit}:${ppi.pursePub}`;
+  }
   export function byPaymentProposalId(proposalId: string): string {
     return `${PendingTaskType.Purchase}:${proposalId}`;
   }
diff --git a/packages/taler-wallet-core/src/wallet-api-types.ts b/packages/taler-wallet-core/src/wallet-api-types.ts
index 3895c944d..093a1b15c 100644
--- a/packages/taler-wallet-core/src/wallet-api-types.ts
+++ b/packages/taler-wallet-core/src/wallet-api-types.ts
@@ -613,7 +613,7 @@ export type InitiatePeerPushPaymentOp = {
 
 /**
  * Check an incoming peer push payment.
- * 
+ *
  * FIXME: Rename to "PrepareIncomingPeerPushPayment"
  */
 export type CheckPeerPushPaymentOp = {
@@ -624,6 +624,8 @@ export type CheckPeerPushPaymentOp = {
 
 /**
  * Accept an incoming peer push payment.
+ *
+ * FIXME: Rename to ConfirmIncomingPeerPushPayment
  */
 export type AcceptPeerPushPaymentOp = {
   op: WalletApiOperation.AcceptPeerPushPayment;
@@ -633,7 +635,7 @@ export type AcceptPeerPushPaymentOp = {
 
 /**
  * Initiate an outgoing peer pull payment.
- * 
+ *
  * FIXME: This does not check anything, so rename to CheckPeerPullPaymentInitiation
  */
 export type PreparePeerPullPaymentOp = {
@@ -654,7 +656,7 @@ export type InitiatePeerPullPaymentOp = {
 /**
  * Prepare for an incoming peer pull payment.
  *
- * FIXME: Rename to "PreparePeerPullPayment"
+ * FIXME: Rename to "PrepareIncomingPeerPullPayment"
  */
 export type CheckPeerPullPaymentOp = {
   op: WalletApiOperation.CheckPeerPullPayment;
@@ -665,7 +667,7 @@ export type CheckPeerPullPaymentOp = {
 /**
  * Accept an incoming peer pull payment (i.e. pay the other party).
  *
- * FIXME: Rename to ConfirmPeerPullPayment
+ * FIXME: Rename to ConfirmIncomingPeerPullPayment
  */
 export type AcceptPeerPullPaymentOp = {
   op: WalletApiOperation.AcceptPeerPullPayment;
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
index 0d02b667b..cbf11d84e 100644
--- a/packages/taler-wallet-core/src/wallet.ts
+++ b/packages/taler-wallet-core/src/wallet.ts
@@ -195,16 +195,17 @@ import {
   processPurchase,
 } from "./operations/pay-merchant.js";
 import {
-  acceptPeerPullPayment,
+  acceptIncomingPeerPullPayment,
   acceptPeerPushPayment,
-  checkPeerPullPayment,
+  prepareIncomingPeerPullPayment,
   checkPeerPushPayment,
   initiatePeerPullPayment,
   initiatePeerPushPayment,
-  preparePeerPullPayment,
+  checkPeerPullPaymentInitiation,
   preparePeerPushPayment,
   processPeerPullInitiation,
   processPeerPushInitiation,
+  processPeerPullDebit,
 } from "./operations/pay-peer.js";
 import { getPendingOperations } from "./operations/pending.js";
 import {
@@ -328,6 +329,8 @@ async function callOperationHandler(
       return await processPeerPushInitiation(ws, pending.pursePub);
     case PendingTaskType.PeerPullInitiation:
       return await processPeerPullInitiation(ws, pending.pursePub);
+    case PendingTaskType.PeerPullDebit:
+      return await processPeerPullDebit(ws, pending.peerPullPaymentIncomingId);
     default:
       return assertUnreachable(pending);
   }
@@ -1440,7 +1443,7 @@ async function dispatchRequestInternal<Op extends WalletApiOperation>(
     }
     case WalletApiOperation.PreparePeerPullPayment: {
       const req = codecForPreparePeerPullPaymentRequest().decode(payload);
-      return await preparePeerPullPayment(ws, req);
+      return await checkPeerPullPaymentInitiation(ws, req);
     }
     case WalletApiOperation.InitiatePeerPullPayment: {
       const req = codecForInitiatePeerPullPaymentRequest().decode(payload);
@@ -1448,11 +1451,11 @@ async function dispatchRequestInternal<Op extends WalletApiOperation>(
     }
     case WalletApiOperation.CheckPeerPullPayment: {
       const req = codecForCheckPeerPullPaymentRequest().decode(payload);
-      return await checkPeerPullPayment(ws, req);
+      return await prepareIncomingPeerPullPayment(ws, req);
     }
     case WalletApiOperation.AcceptPeerPullPayment: {
       const req = codecForAcceptPeerPullPaymentRequest().decode(payload);
-      return await acceptPeerPullPayment(ws, req);
+      return await acceptIncomingPeerPullPayment(ws, req);
     }
     case WalletApiOperation.ApplyDevExperiment: {
       const req = codecForApplyDevExperiment().decode(payload);