Implement tasks 40, 41, 42: web push, notification bus, reminders engine

Task 40 — Web Push (VAPID): - Add web-push package + @types/web-push - pnpm vapid:generate script prints VAPID_PUBLIC_KEY, VAPID_PRIVATE_KEY, NEXT_PUBLIC_VAPID_PUBLIC_KEY - push_subscriptions schema + migration 0013 - _core/push.ts: sendPush() iterates subscriptions, prunes 404/410 stale entries - SW push/notificationclick event handlers added to generated sw.js template - PushOptIn client component on /settings (opt-in, disable, send test) Task 42 — Notification bus + ntfy adapter: - notifications table + notif_push/notif_inapp/notif_ntfy user columns (migration 0013) - _core/notify.ts: notify() fans out to push, in-app DB, and optional ntfy POST - NotificationBell server component in AppNav: unread badge, dropdown inbox, mark-read - NotifyChannelToggles client component in /settings Task 41 — Reminders engine: - fired_at + created_by added to reminders; default channel changed to 'auto' - _core/reminders.ts: scheduleReminder (upsert), cancelReminder, listReminders, tickReminders - tickReminders uses pg_try_advisory_xact_lock for horizontal-scale safety - src/instrumentation.ts starts reminder worker (30s tick) on Node.js boot - Notes actions use scheduleReminder/cancelReminder instead of raw SQL - Calendar createEvent: optional remindMinutesBefore, deleteEvent: cancelReminder - Calendar-shell: "Remind me 30 min before" checkbox on new event form Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-06 16:55:02 -05:00
parent ec20f1bc87
commit b6abe052c9
25 changed files with 986 additions and 85 deletions
@@ -15,3 +15,6 @@ export type { QuickAddItem, SerializedQuickAddItem, SerializedWidgetMeta } from
 export { logActivity, logShareActivity } from "./activity";
 export { createShareLink, resolveShareToken, revokeShareLink } from "./share";
 export type { ShareLinkCapabilities, CreateShareLinkResult } from "./share";
+export { sendPush } from "./push";
+export { notify } from "./notify";
+export { scheduleReminder, cancelReminder, listReminders, startReminderWorker } from "./reminders";
@@ -0,0 +1,52 @@
+import { eq } from "drizzle-orm";
+import { db } from "@/lib/db";
+import { notifications, users } from "./schema";
+import { sendPush } from "./push";
+
+type NotifyPayload = {
+  title: string;
+  body: string;
+  url?: string;
+  channels?: ("push" | "inapp" | "ntfy")[];
+};
+
+export async function notify(userId: string, payload: NotifyPayload) {
+  const channels = payload.channels ?? ["push", "inapp"];
+
+  const [user] = await db
+    .select({ notifPush: users.notifPush, notifInApp: users.notifInApp, notifNtfy: users.notifNtfy })
+    .from(users)
+    .where(eq(users.id, userId))
+    .limit(1);
+
+  if (!user) return;
+
+  const pushEnabled = process.env["VAPID_PUBLIC_KEY"] && process.env["VAPID_PRIVATE_KEY"];
+
+  if (channels.includes("push") && user.notifPush && pushEnabled) {
+    await sendPush(userId, payload).catch((err) =>
+      console.error("[famapp] push channel failed:", err),
+    );
+  }
+
+  if (channels.includes("inapp") && user.notifInApp) {
+    await db.insert(notifications).values({
+      userId,
+      title: payload.title,
+      body: payload.body,
+      url: payload.url,
+    });
+  }
+
+  if (channels.includes("ntfy") && user.notifNtfy) {
+    const ntfyUrl = process.env["NTFY_URL"];
+    const ntfyTopic = process.env["NTFY_TOPIC"];
+    if (ntfyUrl && ntfyTopic) {
+      await fetch(`${ntfyUrl}/${ntfyTopic}`, {
+        method: "POST",
+        headers: { Title: payload.title, "Content-Type": "text/plain" },
+        body: payload.body,
+      }).catch((err) => console.error("[famapp] ntfy delivery failed:", err));
+    }
+  }
+}
@@ -0,0 +1,54 @@
+import webPush from "web-push";
+import { and, eq, inArray } from "drizzle-orm";
+import { db } from "@/lib/db";
+import { pushSubscriptions } from "./schema";
+
+function ensureVapidConfigured() {
+  const subject = process.env["VAPID_SUBJECT"];
+  const publicKey = process.env["VAPID_PUBLIC_KEY"];
+  const privateKey = process.env["VAPID_PRIVATE_KEY"];
+  if (!subject || !publicKey || !privateKey) {
+    throw new Error("VAPID_SUBJECT, VAPID_PUBLIC_KEY, and VAPID_PRIVATE_KEY must be set");
+  }
+  webPush.setVapidDetails(subject, publicKey, privateKey);
+}
+
+export async function sendPush(
+  userId: string,
+  payload: { title: string; body: string; url?: string },
+) {
+  ensureVapidConfigured();
+
+  const subs = await db
+    .select()
+    .from(pushSubscriptions)
+    .where(eq(pushSubscriptions.userId, userId));
+
+  if (subs.length === 0) return;
+
+  const staleIds: string[] = [];
+
+  await Promise.allSettled(
+    subs.map(async (sub) => {
+      try {
+        await webPush.sendNotification(
+          { endpoint: sub.endpoint, keys: { p256dh: sub.p256dh, auth: sub.auth } },
+          JSON.stringify({ title: payload.title, body: payload.body, url: payload.url ?? "/" }),
+        );
+      } catch (err) {
+        const status = (err as { statusCode?: number }).statusCode;
+        if (status === 404 || status === 410) {
+          staleIds.push(sub.id);
+        } else {
+          console.error("[famapp] push delivery failed:", err);
+        }
+      }
+    }),
+  );
+
+  if (staleIds.length > 0) {
+    await db
+      .delete(pushSubscriptions)
+      .where(and(eq(pushSubscriptions.userId, userId), inArray(pushSubscriptions.id, staleIds)));
+  }
+}
@@ -0,0 +1,99 @@
+import { and, eq, inArray, isNull, lte, sql } from "drizzle-orm";
+import { db } from "@/lib/db";
+import { reminders } from "./schema";
+import { notify } from "./notify";
+
+const REMINDER_LOCK_KEY = 7_777_777;
+
+export async function scheduleReminder(input: {
+  householdId: string;
+  entityType: string;
+  entityId: string;
+  fireAt: Date;
+  createdBy: string;
+  channel?: string;
+}) {
+  await db
+    .insert(reminders)
+    .values({
+      householdId: input.householdId,
+      entityType: input.entityType,
+      entityId: input.entityId,
+      fireAt: input.fireAt,
+      channel: input.channel ?? "auto",
+      createdBy: input.createdBy,
+      firedAt: null,
+    })
+    .onConflictDoUpdate({
+      target: [reminders.entityType, reminders.entityId],
+      set: { fireAt: input.fireAt, firedAt: null, createdBy: input.createdBy },
+    });
+}
+
+export async function cancelReminder(entityType: string, entityId: string) {
+  await db
+    .delete(reminders)
+    .where(and(eq(reminders.entityType, entityType), eq(reminders.entityId, entityId)));
+}
+
+export async function listReminders(entityType: string, entityId: string) {
+  return db
+    .select()
+    .from(reminders)
+    .where(and(eq(reminders.entityType, entityType), eq(reminders.entityId, entityId)));
+}
+
+export async function tickReminders() {
+  let dueReminders: (typeof reminders.$inferSelect)[] = [];
+
+  try {
+    await db.transaction(async (tx) => {
+      const lockRows = await tx.execute<{ acquired: boolean }>(
+        sql`SELECT pg_try_advisory_xact_lock(${REMINDER_LOCK_KEY}) AS acquired`,
+      );
+      if (!lockRows[0]?.acquired) return;
+
+      const now = new Date();
+      dueReminders = await tx
+        .select()
+        .from(reminders)
+        .where(and(lte(reminders.fireAt, now), isNull(reminders.firedAt)));
+
+      if (dueReminders.length > 0) {
+        await tx
+          .update(reminders)
+          .set({ firedAt: now })
+          .where(inArray(reminders.id, dueReminders.map((r) => r.id)));
+      }
+    });
+  } catch (err) {
+    console.error("[famapp] reminder tick error:", err);
+    return;
+  }
+
+  await Promise.allSettled(
+    dueReminders.map(async (reminder) => {
+      if (!reminder.createdBy) return;
+      try {
+        await notify(reminder.createdBy, {
+          title: "Reminder",
+          body: `You have a reminder`,
+          url: reminder.entityType === "notes.note" ? `/notes/${reminder.entityId}` : "/",
+          channels: ["push", "inapp"],
+        });
+      } catch (err) {
+        console.error("[famapp] reminder delivery failed:", reminder.id, err);
+      }
+    }),
+  );
+}
+
+let workerTimer: ReturnType<typeof setInterval> | null = null;
+
+export function startReminderWorker() {
+  if (workerTimer) return;
+  workerTimer = setInterval(() => {
+    tickReminders().catch((err) => console.error("[famapp] reminder worker uncaught:", err));
+  }, 30_000);
+  console.log("[famapp] reminder worker started (30s tick)");
+}
@@ -26,6 +26,9 @@ export const users = pgTable("users", {
  theme: text("theme").notNull().default("default"),
  themeMode: text("theme_mode").notNull().default("system"),
  completionVisibilityHours: integer("completion_visibility_hours").notNull().default(24),
+  notifPush: boolean("notif_push").notNull().default(true),
+  notifInApp: boolean("notif_inapp").notNull().default(true),
+  notifNtfy: boolean("notif_ntfy").notNull().default(false),
  createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
 });

@@ -158,7 +161,9 @@ export const reminders = pgTable(
    entityType: text("entity_type").notNull(),
    entityId: uuid("entity_id").notNull(),
    fireAt: timestamp("fire_at", { withTimezone: true }).notNull(),
-    channel: text("channel").notNull().default("in_app"),
+    channel: text("channel").notNull().default("auto"),
+    firedAt: timestamp("fired_at", { withTimezone: true }),
+    createdBy: uuid("created_by").references(() => users.id, { onDelete: "set null" }),
    createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
  },
  (t) => [
@@ -166,3 +171,35 @@ export const reminders = pgTable(
    index("reminders_household_fire_at_idx").on(t.householdId, t.fireAt),
  ],
 );
+
+export const pushSubscriptions = pgTable(
+  "push_subscriptions",
+  {
+    id: uuid("id").primaryKey().defaultRandom(),
+    userId: uuid("user_id")
+      .notNull()
+      .references(() => users.id, { onDelete: "cascade" }),
+    endpoint: text("endpoint").notNull().unique(),
+    p256dh: text("p256dh").notNull(),
+    auth: text("auth").notNull(),
+    userAgent: text("user_agent"),
+    createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
+  },
+  (t) => [index("push_subscriptions_user_idx").on(t.userId)],
+);
+
+export const notifications = pgTable(
+  "notifications",
+  {
+    id: uuid("id").primaryKey().defaultRandom(),
+    userId: uuid("user_id")
+      .notNull()
+      .references(() => users.id, { onDelete: "cascade" }),
+    title: text("title").notNull(),
+    body: text("body").notNull(),
+    url: text("url"),
+    readAt: timestamp("read_at", { withTimezone: true }),
+    createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
+  },
+  (t) => [index("notifications_user_read_idx").on(t.userId, t.readAt)],
+);
@@ -40,6 +40,7 @@ type EventDraft = {
  allDay: boolean;
  location: string;
  notes: string;
+  remindMinutesBefore: number | null;
 };

 const DEFAULT_COLOR = "#2563eb";
@@ -92,6 +93,7 @@ export function CalendarShell({
      allDay,
      location: "",
      notes: "",
+      remindMinutesBefore: 30,
    });
  }

@@ -107,6 +109,7 @@ export function CalendarShell({
      allDay: row.allDay,
      location: row.location ?? "",
      notes: row.notes ?? "",
+      remindMinutesBefore: null,
    });
  }

@@ -148,7 +151,10 @@ export function CalendarShell({
          ),
        );
      } else {
-        const created = await createEvent(payload);
+        const created = await createEvent({
+          ...payload,
+          remindMinutesBefore: selectedEvent.remindMinutesBefore,
+        });
        setLastCalendarId(payload.calendarId);
        setEventRows((current) => [...current, created]);
      }
@@ -489,6 +495,22 @@ export function CalendarShell({
                  setSelectedEvent({ ...selectedEvent, notes: event.target.value })
                }
              />
+              {!selectedEvent.id && (
+                <label className="flex items-center gap-2 text-sm">
+                  <input
+                    type="checkbox"
+                    className="size-4 cursor-pointer"
+                    checked={selectedEvent.remindMinutesBefore !== null}
+                    onChange={(e) =>
+                      setSelectedEvent({
+                        ...selectedEvent,
+                        remindMinutesBefore: e.target.checked ? 30 : null,
+                      })
+                    }
+                  />
+                  Remind me 30 min before
+                </label>
+              )}
              <div className="flex items-center justify-between gap-2 pt-2">
                <div className="flex gap-2">
                  {selectedEvent.id && (
@@ -6,6 +6,7 @@ import { z } from "zod";
 import { db } from "@/lib/db";
 import { getCurrentSession } from "@/lib/session";
 import { logActivity } from "@/modules/_core/activity";
+import { scheduleReminder, cancelReminder } from "@/modules/_core/reminders";
 import { calendarEvents, calendars } from "../schema";
 import { canSeeCalendar } from "./queries";

@@ -23,6 +24,7 @@ const eventBaseInput = z.object({
  allDay: z.boolean().default(false),
  location: z.string().trim().max(300).nullable().optional(),
  notes: z.string().trim().max(3000).nullable().optional(),
+  remindMinutesBefore: z.number().int().min(0).nullable().optional(),
 });

 const eventInput = eventBaseInput.refine((value) => value.endAt >= value.startAt, {
@@ -115,13 +117,17 @@ export async function deleteCalendar(input: { id: string }) {

 export async function createEvent(input: z.input<typeof eventInput>) {
  const parsed = eventInput.parse(input);
-  const { user } = await getCurrentSession();
+  const { user, household } = await getCurrentSession();
  if (!(await canSeeCalendar(user.id, parsed.calendarId))) throw new Error("Forbidden");

  const [event] = await db
    .insert(calendarEvents)
    .values({
-      ...parsed,
+      calendarId: parsed.calendarId,
+      title: parsed.title,
+      startAt: parsed.startAt,
+      endAt: parsed.endAt,
+      allDay: parsed.allDay,
      ownerId: user.id,
      location: parsed.location || null,
      notes: parsed.notes || null,
@@ -129,6 +135,20 @@ export async function createEvent(input: z.input<typeof eventInput>) {
    .returning();

  if (!event) throw new Error("Event was not created");
+
+  if (parsed.remindMinutesBefore != null) {
+    const fireAt = new Date(parsed.startAt.getTime() - parsed.remindMinutesBefore * 60_000);
+    if (fireAt > new Date()) {
+      await scheduleReminder({
+        householdId: household.id,
+        entityType: "calendar.event",
+        entityId: event.id,
+        fireAt,
+        createdBy: user.id,
+      });
+    }
+  }
+
  await logActivity({ entityType: "calendar.event", entityId: event.id, action: "create", payload: { title: event.title } });
  revalidatePath("/calendar");
  return {
@@ -181,6 +201,7 @@ export async function deleteEvent(input: { id: string }) {
  if (!existing) return;
  if (!(await canSeeCalendar(user.id, existing.calendarId))) throw new Error("Forbidden");
  await logActivity({ entityType: "calendar.event", entityId: parsed.id, action: "delete" });
+  await cancelReminder("calendar.event", parsed.id);
  await db.delete(calendarEvents).where(eq(calendarEvents.id, parsed.id));
  revalidatePath("/calendar");
 }
@@ -1,12 +1,12 @@
 "use server";

-import { and, eq } from "drizzle-orm";
+import { eq } from "drizzle-orm";
 import { revalidatePath } from "next/cache";
 import { z } from "zod";
 import { db } from "@/lib/db";
 import { getCurrentSession } from "@/lib/session";
 import { logActivity } from "@/modules/_core/activity";
-import { reminders } from "@/modules/_core/schema";
+import { scheduleReminder, cancelReminder } from "@/modules/_core/reminders";
 import { notes } from "../schema";
 import { canAccessNote, getNote } from "./queries";

@@ -29,63 +29,69 @@ export async function createNote(input: z.input<typeof noteInput>) {
  const parsed = noteInput.parse(input);
  const { household, user } = await getCurrentSession();

-  const [note] = await db.transaction(async (tx) => {
-    const [created] = await tx
-      .insert(notes)
-      .values({
-        householdId: household.id,
-        authorId: user.id,
-        title: parsed.title,
-        body: parsed.body,
-        pinned: parsed.pinned,
-        remindAt: parsed.remindAt ?? null,
-      })
-      .returning();
-
-    if (!created) throw new Error("Note was not created");
-    await syncNoteReminder(tx, {
+  const [note] = await db
+    .insert(notes)
+    .values({
      householdId: household.id,
-      noteId: created.id,
-      remindAt: created.remindAt,
-    });
-    return [created];
-  });
+      authorId: user.id,
+      title: parsed.title,
+      body: parsed.body,
+      pinned: parsed.pinned,
+      remindAt: parsed.remindAt ?? null,
+    })
+    .returning();

-  if (note) await logActivity({ entityType: "notes.note", entityId: note.id, action: "create", payload: { title: note.title } });
+  if (!note) throw new Error("Note was not created");
+
+  if (parsed.remindAt) {
+    await scheduleReminder({
+      householdId: household.id,
+      entityType: "notes.note",
+      entityId: note.id,
+      fireAt: parsed.remindAt,
+      createdBy: user.id,
+    });
+  }
+
+  await logActivity({ entityType: "notes.note", entityId: note.id, action: "create", payload: { title: note.title } });
  revalidatePath("/notes");
  return note;
 }

 export async function updateNote(input: z.input<typeof updateNoteInput>) {
  const parsed = updateNoteInput.parse(input);
-  const { household } = await getCurrentSession();
+  const { household, user } = await getCurrentSession();
  await assertCanAccessNote(parsed.id, household.id);

-  const [note] = await db.transaction(async (tx) => {
-    const [updated] = await tx
-      .update(notes)
-      .set({
-        title: parsed.title,
-        body: parsed.body,
-        pinned: parsed.pinned,
-        remindAt: parsed.remindAt === undefined ? undefined : parsed.remindAt,
-        updatedAt: new Date(),
-      })
-      .where(eq(notes.id, parsed.id))
-      .returning();
+  const [note] = await db
+    .update(notes)
+    .set({
+      title: parsed.title,
+      body: parsed.body,
+      pinned: parsed.pinned,
+      remindAt: parsed.remindAt === undefined ? undefined : parsed.remindAt,
+      updatedAt: new Date(),
+    })
+    .where(eq(notes.id, parsed.id))
+    .returning();

-    if (!updated) throw new Error("Note was not updated");
-    if (parsed.remindAt !== undefined) {
-      await syncNoteReminder(tx, {
+  if (!note) throw new Error("Note was not updated");
+
+  if (parsed.remindAt !== undefined) {
+    if (parsed.remindAt) {
+      await scheduleReminder({
        householdId: household.id,
-        noteId: updated.id,
-        remindAt: updated.remindAt,
+        entityType: "notes.note",
+        entityId: note.id,
+        fireAt: parsed.remindAt,
+        createdBy: user.id,
      });
+    } else {
+      await cancelReminder("notes.note", note.id);
    }
-    return [updated];
-  });
+  }

-  if (note) await logActivity({ entityType: "notes.note", entityId: note.id, action: "update", payload: { title: note.title } });
+  await logActivity({ entityType: "notes.note", entityId: note.id, action: "update", payload: { title: note.title } });
  revalidatePath("/notes");
  revalidatePath(`/notes/${parsed.id}`);
  return note;
@@ -116,12 +122,8 @@ export async function deleteNote(input: { id: string }) {
  const note = await getNote(parsed.id);
  await logActivity({ entityType: "notes.note", entityId: parsed.id, action: "delete", payload: note ? { title: note.title } : undefined });

-  await db.transaction(async (tx) => {
-    await tx
-      .delete(reminders)
-      .where(and(eq(reminders.entityType, "notes.note"), eq(reminders.entityId, parsed.id)));
-    await tx.delete(notes).where(eq(notes.id, parsed.id));
-  });
+  await cancelReminder("notes.note", parsed.id);
+  await db.delete(notes).where(eq(notes.id, parsed.id));

  revalidatePath("/notes");
 }
@@ -129,22 +131,3 @@ export async function deleteNote(input: { id: string }) {
 async function assertCanAccessNote(noteId: string, householdId: string) {
  if (!(await canAccessNote(noteId, householdId))) throw new Error("Forbidden");
 }
-
-async function syncNoteReminder(
-  tx: Parameters<Parameters<typeof db.transaction>[0]>[0],
-  input: { householdId: string; noteId: string; remindAt: Date | null },
-) {
-  await tx
-    .delete(reminders)
-    .where(and(eq(reminders.entityType, "notes.note"), eq(reminders.entityId, input.noteId)));
-
-  if (!input.remindAt) return;
-
-  await tx.insert(reminders).values({
-    householdId: input.householdId,
-    entityType: "notes.note",
-    entityId: input.noteId,
-    fireAt: input.remindAt,
-    channel: "in_app",
-  });
-}