follower query caching

1 files changed, 65 insertions, 55 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);