Open History to see the last commit
1/**
2 * MySQL adapter for Effect SQL, backed by the `mysql2` driver.
3 *
4 * This module provides constructors and layers for a {@link MysqlClient} and
5 * the generic Effect SQL client service. `make` creates a managed mysql2 pool,
6 * checks the connection with `SELECT 1`, maps mysql2 failures to `SqlError`,
7 * supports transaction connections, and exposes streaming queries through
8 * mysql2 query streams. It also provides direct and config-backed layers plus a
9 * MySQL statement compiler.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Config from "effect/Config"
15import * as Context from "effect/Context"
16import * as Duration from "effect/Duration"
17import * as Effect from "effect/Effect"
18import * as Layer from "effect/Layer"
19import * as Reactivity from "effect/reactivity/Reactivity"
20import * as Redacted from "effect/Redacted"
21import type { Scope } from "effect/Scope"
22import * as Semaphore from "effect/Semaphore"
23import * as Client from "effect/sql/SqlClient"
24import type { Connection } from "effect/sql/SqlConnection"
25import {
26 AuthenticationError,
27 AuthorizationError,
28 ConnectionError,
29 ConstraintError,
30 DeadlockError,
31 LockTimeoutError,
32 SqlError,
33 SqlSyntaxError,
34 StatementTimeoutError,
35 UniqueViolation,
36 UnknownError
37} from "effect/sql/SqlError"
38import { asyncPauseResume } from "effect/sql/SqlStream"
39import * as Statement from "effect/sql/Statement"
40import * as Stream from "effect/Stream"
41import * as Mysql from "mysql2"
42
43const ATTR_DB_SYSTEM_NAME = "db.system.name"
44const ATTR_DB_NAMESPACE = "db.namespace"
45const ATTR_SERVER_ADDRESS = "server.address"
46const ATTR_SERVER_PORT = "server.port"
47
48const mysqlErrnoFromCause = (cause: unknown): number | undefined => {
49 if (typeof cause !== "object" || cause === null || !("errno" in cause)) {
50 return undefined
51 }
52 const errno = cause.errno
53 return typeof errno === "number" ? errno : undefined
54}
55
56const mysqlConnectionErrorCodes = new Set([1040, 1042, 1043, 1129, 1130, 1203])
57const mysqlAuthorizationErrorCodes = new Set([1044, 1142, 1143, 1227])
58const mysqlSyntaxErrorCodes = new Set([1054, 1064, 1146])
59const mysqlConstraintErrorCodes = new Set([1022, 1048, 1169, 1216, 1217, 1451, 1452, 1557])
60
61const UNKNOWN_CONSTRAINT = "unknown"
62
63const normalizeConstraintIdentifier = (identifier: unknown): string => {
64 if (typeof identifier !== "string") {
65 return UNKNOWN_CONSTRAINT
66 }
67 const trimmed = identifier.trim()
68 return trimmed.length === 0 ? UNKNOWN_CONSTRAINT : trimmed
69}
70
71const mysqlCauseProperty = (cause: unknown, property: "constraint" | "message" | "sqlMessage"): unknown => {
72 if (typeof cause !== "object" || cause === null || !(property in cause)) {
73 return undefined
74 }
75 return (cause as Record<string, unknown>)[property]
76}
77
78const mysqlDuplicateEntryConstraintFromMessage = (message: unknown): string => {
79 if (typeof message !== "string") {
80 return UNKNOWN_CONSTRAINT
81 }
82 const match = /\bfor key\s+(?:'([^']*)'|\x60([^\x60]*)\x60|([^\s'\x60]+))/i.exec(message)
83 return match === null ?
84 UNKNOWN_CONSTRAINT :
85 normalizeConstraintIdentifier(match[1] ?? match[2] ?? match[3])
86}
87
88const mysqlDuplicateEntryConstraintFromCause = (cause: unknown): string => {
89 const constraint = normalizeConstraintIdentifier(mysqlCauseProperty(cause, "constraint"))
90 if (constraint !== UNKNOWN_CONSTRAINT) {
91 return constraint
92 }
93 const sqlMessageConstraint = mysqlDuplicateEntryConstraintFromMessage(mysqlCauseProperty(cause, "sqlMessage"))
94 if (sqlMessageConstraint !== UNKNOWN_CONSTRAINT) {
95 return sqlMessageConstraint
96 }
97 return mysqlDuplicateEntryConstraintFromMessage(mysqlCauseProperty(cause, "message"))
98}
99
100const classifyError = (
101 cause: unknown,
102 message: string,
103 operation: string
104) => {
105 const props = { cause, message, operation }
106 const errno = mysqlErrnoFromCause(cause)
107 if (errno !== undefined) {
108 if (mysqlConnectionErrorCodes.has(errno)) {
109 return new ConnectionError(props)
110 }
111 if (errno === 1045) {
112 return new AuthenticationError(props)
113 }
114 if (mysqlAuthorizationErrorCodes.has(errno)) {
115 return new AuthorizationError(props)
116 }
117 if (mysqlSyntaxErrorCodes.has(errno)) {
118 return new SqlSyntaxError(props)
119 }
120 if (errno === 1062) {
121 return new UniqueViolation({ ...props, constraint: mysqlDuplicateEntryConstraintFromCause(cause) })
122 }
123 if (mysqlConstraintErrorCodes.has(errno)) {
124 return new ConstraintError(props)
125 }
126 if (errno === 1213) {
127 return new DeadlockError(props)
128 }
129 if (errno === 1205) {
130 return new LockTimeoutError(props)
131 }
132 if (errno === 3024) {
133 return new StatementTimeoutError(props)
134 }
135 }
136 return new UnknownError(props)
137}
138
139/**
140 * Runtime type identifier used to mark `MysqlClient` values.
141 *
142 * @stability unstable
143 * @category type IDs
144 * @since 4.0.0
145 */
146export const TypeId: TypeId = "~@effect/sql-mysql2/MysqlClient"
147
148/**
149 * Type-level identifier used to mark `MysqlClient` values.
150 *
151 * @stability unstable
152 * @category type IDs
153 * @since 4.0.0
154 */
155export type TypeId = "~@effect/sql-mysql2/MysqlClient"
156
157/**
158 * mysql2-backed SQL client service, extending `SqlClient` with its runtime type marker and client configuration.
159 *
160 * @stability unstable
161 * @category services
162 * @since 4.0.0
163 */
164export interface MysqlClient extends Client.SqlClient {
165 readonly [TypeId]: TypeId
166 readonly config: MysqlClientConfig
167}
168
169/**
170 * Service tag for the mysql2 SQL client service.
171 *
172 * **When to use**
173 *
174 * Use to access or provide a mysql2 client through the Effect context.
175 *
176 * @stability unstable
177 * @category services
178 * @since 4.0.0
179 */
180export const MysqlClient = Context.Service<MysqlClient>("@effect/sql-mysql2/MysqlClient")
181
182/**
183 * Configuration for a mysql2 client, including connection URI or connection fields, pool options, span attributes, and query/result name transforms.
184 *
185 * @stability unstable
186 * @category models
187 * @since 4.0.0
188 */
189export interface MysqlClientConfig {
190 /**
191 * Connection URI. Setting this will override the other connection options
192 */
193 readonly url?: Redacted.Redacted | undefined
194
195 readonly host?: string | undefined
196 readonly port?: number | undefined
197 readonly database?: string | undefined
198 readonly username?: string | undefined
199 readonly password?: Redacted.Redacted | undefined
200
201 readonly maxConnections?: number | undefined
202 readonly connectionTTL?: Duration.Input | undefined
203
204 /**
205 * Options passed directly to the `mysql2` connection pool.
206 *
207 * @stability unstable
208 */
209 readonly poolConfig?: Mysql.PoolOptions | undefined
210
211 /**
212 * Use the text protocol instead of prepared statements, for proxies like
213 * Cloudflare Hyperdrive that do not support `COM_STMT_PREPARE`.
214 */
215 readonly disablePreparedStatements?: boolean | undefined
216
217 readonly spanAttributes?: Record<string, unknown> | undefined
218
219 readonly transformResultNames?: ((str: string) => string) | undefined
220 readonly transformQueryNames?: ((str: string) => string) | undefined
221}
222
223/**
224 * Creates a scoped MySQL client backed by a managed mysql2 pool, verifying connectivity and supporting streaming queries through mysql2 query streams.
225 *
226 * @stability unstable
227 * @category constructors
228 * @since 4.0.0
229 */
230export const make = (
231 options: MysqlClientConfig
232): Effect.Effect<MysqlClient, SqlError, Scope | Reactivity.Reactivity> =>
233 Effect.gen(function*() {
234 const compiler = makeCompiler(options.transformQueryNames)
235 const transformRows = options.transformResultNames ?
236 Statement.defaultTransforms(
237 options.transformResultNames
238 ).array :
239 undefined
240 const defaultMethod: "execute" | "query" = options.disablePreparedStatements === true ? "query" : "execute"
241
242 class ConnectionImpl implements Connection {
243 readonly conn: Mysql.PoolConnection
244 // A transaction or reserved connection can be shared by several fibers.
245 // Each statement or stream holds this permit from before its command is
246 // queued until it has settled, so `KILL QUERY` only ever reaches the
247 // statement that was interrupted.
248 readonly permit = Semaphore.makeUnsafe(1)
249 constructor(conn: Mysql.PoolConnection) {
250 this.conn = conn
251 }
252
253 private runRaw(
254 sql: string,
255 values?: ReadonlyArray<any>,
256 rowsAsArray = false,
257 method: "execute" | "query" = defaultMethod
258 ) {
259 return this.permit.withPermit(Effect.callback<unknown, SqlError>((resume) => {
260 const operation = method === "query" ? "executeUnprepared" : "execute"
261 const statement = new ActiveStatement(connectionOptions, this.conn)
262 ;(this.conn as any)[method]({
263 sql,
264 values,
265 rowsAsArray
266 }, (cause: unknown | null, results: unknown, _fields: any) => {
267 // Once interrupted, the result is dropped.
268 if (!statement.settle()) return
269 if (cause) {
270 resume(
271 Effect.fail(new SqlError({ reason: classifyError(cause, "Failed to execute statement", operation) }))
272 )
273 } else {
274 resume(Effect.succeed(results))
275 }
276 })
277 return statement.cancel
278 }))
279 }
280
281 private run(
282 sql: string,
283 values?: ReadonlyArray<any>,
284 rowsAsArray = false,
285 method: "execute" | "query" = defaultMethod
286 ) {
287 return this.runRaw(sql, values, rowsAsArray, method).pipe(
288 Effect.map((results) => Array.isArray(results) ? results : [])
289 )
290 }
291
292 execute(
293 sql: string,
294 params: ReadonlyArray<unknown>,
295 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
296 ) {
297 return transformRows
298 ? Effect.map(this.run(sql, params), transformRows)
299 : this.run(sql, params)
300 }
301 executeRaw(sql: string, params: ReadonlyArray<unknown>) {
302 return this.runRaw(sql, params)
303 }
304 executeValues(sql: string, params: ReadonlyArray<unknown>) {
305 return this.run(sql, params, true)
306 }
307 executeValuesUnprepared(sql: string, params: ReadonlyArray<unknown>) {
308 return this.run(sql, params, true, "query")
309 }
310 executeUnprepared(
311 sql: string,
312 params: ReadonlyArray<unknown>,
313 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
314 ) {
315 return transformRows
316 ? Effect.map(this.run(sql, params, false, "query"), transformRows)
317 : this.run(sql, params, false, "query")
318 }
319 executeStream(
320 sql: string,
321 params: ReadonlyArray<unknown>,
322 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
323 ) {
324 const stream = queryStream(connectionOptions, this.conn, this.permit, sql, params)
325 return transformRows
326 ? Stream.mapArray(stream, (_) => transformRows(_) as any)
327 : stream
328 }
329 }
330
331 // Shared by the pool and the one-off connections that send `KILL QUERY`.
332 const connectionOptions: Mysql.PoolOptions = options.url
333 ? {
334 uri: Redacted.value(options.url),
335 multipleStatements: true,
336 supportBigNumbers: true,
337 connectionLimit: options.maxConnections!,
338 idleTimeout: options.connectionTTL
339 ? Duration.toMillis(Duration.fromInputUnsafe(options.connectionTTL))
340 : undefined as any
341 }
342 : {
343 ...options.poolConfig,
344 host: options.host,
345 port: options.port,
346 database: options.database,
347 user: options.username,
348 password: options.password
349 ? Redacted.value(options.password)
350 : undefined,
351 multipleStatements: true,
352 supportBigNumbers: true,
353 connectionLimit: options.maxConnections ?? options.poolConfig?.connectionLimit,
354 idleTimeout: options.connectionTTL
355 ? Duration.toMillis(Duration.fromInputUnsafe(options.connectionTTL))
356 : options.poolConfig?.idleTimeout
357 } as Mysql.PoolOptions
358 const pool = Mysql.createPool(connectionOptions)
359
360 yield* Effect.acquireRelease(
361 Effect.callback<void, SqlError>((resume) => {
362 ;(pool as any).query("SELECT 1", (cause: Error) => {
363 if (cause) {
364 resume(Effect.fail(
365 new SqlError({
366 reason: classifyError(cause, "MysqlClient: Failed to connect", "connect")
367 })
368 ))
369 } else {
370 resume(Effect.void)
371 }
372 })
373 }),
374 () =>
375 Effect.callback<void>((resume) => {
376 pool.end(() => resume(Effect.void))
377 })
378 ).pipe(
379 Effect.timeoutOrElse({
380 duration: Duration.seconds(5),
381 orElse: () =>
382 Effect.fail(
383 new SqlError({
384 reason: new ConnectionError({
385 message: "MysqlClient: Connection timeout",
386 cause: new Error("connection timeout"),
387 operation: "connect"
388 })
389 })
390 )
391 })
392 )
393
394 const acquireConn = Effect.acquireRelease(
395 Effect.callback<Mysql.PoolConnection, SqlError>((resume) => {
396 let interrupted = false
397 // mysql2 cannot remove a waiter from its queue, so a connection handed
398 // over after interruption goes straight back to the pool.
399 pool.getConnection((cause, conn) => {
400 if (interrupted) {
401 conn?.release()
402 return
403 }
404 if (cause) {
405 resume(
406 Effect.fail(
407 new SqlError({
408 reason: classifyError(cause, "Failed to acquire connection", "acquireConnection")
409 })
410 )
411 )
412 } else {
413 resume(Effect.succeed(conn))
414 }
415 })
416 return Effect.sync(() => {
417 interrupted = true
418 })
419 }),
420 (conn) => Effect.sync(() => conn.release()),
421 { interruptible: true }
422 )
423
424 // Every statement checks out its own connection, so an interrupted
425 // statement can be killed by thread id before the connection is reused.
426 const acquirer = Effect.map(acquireConn, (conn) => new ConnectionImpl(conn))
427
428 const spanAttributes: Array<[string, unknown]> = [
429 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
430 [ATTR_DB_SYSTEM_NAME, "mysql"],
431 [ATTR_SERVER_ADDRESS, options.host ?? "localhost"],
432 [ATTR_SERVER_PORT, options.port ?? 3306]
433 ]
434
435 if (options.database) {
436 spanAttributes.push([ATTR_DB_NAMESPACE, options.database])
437 }
438
439 return Object.assign(
440 yield* Client.make({
441 driver: "mysql2",
442 acquirer,
443 releaseSavepoint: (name) => `RELEASE SAVEPOINT ${name}`,
444 compiler,
445 spanAttributes,
446 transformRows
447 }),
448 { [TypeId]: TypeId as TypeId, config: options }
449 )
450 })
451
452/**
453 * Creates a layer from a `Config`-wrapped MySQL client configuration, providing both `MysqlClient` and `SqlClient`.
454 *
455 * @stability unstable
456 * @category layers
457 * @since 4.0.0
458 */
459export const layerConfig = (
460 config: Config.Wrap<MysqlClientConfig>
461): Layer.Layer<MysqlClient | Client.SqlClient, Config.ConfigError | SqlError> =>
462 Layer.effectContext(
463 Config.unwrap(config).pipe(
464 Effect.flatMap(make),
465 Effect.map((client) =>
466 Context.make(MysqlClient, client).pipe(
467 Context.add(Client.SqlClient, client)
468 )
469 )
470 )
471 ).pipe(Layer.provide(Reactivity.layer))
472
473/**
474 * Creates a layer from a concrete MySQL client configuration, providing both `MysqlClient` and `SqlClient`.
475 *
476 * @stability unstable
477 * @category layers
478 * @since 4.0.0
479 */
480export const layer = (
481 config: MysqlClientConfig
482): Layer.Layer<MysqlClient | Client.SqlClient, Config.ConfigError | SqlError> =>
483 Layer.effectContext(
484 Effect.map(make(config), (client) =>
485 Context.make(MysqlClient, client).pipe(
486 Context.add(Client.SqlClient, client)
487 ))
488 ).pipe(Layer.provide(Reactivity.layer))
489
490/**
491 * Creates the MySQL statement compiler, using `?` placeholders and backtick-escaped identifiers.
492 *
493 * @stability unstable
494 * @category constructors
495 * @since 4.0.0
496 */
497export const makeCompiler = (transform?: (_: string) => string) =>
498 Statement.makeCompiler({
499 dialect: "mysql",
500 placeholder(_) {
501 return `?`
502 },
503 onIdentifier: transform ?
504 function(value, withoutTransform) {
505 return withoutTransform ? escape(value) : escape(transform(value))
506 } :
507 escape,
508 onCustom() {
509 return ["", []]
510 },
511 onRecordUpdate() {
512 return ["", []]
513 }
514 })
515
516const escape = Statement.defaultEscape("`")
517
518/**
519 * How long interrupting a statement waits for `KILL QUERY` and the statement to
520 * settle before destroying the connection instead.
521 */
522const cancelTimeoutMillis = 5000
523
524/**
525 * A statement running on a pool connection: records when the server is done
526 * with it, and can stop it.
527 */
528class ActiveStatement {
529 settled = false
530 private cancelling = false
531 private onSettled: (() => void) | undefined = undefined
532 private readonly connectionOptions: Mysql.ConnectionOptions
533 private readonly conn: Mysql.PoolConnection
534 constructor(connectionOptions: Mysql.ConnectionOptions, conn: Mysql.PoolConnection) {
535 this.connectionOptions = connectionOptions
536 this.conn = conn
537 }
538
539 /**
540 * Records that the server is done with the statement. Returns `true` when
541 * the caller handles the outcome, and `false` when it was already recorded
542 * or `cancel` is waiting for it.
543 */
544 settle(): boolean {
545 if (this.settled) return false
546 this.settled = true
547 if (!this.cancelling) return true
548 this.onSettled?.()
549 return false
550 }
551
552 /**
553 * Stops the statement and waits until it is safe to release the connection.
554 *
555 * Running this marks the statement as being cancelled before anything else,
556 * so a settlement from then on belongs to the cancellation, even one that
557 * happens before the kill is registered. If the statement has already
558 * settled, nothing is killed.
559 *
560 * `KILL QUERY` goes over a one-off connection, because the pool may have no
561 * free slot. The effect completes once the kill attempt has finished and the
562 * statement has settled, so a late kill can never reach another statement on
563 * the connection. Kill failures are ignored, including a kill connection that
564 * cannot be created. If both do not happen within `cancelTimeoutMillis`, the
565 * connection is destroyed, which removes it from the pool.
566 */
567 readonly cancel: Effect.Effect<void> = Effect.suspend(() => {
568 this.cancelling = true
569 return Effect.callback<void>((resume) => {
570 if (this.settled) return resume(Effect.void)
571 let done = false
572 let killed = false
573 let kill: Mysql.Connection | undefined
574 const finish = () => {
575 if (done) return
576 done = true
577 clearTimeout(timer)
578 kill?.destroy()
579 resume(Effect.void)
580 }
581 const onKilled = () => {
582 killed = true
583 if (this.settled) finish()
584 }
585 const timer = setTimeout(() => {
586 this.conn.destroy()
587 finish()
588 }, cancelTimeoutMillis)
589 this.onSettled = () => {
590 if (killed) finish()
591 }
592 try {
593 // A configured `stream` factory runs synchronously here and may throw.
594 kill = Mysql.createConnection({ ...this.connectionOptions })
595 kill.on("error", () => {})
596 ;(kill as any).query("KILL QUERY ?", [this.conn.threadId], onKilled)
597 } catch {
598 onKilled()
599 }
600 })
601 })
602}
603
604function queryStream(
605 connectionOptions: Mysql.ConnectionOptions,
606 conn: Mysql.PoolConnection,
607 permit: Semaphore.Semaphore,
608 sql: string,
609 params?: ReadonlyArray<any>
610) {
611 return asyncPauseResume<any, SqlError>(Effect.fnUntraced(function*(emit) {
612 const statement = new ActiveStatement(connectionOptions, conn)
613 let query: any
614 // The permit is returned as soon as the statement settles, because the
615 // consumer may run its own statements on this connection before the
616 // stream ends. When the stream is cancelled, it is kept until the statement
617 // has been killed and seen to settle.
618 let owned = true
619 const release = Effect.suspend(() => {
620 if (!owned) return Effect.void
621 owned = false
622 return permit.release(1)
623 })
624 // Taking the permit also installs the cleanup, so once the command starts
625 // below there is no point at which interruption could leave it running.
626 yield* Effect.acquireRelease(
627 permit.take(1),
628 () =>
629 Effect.ensuring(
630 Effect.suspend(() => {
631 if (query === undefined) return Effect.void
632 // Destroying the stream resumes a paused connection, so the
633 // remaining rows are read and discarded while the statement is
634 // killed. Rows that were already buffered are replayed
635 // synchronously and can settle the statement before `destroy`
636 // returns, in which case nothing is killed.
637 query.destroy()
638 return statement.cancel
639 }),
640 release
641 ),
642 { interruptible: true }
643 )
644 const command = (conn as any).query(sql, params)
645 query = command.stream()
646 // The command, unlike its stream, reports when the server is done with the
647 // statement, even after the stream has been destroyed.
648 const settle = () => {
649 if (statement.settle()) Effect.runSync(release)
650 }
651 command.once("end", settle)
652 command.once("error", settle)
653
654 let buffer: Array<any> = []
655 let taskPending = false
656 query.on(
657 "error",
658 (cause: unknown) =>
659 emit.fail(new SqlError({ reason: classifyError(cause, "Failed to stream statement", "stream") }))
660 )
661 query.on("data", (row: any) => {
662 buffer.push(row)
663 if (!taskPending) {
664 taskPending = true
665 queueMicrotask(() => {
666 const items = buffer
667 buffer = []
668 emit.array(items)
669 taskPending = false
670 })
671 }
672 })
673 query.on("end", () => {
674 if (buffer.length > 0) {
675 emit.array(buffer)
676 buffer = []
677 }
678 emit.end()
679 })
680
681 return {
682 onPause() {
683 query.pause()
684 },
685 onResume() {
686 query.resume()
687 }
688 }
689 }))
690}