import 'dart:async'; import '../media/ids.dart'; import 'package:flutter/foundation.dart'; import 'package:flutter/widgets.dart'; import 'package:provider/provider.dart'; import '../media/media_item.dart'; import '../mixins/disposable_change_notifier_mixin.dart'; import '../services/watch_state_resolver.dart'; import '../utils/global_key_utils.dart'; import '../utils/watch_state_notifier.dart'; @immutable class HydratedWatchStatePatch { final String globalKey; final WatchStateSnapshot patch; final int updatedAt; final int order; const HydratedWatchStatePatch({ required this.globalKey, required this.patch, required this.updatedAt, required this.order, }); } /// Key for [WatchStateStore]'s observation map: a global key already resolved /// through the client scope its request was issued under. A plain `String` /// would work, but the wrapper keeps the two key spaces from being confused /// with the patch maps' keys at a glance. class _ObservationKey { final String value; const _ObservationKey(this.value); @override bool operator ==(Object other) => other is _ObservationKey && other.value == value; @override int get hashCode => value.hashCode; } class _WatchStatePatchEntry { final WatchStateSnapshot patch; final int updatedAt; final int sequence; final bool isSessionEvent; /// Whether the server had accepted this exact state when the entry was /// recorded. Only an acknowledged entry may be superseded by a later /// authoritative read; an unacknowledged one is a write still owed to the /// server and must outlive any read (#1829). final bool serverAcknowledged; /// Identity used to promote this exact entry once its write settles. /// Promotion never matches by global key: a same-item rewatch may already /// have replaced the entry, and it must not be promoted in its place. final WatchPatchId? patchId; const _WatchStatePatchEntry( this.patch, { required this.updatedAt, required this.sequence, required this.isSessionEvent, this.serverAcknowledged = false, this.patchId, }); _WatchStatePatchEntry acknowledgedAt(int sequence) => _WatchStatePatchEntry( patch, updatedAt: updatedAt, sequence: sequence, isSessionEvent: true, serverAcknowledged: true, patchId: patchId, ); bool isNewerThan(_WatchStatePatchEntry other) { if (updatedAt != other.updatedAt) return updatedAt > other.updatedAt; if (isSessionEvent != other.isSessionEvent) return isSessionEvent; return sequence > other.sequence; } @override bool operator ==(Object other) => other is _WatchStatePatchEntry && other.patch == patch && other.updatedAt == updatedAt && other.sequence == sequence && other.isSessionEvent == isSessionEvent && other.serverAcknowledged == serverAcknowledged && other.patchId == patchId; @override int get hashCode => Object.hash(patch, updatedAt, sequence, isSessionEvent, serverAcknowledged, patchId); } /// The single session-local layer for watch-state freshness. /// /// Server fetches remain the source of truth; [MediaItem] snapshots are never /// hand-mutated to reflect watch events. Instead, every watch event lands here /// as a patch, and consumers resolve items at point of use ([apply] / /// [patchForItem]). Resolution is hierarchy-aware: an item's effective patch /// is the newest among its own and its [MediaItem.parentChain] ancestors', so /// marking a show/season reaches every descendant, while a later per-item /// event still overrides an older container mark. class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin { WatchStateStore() { _subscription = WatchStateNotifier().stream.listen(_onWatchStateEvent); // A second, separately owned subscription: promotions are deliberately // not WatchStateEvents, and BaseNotifier carries exactly one typed // stream, so they cannot arrive on the channel above. _promotionSubscription = WatchPatchPromotionNotifier().stream.listen(_onPromotion); } StreamSubscription? _subscription; StreamSubscription? _promotionSubscription; final Map _patches = {}; final Map _hydratedPatches = {}; String? _activeProfileId; Map _activeClientScopesByServer = const {}; int _sequence = 0; /// Highest watermark at which a qualifying authoritative response returned a /// key, under the client scope its request was issued for. /// /// This records *that* the server re-observed an item, never *what* it /// observed. Storing the observed state instead would need full fidelity /// ([WatchStateSnapshot] cannot hold a container's leaf counts) and would /// make two responses sharing a watermark first-response-wins. A watermark /// has neither problem: it is a single int, and [recordObservations] keeps /// the max, so response order cannot matter. final Map<_ObservationKey, int> _observedAt = {}; /// Snapshot the current sequence *before* issuing an authoritative request; /// pass it back to [recordObservations] when that request succeeds. int get observationWatermark => _sequence; /// Identifies this store instance. Watermarks are store-local, but clients — /// and their in-flight requests — outlive a profile switch, so a response /// minted against the previous store must not be applied to this one. final Object _epoch = Object(); Object get observationEpoch => _epoch; _WatchStatePatchEntry? _exactEntryFor(String globalKey, int? observedAt) { final session = _live(_patches[globalKey], observedAt); final hydrated = _hydratedPatches[globalKey]; if (session == null) return hydrated; if (hydrated == null) return session; return session.isNewerThan(hydrated) ? session : hydrated; } /// Drops an acknowledged session entry the server has re-observed since. /// /// Only acknowledged entries are eligible: an unacknowledged one is a write /// still owed to the server, and a read must never retire it. Hydrated /// entries are the persisted owed-write layer and are likewise untouched — /// filtering happens per layer, before [_exactEntryFor] chooses, so a /// suppressed session entry falls back to an older hydrated one instead of /// suppressing both. static _WatchStatePatchEntry? _live(_WatchStatePatchEntry? entry, int? observedAt) { if (entry == null) return null; if (!entry.serverAcknowledged || observedAt == null) return entry; return observedAt >= entry.sequence ? null : entry; } /// Candidate keys for [globalKey], in the order [_entryFor] consults them: /// the active client scope first, then the public fallback. List _candidateKeys(String globalKey) { final parsed = parseGlobalKey(globalKey); if (parsed != null) { final scoped = _activeClientScopesByServer[parsed.serverId]; if (scoped != null && scoped.isNotEmpty) { return [buildGlobalKey(ServerId(scoped), parsed.ratingKey), globalKey]; } } return [globalKey]; } _WatchStatePatchEntry? _entryFor(String globalKey, [int? observedAt]) { for (final key in _candidateKeys(globalKey)) { final entry = _exactEntryFor(key, observedAt); if (entry != null) return entry; } return null; } /// The watermark at which [globalKey] was last authoritatively observed. /// /// Resolved through the *primary* candidate only — the active client scope /// when there is one, else the public key. Deliberately not a max over both: /// on a user-scoped backend the public server id is shared between users, so /// letting a public observation satisfy a scoped patch would suppress /// another user's watch state. int? _observationFor(String globalKey) => _observedAt[_ObservationKey(_candidateKeys(globalKey).first)]; @visibleForTesting WatchStateSnapshot? patchForGlobalKey(String globalKey) => _entryFor(globalKey, _observationFor(globalKey))?.patch; WatchStateSnapshot? patchForItem(MediaItem item) { if (_patches.isEmpty && _hydratedPatches.isEmpty) return null; // One barrier for the whole resolution: an authoritative read of *this* // item already incorporates any container mark that preceded it, so the // ancestor candidates are judged against the item's own observation too. // Suppressing only the item's own entry would let an older season patch // win and render something worse than either value (#1829 review). final observedAt = _observationFor(item.globalKey); var best = _entryFor(item.globalKey, observedAt); if (item.parentChain.isNotEmpty) { final serverId = serverIdOrNull(item.serverId); for (final parentId in item.parentChain) { // Mirror MediaItem.globalKey's bare-id fallback when serverId is missing. final entry = _entryFor(serverId != null ? buildGlobalKey(serverId, parentId) : parentId, observedAt); if (entry != null && (best == null || entry.isNewerThan(best))) best = entry; } } return best?.patch; } /// Record that a successful authoritative response returned [rows]. /// /// [watermark] must be [observationWatermark] captured immediately before /// the network call, and [epoch] the [observationEpoch] of the store that /// captured it. Entries recorded *during* the request keep a higher /// sequence and survive, so a local action taken mid-flight always wins. /// /// Each row carries the immutable cache scope of the client that fetched /// it: on a user-scoped backend the public server id is shared, so a row /// must only ever reconcile the scope it actually came from. void recordObservations( Iterable<({MediaItem item, String? clientScope})> rows, { required int watermark, required Object epoch, }) { // Watermarks are store-local, but clients — and their in-flight requests // — outlive a profile switch, so a response minted against the previous // store carries a sequence that means nothing here. if (!identical(epoch, _epoch)) return; if (_patches.isEmpty) return; var changed = false; for (final row in rows) { final item = row.item; // An observation is only ever consulted to suppress a patch, so keep // one only while some patch could still apply to this item — its own, // or an ancestor's, which the barrier in [patchForItem] also judges // against this key. That bounds the map by the items the user acted // on rather than by everything ever fetched. if (!_hasSuppressibleEntry(item)) continue; final scope = row.clientScope; final key = _ObservationKey( scope != null && scope.isNotEmpty && scope != item.serverId ? buildGlobalKey(ServerId(scope), item.id) : item.globalKey, ); final existing = _observedAt[key]; if (existing != null && existing >= watermark) continue; _observedAt[key] = watermark; changed = true; } if (changed) safeNotifyListeners(); } bool _hasSuppressibleEntry(MediaItem item) { for (final key in _candidateKeys(item.globalKey)) { if (_patches[key]?.serverAcknowledged ?? false) return true; } if (item.parentChain.isEmpty) return false; final serverId = serverIdOrNull(item.serverId); for (final parentId in item.parentChain) { final parentKey = serverId != null ? buildGlobalKey(serverId, parentId) : parentId; for (final key in _candidateKeys(parentKey)) { if (_patches[key]?.serverAcknowledged ?? false) return true; } } return false; } MediaItem apply(MediaItem item) { return applyPatch(item, patchForItem(item)); } List applyAll(List items) { if (_patches.isEmpty && _hydratedPatches.isEmpty) return items; return [for (final item in items) apply(item)]; } static MediaItem applyPatch(MediaItem item, WatchStateSnapshot? patch) => patch == null ? item : patch.apply(item); void setActiveProfileId(String? profileId) { if (_activeProfileId == profileId) return; _activeProfileId = profileId; if (_patches.isEmpty && _hydratedPatches.isEmpty && _observedAt.isEmpty) return; _patches.clear(); _hydratedPatches.clear(); // Observations are only meaningful against the patches they suppress. _observedAt.clear(); safeNotifyListeners(); } void setActiveClientScopesByServer(Map scopes) { final normalized = { for (final entry in scopes.entries) if (entry.value != null && entry.value!.isNotEmpty && entry.value != entry.key) entry.key: entry.value, }; if (mapEquals(_activeClientScopesByServer, normalized)) return; _activeClientScopesByServer = Map.unmodifiable(normalized); if (_patches.isNotEmpty || _hydratedPatches.isNotEmpty) safeNotifyListeners(); } /// Replace the persisted local-action layer without disturbing newer /// session events. Timestamps preserve freshness across item/ancestor keys. void setHydratedPatches(Iterable patches) { final next = {}; for (final hydrated in patches) { final candidate = _WatchStatePatchEntry( hydrated.patch, updatedAt: hydrated.updatedAt, sequence: hydrated.order, isSessionEvent: false, ); final existing = next[hydrated.globalKey]; if (existing == null || candidate.isNewerThan(existing)) { next[hydrated.globalKey] = candidate; } } if (mapEquals(_hydratedPatches, next)) return; _hydratedPatches ..clear() ..addAll(next); safeNotifyListeners(); } void _onWatchStateEvent(WatchStateEvent event) { final snapshot = WatchStateResolver.fromEvent(event); if (snapshot.isEmpty) return; final activeScope = _activeClientScopesByServer[event.serverId]; final eventScope = event.cacheServerId; if (activeScope != null && activeScope.isNotEmpty && eventScope != null && eventScope.isNotEmpty && eventScope != event.serverId && eventScope != activeScope) { return; } final resolvedScope = activeScope != null && activeScope.isNotEmpty ? activeScope : eventScope; final key = resolvedScope != null && resolvedScope.isNotEmpty && resolvedScope != event.serverId ? buildGlobalKey(ServerId(resolvedScope), event.itemId) : event.globalKey; _patches[key] = _WatchStatePatchEntry( snapshot, updatedAt: DateTime.now().millisecondsSinceEpoch, sequence: ++_sequence, isSessionEvent: true, serverAcknowledged: event.serverAcknowledged, patchId: event.patchId, ); safeNotifyListeners(); } /// The write behind [promotion] has settled, so its entry may now be /// superseded by a later authoritative read. /// /// Matching is by patch identity, never by key: an unknown id — because a /// newer action replaced the entry, or because the process restarted — is a /// deliberate no-op. The fresh sequence matters as much as the flag: an /// observation captured *before* the write landed must not suppress the /// entry it predates. void _onPromotion(WatchPatchPromotion promotion) { final sessionKey = _keyForPatchId(_patches, promotion.patchId); if (sessionKey != null) { _patches[sessionKey] = _patches[sessionKey]!.acknowledgedAt(++_sequence); safeNotifyListeners(); return; } // A hydrated entry is the persisted owed-write layer. Once the write has // landed it no longer belongs there, and leaving it would make it // immortal, since suppression only ever considers session entries. Move // it — but never over a newer session entry on the same key, which would // defeat the exact-identity rule promotion exists for. final hydratedKey = _keyForPatchId(_hydratedPatches, promotion.patchId); if (hydratedKey == null) return; final promoted = _hydratedPatches.remove(hydratedKey)!; final existing = _patches[hydratedKey]; if (existing == null || existing.patchId == promotion.patchId) { _patches[hydratedKey] = promoted.acknowledgedAt(++_sequence); } safeNotifyListeners(); } static String? _keyForPatchId(Map entries, WatchPatchId patchId) { for (final entry in entries.entries) { if (entry.value.patchId == patchId) return entry.key; } return null; } @override void dispose() { _subscription?.cancel(); _subscription = null; _promotionSubscription?.cancel(); _promotionSubscription = null; super.dispose(); } } /// Point-of-use watch-state resolution. All fall back to the item as-is when /// no [WatchStateStore] is in the tree (tests, isolated subtrees). extension WatchStateResolution on BuildContext { /// Build-time resolution: subscribes this context to the item's effective /// patch, so the widget rebuilds when a newer event lands for it (or an /// ancestor). Use in `build`. MediaItem withFreshWatchState(MediaItem item) { try { final patch = select((store) => store.patchForItem(item)); return WatchStateStore.applyPatch(item, patch); } on ProviderNotFoundException { return item; } } /// Point-in-time resolution for handlers and non-build code paths. MediaItem readFreshWatchState(MediaItem item) { try { return read().apply(item); } on ProviderNotFoundException { return item; } } List readFreshWatchStateAll(List items) { try { return read().applyAll(items); } on ProviderNotFoundException { return items; } } }