1import { EnvironmentId } from "@t3tools/contracts";
2import { assert, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";
4import * as Fiber from "effect/Fiber";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as FetchHttpClient from "effect/http/FetchHttpClient";
8import { HttpClient, HttpServer } from "effect/http";
9import * as NetAddress from "effect/net/NetAddress";
10import * as NodeServices from "@effect/platform-node/NodeServices";
11
12import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
13import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
14import * as ServerConfig from "../config.ts";
15import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
16import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
17import * as CliTokenManager from "./CliTokenManager.ts";
18import * as CloudLink from "./CloudLink.ts";
19import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts";
20import {
21 HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET,
22 PUBLISH_AGENT_ACTIVITY_SECRET,
23 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
24 RELAY_URL_SECRET,
25} from "./config.ts";
26
27const encode = (value: string) => new TextEncoder().encode(value);
28
29/** A linked environment whose secret store can refuse writes, and the relay calls it made. */
30const withService = <A, E>(
31 options: {
32 readonly failHoldWrite?: boolean;
33 readonly failActivityWrite?: boolean;
34 readonly relayFails?: boolean;
35 readonly failActivityRead?: boolean;
36 readonly failHoldRead?: boolean;
37 /** The first relay call reports itself, then waits for this before answering. */
38 readonly holdFirstRelayCall?: {
39 readonly started: () => void;
40 readonly released: Promise<void>;
41 };
42 },
43 body: (input: {
44 readonly preferences: { readonly update: CloudLink.CloudLink["Service"]["updatePreferences"] };
45 readonly stored: Map<string, Uint8Array>;
46 readonly relayCalls: Array<boolean>;
47 }) => Effect.Effect<A, E, never>,
48) =>
49 Effect.gen(function* () {
50 const stored = new Map<string, Uint8Array>([
51 [RELAY_URL_SECRET, encode("https://relay.test")],
52 [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, encode("credential")],
53 [HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET, encode("false")],
54 [PUBLISH_AGENT_ACTIVITY_SECRET, encode("false")],
55 ]);
56 const relayCalls: Array<boolean> = [];
57 const fetch: typeof globalThis.fetch = Object.assign(
58 (_input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => {
59 const payload = JSON.parse(String(init?.body)) as { holdWebhooksWhileOffline: boolean };
60 relayCalls.push(payload.holdWebhooksWhileOffline);
61 if (relayCalls.length === 1 && options.holdFirstRelayCall) {
62 options.holdFirstRelayCall.started();
63 return options.holdFirstRelayCall.released.then(() => Response.json(payload));
64 }
65 return options.relayFails
66 ? Promise.resolve(new Response("unavailable", { status: 503 }))
67 : Promise.resolve(Response.json(payload));
68 },
69 { preconnect: () => {} },
70 );
71 const layerDependencies = Layer.mergeAll(
72 Layer.mock(ServerSecretStore.ServerSecretStore)({
73 get: (name) =>
74 (options.failActivityRead && name === PUBLISH_AGENT_ACTIVITY_SECRET) ||
75 (options.failHoldRead && name === HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)
76 ? Effect.fail(
77 new ServerSecretStore.SecretStoreReadError({
78 resource: name,
79 cause: new Error("busy"),
80 }),
81 )
82 : Effect.succeed(Option.fromNullishOr(stored.get(name))),
83 set: (name, value) =>
84 (options.failHoldWrite && name === HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET) ||
85 (options.failActivityWrite && name === PUBLISH_AGENT_ACTIVITY_SECRET)
86 ? Effect.fail(
87 new ServerSecretStore.SecretStorePersistError({
88 resource: name,
89 cause: new Error("read-only"),
90 }),
91 )
92 : Effect.sync(() => void stored.set(name, value)),
93 }),
94 Layer.mock(ServerEnvironment.ServerEnvironment)({
95 getEnvironmentId: Effect.succeed(EnvironmentId.make("environment-1")),
96 }),
97 Layer.mock(AgentAwarenessRelay.AgentAwarenessRelay)({ requestCatchUp: () => Effect.void }),
98 // Saving preferences touches none of the link's other dependencies.
99 Layer.mock(ManagedEndpointRuntime.CloudManagedEndpointRuntime)({}),
100 Layer.mock(EnvironmentAuth.EnvironmentAuth)({}),
101 Layer.mock(CliTokenManager.CloudCliTokenManager)({}),
102 Layer.mock(HttpServer.HttpServer)({
103 address: NetAddress.inetAddressFromIpStringUnsafe("127.0.0.1", 3773),
104 }),
105 Layer.mock(ServerConfig.ServerConfig)({} as ServerConfig.ServerConfig["Service"]),
106 Layer.succeed(
107 HttpClient.HttpClient,
108 HttpClient.make(() => Effect.die("unused")),
109 ),
110 NodeServices.layer,
111 );
112 return yield* Effect.gen(function* () {
113 const link = yield* CloudLink.CloudLink;
114 return yield* body({ preferences: { update: link.updatePreferences }, stored, relayCalls });
115 }).pipe(
116 Effect.provide(CloudLink.layer.pipe(Layer.provide(layerDependencies))),
117 Effect.provideService(FetchHttpClient.Fetch, fetch),
118 );
119 });
120
121it.effect("tells the relay before saving the hold setting locally", () =>
122 withService({}, ({ preferences, stored, relayCalls }) =>
123 Effect.gen(function* () {
124 yield* preferences.update({ publishAgentActivity: true, holdWebhooksWhileOffline: true });
125 assert.deepEqual(relayCalls, [true]);
126 assert.equal(
127 new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
128 "true",
129 );
130 }),
131 ),
132);
133
134it.effect("puts the relay back when the local save fails", () =>
135 withService({ failHoldWrite: true }, ({ preferences, stored, relayCalls }) =>
136 Effect.gen(function* () {
137 const error = yield* preferences
138 .update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
139 .pipe(Effect.flip);
140 assert.deepInclude(error, {
141 _tag: "CloudLinkInternalError",
142 operation: "persist-preferences",
143 });
144 assert.deepEqual(relayCalls, [true, false]);
145 assert.equal(
146 new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
147 "false",
148 );
149 }),
150 ),
151);
152
153it.effect("leaves the relay untouched when the activity setting can't be saved", () =>
154 withService({ failActivityWrite: true }, ({ preferences, stored, relayCalls }) =>
155 Effect.gen(function* () {
156 const error = yield* preferences
157 .update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
158 .pipe(Effect.flip);
159 assert.deepInclude(error, {
160 _tag: "CloudLinkInternalError",
161 operation: "persist-preferences",
162 });
163 assert.deepEqual(relayCalls, []);
164 assert.equal(
165 new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
166 "false",
167 );
168 }),
169 ),
170);
171
172it.effect("keeps both settings unchanged when the relay refuses the hold change", () =>
173 withService({ relayFails: true }, ({ preferences, stored }) =>
174 Effect.gen(function* () {
175 yield* preferences
176 .update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
177 .pipe(Effect.flip);
178 assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
179 assert.equal(
180 new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
181 "false",
182 );
183 }),
184 ),
185);
186
187it.effect("changes nothing when the current activity setting can't be read", () =>
188 withService({ failActivityRead: true, relayFails: true }, ({ preferences, stored, relayCalls }) =>
189 Effect.gen(function* () {
190 const error = yield* preferences
191 .update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
192 .pipe(Effect.flip);
193 assert.deepInclude(error, { _tag: "CloudLinkInternalError", operation: "read-preferences" });
194 assert.deepEqual(relayCalls, []);
195 assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
196 }),
197 ),
198);
199
200it.effect("two overlapping updates are applied one after the other", () => {
201 let release!: () => void;
202 let started!: () => void;
203 const released = new Promise<void>((resolve) => {
204 release = resolve;
205 });
206 const firstAtRelay = new Promise<void>((resolve) => {
207 started = resolve;
208 });
209 return withService({ holdFirstRelayCall: { started, released } }, ({ preferences, stored }) =>
210 Effect.gen(function* () {
211 const first = yield* preferences
212 .update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
213 .pipe(Effect.forkChild);
214 yield* Effect.promise(() => firstAtRelay);
215 const second = yield* preferences
216 .update({ publishAgentActivity: false, holdWebhooksWhileOffline: false })
217 .pipe(Effect.forkChild);
218 // Give the second update every chance to run; it must wait for the first.
219 yield* Effect.yieldNow;
220 assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "true");
221 release();
222 yield* Fiber.join(first);
223 yield* Fiber.join(second);
224 assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
225 assert.equal(
226 new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
227 "false",
228 );
229 }),
230 );
231});
232
233it.effect("changes nothing when the current hold setting can't be read", () =>
234 withService({ failHoldRead: true, failHoldWrite: true }, ({ preferences, stored, relayCalls }) =>
235 Effect.gen(function* () {
236 stored.set(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET, encode("true"));
237 const error = yield* preferences
238 .update({ publishAgentActivity: true, holdWebhooksWhileOffline: false })
239 .pipe(Effect.flip);
240 assert.deepInclude(error, { _tag: "CloudLinkInternalError", operation: "read-preferences" });
241 assert.deepEqual(relayCalls, []);
242 assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "false");
243 }),
244 ),
245);
246
247it.effect("an update the client walks away from still finishes", () => {
248 let release!: () => void;
249 let started!: () => void;
250 const released = new Promise<void>((resolve) => {
251 release = resolve;
252 });
253 const atRelay = new Promise<void>((resolve) => {
254 started = resolve;
255 });
256 return withService({ holdFirstRelayCall: { started, released } }, ({ preferences, stored }) =>
257 Effect.gen(function* () {
258 const update = yield* preferences
259 .update({ publishAgentActivity: true, holdWebhooksWhileOffline: true })
260 .pipe(Effect.forkChild);
261 yield* Effect.promise(() => atRelay);
262 // The interrupt is in flight before the relay answers.
263 const interrupted = yield* Fiber.interrupt(update).pipe(Effect.forkChild);
264 yield* Effect.yieldNow;
265 release();
266 yield* Fiber.join(interrupted);
267 assert.equal(new TextDecoder().decode(stored.get(PUBLISH_AGENT_ACTIVITY_SECRET)), "true");
268 assert.equal(
269 new TextDecoder().decode(stored.get(HOLD_WEBHOOKS_WHILE_OFFLINE_SECRET)),
270 "true",
271 );
272 }),
273 );
274});