useSubscribe, handle emitted requests in sqlite

packages/app/src/Components/Event/Create/util.ts

@@ -8,7 +8,7 @@ export async function sendEventToRelays(
   setResults?: (x: Array<OkResponse>) => void,
 ) {
   if (customRelays) {
-    system.HandleEvent({ ...ev, relays: [] });
+    system.HandleEvent("*", { ...ev, relays: [] });
     return removeUndefined(
       await Promise.all(
         customRelays.map(async r => {

packages/app/src/Components/Feed/TimelineFollows.tsx

@@ -36,10 +36,10 @@ const TimelineFollows = (props: TimelineFollowsProps) => {
   const [latest, setLatest] = useHistoryState(unixNow(), "TimelineFollowsLatest");
   const [limit, setLimit] = useState(50);
   const feed = useFollowsTimelineView(limit);
+  console.log("feed", feed);
   const { muted, isEventMuted } = useModeration();
 
-  const sortedFeed = useMemo(() => orderDescending(feed), [feed]);
-  const oldest = useMemo(() => sortedFeed.at(-1)?.created_at, [sortedFeed]);
+  const oldest = useMemo(() => feed.at(-1)?.created_at, [feed]);
 
   const postsOnly = useCallback(
     (a: NostrEvent) => (props.postsOnly ? !a.tags.some(b => b[0] === "e" || b[0] === "a") : true),
@@ -58,8 +58,8 @@ const TimelineFollows = (props: TimelineFollowsProps) => {
 
   const mixin = useHashtagsFeed();
   const mainFeed = useMemo(() => {
-    return filterPosts((sortedFeed ?? []).filter(a => a.created_at <= latest));
-  }, [sortedFeed, filterPosts, latest, login.follows.timestamp]);
+    return filterPosts((feed ?? []).filter(a => a.created_at <= latest));
+  }, [feed, filterPosts, latest, login.follows.timestamp]);
 
   const findHashTagContext = (a: NostrEvent) => {
     const tag = a.tags.filter(a => a[0] === "t").find(a => login.tags.item.includes(a[1].toLowerCase()))?.[1];
@@ -81,12 +81,12 @@ const TimelineFollows = (props: TimelineFollowsProps) => {
   }, [mixin, mainFeed, postsOnly, isEventMuted]);
 
   const latestFeed = useMemo(() => {
-    return filterPosts((sortedFeed ?? []).filter(a => a.created_at > latest));
-  }, [sortedFeed, latest]);
+    return filterPosts((feed ?? []).filter(a => a.created_at > latest));
+  }, [feed, latest]);
 
   const liveStreams = useMemo(() => {
-    return (sortedFeed ?? []).filter(a => a.kind === EventKind.LiveEvent && findTag(a, "status") === "live");
-  }, [sortedFeed]);
+    return (feed ?? []).filter(a => a.kind === EventKind.LiveEvent && findTag(a, "status") === "live");
+  }, [feed]);
 
   const latestAuthors = useMemo(() => {
     return dedupeByPubkey(latestFeed).map(e => e.pubkey);
@@ -120,7 +120,7 @@ const TimelineFollows = (props: TimelineFollowsProps) => {
         }}
         displayAs={displayAs}
       />
-      {sortedFeed.length > 0 && (
+      {feed.length > 0 && (
         <ShowMoreInView
           onClick={() => {
             setLimit(s => s + 20);

packages/app/src/Components/Trending/TrendingPosts.tsx

@@ -35,7 +35,7 @@ export default function TrendingNotes({ count = Infinity, small = false }: { cou
           console.error(`Event with invalid sig\n\n${ev}\n\nfrom ${trendingNotesUrl}`);
           return;
         }
-        System.HandleEvent(ev as TaggedNostrEvent);
+        System.HandleEvent("*", ev as TaggedNostrEvent);
         return ev;
       }),
     );

packages/app/src/Feed/WorkerRelayView.ts

@@ -4,6 +4,7 @@ import { useRequestBuilder } from "@snort/system-react";
 import { useEffect, useMemo, useState } from "react";
 
 import useLogin from "@/Hooks/useLogin";
+import useSubscribe from "@/Hooks/useSubscribe";
 import { Relay } from "@/system";
 import { Day } from "@/Utils/Const";
 
@@ -77,16 +78,15 @@ export function useFollowsTimelineView(limit = 20) {
   const follows = useLogin(s => s.follows.item);
   const kinds = [EventKind.TextNote, EventKind.Repost, EventKind.Polls];
 
-  const filter = useMemo(() => {
-    return [
-      {
-        authors: follows,
-        kinds,
-        limit,
-      },
-    ];
-  }, [follows, limit]);
-  return useWorkerRelayView("follows-timeline", filter, Day * 7);
+  const filter = useMemo(
+    () => ({
+      authors: follows,
+      kinds,
+      limit,
+    }),
+    [follows, limit],
+  );
+  return useSubscribe("follows-timeline", filter);
 }
 
 export function useNotificationsView() {
@@ -101,7 +101,7 @@ export function useNotificationsView() {
       },
     ];
   }, [publicKey]);
-  return useWorkerRelayView("notifications", req, Day * 30);
+  return useSubscribe("notifications", req[0]);
 }
 
 export function useReactionsView(ids: Array<NostrLink>, leaveOpen = true) {
@@ -123,7 +123,7 @@ export function useReactionsView(ids: Array<NostrLink>, leaveOpen = true) {
     return rb.buildRaw();
   }, [ids]);
 
-  return useWorkerRelayView("reactions", req, undefined);
+  return useSubscribe("reactions", req[0]);
 }
 
 export function useReactionsViewCount(ids: Array<NostrLink>, leaveOpen = true) {
@@ -160,5 +160,5 @@ export function useFollowsContactListView() {
       },
     ];
   }, [follows]);
-  return useWorkerRelayView("follows-contacts-relays", filter, undefined);
+  return useSubscribe("follows-contacts-relays", filter[0]);
 }

packages/app/src/Hooks/useSubscribe.ts

@@ -0,0 +1,31 @@
+import { ReqFilter, RequestBuilder, TaggedNostrEvent } from "@snort/system";
+import inMemoryDB from "@snort/system/src/InMemoryDB";
+import { useRequestBuilder } from "@snort/system-react";
+import { useEffect, useMemo, useState } from "react";
+
+import { System } from "@/system";
+
+export default function useSubscribe(id: string, filter: ReqFilter): TaggedNostrEvent[] {
+  const getEvents = () => inMemoryDB.findArray(filter);
+  const [events, setEvents] = useState(getEvents());
+  const rb = useMemo(() => {
+    const rb = new RequestBuilder(id);
+    rb.withBareFilter(filter);
+    return rb;
+  }, [id, filter]);
+  useRequestBuilder(rb);
+
+  useEffect(() => {
+    const cb = (subId: string) => {
+      if (subId === id) {
+        setEvents(getEvents());
+      }
+    };
+    System.on("event", cb);
+    return () => {
+      System.off("event", cb);
+    };
+  }, [id, filter]);
+
+  return events as Array<TaggedNostrEvent>;
+}

packages/app/src/chat/nip29.ts

@@ -91,7 +91,7 @@ export class Nip29ChatSystem extends ExternalStore<Array<Chat>> implements ChatS
         },
         sendMessage: async (ev, system: SystemInterface) => {
           ev.forEach(async a => {
-            system.HandleEvent({ ...a, relays: [] });
+            system.HandleEvent("*", { ...a, relays: [] });
             await system.WriteOnceToRelay(`wss://${relay}`, a);
           });
         },

packages/app/src/system.ts

@@ -1,4 +1,5 @@
 import { removeUndefined, throwIfOffline } from "@snort/shared";
+import LRUSet from "@snort/shared/src/LRUSet";
 import { mapEventToProfile, NostrEvent, NostrSystem, ProfileLoaderService, socialGraphInstance } from "@snort/system";
 import { WorkerRelayInterface } from "@snort/worker-relay";
 import WorkerRelayPath from "@snort/worker-relay/dist/worker?worker&url";
@@ -70,9 +71,21 @@ export async function initRelayWorker() {
     if (await Relay.init()) {
       if (await Relay.open()) {
         await Relay.migrate();
+        const seen = new LRUSet<string>(100);
         System.on("event", async (_, ev) => {
+          if (seen.has(ev.id)) return;
+          seen.add(ev.id);
           await Relay.event(ev);
         });
+        System.on("request", async (subId, f) => {
+          const evs = await Relay.req(["REQ", "", ...f.filters]);
+          evs.forEach(ev => {
+            seen.add(ev.id);
+            queueMicrotask(() => {
+              System.HandleEvent(subId, { ...ev, relays: [] });
+            });
+          });
+        });
       }
     }
   } catch (e) {

packages/system/src/InMemoryDB.ts

@@ -131,6 +131,10 @@ class InMemoryDB {
     }
   }
 
+  count(filter: Filter): number {
+    return this.findArray(filter).length;
+  }
+
   find(filter: Filter, callback: (event: TaggedNostrEvent) => void): void {
     this.findArray(filter).forEach(event => {
       callback(event);

packages/system/src/connection.ts

@@ -10,6 +10,8 @@ import { NostrEvent, ReqCommand, ReqFilter, TaggedNostrEvent, u256 } from "./nos
 import { RelayInfo } from "./relay-info";
 import EventKind from "./event-kind";
 import { EventExt } from "./event-ext";
+import { getHex64 } from "./utils";
+import inMemoryDB from "./InMemoryDB";
 
 /**
  * Relay settings
@@ -210,6 +212,12 @@ export class Connection extends EventEmitter<ConnectionEvents> {
       }
        */
 
+      const id = getHex64(e.data as string, "id");
+      if (inMemoryDB.has(id)) {
+        this.#log("Already have, skip processing %s", id);
+        return;
+      }
+
       const msg = JSON.parse(e.data as string) as Array<string | NostrEvent | boolean>;
       const tag = msg[0] as string;
       switch (tag) {
@@ -232,7 +240,7 @@ export class Connection extends EventEmitter<ConnectionEvents> {
           } as TaggedNostrEvent;
 
           if (!EventExt.isValid(ev)) {
-            //this.#log("Rejecting invalid event %O", ev);
+            this.#log("Rejecting invalid event %O", ev);
             return;
           }
           this.emit("event", msg[1] as string, ev);

packages/system/src/index.ts

@@ -109,7 +109,7 @@ export interface SystemInterface {
   /**
    * Push an event into the system from external source
    */
-  HandleEvent(ev: TaggedNostrEvent): void;
+  HandleEvent(subId: string, ev: TaggedNostrEvent): void;
 
   /**
    * Send an event to all permanent connections

packages/system/src/nostr-system.ts

@@ -24,12 +24,13 @@ import { RelayMetadataLoader } from "./outbox-model";
 import { Optimizer, DefaultOptimizer } from "./query-optimizer";
 import { ConnectionPool, DefaultConnectionPool } from "./connection-pool";
 import { QueryManager } from "./query-manager";
+import inMemoryDB from "./InMemoryDB";
 
 export interface NostrSystemEvents {
   change: (state: SystemSnapshot) => void;
   auth: (challenge: string, relay: string, cb: (ev: NostrEvent) => void) => void;
   event: (subId: string, ev: TaggedNostrEvent) => void;
-  filters: (filter: BuiltRawReqFilter) => void;
+  request: (subId: string, filter: BuiltRawReqFilter) => void;
 }
 
 export interface NostrsystemProps {
@@ -122,6 +123,7 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
     this.pool.on("event", (_, sub, ev) => {
       ev.relays?.length && this.relayMetricsHandler.onEvent(ev.relays[0]);
       this.emit("event", sub, ev);
+      inMemoryDB.handleEvent(ev);
     });
     this.pool.on("disconnect", (id, code) => {
       const c = this.pool.getConnection(id);
@@ -148,7 +150,7 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
     this.#queryManager.on("trace", t => {
       this.relayMetricsHandler.onTraceReport(t);
     });
-    this.#queryManager.on("filters", (f: BuiltRawReqFilter) => this.emit("filters", f));
+    this.#queryManager.on("request", (subId: string, f: BuiltRawReqFilter) => this.emit("request", subId, f));
   }
 
   get Sockets(): ConnectionStateSnapshot[] {
@@ -189,13 +191,14 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
     return this.#queryManager.query(req);
   }
 
-  HandleEvent(ev: TaggedNostrEvent) {
-    this.emit("event", "*", ev);
+  HandleEvent(subId: string, ev: TaggedNostrEvent) {
+    inMemoryDB.handleEvent(ev);
+    this.emit("event", subId, ev);
     this.#queryManager.handleEvent(ev);
   }
 
   async BroadcastEvent(ev: NostrEvent, cb?: (rsp: OkResponse) => void): Promise<OkResponse[]> {
-    this.HandleEvent({ ...ev, relays: [] });
+    this.HandleEvent("*", { ...ev, relays: [] });
     return await this.pool.broadcast(this, ev, cb);
   }
 

packages/system/src/query-manager.ts

@@ -8,7 +8,7 @@ import { trimFilters } from "./request-trim";
 interface QueryManagerEvents {
   change: () => void;
   trace: (report: TraceReport) => void;
-  filters: (req: BuiltRawReqFilter) => void;
+  request: (subId: string, req: BuiltRawReqFilter) => void;
 }
 
 /**
@@ -57,9 +57,9 @@ export class QueryManager extends EventEmitter<QueryManagerEvents> {
     } else {
       const q = new Query(this.#system, req);
       q.on("trace", r => this.emit("trace", r));
-      q.on("filters", fx => {
+      q.on("request", (id, fx) => {
         this.#send(q, fx);
-        this.emit("filters", fx);
+        this.emit("request", id, fx);
       });
 
       this.#queries.set(req.id, q);
@@ -78,7 +78,7 @@ export class QueryManager extends EventEmitter<QueryManagerEvents> {
   async fetch(req: RequestBuilder, cb?: (evs: Array<TaggedNostrEvent>) => void) {
     const q = new Query(this.#system, req);
     q.on("trace", r => this.emit("trace", r));
-    q.on("filters", fx => {
+    q.on("request", (subId, fx) => {
       this.#send(q, fx);
     });
     if (cb) {

packages/system/src/query.ts

@@ -100,7 +100,7 @@ export interface TraceReport {
 export interface QueryEvents {
   loading: (v: boolean) => void;
   trace: (report: TraceReport) => void;
-  filters: (req: BuiltRawReqFilter) => void;
+  request: (subId: string, req: BuiltRawReqFilter) => void;
   event: (evs: Array<TaggedNostrEvent>) => void;
   end: () => void;
 }
@@ -329,11 +329,11 @@ export class Query extends EventEmitter<QueryEvents> {
     if (!(this.request.options?.skipDiff ?? false) && existing.length > 0) {
       const filters = this.request.buildDiff(this.#system, existing);
       this.#log("Build %s %O", this.id, filters);
-      filters.forEach(f => this.emit("filters", f));
+      filters.forEach(f => this.emit("request", this.id, f));
     } else {
       const filters = this.request.build(this.#system);
       this.#log("Build %s %O", this.id, filters);
-      filters.forEach(f => this.emit("filters", f));
+      filters.forEach(f => this.emit("request", this.id, f));
     }
   }
 

packages/system/src/worker/system-worker.ts

@@ -124,7 +124,7 @@ export class SystemWorker extends EventEmitter<NostrSystemEvents> implements Sys
     this.#workerRpc(WorkerCommand.DisconnectRelay, address);
   }
 
-  HandleEvent(ev: TaggedNostrEvent): void {
+  HandleEvent(subId: string, ev: TaggedNostrEvent): void {
     throw new Error("Method not implemented.");
   }