1import type { ThreadId } from "@t3tools/contracts";
2import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
3import * as Cause from "effect/Cause";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import { HttpClient } from "effect/http";
8
9import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
10import type { PreparedConnection } from "../connection/model.ts";
11import * as ManagedRelay from "../relay/managedRelay.ts";
12import {
13 executeAuthenticatedEnvironmentHttpRequest,
14 withOrchestrationProtocolHeader,
15} from "./environmentHttpAuth.ts";
16import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts";
17
18// Same cold-open budget as the full snapshot path; bounded payloads should fit.
19const DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000;
20
21/**
22 * Load a bounded recent-window thread snapshot over HTTP. Opts into compact
23 * turnItems and restores them, so callers always see the full bounded shape.
24 * Older servers ignore the query and send the full shape.
25 */
26export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn(
27 "clientRuntime.state.fetchEnvironmentBoundedThreadSnapshot",
28)(function* (input: {
29 readonly prepared: PreparedConnection;
30 readonly threadId: ThreadId;
31 readonly signer: Option.Option<ManagedRelay.ManagedRelayDpopSigner["Service"]>;
32 readonly remoteAuthorization?: Option.Option<
33 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization["Service"]
34 >;
35 readonly timeoutMs?: number;
36}) {
37 const endpoint = {
38 params: { threadId: input.threadId },
39 query: { compactTurnItems: "1" },
40 };
41 return yield* executeAuthenticatedEnvironmentHttpRequest({
42 ...input,
43 group: "orchestration",
44 method: "GET",
45 url: (urls) => urls.threadBoundedSnapshot(endpoint),
46 timeoutMs: input.timeoutMs ?? DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS,
47 request: ({ client, headers }) =>
48 client.threadBoundedSnapshot({
49 ...endpoint,
50 headers: withOrchestrationProtocolHeader(headers),
51 }),
52 }).pipe(
53 // Drop the marker with the restore so nothing can restore twice.
54 Effect.map(({ turnItemsOmitLocalVisible, ...snapshot }) => ({
55 ...snapshot,
56 projection: boundedSnapshotProjection({ ...snapshot, turnItemsOmitLocalVisible }),
57 })),
58 );
59});
60
61/**
62 * Shared ThreadSnapshotLoader for clients that render progressive history.
63 * Seeds from the bounded HTTP snapshot when available. Older servers that lack
64 * the bounded route (plain/generic 404) fall back to the existing full HTTP
65 * thread snapshot. Structured EnvironmentResourceNotFoundError from either
66 * endpoint still means missing. Transient failures report `unavailable` so the
67 * socket path remains a last resort for connectivity issues.
68 */
69export const layer: Layer.Layer<
70 ThreadSnapshotLoader.ThreadSnapshotLoader,
71 never,
72 HttpClient.HttpClient
73> = Layer.effect(
74 ThreadSnapshotLoader.ThreadSnapshotLoader,
75 Effect.gen(function* () {
76 const httpClient = yield* HttpClient.HttpClient;
77 const signer = yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner);
78 const remoteAuthorization = yield* Effect.serviceOption(
79 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization,
80 );
81 return ThreadSnapshotLoader.ThreadSnapshotLoader.of({
82 load: (prepared: PreparedConnection, threadId: ThreadId) => {
83 const loadFullFallback = ThreadSnapshotLoader.fetchEnvironmentThreadSnapshot({
84 prepared,
85 threadId,
86 signer,
87 remoteAuthorization,
88 }).pipe(
89 Effect.map((snapshot): ThreadSnapshotLoader.ThreadSnapshotLoadResult => ({
90 _tag: "present",
91 snapshot,
92 })),
93 Effect.provideService(HttpClient.HttpClient, httpClient),
94 Effect.catchTags({
95 EnvironmentResourceNotFoundError: () =>
96 Effect.logDebug(
97 "Full thread snapshot not found over HTTP after bounded fallback; treating the thread as deleted.",
98 ).pipe(
99 Effect.annotateLogs({ threadId }),
100 Effect.as({
101 _tag: "missing",
102 } satisfies ThreadSnapshotLoader.ThreadSnapshotLoadResult),
103 ),
104 }),
105 Effect.catchCause((cause) =>
106 Effect.logWarning(
107 "Could not load the full thread snapshot over HTTP after bounded fallback; using the socket snapshot instead.",
108 ).pipe(
109 Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }),
110 Effect.as({
111 _tag: "unavailable",
112 } satisfies ThreadSnapshotLoader.ThreadSnapshotLoadResult),
113 ),
114 ),
115 );
116
117 return fetchEnvironmentBoundedThreadSnapshot({
118 prepared,
119 threadId,
120 signer,
121 remoteAuthorization,
122 }).pipe(
123 Effect.map((bounded): ThreadSnapshotLoader.ThreadSnapshotLoadResult => ({
124 _tag: "present",
125 snapshot: {
126 snapshotSequence: bounded.snapshotSequence,
127 projection: bounded.projection,
128 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
129 },
130 history: {
131 historyCursor: bounded.historyCursor,
132 hasMoreHistory: bounded.hasMoreHistory,
133 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
134 },
135 })),
136 Effect.provideService(HttpClient.HttpClient, httpClient),
137 Effect.catchTags({
138 EnvironmentResourceNotFoundError: () =>
139 Effect.logDebug(
140 "Bounded thread snapshot not found over HTTP; treating the thread as deleted.",
141 ).pipe(
142 Effect.annotateLogs({ threadId }),
143 Effect.as({
144 _tag: "missing",
145 } satisfies ThreadSnapshotLoader.ThreadSnapshotLoadResult),
146 ),
147 RemoteEnvironmentAuthInvalidJsonError: (error) =>
148 Effect.logDebug(
149 "Bounded thread snapshot returned an invalid response; trying the full HTTP thread snapshot for an older server.",
150 ).pipe(
151 Effect.annotateLogs({ threadId, cause: error.message }),
152 Effect.andThen(loadFullFallback),
153 ),
154 RemoteEnvironmentAuthUndeclaredStatusError: (error) =>
155 error.status === 404
156 ? Effect.logDebug(
157 "Bounded thread snapshot route was not found; trying the full HTTP thread snapshot for an older server.",
158 ).pipe(Effect.annotateLogs({ threadId }), Effect.andThen(loadFullFallback))
159 : Effect.fail(error),
160 }),
161 Effect.catchCause((cause) =>
162 Effect.logWarning(
163 "Could not load the bounded thread snapshot over HTTP; using the socket snapshot instead.",
164 ).pipe(
165 Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }),
166 Effect.as({
167 _tag: "unavailable",
168 } satisfies ThreadSnapshotLoader.ThreadSnapshotLoadResult),
169 ),
170 ),
171 );
172 },
173 });
174 }),
175);