mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: complete monitor ingest migration contracts

Browse Source
This commit is contained in:
AgentRun Codex
2026-07-12 21:06:08 +00:00
parent 8017852670
commit 35a59b27c6
9 changed files with 209 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/monitor-central-persistence.ts
+9 -9
View File
@@ -50,7 +50,7 @@ export interface MonitorCentralStore {
ping(): Promise<void>;
schemaVersion(): Promise<string | null>;
capability(): Promise<{ readonly ingest: boolean; readonly query: boolean }>;
ingest(input: NormalizedIngest): Promise<{ readonly disposition: "inserted" | "idempotent" }>;
ingest(input: NormalizedMonitorCentralIngest): Promise<{ readonly disposition: "inserted" | "idempotent" }>;
updateArtifactAvailability(scope: Required<MonitorScope>, runId: string, locatorIndex: number, reachable: boolean): Promise<void>;
overview(scope: MonitorScope): Promise<Record<string, unknown>>;
runs(scope: MonitorScope, options: RunQueryOptions): Promise<readonly MonitorRun[]>;
@@ -59,7 +59,7 @@ export interface MonitorCentralStore {
findings(scope: MonitorScope, options: FindingQueryOptions): Promise<readonly Record<string, unknown>[]>;
}
interface NormalizedIngest extends Required<Omit<MonitorCentralIngest, "payloadHash">> {
export interface NormalizedMonitorCentralIngest extends Required<Omit<MonitorCentralIngest, "payloadHash">> {
readonly payloadHash: string;
}
@@ -117,7 +117,7 @@ export function createMonitorCentralService(options: MonitorCentralServiceOption
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 normalized = normalizeMonitorCentralIngest(input);
const result = await store.ingest(normalized);
return { ok: true, ...result, identity: identityOf(normalized), payloadHash: normalized.payloadHash, valuesRedacted: true };
};
@@ -280,7 +280,7 @@ export const MONITOR_CENTRAL_MIGRATIONS = [
];
export function createInMemoryMonitorCentralStore(): MonitorCentralStore {
const records = new Map<string, NormalizedIngest>();
const records = new Map<string, NormalizedMonitorCentralIngest>();
return {
async migrate() {},
async ping() {},
@@ -331,10 +331,10 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore {
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 {
export function normalizeMonitorCentralIngest(input: MonitorCentralIngest): NormalizedMonitorCentralIngest {
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.map(immutableLocator), timeline: normalized.timeline })).digest("hex")}` };
return { ...normalized, payloadHash: `sha256:${createHash("sha256").update(monitorCentralStableJson({ status: normalized.status, payload: normalized.payload, findings: normalized.findings, artifactLocators: normalized.artifactLocators.map(immutableLocator), timeline: normalized.timeline })).digest("hex")}` };
}
function immutableLocator(locator: MonitorArtifactLocator): Omit<MonitorArtifactLocator, "reachable"> { const { reachable: _reachable, ...immutable } = locator; return immutable; }
function normalizeLocator(value: MonitorArtifactLocator): MonitorArtifactLocator {
@@ -367,17 +367,17 @@ function identityKey(value: MonitorScope & { readonly runId: string }): string {
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<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: 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 runFrom(value: NormalizedMonitorCentralIngest): 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 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<string, unknown>, right: Record<string, unknown>): 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<string, unknown>, 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<string, unknown>).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => `${JSON.stringify(key)}:${stableJson(item)}`).join(",")}}`; return JSON.stringify(value); }
export function monitorCentralStableJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(monitorCentralStableJson).join(",")}]`; if (value && typeof value === "object") return `{${Object.entries(value as Record<string, unknown>).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => `${JSON.stringify(key)}:${monitorCentralStableJson(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<MonitorScope>, 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<MonitorRun[]> { 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<number> { 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); }
async function updatePostgresArtifactAvailability(transaction: postgres.TransactionSql, input: NormalizedIngest): Promise<void> {
async function updatePostgresArtifactAvailability(transaction: postgres.TransactionSql, input: NormalizedMonitorCentralIngest): Promise<void> {
for (const [index, locator] of input.artifactLocators.entries()) await transaction`
update monitor.artifact_locators
set reachable = ${locator.reachable}