Files
smartmigration/ts/classes.smartmigration.ts
T

625 lines
21 KiB
TypeScript

import * as plugins from './plugins.js';
import { logger as defaultLogger } from './logger.js';
import { MigrationStepBuilder } from './classes.migrationstep.js';
import { buildContext } from './classes.migrationcontext.js';
import { SmartMigrationError, VersionResolver } from './classes.versionresolver.js';
import { Ledger } from './ledgers/classes.ledger.js';
import { MongoLedger } from './ledgers/classes.mongoledger.js';
import { S3Ledger } from './ledgers/classes.s3ledger.js';
import type {
IMigrationLedgerEntry,
IMigrationRunResult,
IMigrationStepDefinition,
IMigrationStepResult,
ISmartMigrationLedgerData,
ISmartMigrationOptions,
} from './interfaces.js';
import type { TLedgerBackend } from './types.js';
export { SmartMigrationError };
const DEFAULT_LEDGER_NAME = 'smartmigration';
const DEFAULT_LOCK_WAIT_MS = 60_000;
const DEFAULT_LOCK_TTL_MS = 600_000;
const LOCK_POLL_INTERVAL_MS = 500;
const MIN_LOCK_HEARTBEAT_MS = 1;
interface IResolvedLedgerState {
currentVersionBefore: string | null;
effectiveCurrentVersion: string;
bootstrapVersionToPersist: string | null;
bootstrapMode: 'none' | 'fresh-install-version' | 'chain-start' | 'target-without-steps';
wasFreshInstall: boolean;
plannedSteps: IMigrationStepDefinition[];
predictedCurrentVersionAfter: string;
}
interface ILockHeartbeat {
getError(): SmartMigrationError | null;
stop(): Promise<void>;
}
/**
* SmartMigration — the runner. See readme.md for the full API.
*
* Lifecycle:
* 1. `new SmartMigration({...})` — creates the runner with the target version.
* 2. `.step('id').from(...).to(...).up(async ctx => { ... })` — register steps
* in the desired execution order.
* 3. `await migration.run()` — invoked at SaaS app startup; idempotent and
* fast on the happy path.
*/
export class SmartMigration {
public readonly settings: Required<
Omit<ISmartMigrationOptions, 'db' | 'bucket' | 'freshInstallVersion' | 'logger'>
> & Pick<ISmartMigrationOptions, 'db' | 'bucket' | 'freshInstallVersion' | 'logger'>;
private steps: IMigrationStepDefinition[] = [];
private ledger: Ledger | null = null;
private instanceId: string;
private log: plugins.smartlog.Smartlog;
constructor(options: ISmartMigrationOptions) {
if (!options || typeof options !== 'object') {
throw new SmartMigrationError('INVALID_OPTIONS', 'SmartMigration requires an options object.');
}
if (!options.targetVersion) {
throw new SmartMigrationError(
'MISSING_TARGET_VERSION',
'SmartMigration requires `targetVersion` in its options.',
);
}
VersionResolver.assertValid(options.targetVersion, 'options.targetVersion');
if (options.freshInstallVersion !== undefined) {
VersionResolver.assertValid(options.freshInstallVersion, 'options.freshInstallVersion');
}
if (!options.db && !options.bucket) {
throw new SmartMigrationError(
'NO_RESOURCES',
'SmartMigration requires at least one of `db` (SmartdataDb) or `bucket` (smartbucket Bucket).',
);
}
const ledgerBackend: TLedgerBackend =
options.ledgerBackend ?? (options.db ? 'mongo' : 's3');
if (ledgerBackend === 'mongo' && !options.db) {
throw new SmartMigrationError(
'LEDGER_BACKEND_MISMATCH',
'ledgerBackend "mongo" requires `db` to be set.',
);
}
if (ledgerBackend === 's3' && !options.bucket) {
throw new SmartMigrationError(
'LEDGER_BACKEND_MISMATCH',
'ledgerBackend "s3" requires `bucket` to be set.',
);
}
if (
options.ledgerName !== undefined &&
(typeof options.ledgerName !== 'string' || options.ledgerName.trim() === '')
) {
throw new SmartMigrationError(
'INVALID_LEDGER_NAME',
'ledgerName must be a non-empty string when provided.',
);
}
if (options.lockWaitMs !== undefined) {
this.assertIntegerOption('INVALID_LOCK_WAIT_MS', 'lockWaitMs', options.lockWaitMs, {
min: 0,
});
}
if (options.lockTtlMs !== undefined) {
this.assertIntegerOption('INVALID_LOCK_TTL_MS', 'lockTtlMs', options.lockTtlMs, {
min: 1,
});
}
this.settings = {
targetVersion: options.targetVersion,
db: options.db,
bucket: options.bucket,
ledgerName: options.ledgerName ?? DEFAULT_LEDGER_NAME,
ledgerBackend,
freshInstallVersion: options.freshInstallVersion,
lockWaitMs: options.lockWaitMs ?? DEFAULT_LOCK_WAIT_MS,
lockTtlMs: options.lockTtlMs ?? DEFAULT_LOCK_TTL_MS,
dryRun: options.dryRun ?? false,
logger: options.logger,
};
this.log = options.logger ?? defaultLogger;
this.instanceId = plugins.randomUUID();
}
// ─── public API ────────────────────────────────────────────────────────────
/** Begin defining a step. Returns a chainable builder. */
public step(id: string): MigrationStepBuilder {
if (!id || typeof id !== 'string') {
throw new SmartMigrationError('INVALID_STEP_ID', 'step(id) requires a non-empty string id.');
}
return new MigrationStepBuilder(this, id);
}
/**
* Register a step. Called by `MigrationStepBuilder.up()`. End users do not
* normally call this directly.
*/
public registerStep(definition: IMigrationStepDefinition): void {
this.steps.push(definition);
}
/** All registered steps in registration order. Mostly useful for tests. */
public getRegisteredSteps(): ReadonlyArray<IMigrationStepDefinition> {
return this.steps;
}
/** Returns the current data version from the ledger, or null if uninitialized. */
public async getCurrentVersion(): Promise<string | null> {
const ledger = await this.ensureLedger();
const data = await ledger.read();
return data.currentVersion;
}
/** Returns the plan that would be executed without actually running it. */
public async plan(): Promise<IMigrationRunResult> {
return this.runInternal({ planOnly: true });
}
/** The startup entry point. Idempotent. Fast no-op if already at target. */
public async run(): Promise<IMigrationRunResult> {
return this.runInternal({ planOnly: false });
}
// ─── internals ─────────────────────────────────────────────────────────────
private async runInternal(opts: { planOnly: boolean }): Promise<IMigrationRunResult> {
VersionResolver.validateChain(this.steps);
const ledger = await this.ensureLedger();
// Fast path: read once, return immediately if already at target.
const earlyData = await ledger.read();
if (
earlyData.currentVersion !== null &&
VersionResolver.equals(earlyData.currentVersion, this.settings.targetVersion)
) {
return {
currentVersionBefore: earlyData.currentVersion,
currentVersionAfter: earlyData.currentVersion,
targetVersion: this.settings.targetVersion,
wasUpToDate: true,
wasFreshInstall: false,
stepsApplied: [],
stepsSkipped: [],
totalDurationMs: 0,
};
}
// For dry runs, skip the lock entirely — we don't write anything.
if (opts.planOnly || this.settings.dryRun) {
return this.computeResultWithoutRun(earlyData);
}
const lockHeld = await this.acquireLockWithBackoff();
if (!lockHeld) {
throw new SmartMigrationError(
'LOCK_TIMEOUT',
`Could not acquire migration lock within ${this.settings.lockWaitMs}ms. Another instance may be running migrations.`,
{ lockWaitMs: this.settings.lockWaitMs },
);
}
const runStart = Date.now();
const applied: IMigrationStepResult[] = [];
let lockHeartbeat: ILockHeartbeat | null = null;
try {
lockHeartbeat = this.startLockHeartbeat(ledger);
// Re-read after acquiring lock (state may have changed while we waited).
let data = await ledger.read();
const resolvedState = await this.resolveLedgerState(data);
if (resolvedState.bootstrapVersionToPersist !== null) {
data.currentVersion = resolvedState.bootstrapVersionToPersist;
await ledger.write(data);
if (resolvedState.bootstrapMode === 'fresh-install-version') {
this.log.log(
'info',
`smartmigration: fresh install detected, jumping to ${resolvedState.bootstrapVersionToPersist}`,
);
}
}
this.assertLockHealthy(lockHeartbeat);
if (resolvedState.plannedSteps.length === 0) {
return {
currentVersionBefore: resolvedState.currentVersionBefore,
currentVersionAfter: resolvedState.predictedCurrentVersionAfter,
targetVersion: this.settings.targetVersion,
wasUpToDate: resolvedState.bootstrapMode !== 'target-without-steps',
wasFreshInstall: resolvedState.wasFreshInstall,
stepsApplied: [],
stepsSkipped: [],
totalDurationMs: Date.now() - runStart,
};
}
let currentVersion = resolvedState.effectiveCurrentVersion;
for (const step of resolvedState.plannedSteps) {
const startedAt = new Date();
const stepStart = Date.now();
let entry: IMigrationLedgerEntry;
try {
// Detect skip-forward resume: the running ledger cursor is past
// this step's fromVersion but hasn't yet reached its toVersion.
// The step handler is being run against data that may already be
// partially in the target shape — handlers must be idempotent.
const isSkipForward = VersionResolver.greaterThan(
currentVersion,
step.fromVersion,
);
if (isSkipForward) {
this.log.log(
'info',
`smartmigration: step "${step.id}" running in skip-forward mode ` +
`(ledger at "${currentVersion}", step starts at "${step.fromVersion}"). ` +
`Step handler must be idempotent.`,
);
}
this.log.log(
'info',
`smartmigration: running step "${step.id}" (${step.fromVersion}${step.toVersion})`,
);
const ctx = buildContext({
step,
options: { ...this.settings, db: this.settings.db, bucket: this.settings.bucket },
ledger,
isDryRun: false,
log: this.log,
});
await step.handler(ctx);
this.assertLockHealthy(lockHeartbeat, { stepId: step.id });
const finishedAt = new Date();
const durationMs = Date.now() - stepStart;
entry = {
id: step.id,
fromVersion: step.fromVersion,
toVersion: step.toVersion,
status: 'applied',
startedAt: startedAt.toISOString(),
finishedAt: finishedAt.toISOString(),
durationMs,
};
// Re-read ledger to pick up any checkpoint writes the step made.
data = await ledger.read();
data.steps[step.id] = entry;
data.currentVersion = step.toVersion;
delete data.checkpoints[step.id];
await ledger.write(data);
// Advance the running cursor used by skip-forward detection.
currentVersion = step.toVersion;
applied.push({ ...entry });
this.log.log(
'info',
`smartmigration: step "${step.id}" applied in ${durationMs}ms`,
);
} catch (err) {
const finishedAt = new Date();
const durationMs = Date.now() - stepStart;
const error = err as Error;
const lockError = this.getLockHealthError(lockHeartbeat, {
stepId: step.id,
originalError: error.message,
stack: error.stack,
});
if (lockError) {
throw lockError;
}
entry = {
id: step.id,
fromVersion: step.fromVersion,
toVersion: step.toVersion,
status: 'failed',
startedAt: startedAt.toISOString(),
finishedAt: finishedAt.toISOString(),
durationMs,
error: { message: error.message, stack: error.stack },
};
// Persist failure to ledger so re-runs see it.
try {
data = await ledger.read();
data.steps[step.id] = entry;
await ledger.write(data);
} catch {
// Ledger write failed — re-throw the original error anyway.
}
this.log.log(
'error',
`smartmigration: step "${step.id}" failed after ${durationMs}ms: ${error.message}`,
);
throw new SmartMigrationError(
'STEP_FAILED',
`Migration step "${step.id}" (${step.fromVersion}${step.toVersion}) failed: ${error.message}`,
{ stepId: step.id, originalError: error.message, stack: error.stack },
);
}
}
this.assertLockHealthy(lockHeartbeat);
const finalData = await ledger.read();
return {
currentVersionBefore: resolvedState.currentVersionBefore,
currentVersionAfter: finalData.currentVersion ?? this.settings.targetVersion,
targetVersion: this.settings.targetVersion,
wasUpToDate: false,
wasFreshInstall: resolvedState.wasFreshInstall,
stepsApplied: applied,
stepsSkipped: [],
totalDurationMs: Date.now() - runStart,
};
} finally {
if (lockHeartbeat) {
await lockHeartbeat.stop();
}
await ledger.releaseLock(this.instanceId).catch((err) => {
this.log.log(
'warn',
`smartmigration: failed to release lock: ${(err as Error).message}`,
);
});
}
}
/**
* Resolve the plan against the current ledger state without acquiring a
* lock or executing anything. Used by `plan()` and `dryRun: true`.
*/
private async computeResultWithoutRun(data: ISmartMigrationLedgerData): Promise<IMigrationRunResult> {
const resolvedState = await this.resolveLedgerState(data);
const skipped: IMigrationStepResult[] = resolvedState.plannedSteps.map((step) => ({
id: step.id,
fromVersion: step.fromVersion,
toVersion: step.toVersion,
status: 'skipped' as const,
startedAt: '',
finishedAt: '',
durationMs: 0,
}));
return {
currentVersionBefore: resolvedState.currentVersionBefore,
currentVersionAfter: resolvedState.predictedCurrentVersionAfter,
targetVersion: this.settings.targetVersion,
wasUpToDate:
resolvedState.plannedSteps.length === 0 &&
resolvedState.bootstrapMode !== 'target-without-steps',
wasFreshInstall: resolvedState.wasFreshInstall,
stepsApplied: [],
stepsSkipped: skipped,
totalDurationMs: 0,
};
}
private async resolveLedgerState(
data: ISmartMigrationLedgerData,
): Promise<IResolvedLedgerState> {
let effectiveCurrentVersion: string;
let bootstrapVersionToPersist: string | null = null;
let bootstrapMode: IResolvedLedgerState['bootstrapMode'] = 'none';
let wasFreshInstall = false;
if (data.currentVersion === null) {
const isFreshInstall = await this.detectFreshInstall();
if (isFreshInstall && this.settings.freshInstallVersion) {
effectiveCurrentVersion = this.settings.freshInstallVersion;
bootstrapVersionToPersist = effectiveCurrentVersion;
bootstrapMode = 'fresh-install-version';
wasFreshInstall = true;
} else if (this.steps.length === 0) {
effectiveCurrentVersion = this.settings.targetVersion;
bootstrapVersionToPersist = effectiveCurrentVersion;
bootstrapMode = 'target-without-steps';
wasFreshInstall = isFreshInstall;
} else {
effectiveCurrentVersion = this.steps[0].fromVersion;
bootstrapVersionToPersist = effectiveCurrentVersion;
bootstrapMode = 'chain-start';
}
} else {
effectiveCurrentVersion = data.currentVersion;
}
const plannedSteps = VersionResolver.equals(
effectiveCurrentVersion,
this.settings.targetVersion,
)
? []
: VersionResolver.computePlan(
this.steps,
effectiveCurrentVersion,
this.settings.targetVersion,
);
return {
currentVersionBefore: data.currentVersion,
effectiveCurrentVersion,
bootstrapVersionToPersist,
bootstrapMode,
wasFreshInstall,
plannedSteps,
predictedCurrentVersionAfter:
plannedSteps.length > 0
? plannedSteps[plannedSteps.length - 1].toVersion
: effectiveCurrentVersion,
};
}
private async ensureLedger(): Promise<Ledger> {
if (this.ledger) return this.ledger;
const ledgerName = this.settings.ledgerName;
if (this.settings.ledgerBackend === 'mongo') {
this.ledger = new MongoLedger(this.settings.db!, ledgerName);
} else {
this.ledger = new S3Ledger(this.settings.bucket!, ledgerName);
}
await this.ledger.init();
return this.ledger;
}
private async acquireLockWithBackoff(): Promise<boolean> {
const ledger = await this.ensureLedger();
const deadline = Date.now() + this.settings.lockWaitMs;
while (Date.now() <= deadline) {
const got = await ledger.acquireLock(this.instanceId, this.settings.lockTtlMs);
if (got) return true;
await this.sleep(LOCK_POLL_INTERVAL_MS);
}
return false;
}
private startLockHeartbeat(ledger: Ledger): ILockHeartbeat {
const intervalMs = this.getLockHeartbeatMs();
let stopped = false;
let resolveStop!: () => void;
let lockError: SmartMigrationError | null = null;
const stopSignal = new Promise<void>((resolve) => {
resolveStop = resolve;
});
const loopPromise = (async () => {
while (!stopped) {
await Promise.race([this.sleep(intervalMs), stopSignal]);
if (stopped) {
return;
}
try {
const renewed = await ledger.renewLock(this.instanceId, this.settings.lockTtlMs);
if (!renewed) {
lockError = new SmartMigrationError(
'LOCK_LOST',
'Lost the migration lock while running steps. Another instance may have taken over.',
{ holderId: this.instanceId },
);
stopped = true;
}
} catch (err) {
const error = err as Error;
lockError = new SmartMigrationError(
'LOCK_LOST',
`Failed to renew the migration lock: ${error.message}`,
{ holderId: this.instanceId, originalError: error.message },
);
stopped = true;
}
}
})();
return {
getError: () => lockError,
stop: async () => {
if (stopped) {
await loopPromise;
return;
}
stopped = true;
resolveStop();
await loopPromise;
},
};
}
/**
* Heuristic fresh-install detector. Returns true when neither mongo nor S3
* contain anything besides smartmigration's own ledger artifacts.
*/
private async detectFreshInstall(): Promise<boolean> {
if (this.settings.db) {
const collections = await this.settings.db.mongoDb
.listCollections({}, { nameOnly: true })
.toArray();
const userCollections = collections.filter(
(c: { name: string }) => !this.isReservedCollectionName(c.name),
);
if (userCollections.length > 0) return false;
}
if (this.settings.bucket) {
for await (const key of this.settings.bucket.listAllObjects('')) {
if (!key.startsWith('.smartmigration/')) {
return false;
}
}
}
return true;
}
private isReservedCollectionName(name: string): boolean {
// smartdata's EasyStore creates a collection named "SmartdataEasyStore".
return name === 'SmartdataEasyStore' || name.startsWith('system.');
}
private getLockHeartbeatMs(): number {
return Math.max(MIN_LOCK_HEARTBEAT_MS, Math.floor(this.settings.lockTtlMs / 3));
}
private getLockHealthError(
lockHeartbeat: ILockHeartbeat | null,
details?: Record<string, unknown>,
): SmartMigrationError | null {
const lockError = lockHeartbeat?.getError();
if (!lockError) {
return null;
}
return new SmartMigrationError(lockError.code, lockError.message, {
...lockError.details,
...details,
});
}
private assertLockHealthy(
lockHeartbeat: ILockHeartbeat | null,
details?: Record<string, unknown>,
): void {
const lockError = this.getLockHealthError(lockHeartbeat, details);
if (lockError) {
throw lockError;
}
}
private assertIntegerOption(
code: string,
optionName: string,
value: number,
constraints: { min: number },
): void {
if (!Number.isInteger(value) || value < constraints.min) {
throw new SmartMigrationError(
code,
`${optionName} must be an integer >= ${constraints.min}.`,
{ [optionName]: value },
);
}
}
private sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
}