show pending incoming amount

4 files changed, 229 insertions, 24 deletions

diff --git a/lib/wallet/cryptoLib.ts b/lib/wallet/cryptoLib.ts
index db82b5cf4..498e1cc53 100644
--- a/lib/wallet/cryptoLib.ts
+++ b/lib/wallet/cryptoLib.ts
@@ -324,6 +324,11 @@ namespace RpcFunctions {
                                               native.EddsaPrivateKey.fromCrock(
                                                 meltCoin.coinPriv)).toCrock();
 
+    let valueOutput = Amounts.getZero(newCoinDenoms[0].value.currency);
+    for (let denom of newCoinDenoms) {
+      valueOutput = Amounts.add(valueOutput, denom.value).amount;
+    }
+
     let refreshSession: RefreshSession = {
       meltCoinPub: meltCoin.coinPub,
       newDenoms: newCoinDenoms.map((d) => d.denom_pub),
@@ -336,6 +341,7 @@ namespace RpcFunctions {
       exchangeBaseUrl,
       transferPrivs,
       finished: false,
+      valueOutput,
     };
 
     return refreshSession;
diff --git a/lib/wallet/query.ts b/lib/wallet/query.ts
index 6255ffb94..3571c32c7 100644
--- a/lib/wallet/query.ts
+++ b/lib/wallet/query.ts
@@ -24,12 +24,19 @@
 "use strict";
 
 
+export interface JoinResult<L,R> {
+  left: L;
+  right: R;
+}
+
+
 export class Store<T> {
   name: string;
   validator?: (v: T) => T;
   storeParams: IDBObjectStoreParameters;
 
-  constructor(name: string, storeParams: IDBObjectStoreParameters, validator?: (v: T) => T) {
+  constructor(name: string, storeParams: IDBObjectStoreParameters,
+              validator?: (v: T) => T) {
     this.name = name;
     this.validator = validator;
     this.storeParams = storeParams;
@@ -53,13 +60,16 @@ export class Index<S extends IDBValidKey,T> {
  */
 export interface QueryStream<T> {
   indexJoin<S,I extends IDBValidKey>(index: Index<I,S>,
-                 keyFn: (obj: T) => I): QueryStream<[T, S]>;
-  filter(f: (x: any) => boolean): QueryStream<T>;
+                                     keyFn: (obj: T) => I): QueryStream<[T, S]>;
+  keyJoin<S,I extends IDBValidKey>(store: Store<S>,
+                                   keyFn: (obj: T) => I): QueryStream<JoinResult<T,S>>;
+  filter(f: (T: any) => boolean): QueryStream<T>;
   reduce<S>(f: (v: T, acc: S) => S, start?: S): Promise<S>;
   flatMap(f: (x: T) => T[]): QueryStream<T>;
   toArray(): Promise<T[]>;
 }
 
+export let AbortTransaction = Symbol("abort_transaction");
 
 /**
  * Get an unresolved promise together with its extracted resolve / reject
@@ -96,11 +106,17 @@ abstract class QueryStreamBase<T> implements QueryStream<T> {
   }
 
   indexJoin<S,I extends IDBValidKey>(index: Index<I,S>,
-                 keyFn: (obj: T) => I): QueryStream<[T, S]> {
+                                     keyFn: (obj: T) => I): QueryStream<[T, S]> {
     this.root.addStoreAccess(index.storeName, false);
     return new QueryStreamIndexJoin(this, index.storeName, index.indexName, keyFn);
   }
 
+  keyJoin<S, I extends IDBValidKey>(store: Store<S>,
+                                    keyFn: (obj: T) => I): QueryStream<JoinResult<T, S>> {
+    this.root.addStoreAccess(store.name, false);
+    return new QueryStreamKeyJoin(this, store.name, keyFn);
+  }
+
   filter(f: (x: any) => boolean): QueryStream<T> {
     return new QueryStreamFilter(this, f);
   }
@@ -234,6 +250,42 @@ class QueryStreamIndexJoin<T, S> extends QueryStreamBase<[T, S]> {
 }
 
 
+class QueryStreamKeyJoin<T, S> extends QueryStreamBase<JoinResult<T, S>> {
+  s: QueryStreamBase<T>;
+  storeName: string;
+  key: any;
+
+  constructor(s: QueryStreamBase<T>, storeName: string,
+              key: any) {
+    super(s.root);
+    this.s = s;
+    this.storeName = storeName;
+    this.key = key;
+  }
+
+  subscribe(f: SubscribeFn) {
+    this.s.subscribe((isDone, value, tx) => {
+      if (isDone) {
+        f(true, undefined, tx);
+        return;
+      }
+      console.log("joining on", this.key(value));
+      let s = tx.objectStore(this.storeName);
+      let req = s.openCursor(IDBKeyRange.only(this.key(value)));
+      req.onsuccess = () => {
+        let cursor = req.result;
+        if (cursor) {
+          f(false, {left:value, right: cursor.value}, tx);
+          cursor.continue();
+        } else {
+          f(true, undefined, tx);
+        }
+      }
+    });
+  }
+}
+
+
 class IterQueryStream<T> extends QueryStreamBase<T> {
   private storeName: string;
   private options: any;
@@ -304,7 +356,8 @@ export class QueryRoot {
     return new IterQueryStream(this, store.name, {});
   }
 
-  iterIndex<S extends IDBValidKey,T>(index: Index<S,T>, only?: S): QueryStream<T> {
+  iterIndex<S extends IDBValidKey,T>(index: Index<S,T>,
+                                     only?: S): QueryStream<T> {
     this.stores.add(index.storeName);
     return new IterQueryStream(this, index.storeName, {
       only,
@@ -326,6 +379,30 @@ export class QueryRoot {
   }
 
 
+  mutate<T>(store: Store<T>, key: any, f: (v: T) => T): QueryRoot {
+    let doPut = (tx: IDBTransaction) => {
+      let reqGet = tx.objectStore(store.name).get(key);
+      reqGet.onsuccess = () => {
+        let r = reqGet.result;
+        let m: T;
+        try {
+          m = f(r);
+        } catch (e) {
+          if (e == AbortTransaction) {
+            tx.abort();
+            return;
+          }
+          throw e;
+        }
+
+        tx.objectStore(store.name).put(m);
+      }
+    };
+    this.addWork(doPut, store.name, true);
+    return this;
+  }
+
+
   /**
    * Add all object from an iterable to the given object store.
    * Fails if the object's key is already present
@@ -380,7 +457,8 @@ export class QueryRoot {
   /**
    * Get one object from a store by its key.
    */
-  getIndexed<I extends IDBValidKey,T>(index: Index<I,T>, key: I): Promise<T|undefined> {
+  getIndexed<I extends IDBValidKey,T>(index: Index<I,T>,
+                                      key: I): Promise<T|undefined> {
     if (key === void 0) {
       throw Error("key must not be undefined");
     }
@@ -388,7 +466,9 @@ export class QueryRoot {
     const {resolve, promise} = openPromise();
 
     const doGetIndexed = (tx: IDBTransaction) => {
-      const req = tx.objectStore(index.storeName).index(index.indexName).get(key);
+      const req = tx.objectStore(index.storeName)
+                    .index(index.indexName)
+                    .get(key);
       req.onsuccess = () => {
         resolve(req.result);
       };
@@ -417,6 +497,9 @@ export class QueryRoot {
       tx.oncomplete = () => {
         resolve();
       };
+      tx.onabort = () => {
+        reject(Error("transaction aborted"));
+      };
       for (let w of this.work) {
         w(tx);
       }
diff --git a/lib/wallet/types.ts b/lib/wallet/types.ts
index 5beff72bd..1edfa3601 100644
--- a/lib/wallet/types.ts
+++ b/lib/wallet/types.ts
@@ -42,6 +42,12 @@ export class AmountJson {
 }
 
 
+export interface SignedAmountJson {
+  amount: AmountJson;
+  isNegative: boolean;
+}
+
+
 export interface ReserveRecord {
   reserve_pub: string;
   reserve_priv: string,
@@ -195,6 +201,12 @@ export interface RefreshSession {
   valueWithFee: AmountJson
 
   /**
+   * Sum of the value of denominations we want
+   * to withdraw in this session, without fees.
+   */
+  valueOutput: AmountJson;
+
+  /**
    * Signature to confirm the melting.
    */
   confirmSig: string;
@@ -308,6 +320,15 @@ export class ExchangeHandle {
   static checked: (obj: any) => ExchangeHandle;
 }
 
+export interface WalletBalance {
+  [currency: string]: WalletBalanceEntry;
+}
+
+export interface WalletBalanceEntry {
+  available: AmountJson;
+  pendingIncoming: AmountJson;
+}
+
 
 @Checkable.Class
 export class Contract {
diff --git a/lib/wallet/wallet.ts b/lib/wallet/wallet.ts
index ca8aa895e..380243b44 100644
--- a/lib/wallet/wallet.ts
+++ b/lib/wallet/wallet.ts
@@ -27,10 +27,11 @@ import {
   IExchangeInfo,
   Denomination,
   Notifier,
-  WireInfo, RefreshSession, ReserveRecord, CoinPaySig
+  WireInfo, RefreshSession, ReserveRecord, CoinPaySig, WalletBalance,
+  WalletBalanceEntry
 } from "./types";
 import {HttpResponse, RequestException} from "./http";
-import {QueryRoot, Store, Index} from "./query";
+import {QueryRoot, Store, Index, JoinResult, AbortTransaction} from "./query";
 import {Checkable} from "./checkable";
 import {canonicalizeBaseUrl} from "./helpers";
 import {ReserveCreationInfo, Amounts} from "./types";
@@ -904,10 +905,31 @@ export class Wallet {
     console.log("creating pre coin at", new Date());
     let preCoin = await this.cryptoApi
                             .createPreCoin(denom, reserve);
+
+    let aborted = false;
+
+    function mutateReserve(r: ReserveRecord) {
+      let currentAmount = r.current_amount;
+      if (!currentAmount) {
+        throw Error("can't withdraw from reserve when current amount is" +
+                    " unknown");
+      }
+      let x = Amounts.sub(currentAmount, preCoin.coinValue);
+      if (x.saturated) {
+        aborted = true;
+        throw AbortTransaction;
+      }
+      r.current_amount = x.amount;
+      return r;
+    }
+
     await this.q()
               .put(Stores.precoins, preCoin)
+              .mutate(Stores.reserves, reserve.reserve_pub, mutateReserve)
               .finish();
-    await this.processPreCoin(preCoin);
+    if (!aborted) {
+      await this.processPreCoin(preCoin);
+    }
   }
 
 
@@ -1155,26 +1177,99 @@ export class Wallet {
    * Retrieve a mapping from currency name to the amount
    * that is currenctly available for spending in the wallet.
    */
-  async getBalances(): Promise<any> {
-    function collectBalances(c: Coin, byCurrency: any) {
+  async getBalances(): Promise<WalletBalance> {
+    function ensureEntry(balance: WalletBalance, currency: string) {
+      let entry: WalletBalanceEntry|undefined = balance[currency];
+      let z = Amounts.getZero(currency);
+      if (!entry) {
+        balance[currency] = entry = {
+          available: z,
+          pendingIncoming: z,
+        };
+      }
+      return entry;
+    }
+
+    function collectBalances(c: Coin, balance: WalletBalance) {
       if (c.suspended) {
-        return byCurrency;
+        return balance;
+      }
+      let currency = c.currentAmount.currency;
+      let entry = ensureEntry(balance, currency);
+      entry.available = Amounts.add(entry.available, c.currentAmount).amount;
+      return balance;
+    }
+
+    function collectPendingWithdraw(r: ReserveRecord, balance: WalletBalance) {
+      if (!r.confirmed) {
+        return balance;
       }
-      let acc: AmountJson = byCurrency[c.currentAmount.currency];
-      if (!acc) {
-        acc = Amounts.getZero(c.currentAmount.currency);
+      let entry = ensureEntry(balance, r.requested_amount.currency);
+      let amount = r.current_amount;
+      if (!amount) {
+        amount = r.requested_amount;
       }
-      byCurrency[c.currentAmount.currency] = Amounts.add(c.currentAmount,
-                                                         acc).amount;
-      return byCurrency;
+      if (Amounts.cmp(smallestWithdraw[r.exchange_base_url], amount) < 0) {
+        entry.pendingIncoming = Amounts.add(entry.pendingIncoming,
+                                            amount).amount;
+      }
+      return balance;
     }
 
-    let byCurrency = await (
-      this.q()
-          .iter(Stores.coins)
-          .reduce(collectBalances, {}));
+    function collectPendingRefresh(r: RefreshSession, balance: WalletBalance) {
+      if (!r.finished) {
+        return balance;
+      }
+      let entry = ensureEntry(balance, r.valueWithFee.currency);
+      entry.pendingIncoming = Amounts.add(entry.pendingIncoming,
+                                          r.valueOutput).amount;
+
+      return balance;
+    }
+
+    function collectSmallestWithdraw(e: IExchangeInfo, sw: any) {
+      let min: AmountJson|undefined;
+      for (let d of e.active_denoms) {
+        let v = Amounts.add(d.value, d.fee_withdraw).amount;
+        if (!min) {
+          min = v;
+          continue;
+        }
+        if (Amounts.cmp(v, min) < 0) {
+          min = v;
+        }
+      }
+      sw[e.baseUrl] = min;
+      return sw;
+    }
+
+    let balance = {};
+    // Mapping from exchange pub to smallest
+    // possible amount we can withdraw
+    let smallestWithdraw: {[baseUrl: string]: AmountJson} = {};
+
+    smallestWithdraw = await (this.q()
+                                  .iter(Stores.exchanges)
+                                  .reduce(collectSmallestWithdraw, {}));
+
+    console.log("smallest withdraw", smallestWithdraw);
+
+    await (this.q()
+               .iter(Stores.coins)
+               .reduce(collectBalances, balance));
+
+    await (this.q()
+               .iter(Stores.refresh)
+               .reduce(collectPendingRefresh, balance));
+
+    console.log("balances collected");
+
+    await (this.q()
+               .iter(Stores.reserves)
+               .reduce(collectPendingWithdraw, balance));
+    console.log("balance", balance);
+    return balance;
 
-    return {balances: byCurrency};
   }