wallet-core: retries for peer pull payments

1 files changed, 57 insertions, 0 deletions

diff --git a/packages/taler-wallet-core/src/operations/pending.ts b/packages/taler-wallet-core/src/operations/pending.ts
index d2066d4fc..d9d62ec65 100644
--- a/packages/taler-wallet-core/src/operations/pending.ts
+++ b/packages/taler-wallet-core/src/operations/pending.ts
@@ -28,6 +28,7 @@ import {
   RefreshCoinStatus,
   OperationStatus,
   OperationStatusRange,
+  PeerPushPaymentInitiationStatus,
 } from "../db.js";
 import {
   PendingOperationsResponse,
@@ -341,6 +342,58 @@ async function gatherBackupPending(
   });
 }
 
+async function gatherPeerPullInitiationPending(
+  ws: InternalWalletState,
+  tx: GetReadOnlyAccess<{
+    peerPullPaymentInitiations: typeof WalletStoresV1.peerPullPaymentInitiations;
+    operationRetries: typeof WalletStoresV1.operationRetries;
+  }>,
+  now: AbsoluteTime,
+  resp: PendingOperationsResponse,
+): Promise<void> {
+  await tx.peerPullPaymentInitiations.iter().forEachAsync(async (pi) => {
+    if (pi.status === OperationStatus.Finished) {
+      return;
+    }
+    const opId = RetryTags.forPeerPullPaymentInitiation(pi);
+    const retryRecord = await tx.operationRetries.get(opId);
+    const timestampDue = retryRecord?.retryInfo.nextRetry ?? AbsoluteTime.now();
+    resp.pendingOperations.push({
+      type: PendingTaskType.PeerPullInitiation,
+      ...getPendingCommon(ws, opId, timestampDue),
+      givesLifeness: true,
+      retryInfo: retryRecord?.retryInfo,
+      pursePub: pi.pursePub,
+    });
+  });
+}
+
+async function gatherPeerPushInitiationPending(
+  ws: InternalWalletState,
+  tx: GetReadOnlyAccess<{
+    peerPushPaymentInitiations: typeof WalletStoresV1.peerPushPaymentInitiations;
+    operationRetries: typeof WalletStoresV1.operationRetries;
+  }>,
+  now: AbsoluteTime,
+  resp: PendingOperationsResponse,
+): Promise<void> {
+  await tx.peerPushPaymentInitiations.iter().forEachAsync(async (pi) => {
+    if (pi.status === PeerPushPaymentInitiationStatus.PurseCreated) {
+      return;
+    }
+    const opId = RetryTags.forPeerPushPaymentInitiation(pi);
+    const retryRecord = await tx.operationRetries.get(opId);
+    const timestampDue = retryRecord?.retryInfo.nextRetry ?? AbsoluteTime.now();
+    resp.pendingOperations.push({
+      type: PendingTaskType.PeerPushInitiation,
+      ...getPendingCommon(ws, opId, timestampDue),
+      givesLifeness: true,
+      retryInfo: retryRecord?.retryInfo,
+      pursePub: pi.pursePub,
+    });
+  });
+}
+
 export async function getPendingOperations(
   ws: InternalWalletState,
 ): Promise<PendingOperationsResponse> {
@@ -359,6 +412,8 @@ export async function getPendingOperations(
       x.depositGroups,
       x.recoupGroups,
       x.operationRetries,
+      x.peerPullPaymentInitiations,
+      x.peerPushPaymentInitiations,
     ])
     .runReadWrite(async (tx) => {
       const resp: PendingOperationsResponse = {
@@ -372,6 +427,8 @@ export async function getPendingOperations(
       await gatherPurchasePending(ws, tx, now, resp);
       await gatherRecoupPending(ws, tx, now, resp);
       await gatherBackupPending(ws, tx, now, resp);
+      await gatherPeerPushInitiationPending(ws, tx, now, resp);
+      await gatherPeerPullInitiationPending(ws, tx, now, resp);
       return resp;
     });
 }