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 * @since 4.0.0
12 */
13import * as Config from "effect/Config"
14import * as Context from "effect/Context"
15import * as Duration from "effect/Duration"
16import * as Effect from "effect/Effect"
17import * as Layer from "effect/Layer"
18import * as Reactivity from "effect/reactivity/Reactivity"
19import * as Redacted from "effect/Redacted"
20import type { Scope } from "effect/Scope"
21import * as Client from "effect/sql/SqlClient"
22import type { Connection } from "effect/sql/SqlConnection"
23import {
24 AuthenticationError,
25 AuthorizationError,
26 ConnectionError,
27 ConstraintError,
28 DeadlockError,
29 LockTimeoutError,
30 SqlError,
31 SqlSyntaxError,
32 StatementTimeoutError,
33 UniqueViolation,
34 UnknownError
35} from "effect/sql/SqlError"
36import { asyncPauseResume } from "effect/sql/SqlStream"
37import * as Statement from "effect/sql/Statement"
38import * as Stream from "effect/Stream"
39import * as Mysql from "mysql2"
40
41const ATTR_DB_SYSTEM_NAME = "db.system.name"
42const ATTR_DB_NAMESPACE = "db.namespace"
43const ATTR_SERVER_ADDRESS = "server.address"
44const ATTR_SERVER_PORT = "server.port"
45
46const mysqlErrnoFromCause = (cause: unknown): number | undefined => {
47 if (typeof cause !== "object" || cause === null || !("errno" in cause)) {
48 return undefined
49 }
50 const errno = cause.errno
51 return typeof errno === "number" ? errno : undefined
52}
53
54const mysqlConnectionErrorCodes = new Set([1040, 1042, 1043, 1129, 1130, 1203])
55const mysqlAuthorizationErrorCodes = new Set([1044, 1142, 1143, 1227])
56const mysqlSyntaxErrorCodes = new Set([1054, 1064, 1146])
57const mysqlConstraintErrorCodes = new Set([1022, 1048, 1169, 1216, 1217, 1451, 1452, 1557])
58
59const UNKNOWN_CONSTRAINT = "unknown"
60
61const normalizeConstraintIdentifier = (identifier: unknown): string => {
62 if (typeof identifier !== "string") {
63 return UNKNOWN_CONSTRAINT
64 }
65 const trimmed = identifier.trim()
66 return trimmed.length === 0 ? UNKNOWN_CONSTRAINT : trimmed
67}
68
69const mysqlCauseProperty = (cause: unknown, property: "constraint" | "message" | "sqlMessage"): unknown => {
70 if (typeof cause !== "object" || cause === null || !(property in cause)) {
71 return undefined
72 }
73 return (cause as Record<string, unknown>)[property]
74}
75
76const mysqlDuplicateEntryConstraintFromMessage = (message: unknown): string => {
77 if (typeof message !== "string") {
78 return UNKNOWN_CONSTRAINT
79 }
80 const match = /\bfor key\s+(?:'([^']*)'|\x60([^\x60]*)\x60|([^\s'\x60]+))/i.exec(message)
81 return match === null ?
82 UNKNOWN_CONSTRAINT :
83 normalizeConstraintIdentifier(match[1] ?? match[2] ?? match[3])
84}
85
86const mysqlDuplicateEntryConstraintFromCause = (cause: unknown): string => {
87 const constraint = normalizeConstraintIdentifier(mysqlCauseProperty(cause, "constraint"))
88 if (constraint !== UNKNOWN_CONSTRAINT) {
89 return constraint
90 }
91 const sqlMessageConstraint = mysqlDuplicateEntryConstraintFromMessage(mysqlCauseProperty(cause, "sqlMessage"))
92 if (sqlMessageConstraint !== UNKNOWN_CONSTRAINT) {
93 return sqlMessageConstraint
94 }
95 return mysqlDuplicateEntryConstraintFromMessage(mysqlCauseProperty(cause, "message"))
96}
97
98const classifyError = (
99 cause: unknown,
100 message: string,
101 operation: string
102) => {
103 const props = { cause, message, operation }
104 const errno = mysqlErrnoFromCause(cause)
105 if (errno !== undefined) {
106 if (mysqlConnectionErrorCodes.has(errno)) {
107 return new ConnectionError(props)
108 }
109 if (errno === 1045) {
110 return new AuthenticationError(props)
111 }
112 if (mysqlAuthorizationErrorCodes.has(errno)) {
113 return new AuthorizationError(props)
114 }
115 if (mysqlSyntaxErrorCodes.has(errno)) {
116 return new SqlSyntaxError(props)
117 }
118 if (errno === 1062) {
119 return new UniqueViolation({ ...props, constraint: mysqlDuplicateEntryConstraintFromCause(cause) })
120 }
121 if (mysqlConstraintErrorCodes.has(errno)) {
122 return new ConstraintError(props)
123 }
124 if (errno === 1213) {
125 return new DeadlockError(props)
126 }
127 if (errno === 1205) {
128 return new LockTimeoutError(props)
129 }
130 if (errno === 3024) {
131 return new StatementTimeoutError(props)
132 }
133 }
134 return new UnknownError(props)
135}
136
137/**
138 * Runtime type identifier used to mark `MysqlClient` values.
139 *
140 * @category type IDs
141 * @since 4.0.0
142 */
143export const TypeId: TypeId = "~@effect/sql-mysql2/MysqlClient"
144
145/**
146 * Type-level identifier used to mark `MysqlClient` values.
147 *
148 * @category type IDs
149 * @since 4.0.0
150 */
151export type TypeId = "~@effect/sql-mysql2/MysqlClient"
152
153/**
154 * mysql2-backed SQL client service, extending `SqlClient` with its runtime type marker and client configuration.
155 *
156 * @category services
157 * @since 4.0.0
158 */
159export interface MysqlClient extends Client.SqlClient {
160 readonly [TypeId]: TypeId
161 readonly config: MysqlClientConfig
162}
163
164/**
165 * Service tag for the mysql2 SQL client service.
166 *
167 * **When to use**
168 *
169 * Use to access or provide a mysql2 client through the Effect context.
170 *
171 * @category services
172 * @since 4.0.0
173 */
174export const MysqlClient = Context.Service<MysqlClient>("@effect/sql-mysql2/MysqlClient")
175
176/**
177 * Configuration for a mysql2 client, including connection URI or connection fields, pool options, span attributes, and query/result name transforms.
178 *
179 * @category models
180 * @since 4.0.0
181 */
182export interface MysqlClientConfig {
183 /**
184 * Connection URI. Setting this will override the other connection options
185 */
186 readonly url?: Redacted.Redacted | undefined
187
188 readonly host?: string | undefined
189 readonly port?: number | undefined
190 readonly database?: string | undefined
191 readonly username?: string | undefined
192 readonly password?: Redacted.Redacted | undefined
193
194 readonly maxConnections?: number | undefined
195 readonly connectionTTL?: Duration.Input | undefined
196
197 /**
198 * Options passed directly to the `mysql2` connection pool.
199 *
200 * @stability unstable
201 */
202 readonly poolConfig?: Mysql.PoolOptions | undefined
203
204 /**
205 * Use the text protocol instead of prepared statements, for proxies like
206 * Cloudflare Hyperdrive that do not support `COM_STMT_PREPARE`.
207 */
208 readonly disablePreparedStatements?: boolean | undefined
209
210 readonly spanAttributes?: Record<string, unknown> | undefined
211
212 readonly transformResultNames?: ((str: string) => string) | undefined
213 readonly transformQueryNames?: ((str: string) => string) | undefined
214}
215
216/**
217 * Creates a scoped MySQL client backed by a managed mysql2 pool, verifying connectivity and supporting streaming queries through mysql2 query streams.
218 *
219 * @category constructors
220 * @since 4.0.0
221 */
222export const make = (
223 options: MysqlClientConfig
224): Effect.Effect<MysqlClient, SqlError, Scope | Reactivity.Reactivity> =>
225 Effect.gen(function*() {
226 const compiler = makeCompiler(options.transformQueryNames)
227 const transformRows = options.transformResultNames ?
228 Statement.defaultTransforms(
229 options.transformResultNames
230 ).array :
231 undefined
232 const defaultMethod: "execute" | "query" = options.disablePreparedStatements === true ? "query" : "execute"
233
234 class ConnectionImpl implements Connection {
235 readonly conn: Mysql.PoolConnection | Mysql.Pool
236 constructor(conn: Mysql.PoolConnection | Mysql.Pool) {
237 this.conn = conn
238 }
239
240 private runRaw(
241 sql: string,
242 values?: ReadonlyArray<any>,
243 rowsAsArray = false,
244 method: "execute" | "query" = defaultMethod
245 ) {
246 return Effect.callback<unknown, SqlError>((resume) => {
247 const operation = method === "query" ? "executeUnprepared" : "execute"
248 ;(this.conn as any)[method]({
249 sql,
250 values,
251 rowsAsArray
252 }, (cause: unknown | null, results: unknown, _fields: any) => {
253 if (cause) {
254 resume(
255 Effect.fail(new SqlError({ reason: classifyError(cause, "Failed to execute statement", operation) }))
256 )
257 } else {
258 resume(Effect.succeed(results))
259 }
260 })
261 })
262 }
263
264 private run(
265 sql: string,
266 values?: ReadonlyArray<any>,
267 rowsAsArray = false,
268 method: "execute" | "query" = defaultMethod
269 ) {
270 return this.runRaw(sql, values, rowsAsArray, method).pipe(
271 Effect.map((results) => Array.isArray(results) ? results : [])
272 )
273 }
274
275 execute(
276 sql: string,
277 params: ReadonlyArray<unknown>,
278 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
279 ) {
280 return transformRows
281 ? Effect.map(this.run(sql, params), transformRows)
282 : this.run(sql, params)
283 }
284 executeRaw(sql: string, params: ReadonlyArray<unknown>) {
285 return this.runRaw(sql, params)
286 }
287 executeValues(sql: string, params: ReadonlyArray<unknown>) {
288 return this.run(sql, params, true)
289 }
290 executeValuesUnprepared(sql: string, params: ReadonlyArray<unknown>) {
291 return this.run(sql, params, true, "query")
292 }
293 executeUnprepared(
294 sql: string,
295 params: ReadonlyArray<unknown>,
296 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
297 ) {
298 return transformRows
299 ? Effect.map(this.run(sql, params, false, "query"), transformRows)
300 : this.run(sql, params, false, "query")
301 }
302 executeStream(
303 sql: string,
304 params: ReadonlyArray<unknown>,
305 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
306 ) {
307 const stream = queryStream(this.conn as any, sql, params)
308 return transformRows
309 ? Stream.mapArray(stream, (_) => transformRows(_) as any)
310 : stream
311 }
312 }
313
314 const pool = options.url
315 ? Mysql.createPool({
316 uri: Redacted.value(options.url),
317 multipleStatements: true,
318 supportBigNumbers: true,
319 connectionLimit: options.maxConnections!,
320 idleTimeout: options.connectionTTL
321 ? Duration.toMillis(Duration.fromInputUnsafe(options.connectionTTL))
322 : undefined as any
323 })
324 : Mysql.createPool({
325 ...options.poolConfig,
326 host: options.host,
327 port: options.port,
328 database: options.database,
329 user: options.username,
330 password: options.password
331 ? Redacted.value(options.password)
332 : undefined,
333 multipleStatements: true,
334 supportBigNumbers: true,
335 connectionLimit: options.maxConnections ?? options.poolConfig?.connectionLimit,
336 idleTimeout: options.connectionTTL
337 ? Duration.toMillis(Duration.fromInputUnsafe(options.connectionTTL))
338 : options.poolConfig?.idleTimeout
339 } as Mysql.PoolOptions)
340
341 yield* Effect.acquireRelease(
342 Effect.callback<void, SqlError>((resume) => {
343 ;(pool as any).query("SELECT 1", (cause: Error) => {
344 if (cause) {
345 resume(Effect.fail(
346 new SqlError({
347 reason: classifyError(cause, "MysqlClient: Failed to connect", "connect")
348 })
349 ))
350 } else {
351 resume(Effect.void)
352 }
353 })
354 }),
355 () =>
356 Effect.callback<void>((resume) => {
357 pool.end(() => resume(Effect.void))
358 })
359 ).pipe(
360 Effect.timeoutOrElse({
361 duration: Duration.seconds(5),
362 orElse: () =>
363 Effect.fail(
364 new SqlError({
365 reason: new ConnectionError({
366 message: "MysqlClient: Connection timeout",
367 cause: new Error("connection timeout"),
368 operation: "connect"
369 })
370 })
371 )
372 })
373 )
374
375 const poolConnection = new ConnectionImpl(pool)
376
377 const acquireConn = Effect.acquireRelease(
378 Effect.callback<Mysql.PoolConnection, SqlError>((resume) => {
379 pool.getConnection((cause, conn) => {
380 if (cause) {
381 resume(
382 Effect.fail(
383 new SqlError({
384 reason: classifyError(cause, "Failed to acquire connection", "acquireConnection")
385 })
386 )
387 )
388 } else {
389 resume(Effect.succeed(conn))
390 }
391 })
392 }),
393 (conn) => Effect.sync(() => conn.release())
394 )
395
396 const transactionAcquirer = Effect.map(
397 acquireConn,
398 (conn) => new ConnectionImpl(conn)
399 )
400
401 const spanAttributes: Array<[string, unknown]> = [
402 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
403 [ATTR_DB_SYSTEM_NAME, "mysql"],
404 [ATTR_SERVER_ADDRESS, options.host ?? "localhost"],
405 [ATTR_SERVER_PORT, options.port ?? 3306]
406 ]
407
408 if (options.database) {
409 spanAttributes.push([ATTR_DB_NAMESPACE, options.database])
410 }
411
412 return Object.assign(
413 yield* Client.make({
414 acquirer: Effect.succeed(poolConnection),
415 transactionAcquirer,
416 releaseSavepoint: (name) => `RELEASE SAVEPOINT ${name}`,
417 compiler,
418 spanAttributes,
419 transformRows
420 }),
421 { [TypeId]: TypeId as TypeId, config: options }
422 )
423 })
424
425/**
426 * Creates a layer from a `Config`-wrapped MySQL client configuration, providing both `MysqlClient` and `SqlClient`.
427 *
428 * @category layers
429 * @since 4.0.0
430 */
431export const layerConfig = (
432 config: Config.Wrap<MysqlClientConfig>
433): Layer.Layer<MysqlClient | Client.SqlClient, Config.ConfigError | SqlError> =>
434 Layer.effectContext(
435 Config.unwrap(config).pipe(
436 Effect.flatMap(make),
437 Effect.map((client) =>
438 Context.make(MysqlClient, client).pipe(
439 Context.add(Client.SqlClient, client)
440 )
441 )
442 )
443 ).pipe(Layer.provide(Reactivity.layer))
444
445/**
446 * Creates a layer from a concrete MySQL client configuration, providing both `MysqlClient` and `SqlClient`.
447 *
448 * @category layers
449 * @since 4.0.0
450 */
451export const layer = (
452 config: MysqlClientConfig
453): Layer.Layer<MysqlClient | Client.SqlClient, Config.ConfigError | SqlError> =>
454 Layer.effectContext(
455 Effect.map(make(config), (client) =>
456 Context.make(MysqlClient, client).pipe(
457 Context.add(Client.SqlClient, client)
458 ))
459 ).pipe(Layer.provide(Reactivity.layer))
460
461/**
462 * Creates the MySQL statement compiler, using `?` placeholders and backtick-escaped identifiers.
463 *
464 * @category constructors
465 * @since 4.0.0
466 */
467export const makeCompiler = (transform?: (_: string) => string) =>
468 Statement.makeCompiler({
469 dialect: "mysql",
470 placeholder(_) {
471 return `?`
472 },
473 onIdentifier: transform ?
474 function(value, withoutTransform) {
475 return withoutTransform ? escape(value) : escape(transform(value))
476 } :
477 escape,
478 onCustom() {
479 return ["", []]
480 },
481 onRecordUpdate() {
482 return ["", []]
483 }
484 })
485
486const escape = Statement.defaultEscape("`")
487
488function queryStream(
489 conn: Mysql.PoolConnection,
490 sql: string,
491 params?: ReadonlyArray<any>
492) {
493 return asyncPauseResume<any, SqlError>(Effect.fnUntraced(function*(emit) {
494 const query = (conn as any).query(sql, params).stream()
495 yield* Effect.addFinalizer(() => Effect.sync(() => query.destroy() as void))
496
497 let buffer: Array<any> = []
498 let taskPending = false
499 query.on(
500 "error",
501 (cause: unknown) =>
502 emit.fail(new SqlError({ reason: classifyError(cause, "Failed to stream statement", "stream") }))
503 )
504 query.on("data", (row: any) => {
505 buffer.push(row)
506 if (!taskPending) {
507 taskPending = true
508 queueMicrotask(() => {
509 const items = buffer
510 buffer = []
511 emit.array(items)
512 taskPending = false
513 })
514 }
515 })
516 query.on("end", () => {
517 if (buffer.length > 0) {
518 emit.array(buffer)
519 buffer = []
520 }
521 emit.end()
522 })
523
524 return {
525 onPause() {
526 query.pause()
527 },
528 onResume() {
529 query.resume()
530 }
531 }
532 }))
533}