Agent/notification domain

.env.example

@@ -27,6 +27,8 @@ PAYSTACK_DVA_PREFERRED_BANK=test-bank
 
 # Email (mock mode below skips Resend and does not require this key)
 RESEND_API_KEY=replace_with_resend_key
+NOTIFICATION_TOKEN_SECRET=replace_with_a_separate_random_secret
+NOTIFICATION_WORKER_SECRET=replace_with_a_separate_worker_secret
 
 # File uploads / Vercel Blob
 BLOB_READ_WRITE_TOKEN=replace_with_local_or_preview_blob_token

actions/user.ts

@@ -8,6 +8,7 @@ import { sendEmail } from "@/lib/services/email.service"
 import { logAuditEvent } from "@/lib/security/audit-log"
 import { isSupportedKycDocumentReference } from "@/lib/security/kyc-documents"
 import User from "@/models/User"
+import { publishNotificationEvent } from "@/lib/notifications/service"
 
 type KycStatus = "none" | "pending" | "approved_stage1" | "pending_stage2" | "approved_stage2" | "rejected"
 type PhysicalMeetingStatus = "none" | "scheduled" | "approved" | "rescheduled" | "completed" | "rejected_stage2"
@@ -50,9 +51,6 @@ function normalizeReason(reason: string | null | undefined) {
   return trimmed.length > 0 ? trimmed.slice(0, 500) : null
 }
 
-function buildNotificationId() {
-  return `notif_${Date.now()}_${Math.random().toString(36).slice(2, 10)}`
-}
 
 function formatMeetingDate(date: Date | null) {
   if (!date) return "your scheduled date"
@@ -436,18 +434,6 @@ export async function updateUserKycStatus(
       reason: normalizedReason,
     })
 
-    if (notification) {
-      user.notifications = Array.isArray(user.notifications) ? user.notifications : []
-      user.notifications.push({
-        id: buildNotificationId(),
-        title: notification.title,
-        message: notification.message,
-        read: false,
-        timestamp: new Date(),
-        link: KYC_NOTIFICATION_LINK[userRole],
-      })
-    }
-
     await user.save()
 
     await logAuditEvent({
@@ -464,7 +450,12 @@ export async function updateUserKycStatus(
     })
 
     if (notification) {
-      await sendKycEmail(user, notification.subject, notification.emailMessage)
+      const decision = user.kycStatus === "rejected" ? "rejected" : user.kycStatus === "approved_stage2" ? "approved" : "review"
+      await publishNotificationEvent({
+        type: "kyc.decision", version: 1, userId: user._id.toString(),
+        eventId: `kyc:${user._id}:${user.kycStatus}:${user.physicalMeetingStatus || "none"}`,
+        occurredAt: new Date().toISOString(), payload: { decision },
+      })
     }
 
     revalidatePath("/dashboard/driver")

app/api/admin/notification-deliveries/route.ts

@@ -0,0 +1,23 @@
+import { NextResponse } from "next/server"
+import { z } from "zod"
+import { requireAuthenticatedUser, finalizeAuthenticatedResponse } from "@/lib/api/route-guard"
+import { parseSearchParams } from "@/lib/api/validation"
+import dbConnect from "@/lib/dbConnect"
+import NotificationDelivery from "@/models/NotificationDelivery"
+
+const querySchema = z.object({
+  userId: z.string().regex(/^[a-f\d]{24}$/i).optional(),
+  status: z.enum(["created", "scheduled", "processing", "delivered", "dead_letter"]).optional(),
+  limit: z.coerce.number().int().min(1).max(100).default(50),
+})
+
+export async function GET(request: Request) {
+  const auth = await requireAuthenticatedUser(request, ["admin"], { forbiddenMessage: "Admin access required" })
+  if ("response" in auth) return auth.response
+  const query = parseSearchParams(request, querySchema)
+  if ("response" in query) return query.response
+  await dbConnect()
+  const filter = { ...(query.data.userId ? { userId: query.data.userId } : {}), ...(query.data.status ? { status: query.data.status } : {}) }
+  const deliveries = await NotificationDelivery.find(filter).select("-html -to").sort({ createdAt: -1 }).limit(query.data.limit).lean()
+  return finalizeAuthenticatedResponse(NextResponse.json({ success: true, deliveries }), auth)
+}

app/api/notifications/process/route.ts

@@ -0,0 +1,8 @@
+import { NextResponse } from "next/server"
+import { processEmailJobs } from "@/lib/notifications/service"
+
+export async function POST(request: Request) {
+  const secret = process.env.NOTIFICATION_WORKER_SECRET
+  if (!secret || request.headers.get("authorization") !== `Bearer ${secret}`) return NextResponse.json({ error: "Unauthorized" }, { status: 401 })
+  return NextResponse.json(await processEmailJobs())
+}

docs/notifications.md

@@ -0,0 +1,29 @@
+# Notification domain
+
+## Event taxonomy
+
+Producers publish versioned events and never call email providers. Version 1 covers funding, investment confirmations, repayments, KYC decisions, payouts, arrears, and contract changes. `eventId` is a stable business-event ID. Payloads contain display-safe labels only—never KYC documents, secrets, bank details, or full financial records.
+
+## Preference matrix
+
+| Category | In-app | Email | Mandatory |
+| --- | --- | --- | --- |
+| Funding | On | On | No |
+| Investment | On | On | No |
+| Repayment | On | On | Due notices |
+| KYC | On | On | Decisions |
+| Payout | On | On | Status changes |
+| Arrears | On | On | Yes |
+| Contract | On | On | Yes |
+
+Mandatory operational notices ignore disabled preferences. Preference links use signed, expiring tokens scoped to one user, category, and email channel; they do not grant account access.
+
+## Template rules
+
+Templates are keyed by event type and integer version. Rendering is deterministic, validates payloads, escapes HTML, and falls back to English while remaining locale-ready. Links use `NEXT_PUBLIC_APP_URL`, require HTTPS outside localhost, and accept internal paths only. Messages direct users to authenticated pages for sensitive details.
+
+## Delivery lifecycle
+
+`publishNotificationEvent` creates one `NotificationDelivery` per channel. The unique `{eventId}:{userId}:{channel}` key makes duplicate events harmless. In-app delivery is independent of email. Email progresses `scheduled → processing → delivered`; failures retry with exponential backoff and enter `dead_letter` after five attempts. Attempts retain timestamps, provider IDs, and redacted errors. A scheduler calls `POST /api/notifications/process` with `Authorization: Bearer $NOTIFICATION_WORKER_SECRET`.
+
+User notification reads are user-scoped; admins may inspect a specified user. Delivery history and dead letters are operational data and must only be exposed to authorized admins.

lib/notifications/__tests__/notification-domain.test.ts

@@ -0,0 +1,21 @@
+import { describe, expect, it } from "vitest"
+import { retryDecision, markRead } from "../policy"
+import { renderNotification } from "../render"
+import { createPreferenceToken, verifyPreferenceToken } from "../security"
+import { enabledChannels, redactError } from "../service"
+import { safeApplicationUrl } from "../templates"
+
+const event = { type: "kyc.decision", version: 1, eventId: "kyc:user:approved", userId: "507f1f77bcf86cd799439011", occurredAt: "2026-01-01T00:00:00Z", payload: { decision: "approved" } } as const
+
+describe("notification domain", () => {
+  it("respects preferences for optional events", () => expect(enabledChannels({ category: "funding", mandatory: false }, { locale: "en", categories: { funding: { email: false } } })).toEqual(["in_app"]))
+  it("keeps mandatory notices enabled", () => expect(enabledChannels({ category: "kyc", mandatory: true }, { locale: "en", categories: { kyc: { email: false, in_app: false } } })).toEqual(["in_app", "email"]))
+  it("uses a deterministic per-channel idempotency shape", () => expect(`${event.eventId}:${event.userId}:email`).toBe(`${event.eventId}:${event.userId}:email`))
+  it("renders deterministically without sensitive KYC details", () => { expect(renderNotification(event)).toEqual(renderNotification(event)); expect(renderNotification(event).html).not.toMatch(/document|passport|secret/i) })
+  it("validates template payloads", () => expect(() => renderNotification({ ...event, payload: { decision: "unknown" as never } })).toThrow())
+  it("rejects unsafe links", () => { expect(() => safeApplicationUrl("//evil.example", "https://chainmove.xyz")).toThrow(); expect(() => safeApplicationUrl("/ok", "http://evil.example")).toThrow() })
+  it("expires scoped preference tokens", () => { const token = createPreferenceToken({ userId: event.userId, category: "funding", channel: "email", exp: 10 }, "secret"); expect(verifyPreferenceToken(token, "secret", 9_000).category).toBe("funding"); expect(() => verifyPreferenceToken(token, "secret", 11_000)).toThrow("Expired") })
+  it("backs off before entering dead letter", () => { const retry = retryDecision(0, 3, new Date(0)); expect(retry.status).toBe("scheduled"); expect(retry.scheduledFor?.getTime()).toBe(60_000); expect(retryDecision(2, 3).status).toBe("dead_letter") })
+  it("changes only selected notifications to read", () => expect(markRead([{ id: "a", read: false }, { id: "b", read: false }], ["a"])).toEqual([{ id: "a", read: true }, { id: "b", read: false }]))
+  it("redacts PII and secrets from provider errors", () => expect(redactError(new Error("john@example.com token=abc document=passport"))).toBe("[REDACTED_EMAIL] token=[REDACTED] document=[REDACTED]"))
+})

lib/notifications/policy.ts

@@ -0,0 +1,10 @@
+export function retryDecision(attemptCount: number, maxAttempts: number, now = new Date()) {
+  const nextCount = attemptCount + 1
+  if (nextCount >= maxAttempts) return { status: "dead_letter" as const, nextCount, scheduledFor: null }
+  return { status: "scheduled" as const, nextCount, scheduledFor: new Date(now.getTime() + Math.min(3600, 30 * 2 ** nextCount) * 1000) }
+}
+
+export function markRead<T extends { id: string; read: boolean }>(records: T[], ids: string[]) {
+  const selected = new Set(ids)
+  return records.map(record => selected.has(record.id) ? { ...record, read: true } : record)
+}

lib/notifications/render.ts

@@ -0,0 +1,32 @@
+import type { NotificationEvent, RenderedNotification } from "./types"
+import { safeApplicationUrl } from "./templates"
+
+const metadata = {
+  "funding.completed": ["funding", false, "Funding received", "/dashboard/investor"],
+  "investment.confirmed": ["investment", false, "Investment confirmed", "/dashboard/investor/investments"],
+  "repayment.due": ["repayment", true, "Repayment due", "/dashboard/driver/repayments"],
+  "kyc.decision": ["kyc", true, "KYC status updated", "/dashboard/driver/kyc/status"],
+  "payout.processed": ["payout", true, "Payout processed", "/dashboard/investor"],
+  "arrears.notice": ["arrears", true, "Account action required", "/dashboard/driver"],
+  "contract.changed": ["contract", true, "Contract updated", "/dashboard/driver"],
+} as const
+
+export function renderNotification(event: NotificationEvent): RenderedNotification {
+  const parsed = (event.type === "kyc.decision" ? schemas.kyc : schemas.text).parse(event.payload)
+  const [category, mandatory, title, path] = metadata[event.type]
+  const p = parsed as Record<string, string>
+  const body = event.type === "funding.completed" ? `Your funding of ${p.amountLabel} was received.`
+    : event.type === "repayment.due" ? `A repayment is due ${p.dueDateLabel}.`
+    : event.type === "kyc.decision" ? `Your KYC status is now ${p.decision}. Sign in for details.`
+    : event.type === "arrears.notice" ? `Please review your account by ${p.actionByLabel}.`
+    : event.type === "contract.changed" ? "A contract associated with your account has changed. Sign in to review it."
+    : event.type === "payout.processed" ? "Your payout status has been updated. Sign in for details."
+    : "Your investment has been confirmed."
+  const actionUrl = safeApplicationUrl(path)
+  const escape = (v: string) => v.replace(/[&<>\"]/g, c => ({ "&": "&amp;", "<": "&lt;", ">": "&gt;", '\"': "&quot;" })[c]!)
+  return { templateKey: event.type, templateVersion: 1, category, mandatory, title, subject: `ChainMove: ${title}`, text: body, html: `<div><h2>${escape(title)}</h2><p>${escape(body)}</p><a href="${escape(actionUrl)}">Open ChainMove</a></div>`, actionUrl }
+}
+
+import { z } from "zod"
+const safeText = z.string().trim().min(1).max(80)
+const schemas = { text: z.record(z.string(), safeText), kyc: z.object({ decision: z.enum(["approved", "rejected", "review"]) }) }

lib/notifications/security.ts

@@ -0,0 +1,17 @@
+import { createHmac, timingSafeEqual } from "node:crypto"
+import type { NotificationCategory } from "./types"
+type Payload = { userId: string; category: NotificationCategory; channel: "email"; exp: number }
+export function createPreferenceToken(payload: Payload, secret = process.env.NOTIFICATION_TOKEN_SECRET || process.env.JWT_SECRET) {
+  if (!secret) throw new Error("Notification token secret is not configured")
+  const body = Buffer.from(JSON.stringify(payload)).toString("base64url")
+  return `${body}.${createHmac("sha256", secret).update(body).digest("base64url")}`
+}
+export function verifyPreferenceToken(token: string, secret = process.env.NOTIFICATION_TOKEN_SECRET || process.env.JWT_SECRET, now = Date.now()): Payload {
+  if (!secret) throw new Error("Notification token secret is not configured")
+  const [body, sig] = token.split("."); if (!body || !sig) throw new Error("Invalid preference token")
+  const expected = createHmac("sha256", secret).update(body).digest(), actual = Buffer.from(sig, "base64url")
+  if (actual.length !== expected.length || !timingSafeEqual(actual, expected)) throw new Error("Invalid preference token")
+  const payload = JSON.parse(Buffer.from(body, "base64url").toString()) as Payload
+  if (!payload.userId || payload.channel !== "email" || payload.exp <= Math.floor(now / 1000)) throw new Error("Expired preference token")
+  return payload
+}

lib/notifications/service.ts

@@ -0,0 +1,76 @@
+import { Resend } from "resend"
+import dbConnect from "@/lib/dbConnect"
+import Notification from "@/models/Notification"
+import NotificationDelivery from "@/models/NotificationDelivery"
+import NotificationPreference from "@/models/NotificationPreference"
+import User from "@/models/User"
+import { renderNotification } from "./render"
+import type { NotificationChannel, NotificationEvent, NotificationPreferences } from "./types"
+
+export function enabledChannels(rendered: { category: string; mandatory: boolean }, preferences?: NotificationPreferences): NotificationChannel[] {
+  const selected = preferences?.categories[rendered.category as keyof NotificationPreferences["categories"]]
+  return (["in_app", "email"] as const).filter(channel => rendered.mandatory || selected?.[channel] !== false)
+}
+
+export async function publishNotificationEvent(event: NotificationEvent, scheduledFor = new Date()) {
+  await dbConnect()
+  const [user, preference] = await Promise.all([User.findById(event.userId).select("email").lean<{ email?: string }>(), NotificationPreference.findOne({ userId: event.userId }).lean()])
+  if (!user) throw new Error("Notification recipient not found")
+  const rendered = renderNotification(event)
+  const channels = enabledChannels(rendered, preference as unknown as NotificationPreferences)
+  const results: unknown[] = []
+  for (const channel of channels) {
+    if (channel === "email" && !user.email) continue
+    const idempotencyKey = `${event.eventId}:${event.userId}:${channel}`
+    const delivery = await NotificationDelivery.findOneAndUpdate({ idempotencyKey }, { $setOnInsert: {
+      idempotencyKey, eventId: event.eventId, eventType: event.type, userId: event.userId, channel,
+      category: rendered.category, mandatory: rendered.mandatory, templateKey: rendered.templateKey,
+      templateVersion: rendered.templateVersion, to: channel === "email" ? user.email : undefined,
+      subject: channel === "email" ? rendered.subject : undefined, html: channel === "email" ? rendered.html : undefined,
+      status: channel === "email" ? "scheduled" : "created", scheduledFor,
+    } }, { upsert: true, new: true })
+    const inAppClaim = channel === "in_app"
+      ? await NotificationDelivery.findOneAndUpdate({ _id: delivery._id, notificationId: { $exists: false }, status: "created" }, { $set: { status: "processing", lockedAt: new Date() } }, { new: true })
+      : null
+    if (inAppClaim) {
+      const notification = await Notification.create({ userId: event.userId, title: rendered.title, message: rendered.text, type: rendered.category, priority: rendered.mandatory ? "high" : "medium", link: new URL(rendered.actionUrl).pathname })
+      await NotificationDelivery.updateOne({ _id: inAppClaim._id }, { $set: { notificationId: notification._id, status: "delivered", deliveredAt: new Date() } })
+    }
+    results.push(delivery)
+  }
+  return results
+}
+
+export interface EmailProvider { send(input: { to: string; subject: string; html: string; idempotencyKey: string }): Promise<{ id: string }> }
+export async function processEmailJobs(provider: EmailProvider = resendProvider(), now = new Date(), limit = 25) {
+  const summary = { delivered: 0, retried: 0, deadLettered: 0 }
+  await dbConnect()
+  for (let i = 0; i < limit; i++) {
+    const job = await NotificationDelivery.findOneAndUpdate({ channel: "email", status: "scheduled", scheduledFor: { $lte: now } }, { $set: { status: "processing", lockedAt: now } }, { sort: { scheduledFor: 1 }, new: true })
+    if (!job) break
+    try {
+      const response = await provider.send({ to: job.to, subject: job.subject, html: job.html, idempotencyKey: job.idempotencyKey })
+      await NotificationDelivery.updateOne({ _id: job._id }, { $set: { status: "delivered", providerId: response.id, deliveredAt: now }, $inc: { attemptCount: 1 }, $push: { attempts: { attemptedAt: now, status: "delivered", providerId: response.id } } })
+      summary.delivered++
+    } catch (error) {
+      const count = job.attemptCount + 1, dead = count >= job.maxAttempts
+      await NotificationDelivery.updateOne({ _id: job._id }, { $set: { status: dead ? "dead_letter" : "scheduled", scheduledFor: new Date(now.getTime() + Math.min(3600, 30 * 2 ** count) * 1000), ...(dead ? { failedAt: now } : {}) }, $inc: { attemptCount: 1 }, $push: { attempts: { attemptedAt: now, status: "failed", error: redactError(error) } } })
+      dead ? summary.deadLettered++ : summary.retried++
+    }
+  }
+  return summary
+}
+
+export function redactError(error: unknown) {
+  const value = error instanceof Error ? error.message : "Provider failure"
+  return value.replace(/[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}/g, "[REDACTED_EMAIL]").replace(/(secret|token|key|document)[=: ]+\S+/gi, "$1=[REDACTED]").slice(0, 500)
+}
+function resendProvider(): EmailProvider {
+  return { async send(input) {
+    if (process.env.ENABLE_MOCK_EMAILS === "true") return { id: `mock-${input.idempotencyKey}` }
+    if (!process.env.RESEND_API_KEY) throw new Error("Email provider is not configured")
+    const { data, error } = await new Resend(process.env.RESEND_API_KEY).emails.send({ from: "ChainMove <onboarding@chainmove.xyz>", to: input.to, subject: input.subject, html: input.html }, { idempotencyKey: input.idempotencyKey })
+    if (error || !data) throw new Error(error?.message || "Email provider failure")
+    return { id: data.id }
+  } }
+}