memind/services/orchestrator/checkpoint.mjs

import { MemorySaver } from '@langchain/langgraph';
import { PostgresSaver } from '@langchain/langgraph-checkpoint-postgres';

const DEFAULT_POSTGRES_SCHEMA = 'memind_orchestrator';

function normalizeMode(value) {
  const mode = String(value ?? 'postgres').trim().toLowerCase();
  if (mode === 'memory' || mode === 'postgres') return mode;
  throw new Error(`Unsupported orchestrator checkpoint mode: ${value}`);
}

function normalizeSchema(value) {
  const schema = String(value ?? DEFAULT_POSTGRES_SCHEMA).trim();
  if (!/^[a-z_][a-z0-9_]{0,62}$/i.test(schema)) {
    throw new Error('Invalid orchestrator PostgreSQL schema');
  }
  return schema;
}

export async function createOrchestratorCheckpoint({
  mode = process.env.MEMIND_ORCHESTRATOR_CHECKPOINT_MODE,
  connectionString = process.env.MEMIND_ORCHESTRATOR_DATABASE_URL,
  schema = process.env.MEMIND_ORCHESTRATOR_DATABASE_SCHEMA,
} = {}) {
  const normalizedMode = normalizeMode(mode);
  if (normalizedMode === 'memory') {
    return {
      kind: 'memory',
      durable: false,
      checkpointer: new MemorySaver(),
      async probe() {
        return true;
      },
      async close() {},
    };
  }

  const normalizedConnectionString = String(connectionString ?? '').trim();
  if (!normalizedConnectionString) {
    const error = new Error(
      'MEMIND_ORCHESTRATOR_DATABASE_URL is required when checkpoint mode is postgres',
    );
    error.code = 'ORCHESTRATOR_DATABASE_URL_REQUIRED';
    throw error;
  }

  const checkpointer = PostgresSaver.fromConnString(normalizedConnectionString, {
    schema: normalizeSchema(schema),
  });
  await checkpointer.setup();
  return {
    kind: 'postgres',
    durable: true,
    checkpointer,
    async probe() {
      await checkpointer.getTuple({
        configurable: {
          thread_id: '__orchestrator_readiness__',
          checkpoint_ns: '',
        },
      });
      return true;
    },
    async close() {
      await checkpointer.end();
    },
  };
}

export const orchestratorCheckpointInternals = {
  DEFAULT_POSTGRES_SCHEMA,
  normalizeMode,
  normalizeSchema,
};