1import type { Contract } from "@prisma/orm-postgres/contract/types";
2import type {
3 ExtractCodecTypes,
4 SqlStorage,
5} from "@prisma/orm-postgres/family-contract/types";
6import type {
7 BindSiteParams,
8 Declaration,
9 ParamsFromDeclaration,
10 PreparedExecution,
11 PreparedFor,
12 PreparedStatement,
13 Runtime,
14 RuntimeQueryable,
15} from "@prisma/orm-postgres/family-runtime";
16import type { CodecTypesBase } from "@prisma/orm-postgres/relational-core/expression";
17import type { SqlQueryPlan } from "@prisma/orm-postgres/relational-core/plan";
18import type { PostgresStaticContext } from "@prisma/orm-postgres/static";
19import * as Effect from "effect/Effect";
20import * as Stream from "effect/Stream";
21import { type ClientError, wrapPrismaError } from "./Errors.ts";
22import type { QueryResult } from "./OrmClient.ts";
23
24type QueryOptions<Params, Row> = Parameters<
25 PreparedStatement<Params, Row>["query"]
26>[2];
27type ExecutionOptions<Params> = Parameters<
28 PreparedExecution<Params>["execute"]
29>[2];
30
31/** A reusable native statement with Effect and Stream consumption. */
32export interface PreparedQuery<Params, Row, E = never, R = never> extends Omit<
33 PreparedStatement<Params, Row>,
34 "query"
35> {
36 query(
37 params: Params,
38 options?: QueryOptions<Params, Row>,
39 ): QueryResult<Row, E, R>;
40}
41
42/** A prepared mutation that reports statistics rather than rows. */
43export interface PreparedMutation<Params, E = never, R = never> extends Omit<
44 PreparedExecution<Params>,
45 "execute"
46> {
47 execute(
48 params: Params,
49 options?: ExecutionOptions<Params>,
50 ): Effect.Effect<
51 Awaited<ReturnType<PreparedExecution<Params>["execute"]>>,
52 E,
53 R
54 >;
55}
56
57export type Prepared<Params, Row, E = never, R = never> =
58 PreparedFor<Params, Row> extends PreparedStatement<Params, Row>
59 ? PreparedQuery<Params, Row, E, R>
60 : PreparedMutation<Params, E, R>;
61
62export interface Prepare<C extends Contract<SqlStorage>, E = never, R = never> {
63 <
64 D extends Declaration<CT>,
65 Row,
66 CT extends CodecTypesBase = ExtractCodecTypes<C>,
67 >(
68 declaration: D,
69 build: (
70 sql: PostgresStaticContext<C>["sql"],
71 params: BindSiteParams<D>,
72 ) => SqlQueryPlan<Row>,
73 ): Effect.Effect<
74 Prepared<ParamsFromDeclaration<D, CT>, Row, ClientError | E, R>,
75 ClientError | E,
76 R
77 >;
78}
79
80export const makePrepare =
81 <C extends Contract<SqlStorage>, E, R>(
82 runtime: Effect.Effect<Runtime, E, R>,
83 sql: PostgresStaticContext<C>["sql"],
84 queryTarget: Effect.Effect<RuntimeQueryable, E, R> = runtime,
85 ): Prepare<C, E, R> =>
86 <
87 D extends Declaration<CT>,
88 Row,
89 CT extends CodecTypesBase = ExtractCodecTypes<C>,
90 >(
91 declaration: D,
92 build: (
93 sql: PostgresStaticContext<C>["sql"],
94 params: BindSiteParams<D>,
95 ) => SqlQueryPlan<Row>,
96 ) =>
97 Effect.gen(function* () {
98 const target = yield* runtime;
99 const statement = yield* Effect.tryPromise({
100 try: () =>
101 target.prepare<D, Row, CT>(declaration, (params) =>
102 build(sql, params),
103 ),
104 catch: wrapPrismaError,
105 });
106 type Params = ParamsFromDeclaration<D, CT>;
107 const metadata = {
108 sql: statement.sql,
109 ast: statement.ast,
110 meta: statement.meta,
111 slots: statement.slots,
112 };
113 const result =
114 "query" in statement
115 ? {
116 ...metadata,
117 query: (params: Params, options?: QueryOptions<Params, Row>) =>
118 Object.assign(
119 Effect.flatMap(queryTarget, (current) =>
120 Effect.tryPromise({
121 try: (signal) =>
122 statement
123 .query(current, params, { ...options, signal })
124 .toArray(),
125 catch: wrapPrismaError,
126 }),
127 ),
128 {
129 stream: Stream.unwrap(
130 Effect.flatMap(queryTarget, (current) =>
131 Effect.try({
132 try: () =>
133 Stream.fromAsyncIterable(
134 statement.query(current, params, options),
135 wrapPrismaError,
136 ),
137 catch: wrapPrismaError,
138 }),
139 ),
140 ),
141 },
142 ),
143 }
144 : {
145 ...metadata,
146 execute: (params: Params, options?: ExecutionOptions<Params>) =>
147 Effect.flatMap(queryTarget, (current) =>
148 Effect.tryPromise({
149 try: (signal) =>
150 statement.execute(current, params, {
151 ...options,
152 signal,
153 }),
154 catch: wrapPrismaError,
155 }),
156 ),
157 };
158 // Prisma selects the consumption method from the plan's affected-count brand.
159 return result as Prepared<Params, Row, ClientError | E, R>;
160 });