memind/services/orchestrator/executor-job-store.test.mjs

import assert from 'node:assert/strict';
import test from 'node:test';
import {
  createExecutorJobPersistence,
  createPostgresExecutorJobStore,
  executorJobStoreInternals,
} from './executor-job-store.mjs';

function record(overrides = {}) {
  return {
    version: 'executor-job-state-v1',
    jobId: 'job-1',
    idempotencyKey: 'idem-1',
    requestFingerprint: 'a'.repeat(64),
    status: 'blocked',
    createdAt: 1000,
    updatedAt: 1000,
    ...overrides,
  };
}

function transactionalPool(query) {
  return {
    query,
    async connect() {
      return {
        query,
        release() {},
      };
    },
  };
}

test('PostgreSQL Executor Job Store creates an isolated schema and job table', async () => {
  const queries = [];
  const store = createPostgresExecutorJobStore({
    schema: 'memind_orchestrator_test',
    pool: {
      async query(sql, params = []) {
        queries.push({ sql, params });
        return { rows: [] };
      },
    },
  });
  await store.ensureSchema();
  assert.equal(store.kind, 'postgres');
  assert.equal(store.durable, true);
  assert.match(queries[0].sql, /CREATE SCHEMA IF NOT EXISTS "memind_orchestrator_test"/);
  assert.match(queries[1].sql, /"memind_orchestrator_test"\."executor_jobs"/);
  assert.match(queries[1].sql, /idempotency_key VARCHAR\(200\) NOT NULL UNIQUE/);
  assert.match(queries[2].sql, /executor_jobs_updated_at_idx/);
  assert.match(queries[3].sql, /"memind_orchestrator_test"\."executor_job_events"/);
  assert.match(queries[3].sql, /UNIQUE \(job_id, sequence\)/);
  assert.match(queries[5].sql, /executor_job_snapshot_imported/);
  assert.match(queries[6].sql, /executor_workers/);
  assert.match(queries[7].sql, /executor_workers_last_seen_idx/);
});

test('PostgreSQL Executor Job Store atomically creates and reloads idempotent state', async () => {
  const stored = record();
  const blockedEvent = {
    version: 'executor-job-event-v1',
    eventId: 'event-1',
    jobId: 'job-1',
    sequence: 1,
    type: 'executor_job_blocked',
    timestamp: 1000,
    data: { status: 'blocked' },
  };
  const calls = [];
  const store = createPostgresExecutorJobStore({
    pool: transactionalPool(async (sql, params = []) => {
      calls.push({ sql, params });
      if (sql.includes('INSERT INTO') && sql.includes('executor_jobs')) {
        return { rows: [{ state_json: stored }] };
      }
      if (sql.includes('INSERT INTO') && sql.includes('executor_job_events')) {
        return { rows: [] };
      }
      if (sql.includes('SELECT job_id') && sql.includes('FOR UPDATE')) {
        return { rows: [{ job_id: 'job-1' }] };
      }
      if (sql.includes('UPDATE')) {
        return { rows: [{ state_json: { ...stored, status: 'cancelled' } }] };
      }
      if (sql.includes('MAX(sequence)')) {
        return { rows: [{ next_sequence: 2 }] };
      }
      if (sql.includes('SELECT event_json')) {
        return { rows: [{ event_json: blockedEvent }] };
      }
      if (sql.includes('WHERE job_id =')) {
        return { rows: [{ state_json: JSON.stringify(stored) }] };
      }
      if (sql.includes('WHERE idempotency_key =')) {
        return { rows: [{ state_json: stored }] };
      }
      return { rows: [] };
    }),
  });

  const created = await store.createIfAbsent(stored, { initialEvent: blockedEvent });
  assert.equal(created.created, true);
  assert.deepEqual(created.record, stored);
  assert.deepEqual(await store.getById('job-1'), stored);
  assert.deepEqual(await store.getByIdempotencyKey('idem-1'), stored);
  assert.equal(
    (await store.update(
      { ...stored, status: 'cancelled' },
      { event: { ...blockedEvent, eventId: 'event-2', type: 'executor_job_cancelled' } },
    )).status,
    'cancelled',
  );
  assert.deepEqual(await store.listEvents('job-1'), [blockedEvent]);
  const jobInsert = calls.find((call) => (
    call.sql.includes('INSERT INTO') && call.sql.includes('executor_jobs')
  ));
  assert.equal(JSON.parse(jobInsert.params[3]).jobId, 'job-1');
  assert.equal(calls.filter((call) => call.sql.trim() === 'BEGIN').length, 2);
  assert.equal(calls.filter((call) => call.sql.trim() === 'COMMIT').length, 2);
});

test('PostgreSQL Executor Job Store returns the existing idempotent record after conflict', async () => {
  const stored = record();
  const store = createPostgresExecutorJobStore({
    pool: transactionalPool(async (sql) => {
      if (sql.includes('INSERT INTO')) return { rows: [] };
      if (sql.includes('WHERE idempotency_key =')) {
        return { rows: [{ state_json: stored }] };
      }
      return { rows: [] };
    }),
  });
  assert.deepEqual(
    await store.createIfAbsent(stored),
    { created: false, record: stored },
  );
});

test('PostgreSQL Executor Job Store deletes a job and cascades its events', async () => {
  const calls = [];
  const store = createPostgresExecutorJobStore({
    pool: {
      async query(sql, params) {
        calls.push({ sql, params });
        return sql.includes('DELETE FROM')
          ? { rows: [{ job_id: 'job-1' }] }
          : { rows: [] };
      },
    },
  });
  assert.equal(await store.deleteById('job-1'), true);
  assert.match(calls[0].sql, /DELETE FROM .*executor_jobs/);
  assert.deepEqual(calls[0].params, ['job-1']);
});

test('PostgreSQL Executor Job Store lists bounded terminal retention candidates', async () => {
  const calls = [];
  const store = createPostgresExecutorJobStore({
    pool: {
      async query(sql, params) {
        calls.push({ sql, params });
        return {
          rows: [{
            job_id: 'job-1',
            workflow_run_id: 'run-1',
            completed_at: '1000',
          }],
        };
      },
    },
  });
  assert.deepEqual(
    await store.listTerminalBefore({ before: 2000, limit: 10 }),
    [{ jobId: 'job-1', workflowRunId: 'run-1', completedAt: 1000 }],
  );
  assert.match(calls[0].sql, /workflowRunId/);
  assert.deepEqual(calls[0].params, [
    ['succeeded', 'failed', 'cancelled', 'timed_out', 'blocked'],
    2000,
    10,
  ]);
});

test('PostgreSQL Executor Job Store rolls back job creation when its initial event fails', async () => {
  const calls = [];
  const eventError = new Error('event insert unavailable');
  const store = createPostgresExecutorJobStore({
    pool: transactionalPool(async (sql) => {
      calls.push(sql.trim());
      if (sql.includes('INSERT INTO') && sql.includes('executor_jobs')) {
        return { rows: [{ state_json: record() }] };
      }
      if (sql.includes('INSERT INTO') && sql.includes('executor_job_events')) {
        throw eventError;
      }
      return { rows: [] };
    }),
  });

  await assert.rejects(
    store.createIfAbsent(record(), {
      initialEvent: {
        eventId: 'event-failing',
        jobId: 'job-1',
        timestamp: 1000,
      },
    }),
    eventError,
  );
  assert.ok(calls.includes('BEGIN'));
  assert.ok(calls.includes('ROLLBACK'));
  assert.ok(!calls.includes('COMMIT'));
});

test('Executor Job persistence supports explicit non-durable memory mode', async () => {
  const persistence = await createExecutorJobPersistence({ mode: 'memory' });
  assert.equal(persistence.kind, 'memory');
  assert.equal(persistence.durable, false);
  assert.equal(persistence.store.durable, false);
  assert.equal(persistence.store.kind, 'memory');
  assert.equal(await persistence.probe(), true);
  await persistence.close();
});

test('Executor Job persistence initializes and closes its PostgreSQL pool', async () => {
  const queries = [];
  let closed = false;
  const pool = {
    async query(sql) {
      queries.push(sql);
      return { rows: [] };
    },
    async end() {
      closed = true;
    },
  };
  const persistence = await createExecutorJobPersistence({
    mode: 'postgres',
    connectionString: 'postgresql://orchestrator:test@postgres/orchestrator',
    schema: 'memind_orchestrator',
    poolFactory(config) {
      assert.equal(
        config.connectionString,
        'postgresql://orchestrator:test@postgres/orchestrator',
      );
      return pool;
    },
  });
  assert.equal(persistence.kind, 'postgres');
  assert.equal(persistence.durable, true);
  assert.equal(persistence.store.kind, 'postgres');
  assert.equal(await persistence.probe(), true);
  assert.match(queries.at(-1), /SELECT 1 AS ok/);
  await persistence.close();
  assert.equal(closed, true);
});

test('Executor Job persistence rejects missing PostgreSQL config and unsafe schema names', async () => {
  let poolCreated = false;
  await assert.rejects(
    createExecutorJobPersistence({ mode: 'postgres', connectionString: '' }),
    (error) => error.code === 'ORCHESTRATOR_DATABASE_URL_REQUIRED',
  );
  await assert.rejects(
    createExecutorJobPersistence({
      mode: 'postgres',
      connectionString: 'postgresql://orchestrator:test@postgres/orchestrator',
      schema: 'public; drop schema public',
      poolFactory() {
        poolCreated = true;
        return null;
      },
    }),
    /Invalid Executor Gateway PostgreSQL schema/,
  );
  assert.equal(poolCreated, false);
  assert.throws(
    () => executorJobStoreInternals.normalizeSchema('public; drop schema public'),
    /Invalid Executor Gateway PostgreSQL schema/,
  );
});

test('Executor Job persistence closes its pool when schema initialization fails', async () => {
  let closed = false;
  const initializationError = new Error('schema unavailable');
  await assert.rejects(
    createExecutorJobPersistence({
      mode: 'postgres',
      connectionString: 'postgresql://orchestrator:test@postgres/orchestrator',
      poolFactory() {
        return {
          async query() {
            throw initializationError;
          },
          async end() {
            closed = true;
          },
        };
      },
    }),
    initializationError,
  );
  assert.equal(closed, true);
});