src/proxy-capture/store-readonly.test.ts

import fs from "node:fs";
import path from "node:path";
import { DatabaseSync } from "node:sqlite";
import { gzipSync } from "node:zlib";
import { afterEach, describe, expect, it } from "vitest";
import { cleanupTempDirs, makeTempDir } from "../../test/helpers/temp-dir.js";
import { OPENCLAW_STATE_SCHEMA_VERSION } from "../state/openclaw-state-db-contract.js";
import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js";
import { createDebugProxyCaptureReader } from "./store-readonly.js";

const cleanupDirs: string[] = [];

afterEach(() => {
  cleanupTempDirs(cleanupDirs);
});

function makeStateEnv(prefix: string): NodeJS.ProcessEnv {
  return { OPENCLAW_STATE_DIR: makeTempDir(cleanupDirs, prefix) };
}

function createCaptureDatabase(env: NodeJS.ProcessEnv, schemaVersion: number): string {
  const databasePath = resolveOpenClawStateSqlitePath(env);
  fs.mkdirSync(path.dirname(databasePath), { recursive: true });
  const db = new DatabaseSync(databasePath);
  try {
    db.exec(`
      PRAGMA user_version = ${schemaVersion};
      CREATE TABLE workspace_attestations (
        workspace_key TEXT PRIMARY KEY,
        attested_at_ms INTEGER NOT NULL,
        updated_at_ms INTEGER NOT NULL
      ) STRICT;
      CREATE TABLE capture_sessions (
        id TEXT NOT NULL PRIMARY KEY,
        started_at INTEGER NOT NULL,
        ended_at INTEGER,
        mode TEXT NOT NULL,
        source_scope TEXT NOT NULL,
        source_process TEXT NOT NULL,
        proxy_url TEXT
      ) STRICT;
      CREATE TABLE capture_blobs (
        blob_id TEXT NOT NULL PRIMARY KEY,
        content_type TEXT,
        encoding TEXT NOT NULL,
        size_bytes INTEGER NOT NULL,
        sha256 TEXT NOT NULL,
        data BLOB NOT NULL,
        created_at INTEGER NOT NULL
      ) STRICT;
      CREATE TABLE capture_events (
        id INTEGER NOT NULL PRIMARY KEY,
        session_id TEXT NOT NULL,
        ts INTEGER NOT NULL,
        source_scope TEXT NOT NULL,
        source_process TEXT NOT NULL,
        protocol TEXT NOT NULL,
        direction TEXT NOT NULL,
        kind TEXT NOT NULL,
        flow_id TEXT NOT NULL,
        method TEXT,
        host TEXT,
        path TEXT,
        status INTEGER,
        close_code INTEGER,
        content_type TEXT,
        headers_json TEXT,
        data_text TEXT,
        data_blob_id TEXT,
        data_sha256 TEXT,
        error_text TEXT,
        meta_json TEXT
      ) STRICT;
      CREATE INDEX capture_events_session_ts_idx ON capture_events(session_id, ts);
      CREATE INDEX capture_events_flow_idx ON capture_events(flow_id, ts);
      INSERT INTO workspace_attestations VALUES ('workspace-1', 100, 200);
      INSERT INTO capture_sessions VALUES (
        'session-1', 1, NULL, 'qa', 'openclaw', 'candidate', NULL
      );
    `);
    db.prepare(
      `INSERT INTO capture_blobs (
         blob_id, content_type, encoding, size_bytes, sha256, data, created_at
       ) VALUES (?, ?, ?, ?, ?, ?, ?)`,
    ).run(
      "blob-1",
      "application/json",
      "gzip",
      11,
      "fixture-sha",
      gzipSync(Buffer.from('{"ok":true}')),
      2,
    );
    db.prepare(
      `INSERT INTO capture_events (
         id, session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id,
         method, host, path, status, content_type, data_blob_id
       ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
    ).run(
      1,
      "session-1",
      3,
      "openclaw",
      "candidate",
      "https",
      "outbound",
      "request",
      "flow-1",
      "POST",
      "slack.com",
      "/api/chat.postMessage",
      200,
      "application/json",
      "blob-1",
    );
  } finally {
    db.close();
  }
  return databasePath;
}

function inspectDatabase(databasePath: string) {
  const db = new DatabaseSync(databasePath, { readOnly: true });
  try {
    return {
      fingerprint: JSON.stringify(
        db
          .prepare(
            `SELECT type, name, tbl_name AS tableName, sql
             FROM sqlite_schema
             WHERE name NOT LIKE 'sqlite_%'
             ORDER BY type, name`,
          )
          .all(),
      ),
      userVersion: db.prepare("PRAGMA user_version").get(),
      workspaceAttestations: db
        .prepare("SELECT * FROM workspace_attestations ORDER BY workspace_key")
        .all(),
    };
  } finally {
    db.close();
  }
}

describe("createDebugProxyCaptureReader", () => {
  it.each([3, 4, 5, 6])(
    "reads schema %s without changing its version, schema, or workspace attestations",
    (schemaVersion) => {
      const env = makeStateEnv(`openclaw-proxy-reader-v${schemaVersion}-`);
      const databasePath = createCaptureDatabase(env, schemaVersion);
      const before = inspectDatabase(databasePath);
      const reader = createDebugProxyCaptureReader({ env });

      expect(reader.getSessionEvents("session-1")).toEqual([
        expect.objectContaining({
          id: 1,
          sessionId: "session-1",
          flowId: "flow-1",
          host: "slack.com",
        }),
      ]);
      expect(reader.readBlob("blob-1")).toBe('{"ok":true}');
      expect(inspectDatabase(databasePath)).toEqual(before);
    },
  );

  it("returns empty results for a missing database without creating state artifacts", () => {
    const env = makeStateEnv("openclaw-proxy-reader-missing-");
    const reader = createDebugProxyCaptureReader({ env });
    const databasePath = resolveOpenClawStateSqlitePath(env);

    expect(reader.getSessionEvents("missing")).toEqual([]);
    expect(reader.readBlob("missing")).toBeNull();
    expect(fs.existsSync(databasePath)).toBe(false);
    expect(fs.existsSync(path.dirname(databasePath))).toBe(false);
  });

  it("observes a database created after the reader", () => {
    const env = makeStateEnv("openclaw-proxy-reader-late-db-");
    const reader = createDebugProxyCaptureReader({ env });
    expect(reader.getSessionEvents("session-1")).toEqual([]);

    createCaptureDatabase(env, 3);

    expect(reader.getSessionEvents("session-1")).toHaveLength(1);
    expect(reader.readBlob("blob-1")).toBe('{"ok":true}');
  });

  it("observes later committed WAL writes on subsequent calls", () => {
    const env = makeStateEnv("openclaw-proxy-reader-wal-");
    const databasePath = createCaptureDatabase(env, 3);
    const writer = new DatabaseSync(databasePath);
    const reader = createDebugProxyCaptureReader({ env });
    try {
      writer.exec("PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0;");
      expect(reader.getSessionEvents("session-1")).toHaveLength(1);

      writer
        .prepare(
          `INSERT INTO capture_events (
             id, session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id
           ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
        )
        .run(2, "session-1", 4, "openclaw", "candidate", "https", "inbound", "response", "flow-1");

      expect(reader.getSessionEvents("session-1").map((event) => event.id)).toEqual([2, 1]);
    } finally {
      writer.close();
    }
  });

  it("projects exact camelCase keys with timestamp, id, and limit ordering", () => {
    const env = makeStateEnv("openclaw-proxy-reader-ordering-");
    const databasePath = createCaptureDatabase(env, 3);
    const writer = new DatabaseSync(databasePath);
    try {
      const insert = writer.prepare(
        `INSERT INTO capture_events (
           id, session_id, ts, source_scope, source_process, protocol, direction, kind, flow_id,
           method, host, path, status, close_code, content_type, headers_json, data_text,
           data_blob_id, data_sha256, error_text, meta_json
         ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      );
      insert.run(
        2,
        "session-1",
        5,
        "openclaw",
        "candidate",
        "https",
        "outbound",
        "request",
        "flow-2",
        "POST",
        "slack.com",
        "/api/chat.postMessage",
        200,
        1000,
        "application/json",
        '{"authorization":"redacted"}',
        '{"text":"older"}',
        "blob-1",
        "sha-2",
        "error-2",
        '{"attempt":2}',
      );
      insert.run(
        3,
        "session-1",
        5,
        "openclaw",
        "candidate",
        "https",
        "inbound",
        "response",
        "flow-3",
        "GET",
        "slack.com",
        "/api/conversations.history",
        201,
        1001,
        "application/json",
        '{"accept":"application/json"}',
        '{"text":"newer"}',
        "blob-1",
        "sha-3",
        "error-3",
        '{"attempt":3}',
      );
      insert.run(
        4,
        "session-1",
        6,
        "openclaw",
        "candidate",
        "https",
        "inbound",
        "response",
        "flow-4",
        null,
        null,
        null,
        null,
        null,
        null,
        null,
        null,
        null,
        null,
        null,
        null,
      );
    } finally {
      writer.close();
    }

    const events = createDebugProxyCaptureReader({ env }).getSessionEvents("session-1", 2);

    expect(events.map((event) => event.id)).toEqual([4, 3]);
    expect(events[1]).toEqual({
      id: 3,
      sessionId: "session-1",
      ts: 5,
      sourceScope: "openclaw",
      sourceProcess: "candidate",
      protocol: "https",
      direction: "inbound",
      kind: "response",
      flowId: "flow-3",
      method: "GET",
      host: "slack.com",
      path: "/api/conversations.history",
      status: 201,
      closeCode: 1001,
      contentType: "application/json",
      headersJson: '{"accept":"application/json"}',
      dataText: '{"text":"newer"}',
      dataBlobId: "blob-1",
      dataSha256: "sha-3",
      errorText: "error-3",
      metaJson: '{"attempt":3}',
    });
  });

  it("propagates newer schema errors", () => {
    const env = makeStateEnv("openclaw-proxy-reader-newer-");
    const databasePath = resolveOpenClawStateSqlitePath(env);
    fs.mkdirSync(path.dirname(databasePath), { recursive: true });
    const db = new DatabaseSync(databasePath);
    db.exec(`PRAGMA user_version = ${OPENCLAW_STATE_SCHEMA_VERSION + 1};`);
    db.close();

    expect(() => createDebugProxyCaptureReader({ env }).getSessionEvents("session-1")).toThrow(
      `uses newer schema version ${OPENCLAW_STATE_SCHEMA_VERSION + 1}`,
    );
  });

  it("propagates database corruption", () => {
    const env = makeStateEnv("openclaw-proxy-reader-corrupt-");
    const databasePath = resolveOpenClawStateSqlitePath(env);
    fs.mkdirSync(path.dirname(databasePath), { recursive: true });
    fs.writeFileSync(databasePath, "not a sqlite database");

    expect(() => createDebugProxyCaptureReader({ env }).readBlob("blob-1")).toThrow();
  });

  it("propagates capture query failures", () => {
    const env = makeStateEnv("openclaw-proxy-reader-sql-");
    const databasePath = resolveOpenClawStateSqlitePath(env);
    fs.mkdirSync(path.dirname(databasePath), { recursive: true });
    const db = new DatabaseSync(databasePath);
    db.exec("PRAGMA user_version = 3;");
    db.close();

    expect(() => createDebugProxyCaptureReader({ env }).getSessionEvents("session-1")).toThrow(
      /no such table: capture_events/u,
    );
  });
});