From 4422a7974721caab13aa368848c11a4dd84fa99d Mon Sep 17 00:00:00 2001 From: Drewry Pope Date: Sat, 31 Oct 2026 11:09:04 -0500 Subject: [PATCH] fix(projector): scope comment identity by kind --- pm-stack/projector/README.md | 20 ++++++++++++ pm-stack/projector/schema.sql | 7 +++-- pm-stack/projector/src/projector.ts | 6 ++-- pm-stack/projector/src/server.ts | 2 +- pm-stack/projector/src/store.ts | 15 ++++++--- pm-stack/projector/test/e2e.test.ts | 37 +++++++++++++++++++++++ pm-stack/projector/test/projector.test.ts | 13 +++++--- pm-stack/projector/test/store.test.ts | 33 ++++++++++++++++++++ 8 files changed, 118 insertions(+), 15 deletions(-) diff --git a/pm-stack/projector/README.md b/pm-stack/projector/README.md index f3ed1f618..aaf4dc864 100644 --- a/pm-stack/projector/README.md +++ b/pm-stack/projector/README.md @@ -43,6 +43,26 @@ general/muse-meta database rather than requiring a dedicated database. Run locally against a SQLite file (`file:` URL) or remotely via `DATABASE_URL` + `TURSO_AUTH_TOKEN`. +### Upgrading the comment identity schema + +Existing databases with the old `pmp_comments(id)` primary key fail startup +with an explicit error. Stop all projector writers and back up the entire +database first, including the human-written `pmp_links` registry. Do not +delete the database: it may contain unrelated fleet data. Using the SQL client +for that database, run: + +```sql +DROP VIEW IF EXISTS pmp_v_subject_activity; +DROP TABLE pmp_comments; +``` + +Start the upgraded projector to recreate the table and view, then run a full +reconciliation for every configured repository and check all stream errors. +The comment and review streams reread their full GitHub collections. This +rebuild cannot recover deleted comments or overwritten historical records +that GitHub no longer returns; retain the backup for those records. Links, +items, deliveries, cursors, and other applications' tables are untouched. + ## Secrets (referenced by name only — never commit values) | Env var | What it is | diff --git a/pm-stack/projector/schema.sql b/pm-stack/projector/schema.sql index 6f3becf6e..51723dd2e 100644 --- a/pm-stack/projector/schema.sql +++ b/pm-stack/projector/schema.sql @@ -44,7 +44,7 @@ CREATE INDEX IF NOT EXISTS pmp_items_kind ON pmp_items(kind); -- comment-activity projection (aggregator requirement #2). -- --------------------------------------------------------------------------- CREATE TABLE IF NOT EXISTS pmp_comments ( - id TEXT PRIMARY KEY, -- GitHub comment/review id + id TEXT NOT NULL, -- GitHub comment/review id ref TEXT NOT NULL, -- item ref this comment belongs to kind TEXT NOT NULL, -- 'issue_comment' | 'review' | 'review_comment' author TEXT, @@ -62,7 +62,8 @@ CREATE TABLE IF NOT EXISTS pmp_comments ( commit_id TEXT, deleted INTEGER NOT NULL DEFAULT 0, -- set to 1 on GitHub delete events source_updated_at TEXT NOT NULL, -- created_at (or updated_at) monotonic guard - fetched_at TEXT NOT NULL + fetched_at TEXT NOT NULL, + PRIMARY KEY (id, kind) ); CREATE INDEX IF NOT EXISTS pmp_comments_ref ON pmp_comments(ref); @@ -154,4 +155,4 @@ CREATE VIEW IF NOT EXISTS pmp_v_subject_activity AS FROM pmp_links l JOIN pmp_items i ON i.ref = l.ref JOIN pmp_comments c ON c.ref = l.ref - ORDER BY c.created_at DESC, c.id DESC; \ No newline at end of file + ORDER BY c.created_at DESC, c.id DESC; diff --git a/pm-stack/projector/src/projector.ts b/pm-stack/projector/src/projector.ts index 24ec023c1..12881b7fa 100644 --- a/pm-stack/projector/src/projector.ts +++ b/pm-stack/projector/src/projector.ts @@ -8,7 +8,7 @@ import type { upsertItem, upsertComment, markCommentDeleted, recordPush, ItemRow export interface ProjectorStore { upsertItem(i: Omit): Promise; upsertComment(c: Omit): Promise; - markCommentDeleted(id: string): Promise; + markCommentDeleted(id: string, kind: CommentRow["kind"]): Promise; recordPush(p: { deliveryId: string; repo: string; @@ -156,7 +156,7 @@ export async function applyEvent( await store.upsertItem(fromIssueOrPr(repo, issue, issueIsPr(issue) ? "pr" : "issue")); addRef(itemRef(repo, issue.number)); if (event === "issue_comment" && payload.action === "deleted") { - await store.markCommentDeleted(String(comment.id)); + await store.markCommentDeleted(String(comment.id), "issue_comment"); } else { await store.upsertComment(commentCommon(comment, repo, itemRef(repo, issue.number), "issue_comment")); } @@ -210,7 +210,7 @@ export async function applyEvent( base.line = typeof comment.line === "number" ? comment.line : null; base.diff_hunk = iso(comment.diff_hunk); if (payload.action === "deleted") { - await store.markCommentDeleted(String(comment.id)); + await store.markCommentDeleted(String(comment.id), "review_comment"); } else { await store.upsertComment(base); } diff --git a/pm-stack/projector/src/server.ts b/pm-stack/projector/src/server.ts index 85211caf6..e79a7226b 100644 --- a/pm-stack/projector/src/server.ts +++ b/pm-stack/projector/src/server.ts @@ -22,7 +22,7 @@ function libsqlStore(db: Client): ProjectorStore { return { upsertItem: (i) => store.upsertItem(db, i), upsertComment: (c) => store.upsertComment(db, c), - markCommentDeleted: (id) => store.markCommentDeleted(db, id), + markCommentDeleted: (id, kind) => store.markCommentDeleted(db, id, kind), recordPush: (p) => store.recordPush(db, p), }; } diff --git a/pm-stack/projector/src/store.ts b/pm-stack/projector/src/store.ts index d20093759..5f31a1cbc 100644 --- a/pm-stack/projector/src/store.ts +++ b/pm-stack/projector/src/store.ts @@ -24,6 +24,13 @@ export function openDb(env: DbEnv): Client { /** Run schema.sql. Idempotent — every statement is CREATE ... IF NOT EXISTS. */ export async function migrate(db: Client, schemaPath?: string): Promise { + const columns = await db.execute("PRAGMA table_info(pmp_comments)"); + if (columns.rows.length > 0) { + const pk = columns.rows.filter((row) => Number(row.pk) > 0).sort((a, b) => Number(a.pk) - Number(b.pk)); + if (pk.length !== 2 || pk[0].name !== "id" || pk[1].name !== "kind") { + throw new Error("Legacy pmp_comments schema: rebuild the comment projection before starting; preserve pmp_links and other tables. See the projector README."); + } + } const path = schemaPath ?? new URL("../schema.sql", import.meta.url).pathname; const sql = await Bun.file(path).text(); // libSQL executes one statement per call. Strip full-line comments first, @@ -204,7 +211,7 @@ export async function upsertComment(db: Client, c: Omit { +export async function markCommentDeleted(db: Client, id: string, kind: CommentRow["kind"]): Promise { await db.execute({ - sql: `UPDATE pmp_comments SET deleted = 1, fetched_at = ? WHERE id = ?`, - args: [now(), id], + sql: `UPDATE pmp_comments SET deleted = 1, fetched_at = ? WHERE id = ? AND kind = ?`, + args: [now(), id, kind], }); } diff --git a/pm-stack/projector/test/e2e.test.ts b/pm-stack/projector/test/e2e.test.ts index a79816ceb..a69d0fcd4 100644 --- a/pm-stack/projector/test/e2e.test.ts +++ b/pm-stack/projector/test/e2e.test.ts @@ -82,6 +82,43 @@ afterAll(() => { }); describe("end-to-end webhook + read model", () => { + test("colliding webhook ids and deletes remain scoped through the server", async () => { + const local = createClient({ url: ":memory:" }); + try { + await migrate(local); + const env = { GITHUB_WEBHOOK_SECRET: SECRET }; + const pr = { number: 7, title: "PR", created_at: "2026-01-01", updated_at: "2026-01-01" }; + const repository = { full_name: "example/repo" }; + const comment = { id: 42, created_at: "2026-01-01", body: "body" }; + let delivery = 0; + const send = async (event: string, payload: object) => { + const body = JSON.stringify(payload); + const sig = await hmacSha256Hex(new TextEncoder().encode(body), SECRET); + const res = await handleRequest(webhookReq(body, event, sig, `collision-${delivery++}`), local, env); + expect(res.status).toBe(200); + }; + await send("issue_comment", { repository, issue: pr, comment, action: "created" }); + await send("pull_request_review", { repository, pull_request: pr, review: comment, action: "submitted" }); + await send("pull_request_review_comment", { repository, pull_request: pr, comment, action: "created" }); + let rows = await local.execute("SELECT kind, deleted FROM pmp_comments ORDER BY kind"); + expect(rows.rows.map((r) => [r.kind, r.deleted])).toEqual([ + ["issue_comment", 0], ["review", 0], ["review_comment", 0], + ]); + await send("issue_comment", { repository, issue: pr, comment, action: "deleted" }); + rows = await local.execute("SELECT kind, deleted FROM pmp_comments ORDER BY kind"); + expect(rows.rows.map((r) => [r.kind, r.deleted])).toEqual([ + ["issue_comment", 1], ["review", 0], ["review_comment", 0], + ]); + await send("pull_request_review_comment", { repository, pull_request: pr, comment, action: "deleted" }); + rows = await local.execute("SELECT kind, deleted FROM pmp_comments ORDER BY kind"); + expect(rows.rows.map((r) => [r.kind, r.deleted])).toEqual([ + ["issue_comment", 1], ["review", 0], ["review_comment", 1], + ]); + } finally { + local.close(); + } + }); + test("fixture sequence applies and produces registry + read-model rows", async () => { const env: ServerEnv = { GITHUB_WEBHOOK_SECRET: SECRET, diff --git a/pm-stack/projector/test/projector.test.ts b/pm-stack/projector/test/projector.test.ts index 59942ba76..bbf7bcf2d 100644 --- a/pm-stack/projector/test/projector.test.ts +++ b/pm-stack/projector/test/projector.test.ts @@ -6,7 +6,7 @@ function fakeStore() { const items = new Map>(); const comments = new Map>(); const pushes: any[] = []; - const deleted: string[] = []; + const deleted: [string, CommentRow["kind"]][] = []; const store: ProjectorStore = { async upsertItem(i) { items.set(i.ref, i); @@ -14,8 +14,8 @@ function fakeStore() { async upsertComment(c) { comments.set(c.id, c); }, - async markCommentDeleted(id) { - deleted.push(id); + async markCommentDeleted(id, kind) { + deleted.push([id, kind]); }, async recordPush(p) { pushes.push(p); @@ -174,7 +174,12 @@ describe("projector: comment deletion + push", () => { }, comment: { id: 700 }, }); - expect(deleted).toContain("700"); + expect(deleted).toEqual([["700", "issue_comment"]]); + await applyEvent(store, "pull_request_review_comment", { + action: "deleted", repository: { full_name: "developing-today/code" }, + pull_request: { number: 300 }, comment: { id: 700 }, + }); + expect(deleted[1]).toEqual(["700", "review_comment"]); }); test("push event records head/before/pusher", async () => { diff --git a/pm-stack/projector/test/store.test.ts b/pm-stack/projector/test/store.test.ts index 1c4de1ec0..58b8a3802 100644 --- a/pm-stack/projector/test/store.test.ts +++ b/pm-stack/projector/test/store.test.ts @@ -7,6 +7,7 @@ import { migrate, upsertItem, upsertComment, + markCommentDeleted, upsertLink, getItems, getComments, @@ -100,6 +101,38 @@ describe("store: items", () => { }); describe("store: comments + links + subject view", () => { + test("same GitHub id is independent across comment kinds and deletion is scoped", async () => { + const base: Omit = { + id: "collision", ref: "test/collision#1", kind: "issue_comment", author: null, + body: "issue", created_at: "2026-01-01", updated_at: null, url: null, + path: null, line: null, diff_hunk: null, review_state: null, submitted_at: null, + commit_id: null, deleted: 0, source_updated_at: "2026-01-01", + }; + await upsertComment(db, base); + await upsertComment(db, { ...base, kind: "review", body: "review" }); + await upsertComment(db, { ...base, kind: "review_comment", body: "line" }); + await upsertComment(db, { ...base, body: "updated issue" }); + let rows = await getComments(db, [base.ref]); + expect(rows.map((r) => [r.kind, r.body]).sort()).toEqual([ + ["issue_comment", "updated issue"], ["review", "review"], ["review_comment", "line"], + ]); + await markCommentDeleted(db, base.id, "issue_comment"); + rows = await getComments(db, [base.ref]); + expect(rows.map((r) => r.kind).sort()).toEqual(["review", "review_comment"]); + await markCommentDeleted(db, base.id, "review_comment"); + rows = await getComments(db, [base.ref]); + expect(rows.map((r) => r.kind)).toEqual(["review"]); + }); + + test("legacy schema is rejected with a clear rebuild instruction", async () => { + const legacy = createClient({ url: ":memory:" }); + try { + await legacy.execute("CREATE TABLE pmp_comments (id TEXT PRIMARY KEY, kind TEXT NOT NULL)"); + await expect(migrate(legacy)).rejects.toThrow(/legacy pmp_comments.*rebuild/i); + } finally { + legacy.close(); + } + }); test("comments upsert and read, newest first", async () => { const base: Omit = { id: "c1",