redlock

12 files changed, 157 insertions, 117 deletions

diff --git a/index.js b/index.js
index ca33b646..fd365e1a 100644
--- a/index.js
+++ b/index.js
@@ -8,6 +8,7 @@ import { getRates, sendRates } from "$lib/rates";
 import { sendHeartbeat } from "$lib/sockets";
 
 import email from "$routes/email";
+import info from "$routes/info";
 import locations from "$routes/locations";
 import nostr from "$routes/nostr";
 import rates from "$routes/rates";
@@ -25,6 +26,7 @@ getRates();
 setInterval(sendRates, 1000);
 setInterval(sendHeartbeat, 2000);
 
+app.get("/balances", info.balances);
 app.post("/email", email.send);
 
 app.get("/rate", rates.last);
diff --git a/lib/db.js b/lib/db.js
index c21e224f..fb99a756 100644
--- a/lib/db.js
+++ b/lib/db.js
@@ -1,25 +1,9 @@
-import { createClient } from "redis";
-
-export const db = createClient({ url: "redis://db" });
-
-db.on("error", err => console.log("db error", err));
-
-await db.connect();
+import { Redis } from "ioredis";
+import Redlock from "redlock";
 
+export const db = new Redis("redis://db", { enableAutoPipelining: true });
 export default db;
 
 export let g = async k => JSON.parse(await db.get(k));
 export let s = (k, v) => db.set(k, JSON.stringify(v));
-
-let retries = {};
-export let t = async (k, f) => {
-  retries[k] = retries[k] ? retries[k]++ : 0;
-  await db.watch(k);
-  let o = f(await g(k));
-  let m = await db.multi();
-  await s(k, o);
-
-  if (retries[k] < 10) {
-    if (!(await m.exec())) setTimeout(() => t(k, f), 100);
-  } else throw new Error("could not obtain lock");
-};
+export let t = (f) => new Redlock([db]).using([db], 5000, f);
diff --git a/lib/lightning.js b/lib/lightning.js
index be5929b7..57667be2 100644
--- a/lib/lightning.js
+++ b/lib/lightning.js
@@ -4,7 +4,8 @@ import { notify } from "$lib/notifications";
 import { callWebhook } from "$lib/webhooks";
 import ln from "$lib/ln";
 import { l, err, warn } from "$lib/logging";
-import { g, s, db } from "$lib/db";
+import { g, s, t, db } from "$lib/db";
+import Redlock from "redlock";
 
 export async function listenForLightning() {
   let {
@@ -50,22 +51,66 @@ export async function listenForLightning() {
 
     invoice.status = "paid";
     invoice.received += received;
+
+    const redlock = new Redlock([db]);
+
     await s(`invoice:${hash}`, invoice);
+    console.log("BAL", await g(`balance:${uid}`));
+    setTimeout(async () => {
+      let lock = await redlock.acquire(["b"], 5000);
+      try {
+        let bal = await g(`balance:${uid}`);
+        console.log("b bal", bal);
+        await new Promise(r => setTimeout(r, 50));
 
-    let t = async () => {
-      await db.watch(`balance:${uid}`);
-      let b = await g(`balance:${uid}`);
-      b += received;
-      let m = await db.multi();
-      await s(`balance:${uid}`, b);
-      if (!(await m.exec())) t();
-    };
-    await t();
+        await s(`balance:${uid}`, bal + 100);
+        console.log("set b bal", bal + 100);
+      } finally {
+        await lock.release();
+      }
+      // try {
+      //   console.log("tweaking balance");
+      //   await s(`balance:${uid}`, 0);
+      //   await new Promise(r => setTimeout(r, 5));
+      //   console.log("mwahaha", await g(`balance:${uid}`));
+      //   await new Promise(r => setTimeout(r, 600));
+      //   console.log("hmm", await g(`balance:${uid}`));
+      // } catch (e) {
+      //   console.log(e);
+      // }
+    }, 50);
+
+    await new Promise(r => setTimeout(r, 5));
+
+    let lock = await redlock.acquire(["a"], 5000);
+    try {
+      let bal = await g(`balance:${uid}`);
+      console.log("a bal", bal, received);
+      await new Promise(r => setTimeout(r, 500));
+
+      await s(`balance:${uid}`, bal + received);
+      console.log("set a bal", bal + received);
+    } finally {
+      await lock.release();
+    }
+
+    // await t(async signal => {
+    //   console.log("acquiring lock");
+    //   let bal = await g(`balance:${uid}`);
+    //   console.log("bal", bal);
+    //   await new Promise(r => setTimeout(r, 500));
+    //
+    //   if (signal.aborted) throw signal.error;
+    //
+    //   await s(`balance:${uid}`, bal + received);
+    //   console.log("set bal", bal + received);
+    // });
+    console.log("done", await g(`balance:${uid}`));
 
     payment.invoice = invoice;
 
     await s(`payment:${hash}`, payment);
-    await db.lPush(`${uid}:payments`, hash);
+    await db.lpush(`${uid}:payments`, hash);
 
     callWebhook(invoice, payment);
 
diff --git a/lib/nostr.js b/lib/nostr.js
index 23ea937b..c8accf85 100644
--- a/lib/nostr.js
+++ b/lib/nostr.js
@@ -63,17 +63,17 @@ export let fillPool = () => {
         await s(`${pubkey}:follows`, JSON.stringify(ev.tags));
         for (let f of ev.tags) {
           let [_, followPubkey] = f;
-          await db.sAdd(`${followPubkey}:followers`, pubkey);
+          await db.sadd(`${followPubkey}:followers`, pubkey);
         }
         coinos.send(ev);
       } else if (sub.includes("followers")) {
         let followed = sub.split(":")[0];
         let { pubkey } = ev;
-        await db.sAdd(`${followed}:followers`, pubkey);
+        await db.sadd(`${followed}:followers`, pubkey);
         coinos.send(ev);
       } else if (sub.includes("notes")) {
         let pubkey = sub.split(":")[0];
-        await db.sAdd(pubkey, ev.id);
+        await db.sadd(pubkey, ev.id);
         await s(`ev:${ev.id}`, JSON.stringify(ev));
       }
     } catch (e) {
diff --git a/lib/sendInternal.js b/lib/sendInternal.js
index 32e94aa2..6054f753 100644
--- a/lib/sendInternal.js
+++ b/lib/sendInternal.js
@@ -39,7 +39,7 @@ export default async ({ amount, hash, memo, tip }, user) => {
     };
 
     await s(`payment:${hash}`, p1);
-    await db.lPush(`${user.id}:payments`, hash);
+    await db.lpush(`${user.id}:payments`, hash);
 
     l("sent internal", user.username, -p1.amount);
     emit(user.id, "payment", p1);
@@ -79,7 +79,7 @@ export default async ({ amount, hash, memo, tip }, user) => {
       invoice.received += amount;
       await s(`invoice:${hash}`, invoice);
       await s(`payment:${id}`, p2);
-      await db.lPush(`${recipient.id}:payments`, id);
+      await db.lpush(`${recipient.id}:payments`, id);
 
       l("received internal", recipient.username, amount);
       emit(recipient.username, "payment", p2);
diff --git a/package.json b/package.json
index 2c1b7258..6f1c45f9 100644
--- a/package.json
+++ b/package.json
@@ -18,6 +18,7 @@
     "fastify": "^4.5.3",
     "file-type": "^18.0.0",
     "got": "^12.5.3",
+    "ioredis": "^5.2.5",
     "jsonwebtoken": "^8.5.1",
     "lnurl": "^0.24.2",
     "nodemon": "^2.0.7",
@@ -32,6 +33,7 @@
     "query-string": "^6.13.1",
     "readable-stream-clone": "^0.0.7",
     "redis": "^4.3.1",
+    "redlock": "5.0.0-beta.2",
     "sharp": "^0.31.1",
     "uuid": "8",
     "web-push": "^3.4.4",
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 5c9f09dd..4c22c13f 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -12,6 +12,7 @@ specifiers:
   fastify: ^4.5.3
   file-type: ^18.0.0
   got: ^12.5.3
+  ioredis: ^5.2.5
   jsonwebtoken: ^8.5.1
   lnurl: ^0.24.2
   nodemon: ^2.0.7
@@ -27,6 +28,7 @@ specifiers:
   query-string: ^6.13.1
   readable-stream-clone: ^0.0.7
   redis: ^4.3.1
+  redlock: 5.0.0-beta.2
   sharp: ^0.31.1
   uuid: '8'
   web-push: ^3.4.4
@@ -44,6 +46,7 @@ dependencies:
   fastify: 4.8.1
   file-type: 18.0.0
   got: 12.5.3
+  ioredis: 5.2.5
   jsonwebtoken: 8.5.1
   lnurl: 0.24.2
   nodemon: 2.0.20
@@ -58,6 +61,7 @@ dependencies:
   query-string: 6.14.1
   readable-stream-clone: 0.0.7
   redis: 4.3.1
+  redlock: 5.0.0-beta.2
   sharp: 0.31.1
   uuid: 8.3.2
   web-push: 3.5.0
@@ -175,6 +179,10 @@ packages:
       - utf-8-validate
     dev: false
 
+  /@ioredis/commands/1.2.0:
+    resolution: {integrity: sha512-Sx1pU8EM64o2BrqNpEO1CNLtKQwyhuXuqyfH7oGKCk+1a33d2r5saW8zNwm3j6BTExtjrv2BxTgzzkMwts6vGg==}
+    dev: false
+
   /@mapbox/node-pre-gyp/1.0.10:
     resolution: {integrity: sha512-4ySo4CjzStuprMwk35H5pPbkymjv1SF3jGLj6rAHp/xT/RF7TL7bd9CTm1xDY49K2qF7jmR/g7k+SkLETP6opA==}
     hasBin: true
@@ -856,6 +864,11 @@ packages:
     resolution: {integrity: sha512-bd2L678uiWATM6m5Z1VzNCErI3jiGzt6HGY8OVICs40JQq/HALfbyNJmp0UDakEY4pMMaN0Ly5om/B1VI/+xfQ==}
     dev: false
 
+  /denque/2.1.0:
+    resolution: {integrity: sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==}
+    engines: {node: '>=0.10'}
+    dev: false
+
   /depd/2.0.0:
     resolution: {integrity: sha512-g7nH6P6dyDioJogAAGprGpCtVImJhpPk/roCzdb3fIh61/s/nPsfR6onyMwkCAR/OlC3yBC0lESvUoQEAssIrw==}
     engines: {node: '>= 0.8'}
@@ -1390,6 +1403,23 @@ packages:
     resolution: {integrity: sha512-JV/yugV2uzW5iMRSiZAyDtQd+nxtUnjeLt0acNdw98kKLrvuRVyB80tsREOE7yvGVgalhZ6RNXCmEHkUKBKxew==}
     dev: false
 
+  /ioredis/5.2.5:
+    resolution: {integrity: sha512-7HKo/ClM2DGLRXdFq8ruS3Uuadensz4A76wPOU0adqlOqd1qkhoLPDaBhmVhUhNGpB+J65/bhLmNB8DDY99HJQ==}
+    engines: {node: '>=12.22.0'}
+    dependencies:
+      '@ioredis/commands': 1.2.0
+      cluster-key-slot: 1.1.0
+      debug: 4.3.4
+      denque: 2.1.0
+      lodash.defaults: 4.2.0
+      lodash.isarguments: 3.1.0
+      redis-errors: 1.2.0
+      redis-parser: 3.0.0
+      standard-as-callback: 2.1.0
+    transitivePeerDependencies:
+      - supports-color
+    dev: false
+
   /ip/2.0.0:
     resolution: {integrity: sha512-WKa+XuLG1A1R0UWhl2+1XQSi+fZWMsYKffMZTTYsiZaUD8k2yDAj5atimTUD2TZkyCkNEeYE5NhFZmupOGtjYQ==}
     dev: false
@@ -1544,10 +1574,18 @@ packages:
       - supports-color
     dev: false
 
+  /lodash.defaults/4.2.0:
+    resolution: {integrity: sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==}
+    dev: false
+
   /lodash.includes/4.3.0:
     resolution: {integrity: sha512-W3Bx6mdkRTGtlJISOvVD/lbqjTlPPUDTMnlXZFnVwi9NKJ6tiAk6LVdlhZMm17VZisqhKcgzpO5Wz91PCt5b0w==}
     dev: false
 
+  /lodash.isarguments/3.1.0:
+    resolution: {integrity: sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==}
+    dev: false
+
   /lodash.isboolean/3.0.3:
     resolution: {integrity: sha512-Bz5mupy2SVbPHURB98VAcw+aHh4vRV5IPNhILUCsOzRmsTmSQ17jIuqopAentWoehktxGd9e/hbIXq980/1QJg==}
     dev: false
@@ -1737,6 +1775,10 @@ packages:
       semver: 7.3.8
     dev: false
 
+  /node-abort-controller/3.0.1:
+    resolution: {integrity: sha512-/ujIVxthRs+7q6hsdjHMaj8hRG9NuWmwrz+JdRwZ14jdFoKSkm+vDsCbF9PLpnSqjaWQJuTmVtcWHNLr+vrOFw==}
+    dev: false
+
   /node-addon-api/2.0.2:
     resolution: {integrity: sha512-Ntyt4AIXyaLIuMHF6IOoTakB3K+RWxwtsHNRxllEoA6vPwP9o4866g6YWDLUdnucilZhmkxiHwHr11gAENw+QA==}
     dev: false
@@ -2129,6 +2171,18 @@ packages:
     engines: {node: '>= 12.13.0'}
     dev: false
 
+  /redis-errors/1.2.0:
+    resolution: {integrity: sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==}
+    engines: {node: '>=4'}
+    dev: false
+
+  /redis-parser/3.0.0:
+    resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==}
+    engines: {node: '>=4'}
+    dependencies:
+      redis-errors: 1.2.0
+    dev: false
+
   /redis/4.3.1:
     resolution: {integrity: sha512-cM7yFU5CA6zyCF7N/+SSTcSJQSRMEKN0k0Whhu6J7n9mmXRoXugfWDBo5iOzGwABmsWKSwGPTU5J4Bxbl+0mrA==}
     dependencies:
@@ -2140,6 +2194,13 @@ packages:
       '@redis/time-series': 1.0.3_@[email protected]
     dev: false
 
+  /redlock/5.0.0-beta.2:
+    resolution: {integrity: sha512-2RDWXg5jgRptDrB1w9O/JgSZC0j7y4SlaXnor93H/UJm/QyDiFgBKNtrh0TI6oCXqYSaSoXxFh6Sd3VtYfhRXw==}
+    engines: {node: '>=12'}
+    dependencies:
+      node-abort-controller: 3.0.1
+    dev: false
+
   /require-from-string/2.0.2:
     resolution: {integrity: sha512-Xf0nWe6RseziFMu+Ap9biiUbmplq6S9/p+7w7YXP/JBHhrUDDUhwa+vANyubuqfZWTveU//DYVGsDG7RKL/vEw==}
     engines: {node: '>=0.10.0'}
@@ -2403,6 +2464,10 @@ packages:
     engines: {node: '>= 10.x'}
     dev: false
 
+  /standard-as-callback/2.1.0:
+    resolution: {integrity: sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==}
+    dev: false
+
   /statuses/2.0.1:
     resolution: {integrity: sha512-RwNA9Z/7PrK06rYLIzFMlaF+l73iwpzsqRIFgbMLbTcLD6cOao82TaWefPXQvB2fOC4AjuYSEndS7N/mTCbkdQ==}
     engines: {node: '>= 0.8'}
diff --git a/routes/info.js b/routes/info.js
index eb42ca35..00a4915e 100644
--- a/routes/info.js
+++ b/routes/info.js
@@ -1,90 +1,32 @@
-import app from "$app";
-import db from "$db";
-import config from "$config";
-import store from "$lib/store";
-
-import bc from "$lib/bitcoin";
-import lq from "$lib/liquid";
+import db from "$lib/db";
 import ln from "$lib/ln";
 
-import sequelize from "@sequelize/core";
-import { SATS } from "$lib/utils";
-
-const { Op } = sequelize;
-
-/**
- * @api {get} /info Request server information
- * @apiName GetInfo
- * @apiGroup Info
- *
- * @apiSuccess {Object} fx Exchange rates relative to USD - fx.CUR is the value of one CUR in USD.
- * @apiSuccess {String[]} networks Array of supported networks; possible values are "bitcoin", "liquid" and "lightning".
- * @apiSuccess {String} clientVersion the current git commit of the ui
- */
-app.get("/info", async (req, res, next) => {
-  const { clientVersion } = config;
-
-  const info = {
-    fx: store.fx,
-    nodes: store.networks,
-    clientVersion
-  };
-
-  res.send(info);
-});
+export default {
+  async balances(req, res) {
+    let lnchannel;
+    let lnwallet;
 
-app.get("/balances", async (req, res, next) => {
-  try {
-    const accounts = await db.Account.findAll({
-      where: {
-        pubkey: null
-      },
-      attributes: [
-        "asset",
-        [sequelize.fn("sum", sequelize.col("balance")), "total"]
-      ],
-      group: ["asset"]
+    const stream = db.scanStream({
+      match: "balance:*",
+      count: 100
     });
 
-    let lnchannel;
-    let lnwallet;
+    let b = 0;
+    for await (let keys of stream) {
+      for (let k of keys) b += parseInt(await db.get(k));
+    }
+    console.log(b)
 
     const funds = await ln.listfunds();
     lnchannel = parseInt(funds.channels.reduce((a, b) => a + b.channel_sat, 0));
     lnwallet = parseInt(funds.outputs.reduce((a, b) => a + b.value, 0));
 
-    let bitcoin, bitcoind;
-    if (config.bitcoin) {
-      bitcoin = parseInt((await bc.getBalance()) * SATS);
-      bitcoind = await bc.getNetworkInfo();
-    }
-
-    let assets, liquid, elementsd;
-    if (config.liquid) {
-      assets = await lq.getBalance();
-      liquid = parseInt(assets.bitcoin * SATS);
-      elementsd = await lq.getNetworkInfo();
-    }
-
     const info = {
-      bitcoind,
-      elementsd,
-      accounts,
-      assets,
-      bitcoin,
-      liquid,
       lnchannel,
-      lnwallet
+      lnwallet,
+      total: lnchannel + lnwallet
     };
 
-    info.total =
-      parseInt(bitcoin) +
-      parseInt(liquid) +
-      parseInt(lnchannel) +
-      parseInt(lnwallet);
-
     res.send(info);
-  } catch (e) {
-    console.log("problem getting balances", e);
   }
-});
+};
diff --git a/routes/invoices.js b/routes/invoices.js
index acd9457e..3e43c57c 100644
--- a/routes/invoices.js
+++ b/routes/invoices.js
@@ -58,7 +58,7 @@ export default {
     );
 
     await s(`invoice:${hash}`, invoice);
-    await db.lPush(`${uid}:invoices`, hash);
+    await db.lpush(`${uid}:invoices`, hash);
 
     if (request_id) {
       let request = await g(`request:${request_id}`);
diff --git a/routes/nostr.js b/routes/nostr.js
index 005a4be9..bac04810 100644
--- a/routes/nostr.js
+++ b/routes/nostr.js
@@ -41,10 +41,10 @@ export default {
 
     await s(`user:${pubkey}`, user);
 
-    let ids = await db.sMembers(pubkey);
+    let ids = await db.smembers(pubkey);
 
     let events = ids.length
-      ? (await db.mGet(ids.map(k => "ev:" + k))).map(JSON.parse)
+      ? (await db.mget(ids.map(k => "ev:" + k))).map(JSON.parse)
       : [];
 
     res.send(events.map(e => ({ ...e, user })));
@@ -101,7 +101,7 @@ export default {
   async followers({ params: { pubkey } }, res) {
     try {
       let pubkeys = [
-        ...new Set([...(await db.sMembers(`${pubkey}:followers`))])
+        ...new Set([...(await db.smembers(`${pubkey}:followers`))])
       ];
 
       let followers = [];
diff --git a/routes/payments.js b/routes/payments.js
index 13c0cdb7..22de79c3 100644
--- a/routes/payments.js
+++ b/routes/payments.js
@@ -21,7 +21,7 @@ export default {
     // if (start) where.createdAt[Op.gte] = new Date(parseInt(start));
     // if (end) where.createdAt[Op.lte] = new Date(parseInt(end));
 
-    let payments = (await db.lRange(`${id}:payments`, 0, -1)) || [];
+    let payments = (await db.lrange(`${id}:payments`, 0, -1)) || [];
     payments = await Promise.all(payments.map(hash => g(`payment:${hash}`)));
     payments = payments.filter(p => p);
     res.send({ payments, total: payments.length });
diff --git a/routes/requests.js b/routes/requests.js
index 79af1fb6..8db373dd 100644
--- a/routes/requests.js
+++ b/routes/requests.js
@@ -15,8 +15,8 @@ export default {
     try {
       const day = new Date(new Date().setDate(new Date().getDate() - 1));
 
-      let invoices = await db.lRange(`${id}:invoices`, 0, -1);
-      let requests = await db.lRange(`${id}:requests`, 0, -1);
+      let invoices = await db.lrange(`${id}:invoices`, 0, -1);
+      let requests = await db.lrange(`${id}:requests`, 0, -1);
 
       res.send({ invoices, requests });
     } catch (e) {