1import * as Cloudflare from "@/Cloudflare";
2import * as Neon from "@/Neon";
3import * as Test from "@/Test/Alchemy";
4import { expect } from "alchemy-test";
5import * as Data from "effect/Data";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
10import * as HttpClient from "effect/http/HttpClient";
11import Stack from "./fixtures/drizzle-workflow/stack.ts";
12import type { Widget } from "./fixtures/drizzle-workflow/schema.ts";
13
14const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
15 providers: Layer.mergeAll(Cloudflare.providers(), Neon.providers()),
16});
17
18const logLevel = Effect.provideService(
19 MinimumLogLevel,
20 process.env.DEBUG ? "Debug" : "Info",
21);
22
23const stack = beforeAll(deploy(Stack));
24afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack));
25
26interface WorkflowStatus {
27 status: string;
28 output?: {
29 inserted: Widget;
30 rowCount: number;
31 widget: Widget | null;
32 };
33 error?: { message?: string } | null;
34}
35
36class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
37 status: number;
38}> {}
39
40// Start a fresh workflow instance and poll until it reaches a terminal state.
41const runToCompletion = (baseUrl: string) =>
42 Effect.gen(function* () {
43 const client = yield* HttpClient.HttpClient;
44
45 // Fresh workers.dev edge takes a few seconds to start serving 200s.
46 const startRes = yield* client.post(`${baseUrl}/workflow/start/1`).pipe(
47 Effect.flatMap((res) =>
48 res.status === 200
49 ? Effect.succeed(res)
50 : Effect.fail(new WorkerNotReady({ status: res.status })),
51 ),
52 Effect.retry({
53 while: (e): e is WorkerNotReady =>
54 e instanceof WorkerNotReady && e.status >= 400 && e.status < 600,
55 schedule: Schedule.max([
56 Schedule.exponential("500 millis"),
57 Schedule.recurs(15),
58 ]),
59 }),
60 );
61 const { instanceId } = (yield* startRes.json) as { instanceId: string };
62 expect(instanceId).toBeTypeOf("string");
63
64 const last = yield* client
65 .get(`${baseUrl}/workflow/status/${instanceId}`)
66 .pipe(
67 Effect.flatMap((res) =>
68 res.status === 200
69 ? Effect.succeed(res)
70 : Effect.fail(new WorkerNotReady({ status: res.status })),
71 ),
72 Effect.flatMap((res) => res.json),
73 Effect.map((json) => json as unknown as WorkflowStatus),
74 Effect.repeat({
75 schedule: Schedule.spaced("2 seconds"),
76 until: (s) => s.status === "complete" || s.status === "errored",
77 times: 30,
78 }),
79 );
80 if (last.status !== "complete") {
81 return yield* Effect.fail(
82 new Error(`workflow ${last.status}: ${JSON.stringify(last.error)}`),
83 );
84 }
85 return last;
86 });
87
88/**
89 * End-to-end regression guard for the ExecutionContext-in-Workflow fix
90 * (PR #515): deploy a Neon project + branch, point a Cloudflare Hyperdrive
91 * at it, host a Workflow that runs `Drizzle.Postgres` queries inside `task`
92 * steps, fire an instance over HTTP, and assert the run completes with the
93 * row it wrote.
94 *
95 * Before the fix the query inside the step dies on a missing
96 * `ExecutionContext` service and the run reports `errored` with no output.
97 */
98test(
99 "Drizzle.Postgres query runs inside a Workflow task (per-run scope provided by the bridge)",
100 Effect.gen(function* () {
101 const { url } = yield* stack;
102 expect(url).toBeTypeOf("string");
103 const baseUrl = url.replace(/\/+$/, "");
104
105 const last = yield* runToCompletion(baseUrl).pipe(
106 Effect.retry({ schedule: Schedule.spaced("3 seconds"), times: 2 }),
107 );
108
109 expect(last.status).toBe("complete");
110 expect(last.error).toBeFalsy();
111 expect(last.output?.rowCount).toBe(1);
112 expect(last.output?.widget).toMatchObject({ id: 1, name: "widget-1" });
113 expect(last.output?.inserted).toMatchObject({ id: 1, name: "widget-1" });
114 }).pipe(logLevel),
115 {
116 tags: [
117 "provider:cloudflare",
118 "provider:cloudflare:hyperdrive",
119 "provider:cloudflare:worker",
120 "provider:cloudflare:workflow",
121 "provider:neon",
122 "provider:neon:branch",
123 "provider:neon:project",
124 "live",
125 ],
126 timeout: 600_000,
127 },
128);
129
130/**
131 * Cross-request regression guard for the build-once bridge: the isolate
132 * builds its layers on the first event and reuses them; each fetch event
133 * builds its own pg pool against its per-event scope and closes it after the
134 * response. Sequential requests must not touch a prior event's pool
135 * ("Cannot use a pool after calling end on the pool") and concurrent
136 * requests must not share I/O objects across request contexts ("Cannot
137 * perform I/O on behalf of a different request").
138 */
139test(
140 "Drizzle.Postgres queries survive sequential and concurrent fetch events",
141 Effect.gen(function* () {
142 const { url } = yield* stack;
143 const baseUrl = url.replace(/\/+$/, "");
144 const client = yield* HttpClient.HttpClient;
145
146 const query = (id: number) =>
147 Effect.gen(function* () {
148 const res = yield* client.get(`${baseUrl}/query/${id}`).pipe(
149 Effect.flatMap((res) =>
150 res.status === 200
151 ? Effect.succeed(res)
152 : Effect.fail(new WorkerNotReady({ status: res.status })),
153 ),
154 Effect.retry({
155 while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
156 schedule: Schedule.max([
157 Schedule.exponential("500 millis"),
158 Schedule.recurs(10),
159 ]),
160 }),
161 );
162 return (yield* res.json) as { rowCount: number };
163 }).pipe(Effect.orDie);
164
165 // Sequential: the second event must not observe the first event's
166 // (already closed) pool.
167 const first = yield* query(1);
168 const second = yield* query(1);
169 expect(first.rowCount).toBeTypeOf("number");
170 expect(second.rowCount).toBeTypeOf("number");
171
172 // Concurrent: every event acquires its own pool on its own scope.
173 const results = yield* Effect.all(
174 [1, 2, 3, 4, 5, 6].map((id) => query(id)),
175 { concurrency: "unbounded" },
176 );
177 for (const body of results) {
178 expect(body.rowCount).toBeTypeOf("number");
179 }
180 }).pipe(logLevel),
181 {
182 tags: [
183 "provider:cloudflare",
184 "provider:cloudflare:hyperdrive",
185 "provider:cloudflare:worker",
186 "provider:cloudflare:workflow",
187 "provider:neon",
188 "provider:neon:branch",
189 "provider:neon:project",
190 "live",
191 ],
192 timeout: 600_000,
193 },
194);