From f7cd67c57242e3cbd3e9c0d7234fb5c01a60b902 Mon Sep 17 00:00:00 2001
From: Alex Gleason <alex@alexgleason.me>
Date: Fri, 25 Aug 2023 14:59:37 -0500
Subject: [PATCH] Make MastoAPI streaming work for public feeds

---
 src/controllers/api/streaming.ts | 39 +++++++++++++++++++++++++++++---
 src/subs.ts                      |  4 ++--
 2 files changed, 38 insertions(+), 5 deletions(-)

diff --git a/src/controllers/api/streaming.ts b/src/controllers/api/streaming.ts
index 412e024..76a23c1 100644
--- a/src/controllers/api/streaming.ts
+++ b/src/controllers/api/streaming.ts
@@ -1,6 +1,10 @@
 import { AppController } from '@/app.ts';
+import { type Event } from '@/deps.ts';
+import { type DittoFilter } from '@/filter.ts';
 import { TOKEN_REGEX } from '@/middleware/auth19.ts';
 import { streamSchema, ws } from '@/stream.ts';
+import { Sub } from '@/subs.ts';
+import { toStatus } from '@/transformers/nostr-to-mastoapi.ts';
 import { bech32ToPubkey } from '@/utils.ts';
 
 const streamingController: AppController = (c) => {
@@ -29,10 +33,30 @@ const streamingController: AppController = (c) => {
     pubkey: bech32ToPubkey(match[1]),
   };
 
-  socket.addEventListener('open', () => {
+  function send(name: string, payload: object) {
+    if (socket.readyState === WebSocket.OPEN) {
+      socket.send(JSON.stringify({
+        event: name,
+        payload: JSON.stringify(payload),
+      }));
+    }
+  }
+
+  socket.addEventListener('open', async () => {
     console.log('websocket: connection opened');
-    if (stream) {
-      ws.subscribe(conn, { stream });
+    if (!stream) return;
+
+    ws.subscribe(conn, { stream });
+
+    const filter = topicToFilter(stream);
+
+    if (filter) {
+      for await (const event of Sub.sub(socket, '1', [filter])) {
+        const status = await toStatus(event);
+        if (status) {
+          send('update', status);
+        }
+      }
     }
   });
 
@@ -46,4 +70,13 @@ const streamingController: AppController = (c) => {
   return response;
 };
 
+function topicToFilter(topic: string): DittoFilter<1> | undefined {
+  switch (topic) {
+    case 'public':
+      return { kinds: [1] };
+    case 'public:local':
+      return { kinds: [1], local: true };
+  }
+}
+
 export { streamingController };
diff --git a/src/subs.ts b/src/subs.ts
index 5c175c3..eebfac1 100644
--- a/src/subs.ts
+++ b/src/subs.ts
@@ -20,7 +20,7 @@ class SubscriptionStore {
    * }
    * ```
    */
-  sub(socket: WebSocket, id: string, filters: DittoFilter[]): Subscription {
+  sub<K extends number>(socket: WebSocket, id: string, filters: DittoFilter<K>[]): Subscription<K> {
     let subs = this.#store.get(socket);
 
     if (!subs) {
@@ -31,7 +31,7 @@ class SubscriptionStore {
     const sub = new Subscription(filters);
 
     this.unsub(socket, id);
-    subs.set(id, sub);
+    subs.set(id, sub as unknown as Subscription);
 
     return sub;
   }