1import type { OrchestrationV2ThreadDetailSnapshot, ThreadId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5
6import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
7import type { PreparedConnection } from "../connection/model.ts";
8import * as ManagedRelay from "../relay/managedRelay.ts";
9import type { RemoteEnvironmentRequestError } from "../rpc/http.ts";
10import {
11 executeAuthenticatedEnvironmentHttpRequest,
12 withOrchestrationProtocolHeader,
13} from "./environmentHttpAuth.ts";
14
15// Long enough for a slow but alive server to finish. On a cold open a timeout
16// makes the socket ask the same server for the same snapshot again, and older
17// turn pages have no fallback, so a short deadline only drops work. The socket
18// fallback is for setups where /api fails but /ws works, such as a proxy that
19// blocks /api. A dead server drops the socket session, which interrupts a
20// cold-open load. Older turn pages wait for this deadline.
21const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 20_000;
22
23/** Progressive history metadata returned by a bounded snapshot loader. */
24export type ThreadSnapshotHistoryMeta = {
25 readonly historyCursor: string | null;
26 readonly hasMoreHistory: boolean;
27 /** Max local turn ordinal from the full projection; optional on older servers. */
28 readonly latestLocalTurnOrdinal?: number | null;
29};
30
31/**
32 * Outcome of an HTTP thread-detail snapshot load.
33 *
34 * - `present`: snapshot body is available (seed projection, resume via socket).
35 * - `missing`: server definitively reported the thread does not exist (404).
36 * - `unavailable`: transport/timeout/5xx/etc.; fall back to the socket path.
37 */
38export type ThreadSnapshotLoadResult =
39 | {
40 readonly _tag: "present";
41 readonly snapshot: OrchestrationV2ThreadDetailSnapshot;
42 readonly history?: ThreadSnapshotHistoryMeta;
43 }
44 | { readonly _tag: "missing" }
45 | { readonly _tag: "unavailable" };
46
47/**
48 * Load a thread's detail snapshot over HTTP instead of embedding it in the
49 * WebSocket subscription's first frame. The response is gzip-compressible by
50 * the transport and keeps the (potentially multi-KB) snapshot off the socket.
51 */
52export const fetchEnvironmentThreadSnapshot = Effect.fn(
53 "clientRuntime.state.fetchEnvironmentThreadSnapshot",
54)(function* (input: {
55 readonly prepared: PreparedConnection;
56 readonly threadId: ThreadId;
57 readonly signer: Option.Option<ManagedRelay.ManagedRelayDpopSigner["Service"]>;
58 readonly remoteAuthorization?: Option.Option<
59 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization["Service"]
60 >;
61 readonly timeoutMs?: number;
62}) {
63 return yield* executeAuthenticatedEnvironmentHttpRequest({
64 ...input,
65 group: "orchestration",
66 method: "GET",
67 url: (urls) => urls.threadSnapshot({ params: { threadId: input.threadId } }),
68 timeoutMs: input.timeoutMs ?? DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS,
69 request: ({ client, headers }) =>
70 client.threadSnapshot({
71 params: { threadId: input.threadId },
72 headers: withOrchestrationProtocolHeader(headers),
73 }),
74 });
75});
76
77export type FetchEnvironmentThreadSnapshotError = RemoteEnvironmentRequestError;
78
79/**
80 * Loads a thread's detail snapshot over HTTP.
81 *
82 * Distinguishes a definitive missing thread (HTTP 404) from transient snapshot
83 * unavailability so the thread state machine can mark the thread deleted without
84 * opening a socket subscription, while still falling back to the socket when the
85 * HTTP path is merely unavailable.
86 */
87export class ThreadSnapshotLoader extends Context.Service<
88 ThreadSnapshotLoader,
89 {
90 readonly load: (
91 prepared: PreparedConnection,
92 threadId: ThreadId,
93 ) => Effect.Effect<ThreadSnapshotLoadResult>;
94 }
95>()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {}