1import * as Effect from "effect/Effect";
2import * as Duration from "effect/Duration";
3import * as Option from "effect/Option";
4import * as Path from "effect/Path";
5import * as Redacted from "effect/Redacted";
6import * as Result from "effect/Result";
7import * as Semaphore from "effect/Semaphore";
8import * as Stream from "effect/Stream";
9import * as ChildProcess from "effect/process/ChildProcess";
10import type { Server, ServerOptions } from "@prisma/dev";
11import type { DatabaseDev } from "./Database.ts";
12
13export interface PrismaDevDatabaseAttrs {
14 directConnectionString: Redacted.Redacted<string>;
15 pooledConnectionString: Redacted.Redacted<string>;
16 accelerateConnectionString: Redacted.Redacted<string>;
17 host: string | null;
18 user: string | null;
19 password: Redacted.Redacted<string> | undefined;
20}
21
22interface PrismaDevDatabaseEntry {
23 optionsKey: string;
24 migrationKey: string | undefined;
25 server: Server;
26 attrs: PrismaDevDatabaseAttrs | undefined;
27}
28
29const servers = new Map<string, PrismaDevDatabaseEntry>();
30const startMutex = Semaphore.makeUnsafe(1);
31const MIGRATION_OUTPUT_LIMIT_BYTES = 1024 * 1024;
32
33const toError = (message: string) => (cause: unknown) =>
34 cause instanceof Error ? cause : new Error(`${message}: ${String(cause)}`);
35
36const importPrismaDev = Effect.tryPromise({
37 try: () => import("@prisma/dev"),
38 catch: toError("Failed to load @prisma/dev"),
39});
40
41const stableJson = (value: unknown) =>
42 Effect.sync(() => JSON.stringify(value) ?? "");
43
44const sanitizeName = (value: string) =>
45 value.replace(/[^a-zA-Z0-9_.-]+/g, "-").replace(/^-+|-+$/g, "") || "alchemy";
46
47const optionsFrom = (databaseId: string, dev: DatabaseDev): ServerOptions => ({
48 name: dev.name ?? `alchemy-${sanitizeName(databaseId)}`,
49 persistenceMode: dev.persistenceMode ?? "stateful",
50 port: dev.port,
51 databasePort: dev.databasePort,
52 shadowDatabasePort: dev.shadowDatabasePort,
53 debug: dev.debug,
54 databaseConnectTimeoutMillis: dev.databaseConnectTimeoutMillis,
55 databaseIdleTimeoutMillis: dev.databaseIdleTimeoutMillis,
56 shadowDatabaseConnectTimeoutMillis: dev.shadowDatabaseConnectTimeoutMillis,
57 shadowDatabaseIdleTimeoutMillis: dev.shadowDatabaseIdleTimeoutMillis,
58});
59
60const parseUrl = (label: string, value: string) =>
61 Effect.try({
62 try: () => new URL(value),
63 catch: toError(`Invalid ${label} connection string`),
64 });
65
66const normalizeConnectionString = Effect.fn(function* (value: string) {
67 const url = yield* parseUrl("local Prisma", value);
68 if (
69 url.hostname === "localhost" ||
70 url.hostname === "::1" ||
71 url.hostname === "[::1]"
72 ) {
73 url.hostname = "127.0.0.1";
74 }
75 return url.toString();
76});
77
78const detailsFrom = Effect.fn(function* (value: string) {
79 const url = yield* parseUrl("direct Prisma", value);
80 return yield* Effect.try({
81 try: () => ({
82 host: url.hostname || null,
83 user: url.username ? decodeURIComponent(url.username) : null,
84 password: url.password
85 ? Redacted.make(decodeURIComponent(url.password))
86 : undefined,
87 }),
88 catch: toError("Invalid direct Prisma credentials"),
89 });
90});
91
92export const prismaDevDatabaseAttrsFromServer = Effect.fn(function* (
93 server: Server,
94) {
95 const direct = yield* normalizeConnectionString(
96 server.database.prismaORMConnectionString ??
97 server.database.connectionString,
98 );
99 const pooled = server.ppg.url;
100 const details = yield* detailsFrom(direct);
101 return {
102 directConnectionString: Redacted.make(direct),
103 pooledConnectionString: Redacted.make(pooled),
104 accelerateConnectionString: Redacted.make(pooled),
105 host: details.host,
106 user: details.user,
107 password: details.password,
108 };
109});
110
111const startServer = Effect.fn(function* (
112 databaseId: string,
113 options: ServerOptions,
114) {
115 const prismaDev = yield* importPrismaDev;
116 return yield* Semaphore.withPermits(
117 startMutex,
118 1,
119 )(
120 Effect.tryPromise({
121 try: () => prismaDev.startPrismaDevServer(options),
122 catch: toError(`Failed to start local Prisma database ${databaseId}`),
123 }),
124 );
125});
126
127const closeEntry = Effect.fn(function* (entry: PrismaDevDatabaseEntry) {
128 yield* Effect.tryPromise({
129 try: () => entry.server.close(),
130 catch: toError("Failed to stop local Prisma database"),
131 });
132});
133
134const collectMigrationOutput = <E>(
135 stream: Stream.Stream<Uint8Array, E>,
136 label: "stdout" | "stderr",
137) =>
138 Stream.runFoldEffect(
139 stream,
140 () => ({ chunks: [] as Uint8Array[], bytes: 0 }),
141 (state, chunk) => {
142 const bytes = state.bytes + chunk.byteLength;
143 return bytes > MIGRATION_OUTPUT_LIMIT_BYTES
144 ? Effect.fail(
145 new Error(
146 `Local Prisma migration ${label} exceeded the ${MIGRATION_OUTPUT_LIMIT_BYTES} byte safety limit. Reduce migration verbosity.`,
147 ),
148 )
149 : Effect.succeed({ chunks: [...state.chunks, chunk], bytes });
150 },
151 ).pipe(
152 Effect.map(({ chunks, bytes }) => {
153 const output = new Uint8Array(bytes);
154 let offset = 0;
155 for (const chunk of chunks) {
156 output.set(chunk, offset);
157 offset += chunk.byteLength;
158 }
159 return new TextDecoder().decode(output);
160 }),
161 );
162
163const sensitiveValuesFromUrl = (value: string) => {
164 try {
165 const url = new URL(value);
166 const values = new Set<string>();
167 const add = (secret: string | undefined) => {
168 if (!secret) return;
169 values.add(secret);
170 try {
171 values.add(decodeURIComponent(secret));
172 } catch {
173 // Keep the original value when it is not valid percent-encoding.
174 }
175 values.add(encodeURIComponent(secret));
176 };
177 add(url.username);
178 add(url.password);
179 for (const [key, secret] of url.searchParams) {
180 if (/(?:api[_-]?key|token|secret|password|credential)/i.test(key)) {
181 add(secret);
182 }
183 }
184 return values;
185 } catch {
186 return new Set<string>();
187 }
188};
189
190const redactMigrationOutput = (
191 output: string,
192 attrs: PrismaDevDatabaseAttrs,
193) => {
194 const password = attrs.password ? Redacted.value(attrs.password) : undefined;
195 const connectionStrings = [
196 Redacted.value(attrs.directConnectionString),
197 Redacted.value(attrs.pooledConnectionString),
198 Redacted.value(attrs.accelerateConnectionString),
199 ];
200 const knownSecrets = [
201 ...connectionStrings,
202 ...connectionStrings.flatMap((value) => [...sensitiveValuesFromUrl(value)]),
203 password,
204 password === undefined ? undefined : encodeURIComponent(password),
205 ]
206 .filter((value): value is string => value !== undefined && value !== "")
207 .sort((left, right) => right.length - left.length);
208 return knownSecrets.reduce(
209 (redacted, secret) => redacted.split(secret).join("[REDACTED]"),
210 output,
211 );
212};
213
214const runMigration = Effect.fn(function* (
215 dev: DatabaseDev,
216 attrs: PrismaDevDatabaseAttrs,
217 timeoutSeconds: number,
218) {
219 const path = yield* Path.Path;
220 const command = dev.migrate;
221 if (command === undefined || command.trim() === "") return;
222
223 const cwd = dev.migrateCwd ? path.resolve(dev.migrateCwd) : path.resolve(".");
224 const resultOption = yield* Effect.scoped(
225 Effect.gen(function* () {
226 const handle = yield* ChildProcess.make(command, [], {
227 shell: true,
228 cwd,
229 env: {
230 DATABASE_URL: Redacted.value(attrs.pooledConnectionString),
231 DIRECT_URL: Redacted.value(attrs.directConnectionString),
232 POOLED_DATABASE_URL: Redacted.value(attrs.pooledConnectionString),
233 },
234 extendEnv: true,
235 stdin: "ignore",
236 stdout: "pipe",
237 stderr: "pipe",
238 // Scoped interruption must terminate the entire POSIX process group
239 // immediately. The Effect process runtime defaults to a detached group
240 // on POSIX, so descendants cannot outlive a timed-out migration.
241 killSignal: "SIGKILL",
242 });
243 return yield* Effect.all(
244 [
245 handle.exitCode,
246 collectMigrationOutput(handle.stdout, "stdout"),
247 collectMigrationOutput(handle.stderr, "stderr"),
248 ] as const,
249 { concurrency: 3 },
250 );
251 }).pipe(Effect.timeoutOption(Duration.seconds(timeoutSeconds))),
252 );
253 if (Option.isNone(resultOption)) {
254 return yield* Effect.fail(
255 new Error(
256 `Local Prisma migration command timed out after ${timeoutSeconds} seconds and was terminated.`,
257 ),
258 );
259 }
260 const [exitCode, rawStdout, rawStderr] = resultOption.value;
261 const stdout = redactMigrationOutput(rawStdout, attrs);
262 const stderr = redactMigrationOutput(rawStderr, attrs);
263 if (exitCode !== 0) {
264 return yield* Effect.fail(
265 new Error(
266 `Local Prisma migration command failed with exit code ${exitCode}${stderr ? `\n${stderr}` : ""}`,
267 ),
268 );
269 }
270 if (stdout) yield* Effect.logDebug("Local Prisma migration output", stdout);
271 if (stderr) yield* Effect.logDebug("Local Prisma migration stderr", stderr);
272});
273
274export const ensurePrismaDevDatabase = Effect.fn(function* (
275 databaseId: string,
276 dev: false | DatabaseDev | undefined,
277) {
278 if (dev === false) {
279 const cached = servers.get(databaseId);
280 if (cached !== undefined) {
281 yield* closeEntry(cached);
282 servers.delete(databaseId);
283 }
284 return undefined;
285 }
286 const config: DatabaseDev = dev ?? {};
287 if (config.provider !== undefined && config.provider !== "@prisma/dev") {
288 return yield* Effect.fail(
289 new Error(
290 `Unsupported Prisma local database provider ${config.provider}`,
291 ),
292 );
293 }
294 const migrateTimeoutSeconds = config.migrateTimeoutSeconds ?? 900;
295 if (!Number.isFinite(migrateTimeoutSeconds) || migrateTimeoutSeconds <= 0) {
296 return yield* Effect.fail(
297 new Error("migrateTimeoutSeconds must be a positive finite number."),
298 );
299 }
300
301 const options = optionsFrom(databaseId, config);
302 const optionsKey = yield* stableJson(options);
303 const cached = servers.get(databaseId);
304 let entry = cached;
305 if (
306 entry === undefined ||
307 entry.optionsKey !== optionsKey ||
308 entry.attrs === undefined
309 ) {
310 if (entry !== undefined) {
311 // Keep the handle tracked when close fails so a later cleanup can retry
312 // instead of leaking an unreachable local database process.
313 yield* closeEntry(entry);
314 servers.delete(databaseId);
315 }
316 const server = yield* startServer(databaseId, options);
317 // `@prisma/dev` binds Postgres/HTTP to 127.0.0.1. Native Linux Docker
318 // must not SYN the bridge IP (UFW INPUT). The workerd sidecar proxy
319 // unix-socket-tunnels these ports into the container netns.
320 const attrsResult = yield* Effect.result(
321 prismaDevDatabaseAttrsFromServer(server),
322 );
323 if (Result.isFailure(attrsResult)) {
324 const closeResult = yield* Effect.result(
325 Effect.tryPromise({
326 try: () => server.close(),
327 catch: toError("Failed to stop invalid local Prisma database"),
328 }),
329 );
330 if (Result.isFailure(closeResult)) {
331 servers.set(databaseId, {
332 optionsKey,
333 migrationKey: undefined,
334 server,
335 attrs: undefined,
336 });
337 return yield* Effect.fail(
338 new AggregateError(
339 [attrsResult.failure, closeResult.failure],
340 `Failed to initialize and stop local Prisma database ${databaseId}`,
341 ),
342 );
343 }
344 return yield* Effect.fail(attrsResult.failure);
345 }
346 const attrs = attrsResult.success;
347 entry = {
348 optionsKey,
349 migrationKey: undefined,
350 server,
351 attrs,
352 };
353 servers.set(databaseId, entry);
354 }
355
356 if (entry.attrs === undefined) {
357 return yield* Effect.fail(
358 new Error(`Local Prisma database ${databaseId} has no usable endpoints.`),
359 );
360 }
361 const attrs = entry.attrs;
362
363 if (config.migrate !== undefined && config.migrate.trim() !== "") {
364 const migrationKey = yield* stableJson({
365 command: config.migrate,
366 cwd: config.migrateCwd,
367 optionsKey,
368 });
369 if (entry.migrationKey !== migrationKey) {
370 yield* runMigration(config, attrs, migrateTimeoutSeconds);
371 entry.migrationKey = migrationKey;
372 }
373 }
374
375 return attrs;
376});
377
378export const closePrismaDevDatabase = Effect.fn(function* (databaseId: string) {
379 const entry = servers.get(databaseId);
380 if (entry === undefined) return;
381 yield* closeEntry(entry);
382 servers.delete(databaseId);
383});
384
385export const closePrismaDevDatabases = Effect.fn(function* () {
386 const ids = Array.from(servers.keys());
387 const failures: Error[] = [];
388 for (const id of ids) {
389 const result = yield* closePrismaDevDatabase(id).pipe(Effect.result);
390 if (Result.isFailure(result)) failures.push(result.failure);
391 }
392 if (failures.length === 1) {
393 return yield* Effect.fail(failures[0]!);
394 }
395 if (failures.length > 1) {
396 return yield* Effect.fail(
397 new AggregateError(
398 failures,
399 `Failed to stop ${failures.length} local Prisma databases`,
400 ),
401 );
402 }
403});