feat(notifications): event→subscription→delivery service + MCP badge feed (closes #536, advances #535)

migrations/0031_notification_subscriptions.sql

@@ -0,0 +1,44 @@
+-- Event-to-subscription-to-delivery notifications (#535). The killer event is a changes_requested
+-- review on a miner's PR (detected in src/notifications/events.ts). `notification_subscriptions` mirrors
+-- `digest_subscriptions` and stores per-channel opt-out (badge is on by default unless paused).
+-- `notification_deliveries` is the idempotent badge read-model: UNIQUE(dedup_key, channel) guarantees a
+-- duplicate webhook / queue retry produces exactly one delivery.
+CREATE TABLE IF NOT EXISTS notification_subscriptions (
+  id TEXT PRIMARY KEY,
+  login TEXT NOT NULL,
+  channel TEXT NOT NULL,
+  status TEXT NOT NULL DEFAULT 'active',
+  destination TEXT,
+  source TEXT NOT NULL DEFAULT 'app',
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE UNIQUE INDEX notification_subscriptions_login_channel_unique
+  ON notification_subscriptions(login, channel);
+CREATE INDEX notification_subscriptions_login_idx ON notification_subscriptions(login);
+
+CREATE TABLE IF NOT EXISTS notification_deliveries (
+  id TEXT PRIMARY KEY,
+  dedup_key TEXT NOT NULL,
+  channel TEXT NOT NULL,
+  recipient_login TEXT NOT NULL,
+  event_type TEXT NOT NULL,
+  repo_full_name TEXT NOT NULL,
+  pull_number INTEGER,
+  title TEXT NOT NULL,
+  body TEXT NOT NULL,
+  deeplink TEXT NOT NULL,
+  actor_login TEXT,
+  status TEXT NOT NULL DEFAULT 'pending',
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  delivered_at TEXT,
+  read_at TEXT
+);
+
+CREATE UNIQUE INDEX notification_deliveries_dedup_channel_unique
+  ON notification_deliveries(dedup_key, channel);
+CREATE INDEX notification_deliveries_recipient_status_idx
+  ON notification_deliveries(recipient_login, status);
+CREATE INDEX notification_deliveries_recipient_channel_created_idx
+  ON notification_deliveries(recipient_login, channel, created_at);

src/db/repositories.ts

@@ -26,6 +26,8 @@ import {
   issueQualityReports,
   issues,
   githubRateLimitObservations,
+  notificationDeliveries,
+  notificationSubscriptions,
   officialMinerDetections,
   pullRequestFiles,
   pullRequestDetailSyncState,
@@ -95,6 +97,10 @@ import type {
   IssueQualityReportRecord,
   JsonValue,
   McpCompatibilityAdoptionSummary,
+  NotificationChannel,
+  NotificationDeliveryRecord,
+  NotificationDeliveryStatus,
+  NotificationSubscriptionRecord,
   ProductUsageActivationFunnel,
   ProductUsageDailyRollupRecord,
   ProductUsageDailyRollupStatus,
@@ -1231,6 +1237,178 @@ export async function countActiveDigestSubscriptions(env: Env): Promise<number>
   return Number(row?.count ?? 0);
 }
 
+export async function upsertNotificationSubscription(
+  env: Env,
+  input: { login: string; channel: NotificationChannel; status?: NotificationSubscriptionRecord["status"]; destination?: string | null; source?: string },
+): Promise<NotificationSubscriptionRecord> {
+  const db = getDb(env.DB);
+  const now = nowIso();
+  const record: NotificationSubscriptionRecord = {
+    id: crypto.randomUUID(),
+    login: input.login.toLowerCase(),
+    channel: input.channel,
+    status: input.status ?? "active",
+    destination: input.destination ?? null,
+    source: input.source ?? "app",
+    createdAt: now,
+    updatedAt: now,
+  };
+  await db
+    .insert(notificationSubscriptions)
+    .values({
+      id: record.id,
+      login: record.login,
+      channel: record.channel,
+      status: record.status,
+      destination: record.destination,
+      source: record.source,
+    })
+    .onConflictDoUpdate({
+      target: [notificationSubscriptions.login, notificationSubscriptions.channel],
+      set: { status: record.status, destination: record.destination, source: record.source, updatedAt: now },
+    });
+  const [row] = await db
+    .select()
+    .from(notificationSubscriptions)
+    .where(and(eq(notificationSubscriptions.login, record.login), eq(notificationSubscriptions.channel, record.channel)))
+    .limit(1);
+  return row ? toNotificationSubscriptionRecord(row) : record;
+}
+
+export async function listNotificationSubscriptionsForLogin(env: Env, login: string): Promise<NotificationSubscriptionRecord[]> {
+  const db = getDb(env.DB);
+  const rows = await db.select().from(notificationSubscriptions).where(eq(notificationSubscriptions.login, login.toLowerCase())).limit(20);
+  return rows.map(toNotificationSubscriptionRecord);
+}
+
+// Idempotency guard: UNIQUE(dedup_key, channel) means a duplicate webhook / queue retry inserts nothing
+// and returns the existing row. Returns whether THIS call created the row (so only the first enqueues delivery).
+export async function insertNotificationDeliveryIfAbsent(
+  env: Env,
+  input: Omit<NotificationDeliveryRecord, "id" | "createdAt" | "deliveredAt" | "readAt" | "status"> & { status?: NotificationDeliveryStatus },
+): Promise<{ delivery: NotificationDeliveryRecord; created: boolean }> {
+  const db = getDb(env.DB);
+  const now = nowIso();
+  const record: NotificationDeliveryRecord = {
+    id: crypto.randomUUID(),
+    dedupKey: input.dedupKey,
+    channel: input.channel,
+    recipientLogin: input.recipientLogin.toLowerCase(),
+    eventType: input.eventType,
+    repoFullName: input.repoFullName,
+    pullNumber: input.pullNumber,
+    title: input.title,
+    body: input.body,
+    deeplink: input.deeplink,
+    actorLogin: input.actorLogin,
+    status: input.status ?? "pending",
+    createdAt: now,
+    deliveredAt: null,
+    readAt: null,
+  };
+  const inserted = await db
+    .insert(notificationDeliveries)
+    .values({
+      id: record.id,
+      dedupKey: record.dedupKey,
+      channel: record.channel,
+      recipientLogin: record.recipientLogin,
+      eventType: record.eventType,
+      repoFullName: record.repoFullName,
+      pullNumber: record.pullNumber,
+      title: record.title,
+      body: record.body,
+      deeplink: record.deeplink,
+      actorLogin: record.actorLogin,
+      status: record.status,
+    })
+    .onConflictDoNothing({ target: [notificationDeliveries.dedupKey, notificationDeliveries.channel] })
+    .returning();
+  if (inserted.length > 0 && inserted[0]) return { delivery: toNotificationDeliveryRecord(inserted[0]), created: true };
+  const [existing] = await db
+    .select()
+    .from(notificationDeliveries)
+    .where(and(eq(notificationDeliveries.dedupKey, record.dedupKey), eq(notificationDeliveries.channel, record.channel)))
+    .limit(1);
+  /* v8 ignore next -- onConflictDoNothing only skips when a row already exists, so the re-select always returns it. */
+  return { delivery: existing ? toNotificationDeliveryRecord(existing) : record, created: false };
+}
+
+export async function countRecentNotificationDeliveries(
+  env: Env,
+  recipientLogin: string,
+  channel: NotificationChannel,
+  sinceIso: string,
+): Promise<number> {
+  const db = getDb(env.DB);
+  const [row] = await db
+    .select({ count: sql<number>`count(*)` })
+    .from(notificationDeliveries)
+    .where(
+      and(
+        eq(notificationDeliveries.recipientLogin, recipientLogin.toLowerCase()),
+        eq(notificationDeliveries.channel, channel),
+        not(eq(notificationDeliveries.status, "suppressed")),
+        gte(notificationDeliveries.createdAt, sinceIso),
+      ),
+    );
+  /* v8 ignore next -- SQL aggregate count always returns one row; fallback protects D1 driver anomalies. */
+  return Number(row?.count ?? 0);
+}
+
+export async function getNotificationDeliveryById(env: Env, id: string): Promise<NotificationDeliveryRecord | null> {
+  const db = getDb(env.DB);
+  const [row] = await db.select().from(notificationDeliveries).where(eq(notificationDeliveries.id, id)).limit(1);
+  return row ? toNotificationDeliveryRecord(row) : null;
+}
+
+export async function markNotificationDeliveryDelivered(env: Env, id: string): Promise<void> {
+  const db = getDb(env.DB);
+  await db
+    .update(notificationDeliveries)
+    .set({ status: "delivered", deliveredAt: nowIso() })
+    .where(and(eq(notificationDeliveries.id, id), eq(notificationDeliveries.status, "pending")));
+}
+
+export async function listNotificationDeliveriesForRecipient(
+  env: Env,
+  recipientLogin: string,
+  options: { channel?: NotificationChannel; unreadOnly?: boolean; limit?: number } = {},
+): Promise<NotificationDeliveryRecord[]> {
+  const db = getDb(env.DB);
+  const conditions: SQL[] = [eq(notificationDeliveries.recipientLogin, recipientLogin.toLowerCase())];
+  if (options.channel) conditions.push(eq(notificationDeliveries.channel, options.channel));
+  if (options.unreadOnly) conditions.push(eq(notificationDeliveries.status, "delivered"));
+  const rows = await db
+    .select()
+    .from(notificationDeliveries)
+    .where(and(...conditions))
+    .orderBy(desc(notificationDeliveries.createdAt))
+    .limit(Math.min(Math.max(options.limit ?? 50, 1), 100));
+  return rows.map(toNotificationDeliveryRecord);
+}
+
+// Marks a recipient's delivered notifications read (the badge-clear action). Scoped to recipientLogin so a
+// caller can never clear another user's notifications. Returns the number of rows transitioned.
+export async function markNotificationDeliveriesRead(
+  env: Env,
+  recipientLogin: string,
+  ids?: string[],
+): Promise<number> {
+  const db = getDb(env.DB);
+  const conditions: SQL[] = [
+    eq(notificationDeliveries.recipientLogin, recipientLogin.toLowerCase()),
+    eq(notificationDeliveries.status, "delivered"),
+  ];
+  if (ids && ids.length > 0) conditions.push(inArray(notificationDeliveries.id, ids));
+  const updated = await db
+    .update(notificationDeliveries)
+    .set({ status: "read", readAt: nowIso() })
+    .where(and(...conditions))
+    .returning({ id: notificationDeliveries.id });
+  return updated.length;
+}
+
 export async function recordProductUsageEvent(
   env: Env,
   event: {
@@ -3681,6 +3859,47 @@ function toDigestSubscriptionRecord(row: typeof digestSubscriptions.$inferSelect
   };
 }
 
+function toNotificationChannel(value: string): NotificationChannel {
+  return value === "email" ? "email" : "badge";
+}
+
+function toNotificationSubscriptionRecord(row: typeof notificationSubscriptions.$inferSelect): NotificationSubscriptionRecord {
+  return {
+    id: row.id,
+    login: row.login,
+    channel: toNotificationChannel(row.channel),
+    status: row.status === "paused" ? "paused" : "active",
+    destination: row.destination ?? null,
+    source: row.source,
+    createdAt: row.createdAt,
+    updatedAt: row.updatedAt,
+  };
+}
+
+function toNotificationDeliveryStatus(value: string): NotificationDeliveryStatus {
+  return value === "delivered" || value === "read" || value === "suppressed" ? value : "pending";
+}
+
+function toNotificationDeliveryRecord(row: typeof notificationDeliveries.$inferSelect): NotificationDeliveryRecord {
+  return {
+    id: row.id,
+    dedupKey: row.dedupKey,
+    channel: toNotificationChannel(row.channel),
+    recipientLogin: row.recipientLogin,
+    eventType: row.eventType,
+    repoFullName: row.repoFullName,
+    pullNumber: row.pullNumber ?? null,
+    title: row.title,
+    body: row.body,
+    deeplink: row.deeplink,
+    actorLogin: row.actorLogin ?? null,
+    status: toNotificationDeliveryStatus(row.status),
+    createdAt: row.createdAt,
+    deliveredAt: row.deliveredAt ?? null,
+    readAt: row.readAt ?? null,
+  };
+}
+
 function toProductUsageEventRecord(row: typeof productUsageEvents.$inferSelect): ProductUsageEventRecord {
   return {
     id: row.id,

src/db/schema.ts

@@ -808,6 +808,50 @@ export const digestSubscriptions = sqliteTable(
   }),
 );
 
+export const notificationSubscriptions = sqliteTable(
+  "notification_subscriptions",
+  {
+    id: text("id").primaryKey(),
+    login: text("login").notNull(),
+    channel: text("channel").notNull(),
+    status: text("status").notNull().default("active"),
+    destination: text("destination"),
+    source: text("source").notNull().default("app"),
+    createdAt: text("created_at").notNull().$defaultFn(() => nowIso()),
+    updatedAt: text("updated_at").notNull().$defaultFn(() => nowIso()),
+  },
+  (table) => ({
+    loginChannel: uniqueIndex("notification_subscriptions_login_channel_unique").on(table.login, table.channel),
+    login: index("notification_subscriptions_login_idx").on(table.login),
+  }),
+);
+
+export const notificationDeliveries = sqliteTable(
+  "notification_deliveries",
+  {
+    id: text("id").primaryKey(),
+    dedupKey: text("dedup_key").notNull(),
+    channel: text("channel").notNull(),
+    recipientLogin: text("recipient_login").notNull(),
+    eventType: text("event_type").notNull(),
+    repoFullName: text("repo_full_name").notNull(),
+    pullNumber: integer("pull_number"),
+    title: text("title").notNull(),
+    body: text("body").notNull(),
+    deeplink: text("deeplink").notNull(),
+    actorLogin: text("actor_login"),
+    status: text("status").notNull().default("pending"),
+    createdAt: text("created_at").notNull().$defaultFn(() => nowIso()),
+    deliveredAt: text("delivered_at"),
+    readAt: text("read_at"),
+  },
+  (table) => ({
+    dedupChannel: uniqueIndex("notification_deliveries_dedup_channel_unique").on(table.dedupKey, table.channel),
+    recipientStatus: index("notification_deliveries_recipient_status_idx").on(table.recipientLogin, table.status),
+    recipientChannelCreated: index("notification_deliveries_recipient_channel_created_idx").on(table.recipientLogin, table.channel, table.createdAt),
+  }),
+);
+
 export const githubAgentCommandAnswers = sqliteTable(
   "github_agent_command_answers",
   {