diff --git a/docs/config/env.md b/docs/config/env.md index e7703a49..f3f6f727 100644 --- a/docs/config/env.md +++ b/docs/config/env.md @@ -120,6 +120,7 @@ rather than as a visible error. | `OXICLOUD_GRANT_CLEANUP_ENABLED` | `true` | Background daemon that deletes expired rows from `storage.role_grants`. The authorization engine already filters expired grants out of every permission check at read time (`expires_at IS NULL OR expires_at > NOW()`), so leaving expired rows in place is a hygiene issue — not a security one. This daemon garbage-collects them daily. Set to `false` to keep every expired grant row forever (uncommon; a fresh install rarely wants this). | | `OXICLOUD_GRANT_CLEANUP_GRACE_DAYS` | `15` | Days past a grant's `expires_at` before the row is eligible for deletion. The grace window preserves the audit / support answer to "what happened to my access?" for a couple of weeks past expiration. Values below 1 are legal but discouraged — the recommendation is **≥ 15 days**. Values above the actual grant TTL used by clients waste index space; a few weeks is the sweet spot. | | `OXICLOUD_GRANT_CLEANUP_INTERVAL_HOURS` | `24` | How often the grant-cleanup daemon fires. Clamped to a minimum of 1 hour. Adjusting this doesn't change what gets deleted — only how promptly. Daily is fine for any realistic grant volume. | +| `OXICLOUD_NOTIFICATIONS_RETENTION_DAYS` | `30` | Retention window for **read** notification rows (`notif.notifications`). The `notifications_cleanup` scheduled job runs daily and deletes rows where `read_at IS NOT NULL` and `read_at < now() - retention_days`. Unread rows are preserved unconditionally — the whole point of the durable table is that a user offline for a month still sees the share-granted notice on next login. Clamped to a minimum of 1 (0 would purge every read row on every tick). Adjust down for compliance-sensitive deployments where "cleared once seen" matters; adjust up when operators expect users to reference old notifications for support. | | `OXICLOUD_WEBDAV_DRIVE_LISTING_PREFIX` | `@drive` | Native WebDAV URL segment that renders the caller's drive list. Sanitized by trimming leading/trailing `/`. Three shapes: (1) default `@drive` — `/webdav/…` addresses the caller's default personal drive (back-compat), `/webdav/@drive/` returns the drive listing, `/webdav/@drive//…` targets a specific drive. (2) empty string `""` — `/webdav/` IS the drive listing, `/webdav//…` targets a specific drive, no default-drive shortcut. (3) any other string (e.g. `drives`) — same shape as `@drive` with that segment substituted. Only drives the caller has Read on via `role_grants` resolve. | ## Storage Entries (multi-entry, recommended) diff --git a/docs/plan/message-bus.md b/docs/plan/message-bus.md index ded7f55f..4525f1d4 100644 --- a/docs/plan/message-bus.md +++ b/docs/plan/message-bus.md @@ -18,9 +18,9 @@ almost no extra scaffolding. ## Status — 2026-09-11 -The `feat/message-bus` branch delivers **D + F + follow-ups shipped -end-to-end** on the FE and BE, verified by S1–S11 in the api-test -smoke suite plus manual multi-user E2E. Live today: +The `feat/message-bus` branch delivers **D + F + Job dashboard live + +follow-ups shipped end-to-end** on the FE and BE, verified by S1–S12 +in the api-test smoke suite plus manual multi-user E2E. Live today: - **Bus core** — `MessageBus` port + `InProcessMessageBus` + `NoopReplicator`. `📤 bus publish` trace under @@ -71,21 +71,29 @@ smoke suite plus manual multi-user E2E. Live today: in every mutation entry point so `$state` reads don't leak into caller `$effect` deps. - **AuthZ tested** — S3 (folder no_read), S4 (nonexistent folder = - anti-enum parity), S9 (cross-user identity topic → `topic_forbidden`). + anti-enum parity), S9 (cross-user identity topic → `topic_forbidden`), + S12 (`job:*` non-admin denied — Class-3 role-scoped gate). - **Ticket tested** — S10 (happy path), S11 (single-use replay rejected). +- **Job dashboard live** — scheduler engine publishes + `JobRunStarted` / `JobRunEnded` on `Topic::Job(name)` around every + dispatch; `useJobTopic` + `AdminJobsPanel` subscribes to every + registered job's topic (keyed on the sorted name set so the 5 s + poll doesn't churn subs). State flips within a network hop instead + of waiting up to POLL_MS. Progress publishes are deferred (see + below); the polling refresh stays as fallback. Active — still under Phase A, ordered by priority: -- **Job dashboard live** (next) — `JobRegistry` publishes step - progress + terminal state on `job:{id}`; the admin jobs view - subscribes and drops its polling. Small; same shape as folder-live. - Value: an operator who triggers a long-running job (backend - migration, thumbnail import, etc.) can navigate to another admin - page and come back without losing progress visibility. - **Notifications table + bell** (E) — topic + producer + auto-sub land here. Same pattern as `:authz`. Larger; unblocks Phase-B `@mentions`. +- **Job progress publishes** (small follow-up to Job dashboard) — + handler-side per-run publisher + 3 s throttle so long jobs + (backend_migration, thumb_derived_import…) push `JobRunProgress` + events. Wire is already in place (`useJobTopic.onProgress`, + `MessageBusEvent::JobRunProgress`); waits for a per-run + `ProgressReporter` handle threaded into `JobHandler::run`. Deferred — see the Roadmap section's `## Deferred` block and the `project_message_bus_reconnect_gap` memory: @@ -1242,13 +1250,13 @@ Ships the infrastructure and the two most visible consumers together. `useReconnect` composable → folder view refetches after WS comes back. Bridges the in-memory-bus "events lost during outage" gap (see `project_message_bus_reconnect_gap` memory). -- **Job dashboard live** — TODO (next slice). `JobRegistry` - publishes step progress and terminal state on `job:{id}`; FE - job dashboard subscribes and replaces polling. Operator value: - once a long-running job is triggered (backend migration, thumb - import, blobs consistency…), the admin can navigate to another - page and come back without losing progress visibility — the WS - push keeps whatever component is subscribed up-to-date. +- **Job dashboard live** — SHIPPED 2026-09-11. Scheduler engine + publishes `JobRunStarted` + `JobRunEnded` on `Topic::Job(name)` + around every dispatch; `useJobTopic` + `AdminJobsPanel` subscribe + to every registered job's topic and flip state within a network + hop. Progress publishes deferred to a follow-up (needs a per-run + `ProgressReporter` threaded into `JobHandler::run`). The 5 s + poll stays as fallback. - **Notifications table + bell** — TODO (Slice E). New `notifications` table + `NotificationService` port; initial ingesters for `share-granted`, `new-login-from-new-device`, @@ -1261,9 +1269,10 @@ Ships the infrastructure and the two most visible consumers together. which this plan sketches but doesn't ship (`rt_ws.rs` today drops binary frames with a debug log). -Deliverables sized ~4 weeks end-to-end. Slice D (folder-live) and -Slice F (ticket flow) landed 2026-09-11. Slices E + collab are the -open work in Phase A. +Deliverables sized ~4 weeks end-to-end. Slice D (folder-live), +Slice F (ticket flow), and Job dashboard live all landed 2026-09-11. +Slice E (notifications bell) + collab are the remaining open work +in Phase A. ### Deferred — everything below is on the shelf diff --git a/example.env b/example.env index f8373e6a..90773277 100644 --- a/example.env +++ b/example.env @@ -406,6 +406,17 @@ DATABASE_URL=postgres://postgres:postgres@localhost:5432/oxicloud #OXICLOUD_GRANT_CLEANUP_GRACE_DAYS=15 #OXICLOUD_GRANT_CLEANUP_INTERVAL_HOURS=24 +# ─── Persistent notifications (bell) ───────────────────────────── +# Retention window (in days) for READ notifications in the bell. +# Unread rows are preserved unconditionally — that's the point of +# the durable table: a user offline for a month still sees the +# share-granted notice on next login. The `notifications_cleanup` +# scheduled job runs daily and deletes read rows older than this. +# +# Minimum 1 day (0 would purge every read row on every tick — the +# service clamps defensively). Default: 30. +#OXICLOUD_NOTIFICATIONS_RETENTION_DAYS=30 + # Enable search functionality (default: true) #OXICLOUD_ENABLE_SEARCH=true diff --git a/frontend/src/lib/api/endpoints/notifications.ts b/frontend/src/lib/api/endpoints/notifications.ts new file mode 100644 index 00000000..84a442db --- /dev/null +++ b/frontend/src/lib/api/endpoints/notifications.ts @@ -0,0 +1,76 @@ +/** + * Persistent notifications (bell) — REST client. + * + * Backs `useNotifications` (composable) and `NotificationBell` + * (component). The bell reads from these; the message bus is a + * cache-invalidation hint that triggers a refetch, not a data path. + * See `docs/plan/message-bus.md § Slice E` for the pattern. + */ +import { apiJson } from '$lib/api/client'; +import { apiFetch } from '$lib/api/client'; +import { getCsrfHeaders } from '$lib/api/csrf'; +import type { + MarkAllReadResponse, + Notification, + NotificationListResponse, + UnreadCountResponse +} from '$lib/api/types'; + +/** List newest-first. Optional `unread` filter, `before` cursor, `limit` cap. */ +export async function listNotifications(opts?: { + unread?: boolean; + before?: 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?.limit !== undefined) q.set('limit', String(opts.limit)); + const suffix = q.toString(); + return apiJson(`/api/notifications${suffix ? `?${suffix}` : ''}`); +} + +/** Badge-only fast path — no payloads fetched. */ +export async function getUnreadCount(): Promise { + const res = await apiJson('/api/notifications/unread'); + return res.unread_count; +} + +/** + * Mark one notification as read. Always resolves — the server responds + * 204 regardless of whether the row existed or belonged to the caller + * (anti-enumeration). Duplicated calls are safe. + */ +export async function markNotificationRead(id: string): Promise { + const res = await apiFetch(`/api/notifications/${encodeURIComponent(id)}/read`, { + method: 'POST', + headers: getCsrfHeaders() + }); + if (!res.ok && res.status !== 204) { + throw new Error(`markNotificationRead failed: HTTP ${res.status}`); + } +} + +/** Bulk mark-all-read. Returns the number of rows the server flipped. */ +export async function markAllNotificationsRead(): Promise { + const res = await apiJson('/api/notifications/read-all', { + method: 'POST', + headers: getCsrfHeaders() + }); + return res.marked; +} + +/** Hard-delete one row. Same anti-enum shape as mark-read. */ +export async function deleteNotification(id: string): Promise { + const res = await apiFetch(`/api/notifications/${encodeURIComponent(id)}`, { + method: 'DELETE', + headers: getCsrfHeaders() + }); + if (!res.ok && res.status !== 204) { + throw new Error(`deleteNotification failed: HTTP ${res.status}`); + } +} + +// Convenience: expose the row shape for consumers that don't want +// to import from `$lib/api/types` too. +export type { Notification }; diff --git a/frontend/src/lib/api/types.ts b/frontend/src/lib/api/types.ts index 0f928a27..d8526160 100644 --- a/frontend/src/lib/api/types.ts +++ b/frontend/src/lib/api/types.ts @@ -934,3 +934,40 @@ export interface ServerConfig { features: ServerFeatures; server_status: ServerStatus; } + +// ─── Notifications (Slice E) ───────────────────────────────────── +// Row shape mirrors `application/dtos` output of the Rust backend. +// `payload` stays a raw JSON object (`Record`) — +// per-kind decoding is a UI concern (kind-specific components read +// what they need from the blob). Adding a new kind server-side does +// NOT churn this file; the FE renders a generic bell row for any +// unknown kind. +export interface Notification { + id: string; + kind: string; + payload: Record; + created_at: string; + /** `null` = unread. */ + read_at: string | null; +} + +export interface NotificationListResponse { + items: Notification[]; + unread_count: number; +} + +export interface UnreadCountResponse { + unread_count: number; +} + +export interface MarkAllReadResponse { + marked: number; +} + +/** Canonical kind slugs — mirror `domain::entities::notification::kind`. */ +export const NOTIFICATION_KIND = { + SHARE_GRANTED: 'share_granted', + NEW_LOGIN_FROM_NEW_DEVICE: 'new_login_from_new_device', + JOB_COMPLETED_FOR_YOU: 'job_completed_for_you', + STORAGE_QUOTA_THRESHOLD: 'storage_quota_threshold' +} as const; diff --git a/frontend/src/lib/components/AppShell.svelte b/frontend/src/lib/components/AppShell.svelte index 2e18ab21..f40016ac 100644 --- a/frontend/src/lib/components/AppShell.svelte +++ b/frontend/src/lib/components/AppShell.svelte @@ -26,6 +26,10 @@ import { session } from '$lib/stores/session.svelte'; import { theme, type Theme } from '$lib/stores/theme.svelte'; import { ui } from '$lib/stores/ui.svelte'; + import { + notifications as persistentNotifications, + useNotifications + } from '$lib/composables/useNotifications.svelte'; import { errorToast } from '$lib/utils/errors'; import { formatBytes } from '$lib/utils/format'; @@ -327,6 +331,82 @@ setTimeout(() => (bellRinging = false), 900); }); + // Persistent notifications (Slice E) — server-backed rows, + // survive reload, delivered via `user:{me}:notifications` bus + // topic + refetched from `GET /api/notifications`. Fires the + // initial hydrate + subscribes to the topic. Independent of the + // transient toast bell above (`ui.notifications`) — that stays + // as-is for upload-progress / one-shot messages; this stream + // carries `share_granted` and friends. + useNotifications(); + + // Merged unread count for the bell badge — transient toasts plus + // persistent unread rows. Same wire and same UX affordance so a + // user sees one number and one bell for both classes. + const totalUnread = $derived(ui.unread + persistentNotifications.unread); + const totalUnreadBadge = $derived(totalUnread > 99 ? '99+' : String(totalUnread)); + + /** Format the server-side `created_at` for a persistent row. */ + function formatPersistentTime(iso: string): string { + try { + return formatTime(new Date(iso).getTime()); + } catch { + return ''; + } + } + + /** Human summary for a persistent notification. Kind-specific + * wording lives here so the DTO stays payload-agnostic. */ + function persistentSummary(row: { kind: string; payload: Record }): string { + switch (row.kind) { + case 'share_granted': { + const role = String(row.payload.role ?? 'a role'); + const resType = String(row.payload.resource_type ?? 'resource'); + return t( + 'notifications.persistent.share_granted', + { role, resType }, + `You were granted ${role} on a ${resType}.` + ); + } + case 'new_login_from_new_device': + return t( + 'notifications.persistent.new_device_login', + 'A new device signed into your account.' + ); + case 'job_completed_for_you': { + const name = String(row.payload.name ?? row.payload.job_name ?? 'a job'); + return t('notifications.persistent.job_completed', { name }, `Job "${name}" finished.`); + } + case 'storage_quota_threshold': + return t( + 'notifications.persistent.quota_threshold', + 'You are approaching your storage quota.' + ); + default: + return t( + 'notifications.persistent.generic', + { kind: row.kind }, + `Notification (${row.kind}).` + ); + } + } + + /** Icon for a persistent row's kind. Falls back to a generic bell. */ + function persistentIcon(kind: string): string { + switch (kind) { + case 'share_granted': + return 'user-plus'; + case 'new_login_from_new_device': + return 'shield-alt'; + case 'job_completed_for_you': + return 'check-circle'; + case 'storage_quota_threshold': + return 'database'; + default: + return 'bell'; + } + } + function openMobileSearch() { searchActive = true; requestAnimationFrame(() => searchInputEl?.focus()); @@ -802,11 +882,19 @@ e.stopPropagation(); notifOpen = !notifOpen; menuOpen = false; - if (notifOpen) ui.markNotificationsRead(); + if (notifOpen) { + ui.markNotificationsRead(); + // Persistent rows stay unread until the user + // explicitly clicks one — opening the panel + // doesn't mark them read (unlike the transient + // toast bell, which resets on view). Keeps the + // bell's badge accurate to "still-relevant + // server-side rows" without a bulk mark-read. + } }} > - {#if ui.unread > 0}{ui.unreadBadge}{/if} + {#if totalUnread > 0}{totalUnreadBadge}{/if}
@@ -827,7 +915,7 @@ {/if}
- {#if ui.notifications.length === 0} + {#if ui.notifications.length === 0 && persistentNotifications.items.length === 0}
{t('notifications.empty', 'No notifications')} @@ -869,6 +957,43 @@
{/each} + {#if persistentNotifications.items.length > 0} + {#if ui.notifications.length > 0} + + {/if} + {#each persistentNotifications.items as row (row.id)} +
void persistentNotifications.markRead(row.id)} + onkeydown={(e) => { + if (e.key === 'Enter' || e.key === ' ') { + e.preventDefault(); + void persistentNotifications.markRead(row.id); + } + }} + > + + + +
+
{persistentSummary(row)}
+
+ {formatPersistentTime(row.created_at)} +
+
+
+ {/each} + {/if} {/if}
@@ -1295,6 +1420,14 @@ white-space: nowrap; } + /* Divider between transient toasts and persistent (server-backed) + rows. Slice E adds a section under the toast list; the divider + is only rendered when both sections have content. */ + .notif-section-divider { + border-top: 1px solid var(--color-border); + margin: 0.5rem 0; + } + /* Bell "ring" animation, replayed when bellRinging toggles on. */ .notif-bell-btn.ring :global(svg), .notif-bell-btn.ring :global(i) { diff --git a/frontend/src/lib/composables/useNotifications.svelte.ts b/frontend/src/lib/composables/useNotifications.svelte.ts new file mode 100644 index 00000000..7e95d0fa --- /dev/null +++ b/frontend/src/lib/composables/useNotifications.svelte.ts @@ -0,0 +1,192 @@ +/** + * Global bell store — persistent notifications (Slice E). + * + * Owns the reactive state for `NotificationBell`. Module-scoped so + * one instance drives every consumer in the SPA (badge in AppShell, + * slide-out panel, admin dashboard hooks, …). Same lifetime as + * `messageBus`: survives every intra-SPA navigation, dies only on + * full reload / tab close. + * + * 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". + */ +import { messageBus } from '$lib/message-bus/client.svelte'; +import { session } from '$lib/stores/session.svelte'; +import { serverConfig } from '$lib/stores/serverConfig.svelte'; +import { + deleteNotification as apiDelete, + getUnreadCount, + listNotifications, + markAllNotificationsRead, + markNotificationRead +} from '$lib/api/endpoints/notifications'; +import type { Notification } from '$lib/api/types'; +import log from 'loglevel'; + +const bellLog = log.getLogger('oxi:notifications'); + +class NotificationsStore { + #items = $state([]); + #unread = $state(0); + #loading = $state(false); + #error = $state(null); + + get items(): Notification[] { + return this.#items; + } + get unread(): number { + return this.#unread; + } + get loading(): boolean { + return this.#loading; + } + get error(): string | null { + return this.#error; + } + + /** + * Fetch the newest page + refresh the badge count. Idempotent — + * safe to call on every bus push, on mount, on visibility return. + */ + async refresh(): Promise { + this.#loading = true; + try { + const res = await listNotifications({ limit: 50 }); + this.#items = res.items; + this.#unread = res.unread_count; + this.#error = null; + } catch (e) { + this.#error = e instanceof Error ? e.message : String(e); + bellLog.warn('notifications refresh failed', e); + } finally { + this.#loading = false; + } + } + + /** Badge-only fast path — avoids fetching payloads. */ + async refreshBadge(): Promise { + try { + this.#unread = await getUnreadCount(); + } catch (e) { + bellLog.warn('badge refresh failed', e); + } + } + + async markRead(id: string): Promise { + // Optimistic update — flip locally, then confirm on the wire. + // Same pattern the folder-view uses on rename: reactive-first, + // server-eventually. A wire failure re-fetches from truth. + const row = this.#items.find((n) => n.id === id); + if (row && row.read_at === null) { + row.read_at = new Date().toISOString(); + this.#unread = Math.max(0, this.#unread - 1); + } + try { + await markNotificationRead(id); + } catch (e) { + bellLog.warn('markRead failed; reconciling', e); + await this.refresh(); + } + } + + async markAllRead(): Promise { + const now = new Date().toISOString(); + for (const row of this.#items) { + if (row.read_at === null) row.read_at = now; + } + this.#unread = 0; + try { + await markAllNotificationsRead(); + } catch (e) { + bellLog.warn('markAllRead failed; reconciling', e); + await this.refresh(); + } + } + + async delete(id: string): Promise { + const idx = this.#items.findIndex((n) => n.id === id); + if (idx >= 0) { + const [removed] = this.#items.splice(idx, 1); + if (removed && removed.read_at === null) { + this.#unread = Math.max(0, this.#unread - 1); + } + } + try { + await apiDelete(id); + } catch (e) { + bellLog.warn('delete failed; reconciling', e); + await this.refresh(); + } + } + + /** Reset — called on logout so a switch-user doesn't inherit the + * previous session's rows. */ + reset(): void { + this.#items = []; + this.#unread = 0; + this.#error = null; + } +} + +/** Module-scoped singleton — one bell state per SPA lifetime. */ +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). + * + * 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`. + */ +export function useNotifications(): void { + $effect(() => { + const userId = session.user?.id; + if (!userId) return; // not logged in — nothing to fetch + // Initial hydrate from DB truth. Runs whether or not the bus + // is enabled — the bell has to work in "polling only" mode + // when OXICLOUD_MESSAGEBUS_ENABLE=false too. + void notifications.refresh(); + }); + + $effect(() => { + if (!serverConfig.features.message_bus) return; + const userId = session.user?.id; + if (!userId) return; + + // The topic is auto-subscribed server-side on WS session open + // (same pattern as `:authz`); this call refcounts up to the + // existing sub, doesn't fire a second `rt.subscribe` frame. + const release = messageBus.subscribe( + `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(); + } + }, + () => { + // Server-evicted (session flipped) — clear local so + // the badge stops showing stale count. + notifications.reset(); + } + ); + + const releaseReconnect = messageBus.onReconnect(() => { + // A push we missed during the outage window is only + // recoverable by rereading the DB. + void notifications.refresh(); + }); + + return () => { + release(); + releaseReconnect(); + }; + }); +} diff --git a/frontend/src/lib/generated/message-bus/NotificationReceivedData.ts b/frontend/src/lib/generated/message-bus/NotificationReceivedData.ts new file mode 100644 index 00000000..2a76e7a7 --- /dev/null +++ b/frontend/src/lib/generated/message-bus/NotificationReceivedData.ts @@ -0,0 +1,8 @@ +// AUTO-GENERATED — do not edit by hand. +// Regenerate with `just asyncapi-ts`. +interface NotificationReceivedData { + created_at: string; + kind: string; + notification_id: string; +} +export type { NotificationReceivedData as default }; diff --git a/frontend/src/lib/generated/message-bus/RtEventKind.ts b/frontend/src/lib/generated/message-bus/RtEventKind.ts index ff42f7f7..df4aaab7 100644 --- a/frontend/src/lib/generated/message-bus/RtEventKind.ts +++ b/frontend/src/lib/generated/message-bus/RtEventKind.ts @@ -7,6 +7,7 @@ enum RtEventKind { FOLDER_RENAMED = 'folder_renamed', FOLDER_MOVED = 'folder_moved', FOLDER_DELETED = 'folder_deleted', + NOTIFICATION_RECEIVED = 'notification_received', JOB_RUN_STARTED = 'job_run_started', JOB_RUN_PROGRESS = 'job_run_progress', JOB_RUN_ENDED = 'job_run_ended' diff --git a/frontend/src/lib/generated/message-bus/RtEventParams.ts b/frontend/src/lib/generated/message-bus/RtEventParams.ts index 38602e61..5c485a8f 100644 --- a/frontend/src/lib/generated/message-bus/RtEventParams.ts +++ b/frontend/src/lib/generated/message-bus/RtEventParams.ts @@ -6,6 +6,7 @@ import type FolderCreatedData from './FolderCreatedData'; import type FolderRenamedData from './FolderRenamedData'; import type FolderMovedData from './FolderMovedData'; import type FolderDeletedData from './FolderDeletedData'; +import type NotificationReceivedData from './NotificationReceivedData'; import type JobRunStartedData from './JobRunStartedData'; import type JobRunProgressData from './JobRunProgressData'; import type JobRunEndedData from './JobRunEndedData'; @@ -22,6 +23,7 @@ interface RtEventParams { | FolderRenamedData | FolderMovedData | FolderDeletedData + | NotificationReceivedData | JobRunStartedData | JobRunProgressData | JobRunEndedData; diff --git a/migrations/20261026000000_notifications.sql b/migrations/20261026000000_notifications.sql new file mode 100644 index 00000000..2ddee7d1 --- /dev/null +++ b/migrations/20261026000000_notifications.sql @@ -0,0 +1,61 @@ +-- notif.notifications — durable per-user notification records. +-- +-- Backs the bell UI and the retention job. The message bus is best-effort +-- (a subscriber offline at publish time misses the push); this table is +-- the truth. Every `NotificationService::create` writes a row AND +-- publishes a `NotificationReceived` event on `user:{user_id}:notifications`. +-- A missed bus event recovers on the next `GET /api/notifications`. +-- +-- See `docs/plan/message-bus.md § Slice E` for the wire contract and +-- retention policy. + +CREATE SCHEMA IF NOT EXISTS notif; + +CREATE TABLE IF NOT EXISTS notif.notifications ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + + -- Recipient. Every row is scoped to exactly one user; a share fanned + -- to N members is N rows. Fanout truncation for very-large groups + -- happens in the ingester (see plan § Notification fanout truncated), + -- not here. + user_id UUID NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE, + + -- Notification kind — a stable slug the FE routes on for icon/label/ + -- action-button choice. New kinds are additive; never repurpose an + -- existing one. Initial kinds: + -- share_granted, new_login_from_new_device, + -- job_completed_for_you, storage_quota_threshold + kind TEXT NOT NULL, + + -- Per-kind opaque JSON with the fields the FE needs to render the + -- row without a follow-up API call (subject name, resource id, + -- action link…). Shape is a per-kind contract owned by the ingester; + -- the DB stays schema-free here so a new field doesn't require a + -- migration. + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + + -- Wall-clock creation stamp. Sort key for the bell. Server-clock, + -- not caller-clock — this is a DB-generated fact. + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + + -- NULL = unread; non-NULL = when the user explicitly marked it + -- read. Retention job deletes rows where read_at IS NOT NULL AND + -- read_at < now() - retention_days. + read_at TIMESTAMPTZ +); + +-- Bell fetch — GET /api/notifications lists a user's rows newest-first, +-- typically capped at ~50, sometimes filtered on unread. This one index +-- covers both the list query and the mark-all-read filter, and the +-- INCLUDE clause keeps common bell renders (id, kind, created_at, +-- read_at) index-only. +CREATE INDEX IF NOT EXISTS notifications_user_created_read + ON notif.notifications (user_id, created_at DESC) + INCLUDE (read_at, kind); + +-- Retention job DELETE — scans read-and-old rows only. Partial keeps +-- the index tiny in the typical steady state where most rows are +-- unread. +CREATE INDEX IF NOT EXISTS notifications_read_at + ON notif.notifications (read_at) + WHERE read_at IS NOT NULL; diff --git a/src/application/ports/message_bus_ports.rs b/src/application/ports/message_bus_ports.rs index a69bf99c..84e5ab9e 100644 --- a/src/application/ports/message_bus_ports.rs +++ b/src/application/ports/message_bus_ports.rs @@ -64,6 +64,16 @@ pub enum Topic { /// the eviction wiring lands (Phase-A follow-up). UserAuthz(Uuid), + /// A user's private notifications channel — poked when a + /// [`MessageBusEvent::NotificationReceived`] event fires. The WS + /// handler auto-subscribes each session at session open (same + /// pattern as [`Topic::UserAuthz`]). Payload is a thin fact + /// (`notification_id` + `kind`); the client refetches the row from + /// `GET /api/notifications` for the details. AuthZ: **strict + /// identity match** — no admin bypass, direct UUID equality, + /// anti-enumeration parity with [`Topic::UserAuthz`]. + UserNotifications(Uuid), + /// A named background job's run lifecycle — start / progress / /// end. Consumed by the admin job dashboard so operators who /// trigger a long-running job (backend migration, thumb import…) @@ -83,6 +93,7 @@ impl Topic { match self { Topic::Folder(id) => format!("folder:{id}"), Topic::UserAuthz(id) => format!("user:{id}:authz"), + Topic::UserNotifications(id) => format!("user:{id}:notifications"), Topic::Job(name) => format!("job:{name}"), } } @@ -97,10 +108,14 @@ impl Topic { return Ok(Topic::Folder(id)); } if let Some(rest) = s.strip_prefix("user:") - && let Some((id_str, "authz")) = rest.rsplit_once(':') + && let Some((id_str, suffix)) = rest.rsplit_once(':') { let id = Uuid::parse_str(id_str).map_err(|_| ParseTopicErr::BadUuid)?; - return Ok(Topic::UserAuthz(id)); + return match suffix { + "authz" => Ok(Topic::UserAuthz(id)), + "notifications" => Ok(Topic::UserNotifications(id)), + _ => Err(ParseTopicErr::Unknown), + }; } if let Some(name) = s.strip_prefix("job:") { // Job names are scheduler-registered short slugs — see @@ -135,6 +150,7 @@ impl Topic { resource: BusResource::Folder(*id), }, Topic::UserAuthz(id) => AuthzCheck::IdentityMatch { user_id: *id }, + Topic::UserNotifications(id) => AuthzCheck::IdentityMatch { user_id: *id }, Topic::Job(_) => AuthzCheck::RoleAdmin, } } @@ -291,6 +307,25 @@ pub enum MessageBusEvent { /// plan's Phase-B roadmap. AuthzChanged { affected_folders: Vec }, + /// A new notification was created for the caller — publishes on + /// [`Topic::UserNotifications`]. Payload is deliberately thin: the + /// FE learns "there's something new to look at" and calls + /// `GET /api/notifications` to load the row. Same recovery path a + /// missed push takes on next mount, so the wire event stays a + /// pure poke — no fields the bell needs to render on its own. + /// + /// `kind` is the notification's registered kind slug + /// (`share_granted`, `job_completed_for_you`, + /// `new_login_from_new_device`, `storage_quota_threshold`, …). + /// The FE may use it to route the toast (high-priority kinds pop + /// a toast; low-priority ones just bump the badge) but never + /// treats it as authoritative — the DB row is the truth. + NotificationReceived { + notification_id: Uuid, + kind: String, + created_at: chrono::DateTime, + }, + /// A background job's run started. Published on /// [`Topic::Job`]. `started_at` is server wall-clock (RFC 3339 /// serialised by serde). Admin dashboard's job-list view uses @@ -497,6 +532,15 @@ mod tests { assert_eq!(Topic::parse(&wire).unwrap(), t); } + #[test] + fn user_notifications_topic_roundtrip() { + let id = Uuid::new_v4(); + let t = Topic::UserNotifications(id); + let wire = t.to_wire_key(); + assert_eq!(wire, format!("user:{id}:notifications")); + assert_eq!(Topic::parse(&wire).unwrap(), t); + } + #[test] fn job_topic_roundtrip() { let t = Topic::Job("backend_migration".to_string()); @@ -539,7 +583,12 @@ mod tests { assert_eq!( Topic::parse(&format!("user:{}", Uuid::new_v4())), Err(ParseTopicErr::Unknown), - "user: without :authz suffix is not a known topic in MVP" + "user: without a known suffix (:authz, :notifications) is not a known topic" + ); + assert_eq!( + Topic::parse(&format!("user:{}:whatever", Uuid::new_v4())), + Err(ParseTopicErr::Unknown), + "an unrecognised suffix rejects — no partial match on the prefix" ); } @@ -563,6 +612,18 @@ mod tests { ); } + #[test] + fn required_perm_user_notifications_is_identity_match() { + // Same strict-privacy gate as :authz — no admin bypass, direct + // UUID equality, anti-enum parity. A regression here would + // let admins snoop on other users' notification streams. + let id = Uuid::new_v4(); + assert_eq!( + Topic::UserNotifications(id).required_perm(), + AuthzCheck::IdentityMatch { user_id: id } + ); + } + #[test] fn event_serializes_with_snake_case_discriminator() { // The `#[serde(tag = "event")]` shape is the WS wire contract for @@ -651,6 +712,14 @@ mod tests { }, "authz_changed", ), + ( + MessageBusEvent::NotificationReceived { + notification_id: Uuid::nil(), + kind: "share_granted".into(), + created_at: chrono::DateTime::::from_timestamp(0, 0).unwrap(), + }, + "notification_received", + ), ( MessageBusEvent::JobRunStarted { name: "backend_migration".into(), diff --git a/src/application/services/mod.rs b/src/application/services/mod.rs index 88710afa..c255f09a 100644 --- a/src/application/services/mod.rs +++ b/src/application/services/mod.rs @@ -25,6 +25,7 @@ pub mod mount_registry; pub mod music_service; pub mod nextcloud_file_id_service; pub mod nextcloud_login_flow_service; +pub mod notification_application_service; pub mod people_service; pub mod places_service; pub mod recent_service; diff --git a/src/application/services/notification_application_service.rs b/src/application/services/notification_application_service.rs new file mode 100644 index 00000000..b7ceaf93 --- /dev/null +++ b/src/application/services/notification_application_service.rs @@ -0,0 +1,127 @@ +//! Orchestrates persistent notifications. +//! +//! `create()` is the single ingester entry point: +//! +//! 1. Insert the row via [`NotificationRepository::create`]. +//! 2. Publish a thin `NotificationReceived` event on +//! `user:{user_id}:notifications` so subscribed sessions refetch +//! immediately. +//! +//! The DB row is the truth (see `docs/plan/message-bus.md § Slice E`). +//! The bus is best-effort — a subscriber offline at publish time +//! recovers on next `GET /api/notifications`. Publish happens AFTER +//! the DB write succeeds, never inside a transaction — the plan's +//! "publish after commit" invariant. +//! +//! Reads (`list_for_user`, `count_unread_for_user`) and state changes +//! (`mark_read`, `mark_all_read`, `delete`) back the REST endpoints in +//! `interfaces/api/handlers/notifications.rs`. Every mutating method +//! is scoped on `user_id` at the SQL layer; the service does not run +//! its own AuthZ check because the identity is by construction +//! (`caller_id == user_id`, extracted from the auth middleware). + +use std::sync::Arc; + +use chrono::Utc; +use uuid::Uuid; + +use crate::application::ports::message_bus_ports::{MessageBus, MessageBusEvent, Topic}; +use crate::common::errors::DomainError; +use crate::domain::entities::notification::{NewNotification, Notification}; +use crate::domain::repositories::notification_repository::{ + NotificationListFilter, NotificationRepository, +}; + +pub struct NotificationApplicationService { + repo: Arc, + bus: Arc, +} + +impl NotificationApplicationService { + pub fn new(repo: Arc, bus: Arc) -> Self { + Self { repo, bus } + } + + /// Insert a row for `new_notif` and publish a thin bus event. + /// Returns the persisted row. This is the ingester-facing method + /// — called from `ShareService::create_grant`, + /// `AuthApplicationService` (new-device login), + /// `SchedulerEngine` (job completed for actor), and the quota + /// threshold hook. + pub async fn create(&self, new_notif: NewNotification) -> Result { + let row = self.repo.create(&new_notif).await?; + + // Publish AFTER the row is durable. Silent no-op if the bus + // is disabled at boot (`OXICLOUD_MESSAGEBUS_ENABLE=false`) — + // the WS route is unmounted so the publish just hits a dead + // sender. The FE bell still works: it reads from the DB on + // mount. See plan § "Slice E". + self.bus.publish( + &Topic::UserNotifications(row.user_id), + MessageBusEvent::NotificationReceived { + notification_id: row.id, + kind: row.kind.clone(), + created_at: row.created_at, + }, + ); + + Ok(row) + } + + /// List notifications for `user_id` newest-first. Default limit at + /// this layer is 50 rows (the repo caps at 500 defensively). + pub async fn list_for_user( + &self, + user_id: Uuid, + filter: NotificationListFilter, + ) -> Result, DomainError> { + self.repo.list_for_user(user_id, &filter).await + } + + /// Unread badge count. + pub async fn count_unread_for_user(&self, user_id: Uuid) -> Result { + self.repo.count_unread_for_user(user_id).await + } + + /// Mark one notification as read. Returns `true` if the row + /// transitioned unread → read (i.e. was owned by `caller_id` and + /// was previously unread). Returns `false` for already-read, + /// missing, or misowned rows — indistinguishable at the wire so + /// enumeration doesn't leak. + pub async fn mark_read( + &self, + notification_id: Uuid, + caller_id: Uuid, + ) -> Result { + self.repo + .mark_read(notification_id, caller_id, Utc::now()) + .await + } + + /// Bulk mark-all-read. Returns rows updated. + pub async fn mark_all_read(&self, caller_id: Uuid) -> Result { + self.repo + .mark_all_read_for_user(caller_id, Utc::now()) + .await + } + + /// Hard-delete one row. Same anti-enumeration semantics as + /// [`mark_read`] — returns `false` for missing / misowned. + pub async fn delete( + &self, + notification_id: Uuid, + caller_id: Uuid, + ) -> Result { + self.repo.delete_by_id(notification_id, caller_id).await + } + + /// Retention job entry point. Called by `notifications_cleanup` + /// on its daily cadence — deletes read rows older than `cutoff`. + /// Unread rows are always preserved. + pub async fn purge_read_before_cutoff( + &self, + cutoff: chrono::DateTime, + ) -> Result { + self.repo.purge_read_before(cutoff).await + } +} diff --git a/src/bin/generate-asyncapi.rs b/src/bin/generate-asyncapi.rs index 52bd67e8..ef3c5f19 100644 --- a/src/bin/generate-asyncapi.rs +++ b/src/bin/generate-asyncapi.rs @@ -141,6 +141,17 @@ fn channels() -> Value { "UnsubscribeRequest": { "$ref": "#/components/messages/RtUnsubscribeRequest" }, } }, + "UserNotifications": { + "address": "user:{userId}:notifications", + "description": "A user's private notifications channel. Identity-scoped: caller_id must equal userId (no admin bypass). Auto-subscribed at session open; the FE bell refetches `GET /api/notifications` when a `notification_received` event fires. The DB row is authoritative — a missed push recovers on the next mount.", + "parameters": { + "userId": { "description": "User UUID — must match the authenticated caller" } + }, + "messages": { + "SubscribeRequest": { "$ref": "#/components/messages/RtSubscribeRequest" }, + "UnsubscribeRequest": { "$ref": "#/components/messages/RtUnsubscribeRequest" }, + } + }, "Job": { "address": "job:{jobName}", "description": "A named background job's run lifecycle — Started / Progress / Ended. Consumed by the admin dashboard so operators who trigger a long-running job (backend migration, thumb import…) can navigate off the admin page and come back without losing progress. AuthZ: admin-only (Class 3 role-scoped) — non-admin gets `topic_forbidden`, indistinguishable on the wire from an unknown topic.", @@ -316,6 +327,7 @@ fn components() -> Value { "FolderRenamedData": folder_renamed_schema(), "FolderMovedData": folder_moved_schema(), "FolderDeletedData": folder_deleted_schema(), + "NotificationReceivedData": notification_received_schema(), "JobRunStartedData": job_run_started_schema(), "JobRunProgressData": job_run_progress_schema(), "JobRunEndedData": job_run_ended_schema(), @@ -598,6 +610,7 @@ fn event_kind_schema() -> Value { "enum": [ "file_created", "file_renamed", "file_moved", "file_deleted", "folder_created", "folder_renamed", "folder_moved", "folder_deleted", + "notification_received", "job_run_started", "job_run_progress", "job_run_ended", ], }) @@ -615,6 +628,7 @@ fn event_data_union_schema() -> Value { ref_schema("FolderRenamedData"), ref_schema("FolderMovedData"), ref_schema("FolderDeletedData"), + ref_schema("NotificationReceivedData"), ref_schema("JobRunStartedData"), ref_schema("JobRunProgressData"), ref_schema("JobRunEndedData"), @@ -732,6 +746,27 @@ fn folder_deleted_schema() -> Value { }) } +// ─────────────────── Notification event payload ────────────────── +// Published on `Topic::UserNotifications(user_id)`. Identity-scoped +// (Class 2) — caller must equal the topic's user_id, no admin +// bypass. Payload is a thin poke: `notification_id` + `kind` + +// `created_at`. The FE bell refetches `GET /api/notifications` on +// receipt for the row's full payload; the DB is the truth, the bus +// event is just an invalidation. + +fn notification_received_schema() -> Value { + json!({ + "type": "object", + "description": "A new notification was created for the caller. Payload is intentionally thin — the FE refetches `GET /api/notifications` for the row's full contents. `kind` is the notification's registered kind slug (`share_granted`, `job_completed_for_you`, `new_login_from_new_device`, `storage_quota_threshold`, …); the FE may use it to route a toast for high-priority kinds but never treats it as authoritative.", + "required": ["notification_id", "kind", "created_at"], + "properties": { + "notification_id": { "type": "string", "format": "uuid" }, + "kind": { "type": "string" }, + "created_at": { "type": "string", "format": "date-time" }, + } + }) +} + // ─────────────────── Job event data payloads ───────────────────── // Published on `Topic::Job(name)`. AuthZ is Class-3 (admin-only) — // non-admins get `topic_forbidden` on subscribe, so these payloads diff --git a/src/common/config.rs b/src/common/config.rs index c6b88781..c094e294 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -2311,6 +2311,17 @@ pub struct FeaturesConfig { /// Enabled by default: expired-auth-row cleanup is a /// security-hygiene default, not opt-in. pub grant_cleanup: GrantCleanupConfig, + + /// Retention window (in days) for read notification rows — + /// `notif.notifications` with `read_at IS NOT NULL`. Unread rows + /// are preserved unconditionally; the `notifications_cleanup` + /// scheduled job deletes read rows older than this on a daily + /// cadence. + /// + /// Env: `OXICLOUD_NOTIFICATIONS_RETENTION_DAYS` (default `30`). + /// Minimum 1 (0 would delete every read row on every tick — the + /// service clamps defensively). + pub notifications_retention_days: u32, } /// Config for the daily expired-grant purge (see @@ -2498,6 +2509,7 @@ impl Default for FeaturesConfig { webdav_drive_listing_prefix: "@drive".to_string(), enable_message_bus: true, // Message bus (WS + ticket) on by default grant_cleanup: GrantCleanupConfig::default(), + notifications_retention_days: 30, // 30 days is the plan's default } } } @@ -3377,6 +3389,15 @@ impl AppConfig { config.features.enable_message_bus = val; } + // Slice E — notification retention. Read as u32 so a + // non-numeric or negative value falls back to the declared + // default (30 days) rather than crashing at boot. + if let Ok(raw) = env::var("OXICLOUD_NOTIFICATIONS_RETENTION_DAYS") + && let Ok(val) = raw.parse::() + { + config.features.notifications_retention_days = val.max(1); + } + if let Ok(enable_search) = env::var("OXICLOUD_ENABLE_SEARCH").map(|v| v.parse::()) && let Ok(val) = enable_search { diff --git a/src/common/di.rs b/src/common/di.rs index 0efe2a65..f2cff698 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -2418,6 +2418,7 @@ impl AppServiceFactory { mock_email_sender: None, // populated below magic_link_invite_service: None, // populated below recipient_notification_service: None, // populated below alongside magic_link_invite_service + notification_service: None, // populated below (Slice E) // Per-caller limits, configurable since the hardcoded ceilings // had no escape hatch for deployments where several actors share // one identity — a CI suite running as a single `admin` shares @@ -2536,6 +2537,42 @@ impl AppServiceFactory { ), )); } + + // Persistent in-app notifications (Slice E). Repo + bus + // are both always available when auth is on; the service + // wraps them into the ingester-facing `create()` + + // bell-facing reads. Always wired under `auth_service` — + // notifications are per-user and require an authenticated + // caller everywhere they surface. + let notif_repo: Arc< + dyn crate::domain::repositories::notification_repository::NotificationRepository, + > = Arc::new( + crate::infrastructure::repositories::pg::NotificationPgRepository::new( + pool.clone(), + ), + ); + let notif_bus: Arc = + app_state.bus.clone(); + let notification_service = Arc::new( + crate::application::services::notification_application_service::NotificationApplicationService::new( + notif_repo, + notif_bus, + ), + ); + app_state.notification_service = Some(notification_service.clone()); + + // Retention sweep — daily; deletes read notifications + // older than OXICLOUD_NOTIFICATIONS_RETENTION_DAYS. Same + // self-registering pattern as `trash_cleanup`. + let retention_days = app_state.core.config.features.notifications_retention_days; + let _ = Arc::new( + crate::infrastructure::services::notifications_cleanup_service::NotificationsCleanupService::new( + notification_service, + retention_days, + ), + ) + .register(&app_state.core.job_registry) + .await; } // 9b. Wire admin settings service when auth is available @@ -3441,6 +3478,16 @@ pub struct AppState { pub recipient_notification_service: Option< Arc, >, + /// Persistent in-app notifications — bell UI, retention job, four + /// initial ingesters (share-granted, new-login-from-new-device, + /// job-completed-for-you, storage-quota-threshold). Always + /// populated when auth is enabled (bell requires an authenticated + /// caller). Wraps a PG repo + the message bus; `create()` writes + /// the row AND publishes on `user:{u}:notifications` in one call. + /// See `docs/plan/message-bus.md § Slice E`. + pub notification_service: Option< + Arc, + >, /// Per-caller sliding-window limiter for `GET /api/users/{id}`. The /// endpoint's primary defense is the visibility check, but a stale /// JWT could in theory iterate UUIDs against the related-by-grant diff --git a/src/domain/entities/mod.rs b/src/domain/entities/mod.rs index a475accb..2a7cdffe 100644 --- a/src/domain/entities/mod.rs +++ b/src/domain/entities/mod.rs @@ -9,6 +9,7 @@ pub mod face; pub mod file; pub mod folder; pub mod magic_link_token; +pub mod notification; pub mod playlist; pub mod session; pub mod share; diff --git a/src/domain/entities/notification.rs b/src/domain/entities/notification.rs new file mode 100644 index 00000000..a251f971 --- /dev/null +++ b/src/domain/entities/notification.rs @@ -0,0 +1,70 @@ +//! In-app notification — one durable row per recipient per event. +//! +//! Backs the bell UI. The message bus poke on +//! `user:{user_id}:notifications` is a fast path; the row is truth. +//! See `docs/plan/message-bus.md § Slice E` for the wire contract. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +/// A stable kind slug. The FE routes on this string for icon / label / +/// action-button choice. New kinds are additive; **never repurpose an +/// existing value** — the FE reads it as an enum-like discriminant. +/// +/// The initial set matches the plan's Slice-E ingester list. Additional +/// values are legal on the wire (an older FE ignores unknown kinds +/// gracefully by falling back to a generic bell row); we still keep the +/// canonical list here so the ingester callsites reach for symbolic +/// constants instead of literal strings. +/// +/// The DB column is plain `TEXT` (see `migrations/20261026000000_notifications.sql`) +/// — no CHECK constraint. Adding a new kind is a code change only, no +/// migration, no downtime. +pub mod kind { + /// A grant was created for the recipient user (they can now access + /// a resource). Payload carries the resource id + role + granter. + pub const SHARE_GRANTED: &str = "share_granted"; + + /// A login succeeded from a device / IP fingerprint the user + /// hasn't seen before. Payload carries the user-agent snippet + /// and the coarsened location if available. + pub const NEW_LOGIN_FROM_NEW_DEVICE: &str = "new_login_from_new_device"; + + /// A background job triggered by the recipient user finished + /// (success or failure). Payload carries the job name and + /// `success: bool`. Clicking navigates to `/admin/jobs/`. + pub const JOB_COMPLETED_FOR_YOU: &str = "job_completed_for_you"; + + /// The recipient's storage quota crossed a warning threshold + /// (e.g. 80 %, 95 %). Payload carries `used_bytes` / `quota_bytes` + /// and the crossed percentage. + pub const STORAGE_QUOTA_THRESHOLD: &str = "storage_quota_threshold"; +} + +/// One notification row. +/// +/// `payload` is a per-kind opaque JSON blob; the DB stays schema-free +/// so a new field never requires a migration. Callers deserialize it +/// against a kind-specific struct on the FE. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Notification { + pub id: Uuid, + pub user_id: Uuid, + pub kind: String, + pub payload: serde_json::Value, + pub created_at: DateTime, + /// `None` = unread; `Some(t)` = when the user explicitly marked it + /// read via `POST /api/notifications/{id}/read` or + /// `POST /api/notifications/read-all`. + pub read_at: Option>, +} + +/// The service-layer input for [`NotificationService::create`]. Split +/// from [`Notification`] because `id` / `created_at` are DB-generated. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NewNotification { + pub user_id: Uuid, + pub kind: String, + pub payload: serde_json::Value, +} diff --git a/src/domain/repositories/mod.rs b/src/domain/repositories/mod.rs index 51a36364..97e414e3 100644 --- a/src/domain/repositories/mod.rs +++ b/src/domain/repositories/mod.rs @@ -6,6 +6,7 @@ pub mod drive_repository; pub mod file_repository; pub mod folder_repository; pub mod magic_link_token_repository; +pub mod notification_repository; pub mod playlist_repository; pub mod session_repository; pub mod settings_repository; diff --git a/src/domain/repositories/notification_repository.rs b/src/domain/repositories/notification_repository.rs new file mode 100644 index 00000000..6055711a --- /dev/null +++ b/src/domain/repositories/notification_repository.rs @@ -0,0 +1,90 @@ +//! Storage port for [`Notification`]. +//! +//! Backs the bell UI. `create` is the only ingester-facing method; +//! `list_for_user` / `mark_read` / `mark_all_read` / `delete_by_id` / +//! `purge_read_before` back the REST endpoints and the retention job. +//! +//! Every method takes `user_id` where relevant so the SQL includes the +//! caller-scope in its WHERE clause — the application service double- +//! checks the requested notification's owner matches the caller, but +//! the repo scoping is defense in depth (a bug that misroutes an id +//! still can't leak another user's row through `mark_read`). + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use uuid::Uuid; + +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". +#[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. + 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. + pub before: Option>, +} + +#[async_trait] +pub trait NotificationRepository: Send + Sync + 'static { + /// Insert a new notification. Returns the persisted row (id + + /// created_at populated). The application service publishes the + /// bus event AFTER this returns Ok — see plan's "publish after + /// commit" invariant. + async fn create(&self, new_notif: &NewNotification) -> Result; + + /// List notifications for `user_id` newest-first, honouring + /// `filter`. Returns an empty Vec (not an error) when the user + /// has none. + async fn list_for_user( + &self, + user_id: Uuid, + filter: &NotificationListFilter, + ) -> Result, DomainError>; + + /// Count unread rows for `user_id`. Backs the bell's unread badge. + /// Separate from `list_for_user` so the badge can render without + /// fetching payloads. + async fn count_unread_for_user(&self, user_id: Uuid) -> Result; + + /// Mark one notification as read. Returns `Ok(true)` if a row + /// transitioned from unread → read (i.e. was owned by `user_id` + /// AND had `read_at IS NULL`); `Ok(false)` if the row didn't + /// exist, was owned by someone else, or was already read. + /// Idempotent from the caller's perspective; the `bool` is for + /// logs / audit only. + async fn mark_read( + &self, + notification_id: Uuid, + user_id: Uuid, + at: DateTime, + ) -> Result; + + /// Bulk mark-all-read. Returns the number of rows updated. + async fn mark_all_read_for_user( + &self, + user_id: Uuid, + at: DateTime, + ) -> Result; + + /// Hard-delete a single row. Same ownership scoping as + /// [`mark_read`]. Returns `Ok(true)` iff a row was deleted. + async fn delete_by_id(&self, notification_id: Uuid, user_id: Uuid) + -> Result; + + /// Retention job: delete every read row whose `read_at` is older + /// than `cutoff`. Returns the number of rows removed. + /// Unread rows are preserved unconditionally — that's the whole + /// point of the durable table. + async fn purge_read_before(&self, cutoff: DateTime) -> Result; +} diff --git a/src/infrastructure/repositories/pg/mod.rs b/src/infrastructure/repositories/pg/mod.rs index 99483610..857bf3d9 100644 --- a/src/infrastructure/repositories/pg/mod.rs +++ b/src/infrastructure/repositories/pg/mod.rs @@ -14,6 +14,7 @@ mod favorites_pg_repository; pub mod file_metadata_repository; mod magic_link_token_pg_repository; mod nextcloud_object_id_repository; +mod notification_pg_repository; mod opaque_pg_repository; pub mod playlist_pg_repository; mod recent_items_pg_repository; @@ -48,6 +49,7 @@ pub use file_metadata_repository::FileMetadataRepository; pub use folder_db_repository::FolderDbRepository; pub use magic_link_token_pg_repository::MagicLinkTokenPgRepository; pub use nextcloud_object_id_repository::NextcloudObjectIdRepository; +pub use notification_pg_repository::NotificationPgRepository; pub use opaque_pg_repository::OpaquePgRepository; pub use playlist_pg_repository::{ AudioMetadataPgRepository, PlaylistItemPgRepository, PlaylistPgRepository, diff --git a/src/infrastructure/repositories/pg/notification_pg_repository.rs b/src/infrastructure/repositories/pg/notification_pg_repository.rs new file mode 100644 index 00000000..ee219496 --- /dev/null +++ b/src/infrastructure/repositories/pg/notification_pg_repository.rs @@ -0,0 +1,281 @@ +//! PostgreSQL implementation of [`NotificationRepository`]. +//! +//! Backs the bell UI plus the daily retention job. All queries scope on +//! `user_id` at the SQL layer so a row misroute in the caller can't +//! leak another user's data through mark_read / delete. Schema lives +//! in `migrations/20261026000000_notifications.sql`. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use sqlx::{PgPool, Row}; +use std::sync::Arc; +use uuid::Uuid; + +use crate::common::errors::{DomainError, ErrorKind}; +use crate::domain::entities::notification::{NewNotification, Notification}; +use crate::domain::repositories::notification_repository::{ + NotificationListFilter, NotificationRepository, +}; + +pub struct NotificationPgRepository { + pool: Arc, +} + +impl NotificationPgRepository { + pub fn new(pool: Arc) -> Self { + Self { pool } + } + + fn map_row(row: &sqlx::postgres::PgRow) -> Result { + let map_err = |field: &str, e: sqlx::Error| { + DomainError::new( + ErrorKind::DatabaseError, + "Notification", + format!("read {field}: {e}"), + ) + }; + Ok(Notification { + id: row.try_get("id").map_err(|e| map_err("id", e))?, + user_id: row.try_get("user_id").map_err(|e| map_err("user_id", e))?, + kind: row.try_get("kind").map_err(|e| map_err("kind", e))?, + payload: row.try_get("payload").map_err(|e| map_err("payload", e))?, + created_at: row + .try_get("created_at") + .map_err(|e| map_err("created_at", e))?, + read_at: row.try_get("read_at").ok(), + }) + } +} + +fn db_err(op: &'static str, e: sqlx::Error) -> DomainError { + DomainError::new( + ErrorKind::DatabaseError, + "Notification", + format!("{op}: {e}"), + ) +} + +#[async_trait] +impl NotificationRepository for NotificationPgRepository { + async fn create(&self, new_notif: &NewNotification) -> Result { + let row = sqlx::query( + r#" + INSERT INTO notif.notifications (user_id, kind, payload) + VALUES ($1::uuid, $2, $3) + RETURNING id, user_id, kind, payload, created_at, read_at + "#, + ) + .bind(new_notif.user_id) + .bind(&new_notif.kind) + .bind(&new_notif.payload) + .fetch_one(self.pool.as_ref()) + .await + .map_err(|e| db_err("create", e))?; + Self::map_row(&row) + } + + async fn list_for_user( + &self, + 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). + 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 + } + } + .map_err(|e| db_err("list_for_user", e))?; + + rows.iter().map(Self::map_row).collect() + } + + async fn count_unread_for_user(&self, user_id: Uuid) -> Result { + let row = sqlx::query( + r#" + SELECT COUNT(*)::bigint AS c + FROM notif.notifications + WHERE user_id = $1::uuid AND read_at IS NULL + "#, + ) + .bind(user_id) + .fetch_one(self.pool.as_ref()) + .await + .map_err(|e| db_err("count_unread_for_user", e))?; + row.try_get::("c") + .map_err(|e| db_err("count_unread_for_user.map", e)) + } + + async fn mark_read( + &self, + notification_id: Uuid, + user_id: Uuid, + at: DateTime, + ) -> Result { + // Guard on read_at IS NULL so a re-issued call from a client + // that's already ack'd the row is a no-op instead of stamping + // a later timestamp over the earlier one. + let res = sqlx::query( + r#" + UPDATE notif.notifications + SET read_at = $3 + WHERE id = $1::uuid + AND user_id = $2::uuid + AND read_at IS NULL + "#, + ) + .bind(notification_id) + .bind(user_id) + .bind(at) + .execute(self.pool.as_ref()) + .await + .map_err(|e| db_err("mark_read", e))?; + Ok(res.rows_affected() == 1) + } + + async fn mark_all_read_for_user( + &self, + user_id: Uuid, + at: DateTime, + ) -> Result { + let res = sqlx::query( + r#" + UPDATE notif.notifications + SET read_at = $2 + WHERE user_id = $1::uuid AND read_at IS NULL + "#, + ) + .bind(user_id) + .bind(at) + .execute(self.pool.as_ref()) + .await + .map_err(|e| db_err("mark_all_read_for_user", e))?; + Ok(res.rows_affected()) + } + + async fn delete_by_id( + &self, + notification_id: Uuid, + user_id: Uuid, + ) -> Result { + let res = sqlx::query( + r#" + DELETE FROM notif.notifications + WHERE id = $1::uuid AND user_id = $2::uuid + "#, + ) + .bind(notification_id) + .bind(user_id) + .execute(self.pool.as_ref()) + .await + .map_err(|e| db_err("delete_by_id", e))?; + Ok(res.rows_affected() == 1) + } + + async fn purge_read_before(&self, cutoff: DateTime) -> Result { + let res = sqlx::query( + r#" + DELETE FROM notif.notifications + WHERE read_at IS NOT NULL AND read_at < $1 + "#, + ) + .bind(cutoff) + .execute(self.pool.as_ref()) + .await + .map_err(|e| db_err("purge_read_before", e))?; + Ok(res.rows_affected()) + } +} diff --git a/src/infrastructure/services/in_process_message_bus.rs b/src/infrastructure/services/in_process_message_bus.rs index b97fb8fc..a5646d73 100644 --- a/src/infrastructure/services/in_process_message_bus.rs +++ b/src/infrastructure/services/in_process_message_bus.rs @@ -199,6 +199,7 @@ fn event_kind(event: &MessageBusEvent) -> &'static str { MessageBusEvent::FolderMoved { .. } => "folder_moved", MessageBusEvent::FolderDeleted { .. } => "folder_deleted", MessageBusEvent::AuthzChanged { .. } => "authz_changed", + MessageBusEvent::NotificationReceived { .. } => "notification_received", MessageBusEvent::JobRunStarted { .. } => "job_run_started", MessageBusEvent::JobRunProgress { .. } => "job_run_progress", MessageBusEvent::JobRunEnded { .. } => "job_run_ended", diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index e3ed53c9..64a60d55 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -39,6 +39,7 @@ pub mod mock_email_sender; pub mod mount_provider_factory; pub mod nextcloud_chunked_upload_service; pub mod noop_face_analyzer; +pub mod notifications_cleanup_service; pub mod oidc_service; #[cfg(feature = "faces-onnx")] pub mod onnx_face_analyzer; diff --git a/src/infrastructure/services/notifications_cleanup_service.rs b/src/infrastructure/services/notifications_cleanup_service.rs new file mode 100644 index 00000000..7153006a --- /dev/null +++ b/src/infrastructure/services/notifications_cleanup_service.rs @@ -0,0 +1,136 @@ +//! `notifications_cleanup` scheduled job — daily retention sweep. +//! +//! Deletes rows from `notif.notifications` where `read_at IS NOT NULL` +//! and older than the retention window. Unread rows are preserved +//! unconditionally (the whole point of the durable table is that a +//! user offline for a month still sees the share-granted notice on +//! next login). +//! +//! Retention window comes from `OXICLOUD_NOTIFICATIONS_RETENTION_DAYS` +//! (default 30), applied at job dispatch — one env var maps to one +//! `retention_days` parameter so an operator can override the default +//! at trigger time without a redeploy. + +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use chrono::Utc; +use tracing::info; + +use crate::application::services::notification_application_service::NotificationApplicationService; +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRegistry, JobRunArgs, Mutates}; + +/// Parameter declaration table. Kept at module scope so +/// `JobHandler::parameters` can return a `'static` slice without +/// stack-allocating each call. +static PARAMETERS: [crate::infrastructure::scheduler::JobParam; 1] = + [crate::infrastructure::scheduler::JobParam::number( + "retention_days", + 30, + "Delete read notifications older than this many days.", + )]; + +pub struct NotificationsCleanupService { + service: Arc, + /// Default retention window in days when the trigger call did NOT + /// supply an explicit `retention_days` parameter. Read from + /// `OXICLOUD_NOTIFICATIONS_RETENTION_DAYS` at boot; the constructor + /// clamps to a minimum of 1 day (0 would purge every read row on + /// every tick). + default_retention_days: i64, +} + +impl NotificationsCleanupService { + pub const JOB_NAME: &'static str = "notifications_cleanup"; + + pub fn new(service: Arc, default_retention_days: u32) -> Self { + Self { + service, + default_retention_days: default_retention_days.max(1) as i64, + } + } + + /// Interval — daily. Same tier as `trash_cleanup`; retention is a + /// "days" concept, so a finer cadence buys nothing. + fn interval() -> Duration { + Duration::from_secs(24 * 3600) + } + + /// Register self with the scheduler. Chained DI helper, same shape + /// as [`TrashCleanupService::register`]. + pub async fn register(self: Arc, registry: &JobRegistry) -> Arc { + registry + .register(self.clone(), Some(Self::interval()), None) + .await; + self + } +} + +#[async_trait] +impl JobHandler for NotificationsCleanupService { + fn name(&self) -> &str { + Self::JOB_NAME + } + + fn description(&self) -> &'static str { + "Deletes read notifications older than the retention window \ + (default 30 days, override via `retention_days` parameter or \ + OXICLOUD_NOTIFICATIONS_RETENTION_DAYS). Unread rows are \ + preserved unconditionally." + } + + fn mutates(&self) -> Mutates { + Mutates::Always + } + + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + // Declared default of 30 days is the SAME literal the config + // block's env fallback uses (`OXICLOUD_NOTIFICATIONS_RETENTION_DAYS` + // default), so an operator who never sets the env sees 30 + // everywhere. The env-derived `default_retention_days` on + // this struct only diverges from 30 when the operator DID + // set the env — see the guard in `run()` below. + &PARAMETERS + } + + async fn run(&self, args: &JobRunArgs) -> JobOutcome { + // `get_number` returns the fallback ONLY when the arg is + // absent — but declared defaults are seeded by the engine + // before `run` runs (see JobRunArgs::normalized_for), so the + // param is always present with either the caller's value or + // the declared 30. We treat "declared default AND env + // override differs" as "use env override" to keep the + // OXICLOUD_NOTIFICATIONS_RETENTION_DAYS knob effective + // without teaching the engine per-instance defaults. + let declared_default = 30_i64; + let raw = args.get_number("retention_days", declared_default); + let retention_days = if raw == declared_default { + self.default_retention_days + } else { + raw + } + .max(1); + let cutoff = Utc::now() - chrono::Duration::days(retention_days); + + match self.service.purge_read_before_cutoff(cutoff).await { + Ok(removed) => { + info!( + target: "audit", + event = "notifications.retention_sweep", + retention_days, + removed, + "🧹 notifications retention sweep: {removed} row(s) purged (retention {retention_days} d)" + ); + JobOutcome::ok_with( + removed, + serde_json::json!({ + "retention_days": retention_days, + "removed": removed, + }), + ) + } + Err(e) => JobOutcome::err(format!("notifications cleanup failed: {e}")), + } + } +} diff --git a/src/interfaces/api/handlers/grant_handler.rs b/src/interfaces/api/handlers/grant_handler.rs index b302fc62..7c053754 100644 --- a/src/interfaces/api/handlers/grant_handler.rs +++ b/src/interfaces/api/handlers/grant_handler.rs @@ -330,6 +330,71 @@ pub async fn create_grant( "🤝 grant created with role '{}'", role.as_str(), ); + // Slice E — persistent in-app notification (bell) for every + // recipient user. Separate channel from the email path below: + // the DB row is authoritative and survives SMTP being down / + // the recipient not having email, and it powers the FE bell + + // unread badge. + // + // Fan out to the resolved user ids: + // - Subject::User(id) → one row for that user + // - Subject::Group(id) → one row per transitive member (uses + // subject_group_service if wired; groups + // without a service configured skip the + // bell but still get email via the + // recipient service below) + // - Subject::Token(_) → no bell row (anonymous share link, no + // target user to route it to) + // + // Every failure here is best-effort — a row-write hiccup logs a + // warn and continues to the email path. The grant row is already + // durable in `role_grants`; the recipient can still discover the + // share via the resources-shared-with-me listing. + if let Some(notif_svc) = state.notification_service.as_ref() { + let recipient_ids: Vec = match subject { + Subject::User(id) => vec![id], + Subject::Group(group_id) => match state.subject_group_service.as_ref() { + Some(sgs) => sgs + .list_transitive_users(group_id) + .await + .unwrap_or_else(|e| { + warn!("group {group_id} member expansion failed; skipping bell: {e}"); + Vec::new() + }), + None => Vec::new(), + }, + Subject::Token(_) => Vec::new(), + }; + for rid in recipient_ids { + // Self-shares (owner grants themselves via a group they + // are also in) would fire a bell on the owner — filter + // that out here. Every other filter (opt-out flag, etc.) + // is deferred; in-app notifications are less intrusive + // than SMTP so the ceremony is lighter. + if rid == caller_id { + continue; + } + let payload = serde_json::json!({ + "granter_id": caller_id, + "resource_type": resource.type_str(), + "resource_id": resource.id(), + "role": role.as_str(), + "expires_at": expires_at, + }); + let new_notif = crate::domain::entities::notification::NewNotification { + user_id: rid, + kind: crate::domain::entities::notification::kind::SHARE_GRANTED.to_string(), + payload, + }; + if let Err(e) = notif_svc.create(new_notif).await { + warn!( + "notification.create failed for share_granted (recipient={rid}, resource={:?}): {e}", + resource + ); + } + } + } + // PR N1 — route the post-grant notification through the unified // RecipientNotificationService. Handles user/group/token subjects // uniformly (Token subjects return an empty outcome set); applies diff --git a/src/interfaces/api/handlers/mod.rs b/src/interfaces/api/handlers/mod.rs index c3b3e7b0..875754a1 100644 --- a/src/interfaces/api/handlers/mod.rs +++ b/src/interfaces/api/handlers/mod.rs @@ -20,6 +20,7 @@ pub mod grant_handler; pub mod i18n_handler; pub mod magic_link_handler; pub mod music_handler; +pub mod notifications_handler; pub mod opaque_auth_handler; pub mod people_handler; pub mod photos_handler; diff --git a/src/interfaces/api/handlers/notifications_handler.rs b/src/interfaces/api/handlers/notifications_handler.rs new file mode 100644 index 00000000..2a9c02b7 --- /dev/null +++ b/src/interfaces/api/handlers/notifications_handler.rs @@ -0,0 +1,222 @@ +//! `/api/notifications/*` — the bell UI's REST surface. +//! +//! Five endpoints back the FE `NotificationBell`: +//! +//! - `GET /api/notifications` — list newest-first; optional +//! `unread=true` filter, `before` cursor, `limit` cap. +//! - `GET /api/notifications/unread` — badge-only fast path (count). +//! - `POST /api/notifications/{id}/read` — mark one as read. +//! - `POST /api/notifications/read-all` — bulk mark-all-read. +//! - `DELETE /api/notifications/{id}` — hard-delete one row. +//! +//! Every endpoint scopes on `auth_user.id` at the SQL layer via the +//! application service, so an id enumeration against +//! `POST /api/notifications/{id}/read` returns the same 204 whether +//! the row exists-and-belongs-to-somebody-else, or doesn't exist at +//! all. Anti-enumeration is the reason the response body doesn't +//! distinguish "already read" from "not yours" — the service returns +//! a bool for our logs, we always return 204 to the wire. + +use std::sync::Arc; + +use axum::{ + Json, + extract::{Path, Query, State}, + http::StatusCode, + response::IntoResponse, +}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; +use uuid::Uuid; + +use crate::application::services::notification_application_service::NotificationApplicationService; +use crate::domain::entities::notification::Notification; +use crate::domain::repositories::notification_repository::NotificationListFilter; +use crate::interfaces::errors::AppError; +use crate::interfaces::middleware::auth::AuthUser; + +/// Wire shape for one notification row. `payload` stays a raw JSON +/// value — per-kind decoding happens on the FE using the `kind` +/// discriminant. +#[derive(Debug, Serialize, ToSchema)] +pub struct NotificationDto { + pub id: Uuid, + pub kind: String, + #[schema(value_type = Object)] + pub payload: serde_json::Value, + pub created_at: DateTime, + /// `null` = unread. + pub read_at: Option>, +} + +impl From for NotificationDto { + fn from(n: Notification) -> Self { + Self { + id: n.id, + kind: n.kind, + payload: n.payload, + created_at: n.created_at, + read_at: n.read_at, + } + } +} + +/// Query params for `GET /api/notifications`. +#[derive(Debug, Deserialize, ToSchema)] +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. + pub before: Option>, + /// Max rows returned. Server-side clamp at 500. + pub limit: Option, +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct ListResponseDto { + pub items: Vec, + /// Unread rows for this user across the whole table — the bell + /// badge reads this. Kept on the list response so a bell open + /// doesn't need a second round-trip for the badge. + pub unread_count: i64, +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct UnreadCountDto { + pub unread_count: i64, +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct MarkAllReadResponseDto { + /// Number of rows that transitioned unread → read. + pub marked: u64, +} + +/// GET /api/notifications +#[utoipa::path( + get, + path = "/api/notifications", + params( + ("unread" = Option, Query, description = "Only return unread rows"), + ("before" = Option>, Query, description = "Cursor — rows strictly before this created_at"), + ("limit" = Option, Query, description = "Max rows (server-side clamp at 500)"), + ), + responses( + (status = 200, description = "List of notifications", body = ListResponseDto), + ), + security(("bearerAuth" = [])), + tag = "notifications" +)] +pub async fn list_notifications( + State(service): State>, + auth_user: AuthUser, + Query(query): Query, +) -> Result, AppError> { + let filter = NotificationListFilter { + limit: query.limit, + unread_only: if query.unread { Some(true) } else { None }, + before: query.before, + }; + let rows = service.list_for_user(auth_user.id, filter).await?; + let unread_count = service.count_unread_for_user(auth_user.id).await?; + Ok(Json(ListResponseDto { + items: rows.into_iter().map(NotificationDto::from).collect(), + unread_count, + })) +} + +/// GET /api/notifications/unread — badge-only fast path. +#[utoipa::path( + get, + path = "/api/notifications/unread", + responses( + (status = 200, description = "Unread count", body = UnreadCountDto), + ), + security(("bearerAuth" = [])), + tag = "notifications" +)] +pub async fn unread_count( + State(service): State>, + auth_user: AuthUser, +) -> Result, AppError> { + let unread_count = service.count_unread_for_user(auth_user.id).await?; + Ok(Json(UnreadCountDto { unread_count })) +} + +/// POST /api/notifications/{id}/read — mark one as read. +/// +/// Always responds 204 regardless of whether the row existed and +/// belonged to the caller — the service's `bool` return is logged +/// (audit reason `notification.marked_read` on success), never +/// surfaced to the wire. +#[utoipa::path( + post, + path = "/api/notifications/{id}/read", + params(("id" = Uuid, Path, description = "Notification id")), + responses((status = 204, description = "Marked read (idempotent, anti-enum)")), + security(("bearerAuth" = [])), + tag = "notifications" +)] +pub async fn mark_read( + State(service): State>, + auth_user: AuthUser, + Path(id): Path, +) -> Result { + let transitioned = service.mark_read(id, auth_user.id).await?; + if transitioned { + tracing::debug!( + target: "oxicloud::notifications", + caller_id = %auth_user.id, + notification_id = %id, + "notification marked read" + ); + } + Ok(StatusCode::NO_CONTENT) +} + +/// POST /api/notifications/read-all — bulk mark-all-read. +#[utoipa::path( + post, + path = "/api/notifications/read-all", + responses((status = 200, description = "Rows marked", body = MarkAllReadResponseDto)), + security(("bearerAuth" = [])), + tag = "notifications" +)] +pub async fn mark_all_read( + State(service): State>, + auth_user: AuthUser, +) -> Result, AppError> { + let marked = service.mark_all_read(auth_user.id).await?; + Ok(Json(MarkAllReadResponseDto { marked })) +} + +/// DELETE /api/notifications/{id} — hard-delete one row. +/// +/// Same anti-enum semantics as `mark_read` — always 204. +#[utoipa::path( + delete, + path = "/api/notifications/{id}", + params(("id" = Uuid, Path, description = "Notification id")), + responses((status = 204, description = "Deleted (idempotent, anti-enum)")), + security(("bearerAuth" = [])), + tag = "notifications" +)] +pub async fn delete_notification( + State(service): State>, + auth_user: AuthUser, + Path(id): Path, +) -> Result { + let deleted = service.delete(id, auth_user.id).await?; + if deleted { + tracing::debug!( + target: "oxicloud::notifications", + caller_id = %auth_user.id, + notification_id = %id, + "notification deleted" + ); + } + Ok(StatusCode::NO_CONTENT) +} diff --git a/src/interfaces/api/handlers/rt_ws.rs b/src/interfaces/api/handlers/rt_ws.rs index 8608b155..a56f1d66 100644 --- a/src/interfaces/api/handlers/rt_ws.rs +++ b/src/interfaces/api/handlers/rt_ws.rs @@ -401,6 +401,20 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc) /// loop then walks the sub set and drops matching topics. Any other /// event kind on this topic is ignored (defensive; shouldn't happen /// in MVP). +/// - For `Topic::UserNotifications(_)`: an incoming +/// `MessageBusEvent::NotificationReceived` is forwarded through the +/// default path — the FE bell listens for `rt.event` on the +/// auto-subscribed identity topic and refetches `GET +/// /api/notifications` when it sees one. Same anti-enumeration +/// invariant as `:authz` (identity-scoped, no admin bypass). /// - For every other topic: bus events are wrapped into a client- /// visible `rt.event` notification and pushed as `SessionOut::Frame`. fn install_subscription( diff --git a/src/interfaces/api/routes.rs b/src/interfaces/api/routes.rs index 84a4c488..d402460c 100644 --- a/src/interfaces/api/routes.rs +++ b/src/interfaces/api/routes.rs @@ -195,6 +195,7 @@ pub fn create_api_routes(app_state: &Arc) -> Router> { let share_service = app_state.share_service.clone(); let favorites_service = app_state.favorites_service.clone(); let recent_service = app_state.recent_service.clone(); + let notification_service = app_state.notification_service.clone(); // authorization is no longer extracted separately — the grants router now // uses app_state directly so handlers can access all services. @@ -409,6 +410,25 @@ pub fn create_api_routes(app_state: &Arc) -> Router> { Router::new() }; + // Notifications bell (Slice E). Mounted only when the service is + // wired (i.e. auth is enabled — bell requires a caller). Non- + // registration path: with the flag off, the routes 404 instead of + // 5xx-ing on a NULL service — matches the OXICLOUD_MESSAGEBUS_ENABLE + // approach for `/api/rt/*` and `OXICLOUD_ENABLE_EXTERNAL_MOUNTS` + // for admin mounts. + let notifications_router = if let Some(ref svc) = notification_service { + use crate::interfaces::api::handlers::notifications_handler; + Router::new() + .route("/", get(notifications_handler::list_notifications)) + .route("/unread", get(notifications_handler::unread_count)) + .route("/read-all", post(notifications_handler::mark_all_read)) + .route("/{id}/read", post(notifications_handler::mark_read)) + .route("/{id}", delete(notifications_handler::delete_notification)) + .with_state(svc.clone()) + } else { + Router::new() + }; + // Create routes for chunked uploads (large files >10MB). // All five handlers are free functions — see chunked_upload_handler.rs for why // #[utoipa::path] cannot be applied to ChunkedUploadHandler impl methods directly. @@ -455,7 +475,8 @@ pub fn create_api_routes(app_state: &Arc) -> Router> { .nest("/shares", share_router) .nest("/grants", grants_router) .nest("/favorites", favorites_router) - .nest("/recent", recent_router); + .nest("/recent", recent_router) + .nest("/notifications", notifications_router); // Photos timeline endpoint — lists all image/video files sorted by capture date { diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh index 0ecd0122..adb68eac 100755 --- a/tests/api/rt_bus_check.sh +++ b/tests/api/rt_bus_check.sh @@ -64,6 +64,39 @@ # unit tests; the seeded suite has no # admin token, and minting one here # would pollute state for other files. +# S13 Notification wire push — user2 subscribes to +# `user:{user2_id}:notifications`; user1 +# creates a grant that targets user2. +# Server must publish one `rt.event` +# with `event="notification_received"`, +# `data.kind="share_granted"`. Guards +# the Slice-E ingester + the auto-sub +# delivery path together — this is the +# only scenario that exercises the +# wire push from the `NotificationService`. +# S14 Notification DB row — after S13's grant, GET +# `/api/notifications` as user2 lists +# at least one row with +# `kind="share_granted"` whose payload +# references the freshly-shared folder, +# and `unread_count >= 1`. Guards the +# authoritative side of the pattern — +# a subscriber offline at publish time +# recovers via this endpoint. +# S15 Cross-user notif deny — user1 subscribes to +# `user:{user2_id}:notifications` +# (an identity-scoped topic that +# resolves to somebody else). Server +# must reject with `topic_forbidden` +# (same wire shape as unknown topic, +# same rule as S9 for `:authz`). +# Guards the strict-privacy Class-2 +# AuthZ gate on Slice-E notifications +# — no admin bypass, direct UUID +# equality only. If this ever accepts +# and delivers events, an admin (or +# anyone else) could snoop on other +# users' notification streams. # # Exit non-zero on any failure — run.sh treats that as a suite failure. # ───────────────────────────────────────────────────────────────────────────── @@ -679,4 +712,111 @@ if ! "$HELPER_BIN" expect-denied \ fi log "S12 OK" -log "All twelve message-bus scenarios passed." +# ── Scenario 13 — Notification wire push (Slice E) ────────────────────────── +# The `share_granted` ingester runs in `grant_handler::create_grant`: +# after `set_role` lands and before the email path, it calls +# `NotificationService::create` for every resolved recipient user. +# `create` writes the DB row AND publishes a thin +# `MessageBusEvent::NotificationReceived` on +# `user:{user_id}:notifications`. This scenario exercises the wire +# path end-to-end: user2 opens a WS + explicitly subscribes to their +# own notifications topic (idempotent with the server's auto-sub), +# user1 fires a fresh grant, user2 sees the one event. +# +# A fresh folder C is used so this scenario is independent of the +# S8 grant/revoke sequence — user2 already has DB rows from S8's +# grants on A + B, but those events fired BEFORE user2's WS opened +# so no wire delivery competes with S13's. +log "S13: create folder C, subscribe user2 to their notifications, expect one share_granted event." +folder_c=$(c_post "$base_url/api/folders" "$user1_token" \ + "$(printf '{"name":"rt_bus_C_%s","parent_id":"%s"}' "$suffix" "$root_id")" | jq -r '.id') +[[ -n "$folder_c" && "$folder_c" != "null" ]] || die "S13: folder C creation failed" + +out_s13="$(mktemp -t rtbus_s13.XXXXXX)" +ready_s13="$(mktemp -t rtbus_s13_ready.XXXXXX)"; rm -f "$ready_s13" +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --token "$user2_token" \ + --subscribe "user:${user2_id}:notifications" \ + --expect-events 1 \ + --timeout 5s \ + --ready-file "$ready_s13" \ + --output "$out_s13" & +helper_pid=$! +wait_ready "$ready_s13" + +grant_c=$(c_post "$base_url/api/grants" "$user1_token" \ + "$(printf '{"subject":{"type":"user","id":"%s"},"resource":{"type":"folder","id":"%s"},"role":"viewer"}' \ + "$user2_id" "$folder_c")") +grant_c_id=$(printf '%s' "$grant_c" | jq -r '.grants[0].id') +[[ -n "$grant_c_id" && "$grant_c_id" != "null" ]] \ + || die "S13: grant on folder C failed: $grant_c" + +if ! wait "$helper_pid"; then + cat "$out_s13" >&2 || true + die "S13: helper did not observe the notification_received event" +fi +[[ "$(jq -r '.events | length' "$out_s13")" == "1" ]] \ + || { cat "$out_s13"; die "S13: expected 1 event, got $(jq -r '.events | length' "$out_s13")"; } +[[ "$(jq -r '.events[0].event' "$out_s13")" == "notification_received" ]] \ + || die "S13: wrong event discriminator: $(jq -r '.events[0].event' "$out_s13")" +[[ "$(jq -r '.events[0].data.kind' "$out_s13")" == "share_granted" ]] \ + || die "S13: wrong notification kind: $(jq -r '.events[0].data.kind' "$out_s13")" +# `notification_id` is a fresh UUID stamped by the DB — check it's +# non-empty and non-null. Value asserted by S14 via GET /api/notifications. +[[ -n "$(jq -r '.events[0].data.notification_id' "$out_s13")" ]] \ + && [[ "$(jq -r '.events[0].data.notification_id' "$out_s13")" != "null" ]] \ + || die "S13: notification_id missing on wire payload" +log "S13 OK" + +# ── Scenario 14 — Notification DB row (Slice E) ───────────────────────────── +# The bus event is best-effort. The DB row is truth: a subscriber +# offline at publish time recovers via `GET /api/notifications`. +# S13 fired a grant on folder C; the ingester wrote a row for user2. +# This scenario reads it back and asserts on shape. +# +# `unread_count` from the same response reflects ALL unread rows, +# including the 2 from S8's grants (folders A + B) — the fresh grant +# in S13 brings the total to >= 3. We assert >= 1 (loose enough to +# not couple to S8's state, tight enough to prove the row landed). +log "S14: GET /api/notifications as user2; expect a share_granted row for folder C." +notifs=$(c_get "$base_url/api/notifications" "$user2_token") +unread=$(printf '%s' "$notifs" | jq -r '.unread_count') +[[ "$unread" -ge 1 ]] \ + || { printf '%s\n' "$notifs" >&2; die "S14: unread_count expected >= 1, got $unread"; } +# Filter for the S13 row: kind == share_granted AND payload.resource_id == folder_c. +match_count=$(printf '%s' "$notifs" | jq --arg fc "$folder_c" \ + '[.items[] | select(.kind == "share_granted" and .payload.resource_id == $fc)] | length') +[[ "$match_count" -ge 1 ]] \ + || { printf '%s\n' "$notifs" >&2; die "S14: no share_granted row for folder C (matches=$match_count)"; } +# The matched row must be unread (read_at is null) — the caller +# hasn't clicked it yet, so the bell would still badge it. +first_read_at=$(printf '%s' "$notifs" | jq -r --arg fc "$folder_c" \ + 'first(.items[] | select(.kind == "share_granted" and .payload.resource_id == $fc)) | .read_at') +[[ "$first_read_at" == "null" ]] \ + || die "S14: matched row unexpectedly marked read: read_at=$first_read_at" +log "S14 OK" + +# ── Scenario 15 — Cross-user notifications identity gate ──────────────────── +# `Topic::UserNotifications(u)` maps to `AuthzCheck::IdentityMatch{u}` +# in `application/ports/message_bus_ports.rs::required_perm`. Direct +# UUID equality only — no admin bypass. A caller subscribing to +# another user's notifications channel MUST be denied with the same +# wire shape (`topic_forbidden`) as an unknown topic — anti-enum. +# +# If this ever regresses (identity check dropped, engine wired on +# this class, admin bypass added) it becomes a privacy leak on par +# with an admin snooping on `:authz` streams. Same guard as S9, +# different topic suffix. +log "S15: user1 subscribes to user:{user2_id}:notifications; expect topic_forbidden." +if ! "$HELPER_BIN" expect-denied \ + --url "$ws_url" \ + --token "$user1_token" \ + --subscribe "user:${user2_id}:notifications" \ + --reason topic_forbidden \ + --timeout 3s; then + die "S15: user1 was NOT denied on user2's notifications topic (identity gate broken?)" +fi +log "S15 OK" + +log "All fifteen message-bus scenarios passed."