import { describe, expect, it } from "vitest"; import { createIncrementalSseParser, type SseParserItem, } from "../../../src/adapters/realtime/sse/sse-parser.ts"; const encoder = new TextEncoder(); function pushText( parser: ReturnType, text: string, ): readonly SseParserItem[] { const result = parser.push(encoder.encode(text)); if (!result.ok) throw new Error(result.error.kind); return result.value; } describe("incremental SSE parser", () => { it("handles BOM, chunk boundaries, CR/LF/CRLF, comments and multi-line data", () => { const parser = createIncrementalSseParser(); const chunks = [ "\uFEFF: ready\r", "\nretry: 2500\revent: resource.updated\n", "id: cursor-1\r\ndata: first\rdata:second\n\n", ]; const items = chunks.flatMap((chunk) => pushText(parser, chunk)); expect(items).toEqual([ { kind: "COMMENT" }, { kind: "RETRY", retryMs: 2_500 }, { kind: "EVENT", eventType: "resource.updated", data: "first\nsecond", id: "cursor-1", hasExplicitId: true, }, ]); expect(parser.finish()).toEqual({ ok: true, value: { items: [], incompleteEventDiscarded: false, }, }); }); it("preserves standard inherited ID state while marking direct IDs", () => { const parser = createIncrementalSseParser(); const items = pushText( parser, "id: cursor-a\ndata: one\n\ndata: two\n\ndata:\n\n", ); expect(items).toEqual([ { kind: "EVENT", eventType: "message", data: "one", id: "cursor-a", hasExplicitId: true, }, { kind: "EVENT", eventType: "message", data: "two", id: "cursor-a", hasExplicitId: false, }, { kind: "EVENT", eventType: "message", data: "", id: "cursor-a", hasExplicitId: false, }, ]); }); it("ignores invalid ID and retry fields without inventing a cursor", () => { const parser = createIncrementalSseParser(); expect( pushText( parser, "id: invalid\u0000cursor\nretry: 99999\ndata: value\n\n", ), ).toEqual([ { kind: "EVENT", eventType: "message", data: "value", id: null, hasExplicitId: false, }, ]); }); it("discards an event that was not terminated by a blank line", () => { const parser = createIncrementalSseParser(); expect(pushText(parser, "data: incomplete\n")).toEqual([]); expect(parser.finish()).toEqual({ ok: true, value: { items: [], incompleteEventDiscarded: true, }, }); expect(parser.push(encoder.encode("data: late\n\n"))).toEqual({ ok: false, error: { kind: "CLOSED", operation: "DECODE", retryable: false, }, }); }); it("fails closed on malformed UTF-8 and parser ceilings", () => { const malformed = createIncrementalSseParser(); expect( malformed.push(new Uint8Array([0xc3, 0x28])), ).toEqual({ ok: false, error: { kind: "MALFORMED_EVENT", operation: "DECODE", retryable: false, }, }); const longLine = createIncrementalSseParser({ maxLineBytes: 4, maxEventBytes: 8, maxIncompleteBufferBytes: 8, maxRetryMs: 100, }); expect(longLine.push(encoder.encode("data:"))).toEqual({ ok: false, error: { kind: "EVENT_TOO_LARGE", operation: "DECODE", retryable: false, }, }); const largeEvent = createIncrementalSseParser({ maxLineBytes: 16, maxEventBytes: 8, maxIncompleteBufferBytes: 16, maxRetryMs: 100, }); expect(largeEvent.push(encoder.encode("data:abc\n"))).toEqual({ ok: false, error: { kind: "EVENT_TOO_LARGE", operation: "DECODE", retryable: false, }, }); const excessiveBatch = createIncrementalSseParser({ maxItemsPerChunk: 2, }); expect(excessiveBatch.push(encoder.encode(":\n:\n:\n"))).toEqual({ ok: false, error: { kind: "QUEUE_OVERFLOW", operation: "DECODE", retryable: false, }, }); const oversizedChunk = createIncrementalSseParser({ maxChunkBytes: 65_536, }); expect( oversizedChunk.push(new Uint8Array(65_537)), ).toEqual({ ok: false, error: { kind: "EVENT_TOO_LARGE", operation: "DECODE", retryable: false, }, }); }); });