1import { HttpServer, NodeHttpServer } from "alchemy/Http";
2import { bootstrap } from "alchemy/Runtime/Bootstrap/Fly";
3import { runProcess } from "alchemy/Runtime/Bootstrap/Process";
4import { createContainerRuntimeContext } from "alchemy/Server/Process";
5import * as Data from "effect/Data";
6import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as Fiber from "effect/Fiber";
9import * as Scope from "effect/Scope";
10import * as Stream from "effect/Stream";
11import { HttpServerRequest } from "effect/http/HttpServerRequest";
12import * as HttpServerResponse from "effect/http/HttpServerResponse";
13
14Effect.runSync(
15 Effect.sync(() => {
16 if (process.env.RECORD_SHUTDOWN_TIMING !== "1") return;
17 let started: number | undefined;
18 const onSignal = () => {
19 started ??= performance.now();
20 };
21 process.prependOnceListener("SIGTERM", onSignal);
22 process.prependOnceListener("SIGINT", onSignal);
23 process.once("exit", () => {
24 if (started !== undefined) {
25 console.error(`shutdown elapsed ms: ${performance.now() - started}`);
26 }
27 });
28 }),
29);
30
31const event = (message: string) => Effect.sync(() => console.log(message));
32let dependencyOpen = true;
33const useDependency = Effect.sync(() => {
34 if (!dependencyOpen) throw new Error("dependency already closed");
35});
36
37const handler = Effect.gen(function* () {
38 const request = yield* HttpServerRequest;
39 if (request.url === "/health") return HttpServerResponse.text("ok");
40 yield* Effect.addFinalizer(() =>
41 useDependency.pipe(Effect.andThen(event("request dependency released"))),
42 );
43 yield* Effect.addFinalizer(() =>
44 useDependency.pipe(
45 Effect.andThen(event("request finalized")),
46 Effect.andThen(
47 process.env.FAIL_REQUEST_FINALIZER === "1"
48 ? Effect.sleep("100 millis").pipe(
49 Effect.andThen(Effect.die(new Error("request cleanup defect"))),
50 )
51 : Effect.void,
52 ),
53 ),
54 );
55 yield* event("request started");
56 if (request.url === "/handler-error") {
57 return yield* Effect.die(new Error("ordinary handler defect"));
58 }
59 if (request.url === "/hang") return yield* Effect.never;
60 if (request.url === "/stream-large") {
61 return HttpServerResponse.stream(
62 Stream.range(0, 15).pipe(
63 Stream.mapEffect((index) =>
64 Effect.sleep("60 millis").pipe(
65 Effect.andThen(useDependency),
66 Effect.andThen(
67 event(index === 15 ? "chunk last" : `chunk ${index}`),
68 ),
69 Effect.andThen(
70 Effect.sync(() =>
71 new TextEncoder().encode(
72 String(index).padStart(2, "0").repeat(8192),
73 ),
74 ),
75 ),
76 ),
77 ),
78 ),
79 );
80 }
81 if (request.url === "/stream" || request.url === "/stream-hang") {
82 const body = Stream.fromArray(["first\n", "second\n", "last\n"]).pipe(
83 Stream.mapEffect((chunk) =>
84 Effect.sleep("300 millis").pipe(
85 Effect.andThen(useDependency),
86 Effect.andThen(event(`chunk ${chunk.trim()}`)),
87 Effect.andThen(Effect.sync(() => new TextEncoder().encode(chunk))),
88 ),
89 ),
90 );
91 return HttpServerResponse.stream(
92 request.url === "/stream-hang"
93 ? body.pipe(Stream.concat(Stream.fromEffect(Effect.never)))
94 : body,
95 );
96 }
97 yield* Effect.sleep("900 millis");
98 yield* useDependency;
99 yield* event("response ready");
100 return HttpServerResponse.text("completed");
101});
102
103class StopFailed extends Data.TaggedError("StopFailed") {}
104
105const worker = (name: string, mode: string) =>
106 Effect.gen(function* () {
107 let clientOpen = true;
108 yield* Effect.addFinalizer(() =>
109 useDependency.pipe(
110 Effect.andThen(
111 Effect.sync(() => {
112 clientOpen = false;
113 }),
114 ),
115 Effect.andThen(event(`${name} client released`)),
116 ),
117 );
118 const workScope = yield* Scope.make();
119 yield* Effect.addFinalizer((exit) =>
120 Scope.close(workScope, exit).pipe(
121 Effect.andThen(event(`${name} work scope closed`)),
122 ),
123 );
124 const pollScope = yield* Scope.make();
125 yield* Effect.addFinalizer((exit) => Scope.close(pollScope, exit));
126 const jobs: Fiber.Fiber<void, never>[] = [];
127 let accepting = true;
128 let claims = 0;
129 const claim = Effect.gen(function* () {
130 if (!accepting) return;
131 const id = ++claims;
132 yield* event(`${name} claim ${id}`);
133 const job = Effect.gen(function* () {
134 yield* Effect.addFinalizer(() => event(`${name} job ${id} finalized`));
135 yield* mode === "job-hang" ? Effect.never : Effect.sleep("900 millis");
136 yield* useDependency;
137 yield* Effect.sync(() => {
138 if (!clientOpen) throw new Error("client already closed");
139 });
140 yield* event(`${name} job ${id} completed`);
141 }).pipe(Effect.scoped, Effect.forkIn(workScope));
142 jobs.push(yield* job);
143 });
144 if (mode !== "stop-delay-empty") yield* claim;
145 const poller = yield* Effect.sleep("150 millis").pipe(
146 Effect.andThen(claim),
147 Effect.forever,
148 Effect.forkIn(pollScope),
149 );
150 const stop = Effect.gen(function* () {
151 yield* event(`${name} stop started`);
152 if (mode === "stop-hang") yield* Effect.never;
153 if (mode === "stop-delay" || mode === "stop-delay-empty") {
154 yield* Effect.sleep("700 millis");
155 }
156 if (mode === "stop-fail") return yield* Effect.fail(new StopFailed());
157 accepting = false;
158 yield* Fiber.interrupt(poller);
159 yield* event(`${name} stopped ${claims}`);
160 });
161 yield* Effect.addFinalizer(() =>
162 stop.pipe(
163 Effect.andThen(
164 Effect.gen(function* () {
165 yield* event(`${name} drain started`);
166 yield* Effect.forEach(jobs, Fiber.join, {
167 concurrency: "unbounded",
168 });
169 yield* useDependency;
170 yield* event(`${name} drained`);
171 }),
172 ),
173 Effect.catchTag("StopFailed", (error) =>
174 Effect.gen(function* () {
175 accepting = false;
176 yield* Fiber.interrupt(poller);
177 yield* event(`${name} checkpoint`);
178 yield* Scope.close(workScope, Exit.void);
179 return yield* Effect.die(error);
180 }),
181 ),
182 ),
183 );
184 yield* event(`${name} ready`);
185 const requestedExit = name === "worker1" ? process.env.RUN_EXIT : undefined;
186 if (mode === "normal" || requestedExit === "normal") return;
187 if (mode === "error-delay") yield* Effect.sleep("600 millis");
188 if (
189 mode === "error" ||
190 mode === "error-delay" ||
191 requestedExit === "error"
192 ) {
193 return yield* Effect.die(new Error("worker failed"));
194 }
195 yield* Effect.never;
196 });
197
198const host = createContainerRuntimeContext("Fly.Service")("fixture");
199const httpProgram = Effect.gen(function* () {
200 const server = yield* HttpServer;
201 yield* server.serve(handler);
202 yield* event("http ready");
203 yield* Effect.never;
204});
205
206const entrypoint = Effect.gen(function* () {
207 yield* Effect.addFinalizer(() =>
208 event("instance finalizing").pipe(
209 Effect.andThen(
210 process.env.HANG_FINALIZER === "1"
211 ? Effect.never
212 : Effect.sleep("40 millis"),
213 ),
214 Effect.andThen(
215 Effect.sync(() => {
216 dependencyOpen = false;
217 }),
218 ),
219 Effect.andThen(event("instance finalized")),
220 Effect.andThen(
221 process.env.FAIL_FINALIZER === "1"
222 ? Effect.die(new Error("dependency finalizer failed"))
223 : Effect.void,
224 ),
225 ),
226 );
227 if (process.env.WORKERS) {
228 const first = worker("worker1", process.env.WORKER_MODE ?? "drain");
229 yield* host.run(
230 process.env.NESTED_WORKER === "1" ? Effect.scoped(first) : first,
231 );
232 if (process.env.WORKERS === "2")
233 yield* host.run(worker("worker2", "drain"));
234 }
235 if (process.env.RUN_ONLY !== "1") yield* host.run(httpProgram);
236 return { RuntimeContext: host };
237});
238
239if (process.env.UNMANAGED_HOST === "1") {
240 await runProcess(
241 "Unmanaged service",
242 httpProgram.pipe(Effect.provide(NodeHttpServer()), Effect.scoped),
243 );
244} else {
245 await bootstrap(entrypoint);
246}