minstrel/flutter_client/lib/cache/mutation_queue.dart

// 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/events.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 'connectivity_provider.dart';
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';
  static const playOffline = 'play.offline';
}

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<void> enqueue(String kind, Map<String, dynamic> 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<int> 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<int>('c');
  }
}

final mutationQueueProvider =
    Provider<MutationQueue>((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<void> 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<String, dynamic> payload;
        try {
          payload = jsonDecode(next.payload) as Map<String, dynamic>;
        } 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<MutationReplayer>((ref) {
  final r = MutationReplayer(ref);
  ref.onDispose(r.dispose);
  r.start();
  return r;
});

/// Type signature for kind-specific replay handlers.
typedef _Handler = Future<void> Function(Ref, Map<String, dynamic>);

/// 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}
/// * play.offline: {'trackId': uuid, 'clientId': str, 'at': iso8601,
///   'durationPlayedMs': int, 'source'?: 'for_you'|'discover'}
final Map<String, _Handler> _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<String>();
    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);
  },
  MutationKinds.playOffline: (ref, p) async {
    final dio = await ref.read(dioProvider.future);
    final api = EventsApi(dio);
    await api.playOffline(
      trackId: p['trackId'] as String,
      clientId: p['clientId'] as String,
      atIso: p['at'] as String,
      durationPlayedMs: (p['durationPlayedMs'] as num).toInt(),
      source: p['source'] 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,
    };