1import { deepEqual } from "@/Diff";
2import { reconcileBlueGreen } from "@/Fly/bluegreen";
3import { makeMachineLeases } from "@/Fly/leases";
4import { alchemyMetadataKeys as keys } from "@/Fly/Metadata";
5import type { ReconcileReplicasInput } from "@/Fly/replicas";
6import { credentials } from "@distilled.cloud/fly-io/Credentials";
7import * as machines from "@distilled.cloud/fly-io/machines";
8import * as Clock from "effect/Clock";
9import * as Effect from "effect/Effect";
10import * as Schema from "effect/Schema";
11import * as HttpClient from "effect/http/HttpClient";
12import type * as HttpClientRequest from "effect/http/HttpClientRequest";
13import * as HttpClientResponse from "effect/http/HttpClientResponse";
14
15export const appName = "protocol-branches";
16export const candidateId = "controlled-candidate";
17export const ownership = {
18 [keys.stack]: "protocol-branches",
19 [keys.stage]: "pure",
20 [keys.id]: "Worker",
21 [keys.type]: "Fly.Machine",
22};
23export const metadata = {
24 ...ownership,
25 [keys.instance]: "controlled-instance",
26 [keys.fqn]: "protocol-branches/pure/Worker",
27 [keys.baseName]: "protocol-worker",
28};
29
30const config: machines.FlyMachineConfig = {
31 image: "registry.test/fixture:latest",
32 checks: { ready: { type: "http", port: 80, path: "/" } },
33 stop_config: { signal: "SIGTERM", timeout: "1s" },
34};
35const input: ReconcileReplicasInput = {
36 appName,
37 id: "Worker",
38 type: "Fly.Machine",
39 resourceInstanceId: metadata[keys.instance],
40 fqn: metadata[keys.fqn],
41 baseName: metadata[keys.baseName],
42 regions: ["ord"],
43 count: 1,
44 disks: [],
45 policy: {
46 bluegreen: true,
47 healthTimeoutMs: 5_000,
48 shutdown: { signal: "SIGTERM", timeout: "1s", timeoutMs: 1_000 },
49 },
50 buildConfig: ({ metadata, mounts }) => ({ ...config, metadata, mounts }),
51 configDrifted: (machine, { metadata, mounts }) =>
52 !deepEqual(machine.config, { ...config, metadata, mounts }),
53};
54
55export const reconcile = Effect.gen(function* () {
56 const leases = yield* makeMachineLeases(appName);
57 return yield* reconcileBlueGreen(input, ownership, metadata, leases);
58}).pipe(Effect.scoped);
59
60export const withControlledClient =
61 (client: HttpClient.HttpClient) =>
62 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
63 effect.pipe(
64 Effect.provideService(HttpClient.HttpClient, client),
65 Effect.provide(
66 credentials({
67 apiKey: "pure-fixture-only",
68 apiBaseUrl: "https://fly.test",
69 }),
70 ),
71 );
72
73export const reply = (
74 request: HttpClientRequest.HttpClientRequest,
75 body: unknown,
76 status = 200,
77) => HttpClientResponse.fromWeb(request, Response.json(body, { status }));
78
79const Body = Schema.Struct({
80 name: Schema.optional(Schema.String),
81 config: Schema.optional(Schema.toType(machines.FlyMachineConfig)),
82 metadata: Schema.optional(Schema.Record(Schema.String, Schema.String)),
83});
84
85export interface ProtocolEvent {
86 method: string;
87 path: string;
88 phase?: string;
89 visible?: boolean;
90}
91
92/** Controlled wire responses are algorithm evidence, never Fly platform observations. */
93export const protocolClient = (
94 options: {
95 conflict?: boolean;
96 hiddenLists?: number;
97 missingImageRef?: "image_ref" | "digest" | "repository";
98 } = {},
99) =>
100 Effect.sync(() => {
101 const events: ProtocolEvent[] = [];
102 let current: machines.Machine | undefined;
103 let hidden = options.hiddenLists ?? 0;
104 const client = HttpClient.make((request) =>
105 Effect.gen(function* () {
106 const now = yield* Clock.currentTimeMillis;
107 return yield* Effect.sync(() => {
108 const path = new URL(request.url).pathname;
109 const body = Schema.decodeUnknownSync(Body)(
110 request.body._tag === "Uint8Array"
111 ? JSON.parse(new TextDecoder().decode(request.body.body))
112 : {},
113 );
114 const event: ProtocolEvent = {
115 method: request.method,
116 path,
117 phase: body.metadata?.[keys.phase],
118 };
119 events.push(event);
120 if (path === `/v1/apps/${appName}/machines`) {
121 if (request.method === "GET") {
122 const visible = current !== undefined && hidden <= 0;
123 event.visible = visible;
124 if (current !== undefined && hidden > 0) hidden--;
125 return reply(request, visible ? [current] : []);
126 }
127 if (request.method === "POST") {
128 if (current)
129 throw new Error("Controller replayed candidate creation");
130 current = {
131 id: candidateId,
132 instance_id: "controlled-machine-version",
133 name: body.name,
134 region: "ord",
135 state: "started",
136 cordoned: true,
137 config: body.config,
138 checks: [{ name: "ready", status: "passing" }],
139 image_ref:
140 options.missingImageRef === "image_ref"
141 ? undefined
142 : {
143 registry: "registry.test",
144 repository:
145 options.missingImageRef === "repository"
146 ? undefined
147 : "fixture",
148 digest:
149 options.missingImageRef === "digest"
150 ? undefined
151 : `sha256:${"a".repeat(64)}`,
152 },
153 };
154 return options.conflict
155 ? reply(request, { error: "Machine name already exists" }, 409)
156 : reply(request, current);
157 }
158 }
159 if (path === `/v1/apps/${appName}/machines/${candidateId}/lease`) {
160 if (request.method === "POST")
161 return reply(request, {
162 data: {
163 nonce: "pure-lease",
164 expires_at: Math.floor(now / 1000) + 120,
165 },
166 });
167 if (request.method === "DELETE") return reply(request, {});
168 }
169 if (
170 path === `/v1/apps/${appName}/machines/${candidateId}` &&
171 request.method === "GET"
172 )
173 return current
174 ? reply(request, current)
175 : reply(request, { error: "not found" }, 404);
176 if (
177 path === `/v1/apps/${appName}/machines/${candidateId}/wait` &&
178 request.method === "GET"
179 )
180 return reply(request, {});
181 if (
182 path === `/v1/apps/${appName}/machines/${candidateId}/metadata` &&
183 request.method === "PATCH" &&
184 current
185 ) {
186 current = {
187 ...current,
188 config: { ...current.config, metadata: body.metadata },
189 };
190 return reply(request, {});
191 }
192 if (
193 path === `/v1/apps/${appName}/machines/${candidateId}/uncordon` &&
194 request.method === "POST" &&
195 current
196 ) {
197 current = { ...current, cordoned: false };
198 return reply(request, {});
199 }
200 throw new Error(
201 `Unexpected controlled request: ${request.method} ${path}`,
202 );
203 });
204 }),
205 );
206 return { client, events };
207 });