Merge pull request #314 from trails-cool/feat/notifications-cursor-pagination

Cursor-based pagination for /notifications

apps/journal/app/jobs/notifications-fanout.integration.test.ts

@@ -75,12 +75,12 @@ describe.skipIf(!runIntegration)("notifications-fanout integration", () => {
     const activityId = await makeActivity(owner, "public", "Public Walk");
     await fanout(activityId);
 
-    expect((await listForUser(a1)).length).toBe(1);
-    expect((await listForUser(a2)).length).toBe(1);
-    expect((await listForUser(p1)).length).toBe(0);
-    expect((await listForUser(p2)).length).toBe(0);
+    expect((await listForUser(a1)).rows.length).toBe(1);
+    expect((await listForUser(a2)).rows.length).toBe(1);
+    expect((await listForUser(p1)).rows.length).toBe(0);
+    expect((await listForUser(p2)).rows.length).toBe(0);
 
-    const a1Rows = await listForUser(a1);
+    const a1Rows = (await listForUser(a1)).rows;
     expect(a1Rows[0]?.type).toBe("activity_published");
     expect((a1Rows[0]?.payload as { activityName?: string })?.activityName).toBe("Public Walk");
   });
@@ -95,7 +95,7 @@ describe.skipIf(!runIntegration)("notifications-fanout integration", () => {
     await fanout(privateAct);
     await fanout(unlistedAct);
 
-    expect((await listForUser(f)).length).toBe(0);
+    expect((await listForUser(f)).rows.length).toBe(0);
   });
 
   it("is idempotent under retry — second fanout doesn't double-insert", async () => {
@@ -107,6 +107,6 @@ describe.skipIf(!runIntegration)("notifications-fanout integration", () => {
     await fanout(activityId);
     await fanout(activityId);
 
-    expect((await listForUser(f)).length).toBe(1);
+    expect((await listForUser(f)).rows.length).toBe(1);
   });
 });

apps/journal/app/lib/notifications.integration.test.ts

@@ -71,8 +71,9 @@ describe.skipIf(!runIntegration)("notifications.server integration", () => {
     });
     expect(second).toBe(false);
 
-    const rows = await listForUser(a);
+    const { rows, nextCursor } = await listForUser(a);
     expect(rows.length).toBe(1);
+    expect(nextCursor).toBeNull();
     expect(rows[0]?.payloadVersion).toBe(1);
     expect(rows[0]?.payload).toMatchObject({ activityId: subject });
   });
@@ -90,7 +91,7 @@ describe.skipIf(!runIntegration)("notifications.server integration", () => {
     });
     expect(await countUnread(a)).toBe(1);
 
-    const all = await listForUser(a);
+    const all = (await listForUser(a)).rows;
     const id = all[0]!.id;
 
     // Foreign user can't mark a's notification read.
@@ -197,13 +198,13 @@ describe.skipIf(!runIntegration)("notifications.server integration", () => {
     const bRow = (await getDb().select().from(users).where(eq(users.id, b)))[0]!;
 
     await followUser(a, bRow.username);
-    let bRows = await listForUser(b);
+    let bRows = (await listForUser(b)).rows;
     expect(bRows.length).toBe(1);
     expect(bRows[0]?.type).toBe("follow_received");
 
     // Re-follow is idempotent at the follow layer — and must NOT emit again.
     await followUser(a, bRow.username);
-    bRows = await listForUser(b);
+    bRows = (await listForUser(b)).rows;
     expect(bRows.length).toBe(1);
   });
 
@@ -213,7 +214,7 @@ describe.skipIf(!runIntegration)("notifications.server integration", () => {
     const bRow = (await getDb().select().from(users).where(eq(users.id, b)))[0]!;
 
     await followUser(a, bRow.username);
-    const bRows = await listForUser(b);
+    const bRows = (await listForUser(b)).rows;
     expect(bRows.length).toBe(1);
     expect(bRows[0]?.type).toBe("follow_request_received");
 
@@ -221,13 +222,64 @@ describe.skipIf(!runIntegration)("notifications.server integration", () => {
     expect(reqs.length).toBe(1);
 
     await approveFollowRequest(b, reqs[0]!.id);
-    const aRows = await listForUser(a);
+    const aRows = (await listForUser(a)).rows;
     expect(aRows.length).toBe(1);
     expect(aRows[0]?.type).toBe("follow_request_approved");
 
     // Idempotent re-approval must not double-emit.
     await approveFollowRequest(b, reqs[0]!.id);
-    const aRowsAfter = await listForUser(a);
+    const aRowsAfter = (await listForUser(a)).rows;
     expect(aRowsAfter.length).toBe(1);
   });
+
+  it("paginates with nextCursor; cursor is stable across pages and ends with null", async () => {
+    const a = await makeUser({ username: `n_pg_a_${Date.now()}` });
+    // Insert 7 notifications, each older than the previous, so we can
+    // page through them with limit: 3. Each row has a unique subject so
+    // the partial-unique constraint doesn't deduplicate them.
+    for (let i = 0; i < 7; i++) {
+      await createNotification({
+        type: "activity_published",
+        recipientUserId: a,
+        actorUserId: null,
+        subjectId: randomUUID(),
+        payload: {
+          activityId: randomUUID(),
+          activityName: `act-${i}`,
+          ownerUsername: "x",
+          ownerDisplayName: null,
+        },
+      });
+    }
+
+    const p1 = await listForUser(a, { limit: 3 });
+    expect(p1.rows.length).toBe(3);
+    expect(p1.nextCursor).not.toBeNull();
+    const p1Names = p1.rows.map((r) => (r.payload as { activityName?: string })?.activityName);
+
+    const p2 = await listForUser(a, { limit: 3, before: p1.nextCursor! });
+    expect(p2.rows.length).toBe(3);
+    expect(p2.nextCursor).not.toBeNull();
+    const p2Names = p2.rows.map((r) => (r.payload as { activityName?: string })?.activityName);
+    // Pages must not overlap.
+    expect(p1Names.some((n) => p2Names.includes(n))).toBe(false);
+
+    const p3 = await listForUser(a, { limit: 3, before: p2.nextCursor! });
+    expect(p3.rows.length).toBe(1);
+    // Only one row left → nextCursor is null (no more pages).
+    expect(p3.nextCursor).toBeNull();
+  });
+
+  it("treats malformed cursor as 'start from top' rather than erroring", async () => {
+    const a = await makeUser({ username: `n_pgbad_a_${Date.now()}` });
+    await createNotification({
+      type: "follow_received",
+      recipientUserId: a,
+      actorUserId: null,
+      subjectId: null,
+      payload: { followerUsername: "x", followerDisplayName: null },
+    });
+    const r = await listForUser(a, { before: "not-a-real-cursor" });
+    expect(r.rows.length).toBe(1);
+  });
 });

apps/journal/app/lib/notifications.server.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import { and, count, desc, eq, isNull, lt, sql } from "drizzle-orm";
+import { and, count, desc, eq, isNull, lt, or, sql } from "drizzle-orm";
 import { getDb } from "./db.ts";
 import { notifications } from "@trails-cool/db/schema/journal";
 import type { NotificationType } from "@trails-cool/db/schema/journal";
@@ -87,14 +87,86 @@ export interface NotificationRow {
   createdAt: Date;
 }
 
-const PAGE_SIZE = 50;
+const DEFAULT_PAGE_SIZE = 50;
+const MAX_PAGE_SIZE = 100;
 
+export interface ListOptions {
+  /**
+   * Opaque cursor from a previous response. When set, returns rows
+   * strictly older than the cursor's `(createdAt, id)` position, so
+   * pagination is stable even when two rows share `created_at`.
+   */
+  before?: string;
+  /** Soft cap; clamped to `[1, MAX_PAGE_SIZE]`. Defaults to 50. */
+  limit?: number;
+}
+
+export interface ListResult {
+  rows: NotificationRow[];
+  /**
+   * Opaque cursor for the next page, or `null` when the caller has
+   * reached the end. Pass back as `before` on the next request.
+   */
+  nextCursor: string | null;
+}
+
+interface CursorShape {
+  ts: string;
+  id: string;
+}
+
+function encodeCursor(c: CursorShape): string {
+  return Buffer.from(JSON.stringify(c), "utf8").toString("base64url");
+}
+
+function decodeCursor(s: string): CursorShape | null {
+  try {
+    const obj = JSON.parse(Buffer.from(s, "base64url").toString("utf8")) as {
+      ts?: unknown;
+      id?: unknown;
+    };
+    if (typeof obj.ts !== "string" || typeof obj.id !== "string") return null;
+    if (Number.isNaN(Date.parse(obj.ts))) return null;
+    return { ts: obj.ts, id: obj.id };
+  } catch {
+    return null;
+  }
+}
+
+/**
+ * Cursor-paginated list of `userId`'s notifications, newest first.
+ * Pass the previous response's `nextCursor` as `before` to fetch the
+ * next page; an unknown / malformed cursor is treated as "start from
+ * the top" rather than 400-ing, since the cursor is opaque to clients.
+ */
 export async function listForUser(
   userId: string,
-  opts: { page?: number } = {},
-): Promise<NotificationRow[]> {
+  opts: ListOptions = {},
+): Promise<ListResult> {
   const db = getDb();
-  const page = Math.max(1, opts.page ?? 1);
+  const limit = Math.min(MAX_PAGE_SIZE, Math.max(1, opts.limit ?? DEFAULT_PAGE_SIZE));
+  const cursor = opts.before ? decodeCursor(opts.before) : null;
+
+  const baseWhere = eq(notifications.recipientUserId, userId);
+  const where = cursor
+    ? and(
+        baseWhere,
+        // (created_at, id) < (cursor.ts, cursor.id)
+        // Keyset comparison broken into the two-row form so Postgres
+        // uses the (recipient, created_at desc) index for the leading
+        // column.
+        or(
+          lt(notifications.createdAt, new Date(cursor.ts)),
+          and(
+            eq(notifications.createdAt, new Date(cursor.ts)),
+            lt(notifications.id, cursor.id),
+          ),
+        ),
+      )
+    : baseWhere;
+
+  // Fetch limit + 1 so we can decide whether a `nextCursor` exists
+  // without a separate count query.
   const rows = await db
     .select({
       id: notifications.id,
@@ -107,11 +179,18 @@ export async function listForUser(
       createdAt: notifications.createdAt,
     })
     .from(notifications)
-    .where(eq(notifications.recipientUserId, userId))
-    .orderBy(desc(notifications.createdAt))
-    .limit(PAGE_SIZE)
-    .offset((page - 1) * PAGE_SIZE);
-  return rows;
+    .where(where)
+    .orderBy(desc(notifications.createdAt), desc(notifications.id))
+    .limit(limit + 1);
+
+  const hasMore = rows.length > limit;
+  const trimmed = hasMore ? rows.slice(0, limit) : rows;
+  const last = trimmed[trimmed.length - 1];
+  const nextCursor = hasMore && last
+    ? encodeCursor({ ts: last.createdAt.toISOString(), id: last.id })
+    : null;
+
+  return { rows: trimmed, nextCursor };
 }
 
 export async function countUnread(userId: string): Promise<number> {

apps/journal/app/routes/notifications.tsx

@@ -14,7 +14,9 @@ export async function loader({ request }: Route.LoaderArgs) {
   const user = await getSessionUser(request);
   if (!user) throw redirect("/auth/login");
 
-  const rows = await listForUser(user.id, { page: 1 });
+  const url = new URL(request.url);
+  const before = url.searchParams.get("before") ?? undefined;
+  const { rows, nextCursor } = await listForUser(user.id, { before });
 
   // Renderer guard: drop activity_published rows whose subject is gone
   // or no longer public (visibility flipped from public → private/unlisted).
@@ -56,6 +58,7 @@ export async function loader({ request }: Route.LoaderArgs) {
         payload,
       };
     }),
+    nextCursor,
   });
 }
 
@@ -145,7 +148,7 @@ function NotificationItem({ row }: { row: Row }) {
 }
 
 export default function Notifications({ loaderData }: Route.ComponentProps) {
-  const { notifications } = loaderData;
+  const { notifications, nextCursor } = loaderData;
   const { t } = useTranslation("journal");
   const markAll = useFetcher();
 
@@ -171,11 +174,23 @@ export default function Notifications({ loaderData }: Route.ComponentProps) {
       {notifications.length === 0 ? (
         <p className="mt-12 text-center text-gray-500">{t("notifications.empty")}</p>
       ) : (
-        <ul className="mt-6 rounded-lg border border-gray-200 bg-white">
-          {notifications.map((n) => (
-            <NotificationItem key={n.id} row={n} />
-          ))}
-        </ul>
+        <>
+          <ul className="mt-6 rounded-lg border border-gray-200 bg-white">
+            {notifications.map((n) => (
+              <NotificationItem key={n.id} row={n} />
+            ))}
+          </ul>
+          {nextCursor && (
+            <div className="mt-4 text-center">
+              <a
+                href={`/notifications?before=${encodeURIComponent(nextCursor)}`}
+                className="text-sm font-medium text-blue-600 hover:underline"
+              >
+                {t("notifications.loadOlder")}
+              </a>
+            </div>
+          )}
+        </>
       )}
     </div>
   );

openspec/changes/notifications/design.md

@@ -235,6 +235,16 @@ EventSource auto-reconnects on transient disconnects; we just open and close.
 
 **Single-process today, Redis-pub/sub when we go multi-process.** The `emitTo(userId, event, data)` interface is the swap point — the in-process `Map` becomes a Redis subscriber that listens to per-user channels, and `emitTo` becomes a publish. The hooks don't change.
 
+### Decision: Cursor-based pagination for `/notifications`
+
+`listForUser` accepts an opaque `before` cursor (`base64url(JSON{ts, id})`) and orders rows by `(created_at DESC, id DESC)`. `id` is a tiebreaker for rows with identical `created_at`, so two callers paging through the list see neither duplicates nor gaps even if a fan-out job inserted a hundred rows at the same instant. Default page size is 50, hard-capped at 100.
+
+**Why cursor, not page-offset:** notifications grow at the head, not the tail — every new event prepends a row. A page-offset query (`OFFSET 50`) would shift under the user as new rows arrive, causing duplicates on page 2. Cursor pagination pins the boundary to a stable position in the data. It also matches the partial unread index (`(recipient_user_id, created_at DESC)`) without needing `OFFSET`'s implicit table scan.
+
+**Why opaque (base64-encoded JSON) over raw `(ts, id)` query params:** clients should not have to know or validate the cursor format. We get to evolve the encoding (e.g. add a `version` field, or include `read_at` for unread-only feeds) without breaking deep links. A malformed cursor is treated as "start from the top" rather than 400, so an old saved URL still renders something useful.
+
+**No total count, deliberately:** the page shows "Load older" while a `nextCursor` exists; once the cursor is `null`, the user has reached the bottom. Computing a total `COUNT(*)` per page load would defeat the cheap-pagination win and is only useful for a "page 5 of 12" UI that we're not building.
+
 ## Risks / Trade-offs
 
 - **Activity fan-out spam**: a user with thousands of followers posting a public activity creates thousands of rows. → At our scale this is invisible; the cost ceiling is documented above. If we ever need to soften, batch the inserts or move to read-time.

openspec/changes/notifications/specs/notifications/spec.md

@@ -70,10 +70,10 @@ The Journal SHALL expose a single server-side helper `linkFor(notification)` ret
 - **THEN** the existing notification rows remain in the database but are filtered out of the recipient's `/notifications` listing (the renderer skips rows whose subject the recipient can no longer see)
 
 ### Requirement: Notifications page and unread count
-The Journal SHALL expose `/notifications` to signed-in users only. The page SHALL list the user's notifications reverse-chronological, with each row showing the actor's display name + handle, a type-specific summary line, the timestamp, and a clickable link to the relevant subject. The navbar SHALL surface an unread count badge linking to `/notifications`; the count is `notifications.read_at IS NULL` for the current user. Logged-out visitors requesting `/notifications` SHALL be redirected to `/auth/login`.
+The Journal SHALL expose `/notifications` to signed-in users only. The page SHALL list the user's notifications reverse-chronological, with each row showing the actor's display name + handle, a type-specific summary line, the timestamp, and a clickable link to the relevant subject. The page SHALL paginate via opaque cursor (a `before` query parameter); when more rows exist past the current page, the page SHALL surface a "Load older" affordance that fetches the next page using the cursor. Page size defaults to 50 and is capped at 100. The navbar SHALL surface an unread count badge linking to `/notifications`; the count is `notifications.read_at IS NULL` for the current user. Logged-out visitors requesting `/notifications` SHALL be redirected to `/auth/login`.
 
 #### Scenario: Logged-in user with notifications
-- **WHEN** a signed-in user with N notifications loads `/notifications`
+- **WHEN** a signed-in user with N notifications (N ≤ page size) loads `/notifications`
 - **THEN** the page lists all N rows reverse-chronological by `created_at`, with unread rows visually distinct from read rows
 
 #### Scenario: Logged-in user with no notifications
@@ -89,6 +89,19 @@ The Journal SHALL expose `/notifications` to signed-in users only. The page SHAL
 - **THEN** the "Notifications" navbar entry renders with a count badge showing K
 - **AND** when K = 0, the entry renders without a badge
 
+#### Scenario: Paginates older notifications via cursor
+- **WHEN** a signed-in user has more notifications than fit in one page and clicks "Load older"
+- **THEN** the next request includes the previous response's cursor as `?before=<cursor>` and the page renders the next batch, strictly older than the cursor's `(created_at, id)` position
+- **AND** when the final page has been reached the response no longer surfaces "Load older"
+
+#### Scenario: Cursor is stable across rows with identical timestamps
+- **WHEN** two notification rows share the exact same `created_at`
+- **THEN** pagination orders them deterministically by `id` (descending) as a tiebreaker, so neither page omits nor duplicates them
+
+#### Scenario: Malformed cursor is ignored, not surfaced as an error
+- **WHEN** a request reaches `/notifications` with a `before` value that doesn't decode to a valid cursor
+- **THEN** the page renders from the top (newest rows) instead of returning a 400, since the cursor is opaque and clients should not need to validate it
+
 ### Requirement: Mark-as-read controls
 A signed-in user SHALL be able to mark an individual notification as read via `POST /api/notifications/:id/read`, and to mark all of their unread notifications as read via `POST /api/notifications/read-all`. Both endpoints are owner-bound: only the recipient may mark their own notifications.
 

openspec/changes/notifications/tasks.md

@@ -7,7 +7,7 @@
 
 ## 2. Server library
 
-- [x] 2.1 Create `apps/journal/app/lib/notifications.server.ts` with: `createNotification`, `listForUser(userId, { page })`, `countUnread(userId)`, `markRead(ownerId, id)`, `markAllRead(ownerId)`, `purgeReadOlderThan(days)`. `createNotification` accepts a typed payload + version per type — TS unions enforce per-type shapes at the call site
+- [x] 2.1 Create `apps/journal/app/lib/notifications.server.ts` with: `createNotification`, `listForUser(userId, { before, limit })` (cursor-paginated, returns `{ rows, nextCursor }`), `countUnread(userId)`, `markRead(ownerId, id)`, `markAllRead(ownerId)`, `purgeReadOlderThan(days)`. `createNotification` accepts a typed payload + version per type — TS unions enforce per-type shapes at the call site
 - [x] 2.2 Define per-type payload TypeScript types (`FollowPayloadV1`, `ApprovalPayloadV1`, `ActivityPayloadV1`) in `apps/journal/app/lib/notifications/payload.ts`. Each generation hook writes `payload_version = 1` plus the matching shape
 - [x] 2.3 Create `apps/journal/app/lib/notifications/link-for.ts` with `linkFor(notification): LinkBundle` returning `{ web, mobile?, email? }` — handles all four v1 types; consults `subject_id` first, falls back to `payload` fields
 - [x] 2.4 Make `markRead` and `markAllRead` owner-bound (predicate includes `recipient_user_id = ownerId`)
@@ -47,7 +47,7 @@
 
 ## 5. Routes + UI
 
-- [x] 5.1 New `/notifications` route (signed-in only; redirect anonymous to `/auth/login`). Loader fetches `listForUser(currentUser.id, { page: 1 })` and `countUnread(currentUser.id)`
+- [x] 5.1 New `/notifications` route (signed-in only; redirect anonymous to `/auth/login`). Loader fetches `listForUser(currentUser.id, { before: searchParams.get("before") })` and surfaces a "Load older" link when `nextCursor` is non-null
 - [x] 5.2 Render: list of cards. Each card shows the actor's display name + handle (live where available, payload-snapshot fallback if the actor row is gone), a type-specific summary line (i18n keyed), the timestamp, and an action whose href comes from `linkFor(notification).web`. Unread cards are visually distinct (e.g. background tint + unread dot)
 - [x] 5.3 Implement click-through: clicking a card POSTs to `/api/notifications/:id/read` and then navigates to `linkFor(notification).web`. Server-side redirect after mark-read is the simplest path
 - [x] 5.4 "Mark all read" button at the top of the page; clicking POSTs to `/api/notifications/read-all` and the page reloads with empty unread state

packages/i18n/src/locales/de.ts

@@ -245,6 +245,7 @@ export default {
       title: "Benachrichtigungen",
       empty: "Noch keine Benachrichtigungen.",
       markAllRead: "Alle als gelesen markieren",
+      loadOlder: "Ältere laden",
       someone: "Jemand",
       summary: {
         followRequestReceived: "{{name}} möchte dir folgen",

packages/i18n/src/locales/en.ts

@@ -245,6 +245,7 @@ export default {
       title: "Notifications",
       empty: "No notifications yet.",
       markAllRead: "Mark all read",
+      loadOlder: "Load older",
       someone: "Someone",
       summary: {
         followRequestReceived: "{{name}} requested to follow you",