From 66980bba4320cc5f9e999967fe3986153cffb645 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?An=C4=B1lcan=20=C3=87ak=C4=B1r?= Date: Wed, 19 Aug 2026 13:07:05 +0300 Subject: [PATCH 1/2] feat(notifications): take notification state from the socket, not a 30s timer The bell polled `GET /notifications` every 30 seconds while an authenticated broadcast socket was already open in the same app, and there was not one line of Echo code in this package. So a notification the server had already published sat invisible for up to a polling interval, and the request that eventually found it asked for a list the socket could have handed over for free. `Notify.startRealtime(channel: ...)` subscribes to the notifiable's private channel and applies each `notification.created` frame straight to the stream. The frame carries the whole row in the same shape `GET /notifications` returns, so it is applied rather than used as a signal to fetch: asking the API for a row that just arrived in full is the round trip this removes. `startPolling()` becomes a no-op while realtime is live, so a consumer keeps wiring it to auth state and never has to branch on whether a socket happens to be up. Degradation runs in both directions, which is most of the code: - No broadcast driver configured returns false and changes nothing. The driver is read from config rather than probed by subscribing, because the null driver ACCEPTS a subscription and silently delivers nothing: an attempt-based check would report success on the one configuration that cannot work, silence the poller, and leave the bell permanently empty. - A dropped connection arms the poller as a stand-in and keeps the subscription, because realtime is still the intent. Without it, a socket that never comes back is a bell that never updates again. - A reconnect drops the fallback and refetches once, since Reverb has no replay. Only `connectionState` is watched, not `onReconnect` as well: a reconnect necessarily transitions the state to `connected`, so both would fetch twice for one event. - A redelivered id replaces the held row instead of appending a duplicate the bell would count twice, and a frame the decoder cannot read is logged and dropped rather than thrown into the driver's listener or allowed to clear the list. `magic` moves to `^0.0.6` for `Echo.connection`. Without a public accessor there is no way to tell an open connection from a closed one, and magic's Reverb driver opens a SECOND WebSocket on a redundant `connect()` rather than refusing it. A `0.0.z` caret pins the patch digit, so `^0.0.5` resolved exactly 0.0.5, which has no such accessor. The channel name is the caller's to supply: this package has no user model and cannot know whose notifications it is receiving. Verified against PUBLISHED magic 0.0.6 with the local path override removed, which is what CI resolves: 294 tests green, analyze clean, format clean. Four guards were mutation-checked and each reddens exactly one test, no two absorbing each other: the polling no-op, the null-driver refusal, the id dedupe, and the offline fallback. The tests drive frames through a channel double implementing the public `BroadcastChannel` contract, because the fake shipped in 0.0.6 discards both the event name and the callback. --- CHANGELOG.md | 12 + README.md | 33 +++ doc/architecture/notification-manager.md | 84 ++++++ doc/basics/laravel-backend-setup.md | 99 +++++++ lib/src/facades/notify.dart | 43 +++ lib/src/notification_manager.dart | 195 +++++++++++++ pubspec.yaml | 6 +- test/notification_realtime_test.dart | 334 +++++++++++++++++++++++ 8 files changed, 805 insertions(+), 1 deletion(-) create mode 100644 test/notification_realtime_test.dart diff --git a/CHANGELOG.md b/CHANGELOG.md index a404571..6a61989 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ ## [Unreleased] +### Added +- **Notification state can arrive over a socket instead of being polled for.** `Notify.startRealtime(channel: ...)` subscribes to the notifiable's private broadcast channel and applies each `notification.created` frame straight to the stream, so a new notification shows up when the server sends it rather than up to 30 seconds later. The frame carries the whole row in the same shape `GET /notifications` returns, so no HTTP follows it. +- `Notify.stopRealtime()`, `Notify.isRealtime` and `Notify.isPolling`. + +### Changed +- **`startPolling()` is a no-op while realtime is live.** A consumer keeps wiring it to auth state and does not have to know whether a socket happens to be up: with one, the 30-second timer is waste on top of a connection that already delivers every row; without one, nothing changes. `stopRealtime()` or a dropped connection restores the timer. +- **`magic` constraint bumped to `^0.0.6`.** `Echo.connection` (the public driver accessor) is the floor for the realtime path: without it there is no way to tell an already-open connection from a closed one, and magic's Reverb driver opens a SECOND WebSocket on a redundant `connect()` instead of refusing it. A `0.0.z` caret pins the patch digit, so `^0.0.5` resolved exactly 0.0.5, which has no such accessor. + +### Notes +- Realtime is opt-in and degrades in both directions. `startRealtime()` returns `false` and changes nothing when no broadcast driver is configured (a `BROADCAST_CONNECTION=null` deployment), and a socket that drops falls back to polling until it returns, at which point the fallback is dropped and the list is refetched once to cover what Reverb cannot replay. +- The channel name is the caller's to supply (`App.Models.User.{id}` by Laravel's default): this package has no user model and cannot know whose notifications it is receiving. See `doc/basics/laravel-backend-setup.md` for the server half. + ## [0.0.2] - 2026-07-26 ### Changed diff --git a/README.md b/README.md index b4625b8..8f192a1 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,7 @@ Managing notifications in Flutter means juggling multiple channels — database | :bell: | **Multi-channel** | Database, Push, and Mail channels through one API | | :iphone: | **OneSignal Push** | iOS, Android, and Web push via `onesignal_flutter` | | :arrows_counterclockwise: | **Real-time Polling** | Background polling with pause/resume/stop lifecycle | +| :satellite: | **Socket Delivery** | Take notification state from a broadcast channel instead, with polling as the fallback | | :dart: | **User Preferences** | Global and per-type channel preference management | | :hammer_and_wrench: | **CLI Tools** | Interactive install, configure, doctor, test, and more | | :gear: | **Config-Driven** | All settings in one Dart config file via `ConfigRepository` | @@ -134,6 +135,10 @@ import 'package:magic_notifications/magic_notifications.dart'; Future onLoginSuccess(User user) async { await Notify.requestPushPermission(); await Notify.initializePush('user_${user.id}'); + + // Prefer the socket; startPolling() is the fallback and no-ops when the + // socket is live, so both calls are safe in either order. + await Notify.startRealtime(channel: 'App.Models.User.${user.id}'); Notify.startPolling(); } ``` @@ -155,11 +160,39 @@ NotificationDropdownWithStream( ```dart Future onLogout() async { + Notify.stopRealtime(); Notify.stopPolling(); await Notify.logoutPush(); } ``` +### Receive Notifications Over a Socket + +`Notify.startRealtime()` subscribes to the notifiable's private broadcast channel +and applies each `notification.created` frame directly to the stream, so a new +notification appears when the server publishes it instead of up to one polling +interval later. It returns `false` and changes nothing when the app has no +broadcast driver configured, which is what keeps `startPolling()` meaningful on a +deployment without a socket. + +```dart +final bool live = await Notify.startRealtime( + channel: 'App.Models.User.${user.id}', +); +``` + +Behaviour worth knowing: + +- The existing list is fetched ONCE on start. A socket only carries what happens + next, so the rows that already exist still have to be read. +- A dropped connection falls back to polling; a reconnect drops the fallback and + refetches once, because Reverb has no replay. +- A redelivered id replaces the held row rather than appending a duplicate. + +The server half (the `broadcast` channel on the notification, the event name, and +the payload shape) is in +[doc/basics/laravel-backend-setup.md](doc/basics/laravel-backend-setup.md). + --- ## CLI Tools diff --git a/doc/architecture/notification-manager.md b/doc/architecture/notification-manager.md index a2d24ce..26a0211 100644 --- a/doc/architecture/notification-manager.md +++ b/doc/architecture/notification-manager.md @@ -7,6 +7,7 @@ - [Push Driver Setup](#push-driver) - [Send Dispatch Flow](#send-flow) - [Polling Orchestration](#polling) +- [Realtime Delivery](#realtime) - [Stream Management](#streams) - [Optimistic Updates with Rollback](#optimistic) @@ -174,6 +175,89 @@ poller.start(); --- +## Realtime Delivery + +Polling is the fallback, not the only path. When the app has a broadcast driver, +the manager can take notification state off the socket instead: + +```dart +final bool live = await Notify.startRealtime( + channel: 'App.Models.User.$userId', +); +``` + +`startRealtime()` does five things in order, and each one answers a specific +failure: + +1. **Refuses when there is no driver.** It reads `broadcasting.default` and + returns `false` for `null`, empty, or the literal `'null'` driver. It does NOT + try to subscribe and see what happens: the null driver accepts a subscription + and silently delivers nothing, so an attempt-based probe would report success + on the one configuration that cannot work, silence the poller, and leave the + bell permanently empty. +2. **Connects only if nothing is connected.** `Echo.connect()` is not idempotent + in magic's Reverb driver: it assigns a fresh channel without closing the + previous one, so a redundant call opens a second WebSocket and leaks the first. + This is why `magic ^0.0.6` is the floor; `Echo.connection` is the accessor that + makes the check possible. +3. **Listens for `notification.created` exactly once.** A second `listen()` for + one event name REPLACES the earlier handler rather than adding to it, so + registering the same event anywhere else would silently drop this one. +4. **Stops the poller.** The socket now covers it. +5. **Fetches the existing list once.** A socket carries only what happens next. + +### The frame is the state + +A `notification.created` frame carries the whole row in the same shape +`GET /notifications` returns, so it is decoded and applied rather than treated as +a signal to fetch: + +```dart +void _applyRealtimeFrame(BroadcastEvent event) { + final incoming = DatabaseNotification.fromMap(event.data); + _notifications = [ + incoming, + ..._notifications.where((n) => n.id != incoming.id), + ]; + _notificationController.add(_notifications); +} +``` + +Newest first, keyed by id, so a redelivery replaces the held row instead of +appending a duplicate the bell would count twice. A payload the decoder cannot +read is logged and dropped: it must not throw into the driver's listener, and it +must not clear what is already held, because a backend one version ahead is a +reason to miss one row and not to empty the list. + +### Both directions of degradation + +`connectionState` is watched (and `onReconnect` deliberately is not, since a +reconnect necessarily transitions the state to `connected` and listening to both +would fetch the same list twice for one event): + +| Transition | What happens | +|------------|--------------| +| anything but `connected` | the poller is armed as a stand-in; the subscription stays, because realtime is still the intent | +| `connected` | the poller is dropped and the list is refetched once, because Reverb has no replay | + +Without the first row, a socket that never comes back is a bell that never +updates again. + +### Lifecycle hooks + +| App Event | Call | +|-----------|------| +| User logged in | `Notify.startRealtime(channel: ...)` then `Notify.startPolling()` | +| User switched account | `Notify.startRealtime(channel: ...)` with the new channel | +| User logged out | `Notify.stopRealtime()` then `Notify.stopPolling()` | + +Both calls on the login row are safe in either order and are idempotent per +channel, so wiring them to an auth-state listener that fires on every login, +logout and restore is the intended shape. `startPolling()` is a no-op while +realtime is live, so a consumer never has to branch on whether a socket is up. + +--- + ## Stream Management Database notifications are distributed via a broadcast `StreamController`: diff --git a/doc/basics/laravel-backend-setup.md b/doc/basics/laravel-backend-setup.md index d5043f9..13d5a8d 100644 --- a/doc/basics/laravel-backend-setup.md +++ b/doc/basics/laravel-backend-setup.md @@ -10,6 +10,7 @@ - [API Controllers](#api-controllers) - [API Routes](#api-routes) - [OneSignal Push Integration](#onesignal-push) +- [Socket Delivery (Broadcast)](#broadcast) - [Sending Notifications](#sending) - [API Contract Reference](#api-contract) - [Testing](#testing) @@ -458,6 +459,104 @@ public function routeNotificationForOneSignal(): array --- +## Socket Delivery (Broadcast) + +The client can take notification state off a socket instead of polling for it +(`Notify.startRealtime`). That needs three things on this side: the `broadcast` +channel on the notification, a payload the client's decoder can read, and channel +authorisation. + +### 1. Add the channel and shape the frame + +```php +use Illuminate\Notifications\Messages\BroadcastMessage; + +class MonitorDownNotification extends Notification +{ + public function via(object $notifiable): array + { + // Alongside `database`, never instead of it: the socket is delivery, the + // row is the record. A client that was offline reads the row over the API. + return ['database', 'broadcast']; + } + + /** + * The wire event name. + * + * Laravel's default is the fully-qualified + * `Illuminate\Notifications\Events\BroadcastNotificationCreated`. Magic's + * Reverb channel matches a listener by EXACT string, so the client would have + * to hardcode a framework internal; this short name is the contract instead. + */ + public function broadcastAs(): string + { + return 'notification.created'; + } + + /** + * The payload, in the SAME shape `GET /notifications` returns a row. + * + * This matters more than it looks. Laravel's default broadcast payload + * FLATTENS the notification data to the top level and adds `id` and `type`, + * while `DatabaseNotification.fromMap` on the client reads `data.title`, + * `data.body` and `data.action_url` from a NESTED `data` key, so the default + * decodes to nothing useful. Building the frame from `toArray()` keeps one + * serializer behind both the API and the socket, so the two cannot drift. + */ + public function toBroadcast(object $notifiable): BroadcastMessage + { + return new BroadcastMessage([ + 'id' => $this->id, + 'type' => static::class, + 'data' => $this->toArray($notifiable), + 'created_at' => now()->toIso8601String(), + 'read_at' => null, + ]); + } +} +``` + +### 2. Authorise the channel + +The default channel for a `Notifiable` is its class name with dots plus its key, +so `App\Models\User` id `42` publishes on `App.Models.User.42`. Authorise it in +`routes/channels.php`, scoped to the one user: + +```php +use App\Models\User; +use Illuminate\Support\Facades\Broadcast; + +Broadcast::channel('App.Models.User.{userId}', function (User $user, string $userId): bool { + // A notification is personal. Comparing as strings keeps a UUID key working. + return (string) $user->id === (string) $userId; +}, ['guards' => ['sanctum']]); +``` + +The `guards` option matters for an API client: the channel-auth request arrives +with a bearer token, not a session cookie, so the default `web` guard would deny +every subscription. + +Override `receivesBroadcastNotificationsOn()` on the notifiable if you want a +different name; the client is told the channel by its caller, so any name works as +long as both sides agree. + +### 3. Point the client at it + +```dart +await Notify.startRealtime(channel: 'App.Models.User.${user.id}'); +``` + +Nothing else changes. `Notify.startPolling()` stays wired to auth state and +becomes a no-op while the socket is live, so a deployment with +`BROADCAST_CONNECTION=null` keeps polling and needs no client change at all. + +> **The row is still the record.** Keep `database` in `via()`. The socket delivers +> to whoever is connected; a client that was closed when the notification fired +> learns about it from the API on its next start, which is the fetch +> `startRealtime()` does once. + +--- + ## Sending Notifications ### From Controller diff --git a/lib/src/facades/notify.dart b/lib/src/facades/notify.dart index e85752a..da92ac9 100644 --- a/lib/src/facades/notify.dart +++ b/lib/src/facades/notify.dart @@ -185,4 +185,47 @@ class Notify { static void resumePolling() { manager.resumePolling(); } + + /// Whether the periodic poller is currently armed. + static bool get isPolling => manager.isPolling; + + // ======================================== + // Realtime + // ======================================== + + /// Receive notification state over the app's broadcast socket instead of + /// polling for it. + /// + /// [channel] is the private channel the backend publishes the user's rows on, + /// `App.Models.User.{id}` by Laravel's default. Wire it to auth state next to + /// [startPolling], which becomes a no-op while this is live: + /// + /// ```dart + /// if (Auth.check()) { + /// await Notify.startRealtime(channel: 'App.Models.User.' + User.current.id); + /// Notify.startPolling(); // the fallback, if there is no socket + /// } else { + /// Notify.stopRealtime(); + /// Notify.stopPolling(); + /// } + /// ``` + /// + /// Returns false when the app has no broadcast driver configured, so the caller + /// above keeps polling. See [NotificationManager.startRealtime]. + static Future startRealtime({ + String? channel, + String event = NotificationManager.realtimeEvent, + }) { + return manager.startRealtime(channel: channel, event: event); + } + + /// Stop receiving notification state over the socket. + /// + /// Call on logout, next to [stopPolling]. + static void stopRealtime() { + manager.stopRealtime(); + } + + /// Whether notification state is currently arriving over a socket. + static bool get isRealtime => manager.isRealtime; } diff --git a/lib/src/notification_manager.dart b/lib/src/notification_manager.dart index 11afc75..133d32d 100644 --- a/lib/src/notification_manager.dart +++ b/lib/src/notification_manager.dart @@ -58,6 +58,17 @@ class NotificationManager { /// Notification poller for periodic fetching NotificationPoller? _poller; + /// The private channel notifications are being received on, or null when the + /// realtime path is not active. + BroadcastChannel? _realtimeChannel; + + /// The channel NAME realtime was started for, retained so a repeat call for the + /// same channel is a no-op and a different one moves the subscription. + String? _realtimeChannelName; + + /// The connection-state subscription that drives the polling fallback. + StreamSubscription? _realtimeConnection; + factory NotificationManager() { return _instance; } @@ -384,6 +395,16 @@ class NotificationManager { /// Creates and starts a poller if one doesn't exist. /// Safe to call multiple times (idempotent). void startPolling() { + // A live socket already delivers every new notification, so a 30-second HTTP + // timer on top of it asks the server for what it has just been told. The + // realtime path does its own single initial fetch and its own refetch on + // reconnect, so there is nothing left for the timer to cover. + // + // This is a NO-OP rather than an error: a consumer wires `startPolling()` to + // its auth state and should not have to know whether a socket happens to be + // up. `stopRealtime()` (or a dropped connection) restores the timer. + if (isRealtime) return; + _poller ??= NotificationPoller(this); _poller!.start(); } @@ -409,4 +430,178 @@ class NotificationManager { void resumePolling() { _poller?.resume(); } + + /// Whether the periodic poller is currently armed and fetching. + bool get isPolling => _poller?.isActive ?? false; + + // ======================================== + // Realtime Notification Methods + // ======================================== + + /// The wire event name a new notification arrives as. + /// + /// The server side is a notification declaring `broadcastAs()`. Laravel's own + /// default is the fully-qualified `Illuminate\Notifications\Events\BroadcastNotificationCreated`, + /// which works but reads badly in a Dart listener and ties the client to a + /// framework internal, so the contract is this short name. + static const String realtimeEvent = 'notification.created'; + + /// Whether notification state is currently arriving over a socket. + bool get isRealtime => _realtimeChannel != null; + + /// Receives notification state from a broadcast channel instead of polling for + /// it. + /// + /// [channel] is the private channel the backend publishes the notifiable's rows + /// on, `App.Models.User.{id}` for a Laravel `Notifiable` that has not overridden + /// `receivesBroadcastNotificationsOn()`. The name has to come from the caller: + /// this package has no user model and cannot know whose notifications these are. + /// + /// Returns false, changing nothing, when the app has no broadcast driver + /// configured. That is the case a `null` [BROADCAST_CONNECTION] deployment is + /// in, and reporting success there would stop the poller and leave the bell + /// permanently empty, which is strictly worse than polling. + /// + /// On success it: + /// + /// 1. connects only if no connection exists. `connect()` is not idempotent in + /// magic's Reverb driver (it assigns a fresh channel without closing the + /// previous one), so a second call opens a second WebSocket and leaks the + /// first; + /// 2. subscribes and listens for [realtimeEvent] exactly once. A second + /// `listen()` for one event REPLACES the earlier handler rather than adding + /// to it, so registering anywhere else would silently drop this one; + /// 3. stops the poller, because the socket now covers it; + /// 4. fetches the existing list ONCE. A socket carries only what happens next, + /// so the rows that already exist have to be read; + /// 5. watches the connection so a drop falls back to polling and a reconnect + /// lifts the fallback and closes the replay gap. + /// + /// Idempotent per channel and safe to call on every auth-state change: the same + /// channel is a no-op, a different one moves the subscription. + Future startRealtime({ + String? channel, + String event = realtimeEvent, + }) async { + if (channel == null || channel.isEmpty) return false; + if (!_broadcastingEnabled()) return false; + if (_realtimeChannelName == channel) return true; + + // A move: drop the previous channel before the first await, so a failed + // connect leaves a clean unsubscribed state the next call retries rather than + // a marker pointing at a channel nothing is listening on. + if (_realtimeChannel != null) stopRealtime(); + + try { + if (!Echo.connection.isConnected) { + await Echo.connect(); + } + final BroadcastChannel subscribed = Echo.private(channel); + subscribed.listen(event, _applyRealtimeFrame); + _realtimeChannel = subscribed; + _realtimeChannelName = channel; + _watchRealtimeConnection(); + } catch (e) { + _safeLogError('Failed to start realtime notifications: $e'); + stopRealtime(); + + return false; + } + + stopPolling(); + await fetchNotifications(); + + return true; + } + + /// Stops receiving notification state over a socket. + /// + /// Leaves the channel and drops the connection watcher, but does NOT touch the + /// connection itself: it is shared with whatever else the app subscribes to. + /// Polling is not restarted here either, because only the caller knows whether + /// the user is still authenticated; a subsequent [startPolling] arms it. + void stopRealtime() { + final BroadcastChannel? channel = _realtimeChannel; + if (channel != null) { + Echo.leave(channel.name); + } + _realtimeChannel = null; + _realtimeChannelName = null; + _realtimeConnection?.cancel(); + _realtimeConnection = null; + } + + /// True when the app has a broadcast driver that can actually deliver a frame. + /// + /// Reads the configured driver rather than trying to subscribe and seeing what + /// happens: the null driver accepts a subscription and silently delivers + /// nothing, so an attempt-based probe would report success on the one + /// configuration that cannot work. + bool _broadcastingEnabled() { + final String? driver = Config.get('broadcasting.default'); + + return driver != null && driver.isNotEmpty && driver != 'null'; + } + + /// Falls back to polling while the socket is away, and lifts the fallback with + /// a refetch when it returns. + /// + /// Only `connectionState` is watched, not `onReconnect` as well: a reconnect + /// necessarily transitions the state to `connected`, so listening to both would + /// fetch the same list twice for one event. + void _watchRealtimeConnection() { + _realtimeConnection?.cancel(); + _realtimeConnection = Echo.connectionState.listen(( + BroadcastConnectionState state, + ) { + if (state == BroadcastConnectionState.connected) { + // The socket is back. Reverb has no replay, so anything published while + // it was down is gone from the stream and only a fetch recovers it. + _poller?.stop(); + _poller = null; + fetchNotifications(); + + return; + } + + // Anything else (reconnecting, disconnected) means frames are not arriving. + // The subscription stays: realtime is still the intent, polling is the + // stand-in. Without it a socket that never comes back is a bell that never + // updates again. + _poller ??= NotificationPoller(this); + _poller!.start(); + }); + } + + /// Applies one `notification.created` frame to the cached list and the stream. + /// + /// The frame carries the whole row in the same shape `GET /notifications` + /// returns, so it is decoded and applied rather than used as a signal to fetch: + /// asking the API for a row that just arrived in full is the round trip this + /// path exists to remove. + /// + /// Newest first, and keyed by id: a redelivery (a socket retry, or the same + /// notification broadcast twice) replaces the held row instead of appending a + /// duplicate the bell would count twice. + /// + /// A payload the decoder cannot read is logged and dropped. It must not throw + /// into the driver's listener, and it must not clear what is already held: a + /// backend one version ahead is a reason to miss one row, not to empty the + /// list. + void _applyRealtimeFrame(BroadcastEvent event) { + try { + final DatabaseNotification incoming = DatabaseNotification.fromMap( + event.data, + ); + _notifications = [ + incoming, + ..._notifications.where( + (DatabaseNotification n) => n.id != incoming.id, + ), + ]; + _notificationController.add(_notifications); + } catch (e) { + _safeLogError('Failed to decode a realtime notification: $e'); + } + } } diff --git a/pubspec.yaml b/pubspec.yaml index 310079e..ab3e38f 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -20,7 +20,11 @@ environment: dependencies: flutter: sdk: flutter - magic: ^0.0.5 + # ^0.0.6 is the floor for `Echo.connection`, which the realtime notification + # path needs to avoid a second WebSocket on an already-open connection. A 0.0.x + # caret pins the patch, so ^0.0.5 resolves exactly 0.0.5 and that release has + # no public connection accessor. + magic: ^0.0.6 fluttersdk_artisan: ^0.0.8 onesignal_flutter: ^5.4.6 diff --git a/test/notification_realtime_test.dart b/test/notification_realtime_test.dart new file mode 100644 index 0000000..e1535a0 --- /dev/null +++ b/test/notification_realtime_test.dart @@ -0,0 +1,334 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:magic/magic.dart'; +import 'package:magic_notifications/magic_notifications.dart'; + +import 'test_helper.dart'; + +/// A channel that keeps its handlers so a test can deliver a frame. +/// +/// The shipped [FakeBroadcastDriver]'s channel discards both the event name and +/// the callback, so a test against it can prove a channel was opened and nothing +/// about what the app does with a frame that arrives on it. +class _RecordingChannel implements BroadcastChannel { + _RecordingChannel(this.name); + + @override + final String name; + + final Map handlers = + {}; + + @override + Stream get events => const Stream.empty(); + + @override + BroadcastChannel listen( + String event, + void Function(BroadcastEvent) callback, + ) { + handlers[event] = callback; + return this; + } + + @override + void stopListening(String event) => handlers.remove(event); + + /// Delivers [data] to the handler registered for [event], the way the driver + /// delivers a real frame. + void emit(String event, Map data) { + handlers[event]?.call( + BroadcastEvent( + event: event, + channel: name, + data: data, + receivedAt: DateTime(2026, 8, 19), + ), + ); + } +} + +/// A driver whose private channels are inspectable and whose connection state is +/// drivable, so the fallback path can be tested at all. +class _RecordingDriver extends FakeBroadcastDriver { + final Map channels = {}; + final List left = []; + final StreamController states = + StreamController.broadcast(); + + @override + Stream get connectionState => states.stream; + + @override + BroadcastChannel private(String name) { + super.private(name); + return channels.putIfAbsent( + 'private-$name', + () => _RecordingChannel('private-$name'), + ); + } + + @override + void leave(String name) { + left.add(name); + channels.remove(name); + super.leave(name); + } +} + +/// Hands out the recording driver in place of the parent's private one. +class _RecordingManager extends FakeBroadcastManager { + final _RecordingDriver spy = _RecordingDriver(); + + @override + BroadcastDriver connection([String? name]) => spy; +} + +void main() { + late NotificationManager manager; + late _RecordingManager echo; + + /// One notification row as the API and the socket both shape it. + Map row({ + String id = 'n1', + String title = 'Incident opened', + String createdAt = '2026-08-19T09:30:00.000Z', + }) => + { + 'id': id, + 'type': 'App\\Notifications\\IncidentOpened', + 'data': { + 'title': title, + 'body': 'API Health is down', + 'action_url': '/incidents/$id', + }, + 'created_at': createdAt, + 'read_at': null, + }; + + setUpAll(() async { + await initMagicForTests(); + }); + + setUp(() { + manager = NotificationManager(); + manager.forgetChannels(); + manager.stopRealtime(); + manager.stopPolling(); + echo = _RecordingManager(); + Magic.app.setInstance('broadcasting', echo); + Config.set('broadcasting.default', 'reverb'); + Http.fake({ + 'notifications': Http.response({'data': []}), + }); + }); + + tearDown(() { + manager.stopRealtime(); + manager.stopPolling(); + Config.forget('broadcasting.default'); + }); + + group('NotificationManager realtime', () { + test('subscribes to the private channel and reports realtime', () async { + final bool started = await manager.startRealtime( + channel: 'App.Models.User.u1', + ); + + expect(started, isTrue); + expect(manager.isRealtime, isTrue); + expect( + echo.spy.channels['private-App.Models.User.u1']!.handlers.keys, + contains('notification.created'), + ); + }); + + test('declines when no broadcast driver is configured', () async { + // A deployment with `BROADCAST_CONNECTION=null` has no socket to receive + // anything on. Reporting success here would silence the poller and leave + // the bell permanently empty, which is worse than polling. + Config.set('broadcasting.default', 'null'); + + final bool started = await manager.startRealtime( + channel: 'App.Models.User.u1', + ); + + expect(started, isFalse); + expect(manager.isRealtime, isFalse); + }); + + test('fetches the existing list once when it starts', () async { + final FakeNetworkDriver driver = Http.fake({ + 'notifications': Http.response({ + 'data': [row(id: 'old')], + }), + }); + + await manager.startRealtime(channel: 'App.Models.User.u1'); + + // The socket only ever carries what happens NEXT, so the list that already + // exists has to be read once. This is the only HTTP the realtime path does + // in its steady state. + expect( + driver.recorded + .where((entry) => entry.$1.url.contains('notifications')) + .length, + 1, + ); + final List current = + await manager.notifications().first; + expect(current.map((DatabaseNotification n) => n.id), ['old']); + }); + + test('a frame prepends to the stream with no further HTTP', () async { + final FakeNetworkDriver driver = Http.fake({ + 'notifications': Http.response({ + 'data': [row(id: 'old')], + }), + }); + await manager.startRealtime(channel: 'App.Models.User.u1'); + final int afterStart = driver.recorded.length; + + echo.spy.channels['private-App.Models.User.u1']!.emit( + 'notification.created', + row(id: 'fresh', title: 'Checkout is down'), + ); + + final List current = + await manager.notifications().first; + // Newest first, and the frame IS the state: asking the API again for a row + // that just arrived in full is the round trip this replaces. + expect( + current.map((DatabaseNotification n) => n.id), + ['fresh', 'old'], + ); + expect(current.first.title, 'Checkout is down'); + expect(driver.recorded.length, afterStart); + }); + + test('a frame the decoder cannot read is dropped, not thrown', () async { + await manager.startRealtime(channel: 'App.Models.User.u1'); + + // A payload shaped by a newer or older backend must not take down the + // listener that delivered it, and must not clear what is already held. + echo.spy.channels['private-App.Models.User.u1']!.emit( + 'notification.created', + {'id': 'broken'}, + ); + + final List current = + await manager.notifications().first; + expect(current, isEmpty); + expect(manager.isRealtime, isTrue); + }); + + test('a redelivered id replaces rather than duplicates', () async { + await manager.startRealtime(channel: 'App.Models.User.u1'); + final _RecordingChannel channel = + echo.spy.channels['private-App.Models.User.u1']!; + + channel.emit('notification.created', row(id: 'n1', title: 'First')); + channel.emit('notification.created', row(id: 'n1', title: 'Corrected')); + + final List current = + await manager.notifications().first; + expect(current, hasLength(1)); + expect(current.single.title, 'Corrected'); + }); + }); + + group('NotificationManager polling under realtime', () { + test('startPolling does not arm the timer while realtime is live', + () async { + await manager.startRealtime(channel: 'App.Models.User.u1'); + + manager.startPolling(); + + // The whole point of the feature: an authenticated socket is already open, + // so a 30-second HTTP timer on top of it is pure waste. + expect(manager.isPolling, isFalse); + }); + + test('startRealtime stops a poller that was already running', () async { + manager.startPolling(); + expect(manager.isPolling, isTrue); + + await manager.startRealtime(channel: 'App.Models.User.u1'); + + expect(manager.isPolling, isFalse); + }); + + test('a lost connection falls back to polling', () async { + await manager.startRealtime(channel: 'App.Models.User.u1'); + expect(manager.isPolling, isFalse); + + echo.spy.states.add(BroadcastConnectionState.reconnecting); + await Future.delayed(Duration.zero); + + // Realtime is still the intent, so the subscription stays; polling is the + // stand-in while the socket is away. Without this, a socket that never + // comes back means a bell that never updates again. + expect(manager.isPolling, isTrue); + expect(manager.isRealtime, isTrue); + }); + + test('reconnecting stops the fallback and closes the replay gap', () async { + final FakeNetworkDriver driver = Http.fake({ + 'notifications': Http.response({'data': []}), + }); + await manager.startRealtime(channel: 'App.Models.User.u1'); + echo.spy.states.add(BroadcastConnectionState.reconnecting); + await Future.delayed(Duration.zero); + final int beforeReconnect = driver.recorded.length; + + echo.spy.states.add(BroadcastConnectionState.connected); + await Future.delayed(Duration.zero); + + // Reverb has no replay, so anything published while the socket was down is + // gone from the stream and only a fetch can recover it. + expect(manager.isPolling, isFalse); + expect(driver.recorded.length, greaterThan(beforeReconnect)); + }); + + test('stopRealtime leaves the channel and lets polling resume', () async { + await manager.startRealtime(channel: 'App.Models.User.u1'); + + manager.stopRealtime(); + + expect(manager.isRealtime, isFalse); + expect(echo.spy.left, contains('private-App.Models.User.u1')); + + manager.startPolling(); + expect(manager.isPolling, isTrue); + }); + + test('starting realtime twice on the same channel is idempotent', () async { + await manager.startRealtime(channel: 'App.Models.User.u1'); + await manager.startRealtime(channel: 'App.Models.User.u1'); + + // The app re-syncs on every auth-state bump, so this runs often. A second + // subscribe on magic's Reverb channel REPLACES the first listener rather + // than adding one, and a second `connect()` opens a second socket. + expect(echo.spy.left, isEmpty); + expect( + echo.spy.subscribedChannels + .where((String c) => c == 'private-App.Models.User.u1') + .length, + 1, + ); + }); + + test('a different channel moves the subscription', () async { + await manager.startRealtime(channel: 'App.Models.User.u1'); + + await manager.startRealtime(channel: 'App.Models.User.u2'); + + expect(echo.spy.left, contains('private-App.Models.User.u1')); + expect( + echo.spy.channels.keys, + contains('private-App.Models.User.u2'), + ); + }); + }); +} From 9e2520d7e2ce910f8c28a8f1172892231342dc2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?An=C4=B1lcan=20=C3=87ak=C4=B1r?= Date: Wed, 19 Aug 2026 14:14:50 +0300 Subject: [PATCH 2/2] fix(notifications): four review findings, one of them a lost notification All four came from the Copilot review on #12 and all four are real. **A frame that landed mid-fetch was clobbered.** `startRealtime()` subscribes and THEN fetches the existing list, so there is a window where a frame prepends to the cached list and the completing read assigns the server's list over the top. The notification was gone until something fetched again. That window is not theoretical: it is the exact moment a backlog of pending notifications is most likely to be publishing. Frames received during a read are now merged back on top, keyed by id, in reverse arrival order so the newest stays at the head. The buffer is cleared in a `finally`, including on a failed read, because the frame was already applied to the cached list as it arrived and must not be re-merged into the next fetch. **The idempotence key ignored the event name.** Keyed on the channel alone, a caller that passed a different `event` for the same channel hit the early return, so the manager silently kept handling the OLD event name: no subscription change, no error, no notifications. The event is part of the key now. **A dartdoc symbol link that is not a symbol.** `[BROADCAST_CONNECTION]` is link markup, not code font, so generated docs carry a broken link. Backticks. **A test-owned broadcast StreamController was never closed**, leaking a handle into every following test. Closed in `tearDown`. Three new tests, each mutation-checked: dropping the mid-fetch merge reddens the two that name it, and reverting the idempotence key reddens only the event one. The mid-fetch test drives the race deterministically rather than by timing, because `fetchNotifications()` sets its in-flight flag synchronously and then suspends on the HTTP await, so emitting between the call and the await lands the frame inside the window every run. 297 tests green, analyze clean, format clean, and re-verified with the local path override removed so the run resolves the PUBLISHED magic 0.0.6 that CI resolves. --- CHANGELOG.md | 4 ++ lib/src/notification_manager.dart | 82 ++++++++++++++++++++++++---- test/notification_realtime_test.dart | 73 +++++++++++++++++++++++++ 3 files changed, 148 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a61989..42bf4fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,10 @@ - **`startPolling()` is a no-op while realtime is live.** A consumer keeps wiring it to auth state and does not have to know whether a socket happens to be up: with one, the 30-second timer is waste on top of a connection that already delivers every row; without one, nothing changes. `stopRealtime()` or a dropped connection restores the timer. - **`magic` constraint bumped to `^0.0.6`.** `Echo.connection` (the public driver accessor) is the floor for the realtime path: without it there is no way to tell an already-open connection from a closed one, and magic's Reverb driver opens a SECOND WebSocket on a redundant `connect()` instead of refusing it. A `0.0.z` caret pins the patch digit, so `^0.0.5` resolved exactly 0.0.5, which has no such accessor. +### Fixed +- **A socket frame that arrived while `fetchNotifications()` was in flight was clobbered by the read.** The frame prepended to the cached list and the server's list was then assigned over the top, so the notification vanished until something fetched again. The window is small and entirely real, because `startRealtime()` subscribes and THEN fetches, which is the exact moment a backlog is most likely to be publishing. Frames received during a read are now merged back on top, keyed by id. +- **`startRealtime()`'s idempotence key now includes the event name.** Keyed on the channel alone, a caller that changed the event for the same channel hit the early return, so the manager silently kept handling the old event name and delivered nothing, with no error. + ### Notes - Realtime is opt-in and degrades in both directions. `startRealtime()` returns `false` and changes nothing when no broadcast driver is configured (a `BROADCAST_CONNECTION=null` deployment), and a socket that drops falls back to polling until it returns, at which point the fallback is dropped and the list is refetched once to cover what Reverb cannot replay. - The channel name is the caller's to supply (`App.Models.User.{id}` by Laravel's default): this package has no user model and cannot know whose notifications it is receiving. See `doc/basics/laravel-backend-setup.md` for the server half. diff --git a/lib/src/notification_manager.dart b/lib/src/notification_manager.dart index 133d32d..602ae4d 100644 --- a/lib/src/notification_manager.dart +++ b/lib/src/notification_manager.dart @@ -69,6 +69,18 @@ class NotificationManager { /// The connection-state subscription that drives the polling fallback. StreamSubscription? _realtimeConnection; + /// The event name realtime was started for, part of the idempotence key so a + /// caller that changes the event for the same channel is not silently ignored. + String? _realtimeEventName; + + /// Whether a [fetchNotifications] read is currently in flight. + bool _fetching = false; + + /// Frames that arrived while a read was in flight, merged back on top of the + /// fetched list so the read cannot clobber them. + final List _framesDuringFetch = + []; + factory NotificationManager() { return _instance; } @@ -140,6 +152,8 @@ class NotificationManager { /// /// Updates the notification stream with fresh data from the API. Future fetchNotifications() async { + _fetching = true; + try { final response = await Http.get('/notifications'); @@ -147,16 +161,58 @@ class NotificationManager { final data = response.data; final List items = data['data'] ?? []; - _notifications = items.map((item) { - return DatabaseNotification.fromMap(item as Map); - }).toList(); + _notifications = _withFramesReceivedDuringFetch( + items.map((item) { + return DatabaseNotification.fromMap(item as Map); + }).toList(), + ); _notificationController.add(_notifications); } } catch (e) { _safeLogError('Failed to fetch notifications: $e'); // Don't throw - just keep current state + } finally { + // Always cleared, including on a failed read: a frame received during the + // window was applied to `_notifications` as it arrived, so it is already + // held and must not be re-merged into the NEXT fetch as well. + _fetching = false; + _framesDuringFetch.clear(); + } + } + + /// The fetched list with every frame received DURING the read merged back on + /// top, newest first and keyed by id. + /// + /// Without this the read clobbers a frame that landed mid-flight: the frame + /// prepends to the cached list, then the server's list is assigned over the top + /// and the notification is gone until something fetches again. The window is + /// small and entirely real, because [startRealtime] subscribes and THEN fetches, + /// which is the exact moment a backlog is most likely to be publishing. + /// + /// Merged in reverse arrival order, so successive prepends leave the newest + /// frame at the head. + List _withFramesReceivedDuringFetch( + List fetched, + ) { + List merged = fetched; + for (final DatabaseNotification frame in _framesDuringFetch.reversed) { + merged = _prependKeyedById(frame, merged); } + + return merged; + } + + /// [incoming] at the head of [into], with any earlier row carrying the same id + /// removed, so a redelivery replaces rather than duplicates. + List _prependKeyedById( + DatabaseNotification incoming, + List into, + ) { + return [ + incoming, + ...into.where((DatabaseNotification n) => n.id != incoming.id), + ]; } /// Fetch paginated notifications from backend. @@ -458,7 +514,7 @@ class NotificationManager { /// this package has no user model and cannot know whose notifications these are. /// /// Returns false, changing nothing, when the app has no broadcast driver - /// configured. That is the case a `null` [BROADCAST_CONNECTION] deployment is + /// configured. That is the case a `null` `BROADCAST_CONNECTION` deployment is /// in, and reporting success there would stop the poller and leave the bell /// permanently empty, which is strictly worse than polling. /// @@ -485,7 +541,9 @@ class NotificationManager { }) async { if (channel == null || channel.isEmpty) return false; if (!_broadcastingEnabled()) return false; - if (_realtimeChannelName == channel) return true; + if (_realtimeChannelName == channel && _realtimeEventName == event) { + return true; + } // A move: drop the previous channel before the first await, so a failed // connect leaves a clean unsubscribed state the next call retries rather than @@ -500,6 +558,7 @@ class NotificationManager { subscribed.listen(event, _applyRealtimeFrame); _realtimeChannel = subscribed; _realtimeChannelName = channel; + _realtimeEventName = event; _watchRealtimeConnection(); } catch (e) { _safeLogError('Failed to start realtime notifications: $e'); @@ -527,6 +586,7 @@ class NotificationManager { } _realtimeChannel = null; _realtimeChannelName = null; + _realtimeEventName = null; _realtimeConnection?.cancel(); _realtimeConnection = null; } @@ -593,12 +653,12 @@ class NotificationManager { final DatabaseNotification incoming = DatabaseNotification.fromMap( event.data, ); - _notifications = [ - incoming, - ..._notifications.where( - (DatabaseNotification n) => n.id != incoming.id, - ), - ]; + // Applied immediately either way, so the bell shows it without waiting; + // the buffer only exists so a read completing after this cannot drop it. + if (_fetching) { + _framesDuringFetch.add(incoming); + } + _notifications = _prependKeyedById(incoming, _notifications); _notificationController.add(_notifications); } catch (e) { _safeLogError('Failed to decode a realtime notification: $e'); diff --git a/test/notification_realtime_test.dart b/test/notification_realtime_test.dart index e1535a0..31e2b79 100644 --- a/test/notification_realtime_test.dart +++ b/test/notification_realtime_test.dart @@ -128,6 +128,9 @@ void main() { manager.stopRealtime(); manager.stopPolling(); Config.forget('broadcasting.default'); + // The recording driver owns a broadcast StreamController per test; leaving it + // open leaks a handle into every following test and hangs stricter runners. + echo.spy.states.close(); }); group('NotificationManager realtime', () { @@ -236,6 +239,76 @@ void main() { expect(current, hasLength(1)); expect(current.single.title, 'Corrected'); }); + + test('a frame that lands mid-fetch survives the fetch', () async { + Http.fake({ + 'notifications': Http.response({ + 'data': [row(id: 'old')], + }), + }); + await manager.startRealtime(channel: 'App.Models.User.u1'); + final _RecordingChannel channel = + echo.spy.channels['private-App.Models.User.u1']!; + + // `fetchNotifications()` sets its in-flight flag synchronously and then + // suspends on the HTTP await, so emitting here lands the frame INSIDE the + // window. Unhandled, the read then assigns the server's list over the top + // and the notification is gone until something fetches again, which is + // exactly the window `startRealtime` opens by subscribing before it fetches. + final Future reading = manager.fetchNotifications(); + channel.emit( + 'notification.created', row(id: 'fresh', title: 'Mid-flight')); + await reading; + + final List current = + await manager.notifications().first; + expect( + current.map((DatabaseNotification n) => n.id), + ['fresh', 'old'], + ); + expect(current.first.title, 'Mid-flight'); + }); + + test('a second frame id from the same window is not duplicated', () async { + Http.fake({ + 'notifications': Http.response({ + 'data': [row(id: 'fresh', title: 'From the API')], + }), + }); + await manager.startRealtime(channel: 'App.Models.User.u1'); + final _RecordingChannel channel = + echo.spy.channels['private-App.Models.User.u1']!; + + // The server's list already carries the row the frame announced, which is + // the common case once a backlog drains. Merging must not leave two. + final Future reading = manager.fetchNotifications(); + channel.emit( + 'notification.created', row(id: 'fresh', title: 'From the socket')); + await reading; + + final List current = + await manager.notifications().first; + expect(current, hasLength(1)); + expect(current.single.title, 'From the socket'); + }); + + test('changing the event for the same channel re-subscribes', () async { + await manager.startRealtime(channel: 'App.Models.User.u1'); + + await manager.startRealtime( + channel: 'App.Models.User.u1', + event: 'notification.pushed', + ); + + // The event is part of the idempotence key. Keyed on the channel alone, the + // early return skipped the re-listen and the manager silently kept handling + // the OLD event name, so a caller whose backend renamed it got nothing and + // no error. + expect( + echo.spy.channels['private-App.Models.User.u1']!.handlers.keys, + contains('notification.pushed'), + ); + }); }); group('NotificationManager polling under realtime', () {