redlock

4 files changed, 65 insertions, 36 deletions

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);