fix: close monitor ingest review gaps

scripts/src/monitor-central-persistence.ts

@@ -46,6 +46,7 @@ export interface MonitorRun extends Required<MonitorScope> {
 }
 
 export interface MonitorCentralStore {
+  close(): Promise<void>;
   migrate(): Promise<void>;
   ping(): Promise<void>;
   schemaVersion(): Promise<string | null>;
@@ -176,6 +177,7 @@ export function createMonitorCentralService(options: MonitorCentralServiceOption
 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 close() { await sql.end({ timeout: 0 }); },
     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`;
@@ -197,9 +199,13 @@ export function createPostgresMonitorCentralStore(databaseUrl: string, options:
     },
     async importSource(manifest, inputs) {
       return sql.begin(async (transaction) => {
-        const existing = await transaction<{ source_fingerprint: string; row_count: string | number }[]>`
-          select source_fingerprint, row_count from monitor.import_manifests where manifest_id = ${manifest.manifestId} for update`;
-        if (existing[0]) {
+        const claimed = await transaction<{ manifest_id: string }[]>`
+          insert into monitor.import_manifests (manifest_id, source_fingerprint, row_count, imported_at)
+          values (${manifest.manifestId}, ${manifest.sourceFingerprint}, ${manifest.rowCount}, now())
+          on conflict (manifest_id) do nothing returning manifest_id`;
+        if (!claimed[0]) {
+          const existing = await transaction<{ source_fingerprint: string; row_count: string | number }[]>`
+            select source_fingerprint, row_count from monitor.import_manifests where manifest_id = ${manifest.manifestId}`;
           if (existing[0].source_fingerprint !== manifest.sourceFingerprint || Number(existing[0].row_count) !== manifest.rowCount) {
             throw Object.assign(new Error("monitor import manifest conflict"), { code: "monitor_import_manifest_conflict" });
           }
@@ -207,9 +213,6 @@ export function createPostgresMonitorCentralStore(databaseUrl: string, options:
         }
         const runs = [] as { disposition: "inserted" | "idempotent" }[];
         for (const input of inputs) runs.push(await writePostgresIngest(transaction, input));
-        await transaction`
-          insert into monitor.import_manifests (manifest_id, source_fingerprint, row_count, imported_at)
-          values (${manifest.manifestId}, ${manifest.sourceFingerprint}, ${manifest.rowCount}, now())`;
         return { disposition: "inserted" as const, runs };
       });
     },
@@ -307,6 +310,7 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore {
   const records = new Map<string, NormalizedMonitorCentralIngest>();
   const manifests = new Map<string, MonitorImportManifest>();
   return {
+    async close() {},
     async migrate() {},
     async ping() {},
     async schemaVersion() { return MONITOR_CENTRAL_SCHEMA_VERSION; },
@@ -361,7 +365,7 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore {
       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 };
+      return value === undefined || !detail ? null : { run: runFrom(detail as MonitorRun), view, value };
     },
     async findings(scope, options) {
       return sortRuns([...records.values()].filter((item) => matches(scope, item)).map(runFrom)).flatMap((run) => {
@@ -416,7 +420,7 @@ function sortRuns(runs: readonly MonitorRun[]): MonitorRun[] { return [...runs].
 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))); }
-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); }
+export function monitorCentralStableJson(value: unknown): string { if (value === undefined) return "null"; if (Array.isArray(value)) return `[${value.map(monitorCentralStableJson).join(",")}]`; if (value && typeof value === "object") return `{${Object.entries(value as Record<string, unknown>).filter(([, item]) => item !== undefined).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => `${JSON.stringify(key)}:${monitorCentralStableJson(item)}`).join(",")}}`; const json = JSON.stringify(value); return json === undefined ? "null" : json; }
 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); }