qinglong/test/back/migrations.test.cjs

require('ts-node/register/transpile-only');

const assert = require('node:assert/strict');
const { afterEach, test } = require('node:test');
const { DataTypes, QueryTypes, Sequelize } = require('sequelize');
const {
  defineSchemaMigrationModel,
} = require('../../back/data/schemaMigration');
const {
  legacyColumnsMigration,
} = require('../../back/migrations/0001-legacy-columns');
const {
  RUN_ATTEMPT_TABLE,
  RUN_EVENT_TABLE,
  RUN_TABLE,
  runSchemaMigration,
} = require('../../back/migrations/0002-run-schema');
const {
  RUNNING_INSTANCE_ATTEMPT_INDEX,
  RUNNING_INSTANCE_RUN_INDEX,
  RUNNING_INSTANCE_TABLE,
  runningInstanceRunReferenceMigration,
} = require('../../back/migrations/0003-running-instance-run-reference');
const {
  RUN_CANCELLATION_REQUEST_INDEX,
  runCancellationRequestMigration,
} = require('../../back/migrations/0004-run-cancellation-request');
const {
  RUN_CANCELLATION_DISPATCH_DUE_INDEX,
  RUN_CANCELLATION_DISPATCH_LEASE_INDEX,
  RUN_CANCELLATION_DISPATCH_TABLE,
  runCancellationDispatchMigration,
} = require('../../back/migrations/0005-run-cancellation-dispatch');
const {
  RUN_ATTEMPT_DEADLINE_INDEX,
  runAttemptDeadlineMigration,
} = require('../../back/migrations/0006-run-attempt-deadline');
const {
  COMPLETION_RECEIPT_JOURNAL_PURGE_INDEX,
  COMPLETION_RECEIPT_JOURNAL_SCAN_INDEX,
  COMPLETION_RECEIPT_JOURNAL_TABLE,
  completionReceiptJournalMigration,
} = require('../../back/migrations/0007-completion-receipt-journal');
const {
  WORKER_REGISTRY_CAPACITY_INDEX,
  WORKER_REGISTRY_LEASE_INDEX,
  WORKER_REGISTRY_TABLE,
  workerRegistryMigration,
} = require('../../back/migrations/0008-worker-registry');
const {
  RUN_DISPATCH_LEASE_EXPIRY_INDEX,
  RUN_DISPATCH_LEASE_TABLE,
  RUN_DISPATCH_LEASE_TOKEN_INDEX,
  RUN_DISPATCH_LEASE_WORKER_INDEX,
  runDispatchLeaseMigration,
} = require('../../back/migrations/0009-run-dispatch-lease');
const { migrations } = require('../../back/migrations');
const { runMigrations } = require('../../back/migrations/runner');

const databases = [];

async function createDatabase() {
  const database = new Sequelize({
    dialect: 'sqlite',
    storage: ':memory:',
    logging: false,
  });
  const migrationModel = defineSchemaMigrationModel(database);
  databases.push(database);
  return { database, migrationModel };
}

afterEach(async () => {
  await Promise.all(databases.splice(0).map((database) => database.close()));
});

test('runs a migration once and records its checksum', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  await queryInterface.createTable('Examples', {
    id: { type: DataTypes.INTEGER, primaryKey: true },
  });

  let calls = 0;
  const migrations = [
    {
      id: '0001-example',
      checksum: 'checksum-v1',
      async up({ queryInterface, transaction }) {
        calls += 1;
        await queryInterface.addColumn(
          'Examples',
          'name',
          { type: DataTypes.STRING },
          { transaction },
        );
      },
    },
  ];

  const options = {
    database,
    migrationModel,
    migrations,
    logger: { info() {} },
  };
  await runMigrations(options);
  await runMigrations(options);

  const description = await queryInterface.describeTable('Examples');
  const applied = await migrationModel.findByPk('0001-example');
  assert.ok(description.name);
  assert.equal(calls, 1);
  assert.equal(applied.checksum, 'checksum-v1');
});

test('rejects a changed checksum for an applied migration', async () => {
  const { database, migrationModel } = await createDatabase();
  const baseOptions = {
    database,
    migrationModel,
    logger: { info() {} },
  };

  await runMigrations({
    ...baseOptions,
    migrations: [
      {
        id: '0001-example',
        checksum: 'checksum-v1',
        async up() {},
      },
    ],
  });

  await assert.rejects(
    runMigrations({
      ...baseOptions,
      migrations: [
        {
          id: '0001-example',
          checksum: 'checksum-v2',
          async up() {},
        },
      ],
    }),
    /Migration checksum mismatch/,
  );
});

test('rolls back the migration record when migration work fails', async () => {
  const { database, migrationModel } = await createDatabase();

  await assert.rejects(
    runMigrations({
      database,
      migrationModel,
      logger: { info() {} },
      migrations: [
        {
          id: '0001-failing',
          checksum: 'checksum-v1',
          async up() {
            throw new Error('migration failed');
          },
        },
      ],
    }),
    /migration failed/,
  );

  assert.equal(await migrationModel.count(), 0);
});

test('rejects duplicate migration ids before touching the database', async () => {
  const { database, migrationModel } = await createDatabase();
  const duplicate = {
    id: '0001-duplicate',
    checksum: 'checksum-v1',
    async up() {},
  };

  await assert.rejects(
    runMigrations({
      database,
      migrationModel,
      migrations: [duplicate, duplicate],
      logger: { info() {} },
    }),
    /Duplicate migration id/,
  );
});

test('upgrades the legacy QingLong tables without swallowing schema errors', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();

  for (const table of ['CrontabViews', 'Subscriptions', 'Crontabs', 'Envs']) {
    await queryInterface.createTable(table, {
      id: { type: DataTypes.INTEGER, primaryKey: true },
    });
  }

  const options = {
    database,
    migrationModel,
    migrations: [legacyColumnsMigration],
    logger: { info() {} },
  };
  await runMigrations(options);
  await runMigrations(options);

  const cronViews = await queryInterface.describeTable('CrontabViews');
  const subscriptions = await queryInterface.describeTable('Subscriptions');
  const crontabs = await queryInterface.describeTable('Crontabs');
  const envs = await queryInterface.describeTable('Envs');

  assert.ok(cronViews.filterRelation);
  assert.ok(cronViews.type);
  assert.ok(subscriptions.proxy);
  assert.ok(subscriptions.autoAddCron);
  assert.ok(subscriptions.autoDelCron);
  assert.ok(crontabs.sub_id);
  assert.ok(crontabs.extra_schedules);
  assert.ok(crontabs.task_before);
  assert.ok(crontabs.task_after);
  assert.ok(crontabs.log_name);
  assert.ok(crontabs.allow_multiple_instances);
  assert.ok(crontabs.work_dir);
  assert.ok(envs.isPinned);
  assert.ok(envs.labels);
  assert.equal(await migrationModel.count(), 1);
});

test('creates the Run aggregate schema with stable uniqueness constraints', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  const options = {
    database,
    migrationModel,
    migrations: [runSchemaMigration],
    logger: { info() {} },
  };

  await runMigrations(options);
  await runMigrations(options);

  const runs = await queryInterface.describeTable(RUN_TABLE);
  const attempts = await queryInterface.describeTable(RUN_ATTEMPT_TABLE);
  const events = await queryInterface.describeTable(RUN_EVENT_TABLE);

  for (const column of [
    'id',
    'project_id',
    'task_id',
    'task_revision',
    'legacy_cron_id',
    'execution_origin',
    'execution_owner',
    'status',
    'version',
    'event_sequence',
    'created_at_ms',
  ]) {
    assert.ok(runs[column], `missing Runs.${column}`);
  }
  for (const column of [
    'id',
    'run_id',
    'attempt',
    'status',
    'executor_type',
    'callback_token_hash',
    'callback_sequence',
    'created_at_ms',
  ]) {
    assert.ok(attempts[column], `missing RunAttempts.${column}`);
  }
  for (const column of [
    'id',
    'run_id',
    'sequence',
    'type',
    'dedupe_key',
    'actor_type',
    'payload',
    'created_at_ms',
  ]) {
    assert.ok(events[column], `missing RunEvents.${column}`);
  }

  const runId = '019f70a0-0000-7000-8000-000000000001';
  const attemptId = '019f70a0-0000-7000-8000-000000000002';
  const baseRun = {
    id: runId,
    project_id: 'default',
    task_id: 'legacy-cron:1',
    task_revision: 'revision-1',
    trigger_type: 'manual',
    execution_origin: 'manual',
    execution_owner: 'legacy',
    status: 'created',
    version: 0,
    event_sequence: 0,
    priority: 0,
    created_at_ms: 1_750_000_000_000,
  };

  await queryInterface.bulkInsert(RUN_TABLE, [
    { ...baseRun, idempotency_key: 'manual-request-1' },
  ]);
  await queryInterface.bulkInsert(RUN_ATTEMPT_TABLE, [
    {
      id: attemptId,
      run_id: runId,
      attempt: 1,
      status: 'claimed',
      executor_type: 'legacy_local',
      callback_sequence: 0,
      created_at_ms: 1_750_000_000_000,
    },
  ]);
  await queryInterface.bulkInsert(RUN_EVENT_TABLE, [
    {
      id: '019f70a0-0000-7000-8000-000000000003',
      run_id: runId,
      sequence: 1,
      type: 'run.created',
      dedupe_key: 'create',
      actor_type: 'compatibility',
      attempt_id: attemptId,
      payload: JSON.stringify({ source: 'migration-test' }),
      created_at_ms: 1_750_000_000_000,
    },
  ]);

  await assert.rejects(
    queryInterface.bulkInsert(RUN_TABLE, [
      {
        ...baseRun,
        id: '019f70a0-0000-7000-8000-000000000004',
        idempotency_key: 'manual-request-1',
      },
    ]),
  );
  await assert.rejects(
    queryInterface.bulkInsert(RUN_TABLE, [
      {
        ...baseRun,
        id: '019f70a0-0000-7000-8000-000000000007',
        version: -1,
        idempotency_key: 'manual-request-2',
      },
    ]),
  );
  await assert.rejects(
    queryInterface.bulkInsert(RUN_ATTEMPT_TABLE, [
      {
        id: '019f70a0-0000-7000-8000-000000000005',
        run_id: runId,
        attempt: 1,
        status: 'claimed',
        executor_type: 'legacy_local',
        callback_sequence: 0,
        created_at_ms: 1_750_000_000_000,
      },
    ]),
  );
  await assert.rejects(
    queryInterface.bulkInsert(RUN_EVENT_TABLE, [
      {
        id: '019f70a0-0000-7000-8000-000000000006',
        run_id: runId,
        sequence: 2,
        type: 'run.created',
        dedupe_key: 'create',
        actor_type: 'compatibility',
        payload: '{}',
        created_at_ms: 1_750_000_000_001,
      },
    ]),
  );

  assert.equal(await migrationModel.count(), 1);
});

test('creates a bounded completion receipt journal linked to Run Attempts', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  await runMigrations({
    database,
    migrationModel,
    migrations: [runSchemaMigration, completionReceiptJournalMigration],
    logger: { info() {} },
  });

  const columns = await queryInterface.describeTable(
    COMPLETION_RECEIPT_JOURNAL_TABLE,
  );
  for (const column of [
    'attempt_id',
    'run_id',
    'state',
    'quarantine_ref',
    'purge_after_ms',
    'registered_at_ms',
    'updated_at_ms',
  ]) {
    assert.ok(columns[column], `missing journal.${column}`);
  }
  const indexes = new Set(
    (await queryInterface.showIndex(COMPLETION_RECEIPT_JOURNAL_TABLE)).map(
      (index) => index.name,
    ),
  );
  assert.ok(indexes.has(COMPLETION_RECEIPT_JOURNAL_SCAN_INDEX));
  assert.ok(indexes.has(COMPLETION_RECEIPT_JOURNAL_PURGE_INDEX));

  await assert.rejects(
    queryInterface.bulkInsert(COMPLETION_RECEIPT_JOURNAL_TABLE, [
      {
        attempt_id: '019f70a0-0000-7000-8000-000000000099',
        run_id: '019f70a0-0000-7000-8000-000000000098',
        state: 'pending',
        registered_at_ms: 1,
        updated_at_ms: 1,
      },
    ]),
    /FOREIGN KEY|constraint/i,
  );
});

test('creates the durable fenced Worker registry and bounded lookup indexes', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  await runMigrations({
    database,
    migrationModel,
    migrations: [workerRegistryMigration],
    logger: { info() {} },
  });

  const columns = await queryInterface.describeTable(WORKER_REGISTRY_TABLE);
  for (const column of [
    'id',
    'session_id',
    'generation',
    'status',
    'version',
    'capabilities_json',
    'capabilities_hash',
    'max_concurrent_runs',
    'available_slots',
    'registered_at_ms',
    'last_heartbeat_at_ms',
    'lease_expires_at_ms',
    'updated_at_ms',
  ]) {
    assert.ok(columns[column], `missing workers.${column}`);
  }
  const indexes = new Set(
    (await queryInterface.showIndex(WORKER_REGISTRY_TABLE)).map(
      (index) => index.name,
    ),
  );
  assert.ok(indexes.has(WORKER_REGISTRY_LEASE_INDEX));
  assert.ok(indexes.has(WORKER_REGISTRY_CAPACITY_INDEX));

  await assert.rejects(
    queryInterface.bulkInsert(WORKER_REGISTRY_TABLE, [
      {
        id: 'worker-invalid',
        session_id: '019f7500-0000-7000-8000-000000000001',
        generation: 0,
        status: 'online',
        version: 0,
        capabilities_json: '{}',
        capabilities_hash: 'a'.repeat(64),
        max_concurrent_runs: 1,
        available_slots: 1,
        registered_at_ms: 1,
        last_heartbeat_at_ms: 1,
        lease_expires_at_ms: 2,
        updated_at_ms: 1,
      },
    ]),
    /constraint/i,
  );
});

test('creates attempt-scoped Run dispatch leases with Worker fencing indexes', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  await runMigrations({
    database,
    migrationModel,
    migrations: [
      runSchemaMigration,
      workerRegistryMigration,
      runDispatchLeaseMigration,
    ],
    logger: { info() {} },
  });

  const columns = await queryInterface.describeTable(RUN_DISPATCH_LEASE_TABLE);
  for (const column of [
    'attempt_id',
    'run_id',
    'status',
    'version',
    'lease_generation',
    'worker_id',
    'worker_session_id',
    'worker_generation',
    'lease_token',
    'expires_at_ms',
    'completed_at_ms',
  ]) {
    assert.ok(columns[column], `missing ${RUN_DISPATCH_LEASE_TABLE}.${column}`);
  }
  const indexes = await queryInterface.showIndex(RUN_DISPATCH_LEASE_TABLE);
  assert.ok(
    indexes.some((index) => index.name === RUN_DISPATCH_LEASE_EXPIRY_INDEX),
  );
  assert.ok(
    indexes.some((index) => index.name === RUN_DISPATCH_LEASE_WORKER_INDEX),
  );
  assert.ok(
    indexes.some(
      (index) => index.name === RUN_DISPATCH_LEASE_TOKEN_INDEX && index.unique,
    ),
  );
  await assert.rejects(
    database.query(
      `INSERT INTO ${RUN_DISPATCH_LEASE_TABLE} (
        attempt_id, run_id, status, version, lease_generation,
        worker_id, worker_session_id, worker_generation, lease_token,
        acquired_at_ms, renewed_at_ms, expires_at_ms, updated_at_ms
      ) VALUES (
        'missing-attempt', 'missing-run', 'leased', -1, 0,
        'missing-worker', '019f7800-0000-7000-8000-000000000001', 0,
        'lease_token_abcdefghijklmnopqrstuvwxyz0123456789', 1, 1, 2, 1
      )`,
    ),
  );
});

test('adds stable Run references to legacy RunningInstance rows', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  await queryInterface.createTable(RUNNING_INSTANCE_TABLE, {
    id: {
      type: DataTypes.INTEGER,
      primaryKey: true,
      autoIncrement: true,
    },
    cron_id: { type: DataTypes.INTEGER, allowNull: false },
    pid: { type: DataTypes.INTEGER, allowNull: true },
    log_path: { type: DataTypes.STRING, allowNull: true },
    started_at: { type: DataTypes.INTEGER, allowNull: false },
    finished_at: { type: DataTypes.INTEGER, allowNull: true },
    status: { type: DataTypes.INTEGER, allowNull: false },
    exit_code: { type: DataTypes.INTEGER, allowNull: true },
  });
  await queryInterface.bulkInsert(RUNNING_INSTANCE_TABLE, [
    { cron_id: 7, started_at: 1_750_000_000, status: 1 },
  ]);

  const options = {
    database,
    migrationModel,
    migrations: [runningInstanceRunReferenceMigration],
    logger: { info() {} },
  };
  await runMigrations(options);
  await runMigrations(options);

  const columns = await queryInterface.describeTable(RUNNING_INSTANCE_TABLE);
  assert.ok(columns.run_id);
  assert.ok(columns.attempt_id);
  assert.equal(columns.run_id.allowNull, true);
  assert.equal(columns.attempt_id.allowNull, true);
  const indexNames = new Set(
    (await queryInterface.showIndex(RUNNING_INSTANCE_TABLE)).map(
      (index) => index.name,
    ),
  );
  assert.ok(indexNames.has(RUNNING_INSTANCE_RUN_INDEX));
  assert.ok(indexNames.has(RUNNING_INSTANCE_ATTEMPT_INDEX));

  const fixture = database.define(
    'RunningInstanceMigrationFixture',
    {
      id: { type: DataTypes.INTEGER, primaryKey: true },
      cron_id: DataTypes.INTEGER,
      run_id: DataTypes.STRING(36),
      attempt_id: DataTypes.STRING(36),
      started_at: DataTypes.INTEGER,
      status: DataTypes.INTEGER,
    },
    { tableName: RUNNING_INSTANCE_TABLE, timestamps: false },
  );
  const legacyRow = await fixture.findByPk(1, { raw: true });
  assert.equal(legacyRow.run_id, null);
  assert.equal(legacyRow.attempt_id, null);

  const runId = '019f7110-0000-7000-8000-000000000001';
  const attemptId = '019f7110-0000-7000-8000-000000000002';
  await queryInterface.bulkInsert(RUNNING_INSTANCE_TABLE, [
    {
      cron_id: 7,
      run_id: runId,
      attempt_id: attemptId,
      started_at: 1_750_000_001,
      status: 0,
    },
  ]);
  await assert.rejects(
    queryInterface.bulkInsert(RUNNING_INSTANCE_TABLE, [
      {
        cron_id: 7,
        run_id: runId,
        attempt_id: attemptId,
        started_at: 1_750_000_002,
        status: 0,
      },
    ]),
  );
  assert.equal(await migrationModel.count(), 1);
});

test('adds durable cancellation requests without rewriting legacy Run rows', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  await runMigrations({
    database,
    migrationModel,
    migrations: [runSchemaMigration],
    logger: { info() {} },
  });

  const runId = '019f7110-0000-7000-8000-000000000004';
  await queryInterface.bulkInsert(RUN_TABLE, [
    {
      id: runId,
      project_id: 'default',
      task_id: 'legacy-cron:7',
      task_revision: 'revision-7',
      trigger_type: 'manual',
      execution_origin: 'manual',
      execution_owner: 'runtime',
      status: 'running',
      version: 3,
      event_sequence: 3,
      priority: 0,
      created_at_ms: 1_750_000_000_000,
      started_at_ms: 1_750_000_000_010,
    },
  ]);

  const options = {
    database,
    migrationModel,
    migrations: [runSchemaMigration, runCancellationRequestMigration],
    logger: { info() {} },
  };
  await runMigrations(options);
  await runMigrations(options);

  const columns = await queryInterface.describeTable(RUN_TABLE);
  assert.ok(columns.cancel_requested_at_ms);
  assert.ok(columns.cancel_reason);
  assert.equal(columns.cancel_requested_at_ms.allowNull, true);
  assert.equal(columns.cancel_reason.allowNull, true);
  const indexNames = new Set(
    (await queryInterface.showIndex(RUN_TABLE)).map((index) => index.name),
  );
  assert.ok(indexNames.has(RUN_CANCELLATION_REQUEST_INDEX));

  const [legacyRow] = await database.query(
    `SELECT cancel_requested_at_ms, cancel_reason FROM ${RUN_TABLE} WHERE id = :runId`,
    {
      replacements: { runId },
      type: QueryTypes.SELECT,
    },
  );
  assert.equal(legacyRow.cancel_requested_at_ms, null);
  assert.equal(legacyRow.cancel_reason, null);
  assert.equal(await migrationModel.count(), 2);
});

test('adds nullable Attempt deadlines without rewriting existing attempts', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  await runMigrations({
    database,
    migrationModel,
    migrations: [runSchemaMigration],
    logger: { info() {} },
  });

  const runId = '019f7110-0000-7000-8000-000000000014';
  const attemptId = '019f7110-0000-7000-8000-000000000015';
  await queryInterface.bulkInsert(RUN_TABLE, [
    {
      id: runId,
      project_id: 'default',
      task_id: 'legacy-cron:8',
      task_revision: 'revision-8',
      trigger_type: 'manual',
      execution_origin: 'manual',
      execution_owner: 'runtime',
      status: 'dispatching',
      version: 2,
      event_sequence: 2,
      priority: 0,
      created_at_ms: 1_750_000_000_000,
    },
  ]);
  await queryInterface.bulkInsert(RUN_ATTEMPT_TABLE, [
    {
      id: attemptId,
      run_id: runId,
      attempt: 1,
      status: 'starting',
      executor_type: 'local_process',
      callback_sequence: 0,
      created_at_ms: 1_750_000_000_000,
    },
  ]);

  const options = {
    database,
    migrationModel,
    migrations: [runSchemaMigration, runAttemptDeadlineMigration],
    logger: { info() {} },
  };
  await runMigrations(options);
  await runMigrations(options);

  const columns = await queryInterface.describeTable(RUN_ATTEMPT_TABLE);
  assert.ok(columns.deadline_at_ms);
  assert.equal(columns.deadline_at_ms.allowNull, true);
  const indexNames = new Set(
    (await queryInterface.showIndex(RUN_ATTEMPT_TABLE)).map(
      (index) => index.name,
    ),
  );
  assert.ok(indexNames.has(RUN_ATTEMPT_DEADLINE_INDEX));
  const [legacyAttempt] = await database.query(
    `SELECT deadline_at_ms FROM ${RUN_ATTEMPT_TABLE} WHERE id = :attemptId`,
    {
      replacements: { attemptId },
      type: QueryTypes.SELECT,
    },
  );
  assert.equal(legacyAttempt.deadline_at_ms, null);
  assert.equal(await migrationModel.count(), 2);
});

test('creates a fenced cancellation dispatch lease bound to a Run Attempt', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();
  await runMigrations({
    database,
    migrationModel,
    migrations: [
      runSchemaMigration,
      runCancellationRequestMigration,
      runCancellationDispatchMigration,
    ],
    logger: { info() {} },
  });

  const columns = await queryInterface.describeTable(
    RUN_CANCELLATION_DISPATCH_TABLE,
  );
  for (const column of [
    'run_id',
    'attempt_id',
    'status',
    'version',
    'dispatch_count',
    'next_attempt_at_ms',
    'lease_owner',
    'lease_token',
    'lease_expires_at_ms',
    'last_result',
    'last_dispatched_at_ms',
    'created_at_ms',
    'updated_at_ms',
  ]) {
    assert.ok(columns[column], `missing cancellation dispatch.${column}`);
  }
  const indexNames = new Set(
    (await queryInterface.showIndex(RUN_CANCELLATION_DISPATCH_TABLE)).map(
      (index) => index.name,
    ),
  );
  assert.ok(indexNames.has(RUN_CANCELLATION_DISPATCH_DUE_INDEX));
  assert.ok(indexNames.has(RUN_CANCELLATION_DISPATCH_LEASE_INDEX));

  const runId = '019f7110-0000-7000-8000-000000000011';
  const attemptId = '019f7110-0000-7000-8000-000000000012';
  await queryInterface.bulkInsert(RUN_TABLE, [
    {
      id: runId,
      project_id: 'default',
      task_id: 'legacy-cron:8',
      task_revision: 'revision-8',
      trigger_type: 'manual',
      execution_origin: 'manual',
      execution_owner: 'runtime',
      status: 'running',
      version: 2,
      event_sequence: 2,
      priority: 0,
      created_at_ms: 1_750_000_000_000,
      cancel_requested_at_ms: 1_750_000_000_100,
      cancel_reason: 'user',
    },
  ]);
  await queryInterface.bulkInsert(RUN_ATTEMPT_TABLE, [
    {
      id: attemptId,
      run_id: runId,
      attempt: 1,
      status: 'running',
      executor_type: 'local_process',
      callback_sequence: 0,
      created_at_ms: 1_750_000_000_010,
    },
  ]);
  await queryInterface.bulkInsert(RUN_CANCELLATION_DISPATCH_TABLE, [
    {
      run_id: runId,
      attempt_id: attemptId,
      status: 'pending',
      version: 0,
      dispatch_count: 0,
      next_attempt_at_ms: 1_750_000_000_100,
      created_at_ms: 1_750_000_000_100,
      updated_at_ms: 1_750_000_000_100,
    },
  ]);
  await assert.rejects(
    queryInterface.bulkInsert(RUN_CANCELLATION_DISPATCH_TABLE, [
      {
        run_id: 'missing-run',
        attempt_id: attemptId,
        status: 'pending',
        version: 0,
        dispatch_count: 0,
        created_at_ms: 1_750_000_000_100,
        updated_at_ms: 1_750_000_000_100,
      },
    ]),
  );
  await assert.rejects(
    queryInterface.bulkUpdate(
      RUN_CANCELLATION_DISPATCH_TABLE,
      { version: -1 },
      { run_id: runId },
    ),
  );
});

test('runs the registered migration chain against a legacy database fixture', async () => {
  const { database, migrationModel } = await createDatabase();
  const queryInterface = database.getQueryInterface();

  for (const table of ['CrontabViews', 'Subscriptions', 'Crontabs', 'Envs']) {
    await queryInterface.createTable(table, {
      id: { type: DataTypes.INTEGER, primaryKey: true },
    });
  }

  const options = {
    database,
    migrationModel,
    logger: { info() {} },
  };
  await runMigrations(options);
  await runMigrations(options);

  const tables = new Set(await queryInterface.showAllTables());
  assert.ok(tables.has(RUN_TABLE));
  assert.ok(tables.has(RUN_ATTEMPT_TABLE));
  assert.ok(tables.has(RUN_EVENT_TABLE));
  assert.ok(tables.has(RUN_CANCELLATION_DISPATCH_TABLE));
  assert.equal(await migrationModel.count(), migrations.length);
});