Merge pull request durable-streams#22 from durable-streams/test1

KyleAMathews · web-flow · commit c5ec6ac01e6e · 2025-12-09T16:13:31.000-07:00
Fix test-ui registry and stream handling
diff --git a/.claude/settings.local.json b/.claude/settings.local.json
@@ -27,7 +27,8 @@
       "Bash(pnpm --dir /Users/kylemathews/programs/ds-worktrees/conformance-tests/packages/test-ui add @durable-streams/writer @durable-streams/client)",
       "Bash(pnpm --filter @durable-streams/test-ui build:*)",
       "Bash(git add:*)",
-      "Bash(git commit:*)"
+      "Bash(git commit:*)",
+      "Bash(pnpm --filter @durable-streams/writer build:*)"
     ],
     "deny": [],
     "ask": []
diff --git a/packages/cli/example-server.ts b/packages/cli/example-server.ts
@@ -15,12 +15,12 @@ const server = new DurableStreamTestServer({
   host: `0.0.0.0`,
 })
 
+const url = await server.start()
+
 // Add hooks to maintain a __registry__ stream for observability
-const hooks = createRegistryHooks(server.store)
+const hooks = createRegistryHooks(server.store, url)
 ;(server as any).options.onStreamCreated = hooks.onStreamCreated
 ;(server as any).options.onStreamDeleted = hooks.onStreamDeleted
-
-const url = await server.start()
 console.log(`✓ Durable Streams server running at ${url}`)
 console.log(`\nYou can now use the CLI to interact with streams:`)
 console.log(`  export STREAM_URL=${url}`)
diff --git a/packages/server/package.json b/packages/server/package.json
@@ -19,6 +19,7 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
+    "@durable-streams/writer": "workspace:*",
     "@neophi/sieve-cache": "^1.0.0",
     "lmdb": "^3.3.0"
   },
diff --git a/packages/server/src/registry-hook.ts b/packages/server/src/registry-hook.ts
@@ -3,6 +3,7 @@
  * This stream records all create/delete events for observability.
  */
 
+import { DurableStream } from "@durable-streams/writer"
 import type { StreamLifecycleHook } from "./types"
 import type { StreamStore } from "./store"
 import type { FileBackedStreamStore } from "./file-store"
@@ -14,14 +15,21 @@ const REGISTRY_PATH = `/v1/stream/__registry__`
  * Any client can read this stream to discover all streams and their lifecycle events.
  */
 export function createRegistryHooks(
-  store: StreamStore | FileBackedStreamStore
+  store: StreamStore | FileBackedStreamStore,
+  serverUrl: string
 ): {
   onStreamCreated: StreamLifecycleHook
   onStreamDeleted: StreamLifecycleHook
 } {
-  const ensureRegistryExists = () => {
+  const registryStream = new DurableStream({
+    url: `${serverUrl}${REGISTRY_PATH}`,
+    contentType: `application/json`,
+  })
+
+  const ensureRegistryExists = async () => {
     if (!store.has(REGISTRY_PATH)) {
-      store.create(REGISTRY_PATH, {
+      await DurableStream.create({
+        url: `${serverUrl}${REGISTRY_PATH}`,
         contentType: `application/json`,
       })
     }
@@ -35,36 +43,28 @@ export function createRegistryHooks(
 
   return {
     onStreamCreated: async (event) => {
-      ensureRegistryExists()
+      await ensureRegistryExists()
 
       const streamName = extractStreamName(event.path)
 
-      const record = JSON.stringify({
+      await registryStream.append({
         type: event.type,
         path: streamName,
         contentType: event.contentType,
         timestamp: event.timestamp,
       })
-
-      await Promise.resolve(
-        store.append(REGISTRY_PATH, Buffer.from(record + `\n`))
-      )
     },
 
     onStreamDeleted: async (event) => {
-      ensureRegistryExists()
+      await ensureRegistryExists()
 
       const streamName = extractStreamName(event.path)
 
-      const record = JSON.stringify({
+      await registryStream.append({
         type: event.type,
         path: streamName,
         timestamp: event.timestamp,
       })
-
-      await Promise.resolve(
-        store.append(REGISTRY_PATH, Buffer.from(record + `\n`))
-      )
     },
   }
 }
diff --git a/packages/test-ui/src/routes/__root.tsx b/packages/test-ui/src/routes/__root.tsx
@@ -8,6 +8,13 @@ interface Stream {
   contentType?: string
 }
 
+interface RegistryEvent {
+  type: `created` | `deleted`
+  path: string
+  contentType?: string
+  timestamp: number
+}
+
 function RootLayout() {
   const [streams, setStreams] = useState<Array<Stream>>([])
   const [newStreamPath, setNewStreamPath] = useState(``)
@@ -36,40 +43,30 @@ function RootLayout() {
         })
       }
 
-      // Read all chunks from the registry
+      // Read all events from the registry
       const loadedStreams: Array<Stream> = []
 
       try {
-        for await (const chunk of registryStream.read({ offset: `-1` })) {
-          console.log(`Registry chunk:`, chunk)
-          if (chunk.data.length > 0) {
-            const text = new TextDecoder().decode(chunk.data)
-            console.log(`Registry text:`, text)
-            const lines = text.trim().split(`\n`).filter(Boolean)
-
-            for (const line of lines) {
-              try {
-                const event = JSON.parse(line)
-                console.log(`Registry event:`, event)
-                if (event.type === `created`) {
-                  loadedStreams.push({
-                    path: event.path,
-                    contentType: event.contentType,
-                  })
-                } else if (event.type === `deleted`) {
-                  const index = loadedStreams.findIndex(
-                    (s) => s.path === event.path
-                  )
-                  if (index !== -1) {
-                    loadedStreams.splice(index, 1)
-                  }
-                }
-              } catch (e) {
-                console.error(`Error parsing registry line:`, line, e)
+        for await (const chunk of registryStream.json<
+          RegistryEvent | Array<RegistryEvent>
+        >()) {
+          const events = Array.isArray(chunk) ? chunk : [chunk]
+
+          for (const event of events) {
+            if (event.type === `created`) {
+              loadedStreams.push({
+                path: event.path,
+                contentType: event.contentType,
+              })
+            } else {
+              const index = loadedStreams.findIndex(
+                (s) => s.path === event.path
+              )
+              if (index !== -1) {
+                loadedStreams.splice(index, 1)
               }
             }
           }
-          console.log(`Loaded streams:`, loadedStreams)
           setStreams(loadedStreams)
         }
       } catch (readErr) {
@@ -106,7 +103,7 @@ function RootLayout() {
   const deleteStream = async (path: string) => {
     if (
       !window.confirm(
-        `Delete stream "${path}"?\n\nThis action cannot be undone.`
+        `Delete stream "${decodeURIComponent(path)}"?\n\nThis action cannot be undone.`
       )
     ) {
       return
@@ -140,7 +137,7 @@ function RootLayout() {
             placeholder="New stream path"
             value={newStreamPath}
             onChange={(e) => setNewStreamPath(e.target.value)}
-            onKeyPress={(e) => e.key === `Enter` && void createStream()}
+            onKeyDown={(e) => e.key === `Enter` && void createStream()}
           />
           <select
             value={newStreamContentType}
@@ -163,14 +160,16 @@ function RootLayout() {
               onClick={() => setSidebarOpen(false)}
             >
               <div>
-                <div className="stream-path">{stream.path}</div>
+                <div className="stream-path">
+                  {decodeURIComponent(stream.path)}
+                </div>
                 <div className="stream-type">
-                  {stream.contentType || `unknown`}
+                  {stream.contentType?.toLowerCase() || `unknown`}
                 </div>
               </div>
               <button
                 className="delete-btn"
-                title={`Delete stream: ${stream.path}`}
+                title={`Delete stream: ${decodeURIComponent(stream.path)}`}
                 onClick={(e) => {
                   e.preventDefault()
                   e.stopPropagation()
diff --git a/packages/test-ui/src/routes/index.tsx b/packages/test-ui/src/routes/index.tsx
@@ -5,5 +5,5 @@ export const Route = createFileRoute(`/`)({
 })
 
 function Index() {
-  return <div className="placeholder" />
+  return null
 }
diff --git a/packages/test-ui/src/routes/stream.$streamPath.tsx b/packages/test-ui/src/routes/stream.$streamPath.tsx
@@ -1,25 +1,34 @@
-import { createFileRoute } from "@tanstack/react-router"
+import { createFileRoute, redirect } from "@tanstack/react-router"
 import { useEffect, useRef, useState } from "react"
 import { DurableStream } from "@durable-streams/writer"
 
 const SERVER_URL = `http://${typeof window !== `undefined` ? window.location.hostname : `localhost`}:8787`
 
 export const Route = createFileRoute(`/stream/$streamPath`)({
   loader: async ({ params }) => {
-    const stream = new DurableStream({
-      url: `${SERVER_URL}/v1/stream/${params.streamPath}`,
-    })
-    const metadata = await stream.head()
-    return {
-      contentType: metadata.contentType || null,
+    try {
+      const streamMetadata = new DurableStream({
+        url: `${SERVER_URL}/v1/stream/${params.streamPath}`,
+      })
+      const metadata = await streamMetadata.head()
+      const stream = new DurableStream({
+        url: `${SERVER_URL}/v1/stream/${params.streamPath}`,
+        contentType: metadata.contentType || undefined,
+      })
+      return {
+        contentType: metadata.contentType || undefined,
+        stream,
+      }
+    } catch {
+      throw redirect({ to: `/` })
     }
   },
   component: StreamViewer,
 })
 
 function StreamViewer() {
   const { streamPath } = Route.useParams()
-  const { contentType } = Route.useLoaderData()
+  const { contentType, stream } = Route.useLoaderData()
   const [messages, setMessages] = useState<
     Array<{ offset: string; data: string }>
   >([])
@@ -43,10 +52,6 @@ function StreamViewer() {
 
     const followStream = async () => {
       try {
-        const stream = new DurableStream({
-          url: `${SERVER_URL}/v1/stream/${streamPath}`,
-        })
-
         for await (const chunk of stream.read({
           offset: `-1`,
           live: `long-poll`,
@@ -80,9 +85,6 @@ function StreamViewer() {
 
     try {
       setError(null)
-      const stream = new DurableStream({
-        url: `${SERVER_URL}/v1/stream/${streamPath}`,
-      })
       await stream.append(writeInput + `\n`)
       setWriteInput(``)
     } catch (err: any) {
@@ -112,17 +114,19 @@ function StreamViewer() {
             Listening for new messages...
           </div>
         )}
-        {isJsonStream ? (
-          messages.map((msg, i) => (
-            <div key={i} className="message">
-              <pre>{msg.data}</pre>
+        {messages.length !== 0 ? (
+          isJsonStream ? (
+            messages.map((msg, i) => (
+              <div key={i} className="message">
+                <pre>{msg.data}</pre>
+              </div>
+            ))
+          ) : (
+            <div className="message">
+              <pre>{messages.map((msg) => msg.data).join(``)}</pre>
             </div>
-          ))
-        ) : (
-          <div className="message">
-            <pre>{messages.map((msg) => msg.data).join(``)}</pre>
-          </div>
-        )}
+          )
+        ) : null}
         <div ref={messagesEndRef} />
       </div>
       {!isRegistryStream && (
diff --git a/packages/writer/src/writer.ts b/packages/writer/src/writer.ts
@@ -119,15 +119,15 @@ export class DurableStream extends BaseStream {
 
     if (!response.ok) {
       const text = await response.text()
-      const headers: Record<string, string> = {}
+      const responseHeaders: Record<string, string> = {}
       response.headers.forEach((v, k) => {
-        headers[k] = v
+        responseHeaders[k] = v
       })
       throw new FetchError(
         response.status,
         text,
         undefined,
-        headers,
+        responseHeaders,
         opts.url,
         response.status === 409
           ? `Stream already exists`
@@ -252,7 +252,8 @@ export class DurableStream extends BaseStream {
       }
     }
 
-    const isJson = normalizeContentType(this.contentType) === `application/json`
+    const isJson =
+      normalizeContentType(this.options.contentType) === `application/json`
 
     // Batch data
     let batchedBody: BodyInit
@@ -286,8 +287,8 @@ export class DurableStream extends BaseStream {
 
     // Send
     const headers: Record<string, string> = {}
-    if (this.contentType) {
-      headers[`content-type`] = this.contentType
+    if (this.options.contentType) {
+      headers[`content-type`] = this.options.contentType
     }
     if (highestSeq) {
       headers[STREAM_SEQ_HEADER] = highestSeq
@@ -301,9 +302,9 @@ export class DurableStream extends BaseStream {
 
     if (!response.ok) {
       const text = await response.text()
-      const headers: Record<string, string> = {}
+      const responseHeaders: Record<string, string> = {}
       response.headers.forEach((v, k) => {
-        headers[k] = v
+        responseHeaders[k] = v
       })
 
       let message: string
@@ -312,7 +313,7 @@ export class DurableStream extends BaseStream {
       } else if (response.status === 409) {
         message = `Sequence conflict`
       } else if (response.status === 400) {
-        message = `Bad request (possibly content-type mismatch)`
+        message = `Bad request (possibly content-type mismatch), ${text}`
       } else {
         message = `Failed to append`
       }
@@ -321,7 +322,7 @@ export class DurableStream extends BaseStream {
         response.status,
         text,
         undefined,
-        headers,
+        responseHeaders,
         this.url,
         message
       )
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml

Original file line number	Diff line number	Diff line change
@@ -5,5 +5,5 @@ export const Route = createFileRoute(`/`)({
`5`	`5`	`})`
`6`	`6`
`7`	`7`	`function Index() {`
`8`		`- return <div className="placeholder" />`
	`8`	`+ return null`
`9`	`9`	`}`