Add beforeAll/afterAll events + file locking

.eslintrc.js

@@ -89,13 +89,15 @@ module.exports = {
 		'no-warning-comments': 'off',
 		'no-dupe-class-members': 'off',
 		'capitalized-comments': 'off',
+		'no-promise-executor-return': 'off',
 
 		'unicorn/catch-error-name': 'off',
 		'unicorn/consistent-function-scoping': 'off',
 		'unicorn/expiring-todo-comments': 'warn',
 		'unicorn/no-fn-reference-in-iterator': 'off',
 		'unicorn/no-null': 'off',
 		'unicorn/prevent-abbreviations': 'off',
+		'unicorn/no-useless-undefined': 'off',
 	},
 	overrides: [
 		{

README.md

@@ -519,6 +519,13 @@ Umzug is an [emittery event emitter](https://www.npmjs.com/package/emittery). Ea
 * `reverting` - A migration is about to be reverted.
 * `reverted` - A migration has successfully been reverted.
 
+These events run at the beginning and end of `up` and `down` calls. They'll receive an object containing a `context` property:
+
+- `beforeAll` - Before any of the migrations are run.
+- `afterAll` - After all the migrations have been executed. Note: this will always run, even if migrations throw an error.
+
+The [`FileLocker` class](./src/file-locker.ts) uses `beforeAll` and `afterAll` to implement a simple filesystem-based locking mechanism.
+
 All events are type-safe, so IDEs will prevent typos and supply strong types for the event payloads.
 
 ## License

src/file-locker.ts

@@ -0,0 +1,89 @@
+import * as fs from 'fs';
+import * as path from 'path';
+import { Umzug } from './umzug';
+
+export interface FileLockerOptions {
+	path: string;
+	fs?: typeof fs;
+}
+
+/**
+ * Simple locker using the filesystem. Only one lock can be held per file. An error will be thrown if the
+ * lock file already exists.
+ *
+ * @example
+ * const umzug = new Umzug({ ... })
+ * FileLocker.attach(umzug, { path: 'path/to/lockfile' })
+ *
+ * @docs
+ * To wait for the lock to be free instead of throwing, you could extend it (the below example uses `setInterval`,
+ * but depending on your use-case, you may want to use a library with retry/backoff):
+ *
+ * @example
+ * class WaitingFileLocker extends FileLocker {
+ *   async getLock() {
+ *     return new Promise(resolve => setInterval(
+ *       () => super.getLock().then(resolve).catch(),
+ *       500,
+ *     )
+ *   }
+ * }
+ *
+ * const locker = new WaitingFileLocker({ path: 'path/to/lockfile' })
+ * locker.attachTo(umzug)
+ */
+export class FileLocker {
+	private readonly lockFile: string;
+	private readonly fs: typeof fs;
+
+	constructor(params: FileLockerOptions) {
+		this.lockFile = params.path;
+		this.fs = params.fs ?? fs;
+	}
+
+	/** Attach `beforeAll` and `afterAll` events to an umzug instance which use the specified filepath */
+	static attach(umzug: Umzug<unknown>, params: FileLockerOptions): void {
+		const locker = new FileLocker(params);
+		locker.attachTo(umzug);
+	}
+
+	/** Attach `beforeAll` and `afterAll` events to an umzug instance */
+	attachTo(umzug: Umzug<unknown>): void {
+		umzug.on('beforeAll', async () => this.getLock());
+		umzug.on('afterAll', async () => this.releaseLock());
+	}
+
+	private async readFile(filepath: string): Promise<string | undefined> {
+		return this.fs.promises.readFile(filepath).then(
+			buf => buf.toString(),
+			() => undefined
+		);
+	}
+
+	private async writeFile(filepath: string, content: string): Promise<void> {
+		await this.fs.promises.mkdir(path.dirname(filepath), { recursive: true });
+		await this.fs.promises.writeFile(filepath, content);
+	}
+
+	private async removeFile(filepath: string): Promise<void> {
+		await this.fs.promises.unlink(filepath);
+	}
+
+	async getLock(): Promise<void> {
+		const existing = await this.readFile(this.lockFile);
+		if (existing) {
+			throw new Error(`Can't acquire lock. ${this.lockFile} exists`);
+		}
+
+		await this.writeFile(this.lockFile, 'lock');
+	}
+
+	async releaseLock(): Promise<void> {
+		const existing = await this.readFile(this.lockFile);
+		if (!existing) {
+			throw new Error(`Nothing to unlock`);
+		}
+
+		await this.removeFile(this.lockFile);
+	}
+}

src/index.ts

@@ -1,2 +1,3 @@
 export * from './umzug';
 export * from './storage';
+export * from './file-locker';

src/umzug.ts

@@ -122,7 +122,8 @@ export type MigrateDownOptions = MergeExclusive<
 >;
 
 export class Umzug<Ctx> extends emittery.Typed<
-	Record<'migrating' | 'migrated' | 'reverting' | 'reverted', MigrationParams<Ctx>>
+	Record<'migrating' | 'migrated' | 'reverting' | 'reverted', MigrationParams<Ctx>> &
+		Record<'beforeAll' | 'afterAll', { context: Ctx }>
 > {
 	private readonly storage: UmzugStorage;
 	private readonly migrations: () => Promise<ReadonlyArray<RunnableMigration<Ctx>>>;
@@ -252,6 +253,15 @@ export class Umzug<Ctx> extends emittery.Typed<
 		return migrations.filter(m => !executedSet.has(m.name));
 	}
 
+	private async withBeforeAfterHooks<T>(cb: () => Promise<T>): Promise<T> {
+		await this.emit('beforeAll', { context: this.context });
+		try {
+			return await cb();
+		} finally {
+			await this.emit('afterAll', { context: this.context });
+		}
+	}
+
 	/**
 	 * Apply migrations. By default, runs all pending migrations.
 	 * @see MigrateUpOptions for other use cases using `to`, `migrations` and `rerun`.
@@ -284,25 +294,27 @@ export class Umzug<Ctx> extends emittery.Typed<
 			return allPending.slice(0, sliceIndex);
 		};
 
-		const toBeApplied = await eligibleMigrations();
+		return this.withBeforeAfterHooks(async () => {
+			const toBeApplied = await eligibleMigrations();
 
-		for (const m of toBeApplied) {
-			const start = Date.now();
-			const params: MigrationParams<Ctx> = { name: m.name, path: m.path, context: this.context };
+			for (const m of toBeApplied) {
+				const start = Date.now();
+				const params: MigrationParams<Ctx> = { name: m.name, path: m.path, context: this.context };
 
-			this.logging({ event: 'migrating', name: m.name });
-			await this.emit('migrating', params);
+				this.logging({ event: 'migrating', name: m.name });
+				await this.emit('migrating', params);
 
-			await m.up(params);
+				await m.up(params);
 
-			await this.storage.logMigration(m.name);
+				await this.storage.logMigration(m.name);
 
-			const duration = (Date.now() - start) / 1000;
-			this.logging({ event: 'migrated', name: m.name, durationSeconds: duration });
-			await this.emit('migrated', params);
-		}
+				const duration = (Date.now() - start) / 1000;
+				this.logging({ event: 'migrated', name: m.name, durationSeconds: duration });
+				await this.emit('migrated', params);
+			}
 
-		return toBeApplied.map(m => ({ name: m.name, path: m.path }));
+			return toBeApplied.map(m => ({ name: m.name, path: m.path }));
+		});
 	}
 
 	/**
@@ -338,25 +350,27 @@ export class Umzug<Ctx> extends emittery.Typed<
 			return executedReversed.slice(0, sliceIndex);
 		};
 
-		const toBeReverted = await eligibleMigrations();
+		return this.withBeforeAfterHooks(async () => {
+			const toBeReverted = await eligibleMigrations();
 
-		for (const m of toBeReverted) {
-			const start = Date.now();
-			const params: MigrationParams<Ctx> = { name: m.name, path: m.path, context: this.context };
+			for (const m of toBeReverted) {
+				const start = Date.now();
+				const params: MigrationParams<Ctx> = { name: m.name, path: m.path, context: this.context };
 
-			this.logging({ event: 'reverting', name: m.name });
-			await this.emit('reverting', params);
+				this.logging({ event: 'reverting', name: m.name });
+				await this.emit('reverting', params);
 
-			await m.down?.(params);
+				await m.down?.(params);
 
-			await this.storage.unlogMigration(m.name);
+				await this.storage.unlogMigration(m.name);
 
-			const duration = Number.parseFloat(((Date.now() - start) / 1000).toFixed(3));
-			this.logging({ event: 'reverted', name: m.name, durationSeconds: duration });
-			await this.emit('reverted', params);
-		}
+				const duration = Number.parseFloat(((Date.now() - start) / 1000).toFixed(3));
+				this.logging({ event: 'reverted', name: m.name, durationSeconds: duration });
+				await this.emit('reverted', params);
+			}
 
-		return toBeReverted.map(m => ({ name: m.name, path: m.path }));
+			return toBeReverted.map(m => ({ name: m.name, path: m.path }));
+		});
 	}
 
 	private findNameIndex(migrations: Array<RunnableMigration<Ctx>>, name: string) {

test/lock.test.ts

@@ -0,0 +1,39 @@
+import { JSONStorage, FileLocker, Umzug } from '../src';
+import * as path from 'path';
+import { fsSyncer } from 'fs-syncer';
+import * as pEvent from 'p-event';
+
+const names = (migrations: Array<{ name: string }>) => migrations.map(m => m.name);
+const delay = async (ms: number) => new Promise(r => setTimeout(r, ms));
+
+describe('locks', () => {
+	const syncer = fsSyncer(path.join(__dirname, 'generated/lock/json'), {});
+	syncer.sync();
+
+	test('file lock', async () => {
+		const umzug = new Umzug({
+			migrations: [1, 2].map(n => ({
+				name: `m${n}`,
+				up: async () => delay(100),
+			})),
+			storage: new JSONStorage({ path: path.join(syncer.baseDir, 'storage.json') }),
+			logger: undefined,
+		});
+
+		FileLocker.attach(umzug, { path: path.join(syncer.baseDir, 'storage.json.lock') });
+
+		expect(syncer.read()).toEqual({});
+
+		const promise1 = umzug.up();
+		await pEvent(umzug, 'migrating');
+		const promise2 = umzug.up();
+
+		await expect(promise2).rejects.toThrowError(/Can't acquire lock. (.*)storage.json.lock exists/);
+		await expect(promise1.then(names)).resolves.toEqual(['m1', 'm2']);
+
+		expect(names(await umzug.executed())).toEqual(['m1', 'm2']);
+		expect(syncer.read()).toEqual({
+			'storage.json': JSON.stringify(['m1', 'm2'], null, 2),
+		});
+	});
+});