Realtime

import { memoryRealtimeAdapter } from "@notifykitjs/core"

const notify = createNotifyKit({
  // ...
  realtime: memoryRealtimeAdapter(),
})

npm install @notifykitjs/realtime-pg

import { pgRealtimeAdapter } from "@notifykitjs/realtime-pg"
import { Pool } from "pg"

const pool = new Pool({ connectionString: process.env.DATABASE_URL })

// The adapter needs a dedicated connection for LISTEN
const listenClient = await pool.connect()

const realtime = pgRealtimeAdapter({
  connection: {
    listen: (channel, handler) => {
      listenClient.on("notification", (msg) => {
        if (msg.channel === channel) handler(msg.payload!)
      })
      return listenClient.query(`LISTEN ${channel}`)
    },
    unlisten: (channel) => listenClient.query(`UNLISTEN ${channel}`),
    notify: (channel, payload) =>
      pool.query("SELECT pg_notify($1, $2)", [channel, payload]),
  },
  heartbeatMs: 60_000, // detect dead connections
  onError: (err) => console.error("realtime error:", err),
})

await realtime.start()

const notify = createNotifyKit({
  // ...
  realtime,
})

npm install @notifykitjs/realtime-ws

import { webSocketRealtimeAdapter } from "@notifykitjs/realtime-ws"

const realtime = webSocketRealtimeAdapter({
  authenticate: async (request) => {
    const token = new URL(request.url).searchParams.get("token")
    const session = await verifyToken(token)
    if (!session) return null
    return {
      recipientId: session.userId,
      tenantId: session.orgId,
    }
  },
  allowedOrigins: ["https://app.example.com"],
  heartbeatMs: 30_000,
  maxConnections: 10_000,
})

server.on("upgrade", async (request, ws) => {
  const identity = await realtime.handleUpgrade(request, ws)
  if (!identity) {
    ws.close(4001, "Unauthorized")
    return
  }

  ws.on("message", (data) => realtime.handleMessage(ws, data.toString()))
  ws.on("close", () => realtime.handleClose(ws))
})

function connectSSE(baseUrl, { onEvent, onStatusChange }) {
  let retryDelay = 1000
  let es = null

  function connect() {
    onStatusChange("connecting")
    es = new EventSource(`${baseUrl}/inbox/stream`, { withCredentials: true })

    es.onopen = () => {
      retryDelay = 1000 // reset backoff
      onStatusChange("connected")
    }

    es.onmessage = (e) => {
      onEvent(JSON.parse(e.data))
    }

    es.onerror = () => {
      es.close()
      onStatusChange("disconnected")
      // Exponential backoff: 1s, 2s, 4s, 8s... max 30s
      setTimeout(connect, retryDelay)
      retryDelay = Math.min(retryDelay * 2, 30_000)
    }
  }

  connect()
  return () => { es?.close() } // cleanup function
}

const client = createNotifyKitClient({
  realtime: true,
  onRealtimeError: (error) => {
    metrics.increment("notifykit.realtime.errors")
    reportError(error)
  },
})

"use client"
import { useInbox, useUnreadCount } from "@notifykitjs/react"

function NotificationBell() {
  // Poll every 10 seconds instead of SSE
  const { items, markAsRead } = useInbox({ pollInterval: 10_000 })
  const { unreadCount } = useUnreadCount({ pollInterval: 10_000 })

  return (
    <div>
      <button>🔔 {unreadCount > 0 && <span>{unreadCount}</span>}</button>
      <ul>
        {items.map(item => (
          <li key={item.id} onClick={() => markAsRead(item.id)}>
            {item.title}
          </li>
        ))}
      </ul>
    </div>
  )
}

"use client"
import { useInbox, useUnreadCount } from "@notifykitjs/react"
import { useState, useEffect, useCallback } from "react"

function useAdaptivePolling() {
  const [interval, setInterval] = useState(30_000) // start slow (30s)

  const onNewItems = useCallback(() => {
    setInterval(5_000) // speed up on activity (5s)
  }, [])

  // Slow down after 2 minutes of no new items
  useEffect(() => {
    const slowDown = setTimeout(() => setInterval(30_000), 120_000)
    return () => clearTimeout(slowDown)
  }, [interval])

  return { interval, onNewItems }
}

function NotificationBell() {
  const { interval, onNewItems } = useAdaptivePolling()
  const { items } = useInbox({
    pollInterval: interval,
    onNewItems, // called when new items appear
  })
  const { unreadCount } = useUnreadCount({ pollInterval: interval })

  // ...render
}

export const realtimeConfig = {
  usePolling: process.env.NEXT_PUBLIC_USE_POLLING === "true",
  pollInterval: 10_000,
}

import { realtimeConfig } from "@/lib/realtime-config"

function NotificationBell() {
  const inboxOptions = realtimeConfig.usePolling
    ? { pollInterval: realtimeConfig.pollInterval }
    : {} // use SSE (default)

  const { items } = useInbox(inboxOptions)
  // ...
}

# Set in your Vercel project settings (or hosting platform)
NEXT_PUBLIC_USE_POLLING=true

# Terminal 1: connect to SSE (grab a session cookie from your browser)
curl -N -H "Cookie: session=..." http://localhost:3000/api/notifykit/inbox/stream

# Expected output (within 30s):
# : heartbeat
#
# After sending a notification:
# event: inbox.created
# data: {"type":"inbox.created","item":{"id":"inb_...","title":"..."}}

import { describe, it, expect, vi } from "vitest"
import { createNotifyKit, memoryAdapter, memoryRealtimeAdapter, fakeEmailProvider } from "@notifykitjs/core"
import { commentMentioned } from "./notifications"

describe("realtime events", () => {
  function setup() {
    const realtime = memoryRealtimeAdapter()
    const notify = createNotifyKit({
      notifications: [commentMentioned] as const,
      database: memoryAdapter(),
      providers: { email: fakeEmailProvider() },
      realtime,
    })
    return { notify, realtime }
  }

  it("fires inbox.created when a notification is sent", async () => {
    const { notify, realtime } = setup()
    const received = vi.fn()

    await notify.upsertRecipient({ id: "alice", email: "a@test.com" })

    // Subscribe BEFORE sending — mimics an open SSE connection
    realtime.subscribe("alice", {}, (event) => received(event))

    await notify.send({
      recipientId: "alice",
      notificationId: "comment_mentioned",
      payload: { actorName: "Rey", postUrl: "/p/1" },
    })

    expect(received).toHaveBeenCalledTimes(1)
    expect(received).toHaveBeenCalledWith(
      expect.objectContaining({
        type: "inbox.created",
        payload: expect.objectContaining({ title: expect.any(String) }),
      })
    )
  })

  it("scopes events by tenant — no cross-org leaks", async () => {
    const { notify, realtime } = setup()
    const orgAEvents = vi.fn()
    const orgBEvents = vi.fn()

    await notify.upsertRecipient({ id: "alice", tenantId: "org_a", email: "a@test.com" })
    await notify.upsertRecipient({ id: "alice", tenantId: "org_b", email: "a@test.com" })

    realtime.subscribe("alice", { tenantId: "org_a" }, orgAEvents)
    realtime.subscribe("alice", { tenantId: "org_b" }, orgBEvents)

    await notify.send({
      recipientId: "alice",
      tenantId: "org_a",
      notificationId: "comment_mentioned",
      payload: { actorName: "Rey", postUrl: "/p/1" },
    })

    expect(orgAEvents).toHaveBeenCalledTimes(1)
    expect(orgBEvents).toHaveBeenCalledTimes(0) // ✓ isolated
  })

  it("fires inbox.updated on mark-read", async () => {
    const { notify, realtime } = setup()
    const received = vi.fn()

    await notify.upsertRecipient({ id: "alice", email: "a@test.com" })
    const result = await notify.send({
      recipientId: "alice",
      notificationId: "comment_mentioned",
      payload: { actorName: "Rey", postUrl: "/p/1" },
    })

    realtime.subscribe("alice", {}, received)
    await notify.inbox.markReadForRecipient(result.inboxItems[0].id, "alice")

    expect(received).toHaveBeenCalledWith(
      expect.objectContaining({
        type: "inbox.updated",
        payload: expect.objectContaining({ readAt: expect.any(Date) }),
      })
    )
  })
})

it("gap-fill: items created while disconnected appear on re-fetch", async () => {
  const { notify } = setup()
  await notify.upsertRecipient({ id: "alice", email: "a@test.com" })

  // Simulate: client was connected, then disconnected
  // (no subscriber active during this send)
  await notify.send({
    recipientId: "alice",
    notificationId: "comment_mentioned",
    payload: { actorName: "Sam", postUrl: "/p/2" },
  })

  // Client reconnects and re-fetches inbox (gap-fill strategy)
  const items = await notify.inbox.list("alice")
  expect(items).toHaveLength(1)
  expect(items[0].title).toContain("Sam")
})