feat: persist blocked executor jobs

services/orchestrator/executor-job-store.mjs

@@ -0,0 +1,191 @@
+import pg from 'pg';
+import { createInMemoryExecutorJobStore } from './executor-gateway.mjs';
+
+const { Pool } = pg;
+const DEFAULT_EXECUTOR_SCHEMA = 'memind_orchestrator';
+const EXECUTOR_JOB_TABLE = 'executor_jobs';
+
+function normalizeMode(value) {
+  const mode = String(value ?? 'postgres').trim().toLowerCase();
+  if (mode === 'memory' || mode === 'postgres') return mode;
+  throw new Error(`Unsupported executor job store mode: ${value}`);
+}
+
+function normalizeSchema(value) {
+  const schema = String(value ?? DEFAULT_EXECUTOR_SCHEMA).trim();
+  if (!/^[a-z_][a-z0-9_]{0,62}$/i.test(schema)) {
+    throw new Error('Invalid Executor Gateway PostgreSQL schema');
+  }
+  return schema;
+}
+
+function parseState(value) {
+  if (value == null) return null;
+  if (typeof value === 'object') return structuredClone(value);
+  try {
+    return JSON.parse(String(value));
+  } catch {
+    return null;
+  }
+}
+
+function projectRow(row) {
+  return parseState(row?.state_json);
+}
+
+export function createPostgresExecutorJobStore({
+  pool,
+  schema = DEFAULT_EXECUTOR_SCHEMA,
+} = {}) {
+  if (!pool?.query) throw new Error('PostgreSQL Executor Job Store requires a pool');
+  const normalizedSchema = normalizeSchema(schema);
+  const table = `"${normalizedSchema}"."${EXECUTOR_JOB_TABLE}"`;
+
+  return {
+    kind: 'postgres',
+    durable: true,
+
+    async ensureSchema() {
+      await pool.query(`CREATE SCHEMA IF NOT EXISTS "${normalizedSchema}"`);
+      await pool.query(`
+        CREATE TABLE IF NOT EXISTS ${table} (
+          job_id VARCHAR(128) PRIMARY KEY,
+          idempotency_key VARCHAR(200) NOT NULL UNIQUE,
+          request_fingerprint CHAR(64) NOT NULL,
+          state_json JSONB NOT NULL,
+          created_at BIGINT NOT NULL,
+          updated_at BIGINT NOT NULL
+        )
+      `);
+      await pool.query(`
+        CREATE INDEX IF NOT EXISTS executor_jobs_updated_at_idx
+          ON ${table} (updated_at DESC)
+      `);
+    },
+
+    async getById(jobId) {
+      const result = await pool.query(
+        `SELECT state_json FROM ${table} WHERE job_id = $1 LIMIT 1`,
+        [String(jobId ?? '').trim()],
+      );
+      return projectRow(result.rows?.[0]);
+    },
+
+    async getByIdempotencyKey(key) {
+      const result = await pool.query(
+        `SELECT state_json FROM ${table} WHERE idempotency_key = $1 LIMIT 1`,
+        [String(key ?? '').trim()],
+      );
+      return projectRow(result.rows?.[0]);
+    },
+
+    async createIfAbsent(record) {
+      const inserted = await pool.query(
+        `INSERT INTO ${table}
+          (job_id, idempotency_key, request_fingerprint, state_json, created_at, updated_at)
+         VALUES ($1, $2, $3, $4::jsonb, $5, $6)
+         ON CONFLICT DO NOTHING
+         RETURNING state_json`,
+        [
+          record.jobId,
+          record.idempotencyKey,
+          record.requestFingerprint,
+          JSON.stringify(record),
+          record.createdAt,
+          record.updatedAt,
+        ],
+      );
+      const created = projectRow(inserted.rows?.[0]);
+      if (created) return { created: true, record: created };
+
+      const existingResult = await pool.query(
+        `SELECT state_json FROM ${table} WHERE idempotency_key = $1 LIMIT 1`,
+        [record.idempotencyKey],
+      );
+      const existing = projectRow(existingResult.rows?.[0]);
+      if (existing) return { created: false, record: existing };
+
+      const error = new Error(`Executor job id already exists: ${record.jobId}`);
+      error.code = 'EXECUTOR_JOB_ID_CONFLICT';
+      error.status = 409;
+      throw error;
+    },
+
+    async update(record) {
+      const result = await pool.query(
+        `UPDATE ${table}
+            SET state_json = $2::jsonb,
+                updated_at = $3
+          WHERE job_id = $1
+          RETURNING state_json`,
+        [record.jobId, JSON.stringify(record), record.updatedAt],
+      );
+      return projectRow(result.rows?.[0]);
+    },
+  };
+}
+
+export async function createExecutorJobPersistence({
+  mode = process.env.MEMIND_ORCHESTRATOR_CHECKPOINT_MODE,
+  connectionString = process.env.MEMIND_ORCHESTRATOR_DATABASE_URL,
+  schema = process.env.MEMIND_ORCHESTRATOR_DATABASE_SCHEMA,
+  poolFactory = (config) => new Pool(config),
+} = {}) {
+  const normalizedMode = normalizeMode(mode);
+  if (normalizedMode === 'memory') {
+    const store = createInMemoryExecutorJobStore();
+    return {
+      kind: 'memory',
+      durable: false,
+      store,
+      async probe() {
+        return true;
+      },
+      async close() {},
+    };
+  }
+
+  const normalizedConnectionString = String(connectionString ?? '').trim();
+  if (!normalizedConnectionString) {
+    const error = new Error(
+      'MEMIND_ORCHESTRATOR_DATABASE_URL is required for PostgreSQL Executor Job Store',
+    );
+    error.code = 'ORCHESTRATOR_DATABASE_URL_REQUIRED';
+    throw error;
+  }
+
+  const normalizedSchema = normalizeSchema(schema);
+  const pool = poolFactory({ connectionString: normalizedConnectionString });
+  const store = createPostgresExecutorJobStore({ pool, schema: normalizedSchema });
+  try {
+    await store.ensureSchema();
+  } catch (error) {
+    try {
+      await pool.end?.();
+    } catch {
+      // Preserve the schema initialization failure.
+    }
+    throw error;
+  }
+  return {
+    kind: 'postgres',
+    durable: true,
+    store,
+    async probe() {
+      await pool.query('SELECT 1 AS ok');
+      return true;
+    },
+    async close() {
+      await pool.end?.();
+    },
+  };
+}
+
+export const executorJobStoreInternals = {
+  DEFAULT_EXECUTOR_SCHEMA,
+  EXECUTOR_JOB_TABLE,
+  normalizeMode,
+  normalizeSchema,
+  parseState,
+  projectRow,
+};