boss/scripts/boss-state-store-maintenance.mjs

#!/usr/bin/env node
import { createHash } from "node:crypto";
import fs from "node:fs/promises";
import path from "node:path";
import process from "node:process";

const snapshotKey = process.env.BOSS_STATE_POSTGRES_KEY?.trim() || "default";
const defaultStateFile = process.env.BOSS_STATE_FILE || path.join(process.cwd(), "data", "boss-state.json");
const defaultBackupDir = process.env.BOSS_STATE_BACKUP_DIR || path.join(path.dirname(defaultStateFile), "backups");
const defaultSchemaFile = path.join(process.cwd(), "scripts", "postgres-state-schema.sql");
const postgresTable = "boss_state_snapshots";

function usage() {
  return [
    "Usage: node scripts/boss-state-store-maintenance.mjs <command> [options]",
    "",
    "Commands:",
    "  describe",
    "  validate-schema [--schema <file>]",
    "  backup-file --input <file> [--output <file>] [--dry-run]",
    "  export-file --input <file> --output <file> [--dry-run]",
    "  migrate-file-to-postgres --input <file> [--dry-run]",
    "  export-postgres-backup --output <file> [--dry-run]",
    "  restore-postgres-backup --input <file> [--dry-run]",
    "  rollback-postgres-to-file --output <file> [--dry-run]",
    "",
    "Environment:",
    "  BOSS_STATE_FILE, BOSS_STATE_STORE, BOSS_DATABASE_URL, BOSS_STATE_POSTGRES_KEY, BOSS_STATE_BACKUP_DIR",
  ].join("\n");
}

function parseArgs(argv) {
  const [command, ...items] = argv;
  const options = { command, dryRun: false };
  for (let index = 0; index < items.length; index += 1) {
    const item = items[index];
    if (item === "--dry-run") {
      options.dryRun = true;
      continue;
    }
    if (item === "--input") {
      options.input = items[index + 1];
      index += 1;
      continue;
    }
    if (item === "--output") {
      options.output = items[index + 1];
      index += 1;
      continue;
    }
    if (item === "--schema") {
      options.schema = items[index + 1];
      index += 1;
      continue;
    }
    throw new Error(`UNKNOWN_OPTION:${item}`);
  }
  return options;
}

function jsonOut(payload) {
  process.stdout.write(`${JSON.stringify(payload, null, 2)}\n`);
}

function timestampSegment() {
  return new Date().toISOString().replace(/[:.]/g, "-");
}

function sha256(text) {
  return createHash("sha256").update(text).digest("hex");
}

function postgresModeEnabled() {
  return process.env.BOSS_STATE_STORE?.trim().toLowerCase() === "postgres";
}

function postgresConfigured() {
  return Boolean(process.env.BOSS_DATABASE_URL?.trim());
}

function requirePostgresMode() {
  if (!postgresModeEnabled()) {
    throw new Error("BOSS_STATE_STORE_POSTGRES_REQUIRED");
  }
}

function requirePostgresDatabaseUrl() {
  if (!postgresConfigured()) {
    throw new Error("BOSS_DATABASE_URL_REQUIRED");
  }
}

function validateStateText(text, source) {
  const parsed = JSON.parse(text);
  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
    throw new Error(`STATE_JSON_INVALID:${source}`);
  }
  return parsed;
}

async function readStateText(filePath) {
  const text = await fs.readFile(filePath, "utf8");
  validateStateText(text, filePath);
  return text;
}

function validatePostgresSchemaText(text, source) {
  const compact = text.replace(/\s+/g, " ").toLowerCase();
  const required = [
    [/create table if not exists boss_state_snapshots/, "table"],
    [/snapshot_key\s+text\s+primary key/, "snapshot_key_primary_key"],
    [/state\s+jsonb\s+not null/, "state_jsonb"],
    [/created_at\s+timestamptz\s+not null\s+default now\(\)/, "created_at"],
    [/updated_at\s+timestamptz\s+not null\s+default now\(\)/, "updated_at"],
    [/create index if not exists boss_state_snapshots_updated_at_idx/, "updated_at_index"],
  ];
  const missing = required.filter(([pattern]) => !pattern.test(compact)).map(([, name]) => name);
  if (missing.length > 0) {
    throw new Error(`POSTGRES_SCHEMA_INVALID:${source}:${missing.join(",")}`);
  }
  return {
    ok: true,
    source,
    table: postgresTable,
    sha256: sha256(text),
  };
}

async function validatePostgresSchema(options) {
  const schema = path.resolve(options.schema || defaultSchemaFile);
  const text = await fs.readFile(schema, "utf8");
  return {
    action: "validate-schema",
    ...validatePostgresSchemaText(text, schema),
  };
}

async function ensurePostgresSchema(client) {
  await client.query(`
    CREATE TABLE IF NOT EXISTS boss_state_snapshots (
      snapshot_key TEXT PRIMARY KEY,
      state JSONB NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `);
  await client.query(`
    CREATE INDEX IF NOT EXISTS boss_state_snapshots_updated_at_idx
      ON boss_state_snapshots (updated_at DESC)
  `);
}

async function withPostgres(handler) {
  requirePostgresMode();
  requirePostgresDatabaseUrl();
  const connectionString = process.env.BOSS_DATABASE_URL.trim();
  const { Client } = await import("pg");
  const client = new Client({ connectionString });
  await client.connect();
  try {
    return await handler(client);
  } finally {
    await client.end();
  }
}

async function backupFile(options) {
  const source = path.resolve(options.input || defaultStateFile);
  const text = await readStateText(source);
  const output = path.resolve(options.output || path.join(defaultBackupDir, `boss-state-${timestampSegment()}.json`));
  if (!options.dryRun) {
    await fs.mkdir(path.dirname(output), { recursive: true });
    await fs.writeFile(output, text, "utf8");
  }
  return {
    ok: true,
    action: "backup-file",
    dryRun: options.dryRun,
    source,
    output,
    bytes: Buffer.byteLength(text),
  };
}

async function exportFile(options) {
  const source = path.resolve(options.input || defaultStateFile);
  if (!options.output) {
    throw new Error("OUTPUT_REQUIRED");
  }
  const text = await readStateText(source);
  const output = path.resolve(options.output);
  if (!options.dryRun) {
    await fs.mkdir(path.dirname(output), { recursive: true });
    await fs.writeFile(output, text, "utf8");
  }
  return {
    ok: true,
    action: "export-file",
    dryRun: options.dryRun,
    source,
    output,
    bytes: Buffer.byteLength(text),
  };
}

async function migrateFileToPostgres(options) {
  requirePostgresMode();
  requirePostgresDatabaseUrl();
  const source = path.resolve(options.input || defaultStateFile);
  const text = await readStateText(source);
  const schema = await validatePostgresSchema({ schema: options.schema });
  if (!options.dryRun) {
    await withPostgres(async (client) => {
      await ensurePostgresSchema(client);
      await client.query(
        `
          INSERT INTO boss_state_snapshots (snapshot_key, state, updated_at)
          VALUES ($1, $2::jsonb, now())
          ON CONFLICT (snapshot_key)
          DO UPDATE SET state = EXCLUDED.state, updated_at = now()
        `,
        [snapshotKey, text],
      );
    });
  }
  return {
    ok: true,
    action: "migrate-file-to-postgres",
    dryRun: options.dryRun,
    source,
    snapshotKey,
    postgresConfigured: postgresConfigured(),
    wouldConnect: !options.dryRun,
    schemaValid: schema.ok,
    schemaSha256: schema.sha256,
    bytes: Buffer.byteLength(text),
    stateSha256: sha256(text),
  };
}

async function rollbackPostgresToFile(options) {
  requirePostgresMode();
  requirePostgresDatabaseUrl();
  const output = path.resolve(options.output || defaultStateFile);
  if (options.dryRun) {
    return {
      ok: true,
      action: "rollback-postgres-to-file",
      dryRun: true,
      output,
      snapshotKey,
      postgresConfigured: postgresConfigured(),
      wouldConnect: false,
    };
  }
  const text = await withPostgres(async (client) => {
    await ensurePostgresSchema(client);
    const result = await client.query("SELECT state FROM boss_state_snapshots WHERE snapshot_key = $1", [snapshotKey]);
    const state = result.rows[0]?.state;
    if (!state) {
      throw new Error("BOSS_POSTGRES_STATE_NOT_FOUND");
    }
    return JSON.stringify(state, null, 2);
  });
  await fs.mkdir(path.dirname(output), { recursive: true });
  await fs.writeFile(output, text, "utf8");
  return {
    ok: true,
    action: "rollback-postgres-to-file",
    dryRun: false,
    output,
    snapshotKey,
    bytes: Buffer.byteLength(text),
    stateSha256: sha256(text),
  };
}

function normalizeBackupPayload(text, source) {
  const parsed = validateStateText(text, source);
  if (parsed.metadata?.format === "boss-state-postgres-backup/v1" && parsed.state) {
    return {
      metadata: parsed.metadata,
      state: parsed.state,
      stateText: JSON.stringify(parsed.state, null, 2),
      bundled: true,
    };
  }
  return {
    metadata: null,
    state: parsed,
    stateText: JSON.stringify(parsed, null, 2),
    bundled: false,
  };
}

async function exportPostgresBackup(options) {
  requirePostgresMode();
  const output = path.resolve(options.output || path.join(defaultBackupDir, `boss-postgres-state-${timestampSegment()}.json`));
  if (options.dryRun) {
    requirePostgresDatabaseUrl();
    return {
      ok: true,
      action: "export-postgres-backup",
      dryRun: true,
      output,
      snapshotKey,
      postgresConfigured: true,
      wouldConnect: false,
    };
  }
  const stateText = await withPostgres(async (client) => {
    await ensurePostgresSchema(client);
    const result = await client.query("SELECT state FROM boss_state_snapshots WHERE snapshot_key = $1", [snapshotKey]);
    const state = result.rows[0]?.state;
    if (!state) {
      throw new Error("BOSS_POSTGRES_STATE_NOT_FOUND");
    }
    return JSON.stringify(state, null, 2);
  });
  const state = validateStateText(stateText, `${postgresTable}:${snapshotKey}`);
  const backup = {
    metadata: {
      format: "boss-state-postgres-backup/v1",
      exportedAt: new Date().toISOString(),
      snapshotKey,
      table: postgresTable,
      stateSha256: sha256(stateText),
      stateBytes: Buffer.byteLength(stateText),
    },
    state,
  };
  const backupText = `${JSON.stringify(backup, null, 2)}\n`;
  await fs.mkdir(path.dirname(output), { recursive: true });
  await fs.writeFile(output, backupText, "utf8");
  return {
    ok: true,
    action: "export-postgres-backup",
    dryRun: false,
    output,
    snapshotKey,
    bytes: Buffer.byteLength(backupText),
    stateSha256: backup.metadata.stateSha256,
  };
}

async function restorePostgresBackup(options) {
  requirePostgresMode();
  requirePostgresDatabaseUrl();
  if (!options.input) {
    throw new Error("INPUT_REQUIRED");
  }
  const source = path.resolve(options.input);
  const text = await fs.readFile(source, "utf8");
  const backup = normalizeBackupPayload(text, source);
  validateStateText(backup.stateText, source);

  if (!options.dryRun) {
    await withPostgres(async (client) => {
      await ensurePostgresSchema(client);
      await client.query(
        `
          INSERT INTO boss_state_snapshots (snapshot_key, state, updated_at)
          VALUES ($1, $2::jsonb, now())
          ON CONFLICT (snapshot_key)
          DO UPDATE SET state = EXCLUDED.state, updated_at = now()
        `,
        [snapshotKey, backup.stateText],
      );
    });
  }

  return {
    ok: true,
    action: "restore-postgres-backup",
    dryRun: options.dryRun,
    source,
    snapshotKey,
    postgresConfigured: postgresConfigured(),
    wouldConnect: !options.dryRun,
    bundled: backup.bundled,
    bytes: Buffer.byteLength(backup.stateText),
    stateSha256: sha256(backup.stateText),
  };
}

async function main() {
  const options = parseArgs(process.argv.slice(2));
  switch (options.command) {
    case "describe":
      jsonOut({
        ok: true,
        action: "describe",
        mode: postgresModeEnabled() ? "postgres" : "file",
        stateFile: path.resolve(defaultStateFile),
        backupDir: path.resolve(defaultBackupDir),
        postgresConfigured: postgresConfigured(),
        postgresTable,
        snapshotKey,
      });
      return;
    case "validate-schema":
      jsonOut(await validatePostgresSchema(options));
      return;
    case "backup-file":
      jsonOut(await backupFile(options));
      return;
    case "export-file":
      jsonOut(await exportFile(options));
      return;
    case "migrate-file-to-postgres":
      jsonOut(await migrateFileToPostgres(options));
      return;
    case "export-postgres-backup":
      jsonOut(await exportPostgresBackup(options));
      return;
    case "restore-postgres-backup":
      jsonOut(await restorePostgresBackup(options));
      return;
    case "rollback-postgres-to-file":
      jsonOut(await rollbackPostgresToFile(options));
      return;
    default:
      throw new Error(`UNKNOWN_COMMAND\n${usage()}`);
  }
}

main().catch((error) => {
  const message = error instanceof Error ? error.message : String(error);
  process.stderr.write(`${message}\n`);
  process.exit(1);
});