1import {
2 EnvironmentId,
3 ORCHESTRATION_PROTOCOL_VERSION,
4 ExecutionEnvironmentDescriptor,
5 WS_METHODS,
6} from "@t3tools/contracts";
7import { describe, expect, it } from "@effect/vitest";
8import * as Effect from "effect/Effect";
9import * as Layer from "effect/Layer";
10import * as Option from "effect/Option";
11import * as Schema from "effect/Schema";
12import * as SubscriptionRef from "effect/SubscriptionRef";
13import * as HttpClient from "effect/http/HttpClient";
14import * as HttpClientResponse from "effect/http/HttpClientResponse";
15import * as Socket from "effect/socket/Socket";
16
17import type { ConnectionCatalogEntry } from "./catalog.ts";
18import { orchestrationProtocolCompatibilityError } from "./compatibility.ts";
19import { PrimaryConnectionTarget } from "./model.ts";
20import { updateOutdatedHost } from "./outdatedHostUpdate.ts";
21import * as EnvironmentRegistry from "./registry.ts";
22import * as ConnectionResolver from "./resolver.ts";
23import * as RelayEnvironmentDiscovery from "../relay/discovery.ts";
24
25const TARGET = new PrimaryConnectionTarget({
26 environmentId: EnvironmentId.make("environment-old"),
27 label: "Build Mac",
28 httpBaseUrl: "https://build.example.test",
29 wsBaseUrl: "wss://build.example.test",
30});
31
32const descriptor = (protocol: number | undefined, serverVersion: string) =>
33 ({
34 environmentId: TARGET.environmentId,
35 label: TARGET.label,
36 platform: { os: "darwin", arch: "arm64" },
37 serverVersion,
38 ...(protocol === undefined ? {} : { orchestrationProtocolVersion: protocol }),
39 capabilities: { repositoryIdentity: true, serverSelfUpdate: "boot-service" },
40 }) satisfies ExecutionEnvironmentDescriptor;
41
42const RpcRequest = Schema.TaggedStruct("Request", {
43 id: Schema.Union([Schema.String, Schema.Number]),
44 payload: Schema.Unknown,
45 tag: Schema.String,
46});
47const isRpcRequest = Schema.is(RpcRequest);
48const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
49const encodeJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
50const encodeDescriptor = Schema.encodeSync(Schema.fromJsonString(ExecutionEnvironmentDescriptor));
51
52type Listener = (event: { readonly type: string; readonly data?: unknown }) => void;
53
54/** Answers update RPCs the way a protocol-1 server does. */
55class OutdatedHostSocket {
56 static readonly OPEN = 1;
57 readyState = 0;
58 readonly requests: Array<typeof RpcRequest.Type> = [];
59 private readonly listeners = new Map<string, Set<Listener>>();
60
61 readonly url: string;
62 private readonly onUpdate: () => void;
63
64 constructor(url: string, onUpdate: () => void) {
65 this.url = url;
66 this.onUpdate = onUpdate;
67 queueMicrotask(() => {
68 this.readyState = OutdatedHostSocket.OPEN;
69 this.emit({ type: "open" });
70 });
71 }
72
73 addEventListener(type: string, listener: Listener) {
74 const listeners = this.listeners.get(type) ?? new Set<Listener>();
75 listeners.add(listener);
76 this.listeners.set(type, listeners);
77 }
78
79 removeEventListener(type: string, listener: Listener) {
80 this.listeners.get(type)?.delete(listener);
81 }
82
83 send(data: string) {
84 const message = decodeJson(data);
85 if (!isRpcRequest(message)) return;
86 this.requests.push(message);
87 if (message.tag !== WS_METHODS.serverUpdateServer) return;
88 this.onUpdate();
89 queueMicrotask(() =>
90 this.emit({
91 type: "message",
92 data: encodeJson({
93 _tag: "Exit",
94 requestId: message.id,
95 exit: {
96 _tag: "Success",
97 value: { targetVersion: "0.0.46", method: "boot-service", updateId: "update-1" },
98 },
99 }),
100 }),
101 );
102 }
103
104 close() {
105 this.readyState = 3;
106 this.emit({ type: "close" });
107 }
108
109 private emit(event: { readonly type: string; readonly data?: unknown }) {
110 for (const listener of this.listeners.get(event.type) ?? []) listener(event);
111 }
112}
113
114describe("updateOutdatedHost", () => {
115 it.effect("updates a protocol-1 host over a bare socket, then switches it back on", () =>
116 Effect.gen(function* () {
117 const blocked = orchestrationProtocolCompatibilityError(descriptor(undefined, "0.0.45"));
118 expect(blocked).toMatchObject({ serverUpdateRequired: true });
119
120 const sockets: Array<OutdatedHostSocket> = [];
121 let served: ExecutionEnvironmentDescriptor = descriptor(undefined, "0.0.45");
122 const entries = yield* SubscriptionRef.make<
123 ReadonlyMap<EnvironmentId, ConnectionCatalogEntry>
124 >(
125 new Map([
126 [
127 TARGET.environmentId,
128 {
129 target: TARGET,
130 profile: Option.none(),
131 enabled: false,
132 unsupportedReason: blocked?.message ?? "",
133 serverUpdateRequired: true,
134 },
135 ],
136 ]),
137 );
138 const calls: Array<string> = [];
139 const registry = EnvironmentRegistry.EnvironmentRegistry.of({
140 entries,
141 setCompatibility: (_environmentId: EnvironmentId, error: unknown) =>
142 Effect.sync(() => calls.push(`compatibility:${error === null ? "clear" : "block"}`)),
143 setEnabled: (_environmentId: EnvironmentId, enabled: boolean) =>
144 Effect.sync(() => calls.push(`enabled:${enabled}`)),
145 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
146 const resolver = ConnectionResolver.ConnectionResolver.of({
147 prepare: () => Effect.die(new Error("The update must bypass the protocol gate.")),
148 prepareForUpdate: () =>
149 Effect.sync(() => served).pipe(
150 Effect.map((current) => ({
151 descriptor: current,
152 prepared: {
153 environmentId: TARGET.environmentId,
154 label: TARGET.label,
155 httpBaseUrl: TARGET.httpBaseUrl,
156 socketUrl: "wss://build.example.test/ws?wsTicket=ticket",
157 httpAuthorization: null,
158 target: TARGET,
159 },
160 })),
161 ),
162 });
163 const httpClient = HttpClient.make((request) =>
164 Effect.sync(() =>
165 HttpClientResponse.fromWeb(request, new Response(encodeDescriptor(served))),
166 ),
167 );
168
169 const result = yield* updateOutdatedHost(
170 TARGET.environmentId,
171 { targetVersion: "0.0.46" },
172 () => Effect.void,
173 ).pipe(
174 Effect.provide(
175 Layer.mergeAll(
176 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, registry),
177 Layer.succeed(ConnectionResolver.ConnectionResolver, resolver),
178 Layer.succeed(
179 RelayEnvironmentDiscovery.RelayEnvironmentDiscovery,
180 RelayEnvironmentDiscovery.RelayEnvironmentDiscovery.of({
181 state: yield* SubscriptionRef.make(
182 RelayEnvironmentDiscovery.EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE,
183 ),
184 refresh: Effect.void,
185 }),
186 ),
187 Layer.succeed(HttpClient.HttpClient, httpClient),
188 Layer.succeed(Socket.WebSocketConstructor, (url) => {
189 // The host relaunches on a compatible protocol once the update lands.
190 const socket = new OutdatedHostSocket(url, () => {
191 served = descriptor(ORCHESTRATION_PROTOCOL_VERSION, "0.0.46");
192 });
193 sockets.push(socket);
194 return socket as unknown as globalThis.WebSocket;
195 }),
196 ),
197 ),
198 );
199
200 expect(sockets[0]?.url).not.toContain("orchestrationProtocol");
201 expect(sockets[0]?.requests.map((request) => request.tag)).toEqual([
202 WS_METHODS.serverUpdateServer,
203 ]);
204 expect(result.targetVersion).toBe("0.0.46");
205 expect(calls).toEqual(["compatibility:clear", "enabled:true"]);
206 }),
207 );
208
209 it.effect("refuses a host that cannot update itself without opening a socket", () =>
210 Effect.gen(function* () {
211 const manual = {
212 ...descriptor(undefined, "0.0.45"),
213 capabilities: { repositoryIdentity: true },
214 };
215 let opened = false;
216 const error = yield* Effect.flip(
217 updateOutdatedHost(TARGET.environmentId, { targetVersion: "0.0.46" }, () => Effect.void),
218 ).pipe(
219 Effect.provide(
220 Layer.mergeAll(
221 Layer.succeed(
222 EnvironmentRegistry.EnvironmentRegistry,
223 EnvironmentRegistry.EnvironmentRegistry.of({
224 entries: yield* SubscriptionRef.make<
225 ReadonlyMap<EnvironmentId, ConnectionCatalogEntry>
226 >(
227 new Map([
228 [
229 TARGET.environmentId,
230 { target: TARGET, profile: Option.none(), enabled: false },
231 ],
232 ]),
233 ),
234 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]),
235 ),
236 Layer.succeed(
237 ConnectionResolver.ConnectionResolver,
238 ConnectionResolver.ConnectionResolver.of({
239 prepare: () => Effect.die(new Error("unused")),
240 prepareForUpdate: () =>
241 Effect.succeed({
242 descriptor: manual,
243 prepared: {
244 environmentId: TARGET.environmentId,
245 label: TARGET.label,
246 httpBaseUrl: TARGET.httpBaseUrl,
247 socketUrl: "wss://build.example.test/ws",
248 httpAuthorization: null,
249 target: TARGET,
250 },
251 }),
252 }),
253 ),
254 Layer.succeed(
255 RelayEnvironmentDiscovery.RelayEnvironmentDiscovery,
256 RelayEnvironmentDiscovery.RelayEnvironmentDiscovery.of({
257 state: yield* SubscriptionRef.make(
258 RelayEnvironmentDiscovery.EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE,
259 ),
260 refresh: Effect.void,
261 }),
262 ),
263 Layer.succeed(
264 HttpClient.HttpClient,
265 HttpClient.make(() => Effect.die(new Error("unused"))),
266 ),
267 Layer.succeed(Socket.WebSocketConstructor, () => {
268 opened = true;
269 throw new Error("unused");
270 }),
271 ),
272 ),
273 );
274 expect(error).toMatchObject({ _tag: "OutdatedHostUpdateError" });
275 expect(opened).toBe(false);
276 }),
277 );
278});