Remove subs.ts & subscription.ts, refactor around it

src/controllers/api/streaming.ts

@@ -5,11 +5,11 @@ import { type AppController } from '@/app.ts';
 import { Conf } from '@/config.ts';
 import { Debug } from '@/deps.ts';
 import { getFeedPubkeys } from '@/queries.ts';
-import { Sub } from '@/subs.ts';
 import { bech32ToPubkey } from '@/utils.ts';
 import { renderReblog, renderStatus } from '@/views/mastodon/statuses.ts';
 import { hydrateEvents } from '@/storages/hydrate.ts';
 import { eventsDB } from '@/storages.ts';
+import { Storages } from '@/storages.ts';
 
 const debug = Debug('ditto:streaming');
 
@@ -38,6 +38,7 @@ const streamingController: AppController = (c) => {
   const upgrade = c.req.header('upgrade');
   const token = c.req.header('sec-websocket-protocol');
   const stream = streamSchema.optional().catch(undefined).parse(c.req.query('stream'));
+  const controller = new AbortController();
 
   if (upgrade?.toLowerCase() !== 'websocket') {
     return c.text('Please use websocket protocol', 400);
@@ -63,33 +64,37 @@ const streamingController: AppController = (c) => {
 
   socket.onopen = async () => {
     if (!stream) return;
+
     const filter = await topicToFilter(stream, c.req.query(), pubkey);
+    if (!filter) return;
 
-    if (filter) {
-      for await (const event of Sub.sub(socket, '1', [filter])) {
-        if (event.kind === 6) {
-          await hydrateEvents({
-            events: [event],
-            storage: eventsDB,
-            signal: AbortSignal.timeout(1000),
-          });
+    for await (const msg of Storages.pubsub.req([filter], { signal: controller.signal })) {
+      if (msg[0] === 'EVENT') {
+        const [event] = await hydrateEvents({
+          events: [msg[2]],
+          storage: eventsDB,
+          signal: AbortSignal.timeout(1000),
+        });
 
-          const status = await renderReblog(event, { viewerPubkey: c.get('pubkey') });
+        if (event.kind === 1) {
+          const status = await renderStatus(event, { viewerPubkey: pubkey });
           if (status) {
             send('update', status);
           }
-          continue;
         }
-        const status = await renderStatus(event, { viewerPubkey: pubkey });
-        if (status) {
-          send('update', status);
+
+        if (event.kind === 6) {
+          const status = await renderReblog(event, { viewerPubkey: pubkey });
+          if (status) {
+            send('update', status);
+          }
         }
       }
     }
   };
 
   socket.onclose = () => {
-    Sub.close(socket);
+    controller.abort();
   };
 
   return response;

src/controllers/nostr/relay.ts

@@ -11,8 +11,7 @@ import {
   clientMsgSchema,
   type ClientREQ,
 } from '@/schemas/nostr.ts';
-import { purifyEvent } from '@/storages/hydrate.ts';
-import { Sub } from '@/subs.ts';
+import { Storages } from '@/storages.ts';
 
 import type { AppController } from '@/app.ts';
 
@@ -29,6 +28,8 @@ type RelayMsg =
 
 /** Set up the Websocket connection. */
 function connectStream(socket: WebSocket) {
+  const controllers = new Map<string, AbortController>();
+
   socket.onmessage = (e) => {
     const result = jsonSchema.pipe(clientMsgSchema).safeParse(e.data);
     if (result.success) {
@@ -39,7 +40,9 @@ function connectStream(socket: WebSocket) {
   };
 
   socket.onclose = () => {
-    Sub.close(socket);
+    for (const controller of controllers.values()) {
+      controller.abort();
+    }
   };
 
   /** Handle client message. */
@@ -64,14 +67,20 @@ function connectStream(socket: WebSocket) {
   async function handleReq([_, subId, ...rest]: ClientREQ): Promise<void> {
     const filters = prepareFilters(rest);
 
+    const controller = new AbortController();
+    controllers.get(subId)?.abort();
+    controllers.set(subId, controller);
+
     for (const event of await eventsDB.query(filters, { limit: FILTER_LIMIT })) {
       send(['EVENT', subId, event]);
     }
 
     send(['EOSE', subId]);
 
-    for await (const event of Sub.sub(socket, subId, filters)) {
-      send(['EVENT', subId, purifyEvent(event)]);
+    for await (const msg of Storages.pubsub.req(filters, { signal: controller.signal })) {
+      if (msg[0] === 'EVENT') {
+        send(['EVENT', subId, msg[2]]);
+      }
     }
   }
 
@@ -93,7 +102,11 @@ function connectStream(socket: WebSocket) {
 
   /** Handle CLOSE. Close the subscription. */
   function handleClose([_, subId]: ClientCLOSE): void {
-    Sub.unsub(socket, subId);
+    const controller = controllers.get(subId);
+    if (controller) {
+      controller.abort();
+      controllers.delete(subId);
+    }
   }
 
   /** Handle COUNT. Return the number of events matching the filters. */