/** Maximum number of reducer applications allowed during a thread resume. */export const THREAD_RESUME_MAX_REPLAY_EVENTS = 128;/** Maximum encoded event JSON allowed during a thread resume. */export const THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES = 1_048_576;/** Encoded payload cost after events have been projected for the wire. */export function threadReplayEncodedBytes(items: ReadonlyArray<unknown>): number { let total = 0; for (const item of items) { const encoded = JSON.stringify(item); total += Buffer.byteLength(encoded ?? "", "utf8"); } return total;}export type ThreadResumePlan = | { readonly mode: "replay"; readonly afterSequence: number; readonly throughSequence: number; } | { readonly mode: "snapshot" };/** * Decide whether a thread subscription should replay the event gap after the * client's cursor or send a fresh snapshot instead. * * A client cursor above the high water mark is stale or invalid. Event count * limits reducer churn while encoded bytes limit a small number of large * updates. Either excess is cheaper to replace with one current snapshot. */export function decideThreadResume(input: { readonly afterSequence: number; readonly highWater: number; readonly replayEventCount: number; readonly replayEncodedBytes: number;}): ThreadResumePlan { if ( input.afterSequence > input.highWater || input.replayEventCount > THREAD_RESUME_MAX_REPLAY_EVENTS || input.replayEncodedBytes > THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES ) { return { mode: "snapshot" }; } return { mode: "replay", afterSequence: input.afterSequence, throughSequence: input.highWater, };}
/** Maximum number of reducer applications allowed during a thread resume. */
export const THREAD_RESUME_MAX_REPLAY_EVENTS = 128;
/** Maximum encoded event JSON allowed during a thread resume. */
export const THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES = 1_048_576;
/** Encoded payload cost after events have been projected for the wire. */
export function threadReplayEncodedBytes(items: ReadonlyArray<unknown>): number {
let total = 0;
for (const item of items) {
const encoded = JSON.stringify(item);
total += Buffer.byteLength(encoded ?? "", "utf8");
}
return total;
}
export type ThreadResumePlan =
| {
readonly mode: "replay";
readonly afterSequence: number;
readonly throughSequence: number;
}
| { readonly mode: "snapshot" };
/**
* Decide whether a thread subscription should replay the event gap after the
* client's cursor or send a fresh snapshot instead.
*
* A client cursor above the high water mark is stale or invalid. Event count
* limits reducer churn while encoded bytes limit a small number of large
* updates. Either excess is cheaper to replace with one current snapshot.
*/
export function decideThreadResume(input: {
readonly afterSequence: number;
readonly highWater: number;
readonly replayEventCount: number;
readonly replayEncodedBytes: number;
}): ThreadResumePlan {
if (
input.afterSequence > input.highWater ||
input.replayEventCount > THREAD_RESUME_MAX_REPLAY_EVENTS ||
input.replayEncodedBytes > THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES
) {
return { mode: "snapshot" };
}
return {
mode: "replay",
afterSequence: input.afterSequence,
throughSequence: input.highWater,
};
}