checkpointer.ts

/**
 * Postgres-backed LangGraph checkpointer (R1, R3, R4).
 *
 * The checkpointer is what makes agent state durable: every super-step of the
 * graph is written to Postgres, keyed by `thread_id`. That is what lets a run
 * survive a process restart and lets a human-in-the-loop interrupt be resumed
 * later — possibly on a different replica.
 *
 * Connection pooling (R4): we build an explicit `pg.Pool` whose `max` is sized
 * from `DATABASE_POOL_MAX`. Keep the sum of pool sizes across all your replicas
 * below the connection limit of your Render Postgres plan to avoid exhaustion.
 *
 * If `DATABASE_URL` is not set we fall back to an in-memory saver so the project
 * still runs on a fresh clone — but state will NOT survive a restart. Set
 * `DATABASE_URL` to get the durable behavior the requirements describe.
 */
import type { BaseCheckpointSaver } from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph";
import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";
import pg from "pg";
import { config } from "./config.js";

let checkpointerPromise: Promise<BaseCheckpointSaver> | undefined;
let pool: pg.Pool | undefined;

async function build(): Promise<BaseCheckpointSaver> {
  if (!config.databaseUrl) {
    console.warn(
      "[checkpointer] DATABASE_URL not set — using in-memory MemorySaver. " +
        "State will NOT persist across restarts. Set DATABASE_URL for durable checkpointing.",
    );
    return new MemorySaver();
  }

  pool = new pg.Pool({
    connectionString: config.databaseUrl,
    max: config.databasePoolMax,
    // Keep idle connections from piling up against the plan's connection limit.
    idleTimeoutMillis: 30_000,
    connectionTimeoutMillis: 10_000,
  });

  const checkpointer = new PostgresSaver(pool);
  // Creates the checkpoint tables / runs migrations. Safe to call repeatedly.
  await checkpointer.setup();
  console.info(`[checkpointer] PostgresSaver ready (pool max=${config.databasePoolMax}).`);
  return checkpointer;
}

/** Lazily construct (and memoize) the shared checkpointer. */
export function getCheckpointer(): Promise<BaseCheckpointSaver> {
  checkpointerPromise ??= build();
  return checkpointerPromise;
}

/** Close the underlying pool. Useful for graceful shutdown and tests. */
export async function closeCheckpointer(): Promise<void> {
  await pool?.end();
  pool = undefined;
  checkpointerPromise = undefined;
}