make queries then-able

3 files changed, 93 insertions, 29 deletions

diff --git a/lib/wallet/query.ts b/lib/wallet/query.ts
index acf9aa44d..c172bbeb7 100644
--- a/lib/wallet/query.ts
+++ b/lib/wallet/query.ts
@@ -69,6 +69,8 @@ export interface QueryStream<T> {
   map<S>(f: (x:T) => S): QueryStream<S>;
   flatMap<S>(f: (x: T) => S[]): QueryStream<S>;
   toArray(): Promise<T[]>;
+
+  then(onfulfill: any, onreject: any): any;
 }
 
 export let AbortTransaction = Symbol("abort_transaction");
@@ -92,7 +94,7 @@ function openPromise<T>() {
 }
 
 
-abstract class QueryStreamBase<T> implements QueryStream<T> {
+abstract class QueryStreamBase<T> implements QueryStream<T>, PromiseLike<void> {
   abstract subscribe(f: (isDone: boolean,
                          value: any,
                          tx: IDBTransaction) => void): void;
@@ -103,11 +105,15 @@ abstract class QueryStreamBase<T> implements QueryStream<T> {
     this.root = root;
   }
 
-  flatMap<S>(f: (x: T) => T[]): QueryStream<S> {
-    return new QueryStreamFlatMap(this, f);
+  then<R>(onfulfilled: (value: void) => R | PromiseLike<R>, onrejected: (reason: any) => R | PromiseLike<R>): PromiseLike<R>  {
+    return this.root.then(onfulfilled, onrejected);
+  }
+
+  flatMap<S>(f: (x: T) => S[]): QueryStream<S> {
+    return new QueryStreamFlatMap<T,S>(this, f);
   }
 
-  map<S>(f: (x: T) => S): QueryStream<T> {
+  map<S>(f: (x: T) => S): QueryStream<S> {
     return new QueryStreamMap(this, f);
   }
 
@@ -193,11 +199,11 @@ class QueryStreamFilter<T> extends QueryStreamBase<T> {
 }
 
 
-class QueryStreamFlatMap<T> extends QueryStreamBase<T> {
+class QueryStreamFlatMap<T,S> extends QueryStreamBase<S> {
   s: QueryStreamBase<T>;
-  flatMapFn: (v: T) => T[];
+  flatMapFn: (v: T) => S[];
 
-  constructor(s: QueryStreamBase<T>, flatMapFn: (v: T) => T[]) {
+  constructor(s: QueryStreamBase<T>, flatMapFn: (v: T) => S[]) {
     super(s.root);
     this.s = s;
     this.flatMapFn = flatMapFn;
@@ -218,11 +224,11 @@ class QueryStreamFlatMap<T> extends QueryStreamBase<T> {
 }
 
 
-class QueryStreamMap<T> extends QueryStreamBase<T> {
-  s: QueryStreamBase<T>;
-  mapFn: (v: T) => T[];
+class QueryStreamMap<S,T> extends QueryStreamBase<T> {
+  s: QueryStreamBase<S>;
+  mapFn: (v: S) => T;
 
-  constructor(s: QueryStreamBase<T>, mapFn: (v: T) => T[]) {
+  constructor(s: QueryStreamBase<S>, mapFn: (v: S) => T) {
     super(s.root);
     this.s = s;
     this.mapFn = mapFn;
@@ -364,7 +370,7 @@ class IterQueryStream<T> extends QueryStreamBase<T> {
 }
 
 
-export class QueryRoot {
+export class QueryRoot implements PromiseLike<void> {
   private work: ((t: IDBTransaction) => void)[] = [];
   private db: IDBDatabase;
   private stores = new Set();
@@ -376,18 +382,26 @@ export class QueryRoot {
    */
   private hasWrite: boolean;
 
+  private finishScheduled: boolean;
+
   constructor(db: IDBDatabase) {
     this.db = db;
   }
 
+  then<R>(onfulfilled: (value: void) => R | PromiseLike<R>, onrejected: (reason: any) => R | PromiseLike<R>): PromiseLike<R> {
+    return this.finish().then(onfulfilled, onrejected);
+  }
+
   iter<T>(store: Store<T>): QueryStream<T> {
     this.stores.add(store.name);
+    this.scheduleFinish();
     return new IterQueryStream(this, store.name, {});
   }
 
   iterIndex<S extends IDBValidKey,T>(index: Index<S,T>,
                                      only?: S): QueryStream<T> {
     this.stores.add(index.storeName);
+    this.scheduleFinish();
     return new IterQueryStream(this, index.storeName, {
       only,
       indexName: index.indexName
@@ -403,6 +417,7 @@ export class QueryRoot {
     let doPut = (tx: IDBTransaction) => {
       tx.objectStore(store.name).put(val);
     };
+    this.scheduleFinish();
     this.addWork(doPut, store.name, true);
     return this;
   }
@@ -427,6 +442,7 @@ export class QueryRoot {
         tx.objectStore(store.name).put(m);
       }
     };
+    this.scheduleFinish();
     this.addWork(doPut, store.name, true);
     return this;
   }
@@ -443,6 +459,7 @@ export class QueryRoot {
         tx.objectStore(store.name).put(obj);
       }
     };
+    this.scheduleFinish();
     this.addWork(doPutAll, store.name, true);
     return this;
   }
@@ -456,6 +473,7 @@ export class QueryRoot {
     const doAdd = (tx: IDBTransaction) => {
       tx.objectStore(store.name).add(val);
     };
+    this.scheduleFinish();
     this.addWork(doAdd, store.name, true);
     return this;
   }
@@ -509,6 +527,13 @@ export class QueryRoot {
                   .then(() => promise);
   }
 
+  private scheduleFinish() {
+    if (!this.finishScheduled) {
+      Promise.resolve().then(() => this.finish());
+      this.finishScheduled = true;
+    }
+  }
+
   /**
    * Finish the query, and start the query in the first place if necessary.
    */
@@ -543,6 +568,7 @@ export class QueryRoot {
     const doDelete = (tx: IDBTransaction) => {
       tx.objectStore(storeName).delete(key);
     };
+    this.scheduleFinish();
     this.addWork(doDelete, storeName, true);
     return this;
   }
diff --git a/lib/wallet/types.ts b/lib/wallet/types.ts
index 2df8094a7..3d21f8229 100644
--- a/lib/wallet/types.ts
+++ b/lib/wallet/types.ts
@@ -64,10 +64,15 @@ export interface ReserveRecord {
    * be higher than the requested_amount
    */
   requested_amount: AmountJson,
+
+
   /**
-   * Amount we've already withdrawn from the reserve.
+   * What's the current amount that sits
+   * in precoins?
    */
-  withdrawn_amount: AmountJson;
+  precoin_amount: AmountJson;
+
+
   confirmed: boolean,
 }
 
diff --git a/lib/wallet/wallet.ts b/lib/wallet/wallet.ts
index 53508cf59..d4c95a6f0 100644
--- a/lib/wallet/wallet.ts
+++ b/lib/wallet/wallet.ts
@@ -33,7 +33,7 @@ import {
 import {HttpResponse, RequestException} from "./http";
 import {QueryRoot, Store, Index, JoinResult, AbortTransaction} from "./query";
 import {Checkable} from "./checkable";
-import {canonicalizeBaseUrl} from "./helpers";
+import {canonicalizeBaseUrl, amountToPretty} from "./helpers";
 import {ReserveCreationInfo, Amounts} from "./types";
 import {PreCoin} from "./types";
 import {CryptoApi} from "./cryptoApi";
@@ -403,10 +403,13 @@ export class Wallet {
     }
   }
 
-  updateExchanges(): void {
+  async updateExchanges(): Promise<void> {
     console.log("updating exchanges");
 
-    let exchangesUrls = this.q().iter(Stores.exchanges).map((e) => e.baseUrl);
+    let exchangesUrls = await this.q()
+                                  .iter(Stores.exchanges)
+                                  .map((e) => e.baseUrl)
+                                  .toArray();
 
     for (let url of exchangesUrls) {
       this.updateExchangeFromUrl(url)
@@ -766,18 +769,18 @@ export class Wallet {
       const coin = await this.withdrawExecute(preCoin);
 
       const 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,
+
+        console.log(`before committing coin: current ${amountToPretty(r.current_amount!)}, precoin: ${amountToPretty(
+          r.precoin_amount)})}`);
+
+        let x = Amounts.sub(r.precoin_amount,
                             preCoin.coinValue,
                             denom!.fee_withdraw);
         if (x.saturated) {
+          console.error("database inconsistent");
           throw AbortTransaction;
         }
-        r.current_amount = x.amount;
+        r.precoin_amount = x.amount;
         return r;
       };
 
@@ -791,7 +794,6 @@ export class Wallet {
       };
 
       await this.q()
-                .put(Stores.precoins, preCoin)
                 .mutate(Stores.reserves, preCoin.reservePub, mutateReserve)
                 .delete("precoins", coin.coinPub)
                 .add(Stores.coins, coin)
@@ -828,7 +830,7 @@ export class Wallet {
       current_amount: null,
       requested_amount: req.amount,
       confirmed: false,
-      withdrawn_amount: Amounts.getZero(req.amount.currency)
+      precoin_amount: Amounts.getZero(req.amount.currency),
     };
 
     const historyEntry = {
@@ -940,17 +942,47 @@ export class Wallet {
   /**
    * Withdraw coins from a reserve until it is empty.
    */
-  private async depleteReserve(reserve: any,
+  private async depleteReserve(reserve: ReserveRecord,
                                exchange: IExchangeInfo): Promise<number> {
+    if (!reserve.current_amount) {
+      throw Error("can't withdraw when amount is unknown");
+    }
     let denomsAvailable: Denomination[] = copy(exchange.active_denoms);
-    let denomsForWithdraw = getWithdrawDenomList(reserve.current_amount,
+    let denomsForWithdraw = getWithdrawDenomList(reserve.current_amount!,
                                                  denomsAvailable);
 
     let ps = denomsForWithdraw.map(async(denom) => {
+      function mutateReserve(r: ReserveRecord): ReserveRecord {
+        let currentAmount = r.current_amount;
+        if (!currentAmount) {
+          throw Error("can't withdraw when amount is unknown");
+        }
+        r.precoin_amount = Amounts.add(r.precoin_amount,
+                                       denom.value,
+                                       denom.fee_withdraw).amount;
+        let result = Amounts.sub(currentAmount,
+                                 denom.value,
+                                 denom.fee_withdraw);
+        if (result.saturated) {
+          console.error("can't create precoin, saturated");
+          throw AbortTransaction;
+        }
+        r.current_amount = result.amount;
+
+        console.log(`after creating precoin: current ${amountToPretty(r.current_amount)}, precoin: ${amountToPretty(
+          r.precoin_amount)})}`);
+
+        return r;
+      }
+
       let preCoin = await this.cryptoApi
                               .createPreCoin(denom, reserve);
+      await this.q()
+                .put(Stores.precoins, preCoin)
+                .mutate(Stores.reserves, reserve.reserve_pub, mutateReserve);
       await this.processPreCoin(preCoin);
     });
+
     await Promise.all(ps);
     return ps.length;
   }
@@ -1219,6 +1251,7 @@ export class Wallet {
       if (!amount) {
         amount = r.requested_amount;
       }
+      amount = Amounts.add(amount, r.precoin_amount).amount;
       if (Amounts.cmp(smallestWithdraw[r.exchange_base_url], amount) < 0) {
         entry.pendingIncoming = Amounts.add(entry.pendingIncoming,
                                             amount).amount;
@@ -1333,7 +1366,7 @@ export class Wallet {
                                           oldDenom.fee_refresh));
 
     function mutateCoin(c: Coin): Coin {
-      let r = Amounts.sub(coin.currentAmount,
+      let r = Amounts.sub(c.currentAmount,
                           refreshSession.valueWithFee);
       if (r.saturated) {
         // Something else must have written the coin value