1import { EnvironmentHttpCommonError, OrchestrationV2ShellResumable } from "@t3tools/contracts";
2import * as Cause from "effect/Cause";
3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as Predicate from "effect/Predicate";
8import * as Schema from "effect/Schema";
9import { HttpClient, HttpClientError, type HttpClientResponse } from "effect/http";
10
11import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
12import type { PreparedConnection } from "../connection/model.ts";
13import * as ManagedRelay from "../relay/managedRelay.ts";
14import {
15 executeAuthenticatedEnvironmentHttpRequest,
16 withOrchestrationProtocolHeader,
17} from "./environmentHttpAuth.ts";
18import {
19 decodeShellSnapshotDeferringPullRequests,
20 type DeferredShellSnapshot,
21} from "./shellPullRequests.ts";
22
23export interface ShellSnapshotOptions {
24 readonly omitIdleSubagents?: boolean;
25 readonly omitOlderSettled?: boolean;
26 /** The cursor of held rows loaded with these omissions; the server may answer `resumable`. */
27 readonly resumeFrom?: number;
28}
29
30/** `projectIdentities` names the repository identities the snapshot carried. */
31export interface ShellHttpSnapshot extends DeferredShellSnapshot {
32 readonly projectIdentities?: string;
33}
34
35export type ShellHttpAnswer = ShellHttpSnapshot | OrchestrationV2ShellResumable;
36
37// On timeout the socket asks the same server for the same full snapshot, so a
38// short deadline only throws the first build away. The socket fallback is for
39// setups where /api fails but /ws works, such as a proxy that blocks /api.
40const DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS = 20_000;
41
42export const fetchEnvironmentShellSnapshot = Effect.fn(
43 "clientRuntime.state.fetchEnvironmentShellSnapshot",
44)(function* (
45 input: {
46 readonly prepared: PreparedConnection;
47 readonly signer: Option.Option<ManagedRelay.ManagedRelayDpopSigner["Service"]>;
48 readonly remoteAuthorization?: Option.Option<
49 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization["Service"]
50 >;
51 readonly timeoutMs?: number;
52 } & ShellSnapshotOptions,
53) {
54 const endpoint = {
55 query: {
56 ...(input.omitIdleSubagents === true ? { omitIdleSubagents: "1" } : {}),
57 ...(input.omitOlderSettled === true ? { omitOlderSettled: "1" } : {}),
58 ...(input.resumeFrom === undefined ? {} : { resumeFrom: String(input.resumeFrom) }),
59 },
60 };
61 return yield* executeAuthenticatedEnvironmentHttpRequest({
62 ...input,
63 group: "orchestration",
64 method: "GET",
65 url: (urls) => urls.shellSnapshot(endpoint),
66 timeoutMs: input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS,
67 request: ({ client, headers }) =>
68 client
69 .shellSnapshot({
70 headers: withOrchestrationProtocolHeader(headers),
71 ...endpoint,
72 responseMode: "response-only",
73 })
74 .pipe(Effect.flatMap(decodeShellSnapshotResponse)),
75 });
76});
77
78const decodeEnvironmentHttpCommonError = Schema.decodeUnknownEffect(
79 Schema.toCodecJson(EnvironmentHttpCommonError),
80);
81
82const isShellResumable = Schema.is(OrchestrationV2ShellResumable);
83
84const decodeShellSnapshotResponse = (
85 response: HttpClientResponse.HttpClientResponse,
86): Effect.Effect<
87 ShellHttpAnswer,
88 HttpClientError.HttpClientError | Schema.SchemaError | EnvironmentHttpCommonError
89> => {
90 if (response.status === 200) {
91 return response.json.pipe(
92 Effect.flatMap((json): Effect.Effect<ShellHttpAnswer, Schema.SchemaError> => {
93 if (isShellResumable(json)) return Effect.succeed(json);
94 const projectIdentities =
95 Predicate.isObject(json) && typeof json.projectIdentities === "string"
96 ? json.projectIdentities
97 : undefined;
98 return decodeShellSnapshotDeferringPullRequests(json).pipe(
99 Effect.map((snapshot) =>
100 projectIdentities === undefined ? snapshot : { ...snapshot, projectIdentities },
101 ),
102 );
103 }),
104 );
105 }
106 const statusError = new HttpClientError.HttpClientError({
107 reason: new HttpClientError.StatusCodeError({ request: response.request, response }),
108 });
109 return response.json.pipe(
110 Effect.flatMap(decodeEnvironmentHttpCommonError),
111 Effect.matchEffect({
112 onFailure: () => Effect.fail(statusError),
113 onSuccess: Effect.fail,
114 }),
115 );
116};
117
118export class ShellSnapshotLoader extends Context.Service<
119 ShellSnapshotLoader,
120 {
121 readonly load: (
122 prepared: PreparedConnection,
123 options?: ShellSnapshotOptions,
124 ) => Effect.Effect<Option.Option<ShellHttpAnswer>>;
125 }
126>()("@t3tools/client-runtime/state/shellSnapshotHttp/ShellSnapshotLoader") {}
127
128export const layer: Layer.Layer<ShellSnapshotLoader, never, HttpClient.HttpClient> = Layer.effect(
129 ShellSnapshotLoader,
130 Effect.gen(function* () {
131 const httpClient = yield* HttpClient.HttpClient;
132 // Resolve the DPoP signer optionally: it is only needed for relay/DPoP
133 // connections, so the loader must not hard-require it.
134 const signer = yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner);
135 const remoteAuthorization = yield* Effect.serviceOption(
136 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization,
137 );
138 return ShellSnapshotLoader.of({
139 load: Effect.fn("ShellSnapshotLoader.load")(function (prepared, options) {
140 return fetchEnvironmentShellSnapshot({
141 prepared,
142 signer,
143 remoteAuthorization,
144 ...options,
145 }).pipe(
146 Effect.map(Option.some<ShellHttpAnswer>),
147 Effect.provideService(HttpClient.HttpClient, httpClient),
148 Effect.catchCause((cause) =>
149 Effect.logWarning(
150 "Could not load the environment shell snapshot over HTTP; using the socket snapshot instead.",
151 ).pipe(
152 Effect.annotateLogs({ cause: Cause.pretty(cause) }),
153 Effect.as(Option.none<ShellHttpAnswer>()),
154 ),
155 ),
156 );
157 }),
158 });
159 }),
160);