1 files changed, 126 insertions, 23 deletions

diff --git a/packages/taler-wallet-core/src/operations/pay-peer.ts b/packages/taler-wallet-core/src/operations/pay-peer.ts
index eda107bea..27363cb3e 100644
--- a/packages/taler-wallet-core/src/operations/pay-peer.ts
+++ b/packages/taler-wallet-core/src/operations/pay-peer.ts
@@ -77,6 +77,7 @@ import {
   PeerPullPaymentIncomingStatus,
   PeerPushPaymentCoinSelection,
   PeerPushPaymentIncomingRecord,
+  PeerPushPaymentIncomingStatus,
   PeerPushPaymentInitiationStatus,
   ReserveRecord,
   WithdrawalGroupStatus,
@@ -619,18 +620,50 @@ export const codecForExchangePurseStatus = (): Codec<ExchangePurseStatus> =>
     .property("balance", codecForAmountString())
     .build("ExchangePurseStatus");
 
-export async function checkPeerPushPayment(
+export async function preparePeerPushCredit(
   ws: InternalWalletState,
   req: CheckPeerPushPaymentRequest,
 ): Promise<CheckPeerPushPaymentResponse> {
-  // FIXME: Check if existing record exists!
-
   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,
+      contractTerms: existing.existingContractTerms,
+      peerPushPaymentIncomingId:
+        existing.existingPushInc.peerPushPaymentIncomingId,
+    };
+  }
+
   const exchangeBaseUrl = uri.exchangeBaseUrl;
 
   await updateExchangeFromUrl(ws, exchangeBaseUrl);
@@ -670,6 +703,8 @@ export async function checkPeerPushPayment(
     dec.contractTerms,
   );
 
+  const withdrawalGroupId = encodeCrock(getRandomBytes(32));
+
   await ws.db
     .mktx((x) => [x.contractTerms, x.peerPushPaymentIncoming])
     .runReadWrite(async (tx) => {
@@ -681,7 +716,8 @@ export async function checkPeerPushPayment(
         pursePub: pursePub,
         timestamp: TalerProtocolTimestamp.now(),
         contractTermsHash,
-        status: OperationStatus.Finished,
+        status: PeerPushPaymentIncomingStatus.Proposed,
+        withdrawalGroupId,
       });
 
       await tx.contractTerms.put({
@@ -754,18 +790,16 @@ async function getMergeReserveInfo(
   return mergeReserveRecord;
 }
 
-export async function acceptPeerPushPayment(
+export async function processPeerPushCredit(
   ws: InternalWalletState,
-  req: AcceptPeerPushPaymentRequest,
-): Promise<AcceptPeerPushPaymentResponse> {
+  peerPushPaymentIncomingId: string,
+): Promise<OperationAttemptResult> {
   let peerInc: PeerPushPaymentIncomingRecord | undefined;
   let contractTerms: PeerContractTerms | undefined;
   await ws.db
     .mktx((x) => [x.contractTerms, x.peerPushPaymentIncoming])
-    .runReadOnly(async (tx) => {
-      peerInc = await tx.peerPushPaymentIncoming.get(
-        req.peerPushPaymentIncomingId,
-      );
+    .runReadWrite(async (tx) => {
+      peerInc = await tx.peerPushPaymentIncoming.get(peerPushPaymentIncomingId);
       if (!peerInc) {
         return;
       }
@@ -773,18 +807,17 @@ export async function acceptPeerPushPayment(
       if (ctRec) {
         contractTerms = ctRec.contractTermsRaw;
       }
+      await tx.peerPushPaymentIncoming.put(peerInc);
     });
 
   if (!peerInc) {
     throw Error(
-      `can't accept unknown incoming p2p push payment (${req.peerPushPaymentIncomingId})`,
+      `can't accept unknown incoming p2p push payment (${peerPushPaymentIncomingId})`,
     );
   }
 
   checkDbInvariant(!!contractTerms);
 
-  await updateExchangeFromUrl(ws, peerInc.exchangeBaseUrl);
-
   const amount = Amounts.parseOrThrow(contractTerms.amount);
 
   const mergeReserveInfo = await getMergeReserveInfo(ws, {
@@ -825,16 +858,17 @@ export async function acceptPeerPushPayment(
 
   const mergeHttpReq = await ws.http.postJson(mergePurseUrl.href, mergeReq);
 
-  logger.info(`merge request: ${j2s(mergeReq)}`);
+  logger.trace(`merge request: ${j2s(mergeReq)}`);
   const res = await readSuccessResponseJsonOrThrow(mergeHttpReq, codecForAny());
-  logger.info(`merge response: ${j2s(res)}`);
+  logger.trace(`merge response: ${j2s(res)}`);
 
-  const wg = await internalCreateWithdrawalGroup(ws, {
+  await internalCreateWithdrawalGroup(ws, {
     amount,
     wgInfo: {
       withdrawalType: WithdrawalRecordType.PeerPushCredit,
       contractTerms,
     },
+    forcedWithdrawalGroupId: peerInc.withdrawalGroupId,
     exchangeBaseUrl: peerInc.exchangeBaseUrl,
     reserveStatus: WithdrawalGroupStatus.QueryingStatus,
     reserveKeyPair: {
@@ -843,10 +877,72 @@ export async function acceptPeerPushPayment(
     },
   });
 
+  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.WithdrawalCreated;
+      }
+      await tx.peerPushPaymentIncoming.put(peerInc);
+    });
+
+  return {
+    type: OperationAttemptResultType.Finished,
+    result: undefined,
+  };
+}
+
+export async function acceptPeerPushPayment(
+  ws: InternalWalletState,
+  req: AcceptPeerPushPaymentRequest,
+): 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 = RetryTags.forPeerPushCredit(peerInc);
+
+  await runOperationWithErrorReporting(ws, retryTag, () =>
+    processPeerPushCredit(ws, req.peerPushPaymentIncomingId),
+  );
+
   return {
     transactionId: makeTransactionId(
       TransactionType.PeerPushCredit,
-      wg.withdrawalGroupId,
+      req.peerPushPaymentIncomingId,
     ),
   };
 }
@@ -1017,7 +1113,7 @@ export async function acceptIncomingPeerPullPayment(
  * Look up information about an incoming peer pull payment.
  * Store the results in the wallet DB.
  */
-export async function prepareIncomingPeerPullPayment(
+export async function preparePeerPullCredit(
   ws: InternalWalletState,
   req: CheckPeerPullPaymentRequest,
 ): Promise<CheckPeerPullPaymentResponse> {
@@ -1135,7 +1231,7 @@ export async function prepareIncomingPeerPullPayment(
   };
 }
 
-export async function processPeerPullInitiation(
+export async function processPeerPullCredit(
   ws: InternalWalletState,
   pursePub: string,
 ): Promise<OperationAttemptResult> {
@@ -1359,6 +1455,8 @@ export async function initiatePeerPullPayment(
 
   const contractKeyPair = await ws.cryptoApi.createEddsaKeypair({});
 
+  const withdrawalGroupId = encodeCrock(getRandomBytes(32));
+
   const mergeReserveRowId = mergeReserveInfo.rowId;
   checkDbInvariant(!!mergeReserveRowId);
 
@@ -1379,6 +1477,7 @@ export async function initiatePeerPullPayment(
         mergeReserveRowId: mergeReserveRowId,
         contractPriv: contractKeyPair.priv,
         contractPub: contractKeyPair.pub,
+        withdrawalGroupId,
       });
       await tx.contractTerms.put({
         contractTermsRaw: contractTerms,
@@ -1394,20 +1493,24 @@ export async function initiatePeerPullPayment(
     ws,
     RetryTags.byPeerPullPaymentInitiationPursePub(pursePair.pub),
     async () => {
-      return processPeerPullInitiation(ws, pursePair.pub);
+      return processPeerPullCredit(ws, pursePair.pub);
     },
   );
 
   // FIXME: Why do we create this only here?
   // What if the previous operation didn't succeed?
 
-  const wg = await internalCreateWithdrawalGroup(ws, {
+  // FIXME: Use a pre-computed withdrawal group ID
+  // so we don't create it multiple times.
+
+  await internalCreateWithdrawalGroup(ws, {
     amount: instructedAmount,
     wgInfo: {
       withdrawalType: WithdrawalRecordType.PeerPullCredit,
       contractTerms,
       contractPriv: contractKeyPair.priv,
     },
+    forcedWithdrawalGroupId: withdrawalGroupId,
     exchangeBaseUrl: exchangeBaseUrl,
     reserveStatus: WithdrawalGroupStatus.QueryingStatus,
     reserveKeyPair: {
@@ -1423,7 +1526,7 @@ export async function initiatePeerPullPayment(
     }),
     transactionId: makeTransactionId(
       TransactionType.PeerPullCredit,
-      wg.withdrawalGroupId,
+      pursePair.pub,
     ),
   };
 }