wallet-core: retries for peer pull payments

1 files changed, 135 insertions, 62 deletions

diff --git a/packages/taler-wallet-core/src/operations/pay-peer.ts b/packages/taler-wallet-core/src/operations/pay-peer.ts
index 670b547ae..68b8eb741 100644
--- a/packages/taler-wallet-core/src/operations/pay-peer.ts
+++ b/packages/taler-wallet-core/src/operations/pay-peer.ts
@@ -340,7 +340,7 @@ export async function preparePeerPushPayment(
   };
 }
 
-export async function processPeerPushOutgoing(
+export async function processPeerPushInitiation(
   ws: InternalWalletState,
   pursePub: string,
 ): Promise<OperationAttemptResult> {
@@ -417,6 +417,7 @@ export async function processPeerPushOutgoing(
         return;
       }
       ppi.status = PeerPushPaymentInitiationStatus.PurseCreated;
+      await tx.peerPushPaymentInitiations.put(ppi);
     });
 
   return {
@@ -428,7 +429,7 @@ export async function processPeerPushOutgoing(
 /**
  * Initiate sending a peer-to-peer push payment.
  */
-export async function initiatePeerToPeerPush(
+export async function initiatePeerPushPayment(
   ws: InternalWalletState,
   req: InitiatePeerPushPaymentRequest,
 ): Promise<InitiatePeerPushPaymentResponse> {
@@ -513,7 +514,7 @@ export async function initiatePeerToPeerPush(
     ws,
     RetryTags.byPeerPushPaymentInitiationPursePub(pursePair.pub),
     async () => {
-      return await processPeerPushOutgoing(ws, pursePair.pub);
+      return await processPeerPushInitiation(ws, pursePair.pub);
     },
   );
 
@@ -935,6 +936,115 @@ export async function checkPeerPullPayment(
   };
 }
 
+export async function processPeerPullInitiation(
+  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");
+  }
+
+  if (pullIni.status === OperationStatus.Finished) {
+    logger.warn("peer pull payment initiation is already finished");
+    return {
+      type: OperationAttemptResultType.Finished,
+      result: undefined,
+    }
+  }
+
+  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,
+  );
+
+  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 = OperationStatus.Finished;
+      await tx.peerPullPaymentInitiations.put(pi2);
+    });
+
+  return {
+    type: OperationAttemptResultType.Finished,
+    result: undefined,
+  };
+}
+
 export async function preparePeerPullPayment(
   ws: InternalWalletState,
   req: PreparePeerPullPaymentRequest,
@@ -967,39 +1077,14 @@ export async function initiatePeerPullPayment(
   const instructedAmount = Amounts.parseOrThrow(
     req.partialContractTerms.amount,
   );
-  const purseExpiration = req.partialContractTerms.purse_expiration;
   const contractTerms = req.partialContractTerms;
 
-  const reservePayto = talerPaytoFromExchangeReserve(
-    req.exchangeBaseUrl,
-    mergeReserveInfo.reservePub,
-  );
-
-  const econtractResp = await ws.cryptoApi.encryptContractForDeposit({
-    contractTerms,
-    pursePriv: pursePair.priv,
-    pursePub: pursePair.pub,
-  });
-
   const hContractTerms = ContractTermsUtil.hashContractTerms(contractTerms);
 
-  const purseFee = Amounts.stringify(
-    Amounts.zeroOfCurrency(instructedAmount.currency),
-  );
+  const contractKeyPair = await ws.cryptoApi.createEddsaKeypair({});
 
-  const sigRes = await ws.cryptoApi.signReservePurseCreate({
-    contractTermsHash: hContractTerms,
-    flags: WalletAccountMergeFlags.CreateWithPurseFee,
-    mergePriv: mergePair.priv,
-    mergeTimestamp: mergeTimestamp,
-    purseAmount: req.partialContractTerms.amount,
-    purseExpiration: purseExpiration,
-    purseFee: purseFee,
-    pursePriv: pursePair.priv,
-    pursePub: pursePair.pub,
-    reservePayto,
-    reservePriv: mergeReserveInfo.reservePriv,
-  });
+  const mergeReserveRowId = mergeReserveInfo.rowId;
+  checkDbInvariant(!!mergeReserveRowId);
 
   await ws.db
     .mktx((x) => [x.peerPullPaymentInitiations, x.contractTerms])
@@ -1010,7 +1095,14 @@ export async function initiatePeerPullPayment(
         exchangeBaseUrl: req.exchangeBaseUrl,
         pursePriv: pursePair.priv,
         pursePub: pursePair.pub,
-        status: OperationStatus.Finished,
+        mergePriv: mergePair.priv,
+        mergePub: mergePair.pub,
+        status: OperationStatus.Pending,
+        contractTerms: contractTerms,
+        mergeTimestamp,
+        mergeReserveRowId: mergeReserveRowId,
+        contractPriv: contractKeyPair.priv,
+        contractPub: contractKeyPair.pub,
       });
       await tx.contractTerms.put({
         contractTermsRaw: contractTerms,
@@ -1018,43 +1110,24 @@ export async function initiatePeerPullPayment(
       });
     });
 
-  const reservePurseReqBody: ExchangeReservePurseRequest = {
-    merge_sig: sigRes.mergeSig,
-    merge_timestamp: mergeTimestamp,
-    h_contract_terms: hContractTerms,
-    merge_pub: mergePair.pub,
-    min_age: 0,
-    purse_expiration: purseExpiration,
-    purse_fee: purseFee,
-    purse_pub: pursePair.pub,
-    purse_sig: sigRes.purseSig,
-    purse_value: req.partialContractTerms.amount,
-    reserve_sig: sigRes.accountSig,
-    econtract: econtractResp.econtract,
-  };
-
-  logger.info(`reserve purse request: ${j2s(reservePurseReqBody)}`);
-
-  const reservePurseMergeUrl = new URL(
-    `reserves/${mergeReserveInfo.reservePub}/purse`,
-    req.exchangeBaseUrl,
-  );
+  // 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 httpResp = await ws.http.postJson(
-    reservePurseMergeUrl.href,
-    reservePurseReqBody,
+  await runOperationWithErrorReporting(
+    ws,
+    RetryTags.byPeerPullPaymentInitiationPursePub(pursePair.pub),
+    async () => {
+      return processPeerPullInitiation(ws, pursePair.pub);
+    },
   );
 
-  const resp = await readSuccessResponseJsonOrThrow(httpResp, codecForAny());
-
-  logger.info(`reserve merge response: ${j2s(resp)}`);
-
   const wg = await internalCreateWithdrawalGroup(ws, {
     amount: instructedAmount,
     wgInfo: {
       withdrawalType: WithdrawalRecordType.PeerPullCredit,
       contractTerms,
-      contractPriv: econtractResp.contractPriv,
+      contractPriv: contractKeyPair.priv,
     },
     exchangeBaseUrl: req.exchangeBaseUrl,
     reserveStatus: WithdrawalGroupStatus.QueryingStatus,
@@ -1067,7 +1140,7 @@ export async function initiatePeerPullPayment(
   return {
     talerUri: constructPayPullUri({
       exchangeBaseUrl: req.exchangeBaseUrl,
-      contractPriv: econtractResp.contractPriv,
+      contractPriv: contractKeyPair.priv,
     }),
     transactionId: makeTransactionId(
       TransactionType.PeerPullCredit,