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
18const DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000;
19
20export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn(
21 "clientRuntime.state.fetchEnvironmentBoundedThreadSnapshot",
22)(function* (input: {
23 readonly prepared: PreparedConnection;
24 readonly threadId: ThreadId;
25 readonly signer: Option.Option<ManagedRelay.ManagedRelayDpopSigner["Service"]>;
26 readonly remoteAuthorization?: Option.Option<
27 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization["Service"]
28 >;
29 readonly timeoutMs?: number;
30}) {
31 const endpoint = {
32 params: { threadId: input.threadId },
33 query: {
34 compactTurnItems: "1",
35 dedupeBodies: "1",
36 compactIds: "1",
37 deferWork: "1",
38 },
39 };
40 return yield* executeAuthenticatedEnvironmentHttpRequest({
41 ...input,
42 group: "orchestration",
43 method: "GET",
44 url: (urls) => urls.threadBoundedSnapshot(endpoint),
45 timeoutMs: input.timeoutMs ?? DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS,
46 request: ({ client, headers }) =>
47 client.threadBoundedSnapshot({
48 ...endpoint,
49 headers: withOrchestrationProtocolHeader(headers),
50 }),
51 }).pipe(
52 Effect.map(({ turnItemsOmitLocalVisible, ...snapshot }) => ({
53 ...snapshot,
54 projection: boundedSnapshotProjection({ ...snapshot, turnItemsOmitLocalVisible }),
55 })),
56 );
57});
58
59/**
60 * A structured EnvironmentResourceNotFoundError means the thread is missing. Any other failure
61 * leaves the snapshot to the socket.
62 */
63export const layer: Layer.Layer<
64 ThreadSnapshotLoader.ThreadSnapshotLoader,
65 never,
66 HttpClient.HttpClient
67> = Layer.effect(
68 ThreadSnapshotLoader.ThreadSnapshotLoader,
69 Effect.gen(function* () {
70 const httpClient = yield* HttpClient.HttpClient;
71 const signer = yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner);
72 const remoteAuthorization = yield* Effect.serviceOption(
73 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization,
74 );
75 return ThreadSnapshotLoader.ThreadSnapshotLoader.of({
76 load: Effect.fn("ThreadSnapshotLoader.load")(function (
77 prepared: PreparedConnection,
78 threadId: ThreadId,
79 ) {
80 return fetchEnvironmentBoundedThreadSnapshot({
81 prepared,
82 threadId,
83 signer,
84 remoteAuthorization,
85 }).pipe(
86 Effect.map((bounded): ThreadSnapshotLoader.ThreadSnapshotLoadResult => ({
87 _tag: "present",
88 snapshot: {
89 snapshotSequence: bounded.snapshotSequence,
90 projection: bounded.projection,
91 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
92 },
93 history: {
94 historyCursor: bounded.historyCursor,
95 hasMoreHistory: bounded.hasMoreHistory,
96 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
97 },
98 })),
99 Effect.provideService(HttpClient.HttpClient, httpClient),
100 Effect.catchTags({
101 EnvironmentResourceNotFoundError: () =>
102 Effect.logDebug(
103 "Bounded thread snapshot not found over HTTP; treating the thread as deleted.",
104 ).pipe(
105 Effect.annotateLogs({ threadId }),
106 Effect.as({
107 _tag: "missing",
108 } satisfies ThreadSnapshotLoader.ThreadSnapshotLoadResult),
109 ),
110 }),
111 Effect.catchCause((cause) =>
112 Effect.logWarning(
113 "Could not load the bounded thread snapshot over HTTP; using the socket snapshot instead.",
114 ).pipe(
115 Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }),
116 Effect.as({
117 _tag: "unavailable",
118 } satisfies ThreadSnapshotLoader.ThreadSnapshotLoadResult),
119 ),
120 ),
121 );
122 }),
123 });
124 }),
125);