Merge pull request durable-streams#20 from durable-streams/json-mode-and-batching

Add JSON mode and batching writer

Author: KyleAMathews

PROTOCOL.md

@@ -400,19 +400,62 @@ Clients **MUST** use the `Stream-Next-Offset` value returned in responses for su
 
 ## 7. Content Types
 
-The protocol supports arbitrary MIME content types. The system operates at the byte level, leaving message framing and interpretation to clients.
+The protocol supports arbitrary MIME content types. Most content types operate at the byte level, leaving message framing and interpretation to clients. The `application/json` content type has special semantics defined below.
 
 **Restriction:**
 
 - SSE mode (Section 5.7) **REQUIRES** `content-type: text/*` or `application/json`
 
 Clients **MAY** use any content type for their streams, including:
 
+- `application/json` for JSON mode with message boundary preservation
 - `application/ndjson` for newline-delimited JSON
 - `application/x-protobuf` for Protocol Buffer messages
 - `text/plain` for plain text
 - Custom types for application-specific formats
 
+### 7.1. JSON Mode
+
+Streams created with `Content-Type: application/json` have special semantics for message boundaries and batch operations.
+
+#### Message Boundaries
+
+For `application/json` streams, servers **MUST** preserve message boundaries. Each POST request stores messages as a distinct unit, and GET responses **MUST** return data as a JSON array containing all messages from the requested offset range.
+
+#### Array Flattening for Batch Operations
+
+When a POST request body contains a JSON array, servers **MUST** flatten exactly one level of the array, treating each element as a separate message. This enables clients to batch multiple messages in a single HTTP request while preserving individual message semantics.
+
+**Examples (direct POST to server):**
+
+- POST body `{"event": "created"}` stores one message: `{"event": "created"}`
+- POST body `[{"event": "a"}, {"event": "b"}]` stores two messages: `{"event": "a"}`, `{"event": "b"}`
+- POST body `[[1,2], [3,4]]` stores two messages: `[1,2]`, `[3,4]`
+- POST body `[[[1,2,3]]]` stores one message: `[[1,2,3]]`
+
+**Note:** Client libraries **MAY** automatically wrap individual values in arrays for batching. For example, a client calling `append({"x": 1})` might send POST body `[{"x": 1}]` to the server, which flattens it to store one message: `{"x": 1}`.
+
+#### Empty Arrays
+
+Servers **MUST** reject POST requests containing empty JSON arrays (`[]`) with `400 Bad Request`. Empty arrays represent no-op operations with no semantic meaning.
+
+#### JSON Validation
+
+Servers **MUST** validate that appended data is valid JSON. If validation fails, servers **MUST** return `400 Bad Request` with an appropriate error message.
+
+#### Response Format
+
+GET responses for `application/json` streams **MUST** return `Content-Type: application/json` with a body containing a JSON array of all messages in the requested range:
+
+```http
+HTTP/1.1 200 OK
+Content-Type: application/json
+
+[{"event":"created"},{"event":"updated"}]
+```
+
+If no messages exist in the range, servers **MUST** return an empty JSON array `[]`.
+
 ## 8. Caching and Collapsing
 
 ### 8.1. Catch-up and Long-poll Reads

README.md

@@ -270,22 +270,23 @@ const stream = await DurableStream.create({
 })
 
 // Append individual JSON values
-await stream.append(JSON.stringify({ event: "user.created", userId: "123" }))
-await stream.append(JSON.stringify({ event: "user.updated", userId: "123" }))
-
-// Read returns parsed JSON array automatically
-const result = await stream.read({ live: "catchup" })
-// result.data = [
-//   { event: "user.created", userId: "123" },
-//   { event: "user.updated", userId: "123" }
-// ]
+await stream.append({ event: "user.created", userId: "123" })
+await stream.append({ event: "user.updated", userId: "123" })
+
+// Read returns parsed JSON array
+for await (const message of stream.json({ live: "catchup" })) {
+  console.log(message)
+  // { event: "user.created", userId: "123" }
+  // { event: "user.updated", userId: "123" }
+}
 ```
 
 In JSON mode:
 
-- Each append must be a valid JSON value
-- The server batches appends into JSON arrays for reads
-- Message boundaries are preserved
+- Each `append()` stores one message
+- Supports all JSON types: objects, arrays, strings, numbers, booleans, null
+- Message boundaries are preserved across reads
+- Reads return JSON arrays of all messages
 - Ideal for structured event streams
 
 ## Offset Semantics

packages/benchmarks/src/index.ts

@@ -312,21 +312,11 @@ export function runBenchmarks(options: BenchmarkOptions): void {
 
         const startTime = performance.now()
 
-        // Create a readable stream that generates a fixed number of chunks
-        let chunksGenerated = 0
-        const messageStream = new ReadableStream({
-          pull(controller) {
-            if (chunksGenerated >= totalChunks) {
-              controller.close()
-              return
-            }
-            controller.enqueue(chunk)
-            chunksGenerated++
-          },
-        })
-
-        // Stream all data through a single connection
-        await stream.appendStream(messageStream)
+        const appends = []
+        for (let i = 0; i < totalChunks; i++) {
+          appends.push(stream.append(chunk))
+        }
+        await Promise.all(appends)
 
         const endTime = performance.now()
 

packages/client/test/stream.test.ts

@@ -305,204 +305,7 @@ describe(`DurableStream`, () => {
     })
   })
 
-  describe(`create`, () => {
-    let mockFetch: Mock<typeof fetch>
-
-    beforeEach(() => {
-      mockFetch = vi.fn()
-    })
-
-    it(`should create stream with PUT`, async () => {
-      mockFetch.mockResolvedValue(
-        new Response(null, {
-          status: 201,
-          headers: { "content-type": `application/json` },
-        })
-      )
-
-      const stream = new DurableStream({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      await stream.create({ contentType: `application/json` })
-
-      expect(mockFetch).toHaveBeenCalledWith(
-        `https://example.com/stream`,
-        expect.objectContaining({
-          method: `PUT`,
-          headers: expect.objectContaining({
-            "content-type": `application/json`,
-          }),
-        })
-      )
-    })
-
-    it(`should throw FetchError on 409`, async () => {
-      mockFetch.mockResolvedValue(
-        new Response(`Already exists`, {
-          status: 409,
-          statusText: `Conflict`,
-        })
-      )
-
-      const stream = new DurableStream({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      await expect(stream.create()).rejects.toThrow(FetchError)
-    })
-  })
-
-  describe(`delete`, () => {
-    let mockFetch: Mock<typeof fetch>
-
-    beforeEach(() => {
-      mockFetch = vi.fn()
-    })
-
-    it(`should delete stream with DELETE`, async () => {
-      mockFetch.mockResolvedValue(new Response(null, { status: 204 }))
-
-      const stream = new DurableStream({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      await stream.delete()
-
-      expect(mockFetch).toHaveBeenCalledWith(
-        `https://example.com/stream`,
-        expect.objectContaining({ method: `DELETE` })
-      )
-    })
-
-    it(`should throw FetchError on 404`, async () => {
-      mockFetch.mockResolvedValue(
-        new Response(`Not found`, {
-          status: 404,
-          statusText: `Not Found`,
-        })
-      )
-
-      const stream = new DurableStream({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      await expect(stream.delete()).rejects.toThrow(FetchError)
-    })
-  })
-
-  describe(`append`, () => {
-    let mockFetch: Mock<typeof fetch>
-
-    beforeEach(() => {
-      mockFetch = vi.fn()
-    })
-
-    it(`should append string data as UTF-8`, async () => {
-      mockFetch.mockResolvedValue(new Response(null, { status: 200 }))
-
-      const stream = new DurableStream({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      await stream.append(`hello world`)
-
-      expect(mockFetch).toHaveBeenCalledWith(
-        `https://example.com/stream`,
-        expect.objectContaining({
-          method: `POST`,
-          body: expect.any(Uint8Array),
-        })
-      )
-
-      const call = mockFetch.mock.calls[0]!
-      const body = call[1]?.body as Uint8Array
-      expect(new TextDecoder().decode(body)).toBe(`hello world`)
-    })
-
-    it(`should append Uint8Array directly`, async () => {
-      mockFetch.mockResolvedValue(new Response(null, { status: 200 }))
-
-      const stream = new DurableStream({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      const data = new Uint8Array([1, 2, 3, 4])
-      await stream.append(data)
-
-      const call = mockFetch.mock.calls[0]!
-      const body = call[1]?.body as Uint8Array
-      expect(body).toEqual(data)
-    })
-
-    it(`should include seq header when provided`, async () => {
-      mockFetch.mockResolvedValue(new Response(null, { status: 200 }))
-
-      const stream = new DurableStream({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      await stream.append(`data`, { seq: `123` })
-
-      expect(mockFetch).toHaveBeenCalledWith(
-        expect.anything(),
-        expect.objectContaining({
-          headers: expect.objectContaining({
-            "Stream-Seq": `123`,
-          }),
-        })
-      )
-    })
-
-    it(`should throw FetchError on 409`, async () => {
-      mockFetch.mockResolvedValue(
-        new Response(`Sequence conflict`, {
-          status: 409,
-          statusText: `Conflict`,
-        })
-      )
-
-      const stream = new DurableStream({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      await expect(stream.append(`data`, { seq: `1` })).rejects.toThrow(
-        FetchError
-      )
-    })
-  })
-
   describe(`static methods`, () => {
-    it(`DurableStream.create should create and return handle`, async () => {
-      const mockFetch = vi.fn().mockResolvedValue(
-        new Response(null, {
-          status: 201,
-          headers: { "content-type": `text/plain` },
-        })
-      )
-
-      const stream = await DurableStream.create({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-        contentType: `text/plain`,
-      })
-
-      expect(stream).toBeInstanceOf(DurableStream)
-      expect(stream.url).toBe(`https://example.com/stream`)
-      expect(mockFetch).toHaveBeenCalledWith(
-        expect.anything(),
-        expect.objectContaining({ method: `PUT` })
-      )
-    })
-
     it(`DurableStream.connect should validate and return handle`, async () => {
       const mockFetch = vi.fn().mockResolvedValue(
         new Response(null, {
@@ -516,7 +319,6 @@ describe(`DurableStream`, () => {
         fetch: mockFetch,
       })
 
-      expect(stream).toBeInstanceOf(DurableStream)
       expect(stream.contentType).toBe(`application/json`)
       expect(mockFetch).toHaveBeenCalledWith(
         expect.anything(),
@@ -544,22 +346,6 @@ describe(`DurableStream`, () => {
       expect(result.contentType).toBe(`text/plain`)
       expect(result.offset).toBe(`5_100`)
     })
-
-    it(`DurableStream.delete should delete without returning handle`, async () => {
-      const mockFetch = vi
-        .fn()
-        .mockResolvedValue(new Response(null, { status: 204 }))
-
-      await DurableStream.delete({
-        url: `https://example.com/stream`,
-        fetch: mockFetch,
-      })
-
-      expect(mockFetch).toHaveBeenCalledWith(
-        expect.anything(),
-        expect.objectContaining({ method: `DELETE` })
-      )
-    })
   })
 
   describe(`auth`, () => {