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

Add files using upload-large-folder tool

34810d2 verified about 1 month ago

11.8 kB

import { gunzipSync } from "node:zlib";
import { normalizeNullableString as normalizeObservedValue } from "@openclaw/normalization-core/string-coerce";
import type { Compilable, InferResult } from "kysely";
import {
  compileSqliteQueryBindings,
  executeSqliteQuerySync,
  executeSqliteQueryTakeFirstSync,
  getNodeSqliteKysely,
} from "../infra/kysely-sync.js";
import { withExistingOpenClawStateDatabaseReadOnly } from "../state/openclaw-state-db-readonly.js";
import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
import type {
  CaptureObservedDimension,
  CaptureQueryPreset,
  CaptureQueryRow,
  CaptureSessionCoverageSummary,
} from "./types.js";

type DebugProxyCaptureDatabase = Pick<
  OpenClawStateKyselyDatabase,
  "capture_sessions" | "capture_events" | "capture_blobs"
>;
type NodeSqliteDatabase = Parameters<typeof getNodeSqliteKysely>[0];

export type DebugProxyCaptureReader = {
  getSessionEvents(sessionId: string, limit?: number): Array<Record<string, unknown>>;
  readBlob(blobId: string): string | null;
};

export function listDebugProxyCaptureSessions(db: NodeSqliteDatabase, limit = 50) {
  const kysely = getNodeSqliteKysely<DebugProxyCaptureDatabase>(db);
  // Preserve the grouped scan that selects sessions with equal start times.
  const sessions = kysely
    .selectFrom("capture_sessions")
    .select(["id", "started_at", "ended_at", "mode", "source_process", "proxy_url"])
    .groupBy("id")
    .orderBy("started_at", "desc")
    .limit(limit)
    .as("s");
  const query = kysely
    .selectFrom(sessions)
    .select([
      "s.id",
      "s.started_at as startedAt",
      "s.ended_at as endedAt",
      "s.mode",
      "s.source_process as sourceProcess",
      "s.proxy_url as proxyUrl",
    ])
    .select((eb) =>
      eb
        .selectFrom("capture_events as e")
        .select((event) => event.fn.count<number>("e.id").as("count"))
        .whereRef("e.session_id", "=", "s.id")
        .as("eventCount"),
    )
    .orderBy("s.started_at", "desc");
  const { compiled, bind } = compileSqliteQueryBindings(() => query);
  // Native reads retain the store's failure ownership without evicting its borrowed DB.
  return db /* sqlite-allow-raw -- Execute Kysely SQL with native failure ownership. */
    .prepare(compiled.sql)
    .all(...bind(undefined)) as InferResult<typeof query>; // SAFETY: Native columns follow this generated projection.
}

export function findDebugProxyCaptureBlobReference(
  db: NodeSqliteDatabase,
  blobId: string,
): string | null {
  const query = getNodeSqliteKysely<DebugProxyCaptureDatabase>(db)
    .selectFrom("capture_events")
    .select("data_blob_id as blobId")
    .where("data_blob_id", "=", blobId)
    .limit(1);
  const { compiled, bind } = compileSqliteQueryBindings(() => query);
  const row = db /* sqlite-allow-raw -- Execute Kysely SQL with native failure ownership. */
    .prepare(compiled.sql)
    .get(...bind(undefined)) as InferResult<typeof query>[number] | undefined; // SAFETY: Native lookup returns the selected nullable text column.
  return row?.blobId || null;
}

export function queryDebugProxyCapturePreset(
  db: NodeSqliteDatabase,
  preset: CaptureQueryPreset,
  sessionId?: string,
): CaptureQueryRow[] {
  const kysely = getNodeSqliteKysely<DebugProxyCaptureDatabase>(db);
  let events = kysely.selectFrom("capture_events");
  if (sessionId) {
    events = events.where("session_id", "=", sessionId);
  }
  const locations = events.select(["host", "path"]);
  const count = kysely.fn.countAll<number>();
  let query: Compilable<CaptureQueryRow>;
  // Aggregate in SQLite so diagnostics do not hydrate whole capture sessions.
  switch (preset) {
    case "double-sends":
      query = locations
        .select(["method", count.as("duplicateCount")])
        .where("kind", "=", "request")
        .groupBy(["host", "path", "method", "data_sha256"])
        .having(count, ">", 1)
        .orderBy("duplicateCount", "desc")
        .orderBy("host", "asc");
      break;
    case "retry-storms":
      query = locations
        .select(count.as("errorCount"))
        .where("kind", "=", "response")
        .where("status", ">=", 429)
        .groupBy(["host", "path"])
        .having(count, ">", 1)
        .orderBy("errorCount", "desc")
        .orderBy("host", "asc");
      break;
    case "cache-busting":
      query = locations
        .select(count.as("variantCount"))
        .where("kind", "=", "request")
        .where((eb) =>
          eb.or([
            eb("path", "like", "%?%"),
            eb("headers_json", "like", "%cache-control%"),
            eb("headers_json", "like", "%pragma%"),
          ]),
        )
        .groupBy(["host", "path"])
        .orderBy("variantCount", "desc")
        .orderBy("host", "asc");
      break;
    case "ws-duplicate-frames":
      query = locations
        .select(count.as("duplicateFrames"))
        .where("kind", "=", "ws-frame")
        .where("direction", "=", "outbound")
        .groupBy(["host", "path", "data_sha256"])
        .having(count, ">", 1)
        .orderBy("duplicateFrames", "desc")
        .orderBy("host", "asc");
      break;
    case "missing-ack":
      query = events
        .select(["flow_id as flowId", "host", "path", count.as("outboundFrames")])
        .where("kind", "=", "ws-frame")
        .where("direction", "=", "outbound")
        .where(
          "flow_id",
          "not in",
          events
            .select("flow_id")
            .where("kind", "=", "ws-frame")
            .where("direction", "=", "inbound"),
        )
        .groupBy(["flow_id", "host", "path"])
        .orderBy("outboundFrames", "desc");
      break;
    case "error-bursts":
      query = locations
        .select(count.as("errorCount"))
        .where("kind", "=", "error")
        .groupBy(["host", "path"])
        .orderBy("errorCount", "desc")
        .orderBy("host", "asc");
      break;
    default:
      return [];
  }
  const { compiled, bind } = compileSqliteQueryBindings(() => query);
  return db /* sqlite-allow-raw -- Execute Kysely SQL with native failure ownership. */
    .prepare(compiled.sql)
    .all(...bind(undefined)) as CaptureQueryRow[]; // SAFETY: Each typed preset selects only text, count, or nullable columns.
}

export function readDebugProxyCaptureSessionEvents(
  db: NodeSqliteDatabase,
  sessionId: string,
  limit = 500,
): Array<Record<string, unknown>> {
  return executeSqliteQuerySync(
    db,
    getNodeSqliteKysely<DebugProxyCaptureDatabase>(db)
      .selectFrom("capture_events")
      .select([
        "id",
        "session_id as sessionId",
        "ts",
        "source_scope as sourceScope",
        "source_process as sourceProcess",
        "protocol",
        "direction",
        "kind",
        "flow_id as flowId",
        "method",
        "host",
        "path",
        "status",
        "close_code as closeCode",
        "content_type as contentType",
        "headers_json as headersJson",
        "data_text as dataText",
        "data_blob_id as dataBlobId",
        "data_sha256 as dataSha256",
        "error_text as errorText",
        "meta_json as metaJson",
      ])
      .where("session_id", "=", sessionId)
      .orderBy("ts", "desc")
      .orderBy("id", "desc")
      .limit(limit),
  ).rows;
}

// Metadata is optional and user/tool supplied, so parse defensively for coverage
// summaries instead of assuming every event has valid JSON.
function parseMetaJson(metaJson: unknown): Record<string, unknown> | null {
  if (typeof metaJson !== "string" || metaJson.trim().length === 0) {
    return null;
  }
  try {
    const parsed = JSON.parse(metaJson) as unknown;
    // SAFETY: Parsed objects, including arrays, are read only through optional label keys.
    return parsed && typeof parsed === "object" ? (parsed as Record<string, unknown>) : null;
  } catch {
    return null;
  }
}

function sortObservedCounts(counts: Map<string, number>): CaptureObservedDimension[] {
  return [...counts.entries()]
    .map(([value, count]) => ({ value, count }))
    .toSorted((left, right) => right.count - left.count || left.value.localeCompare(right.value));
}

export function summarizeDebugProxyCaptureSessionCoverage(
  db: NodeSqliteDatabase,
  sessionId: string,
): CaptureSessionCoverageSummary {
  const { compiled, bind } = compileSqliteQueryBindings<string>((parameter) =>
    getNodeSqliteKysely<DebugProxyCaptureDatabase>(db)
      .selectFrom("capture_events")
      .select(["host", "meta_json as metaJson"])
      .where(
        "session_id",
        "=",
        parameter((value) => value),
      ),
  );
  // Native iteration keeps corruption from evicting the borrowed shared database.
  const rows = db /* sqlite-allow-raw -- Execute Kysely SQL with native failure ownership. */
    .prepare(compiled.sql)
    .iterate(...bind(sessionId));
  const providers = new Map<string, number>();
  const apis = new Map<string, number>();
  const models = new Map<string, number>();
  const hosts = new Map<string, number>();
  const localPeers = new Map<string, number>();
  let totalEvents = 0;
  let unlabeledEventCount = 0;
  try {
    for (const row of rows) {
      totalEvents += 1;
      const meta = parseMetaJson(row.metaJson);
      const provider = normalizeObservedValue(meta?.provider);
      const api = normalizeObservedValue(meta?.api);
      const model = normalizeObservedValue(meta?.model);
      const host = normalizeObservedValue(row.host);
      if (!provider && !api && !model) {
        unlabeledEventCount += 1;
      }
      if (provider) {
        providers.set(provider, (providers.get(provider) ?? 0) + 1);
      }
      if (api) {
        apis.set(api, (apis.get(api) ?? 0) + 1);
      }
      if (model) {
        models.set(model, (models.get(model) ?? 0) + 1);
      }
      if (host) {
        hosts.set(host, (hosts.get(host) ?? 0) + 1);
        // Local model/provider endpoints are useful to surface separately when
        // debugging why cloud-provider labels are absent.
        if (host.startsWith("127.0.0.1:") || host.startsWith("localhost:")) {
          localPeers.set(host, (localPeers.get(host) ?? 0) + 1);
        }
      }
    }
  } catch (error) {
    try {
      rows.return?.();
    } catch {
      // Iterator cleanup must not replace the original read failure.
    }
    throw error;
  }
  return {
    sessionId,
    totalEvents,
    unlabeledEventCount,
    providers: sortObservedCounts(providers),
    apis: sortObservedCounts(apis),
    models: sortObservedCounts(models),
    hosts: sortObservedCounts(hosts),
    localPeers: sortObservedCounts(localPeers),
  };
}

export function readDebugProxyCaptureBlob(db: NodeSqliteDatabase, blobId: string): string | null {
  const row = executeSqliteQueryTakeFirstSync(
    db,
    getNodeSqliteKysely<DebugProxyCaptureDatabase>(db)
      .selectFrom("capture_blobs")
      .select(["encoding", "data"])
      .where("blob_id", "=", blobId),
  );
  if (!row?.data) {
    return null;
  }
  const data = Buffer.from(row.data);
  return (row.encoding === "gzip" ? gunzipSync(data) : data).toString("utf8");
}

/** Read capture rows without joining or mutating the shared-state writer lifecycle. */
export function createDebugProxyCaptureReader(params: {
  env: NodeJS.ProcessEnv;
}): DebugProxyCaptureReader {
  return {
    getSessionEvents(sessionId, limit) {
      return (
        withExistingOpenClawStateDatabaseReadOnly(
          ({ db }) => readDebugProxyCaptureSessionEvents(db, sessionId, limit),
          { env: params.env },
        ) ?? []
      );
    },
    readBlob(blobId) {
      return (
        withExistingOpenClawStateDatabaseReadOnly(
          ({ db }) => readDebugProxyCaptureBlob(db, blobId),
          { env: params.env },
        ) ?? null
      );
    },
  };
}

	import { gunzipSync } from "node:zlib";
	import { normalizeNullableString as normalizeObservedValue } from "@openclaw/normalization-core/string-coerce";
	import type { Compilable, InferResult } from "kysely";
	import {
	compileSqliteQueryBindings,
	executeSqliteQuerySync,
	executeSqliteQueryTakeFirstSync,
	getNodeSqliteKysely,
	} from "../infra/kysely-sync.js";
	import { withExistingOpenClawStateDatabaseReadOnly } from "../state/openclaw-state-db-readonly.js";
	import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
	import type {
	CaptureObservedDimension,
	CaptureQueryPreset,
	CaptureQueryRow,
	CaptureSessionCoverageSummary,
	} from "./types.js";

	type DebugProxyCaptureDatabase = Pick<
	OpenClawStateKyselyDatabase,
	"capture_sessions" \| "capture_events" \| "capture_blobs"
	>;
	type NodeSqliteDatabase = Parameters<typeof getNodeSqliteKysely>[0];

	export type DebugProxyCaptureReader = {
	getSessionEvents(sessionId: string, limit?: number): Array<Record<string, unknown>>;
	readBlob(blobId: string): string \| null;
	};

	export function listDebugProxyCaptureSessions(db: NodeSqliteDatabase, limit = 50) {
	const kysely = getNodeSqliteKysely<DebugProxyCaptureDatabase>(db);
	// Preserve the grouped scan that selects sessions with equal start times.
	const sessions = kysely
	.selectFrom("capture_sessions")
	.select(["id", "started_at", "ended_at", "mode", "source_process", "proxy_url"])
	.groupBy("id")
	.orderBy("started_at", "desc")
	.limit(limit)
	.as("s");
	const query = kysely
	.selectFrom(sessions)
	.select([
	"s.id",
	"s.started_at as startedAt",
	"s.ended_at as endedAt",
	"s.mode",
	"s.source_process as sourceProcess",
	"s.proxy_url as proxyUrl",
	])
	.select((eb) =>
	eb
	.selectFrom("capture_events as e")
	.select((event) => event.fn.count<number>("e.id").as("count"))
	.whereRef("e.session_id", "=", "s.id")
	.as("eventCount"),
	)
	.orderBy("s.started_at", "desc");
	const { compiled, bind } = compileSqliteQueryBindings(() => query);
	// Native reads retain the store's failure ownership without evicting its borrowed DB.
	return db /* sqlite-allow-raw -- Execute Kysely SQL with native failure ownership. */
	.prepare(compiled.sql)
	.all(...bind(undefined)) as InferResult<typeof query>; // SAFETY: Native columns follow this generated projection.
	}

	export function findDebugProxyCaptureBlobReference(
	db: NodeSqliteDatabase,
	blobId: string,
	): string \| null {
	const query = getNodeSqliteKysely<DebugProxyCaptureDatabase>(db)
	.selectFrom("capture_events")
	.select("data_blob_id as blobId")
	.where("data_blob_id", "=", blobId)
	.limit(1);
	const { compiled, bind } = compileSqliteQueryBindings(() => query);
	const row = db /* sqlite-allow-raw -- Execute Kysely SQL with native failure ownership. */
	.prepare(compiled.sql)
	.get(...bind(undefined)) as InferResult<typeof query>[number] \| undefined; // SAFETY: Native lookup returns the selected nullable text column.
	return row?.blobId \|\| null;
	}

	export function queryDebugProxyCapturePreset(
	db: NodeSqliteDatabase,
	preset: CaptureQueryPreset,
	sessionId?: string,
	): CaptureQueryRow[] {
	const kysely = getNodeSqliteKysely<DebugProxyCaptureDatabase>(db);
	let events = kysely.selectFrom("capture_events");
	if (sessionId) {
	events = events.where("session_id", "=", sessionId);
	}
	const locations = events.select(["host", "path"]);
	const count = kysely.fn.countAll<number>();
	let query: Compilable<CaptureQueryRow>;
	// Aggregate in SQLite so diagnostics do not hydrate whole capture sessions.
	switch (preset) {
	case "double-sends":
	query = locations
	.select(["method", count.as("duplicateCount")])
	.where("kind", "=", "request")
	.groupBy(["host", "path", "method", "data_sha256"])
	.having(count, ">", 1)
	.orderBy("duplicateCount", "desc")
	.orderBy("host", "asc");
	break;
	case "retry-storms":
	query = locations
	.select(count.as("errorCount"))
	.where("kind", "=", "response")
	.where("status", ">=", 429)
	.groupBy(["host", "path"])
	.having(count, ">", 1)
	.orderBy("errorCount", "desc")
	.orderBy("host", "asc");
	break;
	case "cache-busting":
	query = locations
	.select(count.as("variantCount"))
	.where("kind", "=", "request")
	.where((eb) =>
	eb.or([
	eb("path", "like", "%?%"),
	eb("headers_json", "like", "%cache-control%"),
	eb("headers_json", "like", "%pragma%"),
	]),
	)
	.groupBy(["host", "path"])
	.orderBy("variantCount", "desc")
	.orderBy("host", "asc");
	break;
	case "ws-duplicate-frames":
	query = locations
	.select(count.as("duplicateFrames"))
	.where("kind", "=", "ws-frame")
	.where("direction", "=", "outbound")
	.groupBy(["host", "path", "data_sha256"])
	.having(count, ">", 1)
	.orderBy("duplicateFrames", "desc")
	.orderBy("host", "asc");
	break;
	case "missing-ack":
	query = events
	.select(["flow_id as flowId", "host", "path", count.as("outboundFrames")])
	.where("kind", "=", "ws-frame")
	.where("direction", "=", "outbound")
	.where(
	"flow_id",
	"not in",
	events
	.select("flow_id")
	.where("kind", "=", "ws-frame")
	.where("direction", "=", "inbound"),
	)
	.groupBy(["flow_id", "host", "path"])
	.orderBy("outboundFrames", "desc");
	break;
	case "error-bursts":
	query = locations
	.select(count.as("errorCount"))
	.where("kind", "=", "error")
	.groupBy(["host", "path"])
	.orderBy("errorCount", "desc")
	.orderBy("host", "asc");
	break;
	default:
	return [];
	}
	const { compiled, bind } = compileSqliteQueryBindings(() => query);
	return db /* sqlite-allow-raw -- Execute Kysely SQL with native failure ownership. */
	.prepare(compiled.sql)
	.all(...bind(undefined)) as CaptureQueryRow[]; // SAFETY: Each typed preset selects only text, count, or nullable columns.
	}

	export function readDebugProxyCaptureSessionEvents(
	db: NodeSqliteDatabase,
	sessionId: string,
	limit = 500,
	): Array<Record<string, unknown>> {
	return executeSqliteQuerySync(
	db,
	getNodeSqliteKysely<DebugProxyCaptureDatabase>(db)
	.selectFrom("capture_events")
	.select([
	"id",
	"session_id as sessionId",
	"ts",
	"source_scope as sourceScope",
	"source_process as sourceProcess",
	"protocol",
	"direction",
	"kind",
	"flow_id as flowId",
	"method",
	"host",
	"path",
	"status",
	"close_code as closeCode",
	"content_type as contentType",
	"headers_json as headersJson",
	"data_text as dataText",
	"data_blob_id as dataBlobId",
	"data_sha256 as dataSha256",
	"error_text as errorText",
	"meta_json as metaJson",
	])
	.where("session_id", "=", sessionId)
	.orderBy("ts", "desc")
	.orderBy("id", "desc")
	.limit(limit),
	).rows;
	}

	// Metadata is optional and user/tool supplied, so parse defensively for coverage
	// summaries instead of assuming every event has valid JSON.
	function parseMetaJson(metaJson: unknown): Record<string, unknown> \| null {
	if (typeof metaJson !== "string" \|\| metaJson.trim().length === 0) {
	return null;
	}
	try {
	const parsed = JSON.parse(metaJson) as unknown;
	// SAFETY: Parsed objects, including arrays, are read only through optional label keys.
	return parsed && typeof parsed === "object" ? (parsed as Record<string, unknown>) : null;
	} catch {
	return null;
	}
	}

	function sortObservedCounts(counts: Map<string, number>): CaptureObservedDimension[] {
	return [...counts.entries()]
	.map(([value, count]) => ({ value, count }))
	.toSorted((left, right) => right.count - left.count \|\| left.value.localeCompare(right.value));
	}

	export function summarizeDebugProxyCaptureSessionCoverage(
	db: NodeSqliteDatabase,
	sessionId: string,
	): CaptureSessionCoverageSummary {
	const { compiled, bind } = compileSqliteQueryBindings<string>((parameter) =>
	getNodeSqliteKysely<DebugProxyCaptureDatabase>(db)
	.selectFrom("capture_events")
	.select(["host", "meta_json as metaJson"])
	.where(
	"session_id",
	"=",
	parameter((value) => value),
	),
	);
	// Native iteration keeps corruption from evicting the borrowed shared database.
	const rows = db /* sqlite-allow-raw -- Execute Kysely SQL with native failure ownership. */
	.prepare(compiled.sql)
	.iterate(...bind(sessionId));
	const providers = new Map<string, number>();
	const apis = new Map<string, number>();
	const models = new Map<string, number>();
	const hosts = new Map<string, number>();
	const localPeers = new Map<string, number>();
	let totalEvents = 0;
	let unlabeledEventCount = 0;
	try {
	for (const row of rows) {
	totalEvents += 1;
	const meta = parseMetaJson(row.metaJson);
	const provider = normalizeObservedValue(meta?.provider);
	const api = normalizeObservedValue(meta?.api);
	const model = normalizeObservedValue(meta?.model);
	const host = normalizeObservedValue(row.host);
	if (!provider && !api && !model) {
	unlabeledEventCount += 1;
	}
	if (provider) {
	providers.set(provider, (providers.get(provider) ?? 0) + 1);
	}
	if (api) {
	apis.set(api, (apis.get(api) ?? 0) + 1);
	}
	if (model) {
	models.set(model, (models.get(model) ?? 0) + 1);
	}
	if (host) {
	hosts.set(host, (hosts.get(host) ?? 0) + 1);
	// Local model/provider endpoints are useful to surface separately when
	// debugging why cloud-provider labels are absent.
	if (host.startsWith("127.0.0.1:") \|\| host.startsWith("localhost:")) {
	localPeers.set(host, (localPeers.get(host) ?? 0) + 1);
	}
	}
	}
	} catch (error) {
	try {
	rows.return?.();
	} catch {
	// Iterator cleanup must not replace the original read failure.
	}
	throw error;
	}
	return {
	sessionId,
	totalEvents,
	unlabeledEventCount,
	providers: sortObservedCounts(providers),
	apis: sortObservedCounts(apis),
	models: sortObservedCounts(models),
	hosts: sortObservedCounts(hosts),
	localPeers: sortObservedCounts(localPeers),
	};
	}

	export function readDebugProxyCaptureBlob(db: NodeSqliteDatabase, blobId: string): string \| null {
	const row = executeSqliteQueryTakeFirstSync(
	db,
	getNodeSqliteKysely<DebugProxyCaptureDatabase>(db)
	.selectFrom("capture_blobs")
	.select(["encoding", "data"])
	.where("blob_id", "=", blobId),
	);
	if (!row?.data) {
	return null;
	}
	const data = Buffer.from(row.data);
	return (row.encoding === "gzip" ? gunzipSync(data) : data).toString("utf8");
	}

	/** Read capture rows without joining or mutating the shared-state writer lifecycle. */
	export function createDebugProxyCaptureReader(params: {
	env: NodeJS.ProcessEnv;
	}): DebugProxyCaptureReader {
	return {
	getSessionEvents(sessionId, limit) {
	return (
	withExistingOpenClawStateDatabaseReadOnly(
	({ db }) => readDebugProxyCaptureSessionEvents(db, sessionId, limit),
	{ env: params.env },
	) ?? []
	);
	},
	readBlob(blobId) {
	return (
	withExistingOpenClawStateDatabaseReadOnly(
	({ db }) => readDebugProxyCaptureBlob(db, blobId),
	{ env: params.env },
	) ?? null
	);
	},
	};
	}