feat(component): read core-backed streams

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",
+    );
+  });
 });

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<TextReadResult, "status" | "error">;
+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")