From ebe467ee92a3ede0535c052cf84168a6b01f83d4 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sat, 12 Sep 2026 00:16:06 +0200 Subject: [PATCH] feat(notification): recover notification since last known event on client resume --- .../src/lib/api/endpoints/notifications.ts | 13 +- .../composables/useNotifications.svelte.ts | 143 +++++++++++++++--- .../lib/composables/useNotifications.test.ts | 58 +++++++ .../repositories/notification_repository.rs | 29 ++-- .../pg/notification_pg_repository.rs | 127 ++++------------ .../api/handlers/notifications_handler.rs | 16 +- tests/api/rt_bus_check.sh | 91 ++++++++++- 7 files changed, 345 insertions(+), 132 deletions(-) create mode 100644 frontend/src/lib/composables/useNotifications.test.ts diff --git a/frontend/src/lib/api/endpoints/notifications.ts b/frontend/src/lib/api/endpoints/notifications.ts index 84a442db..57a2edea 100644 --- a/frontend/src/lib/api/endpoints/notifications.ts +++ b/frontend/src/lib/api/endpoints/notifications.ts @@ -16,15 +16,26 @@ import type { UnreadCountResponse } from '$lib/api/types'; -/** List newest-first. Optional `unread` filter, `before` cursor, `limit` cap. */ +/** + * List newest-first. All filters are optional and additive: + * - `unread` — only rows with `read_at IS NULL` + * - `before` — older-than cursor for "load older page" pagination + * - `after` — newer-than cursor for delta catch-up on WS reconnect + * or tab reactivation (dedup handled at the store layer + * via `mergeById`, since the WS push and the delta fetch + * can race on the same row) + * - `limit` — server-side clamp at 500 rows + */ export async function listNotifications(opts?: { unread?: boolean; before?: string; + after?: string; limit?: number; }): Promise { const q = new URLSearchParams(); if (opts?.unread) q.set('unread', 'true'); if (opts?.before) q.set('before', opts.before); + if (opts?.after) q.set('after', opts.after); if (opts?.limit !== undefined) q.set('limit', String(opts.limit)); const suffix = q.toString(); return apiJson(`/api/notifications${suffix ? `?${suffix}` : ''}`); diff --git a/frontend/src/lib/composables/useNotifications.svelte.ts b/frontend/src/lib/composables/useNotifications.svelte.ts index 7e95d0fa..4a388e0a 100644 --- a/frontend/src/lib/composables/useNotifications.svelte.ts +++ b/frontend/src/lib/composables/useNotifications.svelte.ts @@ -9,9 +9,33 @@ * * Message-bus contract: the FE subscribes to `user:{me}:notifications` * (auto-subscribed server-side on WS session open — no `rt.subscribe` - * frame needed from the client) and refetches the row list whenever - * a `notification_received` event arrives. The DB is truth; the bus - * event just says "there's new data, refresh". + * frame needed from the client) and refetches on every push. The DB + * is truth; the bus event is a cache-invalidation hint. + * + * # Delta catch-up + dedup + * + * Two paths can deliver the SAME row and must not double-count it: + * + * 1. **WS live push** — `notification_received` event → calls + * `refreshDelta(#lastReceivedAt)` which fetches + * `?after=&limit=100`, merges the result into the + * reactive list. + * 2. **Reconnect catch-up** — after a grace-close (tab idle > 60 s) + * or a network drop, the WS reopens and `onReconnect` fires the + * same `refreshDelta(#lastReceivedAt)`. This backfills rows that + * landed while the socket was closed. + * + * The race: a NEW notification created after the reconnect but + * before the delta fetch returns lands via BOTH paths — WS push + * (delta fetch A) and reconnect (delta fetch B). Dedup lives in + * `mergeById`: incoming rows keyed on `id` displace any existing + * entry with the same id, so the row appears exactly once. Server + * `read_at` always wins over local because incoming replaces. + * + * `#lastReceivedAt` is the newest `created_at` we've observed. It + * feeds every delta fetch. Initial `refresh()` seeds it from the + * newest returned row; subsequent merges update it to the newest of + * the incoming set. */ import { messageBus } from '$lib/message-bus/client.svelte'; import { session } from '$lib/stores/session.svelte'; @@ -28,11 +52,37 @@ import log from 'loglevel'; const bellLog = log.getLogger('oxi:notifications'); +/** + * Merge `incoming` rows into `existing`, deduplicating on `id`. + * Where an id appears in both, the incoming (fresh-from-server) + * copy wins — so a `read_at` flip visible in `incoming` correctly + * overrides a stale local unread state. Result stays sorted + * newest-first by `created_at`. + * + * Exported for the unit tests to exercise the race semantics + * without spinning up a full store. + */ +export function mergeById(existing: Notification[], incoming: Notification[]): Notification[] { + if (incoming.length === 0) return existing; + // Local lookup set — pure function, no reactive state involved, + // so `SvelteSet` would add allocations without buying anything. + // eslint-disable-next-line svelte/prefer-svelte-reactivity + const incomingIds = new Set(incoming.map((n) => n.id)); + const kept = existing.filter((n) => !incomingIds.has(n.id)); + // String compare of ISO-8601 UTC timestamps sorts identically + // to Date compare — cheaper, no allocation per row. + return [...incoming, ...kept].sort((a, b) => b.created_at.localeCompare(a.created_at)); +} + class NotificationsStore { #items = $state([]); #unread = $state(0); #loading = $state(false); #error = $state(null); + /** Newest `created_at` we've observed, ISO 8601. Feeds the + * `?after=…` cursor on delta fetches. `null` until the first + * successful `refresh()` seeds it. */ + #lastReceivedAt: string | null = null; get items(): Notification[] { return this.#items; @@ -48,8 +98,9 @@ class NotificationsStore { } /** - * Fetch the newest page + refresh the badge count. Idempotent — - * safe to call on every bus push, on mount, on visibility return. + * Full refresh — replaces the local list with the newest page + * from the server. Used on initial mount + as fallback when a + * delta fetch fails or a mutation reconciliation runs. */ async refresh(): Promise { this.#loading = true; @@ -57,6 +108,7 @@ class NotificationsStore { const res = await listNotifications({ limit: 50 }); this.#items = res.items; this.#unread = res.unread_count; + this.#lastReceivedAt = res.items[0]?.created_at ?? this.#lastReceivedAt; this.#error = null; } catch (e) { this.#error = e instanceof Error ? e.message : String(e); @@ -66,6 +118,48 @@ class NotificationsStore { } } + /** + * Delta fetch — pulls only rows strictly newer than + * `#lastReceivedAt` (or does nothing if we've never fetched yet; + * the caller should fall back to `refresh()` in that case). + * Merges via `mergeById` so a concurrent WS push and reconnect + * catch-up can't double-count a row that landed twice. + * + * Silent no-op when the server returns 0 rows — we're already in + * sync. Updates `#lastReceivedAt` to the newest of the merged set. + */ + async refreshDelta(): Promise { + if (this.#lastReceivedAt === null) { + // Never fetched — fall back to a full refresh so the + // caller doesn't need to distinguish the two cases. + return this.refresh(); + } + try { + // `limit: 100` sized to cover realistic bell traffic per + // hour without paginating; a rare heavy sender who blows + // past 100 in one gap still gets 100 newest and the DB + // row count (unread badge) stays authoritative. + const res = await listNotifications({ + after: this.#lastReceivedAt, + limit: 100 + }); + if (res.items.length > 0) { + this.#items = mergeById(this.#items, res.items); + // Newest of merged set — take the first item's + // created_at since the result is sorted DESC. + this.#lastReceivedAt = res.items[0].created_at; + } + // unread_count is the authoritative live server count — + // always update it even when the delta was empty (a row + // could have been mark-read'd on another device). + this.#unread = res.unread_count; + this.#error = null; + } catch (e) { + this.#error = e instanceof Error ? e.message : String(e); + bellLog.warn('notifications delta failed', e); + } + } + /** Badge-only fast path — avoids fetching payloads. */ async refreshBadge(): Promise { try { @@ -127,6 +221,7 @@ class NotificationsStore { reset(): void { this.#items = []; this.#unread = 0; + this.#lastReceivedAt = null; this.#error = null; } } @@ -135,13 +230,14 @@ class NotificationsStore { export const notifications = new NotificationsStore(); /** - * Wire the bell into a component's lifecycle. Fires an initial fetch - * on mount, subscribes to `user:{me}:notifications` for live pushes, - * refetches on reconnect (bus events lost during outage window). + * Wire the bell into a component's lifecycle. Fires an initial full + * fetch on mount, subscribes to `user:{me}:notifications` for live + * pushes, delta-fetches on reconnect (backfills rows missed during + * grace-close / network gap). * - * Call once from the app root (`+layout.svelte`) — this store is - * global. Additional callers do NOT need to re-mount; they can just - * read `notifications.items` / `notifications.unread`. + * Call once from the app root (`AppShell`) — this store is global. + * Additional callers do NOT need to re-mount; they can just read + * `notifications.items` / `notifications.unread`. */ export function useNotifications(): void { $effect(() => { @@ -165,10 +261,13 @@ export function useNotifications(): void { `user:${userId}:notifications`, (params) => { if (params.event === 'notification_received') { - // Bus event carries only the poke. Refetch the - // list — cheap, gives us the new row with its - // full payload from truth. - void notifications.refresh(); + // Bus event carries only the poke. Delta-fetch + // from `#lastReceivedAt` — cheap when the store + // is caught up, brings the new row with its full + // payload from truth. Dedup via `mergeById` + // handles the race with an in-flight reconnect + // catch-up returning the same row. + void notifications.refreshDelta(); } }, () => { @@ -179,9 +278,17 @@ export function useNotifications(): void { ); const releaseReconnect = messageBus.onReconnect(() => { - // A push we missed during the outage window is only - // recoverable by rereading the DB. - void notifications.refresh(); + // Tab was hidden > 60 s, or network dropped. WS just + // reopened — any bus events published during the gap + // are lost. Backfill via the `?after=` + // cursor. Server's `unread_count` in the response is + // authoritative — a mark-read on another device while + // we were dark shows up here. + // + // Race with a live rt.event that lands milliseconds + // later: `mergeById` deduplicates on `id`, so the + // same row from both paths appears exactly once. + void notifications.refreshDelta(); }); return () => { diff --git a/frontend/src/lib/composables/useNotifications.test.ts b/frontend/src/lib/composables/useNotifications.test.ts new file mode 100644 index 00000000..93566512 --- /dev/null +++ b/frontend/src/lib/composables/useNotifications.test.ts @@ -0,0 +1,58 @@ +import { describe, expect, it } from 'vitest'; +import { mergeById } from './useNotifications.svelte'; +import type { Notification } from '$lib/api/types'; + +function row(id: string, created_at: string, read_at: string | null = null): Notification { + return { + id, + kind: 'share_granted', + payload: {}, + created_at, + read_at + }; +} + +describe('mergeById — WS-push vs delta-fetch race dedup', () => { + it('preserves existing when incoming is empty', () => { + const existing = [row('a', '2026-09-11T10:00:00Z'), row('b', '2026-09-11T09:00:00Z')]; + expect(mergeById(existing, [])).toEqual(existing); + }); + + it('appends non-overlapping incoming and sorts newest-first', () => { + const existing = [row('b', '2026-09-11T09:00:00Z')]; + const incoming = [row('a', '2026-09-11T10:00:00Z')]; + const merged = mergeById(existing, incoming); + expect(merged.map((n) => n.id)).toEqual(['a', 'b']); + }); + + it('dedupes on id — same row from WS push and delta fetch appears once', () => { + // Simulates the race: `x` was delivered live via rt.event + // and appended locally, then the reconnect delta fetch + // returns the same `x` again. Must not double it. + const existing = [row('x', '2026-09-11T10:00:00Z')]; + const incoming = [row('x', '2026-09-11T10:00:00Z')]; + expect(mergeById(existing, incoming)).toHaveLength(1); + }); + + it('lets server value win — read_at flip visible in incoming', () => { + // User marked `x` as read on another device. Local copy is + // stale (still unread). The delta fetch returns the fresh + // row with read_at populated — that must win. + const existing = [row('x', '2026-09-11T10:00:00Z', null)]; + const incoming = [row('x', '2026-09-11T10:00:00Z', '2026-09-11T10:05:00Z')]; + const merged = mergeById(existing, incoming); + expect(merged).toHaveLength(1); + expect(merged[0].read_at).toBe('2026-09-11T10:05:00Z'); + }); + + it('merges mixed overlap correctly', () => { + const existing = [row('b', '2026-09-11T09:00:00Z'), row('a', '2026-09-11T08:00:00Z')]; + const incoming = [ + row('c', '2026-09-11T10:00:00Z'), // new + row('b', '2026-09-11T09:00:00Z', '2026-09-11T09:30:00Z') // updated + ]; + const merged = mergeById(existing, incoming); + expect(merged.map((n) => n.id)).toEqual(['c', 'b', 'a']); + expect(merged[1].read_at).toBe('2026-09-11T09:30:00Z'); + }); +}); diff --git a/src/domain/repositories/notification_repository.rs b/src/domain/repositories/notification_repository.rs index 6055711a..727b69d5 100644 --- a/src/domain/repositories/notification_repository.rs +++ b/src/domain/repositories/notification_repository.rs @@ -18,21 +18,30 @@ use crate::common::errors::DomainError; use crate::domain::entities::notification::{NewNotification, Notification}; /// Optional filter for [`NotificationRepository::list_for_user`]. All -/// fields are additive — `None` means "no restriction on this axis". +/// fields are additive — the default (Default::default) applies no +/// restriction on any axis. #[derive(Debug, Clone, Default)] pub struct NotificationListFilter { /// Cap on rows returned. Default at the service layer is 50; the - /// repo does not impose one so a full-export use case remains - /// possible. + /// repo caps defensively at 500 so a runaway caller can't drag + /// the DB. pub limit: Option, - /// When `Some(true)`, return only rows with `read_at IS NULL`. - /// When `Some(false)`, return only rows with `read_at IS NOT NULL`. - /// `None` returns both. - pub unread_only: Option, - /// When `Some(t)`, return only rows created strictly before `t`. - /// Cursor-style pagination: caller passes the oldest `created_at` - /// from the previous page. + /// `true` → return only rows with `read_at IS NULL`. `false` + /// (default) returns both read and unread. There is no + /// "read-only" filter — no consumer needed it, and adding one + /// bloats the query surface. + pub unread_only: bool, + /// When `Some(t)`, return only rows created strictly BEFORE `t`. + /// Cursor-style pagination for the "load older page" flow: caller + /// passes the oldest `created_at` from the previous page. pub before: Option>, + /// When `Some(t)`, return only rows created strictly AFTER `t`. + /// Delta-catch-up cursor for the "since last seen" flow — used by + /// the FE bell on WS reconnect / tab reactivation to fetch rows + /// that arrived during a disconnect window. Combines with + /// `before` (both applied); combining them semantically bounds + /// the returned range on both sides. + pub after: Option>, } #[async_trait] diff --git a/src/infrastructure/repositories/pg/notification_pg_repository.rs b/src/infrastructure/repositories/pg/notification_pg_repository.rs index ee219496..8db6fd93 100644 --- a/src/infrastructure/repositories/pg/notification_pg_repository.rs +++ b/src/infrastructure/repositories/pg/notification_pg_repository.rs @@ -79,105 +79,36 @@ impl NotificationRepository for NotificationPgRepository { user_id: Uuid, filter: &NotificationListFilter, ) -> Result, DomainError> { - // Dynamic-shape query built to still hit the - // notifications_user_created_read index — every branch keys - // on (user_id, created_at DESC). + // One dynamic query covers every combination of + // (unread_only, before, after). NULL sentinels short-circuit + // the corresponding predicate at planner time, so the + // notifications_user_created_read index still drives the + // scan — the extra `IS NULL` checks are constant-folded. + // + // `before` and `after` combine: passing both bounds the + // returned range on both sides — useful for future + // "paginate a specific window" flows, harmless today when + // callers use one at a time. let limit: i64 = filter.limit.unwrap_or(50).min(500) as i64; - let rows = match (filter.unread_only, filter.before) { - (None, None) => { - sqlx::query( - r#" - SELECT id, user_id, kind, payload, created_at, read_at - FROM notif.notifications - WHERE user_id = $1::uuid - ORDER BY created_at DESC - LIMIT $2 - "#, - ) - .bind(user_id) - .bind(limit) - .fetch_all(self.pool.as_ref()) - .await - } - (Some(true), None) => { - sqlx::query( - r#" - SELECT id, user_id, kind, payload, created_at, read_at - FROM notif.notifications - WHERE user_id = $1::uuid AND read_at IS NULL - ORDER BY created_at DESC - LIMIT $2 - "#, - ) - .bind(user_id) - .bind(limit) - .fetch_all(self.pool.as_ref()) - .await - } - (Some(false), None) => { - sqlx::query( - r#" - SELECT id, user_id, kind, payload, created_at, read_at - FROM notif.notifications - WHERE user_id = $1::uuid AND read_at IS NOT NULL - ORDER BY created_at DESC - LIMIT $2 - "#, - ) - .bind(user_id) - .bind(limit) - .fetch_all(self.pool.as_ref()) - .await - } - (None, Some(before)) => { - sqlx::query( - r#" - SELECT id, user_id, kind, payload, created_at, read_at - FROM notif.notifications - WHERE user_id = $1::uuid AND created_at < $2 - ORDER BY created_at DESC - LIMIT $3 - "#, - ) - .bind(user_id) - .bind(before) - .bind(limit) - .fetch_all(self.pool.as_ref()) - .await - } - (Some(true), Some(before)) => { - sqlx::query( - r#" - SELECT id, user_id, kind, payload, created_at, read_at - FROM notif.notifications - WHERE user_id = $1::uuid AND read_at IS NULL AND created_at < $2 - ORDER BY created_at DESC - LIMIT $3 - "#, - ) - .bind(user_id) - .bind(before) - .bind(limit) - .fetch_all(self.pool.as_ref()) - .await - } - (Some(false), Some(before)) => { - sqlx::query( - r#" - SELECT id, user_id, kind, payload, created_at, read_at - FROM notif.notifications - WHERE user_id = $1::uuid AND read_at IS NOT NULL AND created_at < $2 - ORDER BY created_at DESC - LIMIT $3 - "#, - ) - .bind(user_id) - .bind(before) - .bind(limit) - .fetch_all(self.pool.as_ref()) - .await - } - } + let rows = sqlx::query( + r#" + SELECT id, user_id, kind, payload, created_at, read_at + FROM notif.notifications + WHERE user_id = $1::uuid + AND ($2::bool = FALSE OR read_at IS NULL) + AND ($3::timestamptz IS NULL OR created_at < $3) + AND ($4::timestamptz IS NULL OR created_at > $4) + ORDER BY created_at DESC + LIMIT $5 + "#, + ) + .bind(user_id) + .bind(filter.unread_only) + .bind(filter.before) + .bind(filter.after) + .bind(limit) + .fetch_all(self.pool.as_ref()) + .await .map_err(|e| db_err("list_for_user", e))?; rows.iter().map(Self::map_row).collect() diff --git a/src/interfaces/api/handlers/notifications_handler.rs b/src/interfaces/api/handlers/notifications_handler.rs index 2a9c02b7..f9e9e13a 100644 --- a/src/interfaces/api/handlers/notifications_handler.rs +++ b/src/interfaces/api/handlers/notifications_handler.rs @@ -68,9 +68,15 @@ pub struct ListQuery { /// When `true`, return only unread rows. Default: `false` (both). #[serde(default)] pub unread: bool, - /// Cursor — return rows strictly before this `created_at`. Omit - /// for the newest page. + /// Older-than cursor — return rows strictly BEFORE this + /// `created_at`. Used by the "load older page" pagination flow. + /// Omit for the newest page. pub before: Option>, + /// Newer-than cursor — return rows strictly AFTER this + /// `created_at`. Used by the FE bell on WS reconnect / tab + /// reactivation to catch up on rows that arrived during a + /// disconnect window. Combines with `before` if both are set. + pub after: Option>, /// Max rows returned. Server-side clamp at 500. pub limit: Option, } @@ -101,7 +107,8 @@ pub struct MarkAllReadResponseDto { path = "/api/notifications", params( ("unread" = Option, Query, description = "Only return unread rows"), - ("before" = Option>, Query, description = "Cursor — rows strictly before this created_at"), + ("before" = Option>, Query, description = "Cursor — rows strictly before this created_at (load-older pagination)"), + ("after" = Option>, Query, description = "Cursor — rows strictly after this created_at (delta catch-up on WS reconnect / tab reactivation)"), ("limit" = Option, Query, description = "Max rows (server-side clamp at 500)"), ), responses( @@ -117,8 +124,9 @@ pub async fn list_notifications( ) -> Result, AppError> { let filter = NotificationListFilter { limit: query.limit, - unread_only: if query.unread { Some(true) } else { None }, + unread_only: query.unread, before: query.before, + after: query.after, }; let rows = service.list_for_user(auth_user.id, filter).await?; let unread_count = service.count_unread_for_user(auth_user.id).await?; diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh index adb68eac..3ed851f6 100755 --- a/tests/api/rt_bus_check.sh +++ b/tests/api/rt_bus_check.sh @@ -97,6 +97,25 @@ # and delivers events, an admin (or # anyone else) could snoop on other # users' notification streams. +# S16 Notif ?after= cursor — the delta catch-up cursor the FE +# bell hits on WS reconnect / tab +# reactivation. Snapshots the newest +# row's `created_at` as T0, fires a +# fresh share (folder D) that lands +# exactly one new row, and asserts: +# (a) GET ?after=T0 returns exactly +# one row — the folder-D row +# (guards: predicate applied at +# all; silently-dropped param +# would return everything). +# (b) GET ?after= +# returns exactly zero rows +# (guards: bound is strict `>`, +# not `>=` — a `>=` regression +# would break FE `mergeById` +# dedup because rows would come +# in both via WS push and via +# the delta fetch). # # Exit non-zero on any failure — run.sh treats that as a suite failure. # ───────────────────────────────────────────────────────────────────────────── @@ -819,4 +838,74 @@ if ! "$HELPER_BIN" expect-denied \ fi log "S15 OK" -log "All fifteen message-bus scenarios passed." +# ── Scenario 16 — Notifications `?after=` cursor ──────────────────────────── +# The FE bell fires `refreshDelta()` with `?after=` +# on WS reconnect (grace-close return, network reconnect) and on +# rt.event push. `mergeById` dedups client-side; the server-side +# strict-`>` predicate is what keeps the boundary row from +# double-arriving in the first place. This scenario locks both +# invariants at the wire. +# +# Reuses S13's fresh row on folder C as T0. Creates a new folder D +# and grants it → one row lands strictly after T0. Then two +# assertions: `?after=T0` returns EXACTLY that one row (predicate +# applied), and `?after=` returns ZERO rows +# (strict `>` boundary — the row equal to its own cursor is +# excluded, so no duplicate delivery). +log "S16: cursor delta — grant on folder D, expect ?after=T0 returns 1 row, ?after=T1 returns 0." +# T0 = the S13 row's created_at. Pull it fresh via list so this +# scenario stays self-contained; the S13 row is the newest row +# for user2 at this point (S8 grants are older, S13/C is newest). +notifs_pre=$(c_get "$base_url/api/notifications" "$user2_token") +t0=$(printf '%s' "$notifs_pre" | jq -r --arg fc "$folder_c" \ + 'first(.items[] | select(.kind == "share_granted" and .payload.resource_id == $fc)) | .created_at') +[[ -n "$t0" && "$t0" != "null" ]] \ + || { printf '%s\n' "$notifs_pre" >&2; die "S16: could not resolve T0 from folder C row"; } + +# Fresh folder D, grant to user2 → one share_granted row lands. +folder_d=$(c_post "$base_url/api/folders" "$user1_token" \ + "$(printf '{"name":"rt_bus_D_%s","parent_id":"%s"}' "$suffix" "$root_id")" | jq -r '.id') +[[ -n "$folder_d" && "$folder_d" != "null" ]] || die "S16: folder D creation failed" + +grant_d=$(c_post "$base_url/api/grants" "$user1_token" \ + "$(printf '{"subject":{"type":"user","id":"%s"},"resource":{"type":"folder","id":"%s"},"role":"viewer"}' \ + "$user2_id" "$folder_d")") +grant_d_id=$(printf '%s' "$grant_d" | jq -r '.grants[0].id') +[[ -n "$grant_d_id" && "$grant_d_id" != "null" ]] \ + || die "S16: grant on folder D failed: $grant_d" + +# `after` must URL-encode the ISO timestamp — `:` and `+` are +# reserved. curl's `--data-urlencode`/`-G` handles it cleanly. +notifs_after_t0=$(curl -sS -G \ + -H "Accept: application/json" \ + -H "Authorization: Bearer $user2_token" \ + --data-urlencode "after=$t0" \ + "$base_url/api/notifications") +count_after_t0=$(printf '%s' "$notifs_after_t0" | jq -r '.items | length') +[[ "$count_after_t0" == "1" ]] \ + || { printf '%s\n' "$notifs_after_t0" >&2; die "S16: expected 1 row after T0, got $count_after_t0"; } +# The one row MUST be the folder-D row. +d_kind=$(printf '%s' "$notifs_after_t0" | jq -r '.items[0].kind') +d_resource=$(printf '%s' "$notifs_after_t0" | jq -r '.items[0].payload.resource_id') +[[ "$d_kind" == "share_granted" && "$d_resource" == "$folder_d" ]] \ + || { printf '%s\n' "$notifs_after_t0" >&2; die "S16: post-T0 row not folder D (kind=$d_kind resource=$d_resource)"; } + +# T1 = folder-D row's own created_at. Bound is strict `>`, so a +# query at T1 must return zero rows (the boundary row is excluded). +# This is what keeps the FE's mergeById honest — a `>=` regression +# would return the boundary row here, then the WS push would +# deliver it AGAIN, and only client-side dedup would save us. +t1=$(printf '%s' "$notifs_after_t0" | jq -r '.items[0].created_at') +[[ -n "$t1" && "$t1" != "null" ]] || die "S16: could not resolve T1" + +notifs_after_t1=$(curl -sS -G \ + -H "Accept: application/json" \ + -H "Authorization: Bearer $user2_token" \ + --data-urlencode "after=$t1" \ + "$base_url/api/notifications") +count_after_t1=$(printf '%s' "$notifs_after_t1" | jq -r '.items | length') +[[ "$count_after_t1" == "0" ]] \ + || { printf '%s\n' "$notifs_after_t1" >&2; die "S16: expected 0 rows after T1 (strict '>' bound), got $count_after_t1"; } +log "S16 OK" + +log "All sixteen message-bus scenarios passed."