From e241af61bae6fbb78edd9339898c17f5836112ed Mon Sep 17 00:00:00 2001 From: oktofeesh1 <287075021+oktofeesh1@users.noreply.github.com> Date: Wed, 1 Jul 2026 23:09:08 -0700 Subject: [PATCH] feat(log): add self-host audit PR correlation Emit safe repo and PR-number correlation fields on self-host job audit lines so operators can grep queue lifecycle logs for a specific pull request without logging full payloads. Closes #2087 --- src/selfhost/audit.ts | 100 ++++++++++++++++++++ src/selfhost/pg-queue.ts | 7 +- src/selfhost/sqlite-queue.ts | 7 +- test/unit/selfhost-audit.test.ts | 117 +++++++++++++++++++++++- test/unit/selfhost-pg-queue.test.ts | 6 ++ test/unit/selfhost-sqlite-queue.test.ts | 2 + 6 files changed, 236 insertions(+), 3 deletions(-) diff --git a/src/selfhost/audit.ts b/src/selfhost/audit.ts index fa800188af..f2bb1b2833 100644 --- a/src/selfhost/audit.ts +++ b/src/selfhost/audit.ts @@ -16,12 +16,19 @@ export interface AuditEvent { ts: number; // Unix timestamp (ms) job_id: number | string; payload_type?: string | undefined; // top-level `type` field from the job payload, if present + repo?: string | undefined; + pr_number?: number | undefined; latency_ms: number; // wall time from claim to completion/failure attempts: number; // total attempts consumed (1 = first-try success) error?: string; // last error message, present for job_dead / job_error retry_after_ms?: number; // next retry delay for job_rate_limited } +export interface AuditPayloadContext { + repo?: string | undefined; + pr_number?: number | undefined; +} + /** Emit a single audit event as a JSON line on stdout. */ export function logAudit(ev: AuditEvent, traceParent?: string): void { process.stdout.write(JSON.stringify({ level: "audit", ...ev, ...otelTraceLogFields(traceParent) }) + "\n"); @@ -37,3 +44,96 @@ export function extractPayloadType(payload: string): string | undefined { return undefined; } } + +/** Extract repo / PR correlation labels from a raw job payload. Only safe scalar fields are returned. */ +export function extractPayloadContext(payload: string): AuditPayloadContext | undefined { + let parsed: unknown; + try { + parsed = JSON.parse(payload) as unknown; + } catch { + return undefined; + } + if (!isRecord(parsed)) return undefined; + + const repo = extractRepo(parsed); + const prNumber = extractPrNumber(parsed); + if (repo === undefined && prNumber === undefined) return undefined; + + const context: AuditPayloadContext = {}; + if (repo !== undefined) context.repo = repo; + if (prNumber !== undefined) context.pr_number = prNumber; + return context; +} + +function extractRepo(payload: Record): string | undefined { + const repoFullName = stringField(payload, "repoFullName"); + if (repoFullName !== undefined) return repoFullName; + + const webhookPayload = recordField(payload, "payload"); + if (webhookPayload === undefined) return undefined; + + const repository = recordField(webhookPayload, "repository"); + if (repository === undefined) return undefined; + return stringField(repository, "full_name"); +} + +function extractPrNumber(payload: Record): number | undefined { + const prNumber = numberField(payload, "prNumber"); + if (prNumber !== undefined) return prNumber; + + const webhookPayload = recordField(payload, "payload"); + if (webhookPayload === undefined) return undefined; + + const pullRequest = recordField(webhookPayload, "pull_request"); + if (pullRequest !== undefined) { + const pullRequestNumber = numberField(pullRequest, "number"); + if (pullRequestNumber !== undefined) return pullRequestNumber; + } + + const issue = recordField(webhookPayload, "issue"); + if (issue !== undefined) { + const issueNumber = numberField(issue, "number"); + if (issueNumber !== undefined && recordField(issue, "pull_request") !== undefined) return issueNumber; + } + + const checkRun = recordField(webhookPayload, "check_run"); + if (checkRun !== undefined) { + const checkRunNumber = firstPullRequestNumber(checkRun); + if (checkRunNumber !== undefined) return checkRunNumber; + } + + const checkSuite = recordField(webhookPayload, "check_suite"); + if (checkSuite !== undefined) return firstPullRequestNumber(checkSuite); + + return undefined; +} + +function firstPullRequestNumber(record: Record): number | undefined { + const pullRequests = record.pull_requests; + if (!Array.isArray(pullRequests)) return undefined; + for (const pullRequest of pullRequests) { + if (!isRecord(pullRequest)) continue; + const number = numberField(pullRequest, "number"); + if (number !== undefined) return number; + } + return undefined; +} + +function recordField(record: Record, field: string): Record | undefined { + const value = record[field]; + return isRecord(value) ? value : undefined; +} + +function stringField(record: Record, field: string): string | undefined { + const value = record[field]; + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function numberField(record: Record, field: string): number | undefined { + const value = record[field]; + return typeof value === "number" && Number.isFinite(value) ? value : undefined; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index ccea50ffc6..d4d00a6924 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -3,7 +3,7 @@ // app instances sharing one Postgres can claim jobs concurrently without double-processing. size()/deadCount() // are async (the metrics gauges accept async samplers). import type { Pool } from "pg"; -import { logAudit, extractPayloadType } from "./audit"; +import { logAudit, extractPayloadType, extractPayloadContext } from "./audit"; import { incr } from "./metrics"; import { withReviewSpan } from "./tracing"; import { withOtelSpan } from "./otel"; @@ -385,6 +385,7 @@ export function createPgQueue( return true; } const jobTraceParent = message.type === "github-webhook" ? message.traceParent : undefined; + const payloadContext = extractPayloadContext(job.payload); const rateLimitAdmission = await rateLimitAdmissionDelayMs(message); if (rateLimitAdmission !== null) { const rateLimitMetric = githubRateLimitMetricContext(message, rateLimitAdmission); @@ -437,6 +438,7 @@ export function createPgQueue( ts: Date.now(), job_id: job.id, payload_type: extractPayloadType(job.payload), + ...payloadContext, latency_ms: Date.now() - claimedAt, attempts: Number(job.attempts) + 1, }, jobTraceParent); @@ -477,6 +479,7 @@ export function createPgQueue( ts: Date.now(), job_id: job.id, payload_type: extractPayloadType(job.payload), + ...payloadContext, latency_ms: Date.now() - claimedAt, attempts, retry_after_ms: Math.max(0, retryAfter - Date.now()), @@ -505,6 +508,7 @@ export function createPgQueue( ts: Date.now(), job_id: job.id, payload_type: extractPayloadType(job.payload), + ...payloadContext, latency_ms: Date.now() - claimedAt, attempts, error: errMsg, @@ -527,6 +531,7 @@ export function createPgQueue( ts: Date.now(), job_id: job.id, payload_type: extractPayloadType(job.payload), + ...payloadContext, latency_ms: Date.now() - claimedAt, attempts, error: errMsg, diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 13611086b2..48cd1b35c6 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -4,7 +4,7 @@ // backing store differs. Single-process model: node:sqlite is synchronous + serial, so claim (SELECT→UPDATE) // is atomic with no row-lock dance. import type { SqliteDriver } from "./d1-adapter"; -import { logAudit, extractPayloadType } from "./audit"; +import { logAudit, extractPayloadType, extractPayloadContext } from "./audit"; import { incr } from "./metrics"; import { withReviewSpan } from "./tracing"; import { withOtelSpan } from "./otel"; @@ -325,6 +325,7 @@ export function createSqliteQueue( return true; } const jobTraceParent = message.type === "github-webhook" ? message.traceParent : undefined; + const payloadContext = extractPayloadContext(job.payload); const rateLimitAdmission = rateLimitAdmissionDelayMs(driver, message); if (rateLimitAdmission !== null) { const rateLimitMetric = githubRateLimitMetricContext(message, rateLimitAdmission); @@ -377,6 +378,7 @@ export function createSqliteQueue( ts: Date.now(), job_id: job.id, payload_type: extractPayloadType(job.payload), + ...payloadContext, latency_ms: Date.now() - claimedAt, attempts: job.attempts + 1, }, jobTraceParent); @@ -417,6 +419,7 @@ export function createSqliteQueue( ts: Date.now(), job_id: job.id, payload_type: extractPayloadType(job.payload), + ...payloadContext, latency_ms: Date.now() - claimedAt, attempts, retry_after_ms: Math.max(0, retryAfter - Date.now()), @@ -445,6 +448,7 @@ export function createSqliteQueue( ts: Date.now(), job_id: job.id, payload_type: extractPayloadType(job.payload), + ...payloadContext, latency_ms: Date.now() - claimedAt, attempts, error: errMsg, @@ -467,6 +471,7 @@ export function createSqliteQueue( ts: Date.now(), job_id: job.id, payload_type: extractPayloadType(job.payload), + ...payloadContext, latency_ms: Date.now() - claimedAt, attempts, error: errMsg, diff --git a/test/unit/selfhost-audit.test.ts b/test/unit/selfhost-audit.test.ts index 032fc96288..24bf3bc383 100644 --- a/test/unit/selfhost-audit.test.ts +++ b/test/unit/selfhost-audit.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { logAudit, extractPayloadType } from "../../src/selfhost/audit"; +import { logAudit, extractPayloadContext, extractPayloadType } from "../../src/selfhost/audit"; describe("logAudit", () => { const written: string[] = []; @@ -78,3 +78,118 @@ describe("extractPayloadType", () => { expect(extractPayloadType("{}")).toBeUndefined(); }); }); + +describe("extractPayloadContext", () => { + it("returns top-level repo and PR fields from internal PR jobs", () => { + expect(extractPayloadContext(JSON.stringify({ type: "agent-regate-pr", repoFullName: "JSONbored/gittensory", prNumber: 2087 }))).toEqual({ + repo: "JSONbored/gittensory", + pr_number: 2087, + }); + }); + + it("returns repo-only context for repo-scoped background jobs", () => { + expect(extractPayloadContext(JSON.stringify({ type: "rag-index-repo", repoFullName: "JSONbored/gittensory" }))).toEqual({ + repo: "JSONbored/gittensory", + }); + }); + + it("returns PR-only context when a webhook payload lacks repository data", () => { + expect(extractPayloadContext(JSON.stringify({ type: "github-webhook", payload: { pull_request: { number: 87 } } }))).toEqual({ + pr_number: 87, + }); + }); + + it("returns nested repository and pull request fields from GitHub webhook jobs", () => { + expect( + extractPayloadContext(JSON.stringify({ + type: "github-webhook", + payload: { + repository: { full_name: "JSONbored/gittensory" }, + pull_request: { number: 2087 }, + }, + })), + ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2087 }); + }); + + it("falls back from wrong top-level fields to nested webhook context", () => { + expect( + extractPayloadContext(JSON.stringify({ + type: "github-webhook", + repoFullName: 123, + prNumber: "2087", + payload: { + repository: { full_name: "JSONbored/gittensory" }, + pull_request: { number: 2087 }, + }, + })), + ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2087 }); + }); + + it("extracts PR numbers from PR issue-comment payloads without treating regular issues as PRs", () => { + expect( + extractPayloadContext(JSON.stringify({ + type: "github-webhook", + payload: { + repository: { full_name: "JSONbored/gittensory" }, + issue: { number: 2087, pull_request: {} }, + }, + })), + ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2087 }); + expect( + extractPayloadContext(JSON.stringify({ + type: "github-webhook", + payload: { + repository: { full_name: "JSONbored/gittensory" }, + issue: { number: 2087 }, + }, + })), + ).toEqual({ repo: "JSONbored/gittensory" }); + }); + + it("extracts PR numbers from check-run and check-suite pull request arrays", () => { + expect( + extractPayloadContext(JSON.stringify({ + type: "github-webhook", + payload: { + repository: { full_name: "JSONbored/gittensory" }, + check_run: { pull_requests: [null, { number: "bad" }, { number: 2087 }] }, + }, + })), + ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2087 }); + expect( + extractPayloadContext(JSON.stringify({ + type: "github-webhook", + payload: { + repository: { full_name: "JSONbored/gittensory" }, + check_suite: { pull_requests: [{ number: 2088 }] }, + }, + })), + ).toEqual({ repo: "JSONbored/gittensory", pr_number: 2088 }); + }); + + it("returns undefined for missing, malformed, and non-object payloads", () => { + expect(extractPayloadContext(JSON.stringify({ type: "refresh-registry" }))).toBeUndefined(); + expect(extractPayloadContext("not-json")).toBeUndefined(); + expect(extractPayloadContext("null")).toBeUndefined(); + expect(extractPayloadContext("[]")).toBeUndefined(); + expect(extractPayloadContext('"string"')).toBeUndefined(); + }); + + it("returns undefined for wrong typed and non-finite context fields", () => { + expect( + extractPayloadContext(JSON.stringify({ + type: "github-webhook", + repoFullName: "", + prNumber: "2087", + payload: { + repository: { full_name: "" }, + pull_request: { number: "2087" }, + issue: { number: "2087", pull_request: {} }, + check_run: { pull_requests: "bad" }, + check_suite: { pull_requests: [] }, + }, + })), + ).toBeUndefined(); + expect(extractPayloadContext('{"prNumber":1e999}')).toBeUndefined(); + }); +}); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 71e8c04fe3..f08424d3fe 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -508,6 +508,10 @@ describe("createPgQueue (durable #977)", () => { m.enqueueJob("1", { type: "github-webhook", traceParent: "00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-bbbbbbbbbbbbbbbb-01", + payload: { + repository: { full_name: "JSONbored/gittensory" }, + pull_request: { number: 1629 }, + }, }); const q = createPgQueue(m.pool, async () => undefined); @@ -516,6 +520,8 @@ describe("createPgQueue (durable #977)", () => { const audit = writes.find((line) => line.includes('"event":"job_complete"')); expect(JSON.parse(audit!) as Record).toMatchObject({ + repo: "JSONbored/gittensory", + pr_number: 1629, trace_id: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", }); }); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 3839ac8214..5790101c58 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -98,6 +98,8 @@ describe("createSqliteQueue (durable #980)", () => { const audit = writes.find((line) => line.includes('"event":"job_complete"')); expect(JSON.parse(audit!) as Record).toMatchObject({ + repo: "JSONbored/gittensory", + pr_number: 1629, trace_id: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", }); });