1import * as machines from "@distilled.cloud/fly-io/machines";
2import * as Retry from "@distilled.cloud/fly-io/Retry";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as HttpClient from "effect/http/HttpClient";
7import { throughProxy } from "./transport.ts";
8
9const rolloutKeys = new Set([
10 "alchemy.generation",
11 "alchemy.workload",
12 "alchemy.sequence",
13 "alchemy.count",
14 "alchemy.image",
15 "alchemy.phase",
16 "alchemy.deployment-protocol",
17 "alchemy.readiness-roles",
18 "alchemy.readiness-role",
19 "alchemy.predecessors",
20 "alchemy.idle-policy-restored",
21 "alchemy.checked-instance",
22 "alchemy.min-secrets-version",
23]);
24
25class LegacyWriterLeaseInvalid extends Data.TaggedError(
26 "LegacyWriterLeaseInvalid",
27)<{
28 machineId: string;
29}> {}
30
31/** Explicit legacy-protocol writer model, not an old Alchemy binary or global fence. */
32export const writeLegacyProtocol = (
33 appName: string,
34 machineId: string,
35 runtimeTimeoutMs?: number,
36) =>
37 Effect.gen(function* () {
38 const target = { app_name: appName, machine_id: machineId };
39 const lease = yield* machines
40 .createMachineLease({
41 ...target,
42 description: "Live legacy-protocol writer model",
43 ttl: 120,
44 })
45 .pipe(Retry.none, Effect.timeout("30 seconds"));
46 const nonce = lease.data?.nonce;
47 if (!nonce) return yield* new LegacyWriterLeaseInvalid({ machineId });
48 try {
49 const current = yield* machines.getMachine(target);
50 const config = current.config!;
51 return yield* machines
52 .updateMachine({
53 ...target,
54 lease_nonce: nonce,
55 current_version: current.instance_id,
56 config: {
57 ...config,
58 stop_config: undefined,
59 env: {
60 ...Object.fromEntries(
61 Object.entries(config.env ?? {}).filter(
62 ([key]) => key !== "ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS",
63 ),
64 ),
65 ...(runtimeTimeoutMs === undefined
66 ? {}
67 : {
68 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: String(runtimeTimeoutMs),
69 }),
70 },
71 metadata: Object.fromEntries(
72 Object.entries(config.metadata ?? {}).filter(
73 ([key]) => !rolloutKeys.has(key),
74 ),
75 ),
76 },
77 })
78 .pipe(Retry.none, Effect.timeout("45 seconds"));
79 } finally {
80 yield* machines
81 .machinesReleaseLease({ ...target, lease_nonce: nonce })
82 .pipe(
83 Retry.none,
84 Effect.catchTag("NotFound", () => Effect.void),
85 Effect.timeout("30 seconds"),
86 );
87 }
88 });
89
90export interface StopRequest {
91 machineId: string;
92 signal: string | undefined;
93 timeout: string | undefined;
94}
95
96/** Observe only stop policy fields; forward every original request and response unchanged. */
97export const observeStops = (
98 endpoint: () => string | undefined,
99 record: (request: StopRequest) => void,
100) =>
101 Layer.effect(
102 HttpClient.HttpClient,
103 Effect.gen(function* () {
104 const client = yield* HttpClient.HttpClient;
105 return client.pipe(
106 HttpClient.tapRequest((request) =>
107 Effect.sync(() => {
108 const target = request.url.match(/\/machines\/([^/]+)\/stop$/);
109 if (request.method !== "POST" || !target) return;
110 if (request.body._tag !== "Uint8Array") {
111 throw new Error(
112 "Expected the native SDK's JSON stop request body",
113 );
114 }
115 const body = JSON.parse(
116 new TextDecoder().decode(request.body.body),
117 );
118 record({
119 machineId: target[1]!,
120 signal: typeof body.signal === "string" ? body.signal : undefined,
121 timeout:
122 typeof body.timeout === "string" ? body.timeout : undefined,
123 });
124 }),
125 ),
126 );
127 }),
128 ).pipe(Layer.provideMerge(throughProxy(endpoint)));