1import type {
2 OrchestrationV2ThreadProjection,
3 OrchestrationV2ThreadStreamItem,
4} from "@t3tools/contracts";
5
6import { omitLocalVisibleTurnItems } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
7
8import {
9 buildBoundedThreadProjection,
10 type BoundedProjectionResult,
11} from "./threadHistoryPaging.ts";
12import { projectThreadProjectionForWire } from "./WireProjection.ts";
13
14/** Maximum number of reducer applications allowed during a thread resume. */
15export const THREAD_RESUME_MAX_REPLAY_EVENTS = 128;
16
17/** Maximum encoded event JSON allowed during a thread resume. */
18export const THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES = 1_048_576;
19
20/**
21 * Maximum aggregate persisted payload JSON decoded for one thread resume.
22 * Keep the original preflight budget: decoded objects and temporary strings
23 * cost more than their stored JSON, even if wire projection shrinks them later.
24 */
25export const THREAD_RESUME_MAX_RAW_PAYLOAD_BYTES = 1_048_576;
26
27/** Resource-safety check applied before persisted replay payloads are decoded. */
28export function isThreadReplayRawPayloadSafe(rawPayloadBytes: number): boolean {
29 return rawPayloadBytes <= THREAD_RESUME_MAX_RAW_PAYLOAD_BYTES;
30}
31
32/** Encoded payload cost after events have been projected for the wire. */
33export function threadReplayEncodedBytes(items: ReadonlyArray<unknown>): number {
34 let total = 0;
35 for (const item of items) {
36 const encoded = JSON.stringify(item);
37 total += Buffer.byteLength(encoded ?? "", "utf8");
38 }
39 return total;
40}
41
42type ThreadSnapshotStreamItem = Extract<
43 OrchestrationV2ThreadStreamItem,
44 { readonly kind: "snapshot" }
45>;
46
47/**
48 * Bounded snapshot fields shared by the HTTP route and socket fallbacks. Only
49 * clients that opted in get compact `turnItems` and the marker; everyone else
50 * gets the unchanged representation.
51 */
52export function boundedSnapshotResponseFields(input: {
53 readonly bounded: BoundedProjectionResult;
54 readonly compactTurnItems: boolean;
55}) {
56 const { bounded } = input;
57 const compactProjection = input.compactTurnItems
58 ? omitLocalVisibleTurnItems(bounded.projection)
59 : null;
60 return {
61 projection: compactProjection ?? bounded.projection,
62 historyCursor: bounded.historyCursor,
63 hasMoreHistory: bounded.hasMoreHistory,
64 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
65 payloadBudgetExceeded: bounded.payloadBudgetExceeded,
66 ...(compactProjection === null ? {} : { turnItemsOmitLocalVisible: true as const }),
67 };
68}
69
70/** Build the same bounded, wire-projected snapshot for every socket fallback path. */
71export function buildBoundedThreadStreamSnapshot(input: {
72 readonly snapshotSequence: number;
73 readonly projection: OrchestrationV2ThreadProjection;
74 readonly compactTurnItems?: boolean;
75}): ThreadSnapshotStreamItem {
76 const bounded = buildBoundedThreadProjection({
77 snapshotSequence: input.snapshotSequence,
78 projection: projectThreadProjectionForWire(input.projection),
79 });
80 return {
81 kind: "snapshot",
82 snapshotSequence: input.snapshotSequence,
83 ...boundedSnapshotResponseFields({
84 bounded,
85 compactTurnItems: input.compactTurnItems === true,
86 }),
87 };
88}
89
90export type ThreadResumePlan =
91 | {
92 readonly mode: "replay";
93 readonly afterSequence: number;
94 readonly throughSequence: number;
95 }
96 | { readonly mode: "snapshot" };
97
98/**
99 * Decide whether a thread subscription should replay the event gap after the
100 * client's cursor or send a fresh snapshot instead.
101 *
102 * A client cursor above the high water mark is stale or invalid. Event count
103 * limits reducer churn while encoded bytes limit a small number of large
104 * updates. Either excess is cheaper to replace with one current snapshot.
105 */
106export function decideThreadResume(input: {
107 readonly afterSequence: number;
108 readonly highWater: number;
109 readonly replayEventCount: number;
110 readonly replayEncodedBytes: number;
111}): ThreadResumePlan {
112 if (
113 input.afterSequence > input.highWater ||
114 input.replayEventCount > THREAD_RESUME_MAX_REPLAY_EVENTS ||
115 input.replayEncodedBytes > THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES
116 ) {
117 return { mode: "snapshot" };
118 }
119 return {
120 mode: "replay",
121 afterSequence: input.afterSequence,
122 throughSequence: input.highWater,
123 };
124}