1import { OrchestrationV2ThreadShellJson } from "@t3tools/contracts";
2import * as Effect from "effect/Effect";
3import * as Schema from "effect/Schema";
4
5import { yieldToHostTimer } from "./hostYield.ts";
6import { StoredOrchestrationShellSnapshot } from "./orchestrationCache.ts";
7
8const ROWS_PER_CHUNK = 32;
9
10const encodeThreadChunk = Schema.encodeEffect(Schema.Array(OrchestrationV2ThreadShellJson));
11const EncodedRows = Schema.Array(Schema.Unknown);
12const encodeEnvelope = Schema.encodeEffect(
13 Schema.fromJsonString(
14 StoredOrchestrationShellSnapshot.mapFields((fields) => ({
15 ...fields,
16 snapshot: fields.snapshot.mapFields((snapshotFields) => ({
17 ...snapshotFields,
18 threads: EncodedRows,
19 archivedThreads: EncodedRows,
20 })),
21 })),
22 ),
23);
24
25/**
26 * Shell rows are immutable and the shell reducer keeps unchanged rows by
27 * reference, so each encoder remembers the canonical encoding of rows it has
28 * already encoded successfully and only runs the row codec for new rows.
29 *
30 * `yieldToHost` defaults to a zero-delay timer.
31 */
32export function makeStoredShellSnapshotEncoder(
33 options: { readonly yieldToHost?: Effect.Effect<void> } = {},
34) {
35 const yieldToHost = options.yieldToHost ?? yieldToHostTimer;
36 const encodedRows = new WeakMap<OrchestrationV2ThreadShellJson, unknown>();
37
38 return Effect.fnUntraced(function* (stored: typeof StoredOrchestrationShellSnapshot.Type) {
39 let hasWorked = false;
40 const yieldBetweenChunks = Effect.suspend(() => {
41 if (hasWorked) return yieldToHost;
42 hasWorked = true;
43 return Effect.void;
44 });
45 const encodeMisses =
46 /* Cache encoding runs for shell update batches; avoid tracing each batch. */
47 Effect.fnUntraced(function* (misses: ReadonlyArray<OrchestrationV2ThreadShellJson>) {
48 yield* yieldBetweenChunks;
49 const chunk = yield* encodeThreadChunk(misses);
50 chunk.forEach((row, index) => encodedRows.set(misses[index]!, row));
51 });
52 const encodeRows =
53 /* Cache encoding runs for shell update batches; avoid tracing each batch. */
54 Effect.fnUntraced(function* (rows: ReadonlyArray<OrchestrationV2ThreadShellJson>) {
55 let misses: Array<OrchestrationV2ThreadShellJson> = [];
56 for (const row of rows) {
57 if (encodedRows.has(row)) continue;
58 misses.push(row);
59 if (misses.length === ROWS_PER_CHUNK) {
60 yield* encodeMisses(misses);
61 misses = [];
62 }
63 }
64 if (misses.length > 0) yield* encodeMisses(misses);
65 return rows.map((row) => encodedRows.get(row));
66 });
67
68 const { snapshot } = stored;
69 const threads = yield* encodeRows(snapshot.threads);
70 const archivedThreads = yield* encodeRows(snapshot.archivedThreads);
71 if (hasWorked) yield* yieldToHost;
72 return yield* encodeEnvelope({
73 ...stored,
74 snapshot: { ...snapshot, threads, archivedThreads },
75 });
76 });
77}