WIP: Allow streaming from custom http endpoints

cmd/plug_run.ts

@@ -7,7 +7,7 @@ import { DiskSpacePrimitives } from "$common/spaces/disk_space_primitives.ts";
 import { ServerSystem } from "../server/server_system.ts";
 import { AssetBundlePlugSpacePrimitives } from "$common/spaces/asset_bundle_space_primitives.ts";
 import { LocalShell } from "../server/shell_backend.ts";
-import { Hono } from "hono/mod.ts";
+import { Hono } from "@hono/hono";
 import { DataStore } from "$lib/data/datastore.ts";
 import { DataStoreMQ } from "$lib/data/mq.datastore.ts";
 import { EventHook } from "../common/hooks/event.ts";

common/hooks/event.ts

@@ -65,6 +65,7 @@ export class EventHook implements EventHookI {
                 promises.push((async () => {
                   try {
                     const result = await plug.invoke(name, args);
+                    console.log("Result from event dispatch for ", eventName, " is ", result);
                     if (result !== undefined) {
                       responses.push(result);
                     }

common/spaces/evented_space_primitives.ts

@@ -54,7 +54,7 @@ export class EventedSpacePrimitives implements SpacePrimitives {
       // Some other operation (read, write, list, meta) is already going on
       // this will likely trigger events, so let's not worry about any of that and avoid race condition and inconsistent data.
       console.info(
-        "alreadyFetching is on, skipping even triggering for fetchFileList.",
+        "alreadyFetching is on, skipping event triggering for fetchFileList.",
       );
       return this.wrapped.fetchFileList();
     }

deno.json

@@ -87,6 +87,11 @@
   },
   "imports": {
     "@cliffy/command": "jsr:@cliffy/command@^1.0.0-rc.7",
+    "@hono/hono": "jsr:@hono/hono@^4.6.17",
+    "@hono/hono/streaming": "jsr:@hono/hono@^4.6.17/streaming",
+    "@hono/hono/validator": "jsr:@hono/hono@^4.6.17/validator",
+    "@hono/hono/cors": "jsr:@hono/hono@^4.6.17/cors",
+    "@hono/hono/cookie": "jsr:@hono/hono@^4.6.17/cookie",
     "@silverbulletmd/silverbullet/syscall": "./plug-api/syscall.ts",
     "@silverbulletmd/silverbullet/syscalls": "./plug-api/syscalls.ts",
     "@silverbulletmd/silverbullet/types": "./plug-api/types.ts",
@@ -147,7 +152,6 @@
     "fast-diff": "https://esm.sh/[email protected]",
     "fuse": "https://deno.land/x/[email protected]/dist/fuse.esm.min.js",
     "gitignore-parser": "https://esm.sh/[email protected]",
-    "hono/": "https://deno.land/x/[email protected]/",
     "js-yaml": "https://esm.sh/[email protected]",
     "mimetypes": "https://deno.land/x/[email protected]/mod.ts",
     "porter-stemmer": "https://esm.sh/[email protected]",

server/http_server.ts

@@ -1,6 +1,7 @@
-import { deleteCookie, getCookie, setCookie } from "hono/helper.ts";
-import { cors } from "hono/middleware.ts";
-import { type Context, Hono, validator } from "hono/mod.ts";
+import { deleteCookie, getCookie, setCookie } from "@hono/hono/cookie";
+import { cors } from "@hono/hono/cors";
+import { type Context, Hono } from "@hono/hono";
+import { validator } from "@hono/hono/validator";
 import type { AssetBundle } from "$lib/asset_bundle/bundle.ts";
 import type {
   EndpointRequest,
@@ -22,6 +23,8 @@ import {
 import { base64Encode } from "$lib/crypto.ts";
 import { LockoutTimer } from "./lockout.ts";
 import type { AuthOptions } from "../cmd/server.ts";
+import { streamSSE } from "@hono/hono/streaming";
+
 
 const authenticationExpirySeconds = 60 * 60 * 24 * 7; // 1 week
 
@@ -200,6 +203,7 @@ export class HttpServer {
             ),
             headers: req.header(),
           } as EndpointRequest);
+
         if (responses.length === 0) {
           return ctx.text(
             "No custom endpoint handler is handling this path",
@@ -211,22 +215,92 @@ export class HttpServer {
             500,
           );
         }
+
         const response = responses[0];
         if (response.headers) {
-          for (
-            const [key, value] of Object.entries(
-              response.headers,
-            )
-          ) {
+          for (const [key, value] of Object.entries(response.headers)) {
             ctx.header(key, value);
           }
         }
         ctx.status(response.status || 200);
+
+        console.log("Response body type:", typeof response.body);
+        console.log("Response body:", response.body);
+        console.log("Response headers:", response.headers);
+        console.log("Response status:", response.status);
+
         if (typeof response.body === "string") {
+          console.log("Returning text response");
           return ctx.text(response.body);
         } else if (response.body instanceof Uint8Array) {
+          console.log("Returning Uint8Array response");
           return ctx.body(response.body);
+        // response.body never ends up being a ReadableStream, it's always an empty object.
+        // I think this is because the event system or something else is serializing/deserializing it
+        // } else if (response.body instanceof ReadableStream || response.headers["content-type"]?.includes("text/event-stream")) {
+        } else if (response.body instanceof ReadableStream) {
+          console.log("Received ReadableStream body or SSE stream");
+          console.log("response.body type: ", typeof response.body, "instanceof ReadableStream: ", response.body instanceof ReadableStream);
+          console.log("Full body: ", response.body);
+          console.log("Returning streaming response");
+
+          if (response.headers["content-type"]?.includes("text/event-stream")) {
+            return streamSSE(ctx, async (stream) => {
+              try {
+                // Ensure response.body is a ReadableStream
+                if (response.body instanceof ReadableStream) {
+                  const reader = response.body.getReader();
+                  while (true) {
+                    const { done, value } = await reader.read();
+                    if (done) break;
+
+                    // Convert the chunk to text and write it
+                    const text = new TextDecoder().decode(value);
+                    await stream.write(text);
+                  }
+                } else {
+                  console.error("Response body is not a ReadableStream");
+                }
+              } catch (e) {
+                console.error("Error in SSE stream:", e);
+              }
+            });
+          } else {
+            // For non-SSE streams, use regular streaming
+            ctx.header("Content-Type", response.headers["content-type"] || "application/octet-stream");
+            return ctx.body(response.body);
+          }
+        } else if (response.body?.__type === 'STREAM_PROXY') {
+          // This is kind of a hack. Because of the issue above, we need to initiate the fetch from
+          // inside this function, which means this function becomes a proxy too.
+          console.log("Creating new stream from proxy data method: ", response.body.method, " url: ", response.body.url, " headers: ", response.body.headers, " requestBody: ", response.body.requestBody);
+          const streamResponse = await fetch(response.body.url, {
+            method: response.body.method,
+            headers: response.body.headers,
+            body: JSON.stringify(response.body.requestBody),
+          });
+
+          ctx.status(streamResponse.status);
+
+          if (!streamResponse.ok) {
+            return ctx.text(`Stream proxy error: ${await streamResponse.text()}`, streamResponse.status);
+          }
+
+          return streamSSE(ctx, async (stream) => {
+            try {
+              const reader = streamResponse.body!.getReader();
+              while (true) {
+                const { done, value } = await reader.read();
+                if (done) break;
+                const text = new TextDecoder().decode(value);
+                await stream.write(text);
+              }
+            } catch (e) {
+              console.error("Error in SSE stream:", e);
+            }
+          });
         } else {
+          console.log("Returning JSON response");
           return ctx.json(response.body);
         }
       } catch (e: any) {

web/service_worker.ts

@@ -82,6 +82,11 @@ self.addEventListener("fetch", (event: any) => {
         return fetch(request);
       }
 
+      // Is this a request to a custom endpoint? If so, proxy to the actual server
+      if (location.host == requestUrl.host && requestUrl.pathname.startsWith('/_/')) {
+        return fetch(request);
+      }
+
       // Any request with the X-Sync-Mode header originates from the sync engine: pass it on to the server
       if (request.headers.has("x-sync-mode")) {
         return fetch(request);