1/**
2 * Schema-aware `RequestResolver` helpers for SQL-backed data loading.
3 *
4 * This module represents each lookup or mutation as a `SqlRequest` and batches
5 * concurrent requests into SQL operations. Request payloads are encoded with the
6 * request schema before `execute` is called, and returned rows are decoded with
7 * the result schema before requests are completed. It provides ordered,
8 * grouped, id-based, and side-effect-only resolver constructors, and keeps
9 * batches separated by the active SQL transaction connection.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Arr from "../Array.ts"
15import * as Cause from "../Cause.ts"
16import * as Context from "../Context.ts"
17import * as Effect from "../Effect.ts"
18import * as Equal from "../Equal.ts"
19import * as Exit from "../Exit.ts"
20import * as Hash from "../Hash.ts"
21import * as MutableHashMap from "../MutableHashMap.ts"
22import * as Option from "../Option.ts"
23import * as Request from "../Request.ts"
24import * as RequestResolver from "../RequestResolver.ts"
25import * as Schema from "../Schema.ts"
26import type * as Types from "../Types.ts"
27import * as SqlClient from "./SqlClient.ts"
28import { ResultLengthMismatch } from "./SqlError.ts"
29
30/**
31 * Request type used by SQL request resolvers, carrying the input payload
32 * together with the resolver's result, error, and environment types.
33 *
34 * @stability unstable
35 * @category models
36 * @since 4.0.0
37 */
38export interface SqlRequest<In, A, E, R> extends Request.Request<A, E | Schema.SchemaError, R> {
39 readonly payload: In
40}
41
42const SqlRequestProto = {
43 ...Request.Class.prototype,
44 [Equal.symbol](
45 this: SqlRequest<any, any, any, any>,
46 that: SqlRequest<any, any, any, any>
47 ): boolean {
48 return Equal.equals(this.payload, that.payload)
49 },
50 [Hash.symbol](this: SqlRequest<any, any, any, any>): number {
51 return Hash.hash(this.payload)
52 }
53}
54
55/**
56 * Runs a payload as a `SqlRequest` through a request resolver, either directly
57 * with a payload and resolver or curried by resolver.
58 *
59 * @stability unstable
60 * @category running
61 * @since 4.0.0
62 */
63export const request: {
64 <In, A, E, R>(
65 resolver: RequestResolver.RequestResolver<SqlRequest<In, A, E, R>>
66 ): (payload: In) => Effect.Effect<A, E | Schema.SchemaError, R>
67 <In, A, E, R>(
68 payload: In,
69 resolver: RequestResolver.RequestResolver<SqlRequest<In, A, E, R>>
70 ): Effect.Effect<A, E | Schema.SchemaError, R>
71} = function() {
72 if (arguments.length === 1) {
73 const resolver = arguments[0]
74 return (payload: any) => Effect.request(SqlRequest(payload), resolver)
75 }
76 return Effect.request(SqlRequest(arguments[0]), arguments[1])
77} as any
78
79/**
80 * Constructs a `SqlRequest` from a payload. Equality and hashing are based on
81 * the payload so equal requests can be batched and deduplicated.
82 *
83 * @stability unstable
84 * @category constructors
85 * @since 4.0.0
86 */
87export const SqlRequest = <In, A, E, R>(payload: In): SqlRequest<In, A, E, R> => {
88 const self = Object.create(SqlRequestProto)
89 self.payload = payload
90 return self
91}
92
93/**
94 * Creates a resolver for a SQL query with a request schema and a result schema.
95 *
96 * **Details**
97 *
98 * The request schema is used to validate the input of the query, and the result
99 * schema is used to validate the output of the query. Results are mapped to the
100 * requests in order, so the length of the results must match the length of the
101 * requests.
102 *
103 * @stability unstable
104 * @category resolvers
105 * @since 4.0.0
106 */
107export const ordered = <Req extends Schema.Constraint, Res extends Schema.Constraint, _, E, R>(
108 options: {
109 readonly Request: Req
110 readonly Result: Res
111 readonly execute: (
112 requests: Arr.NonEmptyArray<Req["Encoded"]>
113 ) => Effect.Effect<ReadonlyArray<_>, E, R>
114 }
115): RequestResolver.RequestResolver<
116 SqlRequest<Req["Type"], Res["Type"], E | ResultLengthMismatch, Req["EncodingServices"] | Res["DecodingServices"] | R>
117> => {
118 const decodeArray = Schema.decodeUnknownEffect(Schema.Array(options.Result))
119 return RequestResolver.makeGrouped<
120 SqlRequest<
121 Req["Type"],
122 Res["Type"],
123 E | ResultLengthMismatch,
124 Req["EncodingServices"] | Res["DecodingServices"] | R
125 >,
126 SqlClient.TransactionConnection.Service | undefined
127 >({
128 key: transactionKey,
129 resolver: Effect.fnUntraced(function*(entries) {
130 const [inputs, encodedEntries] = yield* partitionRequests(entries, options.Request)
131 if (!Arr.isArrayNonEmpty(inputs)) return
132 const results = yield* options.execute(inputs).pipe(
133 Effect.provideContext(entries[0].context)
134 )
135 if (results.length !== inputs.length) {
136 return yield* new ResultLengthMismatch({ expected: inputs.length, actual: results.length })
137 }
138 const decodedResults = yield* decodeArray(results).pipe(
139 Effect.provideContext(entries[0].context)
140 )
141 for (let i = 0; i < encodedEntries.length; i++) {
142 encodedEntries[i].completeUnsafe(Exit.succeed(decodedResults[i]))
143 }
144 })
145 })
146}
147
148/**
149 * Creates a batched SQL request resolver that encodes requests, decodes result
150 * rows, groups decoded results by matching request and result keys, and fails a
151 * request with `NoSuchElementError` when no result group exists.
152 *
153 * @stability unstable
154 * @category resolvers
155 * @since 4.0.0
156 */
157export const grouped = <Req extends Schema.Constraint, Res extends Schema.Constraint, K, Row, E, R>(
158 options: {
159 readonly Request: Req
160 readonly RequestGroupKey: (request: Req["Type"]) => K
161 readonly Result: Res
162 readonly ResultGroupKey: (result: Res["Type"], row: Types.NoInfer<Row>) => K
163 readonly execute: (
164 requests: Arr.NonEmptyArray<Req["Encoded"]>
165 ) => Effect.Effect<ReadonlyArray<Row>, E, R>
166 }
167): RequestResolver.RequestResolver<
168 SqlRequest<
169 Req["Type"],
170 Arr.NonEmptyArray<Res["Type"]>,
171 E | Schema.SchemaError | Cause.NoSuchElementError,
172 Req["EncodingServices"] | Res["DecodingServices"] | R
173 >
174> => {
175 const decodeResults = Schema.decodeUnknownEffect(Schema.Array(options.Result))
176
177 return RequestResolver.makeGrouped<
178 SqlRequest<
179 Req["Type"],
180 Arr.NonEmptyArray<Res["Type"]>,
181 E | Schema.SchemaError | Cause.NoSuchElementError,
182 Req["EncodingServices"] | Res["DecodingServices"] | R
183 >,
184 SqlClient.TransactionConnection.Service | undefined
185 >({
186 key: transactionKey,
187 resolver: Effect.fnUntraced(function*(entries) {
188 const [inputs] = yield* partitionRequests(entries, options.Request)
189 if (!Arr.isArrayNonEmpty(inputs)) return
190 const resultMap = MutableHashMap.empty<K, Arr.NonEmptyArray<Res["Type"]>>()
191 const results = yield* options.execute(inputs).pipe(
192 Effect.provideContext(entries[0].context)
193 )
194 const decodedResults = yield* decodeResults(results).pipe(
195 Effect.provideContext(entries[0].context)
196 )
197 for (let i = 0, len = decodedResults.length; i < len; i++) {
198 const result = decodedResults[i]
199 const key = options.ResultGroupKey(result, results[i])
200 const group = MutableHashMap.get(resultMap, key)
201 if (group._tag === "None") {
202 MutableHashMap.set(resultMap, key, [result])
203 } else {
204 group.value.push(result)
205 }
206 }
207 for (let i = 0, len = entries.length; i < len; i++) {
208 const entry = entries[i]
209 const key = options.RequestGroupKey(entry.request.payload)
210 const result = MutableHashMap.get(resultMap, key)
211 entry.completeUnsafe(
212 result._tag === "None" ? constNoSuchElement : Exit.succeed(result.value)
213 )
214 }
215 })
216 })
217}
218
219/**
220 * Creates a batched resolver that fetches rows for encoded ids, decodes
221 * results, completes each matching request using `ResultId`, and fails missing
222 * ids with `NoSuchElementError`.
223 *
224 * @stability unstable
225 * @category resolvers
226 * @since 4.0.0
227 */
228export const findById = <Id extends Schema.Constraint, Res extends Schema.Constraint, Row, E, R>(
229 options: {
230 readonly Id: Id
231 readonly Result: Res
232 readonly ResultId: (result: Res["Type"], row: Types.NoInfer<Row>) => Id["Type"]
233 readonly execute: (
234 requests: Arr.NonEmptyArray<Id["Encoded"]>
235 ) => Effect.Effect<ReadonlyArray<Row>, E, R>
236 }
237): RequestResolver.RequestResolver<
238 SqlRequest<
239 Id["Type"],
240 Res["Type"],
241 E | Schema.SchemaError | Cause.NoSuchElementError,
242 Id["EncodingServices"] | Res["DecodingServices"] | R
243 >
244> => {
245 const decodeResults = Schema.decodeUnknownEffect(Schema.Array(options.Result))
246
247 return RequestResolver.makeGrouped<
248 SqlRequest<
249 Id["Type"],
250 Res["Type"],
251 E | Schema.SchemaError | Cause.NoSuchElementError,
252 Id["EncodingServices"] | Res["DecodingServices"] | R
253 >,
254 SqlClient.TransactionConnection.Service | undefined
255 >({
256 key: transactionKey,
257 resolver: Effect.fnUntraced(function*(entries) {
258 const [inputs, idMap] = yield* partitionRequestsById(entries, options.Id)
259 if (!Arr.isArrayNonEmpty(inputs)) return
260 const results = yield* options.execute(inputs).pipe(
261 Effect.provideContext(entries[0].context)
262 )
263 const decodedResults = yield* decodeResults(results).pipe(
264 Effect.provideContext(entries[0].context)
265 )
266 for (let i = 0; i < decodedResults.length; i++) {
267 const result = decodedResults[i]
268 const id = options.ResultId(result, results[i])
269 const request = MutableHashMap.get(idMap, id)
270 if (request._tag === "None") {
271 continue
272 }
273 MutableHashMap.remove(idMap, id)
274 request.value.completeUnsafe(Exit.succeed(result))
275 }
276 if (MutableHashMap.isEmpty(idMap)) {
277 return
278 }
279 MutableHashMap.forEach(idMap, (request) => {
280 request.completeUnsafe(constNoSuchElement)
281 })
282 })
283 })
284}
285
286const void_ = <Req extends Schema.Constraint, _, E, R>(
287 options: {
288 readonly Request: Req
289 readonly execute: (
290 requests: Arr.NonEmptyArray<Req["Encoded"]>
291 ) => Effect.Effect<ReadonlyArray<_>, E, R>
292 }
293): RequestResolver.RequestResolver<
294 SqlRequest<
295 Req["Type"],
296 void,
297 E | Schema.SchemaError,
298 Req["EncodingServices"] | R
299 >
300> =>
301 RequestResolver.makeGrouped<
302 SqlRequest<
303 Req["Type"],
304 void,
305 E | Schema.SchemaError,
306 Req["EncodingServices"] | R
307 >,
308 SqlClient.TransactionConnection.Service | undefined
309 >({
310 key: transactionKey,
311 resolver: Effect.fnUntraced(function*(entries) {
312 const [inputs] = yield* partitionRequests(entries, options.Request)
313 if (!Arr.isArrayNonEmpty(inputs)) return
314 yield* options.execute(inputs).pipe(
315 Effect.provideContext(entries[0].context)
316 )
317 for (let i = 0; i < entries.length; i++) {
318 entries[i].completeUnsafe(Exit.void)
319 }
320 })
321 })
322
323/**
324 * @stability unstable
325 */
326export {
327 /**
328 * Create a resolver that performs side effects.
329 *
330 * @stability unstable
331 * @category resolvers
332 * @since 4.0.0
333 */
334 void_ as void
335}
336
337const constNoSuchElement = Exit.fail(new Cause.NoSuchElementError())
338
339const partitionRequests = function*<In, A, E, R, InE>(
340 requests: Arr.NonEmptyArray<Request.Entry<SqlRequest<In, A, E, R>>>,
341 schema: Schema.ConstraintCodec<In, InE, R, R>
342) {
343 const len = requests.length
344 const inputs = Arr.empty<InE>()
345 const encodedEntries = Arr.empty<Request.Entry<SqlRequest<In, A, E, R>>>()
346 let entry!: Request.Entry<SqlRequest<In, A, E, R>>
347 const encode = Schema.encodeEffect(schema)
348 const handle = Effect.matchCauseEager({
349 onFailure(cause: Cause.Cause<Schema.SchemaError>) {
350 entry.completeUnsafe(Exit.failCause(cause))
351 },
352 onSuccess(value: InE) {
353 inputs.push(value)
354 encodedEntries.push(entry)
355 }
356 })
357
358 for (let i = 0; i < len; i++) {
359 entry = requests[i]
360 yield (Effect.provideContext(handle(encode(entry.request.payload)), entry.context) as Effect.Effect<void>)
361 }
362
363 return [inputs, encodedEntries] as const
364}
365
366const partitionRequestsById = function*<In, A, E, R, InE>(
367 requests: ReadonlyArray<Request.Entry<SqlRequest<In, A, E, R>>>,
368 schema: Schema.ConstraintCodec<In, InE, R, R>
369) {
370 const len = requests.length
371 const inputs = Arr.empty<InE>()
372 const byIdMap = MutableHashMap.empty<In, Request.Entry<SqlRequest<In, A, E, R>>>()
373
374 for (let i = 0; i < len; i++) {
375 const entry = requests[i]
376 const existing = MutableHashMap.get(byIdMap, entry.request.payload)
377 if (Option.isSome(existing)) {
378 const previous = existing.value
379 MutableHashMap.set(byIdMap, entry.request.payload, {
380 ...previous,
381 completeUnsafe(exit) {
382 previous.completeUnsafe(exit)
383 entry.completeUnsafe(exit)
384 }
385 })
386 } else {
387 MutableHashMap.set(byIdMap, entry.request.payload, entry)
388 }
389 }
390
391 const encode = Schema.encodeEffect(schema)
392 for (const [, entry] of byIdMap) {
393 yield* Effect.provideContext(
394 Effect.matchCauseEager(encode(entry.request.payload), {
395 onFailure(cause) {
396 entry.completeUnsafe(Exit.failCause(cause))
397 },
398 onSuccess(value) {
399 inputs.push(value)
400 }
401 }),
402 entry.context
403 )
404 }
405
406 return [inputs, byIdMap] as const
407}
408
409function transactionKey<A>(entry: Request.Entry<A>): SqlClient.TransactionConnection.Service | undefined {
410 const client = Context.getOrUndefined(entry.context, SqlClient.SqlClient)
411 if (!client) return undefined
412 const conn = Context.getOrUndefined(entry.context, client.transactionService)
413 if (!conn) return undefined
414 return Equal.byReferenceUnsafe(conn)
415}