follower query caching

4 files changed, 97 insertions, 109 deletions

diff --git a/lib/nostr.js b/lib/nostr.js
index 034a4aaa..dc400b7f 100644
--- a/lib/nostr.js
+++ b/lib/nostr.js
@@ -14,15 +14,58 @@ pool.on("open", relay => {
 });
 
 export let coinos;
+let now = () => Math.round(Date.now() / 1000);
+
+let timeouts = {};
+export let q = async (sub, query, timeout = 8000) =>
+  new Promise(async (r, j) => {
+    let rejected;
+    if (!coinos) return j(new Error("relay not initialized"));
+    query.since = await redis.get(`since:${sub}`);
+    if (now() - query.since < 3600) return r();
+
+    coinos.on("eose", s => !rejected && s === sub && r(timeouts[sub].clear()));
+
+    coinos.subscribe(sub, query);
+
+    timeouts[sub] = {
+      async clear() {
+        clearTimeout(this.timer);
+        delete timeouts[sub];
+        coinos.unsubscribe(sub);
+        await redis.set(`since:${sub}`, now());
+      },
+      extend(t = timeout) {
+        clearTimeout(this.timer);
+        this.timer = setTimeout(() => {
+          j(new Error(`query timed out: ${sub} ${Date.now()}`));
+          coinos.unsubscribe(sub);
+          rejected = true;
+        }, t);
+      },
+      timer: undefined
+    };
+
+    timeouts[sub].extend(timeout);
+  });
+
 let seen = [];
-let timeout;
+
+let getUser = async pubkey =>
+  JSON.parse(await redis.get(`user:${pubkey}`)) || {
+    username: pubkey.substr(0, 6),
+    pubkey,
+    anon: true
+  };
+
 pool.on("event", async (relay, sub, ev) => {
+  if (timeouts[sub]) timeouts[sub].extend();
   try {
-    let parsed;
+    let content;
     try {
-      parsed = JSON.parse(ev.content);
+      content = JSON.parse(ev.content);
     } catch (e) {
-      parsed = ev.content;
+      content = ev.content;
     }
 
     if (sub === "live") {
@@ -38,48 +81,25 @@ pool.on("event", async (relay, sub, ev) => {
       if (coinos && ev.kind < 5) {
         coinos.send(["EVENT", ev]);
 
-        let newuser = {
-          username: pubkey.substr(0, 6),
-          pubkey,
-          anon: true,
-          follows: [],
-          followers: []
-        };
-
-        ev.user = JSON.parse(await redis.get(`user:${pubkey}`)) || newuser;
-
-        relay.subscribe("zero", {
-          limit: 1,
-          kinds: [0],
-          authors: [pubkey]
-        });
+        ev.user = await getUser(pubkey);
 
         broadcast("event", ev);
       }
-    } else if (sub === "zero") {
+    } else if (sub.includes("profile")) {
       let { pubkey } = ev;
-      let newuser = {
-        username: pubkey.substr(0, 6),
-        pubkey,
-        anon: true,
-        follows: [],
-        followers: []
-      };
+      let user = await getUser(pubkey);
 
-      let user = JSON.parse(await redis.get(`user:${pubkey}`)) || newuser;
+      if (user.updated > ev.created_at) return;
 
-      if (user.profileUpdated > ev.created_at) return;
+      if (content.name && user.username === pubkey.substr(0, 6))
+        user.username = content.name;
 
-      if (parsed.name && user.username === pubkey.substr(0, 6))
-        user.username = parsed.name;
-
-      delete parsed.name;
+      delete content.name;
 
       user = {
         ...user,
-        ...parsed,
-        initialized: true,
-        profileUpdated: ev.created_at
+        ...content,
+        updated: ev.created_at
       };
 
       await redis.set(`user:${pubkey}`, JSON.stringify(user));
@@ -89,39 +109,29 @@ pool.on("event", async (relay, sub, ev) => {
       await redis.set(`${pubkey}:follows`, JSON.stringify(ev.tags));
       for (let f of ev.tags) {
         let [_, followPubkey] = f;
-        pool.subscribe("zero", {
-          limit: 1,
-          kinds: [0],
-          authors: [followPubkey]
-        });
         await redis.sAdd(`${followPubkey}:followers`, pubkey);
       }
-      store.fetching[pubkey] = false;
     } else if (sub.includes("followers")) {
       let followed = sub.split(":")[0];
       let { pubkey } = ev;
       await redis.sAdd(`${followed}:followers`, pubkey);
-      store.fetching[pubkey] = false;
     } else {
       if (!relay.url.includes("coinos")) return;
 
       let pubkey = sub;
-      clearTimeout(store.timeouts[pubkey]);
+      ev.user = await getUser(pubkey);
 
-      ev.user = JSON.parse(await redis.get(`user:${pubkey}`)) || {
-        username: pubkey.substr(0, 6),
-        pubkey,
-        anon: true
-      };
+      //     console.log(ev.user.username, "UPDATED AT", ev.user.updated)
+      if (!ev.user.updated) {
+        relay.subscribe(`${pubkey}:${ev.id}:profile`, {
+          limit: 1,
+          kinds: [0],
+          authors: [pubkey]
+        });
+      }
 
       await redis.sAdd(pubkey, ev.id);
       await redis.set(`ev:${ev.id}`, JSON.stringify(ev));
-
-      store.fetching[pubkey] = true;
-      store.timeouts[pubkey] = setTimeout(
-        () => (store.fetching[pubkey] = false),
-        100
-      );
     }
   } catch (e) {
     console.log(e);
diff --git a/lib/utils.js b/lib/utils.js
index 08a582eb..0d2d3efb 100644
--- a/lib/utils.js
+++ b/lib/utils.js
@@ -13,7 +13,7 @@ import { coinos } from "$lib/nostr";
 const { Op } = sequelize;
 
 export const sleep = n => new Promise(r => setTimeout(r, n));
-export const wait = async (f, s = 300, n = 1000) => {
+export const wait = async (f, s = 300, n = 50) => {
   let i = 0;
   while (!(await f()) && i < s) (await sleep(n)) && i++;
   if (i >= s) throw new Error("timeout");
diff --git a/routes/events.js b/routes/events.js
index 074579b0..fd2d97cd 100644
--- a/routes/events.js
+++ b/routes/events.js
@@ -1,7 +1,7 @@
 import app from "$app";
 import db from "$db";
 import redis from "$lib/redis";
-import { coinos } from "$lib/nostr";
+import { coinos, q } from "$lib/nostr";
 import store from "$lib/store";
 import { wait } from "$lib/utils";
 
@@ -9,12 +9,6 @@ app.get("/nostr/:pubkey", async (req, res) => {
   try {
     let { pubkey } = req.params;
 
-    store.fetching[pubkey] = true;
-    store.timeouts[pubkey] = setTimeout(
-      () => (store.fetching[pubkey] = false),
-      500
-    );
-
     let user = JSON.parse(await redis.get(`user:${pubkey}`));
 
     if (!user) {
@@ -32,24 +26,13 @@ app.get("/nostr/:pubkey", async (req, res) => {
         followers: []
       };
 
-    let { since } = user;
-
-    coinos.subscribe(pubkey, {
-      since,
+    await q(pubkey, {
       kinds: [1],
       authors: [pubkey]
     });
 
-    coinos.subscribe(`${pubkey}:followers`, {
-      since,
-      kinds: [3],
-      "#p": [pubkey]
-    });
-
-    user.since = Math.round(Date.now() / 1000);
     await redis.set(`user:${pubkey}`, JSON.stringify(user));
 
-    await wait(() => !store.fetching[pubkey], 100, 100);
     let ids = await redis.sMembers(pubkey);
 
     let events = ids.length
diff --git a/routes/users.js b/routes/users.js
index 19060f9f..e40f1b7e 100644
--- a/routes/users.js
+++ b/routes/users.js
@@ -22,7 +22,7 @@ import lq from "$lib/liquid";
 import { emit } from "$lib/sockets";
 import register from "$lib/register";
 import { requirePin } from "$lib/utils";
-import { coinos } from "$lib/nostr";
+import { coinos, q } from "$lib/nostr";
 
 const pick = (O, ...K) => K.reduce((o, k) => ((o[k] = O[k]), o), {});
 
@@ -630,45 +630,40 @@ app.get("/contacts", auth, async function(req, res) {
 
 app.post("/checkPassword", auth, async function(req, res) {
   try {
-  let { user } = req;
-  let result = await bcrypt.compare(req.body.password, user.password);
-  if (!result) throw new Error();
-  res.send(result);
-  } catch(e) {
+    let { user } = req;
+    let result = await bcrypt.compare(req.body.password, user.password);
+    if (!result) throw new Error();
+    res.send(result);
+  } catch (e) {
     res.code(500).send("Invalid password");
-  } 
+  }
 });
 
-let since = {};
 app.get("/:pubkey/follows", async (req, res) => {
   try {
     let { pubkey } = req.params;
     let { tagsonly } = req.query;
 
-    store.fetching[pubkey] = true;
-    store.timeouts[pubkey] = setTimeout(
-      () => (store.fetching[pubkey] = false),
-      500
-    );
-
     let sub = `${pubkey}:follows`;
-    coinos.subscribe(sub, {
-      since: since[sub],
+    await q(sub, {
       limit: 1,
       kinds: [3],
       authors: [pubkey]
     });
 
-    since[sub] = Math.round(Date.now() / 1000);
-
-    await wait(() => !store.fetching[pubkey], 100, 10);
-
     let tags = JSON.parse(await redis.get(`${pubkey}:follows`)) || [];
     if (tagsonly) return res.send(tags);
 
     let follows = [];
     for (let f of tags) {
       let [_, pubkey] = f;
+
+      await q(`${pubkey}:profile:f1`, {
+        limit: 1,
+        kinds: [0],
+        authors: [pubkey]
+      });
+
       let user = JSON.parse(await redis.get(`user:${pubkey}`));
       if (user) follows.push(user);
     }
@@ -679,7 +674,7 @@ app.get("/:pubkey/follows", async (req, res) => {
     res.send(follows);
   } catch (e) {
     console.log(e);
-    res.code(500).send(e.message);
+    res.code(500).send(e && e.message);
   }
 });
 
@@ -687,28 +682,28 @@ app.get("/:pubkey/followers", async (req, res) => {
   try {
     let { pubkey } = req.params;
 
-    store.fetching[pubkey] = true;
-    store.timeouts[pubkey] = setTimeout(
-      () => (store.fetching[pubkey] = false),
-      500
-    );
-
     let sub = `${pubkey}:followers`;
-    coinos.subscribe(sub, {
-      since: since[sub],
+    await q(sub, {
       kinds: [3],
       "#p": [pubkey]
     });
 
-    since[sub] = Math.round(Date.now() / 1000);
-
-    await wait(() => !store.fetching[pubkey], 100, 10);
-
     let pubkeys = await redis.sMembers(`${pubkey}:followers`);
 
     let followers = [];
+
     for (let pubkey of pubkeys) {
       let user = JSON.parse(await redis.get(`user:${pubkey}`));
+      if (!user || !user.updated) {
+        await q(`${pubkey}:profile:f2`, {
+          limit: 1,
+          kinds: [0],
+          authors: [pubkey]
+        });
+
+        user = JSON.parse(await redis.get(`user:${pubkey}`));
+      }
+
       if (user) followers.push(user);
     }
 
@@ -718,6 +713,6 @@ app.get("/:pubkey/followers", async (req, res) => {
     res.send(followers);
   } catch (e) {
     console.log(e);
-    res.code(500).send(e.message);
+    res.code(500).send(e && e.message);
   }
 });