1/**
2 * Runs SQL migrations with `SqlClient`.
3 *
4 * A migrator loads numbered migration effects, records completed ids in a
5 * migrations table, and runs only pending migrations in a transaction. It
6 * creates the table when needed, detects duplicate ids, treats concurrent runs
7 * as locked, and can dump the schema after successful migrations.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import * as Arr from "../Array.ts"
13import * as Data from "../Data.ts"
14import * as Effect from "../Effect.ts"
15import { FileSystem } from "../FileSystem.ts"
16import { pipe } from "../Function.ts"
17import * as Option from "../Option.ts"
18import * as Order from "../Order.ts"
19import { Path } from "../Path.ts"
20import * as Client from "./SqlClient.ts"
21import type { SqlError } from "./SqlError.ts"
22
23/**
24 * Options for running SQL migrations, including the migration loader, optional
25 * schema dump directory, and migrations table name.
26 *
27 * @stability unstable
28 * @category options
29 * @since 4.0.0
30 */
31export interface MigratorOptions<R = never> {
32 readonly loader: Loader<R>
33 readonly schemaDirectory?: string
34 readonly table?: string
35}
36
37/**
38 * Effect that resolves the available migrations for the migrator or fails with a
39 * `MigrationError`.
40 *
41 * @stability unstable
42 * @category models
43 * @since 4.0.0
44 */
45export type Loader<R = never> = Effect.Effect<
46 ReadonlyArray<ResolvedMigration>,
47 MigrationError,
48 R
49>
50
51/**
52 * Tuple produced by a migration loader, containing the migration id, migration
53 * name, and an effect that loads the migration implementation.
54 *
55 * @stability unstable
56 * @category models
57 * @since 4.0.0
58 */
59export type ResolvedMigration = readonly [
60 id: number,
61 name: string,
62 load: Effect.Effect<any, any, Client.SqlClient>
63]
64
65/**
66 * Metadata for a migration recorded in the migrations table, including its id,
67 * name, and creation timestamp.
68 *
69 * @stability unstable
70 * @category models
71 * @since 4.0.0
72 */
73export interface Migration {
74 readonly id: number
75 readonly name: string
76 readonly createdAt: Date
77}
78
79/**
80 * Error raised while loading, validating, locking, or running SQL migrations.
81 *
82 * @stability unstable
83 * @category errors
84 * @since 4.0.0
85 */
86export class MigrationError extends Data.TaggedError("MigrationError")<{
87 readonly _tag: "MigrationError"
88 readonly cause?: unknown
89 readonly kind:
90 | "BadState"
91 | "ImportError"
92 | "Failed"
93 | "Duplicates"
94 | "Locked"
95 readonly message: string
96}> {}
97
98/**
99 * Creates a migrator that ensures the migrations table exists, runs pending
100 * migrations in a transaction, and optionally dumps the schema after successful
101 * migrations.
102 *
103 * @stability unstable
104 * @category constructors
105 * @since 4.0.0
106 */
107export const make = <RD = never>({
108 dumpSchema = () => Effect.void
109}: {
110 dumpSchema?: (
111 path: string,
112 migrationsTable: string
113 ) => Effect.Effect<void, MigrationError, RD>
114}) =>
115<R2 = never>({
116 loader,
117 schemaDirectory,
118 table = "effect_sql_migrations"
119}: MigratorOptions<R2>): Effect.Effect<
120 ReadonlyArray<readonly [id: number, name: string]>,
121 MigrationError | SqlError,
122 Client.SqlClient | RD | R2
123> =>
124 Effect.gen(function*() {
125 const sql = yield* Client.SqlClient
126
127 const ensureMigrationsTable = sql.onDialectOrElse({
128 mssql: () =>
129 sql`IF OBJECT_ID(N'${sql.literal(table)}', N'U') IS NULL
130 CREATE TABLE ${sql(table)} (
131 migration_id INT NOT NULL PRIMARY KEY,
132 name VARCHAR(255) NOT NULL,
133 created_at DATETIME NOT NULL DEFAULT GETDATE()
134 )`,
135 mysql: () =>
136 sql`CREATE TABLE IF NOT EXISTS ${sql(table)} (
137 migration_id INTEGER UNSIGNED NOT NULL,
138 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
139 name VARCHAR(255) NOT NULL,
140 PRIMARY KEY (migration_id)
141)`,
142 pg: () =>
143 Effect.catch(
144 sql`select ${table}::regclass`,
145 () =>
146 sql`CREATE TABLE ${sql(table)} (
147 migration_id integer primary key,
148 created_at timestamp with time zone not null default now(),
149 name text not null
150)`
151 ),
152 orElse: () =>
153 sql`CREATE TABLE IF NOT EXISTS ${sql(table)} (
154 migration_id integer PRIMARY KEY NOT NULL,
155 created_at datetime NOT NULL DEFAULT current_timestamp,
156 name VARCHAR(255) NOT NULL
157)`
158 })
159
160 const insertMigrations = (
161 rows: ReadonlyArray<[id: number, name: string]>
162 ) =>
163 sql`INSERT INTO ${sql(table)} ${
164 sql.insert(
165 rows.map(([migration_id, name]) => ({ migration_id, name }))
166 )
167 }`.withoutTransform
168
169 const latestMigration = Effect.map(
170 sql<{ migration_id: number; name: string; created_at: Date }>`SELECT migration_id, name, created_at FROM ${
171 sql(table)
172 } ORDER BY migration_id DESC`.withoutTransform,
173 (_) =>
174 Option.map(
175 Option.fromNullishOr(_[0] as any),
176 ({ created_at, migration_id, name }): Migration => ({
177 id: migration_id,
178 name,
179 createdAt: created_at
180 })
181 )
182 )
183
184 const loadMigration = ([id, name, load]: ResolvedMigration) =>
185 Effect.catchDefect(load, (_) =>
186 Effect.fail(
187 new MigrationError({
188 kind: "ImportError",
189 message: `Could not import migration "${id}_${name}"\n\n${_}`
190 })
191 )).pipe(
192 Effect.flatMap((_) =>
193 Effect.isEffect(_)
194 ? Effect.succeed(_)
195 : _.default
196 ? Effect.succeed(_.default?.default ?? _.default)
197 : Effect.fail(
198 new MigrationError({
199 kind: "ImportError",
200 message: `Default export not found for migration "${id}_${name}"`
201 })
202 )
203 ),
204 Effect.filterOrFail(
205 (_): _ is Effect.Effect<any, any, Client.SqlClient> => Effect.isEffect(_),
206 () =>
207 new MigrationError({
208 kind: "ImportError",
209 message: `Default export was not an Effect for migration "${id}_${name}"`
210 })
211 )
212 )
213
214 const runMigration = (
215 id: number,
216 name: string,
217 effect: Effect.Effect<unknown, unknown, Client.SqlClient>
218 ) =>
219 Effect.catch(effect, (error: unknown) =>
220 Effect.die(
221 new MigrationError({
222 cause: error,
223 kind: "Failed",
224 message: `Migration "${id}_${name}" failed`
225 })
226 ))
227
228 // === run
229
230 const run = Effect.gen(function*() {
231 yield* sql.onDialectOrElse({
232 pg: () => sql`LOCK TABLE ${sql(table)} IN ACCESS EXCLUSIVE MODE`,
233 orElse: () => Effect.void
234 })
235
236 const [latestMigrationId, current] = yield* Effect.all([
237 Effect.map(
238 latestMigration,
239 Option.match({
240 onNone: () => 0,
241 onSome: (_) => _.id
242 })
243 ),
244 loader
245 ])
246
247 if (new Set(current.map(([id]) => id)).size !== current.length) {
248 return yield* new MigrationError({
249 kind: "Duplicates",
250 message: "Found duplicate migration id's"
251 })
252 }
253
254 const required: Array<ResolvedMigration> = []
255
256 for (const resolved of current) {
257 const [currentId, currentName] = resolved
258 if (currentId <= latestMigrationId) {
259 continue
260 }
261
262 required.push([
263 currentId,
264 currentName,
265 yield* loadMigration(resolved)
266 ])
267 }
268
269 if (required.length > 0) {
270 yield* pipe(
271 insertMigrations(required.map(([id, name]) => [id, name])),
272 Effect.mapError((error): MigrationError | SqlError =>
273 isConstraintConflict(error)
274 ? new MigrationError({
275 kind: "Locked",
276 message: "Migrations already running"
277 })
278 : error
279 )
280 )
281 }
282
283 yield* Effect.forEach(
284 required,
285 ([id, name, effect]) =>
286 Effect.logDebug(`Running migration`).pipe(
287 Effect.flatMap(() => runMigration(id, name, effect)),
288 Effect.annotateLogs("migration_id", String(id)),
289 Effect.annotateLogs("migration_name", name),
290 Effect.withSpan(`Migrator ${id}_${name}`)
291 ),
292 { discard: true }
293 )
294
295 yield* pipe(
296 latestMigration,
297 Effect.flatMap(
298 Option.match({
299 onNone: () => Effect.logDebug(`Migrations complete`),
300 onSome: (_) =>
301 Effect.logDebug(`Migrations complete`).pipe(
302 Effect.annotateLogs("latest_migration_id", _.id.toString()),
303 Effect.annotateLogs("latest_migration_name", _.name)
304 )
305 })
306 )
307 )
308
309 return required.map(([id, name]) => [id, name] as const)
310 })
311
312 yield* ensureMigrationsTable
313
314 const completed = yield* pipe(
315 sql.withTransaction(run),
316 Effect.catchTag("MigrationError", (_) =>
317 _.kind === "Locked"
318 ? Effect.as(Effect.logDebug(_.message), [])
319 : Effect.fail(_))
320 )
321
322 if (schemaDirectory && completed.length > 0) {
323 yield* dumpSchema(`${schemaDirectory}/_schema.sql`, table).pipe(
324 Effect.catchCause((cause) => Effect.logInfo("Could not dump schema", cause))
325 )
326 }
327
328 return completed
329 })
330
331const migrationOrder = Order.make<ResolvedMigration>(([a], [b]) => Order.Number(a, b))
332
333const isConstraintConflict = (error: SqlError): boolean =>
334 error.reason._tag === "ConstraintError" || error.reason._tag === "UniqueViolation"
335
336/**
337 * Creates a migration loader from a glob record of dynamic import functions,
338 * parsing files named `<id>_<name>.js`, `<id>_<name>.ts`,
339 * `<id>_<name>.mjs`, or `<id>_<name>.mts` and sorting migrations by id.
340 *
341 * @stability unstable
342 * @category loaders
343 * @since 4.0.0
344 */
345export const fromGlob = (
346 migrations: Record<string, () => Promise<any>>
347): Loader =>
348 pipe(
349 Object.keys(migrations),
350 Arr.flatMapNullishOr((_) => _.match(/^(?:.*\/)?(\d+)_([^.]+)\.(js|ts|mjs|mts)$/)),
351 Arr.map(
352 ([key, id, name]): ResolvedMigration => [
353 Number(id),
354 name,
355 Effect.promise(() => migrations[key]())
356 ]
357 ),
358 Arr.sort(migrationOrder),
359 Effect.succeed
360 )
361
362/**
363 * Creates a migration loader from a Babel-style glob record, parsing keys such
364 * as `_<id>_<name>Js`, `_<id>_<name>Ts`, `_<id>_<name>Mjs`, or
365 * `_<id>_<name>Mts` and sorting migrations by id.
366 *
367 * @stability unstable
368 * @category loaders
369 * @since 4.0.0
370 */
371export const fromBabelGlob = (migrations: Record<string, any>): Loader =>
372 pipe(
373 Object.keys(migrations),
374 Arr.flatMapNullishOr((_) => _.match(/^_(\d+)_([^.]+?)(Js|Ts|Mjs|Mts)?$/)),
375 Arr.map(
376 ([key, id, name]): ResolvedMigration => [
377 Number(id),
378 name,
379 Effect.succeed(migrations[key])
380 ]
381 ),
382 Arr.sort(migrationOrder),
383 Effect.succeed
384 )
385
386/**
387 * Creates a migration loader from a record of migration effects keyed by
388 * `<id>_<name>`, sorted by migration id.
389 *
390 * @stability unstable
391 * @category loaders
392 * @since 4.0.0
393 */
394export const fromRecord = (migrations: Record<string, Effect.Effect<void, unknown, Client.SqlClient>>): Loader =>
395 pipe(
396 Object.keys(migrations),
397 Arr.flatMapNullishOr((_) => _.match(/^(\d+)_(.+)$/)),
398 Arr.map(
399 ([key, id, name]): ResolvedMigration => [
400 Number(id),
401 name,
402 Effect.succeed(migrations[key])
403 ]
404 ),
405 Arr.sort(migrationOrder),
406 Effect.succeed
407 )
408
409/**
410 * Creates a migration loader that reads a directory with `FileSystem`, imports
411 * files named `<id>_<name>.js`, `<id>_<name>.ts`,
412 * `<id>_<name>.mjs`, or `<id>_<name>.mts`, and sorts migrations by id.
413 *
414 * **Details**
415 *
416 * Requires a `Path` service appropriate for the migration directory's path
417 * syntax. On Windows, prefer a platform-aware implementation such as
418 * `NodePath.layer`; the core `Path.layer` uses POSIX semantics and does not
419 * preserve Windows drive-letter paths.
420 *
421 * @stability unstable
422 * @category loaders
423 * @since 4.0.0
424 */
425export const fromFileSystem: (directory: string) => Loader<FileSystem | Path> = Effect.fnUntraced(function*(directory) {
426 const Fs = yield* FileSystem
427 const path = yield* Path
428 const files = yield* Effect.mapError(
429 Fs.readDirectory(directory),
430 (cause) =>
431 new MigrationError({
432 kind: "Failed",
433 cause,
434 message: "Failed to read migrations directory"
435 })
436 )
437 return files
438 .map((file) => Option.fromNullishOr(file.match(/^(?:.*\/)?(\d+)_([^.]+)\.(js|ts|mjs|mts)$/)))
439 .flatMap(
440 Option.match({
441 onNone: () => [],
442 onSome: ([basename, id, name]): ReadonlyArray<ResolvedMigration> =>
443 [
444 [
445 Number(id),
446 name,
447 // `import` needs a file URL: on Windows an absolute path such as
448 // `D:\migrations\1_init.ts` is rejected by the ESM loader. `orDie` keeps the
449 // failure a defect so `loadMigration` reports it as an import error.
450 Effect.flatMap(Effect.orDie(path.toFileUrl(path.join(directory, basename))), (url) =>
451 Effect.promise(
452 () =>
453 import(
454 /* @vite-ignore */
455 /* webpackIgnore: true */
456 url.href
457 )
458 ))
459 ]
460 ] as const
461 })
462 )
463 .sort(([a], [b]) => a - b)
464})