fix redis/turso webhook e2e (#29)

Author: dustintownsend

.changeset/tame-bags-work.md

@@ -0,0 +1,10 @@
+---
+"@workflow-worlds/redis": patch
+"@workflow-worlds/turso": patch
+---
+
+Fix webhook e2e regressions where webhook endpoints could return 404 in Redis and Turso worlds.
+
+- Redis: avoid closing stream readers before final persisted chunks are drained.
+- Turso: avoid closing stream readers during initial replay before buffered chunks are delivered.
+- Turso: normalize hook metadata nulls to undefined to preserve expected hydration behavior.

packages/redis/src/streamer.ts

@@ -161,6 +161,7 @@ export async function createStreamer(options: {
             eof: boolean;
           }> = [];
           let isClosed = false;
+          let closeSignalReceived = false;
 
           // Handler for pub/sub messages
           const messageHandler = (channel: string, message: string) => {
@@ -170,15 +171,9 @@ export async function createStreamer(options: {
               const parsed = JSON.parse(message);
 
               if (parsed.type === 'close') {
-                if (!isClosed) {
-                  isClosed = true;
-                  cleanup();
-                  try {
-                    controller.close();
-                  } catch {
-                    // Ignore if already closed
-                  }
-                }
+                // A close signal can race ahead of chunk polling. Delay closure
+                // until we have drained persisted entries (including final chunk).
+                closeSignalReceived = true;
               }
               // For chunk notifications, we'll poll for new data
             } catch {
@@ -319,7 +314,7 @@ export async function createStreamer(options: {
               if (!newEntries || newEntries.length === 0) {
                 // Check if stream is now closed
                 const nowClosed = await redis.get(closedKey);
-                if (nowClosed === '1' && !isClosed) {
+                if ((closeSignalReceived || nowClosed === '1') && !isClosed) {
                   isClosed = true;
                   clearInterval(pollInterval);
                   cleanup();

packages/turso/src/storage.ts

@@ -209,7 +209,7 @@ function toHook(row: HookRow, specVersion?: number): Hook {
     hookId: row.hookId,
     runId: row.runId,
     token: row.token,
-    metadata: row.metadata,
+    metadata: row.metadata ?? undefined,
     ownerId: row.ownerId,
     projectId: row.projectId,
     environment: row.environment,

packages/turso/src/streamer.ts

@@ -174,6 +174,7 @@ export function createStreamer(config: StreamerConfig): Streamer {
           // Buffer for chunks that arrive during initial load
           const bufferedEventChunks: StreamChunk[] = [];
           let isLoadingFromStorage = true;
+          let closeRequested = false;
 
           // Handler for new chunks (real-time)
           const chunkHandler = (chunk: StreamChunk) => {
@@ -202,6 +203,12 @@ export function createStreamer(config: StreamerConfig): Streamer {
 
           // Handler for stream close
           const closeHandler = () => {
+            if (isLoadingFromStorage) {
+              // Don't close immediately during initial load. A close event can
+              // race ahead of buffered chunk delivery.
+              closeRequested = true;
+              return;
+            }
             cleanup?.();
             try {
               controller.close();
@@ -274,6 +281,11 @@ export function createStreamer(config: StreamerConfig): Streamer {
                 }
               }
 
+              if (closeRequested) {
+                cleanup?.();
+                controller.close();
+              }
+
             } catch (error) {
               cleanup?.();
               controller.error(error);