1import * as Fly from "@/Fly";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as HttpClient from "effect/http/HttpClient";
5import * as HttpClientRequest from "effect/http/HttpClientRequest";
6import * as http from "node:http";
7import * as https from "node:https";
8
9export interface TransportEvent {
10 sequence: number;
11 stage:
12 | "request"
13 | "completed"
14 | "forwarded"
15 | "dropped"
16 | "cut"
17 | "held"
18 | "upstream-error";
19 method: string;
20 path: string;
21 machineId?: string;
22 name?: string;
23 phase?: string;
24 status?: number;
25 image?: string;
26 digest?: string;
27 instanceId?: string;
28 state?: string;
29 cordoned?: boolean;
30 minSecretsVersion?: number;
31 secretsVersion?: number;
32 checks?: { name?: string; status?: string }[];
33}
34
35export interface FaultRule {
36 match: (event: TransportEvent) => boolean;
37 action: "drop-response" | "cut-request" | "hold-response";
38 remaining: number;
39}
40
41/** No bodies, authorization headers, secret values, or lease nonces enter the journal. */
42export const transportProxy = (upstream = "https://api.machines.dev") =>
43 Effect.acquireRelease(
44 Effect.callback<
45 {
46 url: string;
47 events: TransportEvent[];
48 arm: (rule: FaultRule) => void;
49 clear: () => void;
50 release: () => void;
51 dropHeld: () => void;
52 wait: (
53 match: (event: TransportEvent) => boolean,
54 ) => Effect.Effect<TransportEvent, Error>;
55 close: () => void;
56 },
57 Error
58 >((resume) => {
59 const origin = new URL(upstream);
60 if (origin.protocol !== "https:") {
61 resume(
62 Effect.fail(
63 new Error("The fault proxy requires a real HTTPS upstream"),
64 ),
65 );
66 return;
67 }
68 const events: TransportEvent[] = [];
69 const listeners = new Set<(event: TransportEvent) => void>();
70 const pending = new Set<http.ClientRequest>();
71 const held = new Map<() => void, () => void>();
72 const rules: FaultRule[] = [];
73 let sequence = 0;
74 const record = (event: TransportEvent) => {
75 events.push(event);
76 for (const listener of listeners) listener(event);
77 };
78 const server = http.createServer((incoming, outgoing) => {
79 const chunks: Buffer[] = [];
80 incoming.on("data", (chunk: Buffer) => chunks.push(chunk));
81 incoming.on("error", () => outgoing.destroy());
82 incoming.on("end", () => {
83 const body = Buffer.concat(chunks);
84 const path = new URL(incoming.url!, origin).pathname;
85 const event: TransportEvent = {
86 sequence: ++sequence,
87 stage: "request",
88 method: incoming.method!,
89 path,
90 machineId: path.match(/\/machines\/([^/]+)/)?.[1],
91 };
92 // Only allowlisted identity/phase fields survive request inspection.
93 if (
94 body.length &&
95 incoming.headers["content-type"]?.includes("json")
96 ) {
97 try {
98 const value = JSON.parse(body.toString());
99 if (typeof value.name === "string") event.name = value.name;
100 if (typeof value.config?.image === "string")
101 event.image = value.config.image;
102 if (typeof value.min_secrets_version === "number")
103 event.minSecretsVersion = value.min_secrets_version;
104 const phase =
105 value.metadata?.["alchemy.phase"] ??
106 value.config?.metadata?.["alchemy.phase"];
107 if (typeof phase === "string") event.phase = phase;
108 } catch {
109 // Forward malformed payloads unchanged; the real API decides their validity.
110 }
111 }
112 record(event);
113 const rule = rules.find(
114 (rule) => rule.remaining > 0 && rule.match(event),
115 );
116 if (rule) rule.remaining--;
117 if (rule?.action === "cut-request") {
118 record({ ...event, stage: "cut" });
119 incoming.socket.destroy();
120 return;
121 }
122 const request = https.request(
123 new URL(incoming.url!, origin),
124 {
125 method: incoming.method,
126 headers: {
127 ...incoming.headers,
128 host: origin.host,
129 connection: "close",
130 "accept-encoding": "identity",
131 },
132 },
133 (response) => {
134 const parts: Buffer[] = [];
135 response.on("data", (chunk: Buffer) => parts.push(chunk));
136 response.on("error", () => outgoing.destroy());
137 response.on("end", () => {
138 pending.delete(request);
139 const bytes = Buffer.concat(parts);
140 const completed: TransportEvent = {
141 ...event,
142 status: response.statusCode,
143 };
144 if (response.statusCode! >= 200 && response.statusCode! < 300) {
145 try {
146 const value = JSON.parse(bytes.toString());
147 if (
148 /\/machines(?:\/[^/]+)?$/.test(path) &&
149 !Array.isArray(value)
150 ) {
151 if (typeof value.id === "string")
152 completed.machineId = value.id;
153 if (typeof value.instance_id === "string")
154 completed.instanceId = value.instance_id;
155 if (typeof value.image_ref?.digest === "string")
156 completed.digest = value.image_ref.digest;
157 if (typeof value.state === "string")
158 completed.state = value.state;
159 if (typeof value.cordoned === "boolean")
160 completed.cordoned = value.cordoned;
161 if (Array.isArray(value.checks)) {
162 completed.checks = value.checks.map(
163 (check: { name?: string; status?: string }) => ({
164 name:
165 typeof check.name === "string"
166 ? check.name
167 : undefined,
168 status:
169 typeof check.status === "string"
170 ? check.status
171 : undefined,
172 }),
173 );
174 }
175 }
176 if (/\/secrets(?:\/[^/]+)?$/.test(path)) {
177 const version = value.version ?? value.Version;
178 if (typeof version === "number")
179 completed.secretsVersion = version;
180 }
181 } catch {
182 // Response bytes are never altered to accommodate the observer.
183 }
184 }
185 record({ ...completed, stage: "completed" });
186 const forward = () => {
187 const headers = { ...response.headers };
188 delete headers["transfer-encoding"];
189 delete headers.connection;
190 headers["content-length"] = String(bytes.length);
191 outgoing.writeHead(response.statusCode!, headers);
192 outgoing.end(bytes);
193 record({ ...completed, stage: "forwarded" });
194 };
195 if (rule?.action === "drop-response") {
196 record({ ...completed, stage: "dropped" });
197 incoming.socket.destroy();
198 } else if (rule?.action === "hold-response") {
199 held.set(forward, () => {
200 record({ ...completed, stage: "dropped" });
201 incoming.socket.destroy();
202 });
203 record({ ...completed, stage: "held" });
204 } else forward();
205 });
206 },
207 );
208 pending.add(request);
209 request.setTimeout(90_000, () => request.destroy());
210 request.on("error", () => {
211 pending.delete(request);
212 record({ ...event, stage: "upstream-error" });
213 incoming.socket.destroy();
214 });
215 request.end(body);
216 });
217 });
218 const close = () => {
219 rules.length = 0;
220 held.clear();
221 for (const request of pending) request.destroy();
222 server.closeAllConnections();
223 server.close();
224 };
225 server.on("error", (error) => resume(Effect.fail(error)));
226 server.listen(0, "127.0.0.1", () => {
227 const address = server.address();
228 if (!address || typeof address === "string") {
229 resume(Effect.fail(new Error("Missing proxy listen address")));
230 return;
231 }
232 resume(
233 Effect.succeed({
234 url: `http://127.0.0.1:${address.port}`,
235 events,
236 arm: (rule) => rules.push({ ...rule }),
237 clear: () => {
238 rules.length = 0;
239 },
240 release: () => {
241 for (const forward of held.keys()) forward();
242 held.clear();
243 },
244 dropHeld: () => {
245 for (const drop of held.values()) drop();
246 held.clear();
247 },
248 wait: (match) =>
249 Effect.callback<TransportEvent>((resume) => {
250 const found = events.find(match);
251 if (found) {
252 resume(Effect.succeed(found));
253 return;
254 }
255 const listener = (event: TransportEvent) => {
256 if (match(event)) {
257 listeners.delete(listener);
258 resume(Effect.succeed(event));
259 }
260 };
261 listeners.add(listener);
262 return Effect.sync(() => listeners.delete(listener));
263 }).pipe(
264 Effect.timeout("120 seconds"),
265 Effect.mapError(
266 () =>
267 new Error(
268 "Real API operation barrier was not reached within 120 seconds",
269 ),
270 ),
271 ),
272 close,
273 }),
274 );
275 });
276 return Effect.sync(close);
277 }),
278 (proxy) =>
279 Effect.forEach(
280 Array.from(
281 { length: Math.ceil(proxy.events.length / 50) },
282 (_, index) => index,
283 ),
284 (index) =>
285 Effect.logInfo("Fly transport journal", {
286 endpoint: proxy.url,
287 offset: index * 50,
288 events: proxy.events.slice(index * 50, (index + 1) * 50),
289 }),
290 { discard: true },
291 ).pipe(Effect.andThen(Effect.sync(proxy.close))),
292 );
293
294/** Adapt only transport routing; the original SDK, provider and real state remain intact. */
295export const throughProxy = (endpoint: () => string | undefined) =>
296 Layer.effect(
297 HttpClient.HttpClient,
298 Effect.gen(function* () {
299 const client = yield* HttpClient.HttpClient;
300 return HttpClient.mapRequest(client, (request) => {
301 const url = endpoint();
302 return url && request.url.startsWith("https://api.machines.dev/")
303 ? HttpClientRequest.setUrl(
304 request,
305 request.url.replace("https://api.machines.dev", url),
306 )
307 : request;
308 });
309 }),
310 ).pipe(Layer.provideMerge(Fly.providers()));