feat(webhook): Vercel receiver + MongoDB queue (changeStream consumer)

CLAUDE.md

@@ -1,14 +1,22 @@
 # Claude Development Notes
 
-## Bot Startup (PM2)
+## Bot Startup (oxmgr)
 
-The Slack bot runs on the `sno-bot` branch and should always be running via PM2.
+The Slack bot runs on the `sno-bot` branch and should always be running via oxmgr.
+
+> Migrated from pm2 to oxmgr on 2026-05-12. The npm-installed oxmgr binary
+> ships glibc-linked and requires GLIBC_2.39 (Debian 13+); on Debian 12 we
+> manually swap in the musl static-pie binary from the GitHub release:
+> `/root/.nvm/versions/node/v25.2.1/lib/node_modules/oxmgr/vendor/oxmgr`.
+> Original at `.glibc.bak`. Upstream tracking: Vladimir-Urik/OxMgr#32.
 
 ### Start / Restart
 
 ```bash
 # Start (or restart if already running)
-pm2 start /root/.bun/bin/bun --name comfy-pr-bot --interpreter none -- bot/index.ts --continue
+oxmgr start --name comfy-pr-bot --restart always \
+  --cwd /v1/code/Comfy-Org/Comfy-PR/tree/sno-bot \
+  "/root/.bun/bin/bun bot/index.ts --continue"
 
 # Or use the convenience script (stops old instance first)
 bash bot/up.sh
@@ -17,15 +25,19 @@ bash bot/up.sh
 ### Check Status & Logs
 
 ```bash
-pm2 status comfy-pr-bot
-pm2 logs comfy-pr-bot --lines 50 --nostream
+oxmgr ls                       # one-line status
+oxmgr status comfy-pr-bot      # detailed view
+oxmgr logs comfy-pr-bot        # recent logs
+oxmgr logs comfy-pr-bot -f     # follow live
 ```
 
+Log files are at `/root/.local/share/oxmgr/logs/comfy-pr-bot.{out,err}.log`.
+
 ### Stop
 
 ```bash
-pm2 stop comfy-pr-bot
-pm2 delete comfy-pr-bot
+oxmgr stop comfy-pr-bot
+oxmgr rm   comfy-pr-bot   # also drops the persisted definition
 ```
 
 ### Important Notes
@@ -35,6 +47,7 @@ pm2 delete comfy-pr-bot
 - Port `3475` is used for health checks (env `PRBOT_PORT`)
 - RestartManager watches `bot/`, `src/`, `lib/` for file changes and auto-restarts when idle
 - If the bot crash-loops, check for merge conflicts: `grep -n '<<<<<<' bot/slack-bot.ts`
+- If a single resumed task triggers a SIGKILL loop (memory spike), clear `current-working-tasks` from SlackBotState; helper: `bun tmp/clear-working-tasks.ts`. See project memory `bot-poison-pill-task.md`.
 
 ## Security: Top-Level `await createIndex` Is Intentional
 

app/api/webhook/github/route.ts

@@ -1,3 +1,4 @@
+import { enqueueWebhook } from "@/bot/webhook-queue";
 import { db } from "@/src/db";
 import { createHmac } from "crypto";
 import { type NextRequest, NextResponse } from "next/server";
@@ -88,10 +89,33 @@ export async function POST(request: NextRequest) {
       processed: false,
     };
 
-    // Store to MongoDB
+    // Store to MongoDB. Dual-write: keep the historical
+    // `GithubWebhookEvents` collection for back-compat, and also push into
+    // the unified `webhook_queue` so the VM bot's changeStream consumer can
+    // pick GitHub events up alongside Slack.
     const collection = db.collection("GithubWebhookEvents");
     const result = await collection.insertOne(eventDocument);
 
+    if (deliveryId) {
+      try {
+        await enqueueWebhook({
+          source: "github",
+          eventId: deliveryId,
+          payload,
+          meta: {
+            eventType,
+            hookId,
+            hookInstallationTargetId,
+            hookInstallationTargetType,
+            userAgent: request.headers.get("user-agent"),
+            legacyId: result.insertedId.toString(),
+          },
+        });
+      } catch (qerr) {
+        console.warn("Failed to enqueue github webhook to webhook_queue", qerr);
+      }
+    }
+
     console.log(
       `Stored GitHub webhook event: ${eventType} (delivery: ${deliveryId}, _id: ${result.insertedId})`,
     );

app/api/webhook/notion/route.ts

@@ -0,0 +1,142 @@
+import { enqueueWebhook, ensureWebhookQueueIndexes } from "@/bot/webhook-queue";
+import { db } from "@/src/db";
+import { createHmac, timingSafeEqual } from "crypto";
+import { type NextRequest, NextResponse } from "next/server";
+
+/**
+ * Notion webhook receiver, deployed to Vercel.
+ *
+ * Notion's webhook setup goes through two phases:
+ *
+ * 1. Initial verification: Notion POSTs a JSON body
+ *    `{ "verification_token": "..." }` to the configured URL once. The
+ *    integration setup screen expects you to paste that token back.
+ *    We log the token and also stash it in MongoDB
+ *    (`webhook_notion_verification`) so the operator can retrieve it
+ *    out-of-band without scrubbing logs.
+ *
+ * 2. Steady-state: signed events. The signature header is
+ *    `X-Notion-Signature: sha256=<hex>` over the raw request body, keyed
+ *    by the `verification_token` issued in phase 1. Set the same token
+ *    as `NOTION_WEBHOOK_VERIFICATION_TOKEN` in this Vercel project so
+ *    we can validate.
+ *
+ * @see https://developers.notion.com/reference/webhooks
+ */
+
+export const dynamic = "force-dynamic";
+export const runtime = "nodejs";
+
+function verifyNotionSignature(rawBody: string, signature: string, token: string): boolean {
+  const hmac = createHmac("sha256", token).update(rawBody).digest("hex");
+  const expected = Buffer.from(`sha256=${hmac}`);
+  const received = Buffer.from(signature);
+  if (expected.length !== received.length) return false;
+  return timingSafeEqual(expected, received);
+}
+
+export async function POST(request: NextRequest) {
+  const rawBody = await request.text();
+
+  let payload: {
+    verification_token?: string;
+    type?: string;
+    id?: string;
+    workspace_id?: string;
+    [k: string]: unknown;
+  };
+  try {
+    payload = JSON.parse(rawBody);
+  } catch {
+    return NextResponse.json({ error: "Invalid JSON" }, { status: 400 });
+  }
+
+  // Phase 1: verification handshake. Capture the token so we can register
+  // it in env without copying from logs.
+  if (typeof payload.verification_token === "string" && !payload.type) {
+    try {
+      await db.collection("webhook_notion_verification").insertOne({
+        verification_token: payload.verification_token,
+        receivedAt: new Date(),
+        userAgent: request.headers.get("user-agent"),
+      });
+    } catch (err) {
+      console.warn("notion verification token persist failed", err);
+    }
+    console.log(
+      `[notion-webhook] verification handshake received; token starts with ${payload.verification_token.slice(0, 6)}…`,
+    );
+    return new NextResponse("", { status: 200 });
+  }
+
+  // Phase 2: signed events.
+  const token = process.env.NOTION_WEBHOOK_VERIFICATION_TOKEN;
+  if (!token) {
+    console.error(
+      "[notion-webhook] NOTION_WEBHOOK_VERIFICATION_TOKEN unset — accepting unsigned event for setup, but this is insecure",
+    );
+  } else {
+    const sig = request.headers.get("x-notion-signature") ?? "";
+    if (!sig || !verifyNotionSignature(rawBody, sig, token)) {
+      return new NextResponse("Invalid signature", { status: 401 });
+    }
+  }
+
+  await ensureWebhookQueueIndexes();
+
+  // Notion gives every event a stable id; fall back to a synthetic one so
+  // dedup still works on test deliveries that omit it.
+  const eventId =
+    typeof payload.id === "string" ? payload.id : `${payload.type ?? "unknown"}_${Date.now()}`;
+
+  try {
+    await db.collection("webhook_edge_dedup").insertOne({
+      _id: `notion:${eventId}` as unknown as never,
+      createdAt: new Date(),
+    });
+  } catch (err) {
+    const code = (err as { code?: number }).code;
+    if (code === 11000) {
+      // Duplicate delivery — Notion retries on non-2xx, we may have already
+      // enqueued. Treat as success.
+      return new NextResponse("", { status: 200 });
+    }
+    console.warn("notion dedup insert failed, falling through", err);
+  }
+
+  await enqueueWebhook({
+    source: "notion",
+    eventId,
+    payload,
+    meta: {
+      type: payload.type ?? null,
+      workspace_id: payload.workspace_id ?? null,
+      receivedAt: Date.now(),
+    },
+  });
+
+  return new NextResponse("", { status: 200 });
+}
+
+export async function GET() {
+  try {
+    await db.admin().ping();
+    const col = db.collection("webhook_queue");
+    const total = await col.countDocuments({ source: "notion" });
+    const pending = await col.countDocuments({ source: "notion", processed: false });
+    const verifTokens = await db.collection("webhook_notion_verification").countDocuments();
+    return NextResponse.json({
+      status: "ok",
+      collection: "webhook_queue",
+      notion: { total, pending, verificationTokensReceived: verifTokens },
+    });
+  } catch (error) {
+    return NextResponse.json(
+      {
+        status: "error",
+        message: error instanceof Error ? error.message : String(error),
+      },
+      { status: 500 },
+    );
+  }
+}

app/api/webhook/slack/route.ts

@@ -0,0 +1,137 @@
+import { enqueueWebhook, ensureWebhookQueueIndexes } from "@/bot/webhook-queue";
+import { db } from "@/src/db";
+import { createHmac, timingSafeEqual } from "crypto";
+import { type NextRequest, NextResponse } from "next/server";
+
+/**
+ * Slack webhook receiver, deployed to Vercel.
+ *
+ * The local bot on the VM tails MongoDB's `webhook_queue` via changeStream,
+ * so this endpoint stays up even when the bot is restarting/down — no
+ * Slack retries get dropped.
+ *
+ * URL verification (initial setup): respond with the challenge directly.
+ * event_callback: verify HMAC, enqueue, 200.
+ *
+ * @see https://docs.slack.dev/apis/events-api/
+ */
+
+export const dynamic = "force-dynamic";
+export const runtime = "nodejs";
+
+function verifySlackSignature(
+  body: string,
+  timestamp: string,
+  signature: string,
+  secret: string,
+): boolean {
+  // Slack guards against >5min replay attacks; we mirror that.
+  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false;
+  const hmac = createHmac("sha256", secret).update(`v0:${timestamp}:${body}`).digest("hex");
+  const expected = Buffer.from(`v0=${hmac}`);
+  const received = Buffer.from(signature);
+  if (expected.length !== received.length) return false;
+  return timingSafeEqual(expected, received);
+}
+
+export async function POST(request: NextRequest) {
+  const body = await request.text();
+  const timestamp = request.headers.get("x-slack-request-timestamp") ?? "";
+  const signature = request.headers.get("x-slack-signature") ?? "";
+
+  const secret = process.env.SLACK_SIGNING_SECRET;
+  if (!secret) {
+    console.error("SLACK_SIGNING_SECRET is not set in this Vercel deployment");
+    return NextResponse.json({ error: "server misconfigured" }, { status: 500 });
+  }
+
+  if (!verifySlackSignature(body, timestamp, signature, secret)) {
+    return new NextResponse("Invalid signature", { status: 401 });
+  }
+
+  let payload: {
+    type?: string;
+    challenge?: string;
+    event?: { team?: string; channel?: string; ts?: string; event_ts?: string };
+    team_id?: string;
+    event_id?: string;
+  };
+  try {
+    payload = JSON.parse(body);
+  } catch {
+    return NextResponse.json({ error: "Invalid JSON" }, { status: 400 });
+  }
+
+  // URL verification challenge — Slack does this when you set the
+  // Request URL in the app config.
+  if (payload.type === "url_verification" && payload.challenge) {
+    return NextResponse.json({ challenge: payload.challenge });
+  }
+
+  if (payload.type === "event_callback") {
+    const retryNum = request.headers.get("x-slack-retry-num");
+    const retryReason = request.headers.get("x-slack-retry-reason");
+    const eventId =
+      payload.event_id ||
+      `${payload.event?.channel ?? "-"}_${payload.event?.event_ts ?? payload.event?.ts ?? "-"}`;
+
+    // Make sure indexes exist (no-op after first request per instance).
+    await ensureWebhookQueueIndexes();
+
+    // Edge-level dedup so a Slack retry storm doesn't insert N copies.
+    // The actual queue is content-deduped by the bot consumer.
+    try {
+      const dedupCol = db.collection("webhook_edge_dedup");
+      // _id is a string; the unique constraint is built-in, so a duplicate
+      // throws code 11000.
+      await dedupCol.insertOne({
+        _id: `slack:${eventId}` as unknown as never,
+        createdAt: new Date(),
+      });
+    } catch (err) {
+      const code = (err as { code?: number }).code;
+      if (code === 11000) {
+        console.log(`slack webhook dedup hit for ${eventId} (retry=${retryNum})`);
+        return new NextResponse("", { status: 200 });
+      }
+      console.warn("dedup insert failed, falling through", err);
+    }
+
+    await enqueueWebhook({
+      source: "slack",
+      eventId,
+      payload,
+      meta: {
+        retryNum: retryNum ?? null,
+        retryReason: retryReason ?? null,
+        receivedAt: Date.now(),
+      },
+    });
+  }
+
+  // Always 200 the webhook within a few seconds — Slack times out at 3s
+  // and retries up to 3 times if it doesn't get one.
+  return new NextResponse("", { status: 200 });
+}
+
+export async function GET() {
+  try {
+    await db.admin().ping();
+    const col = db.collection("webhook_queue");
+    const total = await col.countDocuments({ source: "slack" });
+    const pending = await col.countDocuments({ source: "slack", processed: false });
+    return NextResponse.json({
+      status: "ok",
+      collection: "webhook_queue",
+      slack: { total, pending },
+    });
+  } catch (error) {
+    return NextResponse.json(
+      {
+        status: "error",
+        message: error instanceof Error ? error.message : String(error),
+      },
+      { status: 500 },
+    );
+  }
+}