include (pending) wallet balance in pending ops response

4 files changed, 137 insertions, 100 deletions

diff --git a/src/operations/balance.ts b/src/operations/balance.ts
index d12fbaf7c..03d1b2a9f 100644
--- a/src/operations/balance.ts
+++ b/src/operations/balance.ts
@@ -18,7 +18,7 @@
  * Imports.
  */
 import { WalletBalance, WalletBalanceEntry } from "../types/walletTypes";
-import { Database } from "../util/query";
+import { Database, TransactionHandle } from "../util/query";
 import { InternalWalletState } from "./state";
 import { Stores, TipRecord, CoinStatus } from "../types/dbTypes";
 import * as Amounts from "../util/amounts";
@@ -28,13 +28,14 @@ import { Logger } from "../util/logging";
 const logger = new Logger("withdraw.ts");
 
 /**
- * Get detailed balance information, sliced by exchange and by currency.
+ * Get balance information.
  */
-export async function getBalances(
+export async function getBalancesInsideTransaction(
   ws: InternalWalletState,
+  tx: TransactionHandle,
 ): Promise<WalletBalance> {
-  logger.trace("starting to compute balance");
-  /**
+
+    /**
    * Add amount to a balance field, both for
    * the slicing by exchange and currency.
    */
@@ -73,76 +74,85 @@ export async function getBalances(
     byExchange: {},
   };
 
-  await ws.db.runWithReadTransaction(
-    [Stores.coins, Stores.refreshGroups, Stores.reserves, Stores.purchases, Stores.withdrawalSession],
-    async tx => {
-      await tx.iter(Stores.coins).forEach(c => {
-        if (c.suspended) {
-          return;
-        }
-        if (c.status === CoinStatus.Fresh) {
-          addTo(balanceStore, "available", c.currentAmount, c.exchangeBaseUrl);
-        }
-      });
-      await tx.iter(Stores.refreshGroups).forEach(r => {
-        // Don't count finished refreshes, since the refresh already resulted
-        // in coins being added to the wallet.
-        if (r.timestampFinished) {
-          return;
-        }
-        for (let i = 0; i < r.oldCoinPubs.length; i++) {
-          const session = r.refreshSessionPerCoin[i];
-          if (session) {
-            addTo(
-              balanceStore,
-              "pendingIncoming",
-              session.amountRefreshOutput,
-              session.exchangeBaseUrl,
-            );
-            addTo(
-              balanceStore,
-              "pendingIncomingRefresh",
-              session.amountRefreshOutput,
-              session.exchangeBaseUrl,
-            );
-          }
-        }
-      });
-
-      await tx.iter(Stores.withdrawalSession).forEach(wds => {
-        let w = wds.totalCoinValue;
-        for (let i = 0; i < wds.planchets.length; i++) {
-          if (wds.withdrawn[i]) {
-            const p = wds.planchets[i];
-            if (p) {
-              w = Amounts.sub(w, p.coinValue).amount;
-            }
-          }
-        }
+  await tx.iter(Stores.coins).forEach(c => {
+    if (c.suspended) {
+      return;
+    }
+    if (c.status === CoinStatus.Fresh) {
+      addTo(balanceStore, "available", c.currentAmount, c.exchangeBaseUrl);
+    }
+  });
+  await tx.iter(Stores.refreshGroups).forEach(r => {
+    // Don't count finished refreshes, since the refresh already resulted
+    // in coins being added to the wallet.
+    if (r.timestampFinished) {
+      return;
+    }
+    for (let i = 0; i < r.oldCoinPubs.length; i++) {
+      const session = r.refreshSessionPerCoin[i];
+      if (session) {
         addTo(
           balanceStore,
           "pendingIncoming",
-          w,
-          wds.exchangeBaseUrl,
+          session.amountRefreshOutput,
+          session.exchangeBaseUrl,
+        );
+        addTo(
+          balanceStore,
+          "pendingIncomingRefresh",
+          session.amountRefreshOutput,
+          session.exchangeBaseUrl,
         );
-      });
+      }
+    }
+  });
 
-      await tx.iter(Stores.purchases).forEach(t => {
-        if (t.timestampFirstSuccessfulPay) {
-          return;
-        }
-        for (const c of t.payReq.coins) {
-          addTo(
-            balanceStore,
-            "pendingPayment",
-            Amounts.parseOrThrow(c.contribution),
-            c.exchange_url,
-          );
+  await tx.iter(Stores.withdrawalSession).forEach(wds => {
+    let w = wds.totalCoinValue;
+    for (let i = 0; i < wds.planchets.length; i++) {
+      if (wds.withdrawn[i]) {
+        const p = wds.planchets[i];
+        if (p) {
+          w = Amounts.sub(w, p.coinValue).amount;
         }
-      });
-    },
-  );
+      }
+    }
+    addTo(balanceStore, "pendingIncoming", w, wds.exchangeBaseUrl);
+  });
+
+  await tx.iter(Stores.purchases).forEach(t => {
+    if (t.timestampFirstSuccessfulPay) {
+      return;
+    }
+    for (const c of t.payReq.coins) {
+      addTo(
+        balanceStore,
+        "pendingPayment",
+        Amounts.parseOrThrow(c.contribution),
+        c.exchange_url,
+      );
+    }
+  });
 
-  logger.trace("computed balances:", balanceStore);
   return balanceStore;
 }
+
+/**
+ * Get detailed balance information, sliced by exchange and by currency.
+ */
+export async function getBalances(
+  ws: InternalWalletState,
+): Promise<WalletBalance> {
+  logger.trace("starting to compute balance");
+
+  return await ws.db.runWithReadTransaction([
+    Stores.coins,
+    Stores.refreshGroups,
+    Stores.reserves,
+    Stores.purchases,
+    Stores.withdrawalSession,
+  ],
+  async tx => {
+    return getBalancesInsideTransaction(ws, tx);
+  });
+}
diff --git a/src/operations/pending.ts b/src/operations/pending.ts
index 9e2ff6b32..fce9a3bfb 100644
--- a/src/operations/pending.ts
+++ b/src/operations/pending.ts
@@ -28,9 +28,16 @@ import {
   PendingOperationType,
   ExchangeUpdateOperationStage,
 } from "../types/pending";
-import { Duration, getTimestampNow, Timestamp, getDurationRemaining, durationMin } from "../util/time";
+import {
+  Duration,
+  getTimestampNow,
+  Timestamp,
+  getDurationRemaining,
+  durationMin,
+} from "../util/time";
 import { TransactionHandle } from "../util/query";
 import { InternalWalletState } from "./state";
+import { getBalances, getBalancesInsideTransaction } from "./balance";
 
 function updateRetryDelay(
   oldDelay: Duration,
@@ -38,7 +45,7 @@ function updateRetryDelay(
   retryTimestamp: Timestamp,
 ): Duration {
   const remaining = getDurationRemaining(retryTimestamp, now);
-  const nextDelay =  durationMin(oldDelay, remaining);
+  const nextDelay = durationMin(oldDelay, remaining);
   return nextDelay;
 }
 
@@ -110,14 +117,14 @@ async function gatherExchangePending(
         });
         break;
       case ExchangeUpdateStatus.FinalizeUpdate:
-          resp.pendingOperations.push({
-            type: PendingOperationType.ExchangeUpdate,
-            givesLifeness: false,
-            stage: ExchangeUpdateOperationStage.FinalizeUpdate,
-            exchangeBaseUrl: e.baseUrl,
-            lastError: e.lastError,
-            reason: e.updateReason || "unknown",
-          });
+        resp.pendingOperations.push({
+          type: PendingOperationType.ExchangeUpdate,
+          givesLifeness: false,
+          stage: ExchangeUpdateOperationStage.FinalizeUpdate,
+          exchangeBaseUrl: e.baseUrl,
+          lastError: e.lastError,
+          reason: e.updateReason || "unknown",
+        });
         break;
       default:
         resp.pendingOperations.push({
@@ -400,15 +407,10 @@ async function gatherPurchasePending(
 
 export async function getPendingOperations(
   ws: InternalWalletState,
-  onlyDue: boolean = false,
+  { onlyDue = false } = {},
 ): Promise<PendingOperationsResponse> {
-  const resp: PendingOperationsResponse = {
-    nextRetryDelay: { d_ms: Number.MAX_SAFE_INTEGER },
-    onlyDue: onlyDue,
-    pendingOperations: [],
-  };
   const now = getTimestampNow();
-  await ws.db.runWithReadTransaction(
+  return await ws.db.runWithReadTransaction(
     [
       Stores.exchanges,
       Stores.reserves,
@@ -420,6 +422,13 @@ export async function getPendingOperations(
       Stores.purchases,
     ],
     async tx => {
+      const walletBalance = await getBalancesInsideTransaction(ws, tx);
+      const resp: PendingOperationsResponse = {
+        nextRetryDelay: { d_ms: Number.MAX_SAFE_INTEGER },
+        onlyDue: onlyDue,
+        walletBalance,
+        pendingOperations: [],
+      };
       await gatherExchangePending(tx, now, resp, onlyDue);
       await gatherReservePending(tx, now, resp, onlyDue);
       await gatherRefreshPending(tx, now, resp, onlyDue);
@@ -427,7 +436,7 @@ export async function getPendingOperations(
       await gatherProposalPending(tx, now, resp, onlyDue);
       await gatherTipPending(tx, now, resp, onlyDue);
       await gatherPurchasePending(tx, now, resp, onlyDue);
+      return resp;
     },
   );
-  return resp;
 }
diff --git a/src/types/pending.ts b/src/types/pending.ts
index 3c169c2c4..b86c7797b 100644
--- a/src/types/pending.ts
+++ b/src/types/pending.ts
@@ -21,7 +21,7 @@
 /**
  * Imports.
  */
-import { OperationError } from "./walletTypes";
+import { OperationError, WalletBalance } from "./walletTypes";
 import { WithdrawalSource, RetryInfo, ReserveRecordStatus } from "./dbTypes";
 import { Timestamp, Duration } from "../util/time";
 
@@ -231,7 +231,19 @@ export interface PendingOperationInfoCommon {
  * Response returned from the pending operations API.
  */
 export interface PendingOperationsResponse {
+  /**
+   * List of pending operations.
+   */
   pendingOperations: PendingOperationInfo[];
+
+  /**
+   * Current wallet balance, including pending balances.
+   */
+  walletBalance: WalletBalance;
+
+  /**
+   * When is the next pending operation due to be re-tried?
+   */
   nextRetryDelay: Duration;
 
   /**
diff --git a/src/wallet.ts b/src/wallet.ts
index 32a92cee0..12bc2ccbb 100644
--- a/src/wallet.ts
+++ b/src/wallet.ts
@@ -82,7 +82,10 @@ import {
   getExchangePaytoUri,
   acceptExchangeTermsOfService,
 } from "./operations/exchanges";
-import { processReserve, createTalerWithdrawReserve } from "./operations/reserves";
+import {
+  processReserve,
+  createTalerWithdrawReserve,
+} from "./operations/reserves";
 
 import { InternalWalletState } from "./operations/state";
 import { createReserve, confirmReserve } from "./operations/reserves";
@@ -111,7 +114,6 @@ import {
 } from "./operations/refund";
 import { durationMin, Duration } from "./util/time";
 
-
 const builtinCurrencies: CurrencyRecord[] = [
   {
     auditors: [
@@ -225,7 +227,7 @@ export class Wallet {
    */
   public async runPending(forceNow: boolean = false): Promise<void> {
     const onlyDue = !forceNow;
-    const pendingOpsResponse = await this.getPendingOperations(onlyDue);
+    const pendingOpsResponse = await this.getPendingOperations({ onlyDue });
     for (const p of pendingOpsResponse.pendingOperations) {
       try {
         await this.processOnePendingOperation(p, forceNow);
@@ -260,7 +262,7 @@ export class Wallet {
     await p;
   }
 
-    /**
+  /**
    * Run the wallet until there are no more pending operations that give
    * liveness left.  The wallet will be in a stopped state when this function
    * returns without resolving to an exception.
@@ -304,10 +306,10 @@ export class Wallet {
   private async runRetryLoopImpl(): Promise<void> {
     while (!this.stopped) {
       console.log("running wallet retry loop iteration");
-      let pending = await this.getPendingOperations(true);
+      let pending = await this.getPendingOperations({ onlyDue: true });
       console.log("pending ops", JSON.stringify(pending, undefined, 2));
       if (pending.pendingOperations.length === 0) {
-        const allPending = await this.getPendingOperations(false);
+        const allPending = await this.getPendingOperations({ onlyDue: false });
         let numPending = 0;
         let numGivingLiveness = 0;
         for (const p of allPending.pendingOperations) {
@@ -324,7 +326,7 @@ export class Wallet {
           // Wait for 5 seconds
           dt = { d_ms: 5000 };
         } else {
-          dt = durationMin({ d_ms: 5000}, allPending.nextRetryDelay);
+          dt = durationMin({ d_ms: 5000 }, allPending.nextRetryDelay);
         }
         const timeout = this.timerGroup.resolveAfter(dt);
         this.ws.notify({
@@ -524,11 +526,11 @@ export class Wallet {
     return getHistory(this.ws, historyQuery);
   }
 
-  async getPendingOperations(
-    onlyDue: boolean = false,
-  ): Promise<PendingOperationsResponse> {
+  async getPendingOperations({ onlyDue = false } = {}): Promise<
+    PendingOperationsResponse
+  > {
     return this.ws.memoGetPending.memo(() =>
-      getPendingOperations(this.ws, onlyDue),
+      getPendingOperations(this.ws, { onlyDue }),
     );
   }
 
@@ -702,7 +704,11 @@ export class Wallet {
     selectedExchange: string,
   ): Promise<AcceptWithdrawalResponse> {
     try {
-      return createTalerWithdrawReserve(this.ws, talerWithdrawUri, selectedExchange);
+      return createTalerWithdrawReserve(
+        this.ws,
+        talerWithdrawUri,
+        selectedExchange,
+      );
     } finally {
       this.latch.trigger();
     }