feat(monitor): add central PostgreSQL persistence API
This commit is contained in:
AgentRun Codex
@@ -0,0 +1,15 @@
|
||||
// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
|
||||
// Responsibility: Explicit DATABASE_URL 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 }));
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
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";
|
||||
|
||||
function service() { return createMonitorCentralService({ store: createInMemoryMonitorCentralStore() }); }
|
||||
async function ingest(target: ReturnType<typeof service>, runId: string, updatedAt: string, scope: Record<string, string> = {}) {
|
||||
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" }] });
|
||||
}
|
||||
|
||||
test("schema covers central persistence objects and does not store artifact blobs", () => {
|
||||
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);
|
||||
});
|
||||
|
||||
test("ingest is transactional in contract, idempotent by payload hash, and rejects conflicts", 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" });
|
||||
});
|
||||
|
||||
test("global and scoped latest use updatedAt then runId descending", async () => {
|
||||
const target = service();
|
||||
await ingest(target, "run-a", "2026-07-12T10:00:00.000Z");
|
||||
await ingest(target, "run-b", "2026-07-12T10:00:00.000Z");
|
||||
await ingest(target, "run-c", "2026-07-12T09:00:00.000Z", { sentinelId: "sentinel-b" });
|
||||
const overview = await (await target.fetch(new Request("http://local/api/overview"))).json() as any;
|
||||
const scoped = await (await target.fetch(new Request("http://local/api/overview?sentinelId=sentinel-b"))).json() as any;
|
||||
assert.equal(overview.overview.latest.runId, "run-b");
|
||||
assert.equal(scoped.overview.latest.runId, "run-c");
|
||||
});
|
||||
|
||||
test("runs and findings are bounded, cursor-paged, and scope-isolated", 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);
|
||||
});
|
||||
|
||||
test("run detail returns payload, locator metadata, timeline, and no secret values", 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" });
|
||||
assert.equal(response.valuesRedacted, true);
|
||||
});
|
||||
|
||||
test("health is redacted and missing DATABASE_URL 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", () => {
|
||||
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 });
|
||||
});
|
||||
@@ -0,0 +1,331 @@
|
||||
// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
|
||||
// Responsibility: Host PostgreSQL-backed Monitor 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 interface MonitorScope {
|
||||
readonly sentinelId?: string;
|
||||
readonly node?: string;
|
||||
readonly lane?: string;
|
||||
}
|
||||
|
||||
export interface MonitorArtifactLocator {
|
||||
readonly locator: string;
|
||||
readonly sha256?: string;
|
||||
readonly owner?: string;
|
||||
readonly sizeBytes?: number;
|
||||
readonly kind?: string;
|
||||
}
|
||||
|
||||
export interface MonitorCentralIngest {
|
||||
readonly sentinelId: string;
|
||||
readonly node: string;
|
||||
readonly lane: string;
|
||||
readonly runId: string;
|
||||
readonly updatedAt?: string;
|
||||
readonly status?: string;
|
||||
readonly payload: Record<string, unknown>;
|
||||
readonly payloadHash?: string;
|
||||
readonly findings?: readonly Record<string, unknown>[];
|
||||
readonly artifactLocators?: readonly MonitorArtifactLocator[];
|
||||
readonly timeline?: readonly Record<string, unknown>[];
|
||||
}
|
||||
|
||||
export interface MonitorRun extends MonitorScope {
|
||||
readonly runId: string;
|
||||
readonly updatedAt: string;
|
||||
readonly status: string;
|
||||
readonly payloadHash: string;
|
||||
}
|
||||
|
||||
export interface MonitorCentralStore {
|
||||
migrate(): Promise<void>;
|
||||
ping(): Promise<void>;
|
||||
schemaVersion(): Promise<string | null>;
|
||||
ingest(input: Required<MonitorCentralIngest>): Promise<{ readonly disposition: "inserted" | "idempotent" }>;
|
||||
overview(scope: MonitorScope): Promise<Record<string, unknown>>;
|
||||
runs(scope: MonitorScope, options: QueryOptions): Promise<readonly MonitorRun[]>;
|
||||
run(scope: MonitorScope, runId: string): Promise<Record<string, unknown> | null>;
|
||||
view(scope: MonitorScope, runId: string, view: string): Promise<Record<string, unknown> | null>;
|
||||
findings(scope: MonitorScope, options: QueryOptions): Promise<readonly Record<string, unknown>[]>;
|
||||
}
|
||||
|
||||
export interface QueryOptions {
|
||||
readonly limit: number;
|
||||
readonly cursor: Cursor | null;
|
||||
}
|
||||
|
||||
interface Cursor {
|
||||
readonly updatedAt: string;
|
||||
readonly runId: string;
|
||||
}
|
||||
|
||||
export interface MonitorCentralService {
|
||||
ingest(input: MonitorCentralIngest): Promise<Record<string, unknown>>;
|
||||
health(): Promise<Record<string, unknown>>;
|
||||
fetch(request: Request): Promise<Response>;
|
||||
}
|
||||
|
||||
export interface MonitorCentralServiceOptions {
|
||||
readonly databaseUrl?: string | null;
|
||||
readonly store?: MonitorCentralStore;
|
||||
}
|
||||
|
||||
export class MonitorCentralConfigurationError extends Error {
|
||||
readonly code = "monitor_central_database_url_missing";
|
||||
|
||||
constructor() {
|
||||
super("Monitor central service requires an explicit DATABASE_URL");
|
||||
this.name = "MonitorCentralConfigurationError";
|
||||
}
|
||||
|
||||
toJSON(): Record<string, unknown> {
|
||||
return { ok: false, error: { code: this.code, message: this.message }, valuesRedacted: true };
|
||||
}
|
||||
}
|
||||
|
||||
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<string, unknown>[] = [{ 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);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function createPostgresMonitorCentralStore(databaseUrl: string): MonitorCentralStore {
|
||||
const sql = postgres(databaseUrl, { max: 4, idle_timeout: 10 });
|
||||
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`;
|
||||
},
|
||||
async ping() { await sql`select 1`; },
|
||||
async schemaVersion() {
|
||||
const rows = await sql<{ version: string }[]>`select version from monitor.schema_migrations order by applied_at desc limit 1`;
|
||||
return rows[0]?.version ?? null;
|
||||
},
|
||||
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`
|
||||
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})`;
|
||||
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)`;
|
||||
for (const [index, finding] of input.findings.entries()) await transaction`
|
||||
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})`;
|
||||
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)`;
|
||||
return { disposition: "inserted" as const };
|
||||
});
|
||||
},
|
||||
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 };
|
||||
},
|
||||
async runs(scope, options) { return queryRows(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 row = rows[0] as Record<string, unknown> | 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),
|
||||
]);
|
||||
return { ...runRow(row), payload: row.payload, findings: findings.map((item) => item.finding), artifactLocators: locators.map(locatorRow), timeline: timeline.map((item) => item.event) };
|
||||
},
|
||||
async view(scope, runId, view) {
|
||||
const detail = await this.run(scope, runId);
|
||||
const views = detail?.payload && typeof detail.payload === "object" ? (detail.payload as Record<string, unknown>).views : null;
|
||||
const value = views && typeof views === "object" ? (views as Record<string, unknown>)[view] : undefined;
|
||||
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 }));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
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 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.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<string, Required<MonitorCentralIngest>>();
|
||||
return {
|
||||
async migrate() {},
|
||||
async ping() {},
|
||||
async schemaVersion() { return MONITOR_CENTRAL_SCHEMA_VERSION; },
|
||||
async ingest(input) {
|
||||
const key = identityKey(input);
|
||||
const existing = records.get(key);
|
||||
if (existing) {
|
||||
if (existing.payloadHash !== input.payloadHash) throw conflict(input);
|
||||
return { disposition: "idempotent" };
|
||||
}
|
||||
records.set(key, structuredClone(input));
|
||||
return { disposition: "inserted" };
|
||||
},
|
||||
async overview(scope) {
|
||||
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 run(scope, runId) {
|
||||
const value = [...records.values()].find((item) => matches(scope, item) && item.runId === runId);
|
||||
return value ? { ...runFrom(value), payload: value.payload, findings: value.findings, artifactLocators: value.artifactLocators, timeline: value.timeline } : null;
|
||||
},
|
||||
async view(scope, runId, view) {
|
||||
const detail = await this.run(scope, runId);
|
||||
const views = detail?.payload && typeof detail.payload === "object" ? (detail.payload as Record<string, unknown>).views : null;
|
||||
const value = views && typeof views === "object" ? (views as Record<string, unknown>)[view] : undefined;
|
||||
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) => {
|
||||
const source = records.get(identityKey(run))!;
|
||||
return source.findings.map((finding) => ({ ...run, finding }));
|
||||
}).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 normalizeIngest(input: MonitorCentralIngest): Required<MonitorCentralIngest> {
|
||||
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 } : {}) };
|
||||
}
|
||||
|
||||
function scopeFrom(url: URL): MonitorScope { return { sentinelId: optional(url, "sentinelId"), node: optional(url, "node"), lane: optional(url, "lane") }; }
|
||||
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<string, unknown>[]): Record<string, unknown> { 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 json(body: Record<string, unknown>, status = 200): Response { return Response.json(body, { status }); }
|
||||
function structuredError(error: unknown): Record<string, unknown> { return error instanceof MonitorCentralConfigurationError ? error.toJSON().error as Record<string, unknown> : { code: (error as { code?: string })?.code ?? "invalid_request", message: errorMessage(error) }; }
|
||||
function errorMessage(error: unknown): string { return error instanceof Error ? error.message.replace(/postgres(?:ql)?:\/\/[^\s]+/giu, "postgres://<redacted>") : "unknown error"; }
|
||||
function identityOf(value: MonitorScope & { readonly runId: string }): Record<string, unknown> { 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 publicScope(scope: MonitorScope): Record<string, unknown> { 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<MonitorCentralIngest>): MonitorRun { return { ...identityOf(value), runId: value.runId, updatedAt: value.updatedAt, status: value.status, payloadHash: value.payloadHash } as MonitorRun; }
|
||||
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<string, unknown>).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<MonitorRun[]> {
|
||||
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<number> { 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<string, unknown>): { 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 runRow(row: Record<string, unknown>): 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<string, unknown>): 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) } : {}) }; }
|
||||
Reference in New Issue
Block a user