import { ConvexError, v } from 'convex/values'; import type { Doc, Id } from './_generated/dataModel'; import type { MutationCtx } from './_generated/server'; import { internal } from './_generated/api'; import { internalMutation, internalQuery, mutation, query, } from './_generated/server'; import { getRequiredUserId } from './model'; export type NotificationKind = | 'maintenance_thread' | 'agent_turn_finished' | 'agent_needs_input' | 'sync_failed' | 'connection_needs_reauth'; // Maps a kind to the notificationPreferences field that gates its email. const EMAIL_PREF_FIELD: Record< NotificationKind, keyof Doc<'notificationPreferences'> > = { maintenance_thread: 'emailMaintenance', agent_turn_finished: 'emailAgentTurnFinished', agent_needs_input: 'emailAgentNeedsInput', sync_failed: 'emailSyncFailed', connection_needs_reauth: 'emailConnectionReauth', }; const kindValidator = v.union( v.literal('maintenance_thread'), v.literal('agent_turn_finished'), v.literal('agent_needs_input'), v.literal('sync_failed'), v.literal('connection_needs_reauth'), ); const emitArgs = { userId: v.id('users'), kind: kindValidator, title: v.string(), body: v.string(), link: v.optional(v.string()), spoonId: v.optional(v.id('spoons')), threadId: v.optional(v.id('threads')), }; export interface EmitNotificationArgs { userId: Id<'users'>; kind: NotificationKind; title: string; body: string; link?: string; spoonId?: Id<'spoons'>; threadId?: Id<'threads'>; } /** * Single choke point for creating a notification. Inserts the row, then * best-effort schedules an email when the user's preferences allow it and the * user has an email address. Callable directly inside any mutation; actions go * through the `emit` wrapper below. */ export async function emitNotification( ctx: MutationCtx, args: EmitNotificationArgs, ): Promise> { const notificationId = await ctx.db.insert('notifications', { userId: args.userId, kind: args.kind, title: args.title, body: args.body, link: args.link, spoonId: args.spoonId, threadId: args.threadId, createdAt: Date.now(), }); const prefs = await ctx.db .query('notificationPreferences') .withIndex('by_user', (q) => q.eq('userId', args.userId)) .unique(); // Unset preferences default to enabled, so email is allowed unless a stored // preference explicitly opts out (globally via emailEnabled, or per-kind). const emailAllowed = prefs?.emailEnabled !== false && prefs?.[EMAIL_PREF_FIELD[args.kind]] !== false; if (emailAllowed) { const user = await ctx.db.get(args.userId); if (user?.email) { await ctx.scheduler.runAfter( 0, internal.notificationsNode.sendNotificationEmail, { notificationId }, ); } } return notificationId; } /** * Thin internalMutation wrapper so `'use node'` actions (e.g. githubSync) can * emit via `ctx.runMutation(internal.notifications.emit, ...)`. */ export const emit = internalMutation({ args: emitArgs, handler: async (ctx, args): Promise> => await emitNotification(ctx, args), }); export const markEmailed = internalMutation({ args: { notificationId: v.id('notifications') }, handler: async (ctx, { notificationId }) => { await ctx.db.patch(notificationId, { emailedAt: Date.now() }); }, }); export const getForEmail = internalQuery({ args: { notificationId: v.id('notifications') }, handler: async ( ctx, { notificationId }, ): Promise<{ to: string | null; title: string; body: string; link: string | null; }> => { const notification = await ctx.db.get(notificationId); if (!notification) { return { to: null, title: '', body: '', link: null }; } const user = await ctx.db.get(notification.userId); return { to: user?.email ?? null, title: notification.title, body: notification.body, link: notification.link ?? null, }; }, }); /** * The authed caller's notifications, newest-first. Scoped to the caller via * `getRequiredUserId`; never accepts a userId argument. */ export const listMine = query({ args: { limit: v.optional(v.number()) }, handler: async (ctx, args): Promise[]> => { const userId = await getRequiredUserId(ctx); return await ctx.db .query('notifications') .withIndex('by_user_created', (q) => q.eq('userId', userId)) .order('desc') .take(args.limit ?? 30); }, }); /** * How many of the caller's notifications are unread. Bounded at 100 via the * `by_user_unread` index so the header bell stays cheap. */ export const unreadCount = query({ args: {}, handler: async (ctx): Promise => { const userId = await getRequiredUserId(ctx); const unread = await ctx.db .query('notifications') .withIndex('by_user_unread', (q) => q.eq('userId', userId).eq('readAt', undefined), ) .take(100); return unread.length; }, }); /** * Marks a single notification read. Loads the row and asserts it belongs to the * caller before patching, so a user can never mark another user's notification * read (indistinguishable from a missing row). */ export const markRead = mutation({ args: { notificationId: v.id('notifications') }, handler: async (ctx, { notificationId }): Promise => { const userId = await getRequiredUserId(ctx); const notification = await ctx.db.get(notificationId); if (notification?.userId !== userId) { throw new ConvexError('Notification not found.'); } if (notification.readAt === undefined) { await ctx.db.patch(notificationId, { readAt: Date.now() }); } }, }); /** * Marks all of the caller's unread notifications read. Iterates the caller's * unread rows via the `by_user_unread` index; only ever touches the caller's * own notifications. */ export const markAllRead = mutation({ args: {}, handler: async (ctx): Promise => { const userId = await getRequiredUserId(ctx); const unread = await ctx.db .query('notifications') .withIndex('by_user_unread', (q) => q.eq('userId', userId).eq('readAt', undefined), ) .collect(); const now = Date.now(); for (const notification of unread) { await ctx.db.patch(notification._id, { readAt: now }); } }, });