1// The Prisma runtime peers are loaded on first use. This module is not
2// re-exported from `alchemy/Prisma`; import `alchemy/Prisma/ORM/Postgres`.
3import type { SqlStorage } from "@prisma/orm-postgres/family-contract/types";
4import type { Contract } from "@prisma/orm-postgres/contract/types";
5import type {
6 Runtime,
7 RuntimeConnection,
8 RuntimeTransaction,
9} from "@prisma/orm-postgres/family-runtime";
10import type {
11 SqlExecutionPlan,
12 SqlQueryPlan,
13} from "@prisma/orm-postgres/relational-core/plan";
14import type {
15 PostgresClient,
16 PostgresOptionsBase,
17} from "@prisma/orm-postgres/runtime";
18import type { PostgresStaticContext } from "@prisma/orm-postgres/static";
19import * as Effect from "effect/Effect";
20import * as Redacted from "effect/Redacted";
21import * as Stream from "effect/Stream";
22import { makeExecutionMemo } from "../../Runtime/ExecutionMemo.ts";
23import { type ClientError, RollbackError, wrapPrismaError } from "./Errors.ts";
24import { type EffectOrm, makeOrmProxy } from "./OrmClient.ts";
25import { type Prepare, makePrepare } from "./Prepared.ts";
26export type {
27 Prepare,
28 Prepared,
29 PreparedQuery,
30 PreparedMutation,
31} from "./Prepared.ts";
32
33export * from "./Errors.ts";
34export type { EffectCollection, EffectOrm, WhereFilter } from "./OrmClient.ts";
35
36/** A Prisma Postgres contract, authored directly or described by emitted types. */
37export type AnyPostgresContract = Contract<SqlStorage>;
38
39/** A plan produced by the `sql` builder lane (`db.sql...build()`) or `raw`. */
40export type Plan<Row> = SqlQueryPlan<Row> | SqlExecutionPlan<Row>;
41
42export type PostgresConfig<C extends AnyPostgresContract> =
43 PostgresOptionsBase &
44 (
45 | {
46 /** The TypeScript contract returned by Prisma's `defineContract`. */
47 readonly contract: C;
48 readonly contractJson?: never;
49 }
50 | {
51 /** Canonical JSON paired with Prisma's emitted Contract declaration. */
52 readonly contractJson: unknown;
53 readonly contract?: never;
54 }
55 );
56
57/**
58 * The transaction scope handed to {@link PostgresDatabase.transaction}'s
59 * callback: the same `orm`/`execute` surfaces bound to the open
60 * transaction, plus a typed `rollback`.
61 */
62export interface PostgresTransaction<C extends AnyPostgresContract> {
63 /** The orm lane, executing on this transaction's connection. */
64 readonly orm: EffectOrm<C>;
65 /** Run a `sql`-lane or `raw` plan on this transaction's connection. */
66 execute<Row>(plan: Plan<Row>): Effect.Effect<Row[], ClientError>;
67 /** Prepare a typed statement whose executions use this transaction. */
68 readonly prepare: Prepare<C>;
69 /**
70 * Abort the transaction: rolls back and fails the `transaction` effect
71 * with {@link RollbackError} (catchable by tag).
72 */
73 rollback(): Effect.Effect<never, RollbackError>;
74}
75
76export interface PostgresDatabase<
77 C extends AnyPostgresContract,
78 E = never,
79 R = never,
80> {
81 /**
82 * The native Prisma client for the current execution. Prefer the Effect
83 * methods for typed failures and repeatable execution.
84 */
85 readonly client: Effect.Effect<PostgresClient<C>, E, R>;
86 /**
87 * Run promise-land code against the execution's client:
88 *
89 * ```typescript
90 * const rows = yield* db.use((c) => c.orm.public.User.groupBy(...)...);
91 * ```
92 */
93 readonly use: <A>(
94 f: (client: PostgresClient<C>) => PromiseLike<A>,
95 ) => Effect.Effect<A, ClientError | E, R>;
96 /**
97 * The Effect-native orm lane — chain like Prisma, yield the terminal:
98 *
99 * ```typescript
100 * const user = yield* db.orm.public.User.where({ email }).include("posts").first();
101 * ```
102 */
103 readonly orm: EffectOrm<C, ClientError | E, R>;
104 /**
105 * The pure `sql` builder lane (no connection — plans are data):
106 * `db.sql.public.user.select("id", "email").build()`.
107 */
108 readonly sql: PostgresStaticContext<C>["sql"];
109 /** The raw SQL lane (pure): ``db.raw.sql`SELECT ...` `` builds plans. */
110 readonly raw: PostgresStaticContext<C>["raw"];
111 /** Execute a plan, buffering all rows. */
112 execute<Row>(plan: Plan<Row>): Effect.Effect<Row[], ClientError | E, R>;
113 /** Execute a plan as a row stream (each run re-executes the plan). */
114 stream<Row>(plan: Plan<Row>): Stream.Stream<Row, ClientError | E, R>;
115 /** Prepare a typed statement; each query resolves the current execution's client. */
116 readonly prepare: Prepare<C, E, R>;
117 /**
118 * Run `f` inside a database transaction on a dedicated connection.
119 * Commits on success; rolls back on failure or interruption. Yield
120 * `tx.rollback()` to abort with a typed {@link RollbackError}.
121 */
122 transaction<A, E2, R2>(
123 f: (tx: PostgresTransaction<C>) => Effect.Effect<A, E2, R2>,
124 ): Effect.Effect<A, E2 | ClientError | E, R | R2>;
125}
126
127/**
128 * Open a Prisma ORM v8 Postgres client from a connection URL, with
129 * Effect-native query surfaces over Prisma's own builders and engine.
130 *
131 * The client is built at most once per execution — a Worker
132 * `fetch`/`queue`/`scheduled` event, a Durable Object call, a Workflow run,
133 * or a Lambda invocation — and memoized on the execution's `Scope` (via
134 * {@link makeExecutionMemo}), with `close()` registered as a scope finalizer
135 * so the underlying `pg` pool never outlives its event. That per-execution
136 * lifecycle is what makes the client safe on workerd, where sockets are
137 * pinned to the creating request's IoContext. Construction does no I/O
138 * (Prisma connects lazily on the first query), so deploy/plan-time
139 * evaluations never touch the database.
140 *
141 * Model types infer directly from the native TypeScript contract, together
142 * with the connection source's error and requirement channels. No generated
143 * application imports are required. For Prisma rc.11, author contracts with
144 * `defineContract` from `alchemy/Prisma/ORM` to retain metadata
145 * lost by the upstream declarations. It uses Prisma's native runtime builders.
146 *
147 *
148 * ```typescript
149 * import * as PrismaPostgres from "alchemy/Prisma/ORM/Postgres";
150 * import { contract } from "./prisma/contract.ts";
151 *
152 * const connection = yield* Cloudflare.Hyperdrive.Connect(hyperdrive);
153 * const db = yield* PrismaPostgres.Postgres(
154 * connection.connectionString,
155 * { contract },
156 * );
157 *
158 * fetch: Effect.gen(function* () {
159 * // orm lane — queries ARE Effects, with typed errors
160 * const user = yield* db.orm.public.User.where({ email }).include("posts").first();
161 * const made = yield* db.orm.public.Post.create({ title, authorId: user.id });
162 *
163 * // sql builder lane — pure plans, Effect executor
164 * const rows = yield* db.execute(db.sql.public.user.select("id", "email").build());
165 *
166 * // transactions — commit on success, rollback on failure/interrupt
167 * yield* db.transaction((tx) =>
168 * Effect.gen(function* () {
169 * const u = yield* tx.orm.public.User.create({ email });
170 * if (!u) return yield* tx.rollback();
171 * return u;
172 * }),
173 * );
174 * });
175 * ```
176 *
177 * ### PSL Contracts
178 * **Example:** Use a generated contract-bound factory
179 * ```typescript
180 * import { makeDatabase } from "./prisma/generated/client.ts";
181 * const db = yield* makeDatabase(connection.connectionString);
182 * const users = yield* db.orm.public.User.select("id", "email").all();
183 * ```
184 *
185 * Run `alchemy prisma generate` with `withEffect` registered in the ORM
186 * configuration. The factory delegates to this runtime with Prisma's emitted
187 * `Contract` type and `contractJson`; query behavior and cleanup are identical.
188 * Generated `schemas.ts` can be imported independently of this client.
189 *
190 * ### Prepared Queries
191 * **Example:** Reuse a query with typed parameters
192 * ```typescript
193 * const findUser = yield* db.prepare({ email: "pg/text@1" }, (sql, params) =>
194 * sql.public.user
195 * .select("id", "email")
196 * .where((fields, fns) => fns.eq(fields.email, params.email))
197 * .build(),
198 * );
199 * const users = yield* findUser.query({ email: "alice@example.com" });
200 * const rows = findUser.query({ email: "alice@example.com" }).stream;
201 * ```
202 *
203 * Prepared queries resolve the current execution's client for each run. Inside
204 * `db.transaction`, `tx.prepare` binds executions to that transaction instead.
205 * An affected-count plan returns a prepared mutation with `execute(params)`;
206 * a row-returning plan returns a prepared query with `query(params)`.
207 *
208 * The Stream surfaces use Prisma's async iterators. The long-lived Postgres
209 * driver buffers raw results, so these streams do not guarantee bounded memory.
210 *
211 * Queries are lazy and re-runnable: each evaluation replays the chain
212 * against the execution's client, so `Effect.retry` re-issues the query.
213 * Failures surface as granular tagged errors: the SQL-standard integrity
214 * violations each get their own tag (`Prisma.UniqueViolationError`,
215 * `Prisma.ForeignKeyViolationError`, `Prisma.NotNullViolationError`,
216 * `Prisma.CheckViolationError`), other statement failures are
217 * `Prisma.QueryError` (with the normalized `sqlState`), connection
218 * failures are `Prisma.ConnectionError` (with the driver's
219 * `transient` verdict), and Prisma's structured codes split by
220 * category into `Prisma.OrmError` / `Prisma.RuntimeError`
221 * with an autocompleting `code` field.
222 *
223 * @binding
224 * @category ORM
225 */
226export const Postgres = <C extends AnyPostgresContract, E = never, R = never>(
227 connectionString: Effect.Effect<Redacted.Redacted<string>, E, R>,
228 config: PostgresConfig<C>,
229): Effect.Effect<PostgresDatabase<C, E, R>> =>
230 Effect.gen(function* () {
231 const { contract, contractJson, ...options } = config;
232 const [{ default: postgres }, { default: postgresStatic }, { orm }] =
233 yield* Effect.promise(() =>
234 Promise.all([
235 import("@prisma/orm-postgres/runtime"),
236 import("@prisma/orm-postgres/static"),
237 import("@prisma/orm-postgres/orm-client"),
238 ]),
239 );
240
241 // Pure static context: the typed sql/raw builders and the codec
242 // machinery, with no driver and no connection behind them.
243 const statics = postgresStatic<C>({
244 contractJson: contract ?? contractJson,
245 ...(options.extensions === undefined
246 ? {}
247 : { extensions: options.extensions }),
248 });
249
250 const client = yield* makeExecutionMemo(
251 Effect.gen(function* () {
252 const url = Redacted.value(yield* connectionString);
253 const instance =
254 contract === undefined
255 ? postgres<C>({ contractJson, url, ...options })
256 : postgres({ contract, url, ...options });
257 yield* Effect.addFinalizer(() =>
258 Effect.tryPromise(() => instance.close()).pipe(Effect.ignore),
259 );
260 return instance;
261 }),
262 );
263
264 const execute = <Row>(
265 plan: Plan<Row>,
266 ): Effect.Effect<Row[], ClientError | E, R> =>
267 Effect.flatMap(client, (c) =>
268 Effect.tryPromise({
269 try: (signal) => c.runtime().query(plan, { signal }).toArray(),
270 catch: wrapPrismaError,
271 }),
272 );
273
274 const stream = <Row>(
275 plan: Plan<Row>,
276 ): Stream.Stream<Row, ClientError | E, R> =>
277 // unwrap re-evaluates per run, so each run gets a fresh
278 // AsyncIterableResult (they are single-consumption).
279 Stream.unwrap(
280 Effect.map(client, (c) =>
281 Stream.fromAsyncIterable(
282 c.runtime().query(plan) as AsyncIterable<Row>,
283 wrapPrismaError,
284 ),
285 ),
286 );
287
288 const makeTransactionScope = (
289 txn: RuntimeTransaction,
290 runtime: Runtime,
291 ) => {
292 const txOrm = orm<C>({
293 runtime: txn,
294 context: statics.context,
295 });
296 const tx: PostgresTransaction<C> = {
297 orm: makeOrmProxy<C, never, never>(Effect.sync(() => txOrm)),
298 prepare: makePrepare(
299 Effect.succeed(runtime),
300 statics.sql,
301 Effect.succeed(txn),
302 ),
303 execute: <Row>(plan: Plan<Row>) =>
304 Effect.tryPromise({
305 try: (signal) => txn.query(plan, { signal }).toArray(),
306 catch: wrapPrismaError,
307 }),
308 rollback: () => Effect.fail(new RollbackError()),
309 };
310 return tx;
311 };
312
313 const transaction = <A, E2, R2>(
314 f: (tx: PostgresTransaction<C>) => Effect.Effect<A, E2, R2>,
315 ): Effect.Effect<A, E2 | ClientError | E, R | R2> =>
316 Effect.flatMap(client, (c) =>
317 Effect.acquireUseRelease(
318 Effect.tryPromise({
319 try: async () => {
320 const connection: RuntimeConnection = await c
321 .runtime()
322 .connection();
323 try {
324 const txn = await connection.transaction();
325 return { connection, txn, committed: false };
326 } catch (error) {
327 await connection.destroy(error).catch(() => {});
328 throw error;
329 }
330 },
331 catch: wrapPrismaError,
332 }),
333 (scope) =>
334 f(makeTransactionScope(scope.txn, c.runtime())).pipe(
335 Effect.flatMap((value) =>
336 Effect.tryPromise({
337 try: async () => {
338 await scope.txn.commit();
339 scope.committed = true;
340 return value;
341 },
342 catch: wrapPrismaError,
343 }),
344 ),
345 ),
346 (scope) =>
347 // Rollback on any non-committed exit (failure, interrupt, or
348 // commit failure), then return the connection to the pool —
349 // destroying it if cleanup itself failed.
350 Effect.promise(async () => {
351 try {
352 if (!scope.committed) await scope.txn.rollback();
353 await scope.connection.release();
354 } catch (error) {
355 await scope.connection.destroy(error).catch(() => {});
356 }
357 }),
358 ),
359 );
360
361 const db: PostgresDatabase<C, E, R> = {
362 client,
363 use: (f) =>
364 Effect.flatMap(client, (c) =>
365 Effect.tryPromise({
366 try: () => Promise.resolve(f(c)),
367 catch: wrapPrismaError,
368 }),
369 ),
370 orm: makeOrmProxy<C, ClientError | E, R>(
371 Effect.map(client, (c) => c.orm),
372 ),
373 sql: statics.sql,
374 raw: statics.raw,
375 prepare: makePrepare(
376 Effect.map(client, (c) => c.runtime()),
377 statics.sql,
378 ),
379 execute,
380 stream,
381 transaction,
382 };
383 return db;
384 });