From 3a4e41a2485aed2c5bd1ef1ccdd81ae5f90e5ad4 Mon Sep 17 00:00:00 2001 From: AgentRun Codex Date: Sun, 12 Jul 2026 20:16:59 +0000 Subject: [PATCH] fix(monitor): harden central ingest contracts --- scripts/monitor-central-service.ts | 35 +- .../src/monitor-central-persistence.test.ts | 108 +++--- scripts/src/monitor-central-persistence.ts | 313 ++++++++++-------- 3 files changed, 252 insertions(+), 204 deletions(-) diff --git a/scripts/monitor-central-service.ts b/scripts/monitor-central-service.ts index 967b489a..790b8e17 100644 --- a/scripts/monitor-central-service.ts +++ b/scripts/monitor-central-service.ts @@ -1,15 +1,28 @@ // SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence. -// Responsibility: Explicit DATABASE_URL bootstrap for the independent central Monitor HTTP service. +// Responsibility: Environment-projected bootstrap for the independent central Monitor HTTP service. import { createMonitorCentralService, createPostgresMonitorCentralStore, MonitorCentralConfigurationError } from "./src/monitor-central-persistence"; -const databaseUrl = process.env.DATABASE_URL; -if (!databaseUrl) { - console.log(JSON.stringify(new MonitorCentralConfigurationError().toJSON())); - process.exitCode = 2; -} else { - const store = createPostgresMonitorCentralStore(databaseUrl); - await store.migrate(); - const service = createMonitorCentralService({ store }); - const server = Bun.serve({ hostname: process.env.MONITOR_CENTRAL_HOST ?? "127.0.0.1", port: Number(process.env.MONITOR_CENTRAL_PORT ?? "8787"), fetch: service.fetch }); - console.log(JSON.stringify({ ok: true, service: "monitor-central", port: server.port, valuesRedacted: true })); +function requiredString(name: string): string { + const value = process.env[name]; + if (!value) throw Object.assign(new Error(`${name} is required`), { code: "monitor_central_runtime_config_missing" }); + return value; +} + +function requiredPositiveInteger(name: string): number { + const value = Number(requiredString(name)); + if (!Number.isSafeInteger(value) || value < 1) throw Object.assign(new Error(`${name} must be a positive integer`), { code: "monitor_central_runtime_config_invalid" }); + return value; +} + +try { + const databaseUrl = requiredString("DATABASE_URL"); + const store = createPostgresMonitorCentralStore(databaseUrl, { poolMax: requiredPositiveInteger("MONITOR_CENTRAL_PG_POOL_MAX"), idleTimeoutSeconds: requiredPositiveInteger("MONITOR_CENTRAL_PG_IDLE_TIMEOUT_SECONDS") }); + await store.migrate(); + const service = createMonitorCentralService({ store, pageSize: requiredPositiveInteger("MONITOR_CENTRAL_PAGE_SIZE") }); + const server = Bun.serve({ hostname: requiredString("MONITOR_CENTRAL_HOST"), port: requiredPositiveInteger("MONITOR_CENTRAL_PORT"), fetch: service.fetch }); + console.log(JSON.stringify({ ok: true, service: "monitor-central", port: server.port, valuesRedacted: true })); +} catch (error) { + const structured = error instanceof MonitorCentralConfigurationError ? error.toJSON() : { ok: false, error: { code: (error as { code?: string })?.code ?? "monitor_central_runtime_config_invalid", message: error instanceof Error ? error.message : "invalid configuration" }, valuesRedacted: true }; + console.log(JSON.stringify(structured)); + process.exitCode = 2; } diff --git a/scripts/src/monitor-central-persistence.test.ts b/scripts/src/monitor-central-persistence.test.ts index 5a573866..5578f9e5 100644 --- a/scripts/src/monitor-central-persistence.test.ts +++ b/scripts/src/monitor-central-persistence.test.ts @@ -2,26 +2,42 @@ import assert from "node:assert/strict"; import { spawnSync } from "node:child_process"; import { test } from "bun:test"; -import { createInMemoryMonitorCentralStore, createMonitorCentralService, MONITOR_CENTRAL_MIGRATIONS, MonitorCentralConfigurationError } from "./monitor-central-persistence"; +import { createInMemoryMonitorCentralStore, createMonitorCentralService, mapPostgresFindingRow, MONITOR_CENTRAL_MIGRATIONS, MonitorCentralConfigurationError } from "./monitor-central-persistence"; -function service() { return createMonitorCentralService({ store: createInMemoryMonitorCentralStore() }); } -async function ingest(target: ReturnType, runId: string, updatedAt: string, scope: Record = {}) { - return target.ingest({ sentinelId: scope.sentinelId ?? "sentinel-a", node: scope.node ?? "NC01", lane: scope.lane ?? "v03", runId, updatedAt, status: "succeeded", payload: { runId, views: { summary: { state: "ready" } } }, findings: [{ id: `finding-${runId}` }], artifactLocators: [{ locator: `s3://evidence/${runId}`, sha256: "sha256:abc", owner: "runner", sizeBytes: 12, kind: "report" }], timeline: [{ phase: "done" }] }); -} +function service() { return createMonitorCentralService({ store: createInMemoryMonitorCentralStore(), pageSize: 50 }); } +function locator(runId: string) { return { ownerNode: "NC01", ownerLane: "v03", ownerPvc: "sentinel-pvc", stateDir: `/state/${runId}`, relativePath: "analysis/report.json", sha256: "sha256:abc", sizeBytes: 12, kind: "report", reachable: true }; } +function input(runId: string, updatedAt = "2026-07-12T10:00:00.000Z", scope: Record = {}) { return { sentinelId: scope.sentinelId ?? "sentinel-a", node: scope.node ?? "NC01", lane: scope.lane ?? "v03", runId, updatedAt, status: "succeeded", payload: { runId, views: { summary: { state: "ready" } } }, findings: [{ id: `${runId}-0` }, { id: `${runId}-1` }], artifactLocators: [locator(runId)], timeline: [{ phase: "done" }] }; } +async function ingest(target: ReturnType, runId: string, updatedAt?: string, scope?: Record) { return target.ingest(input(runId, updatedAt, scope)); } -test("schema covers central persistence objects and does not store artifact blobs", () => { +test("schema covers runner FK, locator reachability, and atomic insert contract", () => { const schema = MONITOR_CENTRAL_MIGRATIONS.join("\n"); for (const name of ["schema_migrations", "runners", "runs", "findings", "run_payloads", "artifact_locators", "timeline_events", "import_manifests"]) assert.match(schema, new RegExp(`monitor\\.${name}`)); - assert.doesNotMatch(schema, /blob|bytea/i); + assert.match(schema, /foreign key \(sentinel_id, node, lane\) references monitor\.runners/u); + assert.match(schema, /owner_pvc.*state_dir.*relative_path.*reachable/u); + assert.doesNotMatch(schema, /blob|bytea/iu); }); -test("ingest is transactional in contract, idempotent by payload hash, and rejects conflicts", async () => { +test("HTTP terminal ingest is unbound-safe, ignores caller hash, and maps conflict to 409", async () => { const target = service(); - const first = await ingest(target, "run-1", "2026-07-12T10:00:00.000Z"); - const repeated = await ingest(target, "run-1", "2026-07-12T10:00:00.000Z"); - assert.equal(first.disposition, "inserted"); - assert.equal(repeated.disposition, "idempotent"); - await assert.rejects(() => target.ingest({ sentinelId: "sentinel-a", node: "NC01", lane: "v03", runId: "run-1", payload: { changed: true } }), { code: "payload_hash_conflict" }); + const fetch = target.fetch; + const first = await fetch(new Request("http://local/api/ingest", { method: "POST", body: JSON.stringify(input("run-1")) })); + const repeated = await fetch(new Request("http://local/api/ingest", { method: "POST", body: JSON.stringify({ ...input("run-1"), payloadHash: "sha256:forged" }) })); + const conflict = await fetch(new Request("http://local/api/ingest", { method: "POST", body: JSON.stringify({ ...input("run-1"), status: "blocked", payloadHash: "sha256:forged" }) })); + assert.equal(first.status, 202); + assert.equal((await first.json() as any).disposition, "inserted"); + assert.equal((await repeated.json() as any).disposition, "idempotent"); + assert.equal(conflict.status, 409); + assert.equal((await conflict.json() as any).error.code, "payload_hash_conflict"); + assert.equal((await fetch(new Request("http://local/health"))).status, 200); +}); + +test("full terminal envelope controls hash and concurrent same identity converges", async () => { + const target = service(); + const base = { ...input("run-envelope"), payloadHash: "sha256:caller-controlled" }; + const results = await Promise.all([target.ingest(base), target.ingest(base)]); + assert.deepEqual(new Set(results.map((item) => item.disposition)), new Set(["inserted", "idempotent"])); + await assert.rejects(() => target.ingest({ ...base, findings: [{ id: "changed" }] }), { code: "payload_hash_conflict" }); + await assert.rejects(() => target.ingest({ ...base, artifactLocators: [{ ...locator("run-envelope"), reachable: false }] }), { code: "payload_hash_conflict" }); }); test("global and scoped latest use updatedAt then runId descending", async () => { @@ -35,47 +51,47 @@ test("global and scoped latest use updatedAt then runId descending", async () => assert.equal(scoped.overview.latest.runId, "run-c"); }); -test("runs and findings are bounded, cursor-paged, and scope-isolated", async () => { +test("detail and view require full composite identity", async () => { const target = service(); - await ingest(target, "run-1", "2026-07-12T10:00:00.000Z"); - await ingest(target, "run-2", "2026-07-12T11:00:00.000Z"); - await ingest(target, "run-3", "2026-07-12T12:00:00.000Z", { lane: "v04" }); - const first = await (await target.fetch(new Request("http://local/api/runs?lane=v03&limit=1"))).json() as any; - const second = await (await target.fetch(new Request(`http://local/api/runs?lane=v03&limit=1000&cursor=${first.nextCursor}`))).json() as any; - const findings = await (await target.fetch(new Request("http://local/api/findings?lane=v03"))).json() as any; - assert.equal(first.items.length, 1); - assert.equal(first.items[0].runId, "run-2"); - assert.equal(second.items.length, 1); - assert.equal(second.items[0].runId, "run-1"); - assert.equal(findings.items.length, 2); + await ingest(target, "shared", undefined, { sentinelId: "sentinel-a" }); + await ingest(target, "shared", undefined, { sentinelId: "sentinel-b" }); + const missing = await target.fetch(new Request("http://local/api/runs/shared")); + const detail = await (await target.fetch(new Request("http://local/api/runs/shared?sentinelId=sentinel-b&node=NC01&lane=v03"))).json() as any; + const view = await (await target.fetch(new Request("http://local/api/runs/shared/views/summary?sentinelId=sentinel-b&node=NC01&lane=v03"))).json() as any; + assert.equal(missing.status, 400); + assert.equal((await missing.json() as any).error.code, "scope_required"); + assert.equal(detail.run.sentinelId, "sentinel-b"); + assert.equal(view.view.value.state, "ready"); }); -test("run detail returns payload, locator metadata, timeline, and no secret values", async () => { +test("findings cursor contains finding index and does not skip a same-run page", async () => { const target = service(); - await ingest(target, "run-detail", "2026-07-12T10:00:00.000Z"); - const response = await (await target.fetch(new Request("http://local/api/runs/run-detail"))).json() as any; - assert.equal(response.run.payload.runId, "run-detail"); - assert.deepEqual(response.run.artifactLocators[0], { locator: "s3://evidence/run-detail", sha256: "sha256:abc", owner: "runner", sizeBytes: 12, kind: "report" }); + await ingest(target, "run-findings"); + const first = await (await target.fetch(new Request("http://local/api/findings?limit=1"))).json() as any; + const second = await (await target.fetch(new Request(`http://local/api/findings?limit=1&cursor=${first.nextCursor}`))).json() as any; + assert.equal(first.items[0].findingIndex, 1); + assert.equal(second.items[0].findingIndex, 0); +}); + +test("PG finding mapping preserves selected status and payload hash", () => { + const row = mapPostgresFindingRow({ sentinel_id: "sentinel-a", node: "NC01", lane: "v03", run_id: "run-pg", updated_at: "2026-07-12T10:00:00.000Z", status: "blocked", payload_hash: "sha256:terminal", finding_index: 2, finding: { id: "pg" } }); + assert.deepEqual(row, { sentinelId: "sentinel-a", node: "NC01", lane: "v03", runId: "run-pg", updatedAt: "2026-07-12T10:00:00.000Z", status: "blocked", payloadHash: "sha256:terminal", findingIndex: 2, finding: { id: "pg" } }); +}); + +test("run detail returns normalized locator metadata without artifact bytes", async () => { + const target = service(); + await ingest(target, "run-detail"); + const response = await (await target.fetch(new Request("http://local/api/runs/run-detail?sentinelId=sentinel-a&node=NC01&lane=v03"))).json() as any; + assert.deepEqual(response.run.artifactLocators[0], locator("run-detail")); assert.equal(response.valuesRedacted, true); }); -test("health is redacted and missing DATABASE_URL fails structurally", async () => { +test("configuration is explicit and missing environment projection fails structurally", async () => { const target = service(); - const health = await target.health(); - assert.equal(health.ok, true); - assert.equal(health.valuesRedacted, true); - assert.throws(() => createMonitorCentralService({ databaseUrl: "" }), MonitorCentralConfigurationError); -}); - -test("view contract reshapes an existing report payload without reading runner artifacts", async () => { - const target = service(); - await ingest(target, "run-view", "2026-07-12T10:00:00.000Z"); - const response = await (await target.fetch(new Request("http://local/api/runs/run-view/views/summary"))).json() as any; - assert.deepEqual(response.view, { runId: "run-view", name: "summary", value: { state: "ready" } }); -}); - -test("service entry exits with a structured redacted error when DATABASE_URL is absent", () => { + assert.equal((await target.health()).ok, true); + assert.throws(() => createMonitorCentralService({ databaseUrl: "", pageSize: 50 }), MonitorCentralConfigurationError); + assert.throws(() => createMonitorCentralService({ store: createInMemoryMonitorCentralStore() }), { code: "monitor_central_page_size_missing" }); const result = spawnSync("bun", ["scripts/monitor-central-service.ts"], { cwd: process.cwd(), encoding: "utf8", env: { PATH: process.env.PATH ?? "" } }); assert.equal(result.status, 2); - assert.deepEqual(JSON.parse(result.stdout), { ok: false, error: { code: "monitor_central_database_url_missing", message: "Monitor central service requires an explicit DATABASE_URL" }, valuesRedacted: true }); + assert.equal(JSON.parse(result.stdout).error.code, "monitor_central_runtime_config_missing"); }); diff --git a/scripts/src/monitor-central-persistence.ts b/scripts/src/monitor-central-persistence.ts index f27bd2a0..79188964 100644 --- a/scripts/src/monitor-central-persistence.ts +++ b/scripts/src/monitor-central-persistence.ts @@ -1,11 +1,10 @@ // SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence. -// Responsibility: Host PostgreSQL-backed Monitor ingest, bounded read models, and health contracts. +// Responsibility: Host PostgreSQL-backed Monitor terminal ingest, bounded read models, and health contracts. import { createHash } from "node:crypto"; import postgres from "postgres"; export const MONITOR_CENTRAL_SPEC_REF = "PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence"; -export const MONITOR_CENTRAL_SCHEMA_VERSION = "2026-07-12-p17-1"; -const MAX_PAGE_SIZE = 100; +export const MONITOR_CENTRAL_SCHEMA_VERSION = "2026-07-12-p17-2"; export interface MonitorScope { readonly sentinelId?: string; @@ -14,11 +13,15 @@ export interface MonitorScope { } export interface MonitorArtifactLocator { - readonly locator: string; - readonly sha256?: string; - readonly owner?: string; - readonly sizeBytes?: number; - readonly kind?: string; + readonly ownerNode: string; + readonly ownerLane: string; + readonly ownerPvc: string; + readonly stateDir: string; + readonly relativePath: string; + readonly sha256: string; + readonly sizeBytes: number; + readonly kind: string; + readonly reachable: boolean; } export interface MonitorCentralIngest { @@ -35,7 +38,7 @@ export interface MonitorCentralIngest { readonly timeline?: readonly Record[]; } -export interface MonitorRun extends MonitorScope { +export interface MonitorRun extends Required { readonly runId: string; readonly updatedAt: string; readonly status: string; @@ -46,24 +49,37 @@ export interface MonitorCentralStore { migrate(): Promise; ping(): Promise; schemaVersion(): Promise; - ingest(input: Required): Promise<{ readonly disposition: "inserted" | "idempotent" }>; + ingest(input: NormalizedIngest): Promise<{ readonly disposition: "inserted" | "idempotent" }>; overview(scope: MonitorScope): Promise>; - runs(scope: MonitorScope, options: QueryOptions): Promise; - run(scope: MonitorScope, runId: string): Promise | null>; - view(scope: MonitorScope, runId: string, view: string): Promise | null>; - findings(scope: MonitorScope, options: QueryOptions): Promise[]>; + runs(scope: MonitorScope, options: RunQueryOptions): Promise; + run(scope: Required, runId: string): Promise | null>; + view(scope: Required, runId: string, view: string): Promise | null>; + findings(scope: MonitorScope, options: FindingQueryOptions): Promise[]>; } -export interface QueryOptions { - readonly limit: number; - readonly cursor: Cursor | null; +interface NormalizedIngest extends Required> { + readonly payloadHash: string; } -interface Cursor { +interface RunCursor { readonly updatedAt: string; readonly runId: string; } +interface FindingCursor extends RunCursor { + readonly findingIndex: number; +} + +export interface RunQueryOptions { + readonly limit: number; + readonly cursor: RunCursor | null; +} + +export interface FindingQueryOptions { + readonly limit: number; + readonly cursor: FindingCursor | null; +} + export interface MonitorCentralService { ingest(input: MonitorCentralIngest): Promise>; health(): Promise>; @@ -73,6 +89,13 @@ export interface MonitorCentralService { export interface MonitorCentralServiceOptions { readonly databaseUrl?: string | null; readonly store?: MonitorCentralStore; + readonly pageSize?: number; + readonly postgres?: PostgresMonitorCentralStoreOptions; +} + +export interface PostgresMonitorCentralStoreOptions { + readonly poolMax: number; + readonly idleTimeoutSeconds: number; } export class MonitorCentralConfigurationError extends Error { @@ -89,61 +112,60 @@ export class MonitorCentralConfigurationError extends Error { } export function createMonitorCentralService(options: MonitorCentralServiceOptions = {}): MonitorCentralService { - const store = options.store ?? createPostgresMonitorCentralStore(requireDatabaseUrl(options.databaseUrl)); - return { - async ingest(input) { - const normalized = normalizeIngest(input); - const result = await store.ingest(normalized); - return { ok: true, ...result, identity: identityOf(normalized), payloadHash: normalized.payloadHash, valuesRedacted: true }; - }, - async health() { - const checks: Record[] = [{ name: "configuration", ok: true, valuesRedacted: true }]; - try { - await store.ping(); - checks.push({ name: "postgres", ok: true, valuesRedacted: true }); - const schemaVersion = await store.schemaVersion(); - checks.push({ name: "schema", ok: schemaVersion === MONITOR_CENTRAL_SCHEMA_VERSION, version: schemaVersion, valuesRedacted: true }); - checks.push({ name: "ingest", ok: true, valuesRedacted: true }); - checks.push({ name: "query", ok: true, valuesRedacted: true }); - } catch (error) { - checks.push({ name: "postgres", ok: false, code: "postgres_unavailable", message: errorMessage(error), valuesRedacted: true }); - } - return { ok: checks.every((item) => item.ok === true), checks, valuesRedacted: true }; - }, - async fetch(request) { - const url = new URL(request.url); - try { - if (request.method === "GET" && url.pathname === "/health") return json(await this.health()); - if (request.method === "GET" && url.pathname === "/api/overview") return json({ ok: true, overview: await store.overview(scopeFrom(url)), valuesRedacted: true }); - if (request.method === "GET" && url.pathname === "/api/runs") return json(page(await store.runs(scopeFrom(url), queryOptions(url)))); - if (request.method === "GET" && url.pathname.startsWith("/api/runs/")) { - const suffix = decodeURIComponent(url.pathname.slice("/api/runs/".length)); - const viewMatch = suffix.match(/^([^/]+)\/views\/([^/]+)$/u); - if (viewMatch) { - const view = await store.view(scopeFrom(url), viewMatch[1], viewMatch[2]); - return view ? json({ ok: true, view, valuesRedacted: true }) : json({ ok: false, error: { code: "view_not_found" }, valuesRedacted: true }, 404); - } - const run = await store.run(scopeFrom(url), suffix); - return run ? json({ ok: true, run, valuesRedacted: true }) : json({ ok: false, error: { code: "run_not_found" }, valuesRedacted: true }, 404); - } - if (request.method === "GET" && url.pathname === "/api/findings") return json(page(await store.findings(scopeFrom(url), queryOptions(url)))); - return json({ ok: false, error: { code: "not_found" }, valuesRedacted: true }, 404); - } catch (error) { - return json({ ok: false, error: structuredError(error), valuesRedacted: true }, error instanceof MonitorCentralConfigurationError ? 503 : 400); - } - }, + const pageSize = requirePositiveInteger(options.pageSize, "monitor_central_page_size_missing"); + const store = options.store ?? createPostgresMonitorCentralStore(requireDatabaseUrl(options.databaseUrl), requirePostgresOptions(options.postgres)); + const ingest = async (input: MonitorCentralIngest) => { + const normalized = normalizeIngest(input); + const result = await store.ingest(normalized); + return { ok: true, ...result, identity: identityOf(normalized), payloadHash: normalized.payloadHash, valuesRedacted: true }; }; + const health = async () => { + const checks: Record[] = [{ name: "configuration", ok: true, valuesRedacted: true }]; + try { + await store.ping(); + checks.push({ name: "postgres", ok: true, valuesRedacted: true }); + const schemaVersion = await store.schemaVersion(); + checks.push({ name: "schema", ok: schemaVersion === MONITOR_CENTRAL_SCHEMA_VERSION, version: schemaVersion, valuesRedacted: true }); + checks.push({ name: "ingest", ok: true, valuesRedacted: true }); + checks.push({ name: "query", ok: true, valuesRedacted: true }); + } catch (error) { + checks.push({ name: "postgres", ok: false, code: "postgres_unavailable", message: errorMessage(error), valuesRedacted: true }); + } + return { ok: checks.every((item) => item.ok === true), checks, valuesRedacted: true }; + }; + const fetch = async (request: Request): Promise => { + const url = new URL(request.url); + try { + if (request.method === "POST" && url.pathname === "/api/ingest") return json(await ingest(await requestRecord(request)), 202); + if (request.method === "GET" && url.pathname === "/health") return json(await health()); + if (request.method === "GET" && url.pathname === "/api/overview") return json({ ok: true, overview: await store.overview(scopeFrom(url)), valuesRedacted: true }); + if (request.method === "GET" && url.pathname === "/api/runs") return json(runPage(await store.runs(scopeFrom(url), runQueryOptions(url, pageSize)))); + if (request.method === "GET" && url.pathname === "/api/findings") return json(findingPage(await store.findings(scopeFrom(url), findingQueryOptions(url, pageSize)))); + if (request.method === "GET" && url.pathname.startsWith("/api/runs/")) { + const suffix = decodeURIComponent(url.pathname.slice("/api/runs/".length)); + const scope = requireFullScope(scopeFrom(url)); + const viewMatch = suffix.match(/^([^/]+)\/views\/([^/]+)$/u); + if (viewMatch) { + const view = await store.view(scope, viewMatch[1], viewMatch[2]); + return view ? json({ ok: true, view, valuesRedacted: true }) : json({ ok: false, error: { code: "view_not_found" }, valuesRedacted: true }, 404); + } + const run = await store.run(scope, suffix); + return run ? json({ ok: true, run, valuesRedacted: true }) : json({ ok: false, error: { code: "run_not_found" }, valuesRedacted: true }, 404); + } + return json({ ok: false, error: { code: "not_found" }, valuesRedacted: true }, 404); + } catch (error) { + return json({ ok: false, error: structuredError(error), valuesRedacted: true }, httpStatus(error)); + } + }; + return { ingest, health, fetch }; } -export function createPostgresMonitorCentralStore(databaseUrl: string): MonitorCentralStore { - const sql = postgres(databaseUrl, { max: 4, idle_timeout: 10 }); +export function createPostgresMonitorCentralStore(databaseUrl: string, options: PostgresMonitorCentralStoreOptions): MonitorCentralStore { + const sql = postgres(databaseUrl, { max: requirePositiveInteger(options.poolMax, "monitor_central_pg_pool_max_missing"), idle_timeout: requirePositiveInteger(options.idleTimeoutSeconds, "monitor_central_pg_idle_timeout_missing") }); return { async migrate() { for (const statement of MONITOR_CENTRAL_MIGRATIONS) await sql.unsafe(statement); - await sql` - insert into monitor.schema_migrations (version, applied_at) - values (${MONITOR_CENTRAL_SCHEMA_VERSION}, now()) - on conflict (version) do nothing`; + await sql`insert into monitor.schema_migrations (version, applied_at) values (${MONITOR_CENTRAL_SCHEMA_VERSION}, now()) on conflict (version) do nothing`; }, async ping() { await sql`select 1`; }, async schemaVersion() { @@ -152,21 +174,22 @@ export function createPostgresMonitorCentralStore(databaseUrl: string): MonitorC }, async ingest(input) { return sql.begin(async (transaction) => { - const existing = await transaction<{ payload_hash: string }[]>` - select payload_hash from monitor.runs - where sentinel_id = ${input.sentinelId} and node = ${input.node} and lane = ${input.lane} and run_id = ${input.runId} - for update`; - if (existing[0]) { - if (existing[0].payload_hash !== input.payloadHash) throw conflict(input); - return { disposition: "idempotent" as const }; - } await transaction` insert into monitor.runners (sentinel_id, node, lane, updated_at) values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.updatedAt}) - on conflict (sentinel_id, node, lane) do update set updated_at = excluded.updated_at`; - await transaction` + on conflict (sentinel_id, node, lane) do update set updated_at = greatest(monitor.runners.updated_at, excluded.updated_at)`; + const inserted = await transaction<{ payload_hash: string }[]>` insert into monitor.runs (sentinel_id, node, lane, run_id, updated_at, status, payload_hash) - values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.runId}, ${input.updatedAt}, ${input.status}, ${input.payloadHash})`; + values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.runId}, ${input.updatedAt}, ${input.status}, ${input.payloadHash}) + on conflict (sentinel_id, node, lane, run_id) do nothing + returning payload_hash`; + if (!inserted[0]) { + const existing = await transaction<{ payload_hash: string }[]>` + select payload_hash from monitor.runs + where sentinel_id = ${input.sentinelId} and node = ${input.node} and lane = ${input.lane} and run_id = ${input.runId}`; + if (existing[0]?.payload_hash === input.payloadHash) return { disposition: "idempotent" as const }; + throw conflict(input); + } await transaction` insert into monitor.run_payloads (sentinel_id, node, lane, run_id, payload_hash, payload) values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.runId}, ${input.payloadHash}, ${JSON.stringify(input.payload)}::jsonb)`; @@ -174,8 +197,8 @@ export function createPostgresMonitorCentralStore(databaseUrl: string): MonitorC insert into monitor.findings (sentinel_id, node, lane, run_id, finding_index, finding) values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.runId}, ${index}, ${JSON.stringify(finding)}::jsonb)`; for (const [index, locator] of input.artifactLocators.entries()) await transaction` - insert into monitor.artifact_locators (sentinel_id, node, lane, run_id, locator_index, locator, sha256, owner_name, size_bytes, kind) - values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.runId}, ${index}, ${locator.locator}, ${locator.sha256 ?? null}, ${locator.owner ?? null}, ${locator.sizeBytes ?? null}, ${locator.kind ?? null})`; + insert into monitor.artifact_locators (sentinel_id, node, lane, run_id, locator_index, owner_node, owner_lane, owner_pvc, state_dir, relative_path, sha256, size_bytes, kind, reachable) + values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.runId}, ${index}, ${locator.ownerNode}, ${locator.ownerLane}, ${locator.ownerPvc}, ${locator.stateDir}, ${locator.relativePath}, ${locator.sha256}, ${locator.sizeBytes}, ${locator.kind}, ${locator.reachable})`; for (const [index, event] of input.timeline.entries()) await transaction` insert into monitor.timeline_events (sentinel_id, node, lane, run_id, event_index, event) values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.runId}, ${index}, ${JSON.stringify(event)}::jsonb)`; @@ -183,21 +206,20 @@ export function createPostgresMonitorCentralStore(databaseUrl: string): MonitorC }); }, async overview(scope) { - const rows = await queryRows(sql, scope, { limit: 1, cursor: null }); - const count = await countRuns(sql, scope); - return { scope: publicScope(scope), runCount: count, latest: rows[0] ?? null }; + const rows = await queryRuns(sql, scope, { limit: 1, cursor: null }); + return { scope: publicScope(scope), runCount: await countRuns(sql, scope), latest: rows[0] ?? null }; }, - async runs(scope, options) { return queryRows(sql, scope, options); }, + async runs(scope, options) { return queryRuns(sql, scope, options); }, async run(scope, runId) { - const clause = whereClause(scope, runId); - const rows = await sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash, p.payload from monitor.runs r join monitor.run_payloads p using (sentinel_id, node, lane, run_id) where ${clause.text} limit 1`, clause.values); + const where = identityWhere(scope, runId); + const rows = await sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash, p.payload from monitor.runs r join monitor.run_payloads p using (sentinel_id, node, lane, run_id) where ${where.text} limit 1`, where.values); const row = rows[0] as Record | undefined; if (!row) return null; const identity = rowIdentity(row); const [findings, locators, timeline] = await Promise.all([ - sql.unsafe(`select finding from monitor.findings where ${identity.where.text} order by finding_index`, identity.where.values), - sql.unsafe(`select locator, sha256, owner_name, size_bytes, kind from monitor.artifact_locators where ${identity.where.text} order by locator_index`, identity.where.values), - sql.unsafe(`select event from monitor.timeline_events where ${identity.where.text} order by event_index`, identity.where.values), + sql.unsafe(`select finding from monitor.findings where ${identity.text} order by finding_index`, identity.values), + sql.unsafe(`select owner_node, owner_lane, owner_pvc, state_dir, relative_path, sha256, size_bytes, kind, reachable from monitor.artifact_locators where ${identity.text} order by locator_index`, identity.values), + sql.unsafe(`select event from monitor.timeline_events where ${identity.text} order by event_index`, identity.values), ]); return { ...runRow(row), payload: row.payload, findings: findings.map((item) => item.finding), artifactLocators: locators.map(locatorRow), timeline: timeline.map((item) => item.event) }; }, @@ -208,12 +230,16 @@ export function createPostgresMonitorCentralStore(databaseUrl: string): MonitorC return value === undefined ? null : { runId, name: view, value }; }, async findings(scope, options) { - const where = whereClause(scope); - const cursor = options.cursor ? `and (r.updated_at, r.run_id) < ($${where.values.length + 1}, $${where.values.length + 2})` : ""; - const values = options.cursor ? [...where.values, options.cursor.updatedAt, options.cursor.runId, options.limit] : [...where.values, options.limit]; - const limitPosition = values.length; - const rows = await sql.unsafe(`select f.finding, r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at from monitor.findings f join monitor.runs r using (sentinel_id, node, lane, run_id) where ${where.text} ${cursor} order by r.updated_at desc, r.run_id desc, f.finding_index asc limit $${limitPosition}`, values); - return rows.map((row) => ({ ...runRow(row), finding: row.finding })); + const where = scopeWhere(scope); + const values: unknown[] = [...where.values]; + let cursor = ""; + if (options.cursor) { + values.push(options.cursor.updatedAt, options.cursor.runId, options.cursor.findingIndex); + cursor = `and (r.updated_at, r.run_id, f.finding_index) < ($${values.length - 2}, $${values.length - 1}, $${values.length})`; + } + values.push(options.limit); + const rows = await sql.unsafe(`select f.finding, f.finding_index, r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash from monitor.findings f join monitor.runs r using (sentinel_id, node, lane, run_id) where ${where.text} ${cursor} order by r.updated_at desc, r.run_id desc, f.finding_index desc limit $${values.length}`, values); + return rows.map(mapPostgresFindingRow); }, }; } @@ -222,17 +248,17 @@ export const MONITOR_CENTRAL_MIGRATIONS = [ "create schema if not exists monitor", "create table if not exists monitor.schema_migrations (version text primary key, applied_at timestamptz not null)", "create table if not exists monitor.runners (sentinel_id text not null, node text not null, lane text not null, updated_at timestamptz not null, primary key (sentinel_id, node, lane))", - "create table if not exists monitor.runs (sentinel_id text not null, node text not null, lane text not null, run_id text not null, updated_at timestamptz not null, status text not null, payload_hash text not null, primary key (sentinel_id, node, lane, run_id))", + "create table if not exists monitor.runs (sentinel_id text not null, node text not null, lane text not null, run_id text not null, updated_at timestamptz not null, status text not null, payload_hash text not null, primary key (sentinel_id, node, lane, run_id), foreign key (sentinel_id, node, lane) references monitor.runners (sentinel_id, node, lane))", "create index if not exists monitor_runs_latest_idx on monitor.runs (updated_at desc, run_id desc)", "create table if not exists monitor.run_payloads (sentinel_id text not null, node text not null, lane text not null, run_id text not null, payload_hash text not null, payload jsonb not null, primary key (sentinel_id, node, lane, run_id), foreign key (sentinel_id, node, lane, run_id) references monitor.runs on delete cascade)", "create table if not exists monitor.findings (sentinel_id text not null, node text not null, lane text not null, run_id text not null, finding_index integer not null, finding jsonb not null, primary key (sentinel_id, node, lane, run_id, finding_index), foreign key (sentinel_id, node, lane, run_id) references monitor.runs on delete cascade)", - "create table if not exists monitor.artifact_locators (sentinel_id text not null, node text not null, lane text not null, run_id text not null, locator_index integer not null, locator text not null, sha256 text, owner_name text, size_bytes bigint, kind text, primary key (sentinel_id, node, lane, run_id, locator_index), foreign key (sentinel_id, node, lane, run_id) references monitor.runs on delete cascade)", + "create table if not exists monitor.artifact_locators (sentinel_id text not null, node text not null, lane text not null, run_id text not null, locator_index integer not null, owner_node text not null, owner_lane text not null, owner_pvc text not null, state_dir text not null, relative_path text not null, sha256 text not null, size_bytes bigint not null, kind text not null, reachable boolean not null, primary key (sentinel_id, node, lane, run_id, locator_index), foreign key (sentinel_id, node, lane, run_id) references monitor.runs on delete cascade)", "create table if not exists monitor.timeline_events (sentinel_id text not null, node text not null, lane text not null, run_id text not null, event_index integer not null, event jsonb not null, primary key (sentinel_id, node, lane, run_id, event_index), foreign key (sentinel_id, node, lane, run_id) references monitor.runs on delete cascade)", "create table if not exists monitor.import_manifests (manifest_id text primary key, source_fingerprint text not null, row_count bigint not null, imported_at timestamptz not null)", ]; export function createInMemoryMonitorCentralStore(): MonitorCentralStore { - const records = new Map>(); + const records = new Map(); return { async migrate() {}, async ping() {}, @@ -251,9 +277,9 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore { const runs = await this.runs(scope, { limit: 1, cursor: null }); return { scope: publicScope(scope), runCount: [...records.values()].filter((item) => matches(scope, item)).length, latest: runs[0] ?? null }; }, - async runs(scope, options) { return sortRuns([...records.values()].filter((item) => matches(scope, item)).map(runFrom)).filter((run) => afterCursor(run, options.cursor)).slice(0, options.limit); }, + async runs(scope, options) { return sortRuns([...records.values()].filter((item) => matches(scope, item)).map(runFrom)).filter((run) => afterRunCursor(run, options.cursor)).slice(0, options.limit); }, async run(scope, runId) { - const value = [...records.values()].find((item) => matches(scope, item) && item.runId === runId); + const value = records.get(identityKey({ ...scope, runId })); return value ? { ...runFrom(value), payload: value.payload, findings: value.findings, artifactLocators: value.artifactLocators, timeline: value.timeline } : null; }, async view(scope, runId, view) { @@ -263,69 +289,62 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore { return value === undefined ? null : { runId, name: view, value }; }, async findings(scope, options) { - return sortRuns([...records.values()].filter((item) => matches(scope, item)).map(runFrom)).filter((run) => afterCursor(run, options.cursor)).flatMap((run) => { + return sortRuns([...records.values()].filter((item) => matches(scope, item)).map(runFrom)).flatMap((run) => { const source = records.get(identityKey(run))!; - return source.findings.map((finding) => ({ ...run, finding })); - }).slice(0, options.limit); + return source.findings.map((finding, findingIndex) => ({ ...run, findingIndex, finding })); + }).sort(compareFindings).filter((finding) => afterFindingCursor(finding, options.cursor)).slice(0, options.limit); }, }; } -function requireDatabaseUrl(databaseUrl: string | null | undefined): string { - const value = databaseUrl ?? process.env.DATABASE_URL; - if (!value) throw new MonitorCentralConfigurationError(); - return value; +function requireDatabaseUrl(databaseUrl: string | null | undefined): string { const value = databaseUrl ?? process.env.DATABASE_URL; if (!value) throw new MonitorCentralConfigurationError(); return value; } +function requirePostgresOptions(value: PostgresMonitorCentralStoreOptions | undefined): PostgresMonitorCentralStoreOptions { if (!value) throw Object.assign(new Error("Monitor central PostgreSQL pool configuration must be injected"), { code: "monitor_central_pg_options_missing" }); return value; } +function requirePositiveInteger(value: number | undefined, code: string): number { if (!Number.isSafeInteger(value) || value < 1) throw Object.assign(new Error(`${code} must be a positive integer`), { code }); return value; } +function normalizeIngest(input: MonitorCentralIngest): NormalizedIngest { + for (const key of ["sentinelId", "node", "lane", "runId"] as const) if (!input[key]) throw invalid(`missing required ingest field: ${key}`); + const normalized = { ...input, updatedAt: input.updatedAt ?? new Date().toISOString(), status: input.status ?? "unknown", findings: input.findings ?? [], artifactLocators: (input.artifactLocators ?? []).map(normalizeLocator), timeline: input.timeline ?? [] }; + return { ...normalized, payloadHash: `sha256:${createHash("sha256").update(stableJson({ status: normalized.status, payload: normalized.payload, findings: normalized.findings, artifactLocators: normalized.artifactLocators, timeline: normalized.timeline })).digest("hex")}` }; } - -function normalizeIngest(input: MonitorCentralIngest): Required { - for (const key of ["sentinelId", "node", "lane", "runId"] as const) if (!input[key]) throw new Error(`missing required ingest field: ${key}`); - const payloadHash = input.payloadHash ?? `sha256:${createHash("sha256").update(stableJson(input.payload)).digest("hex")}`; - return { ...input, updatedAt: input.updatedAt ?? new Date().toISOString(), status: input.status ?? "unknown", payloadHash, findings: input.findings ?? [], artifactLocators: (input.artifactLocators ?? []).map(normalizeLocator), timeline: input.timeline ?? [] }; -} - function normalizeLocator(value: MonitorArtifactLocator): MonitorArtifactLocator { - if (!value.locator) throw new Error("artifact locator must include locator"); - return { locator: value.locator, ...(value.sha256 ? { sha256: value.sha256 } : {}), ...(value.owner ? { owner: value.owner } : {}), ...(Number.isFinite(value.sizeBytes) ? { sizeBytes: value.sizeBytes } : {}), ...(value.kind ? { kind: value.kind } : {}) }; + for (const key of ["ownerNode", "ownerLane", "ownerPvc", "stateDir", "relativePath", "sha256", "kind"] as const) if (!value[key]) throw invalid(`artifact locator must include ${key}`); + if (!Number.isSafeInteger(value.sizeBytes) || value.sizeBytes < 0) throw invalid("artifact locator sizeBytes must be a non-negative integer"); + if (typeof value.reachable !== "boolean") throw invalid("artifact locator reachable must be boolean"); + return { ...value }; } - +async function requestRecord(request: Request): Promise> { const value = await request.json(); if (!value || typeof value !== "object" || Array.isArray(value)) throw invalid("ingest request body must be an object"); return value as Record; } function scopeFrom(url: URL): MonitorScope { return { sentinelId: optional(url, "sentinelId"), node: optional(url, "node"), lane: optional(url, "lane") }; } +function requireFullScope(scope: MonitorScope): Required { if (!scope.sentinelId || !scope.node || !scope.lane) throw Object.assign(invalid("run detail requires sentinelId, node, and lane"), { code: "scope_required" }); return scope as Required; } function optional(url: URL, name: string): string | undefined { return url.searchParams.get(name) || undefined; } -function queryOptions(url: URL): QueryOptions { return { limit: boundedLimit(url.searchParams.get("limit")), cursor: parseCursor(url.searchParams.get("cursor")) }; } -function boundedLimit(value: string | null): number { const parsed = Number(value ?? 50); return Number.isInteger(parsed) && parsed > 0 ? Math.min(parsed, MAX_PAGE_SIZE) : 50; } -function parseCursor(value: string | null): Cursor | null { if (!value) return null; try { const decoded = JSON.parse(Buffer.from(value, "base64url").toString("utf8")); if (typeof decoded.updatedAt !== "string" || typeof decoded.runId !== "string") throw new Error("invalid cursor"); return decoded; } catch { throw new Error("invalid cursor"); } } -function page(items: readonly Record[]): Record { const last = items.at(-1); return { ok: true, items, nextCursor: last ? Buffer.from(JSON.stringify({ updatedAt: last.updatedAt, runId: last.runId })).toString("base64url") : null, valuesRedacted: true }; } +function boundedLimit(value: string | null, pageSize: number): number { const parsed = Number(value ?? pageSize); return Number.isInteger(parsed) && parsed > 0 ? Math.min(parsed, pageSize) : pageSize; } +function runQueryOptions(url: URL, pageSize: number): RunQueryOptions { return { limit: boundedLimit(url.searchParams.get("limit"), pageSize), cursor: parseRunCursor(url.searchParams.get("cursor")) }; } +function findingQueryOptions(url: URL, pageSize: number): FindingQueryOptions { return { limit: boundedLimit(url.searchParams.get("limit"), pageSize), cursor: parseFindingCursor(url.searchParams.get("cursor")) }; } +function parseRunCursor(value: string | null): RunCursor | null { return parseCursor(value, false) as RunCursor | null; } +function parseFindingCursor(value: string | null): FindingCursor | null { return parseCursor(value, true) as FindingCursor | null; } +function parseCursor(value: string | null, finding: boolean): RunCursor | FindingCursor | null { if (!value) return null; try { const decoded = JSON.parse(Buffer.from(value, "base64url").toString("utf8")); if (typeof decoded.updatedAt !== "string" || typeof decoded.runId !== "string" || (finding && !Number.isInteger(decoded.findingIndex))) throw new Error("invalid cursor"); return decoded; } catch { throw invalid("invalid cursor"); } } +function runPage(items: readonly MonitorRun[]): Record { const last = items.at(-1); return { ok: true, items, nextCursor: last ? cursor(last) : null, valuesRedacted: true }; } +function findingPage(items: readonly Record[]): Record { const last = items.at(-1); return { ok: true, items, nextCursor: last ? cursor({ updatedAt: last.updatedAt, runId: last.runId, findingIndex: last.findingIndex }) : null, valuesRedacted: true }; } +function cursor(value: unknown): string { return Buffer.from(JSON.stringify(value)).toString("base64url"); } function json(body: Record, status = 200): Response { return Response.json(body, { status }); } +function httpStatus(error: unknown): number { return (error as { code?: string })?.code === "payload_hash_conflict" ? 409 : error instanceof MonitorCentralConfigurationError ? 503 : 400; } function structuredError(error: unknown): Record { return error instanceof MonitorCentralConfigurationError ? error.toJSON().error as Record : { code: (error as { code?: string })?.code ?? "invalid_request", message: errorMessage(error) }; } +function invalid(message: string): Error & { code: string } { return Object.assign(new Error(message), { code: "invalid_request" }); } function errorMessage(error: unknown): string { return error instanceof Error ? error.message.replace(/postgres(?:ql)?:\/\/[^\s]+/giu, "postgres://") : "unknown error"; } function identityOf(value: MonitorScope & { readonly runId: string }): Record { return { sentinelId: value.sentinelId, node: value.node, lane: value.lane, runId: value.runId }; } function identityKey(value: MonitorScope & { readonly runId: string }): string { return [value.sentinelId, value.node, value.lane, value.runId].join("\u0000"); } -function conflict(value: MonitorCentralIngest): Error & { code: string } { return Object.assign(new Error("ingest identity already has a different payload hash"), { code: "payload_hash_conflict", identity: identityOf(value) }); } +function conflict(value: MonitorCentralIngest): Error & { code: string } { return Object.assign(new Error("ingest identity already has a different terminal envelope"), { code: "payload_hash_conflict", identity: identityOf(value) }); } function publicScope(scope: MonitorScope): Record { return { sentinelId: scope.sentinelId ?? null, node: scope.node ?? null, lane: scope.lane ?? null }; } function matches(scope: MonitorScope, value: MonitorScope): boolean { return (!scope.sentinelId || scope.sentinelId === value.sentinelId) && (!scope.node || scope.node === value.node) && (!scope.lane || scope.lane === value.lane); } -function runFrom(value: Required): MonitorRun { return { ...identityOf(value), runId: value.runId, updatedAt: value.updatedAt, status: value.status, payloadHash: value.payloadHash } as MonitorRun; } +function runFrom(value: NormalizedIngest): MonitorRun { return { sentinelId: value.sentinelId, node: value.node, lane: value.lane, runId: value.runId, updatedAt: value.updatedAt, status: value.status, payloadHash: value.payloadHash }; } function sortRuns(runs: readonly MonitorRun[]): MonitorRun[] { return [...runs].sort((left, right) => right.updatedAt.localeCompare(left.updatedAt) || right.runId.localeCompare(left.runId)); } -function afterCursor(run: MonitorRun, cursor: Cursor | null): boolean { return !cursor || run.updatedAt < cursor.updatedAt || (run.updatedAt === cursor.updatedAt && run.runId < cursor.runId); } -function stableJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`; if (value && typeof value === "object") return `{${Object.entries(value as Record).sort(([a], [b]) => a.localeCompare(b)).map(([key, item]) => `${JSON.stringify(key)}:${stableJson(item)}`).join(",")}}`; return JSON.stringify(value); } - -function whereClause(scope: MonitorScope, runId?: string): { readonly text: string; readonly values: readonly string[] } { - const entries = [...Object.entries({ sentinel_id: scope.sentinelId, node: scope.node, lane: scope.lane, run_id: runId }).filter(([, value]) => value)]; - return { text: entries.length ? entries.map(([key], index) => `r.${key} = $${index + 1}`).join(" and ") : "true", values: entries.map(([, value]) => String(value)) }; -} -async function queryRows(sql: postgres.Sql, scope: MonitorScope, options: QueryOptions): Promise { - const where = whereClause(scope); - const cursor = options.cursor ? `and (r.updated_at, r.run_id) < ($${where.values.length + 1}, $${where.values.length + 2})` : ""; - const values = options.cursor ? [...where.values, options.cursor.updatedAt, options.cursor.runId, options.limit] : [...where.values, options.limit]; - const rows = await sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash from monitor.runs r where ${where.text} ${cursor} order by r.updated_at desc, r.run_id desc limit $${values.length}`, values); - return rows.map(runRow); -} -async function countRuns(sql: postgres.Sql, scope: MonitorScope): Promise { const where = whereClause(scope); const rows = await sql.unsafe(`select count(*)::int as count from monitor.runs r where ${where.text}`, where.values); return Number(rows[0]?.count ?? 0); } -function rowIdentity(row: Record): { readonly where: { readonly text: string; readonly values: readonly string[] } } { - return { - where: { - text: "sentinel_id = $1 and node = $2 and lane = $3 and run_id = $4", - values: [String(row.sentinel_id), String(row.node), String(row.lane), String(row.run_id)], - }, - }; -} +function afterRunCursor(run: MonitorRun, cursorValue: RunCursor | null): boolean { return !cursorValue || run.updatedAt < cursorValue.updatedAt || (run.updatedAt === cursorValue.updatedAt && run.runId < cursorValue.runId); } +function compareFindings(left: Record, right: Record): number { return String(right.updatedAt).localeCompare(String(left.updatedAt)) || String(right.runId).localeCompare(String(left.runId)) || Number(right.findingIndex) - Number(left.findingIndex); } +function afterFindingCursor(finding: Record, cursorValue: FindingCursor | null): boolean { if (!cursorValue) return true; return String(finding.updatedAt) < cursorValue.updatedAt || (finding.updatedAt === cursorValue.updatedAt && (String(finding.runId) < cursorValue.runId || (finding.runId === cursorValue.runId && Number(finding.findingIndex) < cursorValue.findingIndex))); } +function stableJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`; if (value && typeof value === "object") return `{${Object.entries(value as Record).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => `${JSON.stringify(key)}:${stableJson(item)}`).join(",")}}`; return JSON.stringify(value); } +function scopeWhere(scope: MonitorScope): { readonly text: string; readonly values: readonly string[] } { const entries = Object.entries({ sentinel_id: scope.sentinelId, node: scope.node, lane: scope.lane }).filter(([, value]) => value); return { text: entries.length ? entries.map(([key], index) => `r.${key} = $${index + 1}`).join(" and ") : "true", values: entries.map(([, value]) => String(value)) }; } +function identityWhere(scope: Required, runId: string): { readonly text: string; readonly values: readonly string[] } { return { text: "r.sentinel_id = $1 and r.node = $2 and r.lane = $3 and r.run_id = $4", values: [scope.sentinelId, scope.node, scope.lane, runId] }; } +async function queryRuns(sql: postgres.Sql, scope: MonitorScope, options: RunQueryOptions): Promise { const where = scopeWhere(scope); const values: unknown[] = [...where.values]; let cursorClause = ""; if (options.cursor) { values.push(options.cursor.updatedAt, options.cursor.runId); cursorClause = `and (r.updated_at, r.run_id) < ($${values.length - 1}, $${values.length})`; } values.push(options.limit); const rows = await sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash from monitor.runs r where ${where.text} ${cursorClause} order by r.updated_at desc, r.run_id desc limit $${values.length}`, values); return rows.map(runRow); } +async function countRuns(sql: postgres.Sql, scope: MonitorScope): Promise { const where = scopeWhere(scope); const rows = await sql.unsafe(`select count(*)::int as count from monitor.runs r where ${where.text}`, where.values); return Number(rows[0]?.count ?? 0); } +function rowIdentity(row: Record): { readonly text: string; readonly values: readonly string[] } { return { text: "sentinel_id = $1 and node = $2 and lane = $3 and run_id = $4", values: [String(row.sentinel_id), String(row.node), String(row.lane), String(row.run_id)] }; } function runRow(row: Record): MonitorRun { return { sentinelId: String(row.sentinel_id), node: String(row.node), lane: String(row.lane), runId: String(row.run_id), updatedAt: new Date(String(row.updated_at)).toISOString(), status: String(row.status), payloadHash: String(row.payload_hash) }; } -function locatorRow(row: Record): MonitorArtifactLocator { return { locator: String(row.locator), ...(row.sha256 ? { sha256: String(row.sha256) } : {}), ...(row.owner_name ? { owner: String(row.owner_name) } : {}), ...(row.size_bytes !== null ? { sizeBytes: Number(row.size_bytes) } : {}), ...(row.kind ? { kind: String(row.kind) } : {}) }; } +export function mapPostgresFindingRow(row: Record): Record { return { ...runRow(row), findingIndex: Number(row.finding_index), finding: row.finding }; } +function locatorRow(row: Record): MonitorArtifactLocator { return { ownerNode: String(row.owner_node), ownerLane: String(row.owner_lane), ownerPvc: String(row.owner_pvc), stateDir: String(row.state_dir), relativePath: String(row.relative_path), sha256: String(row.sha256), sizeBytes: Number(row.size_bytes), kind: String(row.kind), reachable: row.reachable === true }; }