feat(notifications): add emit helper + UseSend email dispatch

packages/backend/convex/notifications.ts

@@ -0,0 +1,141 @@
+import { 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 } from './_generated/server';
+
+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<Id<'notifications'>> {
+  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<Id<'notifications'>> =>
+    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,
+    };
+  },
+});

packages/backend/convex/notificationsNode.ts

@@ -0,0 +1,61 @@
+'use node';
+
+import { v } from 'convex/values';
+import { UseSend } from 'usesend-js';
+
+import { internal } from './_generated/api';
+import { internalAction } from './_generated/server';
+
+/**
+ * Best-effort email dispatch for a notification. Runs the getForEmail query,
+ * sends via UseSend, and marks the notification emailed on success. Missing
+ * UseSend env is logged and swallowed — email is never allowed to throw.
+ */
+export const sendNotificationEmail = internalAction({
+  args: { notificationId: v.id('notifications') },
+  handler: async (ctx, { notificationId }) => {
+    const { to, title, body, link } = await ctx.runQuery(
+      internal.notifications.getForEmail,
+      { notificationId },
+    );
+
+    if (!to) return;
+
+    const apiKey = process.env.USESEND_API_KEY;
+    const useSendUrl = process.env.USESEND_URL;
+    const from = process.env.USESEND_FROM_EMAIL;
+    if (!apiKey || !useSendUrl || !from) {
+      console.warn(
+        'Skipping notification email: USESEND_API_KEY, USESEND_URL, and USESEND_FROM_EMAIL must be set.',
+      );
+      return;
+    }
+
+    const text = link ? `${body}\n\n${link}` : body;
+    const html = `
+      <div style="max-width: 600px; margin: 0 auto; font-family: Arial, sans-serif;">
+        <h2>${title}</h2>
+        <p>${body}</p>
+        ${link ? `<p><a href="${link}">${link}</a></p>` : ''}
+      </div>
+    `;
+
+    const useSend = new UseSend(apiKey, useSendUrl);
+    const result = await useSend.emails.send({
+      from,
+      to: [to],
+      subject: title,
+      text,
+      html,
+    });
+
+    if (result.error) {
+      console.error('UseSend error: ' + JSON.stringify(result.error));
+      return;
+    }
+
+    await ctx.runMutation(internal.notifications.markEmailed, {
+      notificationId,
+    });
+  },
+});

packages/backend/tests/unit/notifications.test.ts

@@ -1,6 +1,7 @@
 import { convexTest } from 'convex-test';
 import { describe, expect, test } from 'vitest';
 
+import { internal } from '../../convex/_generated/api';
 import schema from '../../convex/schema';
 
 const modules = import.meta.glob('../../convex/**/*.*s');
@@ -70,3 +71,105 @@ describe('notifications schema', () => {
     expect(typeof preference?.updatedAt).toBe('number');
   });
 });
+
+describe('emitNotification', () => {
+  test('inserts an unread notification row for the user', async () => {
+    const t = convexTest(schema, modules);
+
+    const userId = await t.run(
+      async (ctx) =>
+        await ctx.db.insert('users', { email: 'emit@b.c', name: 'Emit' }),
+    );
+
+    const notificationId = await t.mutation(internal.notifications.emit, {
+      userId,
+      kind: 'agent_turn_finished',
+      title: 'Agent finished',
+      body: 'Your agent completed its turn.',
+      link: '/threads/abc',
+    });
+
+    const notification = await t.run(
+      async (ctx) => await ctx.db.get(notificationId),
+    );
+
+    expect(notification).not.toBeNull();
+    expect(notification?.userId).toBe(userId);
+    expect(notification?.kind).toBe('agent_turn_finished');
+    expect(notification?.title).toBe('Agent finished');
+    expect(notification?.body).toBe('Your agent completed its turn.');
+    expect(notification?.link).toBe('/threads/abc');
+    expect(notification?.readAt).toBeUndefined();
+    expect(notification?.emailedAt).toBeUndefined();
+    expect(typeof notification?.createdAt).toBe('number');
+  });
+
+  test('does not email when emailEnabled is false', async () => {
+    const t = convexTest(schema, modules);
+
+    const userId = await t.run(async (ctx) => {
+      const userId = await ctx.db.insert('users', {
+        email: 'noemail@b.c',
+        name: 'NoEmail',
+      });
+      await ctx.db.insert('notificationPreferences', {
+        userId,
+        emailEnabled: false,
+        updatedAt: Date.now(),
+      });
+      return userId;
+    });
+
+    const notificationId = await t.mutation(internal.notifications.emit, {
+      userId,
+      kind: 'sync_failed',
+      title: 'Sync failed',
+      body: 'Your sync failed.',
+    });
+
+    // No email should have been scheduled; draining is a safe no-op.
+    await t.finishAllScheduledFunctions(() => undefined);
+
+    const notification = await t.run(
+      async (ctx) => await ctx.db.get(notificationId),
+    );
+    expect(notification?.emailedAt).toBeUndefined();
+  });
+
+  test('getForEmail returns the recipient shape and markEmailed patches emailedAt', async () => {
+    const t = convexTest(schema, modules);
+
+    const { notificationId } = await t.run(async (ctx) => {
+      const userId = await ctx.db.insert('users', {
+        email: 'foremail@b.c',
+        name: 'ForEmail',
+      });
+      const notificationId = await ctx.db.insert('notifications', {
+        userId,
+        kind: 'agent_needs_input',
+        title: 'Needs input',
+        body: 'The agent needs your input.',
+        link: '/threads/xyz',
+        createdAt: Date.now(),
+      });
+      return { notificationId };
+    });
+
+    const forEmail = await t.query(internal.notifications.getForEmail, {
+      notificationId,
+    });
+    expect(forEmail).toEqual({
+      to: 'foremail@b.c',
+      title: 'Needs input',
+      body: 'The agent needs your input.',
+      link: '/threads/xyz',
+    });
+
+    await t.mutation(internal.notifications.markEmailed, { notificationId });
+
+    const notification = await t.run(
+      async (ctx) => await ctx.db.get(notificationId),
+    );
+    expect(typeof notification?.emailedAt).toBe('number');
+  });
+});