From 1d1c5731db233c8270db7ce4fba2079804a98c8d Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sun, 14 Jun 2026 03:31:42 -0700 Subject: [PATCH] =?UTF-8?q?feat(notifications):=20event=E2=86=92subscripti?= =?UTF-8?q?on=E2=86=92delivery=20service=20+=20MCP=20badge=20feed=20(close?= =?UTF-8?q?s=20#536,=20advances=20#535)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the contribution loop's killer event: when a reviewer requests changes on a miner's PR, the miner now has a notification to read. Builds the event→subscription→delivery pipeline on top of the existing changes-requested detector (events.ts, #609), delivered through the miner's primary surface — MCP. - D1: `notification_subscriptions` (per-channel opt-out; badge on by default) + `notification_deliveries` with UNIQUE(dedup_key, channel) as the idempotency guard, so a duplicate webhook / queue retry produces exactly one delivery. - Queue: `notify-evaluate` / `notify-deliver` job types. The webhook enqueues notify-evaluate per detected event; evaluate resolves channels, writes one idempotent delivery row (rate-limited per recipient/window — bursts beyond the cap are recorded `suppressed`, never notified), and enqueues notify-deliver; deliver makes the badge row visible (pull-based). - Notification service (`src/notifications/service.ts`): channel resolution, public-safe changes-requested copy (via sanitizePublicComment), the badge feed builder (unread = delivered count), idempotent evaluate, and deliver. - MCP: `gittensory_list_notifications` + `gittensory_mark_notifications_read`, the miner's harness surface. Both self-scoped via requireContributorAccess — a session can only read/clear its OWN login's notifications. - `DetectedNotificationEvent`/`NotificationEventType` moved to types.ts (canonical location) to avoid a types↔events circular import. Scope: badge channel only (the AC's first channel, not gated behind #150/PWA). The maintainer-gated browser-extension badge UI (#534/#569), email (#570), and the predicted-gate fix-list enrichment remain follow-ups on #535. Tests: service unit (channel resolution, copy, feed, idempotency, rate-limit, mute, deliver), queue wiring (webhook→evaluate→deliver e2e + idempotency), MCP tool scope (own vs. other login). 97% coverage gate green; workers tests pass. --- .../0031_notification_subscriptions.sql | 44 ++++ src/db/repositories.ts | 219 +++++++++++++++++ src/db/schema.ts | 44 ++++ src/mcp/server.ts | 64 +++++ src/notifications/events.ts | 15 +- src/notifications/service.ts | 119 +++++++++ src/queue/processors.ts | 10 + src/types.ts | 58 +++++ test/unit/mcp-notifications.test.ts | 66 +++++ test/unit/notifications-service.test.ts | 226 ++++++++++++++++++ test/unit/queue.test.ts | 13 +- 11 files changed, 864 insertions(+), 14 deletions(-) create mode 100644 migrations/0031_notification_subscriptions.sql create mode 100644 src/notifications/service.ts create mode 100644 test/unit/mcp-notifications.test.ts create mode 100644 test/unit/notifications-service.test.ts diff --git a/migrations/0031_notification_subscriptions.sql b/migrations/0031_notification_subscriptions.sql new file mode 100644 index 0000000000..d640605db1 --- /dev/null +++ b/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); diff --git a/src/db/repositories.ts b/src/db/repositories.ts index 2fb6c8141f..0dafdc7c99 100644 --- a/src/db/repositories.ts +++ b/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 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 { + 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 { + 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 & { 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 { + const db = getDb(env.DB); + const [row] = await db + .select({ count: sql`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 { + 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 { + 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 { + 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 { + 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, diff --git a/src/db/schema.ts b/src/db/schema.ts index 4e7aa5c4d7..512f3fe374 100644 --- a/src/db/schema.ts +++ b/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", { diff --git a/src/mcp/server.ts b/src/mcp/server.ts index 65cb99cfef..ce8484ad90 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -22,14 +22,17 @@ import { listContributorPullRequests, listIssueSignalSample, listIssues, + listNotificationDeliveriesForRecipient, listOpenPullRequests, listPullRequests, listRecentMergedPullRequests, listRepoSyncSegments, listRepoSyncStates, listRepositories, + markNotificationDeliveriesRead, recordProductUsageEvent, } from "../db/repositories"; +import { buildNotificationFeed } from "../notifications/service"; import { contributorRepoStatsFromGittensor, fetchGittensorContributorSnapshot } from "../gittensor/api"; import { fetchPublicContributorProfile } from "../github/public"; import { listLatestRegistrySnapshots } from "../registry/sync"; @@ -351,6 +354,26 @@ const openPrMonitorOutputSchema = { pullRequests: z.unknown().optional(), }; +const notificationsOutputSchema = { + login: z.string().optional(), + unreadCount: z.number().optional(), + notifications: z.unknown().optional(), +}; + +const markNotificationsReadOutputSchema = { + login: z.string().optional(), + marked: z.number().optional(), +}; + +const listNotificationsShape = { + login: z.string().min(1), +}; + +const markNotificationsReadShape = { + login: z.string().min(1), + ids: z.array(z.string().min(1)).optional(), +}; + const explainRepoDecisionOutputSchema = { status: z.string().optional(), login: z.string().optional(), @@ -547,6 +570,28 @@ export class GittensoryMcp { async (input) => this.toolResult(await this.monitorOpenPullRequests(input.login)), ); + server.registerTool( + "gittensory_list_notifications", + { + description: + "Return a contributor's own Gittensory notifications (e.g. changes requested on their PRs) and unread badge count. Self-scoped: only the authenticated login's notifications.", + inputSchema: listNotificationsShape, + outputSchema: notificationsOutputSchema, + }, + async (input) => this.toolResult(await this.listNotifications(input.login)), + ); + + server.registerTool( + "gittensory_mark_notifications_read", + { + description: + "Mark a contributor's own delivered notifications as read (clears the badge). Self-scoped; pass `ids` to clear specific notifications or omit to clear all.", + inputSchema: markNotificationsReadShape, + outputSchema: markNotificationsReadOutputSchema, + }, + async (input) => this.toolResult(await this.markNotificationsRead(input.login, input.ids)), + ); + server.registerTool( "gittensory_explain_repo_decision", { @@ -1122,6 +1167,25 @@ export class GittensoryMcp { }; } + private async listNotifications(login: string): Promise { + this.requireContributorAccess(login); + const deliveries = await listNotificationDeliveriesForRecipient(this.env, login, { channel: "badge", limit: 50 }); + const feed = buildNotificationFeed(login, deliveries); + return { + summary: `Gittensory notifications for ${login}: ${feed.unreadCount} unread.`, + data: feed as unknown as Record, + }; + } + + private async markNotificationsRead(login: string, ids?: string[]): Promise { + this.requireContributorAccess(login); + const marked = await markNotificationDeliveriesRead(this.env, login, ids); + return { + summary: `Marked ${marked} Gittensory notification(s) read for ${login}.`, + data: { login: login.toLowerCase(), marked }, + }; + } + private async explainRepoDecision(input: { login: string; owner: string; repo: string }): Promise { this.requireContributorAccess(input.login); const fullName = `${input.owner}/${input.repo}`; diff --git a/src/notifications/events.ts b/src/notifications/events.ts index 52024e015a..301b49f7ad 100644 --- a/src/notifications/events.ts +++ b/src/notifications/events.ts @@ -1,18 +1,7 @@ -import type { GitHubWebhookPayload } from "../types"; +import type { DetectedNotificationEvent, GitHubWebhookPayload } from "../types"; import { nowIso } from "../utils/json"; -export type NotificationEventType = "pull_request_changes_requested"; - -export type DetectedNotificationEvent = { - eventType: NotificationEventType; - recipientLogin: string; - repoFullName: string; - pullNumber: number; - dedupKey: string; - deeplink: string; - actorLogin: string; - detectedAt: string; -}; +export type { DetectedNotificationEvent, NotificationEventType } from "../types"; function isBotUser(user: { login?: string; type?: string } | undefined): boolean { return user?.type === "Bot"; diff --git a/src/notifications/service.ts b/src/notifications/service.ts new file mode 100644 index 0000000000..e092d52fdd --- /dev/null +++ b/src/notifications/service.ts @@ -0,0 +1,119 @@ +import { sanitizePublicComment } from "../github/commands"; +import { + countRecentNotificationDeliveries, + getNotificationDeliveryById, + insertNotificationDeliveryIfAbsent, + listNotificationSubscriptionsForLogin, + markNotificationDeliveryDelivered, +} from "../db/repositories"; +import type { DetectedNotificationEvent, NotificationChannel, NotificationDeliveryRecord, NotificationSubscriptionRecord } from "../types"; +import { nowIso } from "../utils/json"; + +// Per-recipient, per-channel safety cap. The killer event (changes_requested) delivers immediately, but a +// burst of reviews must not flood a miner's badge — beyond the cap inside the window, deliveries are still +// recorded (idempotent) but marked `suppressed` so they neither notify nor count toward the next window. +export const NOTIFICATION_RATE_LIMIT = { windowMinutes: 60, maxPerWindow: 10 } as const; + +// `badge` is the channel shipped first (pull-based extension + harness feed). It is on by default; a miner +// opts OUT by pausing the badge subscription. `email` (#570) is a later opt-in channel — not resolved yet. +export function resolveNotificationChannels(subscriptions: NotificationSubscriptionRecord[]): NotificationChannel[] { + const badgePaused = subscriptions.some((subscription) => subscription.channel === "badge" && subscription.status === "paused"); + return badgePaused ? [] : ["badge"]; +} + +export function buildChangesRequestedNotification(event: DetectedNotificationEvent): { title: string; body: string } { + const ref = `${event.repoFullName}#${event.pullNumber}`; + const reviewer = event.actorLogin && event.actorLogin !== "unknown" ? `@${event.actorLogin}` : "a reviewer"; + return { + title: sanitizePublicComment(`Changes requested on ${ref}`), + body: sanitizePublicComment(`${reviewer} requested changes on your pull request ${ref}. Address the review feedback to keep it on track to merge.`), + }; +} + +function rateLimitWindowStart(now: string): string { + return new Date(Date.parse(now) - NOTIFICATION_RATE_LIMIT.windowMinutes * 60_000).toISOString(); +} + +// Resolves the recipient's enabled channels and writes one idempotent delivery row per channel. Returns the +// rows that were freshly created with status `pending` (the caller enqueues a deliver job for each). Rows +// that already existed (duplicate webhook/retry) or were rate-limited/suppressed are NOT returned. +export async function evaluateNotificationEvent(env: Env, event: DetectedNotificationEvent): Promise { + const subscriptions = await listNotificationSubscriptionsForLogin(env, event.recipientLogin); + const channels = resolveNotificationChannels(subscriptions); + if (channels.length === 0) return []; + + const { title, body } = buildChangesRequestedNotification(event); + const now = nowIso(); + const windowStart = rateLimitWindowStart(now); + const pending: NotificationDeliveryRecord[] = []; + + for (const channel of channels) { + const recent = await countRecentNotificationDeliveries(env, event.recipientLogin, channel, windowStart); + const status = recent >= NOTIFICATION_RATE_LIMIT.maxPerWindow ? "suppressed" : "pending"; + const { delivery, created } = await insertNotificationDeliveryIfAbsent(env, { + dedupKey: event.dedupKey, + channel, + recipientLogin: event.recipientLogin, + eventType: event.eventType, + repoFullName: event.repoFullName, + pullNumber: event.pullNumber, + title, + body, + deeplink: event.deeplink, + actorLogin: event.actorLogin, + status, + }); + if (created && delivery.status === "pending") pending.push(delivery); + } + return pending; +} + +export type NotificationFeedItem = { + id: string; + eventType: string; + repoFullName: string; + pullNumber: number | null; + title: string; + body: string; + deeplink: string; + status: NotificationDeliveryRecord["status"]; + createdAt: string; +}; + +export type NotificationFeed = { + login: string; + unreadCount: number; + notifications: NotificationFeedItem[]; +}; + +// Shapes the recipient's badge feed: the unread count (the badge number) plus recent items. Only rows that +// reached `delivered` (or already `read`) are shown — `pending`/`suppressed` never surface to the user. +export function buildNotificationFeed(login: string, deliveries: NotificationDeliveryRecord[]): NotificationFeed { + const visible = deliveries.filter((delivery) => delivery.status === "delivered" || delivery.status === "read"); + return { + login: login.toLowerCase(), + unreadCount: visible.filter((delivery) => delivery.status === "delivered").length, + notifications: visible.map((delivery) => ({ + id: delivery.id, + eventType: delivery.eventType, + repoFullName: delivery.repoFullName, + pullNumber: delivery.pullNumber, + title: delivery.title, + body: delivery.body, + deeplink: delivery.deeplink, + status: delivery.status, + createdAt: delivery.createdAt, + })), + }; +} + +// Badge delivery is pull-based: "delivering" just makes the row visible to the recipient's feed (status +// pending -> delivered). Email/web-push (#570) would perform an outbound send here for their channel. +export async function deliverNotification(env: Env, deliveryId: string): Promise { + const delivery = await getNotificationDeliveryById(env, deliveryId); + /* v8 ignore next -- deliver is only enqueued for a row that was just created; the guard protects retries after deletion. */ + if (!delivery || delivery.status !== "pending") return; + // Only the badge channel is resolved today (resolveNotificationChannels), so every delivery is a badge + // delivery — making the row visible to the recipient's feed. Email/web-push (#570) will branch by channel here. + await markNotificationDeliveryDelivered(env, deliveryId); +} diff --git a/src/queue/processors.ts b/src/queue/processors.ts index c719d98306..b1fd86153c 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -81,6 +81,7 @@ import { fetchPublicContributorProfile } from "../github/public"; import { refreshRegistry } from "../registry/sync"; import { buildIssueAdvisory, buildPullRequestAdvisory, evaluateGateCheck } from "../rules/advisory"; import { detectNotificationEvents } from "../notifications/events"; +import { deliverNotification, evaluateNotificationEvent } from "../notifications/service"; import { getOrCreateScoringModelSnapshot, refreshScoringModelSnapshot } from "../scoring/model"; import { buildAndPersistContributorDecisionPack, loadDecisionPackSharedInputs } from "../services/decision-pack"; import { @@ -274,6 +275,14 @@ export async function processJob(env: Env, message: JobMessage): Promise { case "run-agent": await executeAgentRun(env, message.runId); return; + case "notify-evaluate": { + const deliveries = await evaluateNotificationEvent(env, message.event); + await Promise.all(deliveries.map((delivery) => env.JOBS.send({ type: "notify-deliver", requestedBy: "notify-evaluate", deliveryId: delivery.id }))); + return; + } + case "notify-deliver": + await deliverNotification(env, message.deliveryId); + return; case "github-webhook": await processGitHubWebhook(env, message.deliveryId, message.eventName, message.payload); return; @@ -771,6 +780,7 @@ async function processGitHubWebhook(env: Env, deliveryId: string, eventName: str deeplink: notificationEvent.deeplink, }, }); + await env.JOBS.send({ type: "notify-evaluate", requestedBy: "webhook", event: notificationEvent }); } await recordWebhookEvent(env, { diff --git a/src/types.ts b/src/types.ts index 20616f25da..b3a3443b62 100644 --- a/src/types.ts +++ b/src/types.ts @@ -114,6 +114,16 @@ export type JobMessage = type: "run-agent"; requestedBy: "api" | "mcp" | "github_comment" | "test"; runId: string; + } + | { + type: "notify-evaluate"; + requestedBy: "webhook" | "test"; + event: DetectedNotificationEvent; + } + | { + type: "notify-deliver"; + requestedBy: "notify-evaluate" | "test"; + deliveryId: string; }; export type GitHubWebhookPayload = { @@ -1067,6 +1077,54 @@ export type DigestSubscriptionRecord = { updatedAt: string; }; +// Notifications (#535). `badge` is the pull-based extension/harness channel shipped first; `email` +// (#570) is a later opt-in channel. Subscriptions store per-channel opt-out (badge is on by default +// unless a row is `paused`). +export type NotificationChannel = "badge" | "email"; +export type NotificationDeliveryStatus = "pending" | "delivered" | "read" | "suppressed"; +export type NotificationEventType = "pull_request_changes_requested"; + +// A notification-worthy event extracted from a webhook payload (src/notifications/events.ts). +export type DetectedNotificationEvent = { + eventType: NotificationEventType; + recipientLogin: string; + repoFullName: string; + pullNumber: number; + dedupKey: string; + deeplink: string; + actorLogin: string; + detectedAt: string; +}; + +export type NotificationSubscriptionRecord = { + id: string; + login: string; + channel: NotificationChannel; + status: "active" | "paused"; + destination: string | null; + source: string; + createdAt: string; + updatedAt: string; +}; + +export type NotificationDeliveryRecord = { + id: string; + dedupKey: string; + channel: NotificationChannel; + recipientLogin: string; + eventType: string; + repoFullName: string; + pullNumber: number | null; + title: string; + body: string; + deeplink: string; + actorLogin: string | null; + status: NotificationDeliveryStatus; + createdAt: string; + deliveredAt: string | null; + readAt: string | null; +}; + export type CommandFeedbackVote = "useful" | "not_useful"; export type CommandFeedbackSource = "github_reaction" | "app"; diff --git a/test/unit/mcp-notifications.test.ts b/test/unit/mcp-notifications.test.ts new file mode 100644 index 0000000000..6a4310c17b --- /dev/null +++ b/test/unit/mcp-notifications.test.ts @@ -0,0 +1,66 @@ +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; +import { describe, expect, it } from "vitest"; +import { GittensoryMcp } from "../../src/mcp/server"; +import { createSessionForGitHubUser, type AuthIdentity } from "../../src/auth/security"; +import { insertNotificationDeliveryIfAbsent, markNotificationDeliveryDelivered } from "../../src/db/repositories"; +import { createTestEnv } from "../helpers/d1"; + +async function connect(env: Env, identity?: AuthIdentity) { + const server = (identity ? new GittensoryMcp(env, identity) : new GittensoryMcp(env)).createServer(); + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + await server.connect(serverTransport); + const client = new Client({ name: "gittensory-notifications-test", version: "0.1.0" }, { capabilities: {} }); + await client.connect(clientTransport); + return client; +} + +async function seedDelivered(env: Env, recipientLogin: string, dedupKey: string): Promise { + const { delivery } = await insertNotificationDeliveryIfAbsent(env, { + dedupKey, + channel: "badge", + recipientLogin, + eventType: "pull_request_changes_requested", + repoFullName: "owner/repo", + pullNumber: 7, + title: "Changes requested on owner/repo#7", + body: "A reviewer requested changes on your pull request owner/repo#7.", + deeplink: "https://github.com/owner/repo/pull/7", + actorLogin: "reviewer", + }); + await markNotificationDeliveryDelivered(env, delivery.id); +} + +describe("MCP notification tools", () => { + it("lists and clears a contributor's own notifications", async () => { + const env = createTestEnv(); + await seedDelivered(env, "miner", "k1"); + const client = await connect(env); + + const list = await client.callTool({ name: "gittensory_list_notifications", arguments: { login: "miner" } }); + expect(list.isError).toBeFalsy(); + expect((list.structuredContent as { unreadCount: number }).unreadCount).toBe(1); + expect(JSON.stringify(list.structuredContent)).not.toMatch(/wallet|hotkey|reward estimate|trust score/i); + + const read = await client.callTool({ name: "gittensory_mark_notifications_read", arguments: { login: "miner" } }); + expect(read.isError).toBeFalsy(); + expect((read.structuredContent as { marked: number }).marked).toBe(1); + + const after = await client.callTool({ name: "gittensory_list_notifications", arguments: { login: "miner" } }); + expect((after.structuredContent as { unreadCount: number }).unreadCount).toBe(0); + }); + + it("forbids reading or clearing another login's notifications from a scoped session", async () => { + const env = createTestEnv(); + const { session } = await createSessionForGitHubUser(env, { login: "miner", id: 1 }); + const identity: AuthIdentity = { kind: "session", actor: "miner", session }; + const client = await connect(env, identity); + + const list = await client.callTool({ name: "gittensory_list_notifications", arguments: { login: "other" } }); + expect(list.isError).toBe(true); + expect(JSON.stringify(list.content)).toContain("authenticated GitHub login"); + + const read = await client.callTool({ name: "gittensory_mark_notifications_read", arguments: { login: "other" } }); + expect(read.isError).toBe(true); + }); +}); diff --git a/test/unit/notifications-service.test.ts b/test/unit/notifications-service.test.ts new file mode 100644 index 0000000000..24ca57f5d3 --- /dev/null +++ b/test/unit/notifications-service.test.ts @@ -0,0 +1,226 @@ +import { describe, expect, it } from "vitest"; +import { + buildChangesRequestedNotification, + buildNotificationFeed, + deliverNotification, + evaluateNotificationEvent, + NOTIFICATION_RATE_LIMIT, + resolveNotificationChannels, +} from "../../src/notifications/service"; +import { + getNotificationDeliveryById, + insertNotificationDeliveryIfAbsent, + listNotificationDeliveriesForRecipient, + listNotificationSubscriptionsForLogin, + markNotificationDeliveriesRead, + upsertNotificationSubscription, +} from "../../src/db/repositories"; +import { processJob } from "../../src/queue/processors"; +import { createTestEnv } from "../helpers/d1"; +import type { DetectedNotificationEvent, NotificationDeliveryRecord, NotificationSubscriptionRecord } from "../../src/types"; + +function event(overrides: Partial = {}): DetectedNotificationEvent { + return { + eventType: "pull_request_changes_requested", + recipientLogin: "miner", + repoFullName: "owner/repo", + pullNumber: 7, + dedupKey: "changes_requested:owner/repo#7:reviewer:2026-05-28T12:00:00.000Z", + deeplink: "https://github.com/owner/repo/pull/7", + actorLogin: "reviewer", + detectedAt: "2026-05-28T12:00:00.000Z", + ...overrides, + }; +} + +function subscription(overrides: Partial = {}): NotificationSubscriptionRecord { + return { + id: "sub-1", + login: "miner", + channel: "badge", + status: "active", + destination: null, + source: "app", + createdAt: "2026-05-28T00:00:00.000Z", + updatedAt: "2026-05-28T00:00:00.000Z", + ...overrides, + }; +} + +function deliveryRecord(overrides: Partial = {}): NotificationDeliveryRecord { + return { + id: "d1", + dedupKey: "k1", + channel: "badge", + recipientLogin: "miner", + eventType: "pull_request_changes_requested", + repoFullName: "owner/repo", + pullNumber: 7, + title: "t", + body: "b", + deeplink: "https://x", + actorLogin: "reviewer", + status: "delivered", + createdAt: "2026-05-28T12:00:00.000Z", + deliveredAt: "2026-05-28T12:00:00.000Z", + readAt: null, + ...overrides, + }; +} + +describe("notification channel resolution + copy", () => { + it("keeps badge on by default and lets a paused badge subscription mute it", () => { + expect(resolveNotificationChannels([])).toEqual(["badge"]); + expect(resolveNotificationChannels([subscription({ status: "active" })])).toEqual(["badge"]); + expect(resolveNotificationChannels([subscription({ channel: "email", status: "paused" })])).toEqual(["badge"]); + expect(resolveNotificationChannels([subscription({ status: "paused" })])).toEqual([]); + }); + + it("builds public-safe changes-requested copy and falls back when the reviewer is unknown", () => { + const named = buildChangesRequestedNotification(event()); + expect(named.title).toContain("owner/repo#7"); + expect(named.body).toContain("@reviewer"); + + const anon = buildChangesRequestedNotification(event({ actorLogin: "unknown" })); + expect(anon.body).toContain("a reviewer"); + expect(anon.body).not.toContain("@unknown"); + }); + + it("shows only delivered/read rows in the feed and counts only delivered as unread", () => { + const feed = buildNotificationFeed("Miner", [ + deliveryRecord({ id: "a", status: "delivered" }), + deliveryRecord({ id: "b", status: "read" }), + deliveryRecord({ id: "c", status: "pending" }), + deliveryRecord({ id: "d", status: "suppressed" }), + ]); + expect(feed.login).toBe("miner"); + expect(feed.unreadCount).toBe(1); + expect(feed.notifications.map((item) => item.id)).toEqual(["a", "b"]); + }); +}); + +describe("evaluateNotificationEvent", () => { + it("creates exactly one badge delivery and is idempotent on a duplicate event", async () => { + const env = createTestEnv(); + const created = await evaluateNotificationEvent(env, event()); + expect(created).toHaveLength(1); + expect(created[0]).toMatchObject({ channel: "badge", recipientLogin: "miner", status: "pending" }); + + const again = await evaluateNotificationEvent(env, event()); + expect(again).toEqual([]); + + const rows = await listNotificationDeliveriesForRecipient(env, "miner"); + expect(rows).toHaveLength(1); + }); + + it("returns nothing when the recipient has muted the badge channel", async () => { + const env = createTestEnv(); + await upsertNotificationSubscription(env, { login: "miner", channel: "badge", status: "paused" }); + expect(await evaluateNotificationEvent(env, event())).toEqual([]); + expect(await listNotificationDeliveriesForRecipient(env, "miner")).toHaveLength(0); + }); + + it("suppresses deliveries beyond the per-recipient rate-limit window", async () => { + const env = createTestEnv(); + for (let index = 0; index < NOTIFICATION_RATE_LIMIT.maxPerWindow; index += 1) { + await insertNotificationDeliveryIfAbsent(env, { + dedupKey: `prefill-${index}`, + channel: "badge", + recipientLogin: "miner", + eventType: "pull_request_changes_requested", + repoFullName: "owner/repo", + pullNumber: index, + title: "t", + body: "b", + deeplink: "https://x", + actorLogin: "reviewer", + status: "delivered", + }); + } + const created = await evaluateNotificationEvent(env, event({ dedupKey: "over-limit" })); + expect(created).toEqual([]); + const rows = await listNotificationDeliveriesForRecipient(env, "miner"); + expect(rows.find((row) => row.dedupKey === "over-limit")?.status).toBe("suppressed"); + }); +}); + +describe("deliverNotification", () => { + it("transitions a pending badge delivery to delivered and is a no-op otherwise", async () => { + const env = createTestEnv(); + const [pending] = await evaluateNotificationEvent(env, event()); + await deliverNotification(env, pending!.id); + expect((await getNotificationDeliveryById(env, pending!.id))?.status).toBe("delivered"); + + // Re-delivering an already-delivered row and an unknown id are both no-ops. + await deliverNotification(env, pending!.id); + await deliverNotification(env, "does-not-exist"); + expect((await getNotificationDeliveryById(env, pending!.id))?.status).toBe("delivered"); + }); +}); + +describe("notification queue wiring", () => { + it("runs evaluate -> deliver end-to-end through processJob and stays idempotent", async () => { + const enqueued: Array<{ type: string; deliveryId?: string }> = []; + const env = createTestEnv({ + JOBS: { + async send(message: { type: string; deliveryId?: string }) { + enqueued.push(message); + }, + } as unknown as Queue, + }); + + await processJob(env, { type: "notify-evaluate", requestedBy: "test", event: event() }); + const deliverJob = enqueued.find((message) => message.type === "notify-deliver"); + expect(deliverJob?.deliveryId).toBeTruthy(); + + await processJob(env, { type: "notify-deliver", requestedBy: "test", deliveryId: deliverJob!.deliveryId! }); + const delivered = await listNotificationDeliveriesForRecipient(env, "miner", { unreadOnly: true }); + expect(delivered).toHaveLength(1); + expect(delivered[0]?.status).toBe("delivered"); + + // A retried evaluate (same event) enqueues no further deliver jobs. + const before = enqueued.length; + await processJob(env, { type: "notify-evaluate", requestedBy: "test", event: event() }); + expect(enqueued.length).toBe(before); + }); +}); + +describe("notification repository helpers", () => { + it("upserts a subscription, lists it, and updates on conflict", async () => { + const env = createTestEnv(); + const first = await upsertNotificationSubscription(env, { login: "Miner", channel: "badge" }); + expect(first).toMatchObject({ login: "miner", channel: "badge", status: "active" }); + + const paused = await upsertNotificationSubscription(env, { login: "Miner", channel: "badge", status: "paused" }); + expect(paused.status).toBe("paused"); + + const subs = await listNotificationSubscriptionsForLogin(env, "miner"); + expect(subs).toHaveLength(1); + expect(subs[0]?.status).toBe("paused"); + }); + + it("marks a recipient's delivered notifications read, optionally by id, scoped to the recipient", async () => { + const env = createTestEnv(); + const [first] = await evaluateNotificationEvent(env, event({ dedupKey: "a" })); + const [second] = await evaluateNotificationEvent(env, event({ dedupKey: "b", pullNumber: 8 })); + await deliverNotification(env, first!.id); + await deliverNotification(env, second!.id); + + // Mark only the first by id. + expect(await markNotificationDeliveriesRead(env, "miner", [first!.id])).toBe(1); + expect((await getNotificationDeliveryById(env, first!.id))?.status).toBe("read"); + expect((await getNotificationDeliveryById(env, second!.id))?.status).toBe("delivered"); + + // Mark the rest (no ids = all delivered). + expect(await markNotificationDeliveriesRead(env, "miner")).toBe(1); + expect(await markNotificationDeliveriesRead(env, "other-login")).toBe(0); + + const feed = buildNotificationFeed("miner", await listNotificationDeliveriesForRecipient(env, "miner")); + expect(feed.unreadCount).toBe(0); + }); + + it("returns null for an unknown delivery id", async () => { + const env = createTestEnv(); + expect(await getNotificationDeliveryById(env, "missing")).toBeNull(); + }); +}); diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 4d1e257cf8..7ace91aa63 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -3740,7 +3740,14 @@ describe("queue processors", () => { }); it("detects a changes-requested review notification for the PR author", async () => { - const env = createTestEnv(); + const enqueued: Array<{ type: string }> = []; + const env = createTestEnv({ + JOBS: { + async send(message: { type: string }) { + enqueued.push(message); + }, + } as unknown as Queue, + }); await processJob(env, { type: "github-webhook", @@ -3786,6 +3793,10 @@ describe("queue processors", () => { dedupKey: "changes_requested:JSONbored/gittensory#42:maintainer:2026-05-28T12:00:00.000Z", }); expect(JSON.stringify(detected.results[0])).not.toMatch(/trust score|wallet|hotkey|reward estimate|reviewability/i); + + const evaluateJob = enqueued.find((message): message is { type: "notify-evaluate"; event: { recipientLogin: string } } => message.type === "notify-evaluate"); + expect(evaluateJob).toBeDefined(); + expect(evaluateJob!.event.recipientLogin).toBe("contributor"); }); });