smartmigration/ts/classes.smartmigration.ts

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;

/**
 * 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.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.',
      );
    }

    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 wasFreshInstall = false;
    let currentVersionBefore: string | null = null;

    try {
      // Re-read after acquiring lock (state may have changed while we waited).
      let data = await ledger.read();
      currentVersionBefore = data.currentVersion;

      // Resolve initial version.
      let currentVersion: string;
      if (data.currentVersion === null) {
        const fresh = await this.detectFreshInstall();
        if (fresh && this.settings.freshInstallVersion) {
          wasFreshInstall = true;
          currentVersion = this.settings.freshInstallVersion;
          VersionResolver.assertValid(currentVersion, 'freshInstallVersion');
          data.currentVersion = currentVersion;
          await ledger.write(data);
          this.log.log('info', `smartmigration: fresh install detected, jumping to ${currentVersion}`);
        } else {
          if (this.steps.length === 0) {
            // No steps and no current version — nothing to do.
            data.currentVersion = this.settings.targetVersion;
            await ledger.write(data);
            return {
              currentVersionBefore: null,
              currentVersionAfter: this.settings.targetVersion,
              targetVersion: this.settings.targetVersion,
              wasUpToDate: false,
              wasFreshInstall: true,
              stepsApplied: [],
              stepsSkipped: [],
              totalDurationMs: Date.now() - runStart,
            };
          }
          currentVersion = this.steps[0].fromVersion;
          data.currentVersion = currentVersion;
          await ledger.write(data);
        }
      } else {
        currentVersion = data.currentVersion;
      }

      // Already at target after fresh-install resolution?
      if (VersionResolver.equals(currentVersion, this.settings.targetVersion)) {
        return {
          currentVersionBefore,
          currentVersionAfter: currentVersion,
          targetVersion: this.settings.targetVersion,
          wasUpToDate: true,
          wasFreshInstall,
          stepsApplied: [],
          stepsSkipped: [],
          totalDurationMs: Date.now() - runStart,
        };
      }

      const plan = VersionResolver.computePlan(
        this.steps,
        currentVersion,
        this.settings.targetVersion,
      );

      for (const step of plan) {
        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);

          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;
          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;
          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 },
          );
        }
      }

      const finalData = await ledger.read();
      return {
        currentVersionBefore,
        currentVersionAfter: finalData.currentVersion ?? this.settings.targetVersion,
        targetVersion: this.settings.targetVersion,
        wasUpToDate: false,
        wasFreshInstall,
        stepsApplied: applied,
        stepsSkipped: [],
        totalDurationMs: Date.now() - runStart,
      };
    } finally {
      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 computeResultWithoutRun(data: ISmartMigrationLedgerData): IMigrationRunResult {
    const currentVersion =
      data.currentVersion ??
      (this.steps.length > 0 ? this.steps[0].fromVersion : this.settings.targetVersion);

    if (VersionResolver.equals(currentVersion, this.settings.targetVersion)) {
      return {
        currentVersionBefore: data.currentVersion,
        currentVersionAfter: currentVersion,
        targetVersion: this.settings.targetVersion,
        wasUpToDate: true,
        wasFreshInstall: false,
        stepsApplied: [],
        stepsSkipped: [],
        totalDurationMs: 0,
      };
    }

    const plan = VersionResolver.computePlan(
      this.steps,
      currentVersion,
      this.settings.targetVersion,
    );

    const skipped: IMigrationStepResult[] = plan.map((step) => ({
      id: step.id,
      fromVersion: step.fromVersion,
      toVersion: step.toVersion,
      status: 'skipped' as const,
      startedAt: '',
      finishedAt: '',
      durationMs: 0,
    }));

    return {
      currentVersionBefore: data.currentVersion,
      currentVersionAfter: currentVersion,
      targetVersion: this.settings.targetVersion,
      wasUpToDate: false,
      wasFreshInstall: false,
      stepsApplied: [],
      stepsSkipped: skipped,
      totalDurationMs: 0,
    };
  }

  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;
  }

  /**
   * 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) {
      const cursor = (this.settings.bucket as any).createCursor('', { pageSize: 5 });
      const batch = (await cursor.next()) as string[] | undefined;
      if (batch && batch.length > 0) {
        const userKeys = batch.filter((k) => !k.startsWith('.smartmigration/'));
        if (userKeys.length > 0) return false;
      }
    }
    return true;
  }

  private isReservedCollectionName(name: string): boolean {
    // smartdata's EasyStore creates a collection named "SmartdataEasyStore".
    return name === 'SmartdataEasyStore' || name.startsWith('system.');
  }

  private sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }
}