dezren39 · Oct 31, 2026
@@ -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                                                                |

@@ -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;
+  ORDER BY c.created_at DESC, c.id DESC;
@@ -8,7 +8,7 @@ import type { upsertItem, upsertComment, markCommentDeleted, recordPush, ItemRow
 export interface ProjectorStore {
   upsertItem(i: Omit<ItemRow, "fetched_at">): Promise<void>;
   upsertComment(c: Omit<CommentRow, "fetched_at">): Promise<void>;
-  markCommentDeleted(id: string): Promise<void>;
+  markCommentDeleted(id: string, kind: CommentRow["kind"]): Promise<void>;
   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);
         }

@@ -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),
   };
 }

@@ -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<void> {
+  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<CommentRow, "fetched_at"
             line, diff_hunk, review_state, submitted_at, commit_id, deleted,
             source_updated_at, fetched_at
           ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
-          ON CONFLICT(id) DO UPDATE SET
+          ON CONFLICT(id, kind) DO UPDATE SET
             ref = excluded.ref,
             kind = excluded.kind,
             author = excluded.author,
@@ -244,10 +251,10 @@ export async function upsertComment(db: Client, c: Omit<CommentRow, "fetched_at"
   });
 }
 
-export async function markCommentDeleted(db: Client, id: string): Promise<void> {
+export async function markCommentDeleted(db: Client, id: string, kind: CommentRow["kind"]): Promise<void> {
   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],
   });
 }
 

@@ -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,

@@ -6,16 +6,16 @@ function fakeStore() {
   const items = new Map<string, Omit<ItemRow, "fetched_at">>();
   const comments = new Map<string, Omit<CommentRow, "fetched_at">>();
   const pushes: any[] = [];
-  const deleted: string[] = [];
+  const deleted: [string, CommentRow["kind"]][] = [];
   const store: ProjectorStore = {
     async upsertItem(i) {
       items.set(i.ref, i);
     },
     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 () => {

@@ -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<CommentRow, "fetched_at"> = {
+      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<CommentRow, "fetched_at"> = {
       id: "c1",