Sending notifications

await notify.upsertRecipient({
  id: user.id,
  email: user.email,
  name: user.name,
})

const result = await notify.send({
  recipientId: user.id,
  notificationId: "comment_mentioned",
  payload: {
    actorName: "Rey",
    postTitle: "Launch Plan",
    postUrl: "/posts/42",
  },
})

// ❌ Unknown notification id
await notify.send({
  recipientId: "u_1",
  notificationId: "wrong_id",  // TS error: not assignable
  payload: {},
})

// ❌ Wrong payload shape
await notify.send({
  recipientId: "u_1",
  notificationId: "comment_mentioned",
  payload: { actorName: 42 },  // TS error: number is not string
})

import { setTimeoutQueue } from "@notifykitjs/core"

const notify = createNotifyKit({
  // ...
  queue: setTimeoutQueue(),
  retry: { maxAttempts: 5, delayMs: (n) => 500 * 2 ** (n - 1) },
})

// Wait for in-flight jobs before shutdown:
await notify.drain()

createNotifyKit({
  // ...
  on: {
    "notification.created": ({ notification }) =>
      metrics.inc("notifications.created"),
    "delivery.sent": ({ delivery }) =>
      metrics.inc("delivery.sent", { channel: delivery.channel }),
    "delivery.failed": ({ delivery, error }) =>
      sentry.captureException(error),
  },
})

// Server action — persist intent with the business mutation
"use server"
import { notify } from "@/lib/notifykit"

export async function addComment(postId: string, body: string) {
  return db.transaction(async tx => {
    const comment = await tx.comments.create({ postId, body, authorId: session.userId })

    for (const userId of extractMentions(body)) {
      await tx.notificationOutbox.create({
        notificationId: "comment_mentioned",
        recipientId: userId,
        payload: {
          actorName: session.userName,
          postTitle: comment.postTitle,
          postUrl: `/posts/${postId}`,
        },
        idempotencyKey: `mention:${comment.id}:${userId}`,
      })
    }

    return comment
  })
}

// A durable worker claims the outbox row, then awaits notify.send({
//   ...row,
//   payload: row.payload,
// }) before marking it complete.

// Notify all team members when a project ships
const members = await db.teamMembers.findMany({ where: { teamId } })

const results = await Promise.allSettled(
  members
    .filter(m => m.userId !== actor.id) // don't notify the person who did it
    .map(m =>
      notify.send({
        recipientId: m.userId,
        notificationId: "project_shipped",
        payload: { projectName, actorName: actor.name },
        idempotencyKey: `shipped:${projectId}:${m.userId}`,
      })
    )
)

const failed = results.filter(r => r.status === "rejected")
if (failed.length) logger.warn(`${failed.length} sends failed`, { projectId })

// Robust broadcast with error classification
async function broadcastWithRecovery(recipientIds: string[], notification: SendInput) {
  const results = await Promise.allSettled(
    recipientIds.map(id =>
      notify.send({ ...notification, recipientId: id, idempotencyKey: `${notification.idempotencyKey}:${id}` })
    )
  )

  const succeeded = results.filter(r => r.status === "fulfilled")
  const failed = results
    .map((r, i) => ({ result: r, recipientId: recipientIds[i] }))
    .filter(({ result }) => result.status === "rejected")

  // Classify failures
  const retryable = failed.filter(({ result }) =>
    result.status === "rejected" && isTransient(result.reason)
  )
  const deadLettered = failed.filter(({ result }) =>
    result.status === "rejected" && !isTransient(result.reason)
  )

  // Log and alert
  if (failed.length > 0) {
    logger.warn("Broadcast partial failure", {
      total: recipientIds.length,
      succeeded: succeeded.length,
      retryable: retryable.length,
      deadLettered: deadLettered.length,
    })
  }

  // Alert on high failure rate
  const failureRate = failed.length / recipientIds.length
  if (failureRate > 0.2) {
    await alertOncall("Broadcast failure rate >20%", { failureRate, notification })
  }

  // Retry transient failures (once, with backoff)
  if (retryable.length > 0) {
    await delay(2000)
    await Promise.allSettled(
      retryable.map(({ recipientId }) =>
        notify.send({ ...notification, recipientId, idempotencyKey: `${notification.idempotencyKey}:${recipientId}` })
      )
    )
  }

  return { succeeded: succeeded.length, failed: failed.length, retried: retryable.length }
}

function isTransient(error: unknown): boolean {
  const message = error instanceof Error ? error.message : ""
  return message.includes("timeout") || message.includes("ECONNRESET") || message.includes("503")
}

// ❌ Untracked fire-and-forget can disappear on crash or deploy
"use server"
export async function addComment(postId: string, body: string) {
  const comment = await db.comments.create({ postId, body })
  void notify.send({ recipientId: mentioned, ... })
  return comment
}

// ✅ Persist notification intent with the business operation
"use server"
export async function addComment(postId: string, body: string) {
  return db.transaction(async tx => {
    const comment = await tx.comments.create({ postId, body })
    await tx.notificationOutbox.create({
      type: "comment_mentioned",
      recipientId: mentioned,
      payload: { postId, commentId: comment.id },
    })
    return comment
  })
}

// ❌ Generic dedup key — silences ALL comment notifications globally
await notify.send({
  recipientId: userId,
  notificationId: "comment_mentioned",
  payload,
  dedupeKey: "comment",  // Only one comment notification per 5 min, ever
})

// ✅ Scoped dedup key — collapses only rapid duplicates for the same context
await notify.send({
  recipientId: userId,
  notificationId: "comment_mentioned",
  payload,
  dedupeKey: `comment:${postId}:${actorId}:${recipientId}`,
})

// ❌ No idempotency on a webhook handler — retries cause duplicates
export async function POST(req: Request) {
  const event = await req.json()
  await notify.send({
    recipientId: event.userId,
    notificationId: "payment_received",
    payload: { amount: event.amount },
  })
  return Response.json({ ok: true })
}

// ✅ Idempotency key derived from the event — retries are safe no-ops
export async function POST(req: Request) {
  const event = await req.json()
  await notify.send({
    recipientId: event.userId,
    notificationId: "payment_received",
    payload: { amount: event.amount },
    idempotencyKey: `stripe:${event.id}`,
  })
  return Response.json({ ok: true })
}

// lib/notifications/on-comment-posted.ts
import { notify } from "@/lib/notifykit"

export async function onCommentPosted(comment: {
  id: string
  postId: string
  authorId: string
  body: string
  postAuthorId: string
}) {
  const mentions = extractMentions(comment.body)
  const watchers = await db.watchers.findMany({ where: { postId: comment.postId } })
  const actor = await db.users.findFirst({ where: { id: comment.authorId } })

  // 1. Notify mentioned users (highest priority)
  const mentionSends = mentions
    .filter(userId => userId !== comment.authorId) // don't notify yourself
    .map(userId =>
      notify.send({
        recipientId: userId,
        notificationId: "comment_mentioned",
        payload: { actorName: actor.name, postUrl: `/posts/${comment.postId}` },
        dedupeKey: `mention:${comment.postId}:${comment.authorId}:${userId}`,
        dedupeWindowMs: 5 * 60_000,
      })
    )

  // 2. Notify post author (unless they wrote the comment)
  const authorSend = comment.authorId !== comment.postAuthorId
    ? notify.send({
        recipientId: comment.postAuthorId,
        notificationId: "comment_on_your_post",
        payload: { actorName: actor.name, postUrl: `/posts/${comment.postId}` },
        dedupeKey: `reply:${comment.postId}:${comment.authorId}`,
        dedupeWindowMs: 5 * 60_000,
      })
    : null

  // 3. Notify watchers (lowest priority — often digested)
  const alreadyNotified = new Set([...mentions, comment.postAuthorId, comment.authorId])
  const watcherSends = watchers
    .filter(w => !alreadyNotified.has(w.userId))
    .map(w =>
      notify.send({
        recipientId: w.userId,
        notificationId: "comment_on_watched",
        payload: { actorName: actor.name, postUrl: `/posts/${comment.postId}` },
      })
    )

  await Promise.allSettled([...mentionSends, authorSend, ...watcherSends].filter(Boolean))
}

// Server action — fire after the mutation
"use server"
import { onCommentPosted } from "@/lib/notifications/on-comment-posted"

export async function addComment(postId: string, body: string) {
  const comment = await db.comments.create({ ... })

  // Fire-and-forget — don't block the user response
  void onCommentPosted(comment)

  return comment
}

// Or from a background job (webhook, queue worker):
export async function handleCommentWebhook(payload: CommentEvent) {
  await onCommentPosted(payload.comment) // safe to await — no user waiting
}

// test/helpers/notifykit.ts
import { createNotifyKit, memoryAdapter, fakeEmailProvider, channel, notification } from "@notifykitjs/core"

export const commentMentioned = notification({
  id: "comment_mentioned",
  payload: { actorName: "string", postTitle: "string", postUrl: "string" },
  channels: [
    channel.inbox()({ title: "{{actorName}} mentioned you", body: "In {{postTitle}}", actionUrl: "{{postUrl}}" }),
    channel.email()({ subject: "{{actorName}} mentioned you", body: "Open {{postUrl}}" }),
  ],
})

export function createTestNotify() {
  return createNotifyKit({
    notifications: [commentMentioned] as const,
    database: memoryAdapter(),
    providers: { email: fakeEmailProvider() },
  })
}

import { describe, it, expect, beforeEach } from "vitest"
import { createTestNotify } from "./helpers/notifykit"

describe("comment mention sends", () => {
  let notify: ReturnType<typeof createTestNotify>

  beforeEach(async () => {
    notify = createTestNotify()
    await notify.upsertRecipient({ id: "alice", email: "alice@test.com" })
  })

  it("delivers to inbox and email", async () => {
    const result = await notify.send({
      recipientId: "alice",
      notificationId: "comment_mentioned",
      payload: { actorName: "Rey", postTitle: "Launch", postUrl: "/p/1" },
    })

    expect(result.inboxItems).toHaveLength(1)
    expect(result.inboxItems[0].title).toBe("Rey mentioned you")
    expect(result.deliveries).toHaveLength(1)
    expect(result.deliveries[0].channel).toBe("email")
    expect(result.deliveries[0].status).toBe("sent")
    expect(result.skipped).toHaveLength(0)
  })

  it("deduplicates within window", async () => {
    const opts = {
      recipientId: "alice" as const,
      notificationId: "comment_mentioned" as const,
      payload: { actorName: "Rey", postTitle: "Launch", postUrl: "/p/1" },
      dedupeKey: "mention:p1:rey",
      dedupeWindowMs: 60_000,
    }

    const first = await notify.send(opts)
    const second = await notify.send(opts)

    expect(first.inboxItems).toHaveLength(1)
    expect(second.inboxItems).toHaveLength(0) // deduped
  })

  it("idempotency returns original result", async () => {
    const opts = {
      recipientId: "alice" as const,
      notificationId: "comment_mentioned" as const,
      payload: { actorName: "Rey", postTitle: "Launch", postUrl: "/p/1" },
      idempotencyKey: "job:abc",
    }

    const first = await notify.send(opts)
    const replay = await notify.send(opts)

    expect(replay.idempotent).toBe(true)
    expect(replay.notification?.id).toBe(first.notification?.id)
  })
})

// lib/notifications/on-comment-posted.test.ts
import { describe, it, expect, vi } from "vitest"
import { onCommentPosted } from "./on-comment-posted"

// Mock the notify instance
const mockSend = vi.fn().mockResolvedValue({ inboxItems: [], deliveries: [] })
vi.mock("@/lib/notifykit", () => ({
  notify: { send: (...args) => mockSend(...args) },
}))

describe("onCommentPosted", () => {
  it("notifies mentioned users but not the author", async () => {
    await onCommentPosted({
      id: "c1",
      postId: "p1",
      authorId: "rey",
      mentions: ["alice", "bob", "rey"], // rey should be filtered
      postTitle: "Launch Plan",
    })

    expect(mockSend).toHaveBeenCalledTimes(2) // alice + bob, not rey
    expect(mockSend).not.toHaveBeenCalledWith(
      expect.objectContaining({ recipientId: "rey" })
    )
  })

  it("uses per-mention dedup keys", async () => {
    await onCommentPosted({
      id: "c1",
      postId: "p1",
      authorId: "rey",
      mentions: ["alice"],
      postTitle: "Launch",
    })

    expect(mockSend).toHaveBeenCalledWith(
      expect.objectContaining({
        dedupeKey: "mention:p1:rey:alice",
      })
    )
  })
})