// Outbound mutation queue for offline-resilient REST calls. // // Controllers (LikesController, MyQuarantineController, the add-to- // playlist sheet, the Discover request flow, the Requests cancel // action) write their optimistic local state to drift first and try // the corresponding REST call. On failure they enqueue here. The // replayer drains the queue when connectivity comes back; the user's // intent persists across network loss without rolling back their // visible action. // // Why this is separate from SyncController: sync ingests AUTHORITATIVE // SERVER state into drift. The mutation queue carries USER INTENT // outbound to the server. They flow in opposite directions and // shouldn't be conflated — sync wins on conflict (server is the // source of truth), and a mutation that fails forever (5 attempts) // is dropped on the trust that next sync will reconcile drift to // match the server's actual state. // // Drop semantics: 5 attempts then drop. The drop is intentional — // holding onto a forever-failing mutation just delays sync's // reconciliation. The user's optimistic drift state will diverge // from server, and the next library_changes delta corrects it. import 'dart:async'; import 'dart:convert'; import 'package:dio/dio.dart' show DioException, DioExceptionType; import 'package:drift/drift.dart'; import 'package:flutter/foundation.dart' show debugPrint; import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../api/endpoints/discover.dart'; import '../api/endpoints/likes.dart'; import '../library/library_providers.dart' show dioProvider; import '../likes/likes_provider.dart' show likesApiProvider; import '../models/lidarr.dart' show LidarrRequestKind; import '../playlists/playlists_provider.dart' show playlistsApiProvider; import '../quarantine/quarantine_provider.dart' show quarantineApiProvider; import '../requests/requests_provider.dart' show requestsApiProvider; import 'audio_cache_manager.dart' show appDbProvider; import 'db.dart'; /// Stable kind constants. Each controller emits one of these via /// MutationQueue.enqueue; the replayer dispatches by kind. class MutationKinds { MutationKinds._(); static const likeAdd = 'like.add'; static const likeRemove = 'like.remove'; static const quarantineFlag = 'quarantine.flag'; static const quarantineUnflag = 'quarantine.unflag'; static const playlistAppend = 'playlist.append'; static const requestCreate = 'request.create'; static const requestCancel = 'request.cancel'; } class MutationQueue { MutationQueue(this._ref); final Ref _ref; /// Persist a mutation for replay. Caller has already done the /// optimistic local mutation (drift write, in-memory state update) /// — this only records the REST call that needs to fire eventually. Future enqueue(String kind, Map payload) async { final db = _ref.read(appDbProvider); await db.into(db.cachedMutations).insert( CachedMutationsCompanion.insert( kind: kind, payload: jsonEncode(payload), ), ); // Nudge the replayer in case we're online right now. unawaited(_ref.read(mutationReplayerProvider).drain()); } /// Count of pending mutations (attempts < 5). Diagnostic / future /// UI surface for a "syncing N pending changes" indicator. Future pendingCount() async { final db = _ref.read(appDbProvider); final rows = await db.customSelect( 'SELECT COUNT(*) AS c FROM cached_mutations WHERE attempts < 5', ).get(); return rows.first.read('c'); } } final mutationQueueProvider = Provider((ref) => MutationQueue(ref)); class MutationReplayer { MutationReplayer(this._ref); final Ref _ref; Timer? _initialTimer; Timer? _intervalTimer; bool _running = false; bool _disposed = false; /// Periodic poll cadence. The connectivity listener catches most /// transitions; this is the belt-and-suspenders for cases where /// connectivity_plus misses a state change (e.g. flaky Wi-Fi). static const _interval = Duration(minutes: 1); /// Max replay attempts before we drop a mutation. 5 is enough to /// span a few reconnection cycles without holding onto a forever- /// broken request. See file header for the drop rationale. static const _maxAttempts = 5; void start() { // Drain shortly after launch in case there are queued mutations // from a prior session that died offline. 3s lets the auth + // server-url + dio providers finish their async warmup. _initialTimer = Timer(const Duration(seconds: 3), drain); // Periodic ticker is the only reconnect signal. We previously had // a ref.listen(connectivityProvider, …) edge trigger here, but // subscribing at start-time eagerly mounted the connectivity // StreamProvider and leaked its initial-timeout Timer through // tests that never reach the auth state. The drain() loop itself // gates on connectivity via the .future read below. _intervalTimer = Timer.periodic(_interval, (_) => drain()); } /// Walk the queue oldest-first, replaying each mutation. Stops on /// any transient network error (we'll retry on the next tick); a /// permanent error (404 etc.) increments the attempt counter and /// moves on to the next mutation. This is public so MutationQueue. /// enqueue can trigger an immediate attempt after writing. Future drain() async { if (_disposed || _running) return; final online = await _ref .read(connectivityProvider.future) .timeout(const Duration(seconds: 3), onTimeout: () => true); if (!online) return; _running = true; try { while (!_disposed) { final db = _ref.read(appDbProvider); final next = await (db.select(db.cachedMutations) ..where((t) => t.attempts.isSmallerThanValue(_maxAttempts)) ..orderBy([(t) => OrderingTerm.asc(t.createdAt)]) ..limit(1)) .getSingleOrNull(); if (next == null) break; final handler = _handlers[next.kind]; if (handler == null) { debugPrint('mutation_replayer: unknown kind ${next.kind}, dropping'); await (db.delete(db.cachedMutations) ..where((t) => t.id.equals(next.id))) .go(); continue; } Map payload; try { payload = jsonDecode(next.payload) as Map; } catch (e) { debugPrint('mutation_replayer: bad payload for ${next.kind}: $e'); await (db.delete(db.cachedMutations) ..where((t) => t.id.equals(next.id))) .go(); continue; } try { await handler(_ref, payload); await (db.delete(db.cachedMutations) ..where((t) => t.id.equals(next.id))) .go(); } on DioException catch (e) { await (db.update(db.cachedMutations) ..where((t) => t.id.equals(next.id))) .write(CachedMutationsCompanion( attempts: Value(next.attempts + 1), lastAttemptAt: Value(DateTime.now()), )); if (_isTransient(e)) { // Network is flaky again — bail and try the whole queue // later. The current mutation will be retried next tick. break; } // Permanent failure (4xx/5xx with a real response). Leave // the row with bumped attempts and move on to the next so // one bad mutation doesn't block the others. } catch (e, st) { debugPrint('mutation_replayer: unexpected error: $e\n$st'); await (db.update(db.cachedMutations) ..where((t) => t.id.equals(next.id))) .write(CachedMutationsCompanion( attempts: Value(next.attempts + 1), lastAttemptAt: Value(DateTime.now()), )); } } } finally { _running = false; } } bool _isTransient(DioException e) { switch (e.type) { case DioExceptionType.connectionError: case DioExceptionType.connectionTimeout: case DioExceptionType.sendTimeout: case DioExceptionType.receiveTimeout: return true; default: return false; } } void dispose() { _disposed = true; _initialTimer?.cancel(); _intervalTimer?.cancel(); } } final mutationReplayerProvider = Provider((ref) { final r = MutationReplayer(ref); ref.onDispose(r.dispose); r.start(); return r; }); /// Type signature for kind-specific replay handlers. typedef _Handler = Future Function(Ref, Map); /// Kind → handler dispatch. Each handler decodes its payload and /// re-fires the original REST call. Throws on failure (caught by the /// replayer's drain loop above). /// /// Payload shapes (see also enqueue call sites for the writer side): /// * like.add / like.remove: {'kind': 'track'|'album'|'artist', 'id': uuid} /// * quarantine.flag: {'trackId': uuid, 'reason': str, 'notes': str} /// * quarantine.unflag: {'trackId': uuid} /// * playlist.append: {'playlistId': uuid, 'trackIds': [uuid, …]} /// * request.create: {full createRequest args; see DiscoverApi} /// * request.cancel: {'id': uuid} final Map _handlers = { MutationKinds.likeAdd: (ref, p) async { final api = await ref.read(likesApiProvider.future); await api.like(_likeKindFromString(p['kind'] as String), p['id'] as String); }, MutationKinds.likeRemove: (ref, p) async { final api = await ref.read(likesApiProvider.future); await api.unlike( _likeKindFromString(p['kind'] as String), p['id'] as String); }, MutationKinds.quarantineFlag: (ref, p) async { final api = await ref.read(quarantineApiProvider.future); await api.flag( p['trackId'] as String, p['reason'] as String, notes: (p['notes'] as String?) ?? '', ); }, MutationKinds.quarantineUnflag: (ref, p) async { final api = await ref.read(quarantineApiProvider.future); await api.unflag(p['trackId'] as String); }, MutationKinds.playlistAppend: (ref, p) async { final api = await ref.read(playlistsApiProvider.future); final ids = (p['trackIds'] as List).cast(); await api.appendTracks(p['playlistId'] as String, ids); }, MutationKinds.requestCreate: (ref, p) async { final dio = await ref.read(dioProvider.future); final api = DiscoverApi(dio); await api.createRequest( kind: _lidarrKindFromString(p['kind'] as String), artistMbid: p['artistMbid'] as String, artistName: p['artistName'] as String, albumMbid: p['albumMbid'] as String?, albumTitle: p['albumTitle'] as String?, trackMbid: p['trackMbid'] as String?, trackTitle: p['trackTitle'] as String?, ); }, MutationKinds.requestCancel: (ref, p) async { final api = await ref.read(requestsApiProvider.future); await api.cancel(p['id'] as String); }, }; LikeKind _likeKindFromString(String s) => switch (s) { 'album' => LikeKind.album, 'artist' => LikeKind.artist, _ => LikeKind.track, }; LidarrRequestKind _lidarrKindFromString(String s) => switch (s) { 'album' => LidarrRequestKind.album, 'track' => LidarrRequestKind.track, _ => LidarrRequestKind.artist, };