diff --git a/src/component/lib.test.ts b/src/component/lib.test.ts index 470d151..f4270f9 100644 --- a/src/component/lib.test.ts +++ b/src/component/lib.test.ts @@ -6,6 +6,7 @@ import { describe, expect, it } from "vitest"; import { api } from "./_generated/api.js"; import schema from "./schema.js"; +import { textStreams } from "./streams.js"; const modules = import.meta.glob("./**/*.ts"); @@ -129,4 +130,84 @@ describe("bounded chunk reads", () => { t.query(api.lib.read, { streamId, cursor: null, numItems: 129 }), ).rejects.toThrow(); }); + + it("reads and reconstructs a core-backed facade across pages", async () => { + const t = convexTest(schema, modules); + const events = Array.from({ length: 129 }, (_, index) => `${index},`); + const streamId = await t.run(async (ctx) => { + const core = await textStreams.create(ctx); + const facadeId = await ctx.db.insert("streams", { + status: "done", + coreId: core.streamId, + }); + for (const [index, event] of events.entries()) { + await textStreams.append(ctx, { + streamId: core.streamId, + event, + ...(index === events.length - 1 ? { complete: true as const } : {}), + }); + } + return facadeId; + }); + + const first = await t.query(api.lib.read, { + streamId, + cursor: null, + numItems: 128, + }); + const second = await t.query(api.lib.read, { + streamId, + cursor: first.continueCursor, + numItems: 128, + }); + + expect(first.streamId).toBe(streamId); + expect(first.page.map(({ event }) => event)).toEqual(events.slice(0, 128)); + expect(first.caughtUp).toBe(false); + expect(second).toMatchObject({ + streamId, + page: [{ event: events[128] }], + caughtUp: true, + status: "done", + }); + await expect(t.query(api.lib.getStreamText, { streamId })).resolves.toEqual( + { text: events.join(""), status: "done" }, + ); + }); + + it("requires paged reads for a live core-backed facade", async () => { + const t = convexTest(schema, modules); + const streamId = await t.run(async (ctx) => { + const core = await textStreams.create(ctx); + return await ctx.db.insert("streams", { + status: "streaming", + coreId: core.streamId, + }); + }); + + await expect(t.query(api.lib.getStreamText, { streamId })).rejects.toThrow( + "readStreamRequired", + ); + }); + + it("rejects a terminal core snapshot over the serialized byte limit", async () => { + const t = convexTest(schema, modules); + const streamId = await t.run(async (ctx) => { + const core = await textStreams.create(ctx); + const facadeId = await ctx.db.insert("streams", { + status: "done", + coreId: core.streamId, + }); + await textStreams.append(ctx, { + streamId: core.streamId, + event: "\u0000".repeat(900_000), + complete: true, + }); + return facadeId; + }); + + await expect(t.query(api.lib.getStreamText, { streamId })).rejects.toThrow( + "fullBodyLimitExceeded", + ); + }); }); diff --git a/src/component/lib.ts b/src/component/lib.ts index bc5f9a4..9c7abe9 100644 --- a/src/component/lib.ts +++ b/src/component/lib.ts @@ -4,6 +4,7 @@ import { paginator } from "convex-helpers/server/pagination"; import { stream as queryStream } from "convex-helpers/server/stream"; import { ConvexError, convexToJson, jsonToConvex, v } from "convex/values"; import { streamReadLifecycleFromStatus } from "./status.js"; +import { textStreams } from "./streams.js"; import { internal } from "./_generated/api.js"; import type { Doc, Id } from "./_generated/dataModel.js"; import { internalMutation, mutation, query } from "./_generated/server.js"; @@ -13,7 +14,10 @@ const MAX_READ_ITEMS = 128; // The read loop checks this after appending a document, so one legal final // document may exceed the threshold. const MAX_READ_SCAN_BYTES = 256 * 1024; +const MAX_FULL_BODY_EVENTS = 4096; +const MAX_FULL_BODY_JSON_BYTES = 4 * 1024 * 1024; const READ_CURSOR_PREFIX = "pts1:"; +const textEncoder = new TextEncoder(); type ReadCursor = { streamId: Id<"streams">; @@ -24,6 +28,7 @@ type ReadCursor = { type ReadIndexKey = [Id<"streams">, number, string]; type TextReadResult = StreamReadResult<"streams", string>; type TextReadBase = Omit; +type CoreTextReadResult = StreamReadResult<"textStreams", string>; function fail(code: string, message: string): never { throw new ConvexError({ code, message }); @@ -216,7 +221,6 @@ export const getStreamStatus = query({ }); // Get the full text of a stream. -// Involves concatenating all the chunks. export const getStreamText = query({ args: { streamId: v.id("streams"), @@ -230,6 +234,45 @@ export const getStreamText = query({ if (!stream) { throw new Error("Stream not found"); } + if (stream.coreId !== undefined) { + if (stream.status === "pending" || stream.status === "streaming") { + fail( + "readStreamRequired", + "Use readStream for live stream consumption.", + ); + } + const events: string[] = []; + let eventCount = 0; + let serializedBytes = 0; + let cursor: string | null = null; + for (;;) { + const result: CoreTextReadResult = await textStreams.read(ctx, { + streamId: stream.coreId, + cursor, + numItems: MAX_READ_ITEMS, + }); + for (const { event } of result.page) { + eventCount += 1; + serializedBytes += textEncoder.encode( + JSON.stringify(convexToJson(event)), + ).byteLength; + if ( + eventCount > MAX_FULL_BODY_EVENTS || + serializedBytes > MAX_FULL_BODY_JSON_BYTES + ) { + fail( + "fullBodyLimitExceeded", + "Full stream snapshot exceeds its limit.", + ); + } + events.push(event); + } + if (result.caughtUp) { + return { text: events.join(""), status: stream.status }; + } + cursor = result.continueCursor; + } + } let text = ""; if (stream.status !== "pending") { const chunks = await ctx.db @@ -262,6 +305,14 @@ export const read = query({ if (stream === null) { fail("streamNotFound", "Stream not found."); } + if (stream.coreId !== undefined) { + const result = await textStreams.read(ctx, { + streamId: stream.coreId, + cursor: args.cursor, + numItems: limit, + }); + return { ...result, streamId: stream._id }; + } const cursor = decodeReadCursor(args.cursor, stream._id); const chunkStream = queryStream(ctx.db, schema) .query("chunks")