1/**
2 * Connects Effect SQL to PGlite, the embedded PostgreSQL-compatible database
3 * from `@electric-sql/pglite`.
4 *
5 * This module can create a managed PGlite instance or wrap an existing one and
6 * expose it as both `PgliteClient` and the generic Effect SQL client. The client
7 * runs PostgreSQL-style SQL, adds helpers for JSON values and LISTEN/NOTIFY
8 * messages, can dump the PGlite data directory, and can refresh PGlite array
9 * types. It also provides layers and maps common PostgreSQL-style failures into
10 * Effect SQL errors.
11 *
12 * @stability unstable
13 * @since 4.0.0
14 */
15import { PGlite, type PGliteInterface, type PGliteOptions } from "@electric-sql/pglite"
16import * as Config from "effect/Config"
17import * as Context from "effect/Context"
18import * as Effect from "effect/Effect"
19import * as Fiber from "effect/Fiber"
20import * as Layer from "effect/Layer"
21import * as Queue from "effect/Queue"
22import * as Reactivity from "effect/reactivity/Reactivity"
23import * as Scope from "effect/Scope"
24import * as Semaphore from "effect/Semaphore"
25import * as Client from "effect/sql/SqlClient"
26import type { Connection } from "effect/sql/SqlConnection"
27import {
28 AuthenticationError,
29 AuthorizationError,
30 ConnectionError,
31 ConstraintError,
32 DeadlockError,
33 LockTimeoutError,
34 SerializationError,
35 SqlError,
36 SqlSyntaxError,
37 StatementTimeoutError,
38 UniqueViolation,
39 UnknownError
40} from "effect/sql/SqlError"
41import type { Custom, Fragment } from "effect/sql/Statement"
42import * as Statement from "effect/sql/Statement"
43import * as Stream from "effect/Stream"
44
45/**
46 * Runtime type identifier used to mark `PgliteClient` values.
47 *
48 * @stability unstable
49 * @category type IDs
50 * @since 4.0.0
51 */
52export const TypeId: TypeId = "~@effect/sql-pglite/PgliteClient"
53
54/**
55 * Type-level identifier used to mark `PgliteClient` values.
56 *
57 * @stability unstable
58 * @category type IDs
59 * @since 4.0.0
60 */
61export type TypeId = "~@effect/sql-pglite/PgliteClient"
62
63/**
64 * PGlite-backed PostgreSQL client service, extending `SqlClient` with access to the PGlite instance, JSON fragments, LISTEN/NOTIFY, data directory dumps, and array type refresh.
65 *
66 * @stability unstable
67 * @category services
68 * @since 4.0.0
69 */
70export interface PgliteClient extends Client.SqlClient {
71 readonly [TypeId]: TypeId
72 readonly config: PgliteClientConfig
73 readonly pglite: PGliteInterface
74 readonly json: (_: unknown) => Fragment
75 /**
76 * Subscribes to a PGlite notification channel.
77 *
78 * **Details**
79 *
80 * The effect completes after the listener is installed. Notifications are
81 * buffered in the returned dequeue, and the subscription remains active
82 * until the required scope closes.
83 */
84 readonly listen: (
85 channel: string
86 ) => Effect.Effect<Queue.Dequeue<string>, SqlError, Scope.Scope>
87 readonly notify: (channel: string, payload: string) => Effect.Effect<void, SqlError>
88 readonly dumpDataDir: (compression?: "none" | "gzip" | "auto") => Effect.Effect<File | Blob, SqlError>
89 readonly refreshArrayTypes: Effect.Effect<void, SqlError>
90}
91
92/**
93 * Service tag for the PGlite client service.
94 *
95 * **When to use**
96 *
97 * Use to access or provide a PGlite client through the Effect context.
98 *
99 * @stability unstable
100 * @category services
101 * @since 4.0.0
102 */
103export const PgliteClient = Context.Service<PgliteClient>("@effect/sql-pglite/PgliteClient")
104
105/**
106 * Configuration for a PGlite client, either by supplying PGlite creation options or an existing live PGlite client.
107 *
108 * @stability unstable
109 * @category models
110 * @since 4.0.0
111 */
112export type PgliteClientConfig = PgliteClientConfig.Create | PgliteClientConfig.Live
113
114/**
115 * Namespace containing the configuration variants for `PgliteClient`.
116 *
117 * @stability unstable
118 * @since 4.0.0
119 */
120export declare namespace PgliteClientConfig {
121 /**
122 * Shared PGlite client options for span attributes, query/result name transformations, and JSON value transformation.
123 *
124 * @stability unstable
125 * @category models
126 * @since 4.0.0
127 */
128 export interface Base {
129 readonly spanAttributes?: Record<string, unknown> | undefined
130 readonly transformResultNames?: ((str: string) => string) | undefined
131 readonly transformQueryNames?: ((str: string) => string) | undefined
132 readonly transformJson?: boolean | undefined
133 }
134
135 /**
136 * Configuration used to create a managed PGlite instance from PGlite constructor options.
137 *
138 * @stability unstable
139 * @category models
140 * @since 4.0.0
141 */
142 export interface Create extends Base, PGliteOptions {}
143
144 /**
145 * Configuration that uses an existing PGlite client. The supplied `liveClient` is caller-owned and is not closed by the Effect client.
146 *
147 * @stability unstable
148 * @category models
149 * @since 4.0.0
150 */
151 export interface Live extends Base {
152 readonly liveClient: PGliteInterface
153 }
154
155 /**
156 * Config-friendly subset of PGlite creation options, including data directory, username, database, relaxed durability, and shared transform options.
157 *
158 * @stability unstable
159 * @category models
160 * @since 4.0.0
161 */
162 export interface ConfigBase extends Base {
163 readonly dataDir?: string | undefined
164 readonly username?: string | undefined
165 readonly database?: string | undefined
166 readonly relaxedDurability?: boolean | undefined
167 }
168}
169
170/**
171 * Creates a scoped PGlite SQL client. When no live client is supplied it creates and closes a PGlite instance; when `liveClient` is supplied, the caller retains ownership.
172 *
173 * @stability unstable
174 * @category constructors
175 * @since 4.0.0
176 */
177export const make = (
178 options: PgliteClientConfig = {}
179): Effect.Effect<PgliteClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
180 Effect.gen(function*() {
181 const pglite = "liveClient" in options
182 ? options.liveClient
183 : yield* Effect.acquireRelease(
184 Effect.tryPromise({
185 try: async () => {
186 const pg = new PGlite(options)
187 await pg.waitReady
188 return pg
189 },
190 catch: (cause) => new SqlError({ reason: classifyError(cause, "PgliteClient: Failed to connect", "connect") })
191 }),
192 (pg) => Effect.promise(() => pg.close()).pipe(Effect.timeoutOption(1000)),
193 { interruptible: true }
194 )
195
196 return yield* fromClient({ ...options, liveClient: pglite })
197 })
198
199/**
200 * Builds a `PgliteClient` around an existing PGlite instance, adding SQL client operations, LISTEN/NOTIFY, dump helpers, and serialized access.
201 *
202 * @stability unstable
203 * @category constructors
204 * @since 4.0.0
205 */
206export const fromClient = (
207 options:
208 & PgliteClientConfig.Base
209 & {
210 readonly liveClient: PGliteInterface
211 }
212): Effect.Effect<PgliteClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
213 Effect.gen(function*() {
214 const pglite = options.liveClient
215 const compiler = makeCompiler(options.transformQueryNames, options.transformJson)
216 const transformRows = options.transformResultNames
217 ? Statement.defaultTransforms(options.transformResultNames, options.transformJson).array
218 : undefined
219
220 const spanAttributes: Array<[string, unknown]> = [
221 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
222 [ATTR_DB_SYSTEM_NAME, "postgresql"]
223 ]
224
225 const connection = new PgliteConnection(pglite)
226 const semaphore = Semaphore.makeUnsafe(1)
227 const acquirer = semaphore.withPermits(1)(Effect.succeed(connection))
228 const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
229 const fiber = Fiber.getCurrent()!
230 const scope = Context.getUnsafe(fiber.context, Scope.Scope)
231 return Effect.as(
232 Effect.tap(
233 restore(semaphore.take(1)),
234 () => Scope.addFinalizer(scope, semaphore.release(1))
235 ),
236 connection
237 )
238 })
239
240 const config = options as PgliteClientConfig
241 const client = yield* Client.make({
242 acquirer,
243 compiler,
244 transactionAcquirer,
245 releaseSavepoint: (name) => `RELEASE SAVEPOINT ${name}`,
246 spanAttributes,
247 transformRows
248 })
249
250 return Object.assign(
251 client,
252 {
253 [TypeId]: TypeId as TypeId,
254 config,
255 pglite,
256 json: (_: unknown) => Statement.fragment([PgJson(_)]),
257 listen: Effect.fnUntraced(function*(channel: string) {
258 const queue = yield* Queue.unbounded<string>()
259 yield* Effect.acquireRelease(
260 Effect.tryPromise({
261 try: () =>
262 pglite.listen(channel, (payload) => {
263 Queue.offerUnsafe(queue, payload)
264 }),
265 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to listen", "listen") })
266 }),
267 (unlisten) => Effect.promise(() => unlisten()),
268 { interruptible: true }
269 )
270 yield* Effect.addFinalizer(() => Queue.shutdown(queue))
271 return queue
272 }),
273 notify: (channel: string, payload: string) =>
274 Effect.tryPromise({
275 try: () => pglite.exec(`NOTIFY ${escape(channel)}, ${escapeLiteral(payload)}`),
276 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to notify", "notify") })
277 }).pipe(
278 semaphore.withPermit,
279 Effect.asVoid
280 ),
281 dumpDataDir: (compression?: "none" | "gzip" | "auto") =>
282 semaphore.withPermit(
283 Effect.tryPromise({
284 try: () => pglite.dumpDataDir(compression),
285 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to dump data dir", "dumpDataDir") })
286 })
287 ),
288 refreshArrayTypes: semaphore.withPermit(
289 Effect.tryPromise({
290 try: () => pglite.refreshArrayTypes(),
291 catch: (cause) =>
292 new SqlError({ reason: classifyError(cause, "Failed to refresh array types", "refreshArrayTypes") })
293 })
294 )
295 }
296 )
297 })
298
299class PgliteConnection implements Connection {
300 readonly pglite: PGliteInterface
301 constructor(pglite: PGliteInterface) {
302 this.pglite = pglite
303 }
304
305 private run(method: string, sql: string, params: ReadonlyArray<unknown>) {
306 return Effect.map(
307 Effect.tryPromise({
308 try: () => this.pglite.query<any>(sql, params as Array<any>),
309 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", method) })
310 }),
311 (result) => result.rows
312 )
313 }
314 execute(
315 sql: string,
316 params: ReadonlyArray<unknown>,
317 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
318 ) {
319 return transformRows
320 ? Effect.map(this.run("execute", sql, params), transformRows)
321 : this.run("execute", sql, params)
322 }
323 executeRaw(sql: string, params: ReadonlyArray<unknown>) {
324 return Effect.tryPromise({
325 try: () => this.pglite.query<any>(sql, params as Array<any>),
326 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", "executeRaw") })
327 })
328 }
329 executeValues(sql: string, params: ReadonlyArray<unknown>) {
330 return Effect.map(
331 Effect.tryPromise({
332 try: () => this.pglite.query<any>(sql, params as Array<any>, { rowMode: "array" }),
333 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", "executeValues") })
334 }),
335 (result) => result.rows as ReadonlyArray<ReadonlyArray<any>>
336 )
337 }
338 executeValuesUnprepared(sql: string, params: ReadonlyArray<unknown>) {
339 return Effect.map(
340 Effect.tryPromise({
341 try: () => this.pglite.query<any>(sql, params as Array<any>, { rowMode: "array" }),
342 catch: (cause) =>
343 new SqlError({ reason: classifyError(cause, "Failed to execute statement", "executeValuesUnprepared") })
344 }),
345 (result) => result.rows as ReadonlyArray<ReadonlyArray<any>>
346 )
347 }
348 executeUnprepared(
349 sql: string,
350 params: ReadonlyArray<unknown>,
351 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
352 ) {
353 return transformRows
354 ? Effect.map(this.run("executeUnprepared", sql, params), transformRows)
355 : this.run("executeUnprepared", sql, params)
356 }
357 executeStream(
358 sql: string,
359 params: ReadonlyArray<unknown>,
360 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
361 ) {
362 const eff = transformRows
363 ? Effect.map(this.run("executeStream", sql, params), transformRows)
364 : this.run("executeStream", sql, params)
365 return Stream.fromArrayEffect(eff)
366 }
367}
368
369/**
370 * Creates a layer from an effect that acquires a `PgliteClient`, providing both `PgliteClient` and `SqlClient`.
371 *
372 * @stability unstable
373 * @category layers
374 * @since 4.0.0
375 */
376export const layerFrom = <E, R>(
377 acquire: Effect.Effect<PgliteClient, E, R>
378): Layer.Layer<PgliteClient | Client.SqlClient, E, Exclude<R, Scope.Scope | Reactivity.Reactivity>> =>
379 Layer.effectContext(
380 Effect.map(acquire, (client) =>
381 Context.make(PgliteClient, client).pipe(
382 Context.add(Client.SqlClient, client)
383 ))
384 ).pipe(Layer.provide(Reactivity.layer)) as any
385
386/**
387 * Creates a layer from a `Config`-wrapped PGlite client configuration, providing both `PgliteClient` and `SqlClient`.
388 *
389 * @stability unstable
390 * @category layers
391 * @since 4.0.0
392 */
393export const layerConfig: (
394 config: Config.Wrap<PgliteClientConfig.ConfigBase>
395) => Layer.Layer<PgliteClient | Client.SqlClient, Config.ConfigError | SqlError> = (
396 config: Config.Wrap<PgliteClientConfig.ConfigBase>
397): Layer.Layer<PgliteClient | Client.SqlClient, Config.ConfigError | SqlError> =>
398 layerFrom(Effect.flatMap(
399 Config.unwrap(config),
400 (resolved) => make(resolved as PgliteClientConfig)
401 ))
402
403/**
404 * Creates a layer from a concrete PGlite client configuration, providing both `PgliteClient` and `SqlClient`.
405 *
406 * @stability unstable
407 * @category layers
408 * @since 4.0.0
409 */
410export const layer = (
411 config?: PgliteClientConfig | undefined
412): Layer.Layer<PgliteClient | Client.SqlClient, SqlError> => layerFrom(make(config))
413
414/**
415 * Creates the PGlite statement compiler, using PostgreSQL `$1` placeholders, double-quoted identifiers, returning clauses, and optional JSON value transformation.
416 *
417 * @stability unstable
418 * @category constructors
419 * @since 4.0.0
420 */
421export const makeCompiler = (
422 transform?: (_: string) => string,
423 transformJson = true
424): Statement.Compiler => {
425 const transformValue = transformJson && transform
426 ? Statement.defaultTransforms(transform).value
427 : undefined
428
429 return Statement.makeCompiler<PgCustom>({
430 dialect: "pg",
431 placeholder(_) {
432 return `$${_}`
433 },
434 onIdentifier: transform ?
435 function(value, withoutTransform) {
436 return withoutTransform ? escape(value) : escape(transform(value))
437 } :
438 escape,
439 onRecordUpdate(placeholders, valueAlias, valueColumns, values, returning) {
440 return [
441 `(values ${placeholders}) AS ${valueAlias}${valueColumns}${returning ? ` RETURNING ${returning[0]}` : ""}`,
442 returning ?
443 values.flat().concat(returning[1]) :
444 values.flat()
445 ]
446 },
447 onCustom(type, placeholder, withoutTransform) {
448 switch (type.kind) {
449 case "PgJson": {
450 const value = withoutTransform || transformValue === undefined
451 ? type.paramA
452 : transformValue(type.paramA)
453 return [
454 placeholder(undefined),
455 [typeof value === "string" ? JSON.stringify(value) : value]
456 ]
457 }
458 }
459 }
460 })
461}
462
463const escape = Statement.defaultEscape("\"")
464const escapeLiteral = (value: string) => `'${value.replace(/'/g, "''")}'`
465
466/**
467 * PGlite-specific custom statement fragments supported by the compiler, currently JSON parameter fragments.
468 *
469 * @stability unstable
470 * @category models
471 * @since 4.0.0
472 */
473export type PgCustom = PgJson
474
475/**
476 * @category models
477 * @since 4.0.0
478 */
479interface PgJson extends Custom<"PgJson", unknown> {}
480const PgJson = Statement.custom<PgJson>("PgJson")
481
482const ATTR_DB_SYSTEM_NAME = "db.system.name"
483
484const pgCodeFromCause = (cause: unknown): string | undefined => {
485 if (typeof cause !== "object" || cause === null || !("code" in cause)) {
486 return undefined
487 }
488 const code = cause.code
489 return typeof code === "string" ? code : undefined
490}
491
492const pgConstraintFromCause = (cause: unknown): string => {
493 if (typeof cause !== "object" || cause === null || !("constraint" in cause)) {
494 return "unknown"
495 }
496 const constraint = cause.constraint
497 if (typeof constraint !== "string") {
498 return "unknown"
499 }
500 const normalized = constraint.trim()
501 return normalized.length === 0 ? "unknown" : normalized
502}
503
504const classifyError = (
505 cause: unknown,
506 message: string,
507 operation: string
508) => {
509 const props = { cause, message, operation }
510 const code = pgCodeFromCause(cause)
511 if (code !== undefined) {
512 if (code.startsWith("08")) {
513 return new ConnectionError(props)
514 }
515 if (code.startsWith("28")) {
516 return new AuthenticationError(props)
517 }
518 if (code === "42501") {
519 return new AuthorizationError(props)
520 }
521 if (code.startsWith("42")) {
522 return new SqlSyntaxError(props)
523 }
524 if (code === "23505") {
525 return new UniqueViolation({ ...props, constraint: pgConstraintFromCause(cause) })
526 }
527 if (code.startsWith("23")) {
528 return new ConstraintError(props)
529 }
530 if (code === "40P01") {
531 return new DeadlockError(props)
532 }
533 if (code === "40001") {
534 return new SerializationError(props)
535 }
536 if (code === "55P03") {
537 return new LockTimeoutError(props)
538 }
539 if (code === "57014") {
540 return new StatementTimeoutError(props)
541 }
542 }
543 return new UnknownError(props)
544}