import 'dart:async'; import 'package:connectivity_plus/connectivity_plus.dart'; import 'package:sqflite/sqflite.dart'; // SQLite table and column names — kept as constants to avoid typos and make // schema migrations easy to spot. const _kTable = 'progress_queue'; const _kColId = 'id'; const _kColMediaId = 'media_id'; const _kColPositionSeconds = 'position_seconds'; const _kColFinished = 'finished'; const _kColQueuedAt = 'queued_at'; // Database schema version. Bump when columns change so onUpgrade fires. const _kDbVersion = 1; // Database filename stored in the default sqflite databases path. const _kDbName = 'progress_queue.db'; // --------------------------------------------------------------------------- // ProgressSyncClient — narrow interface (ISP) // --------------------------------------------------------------------------- /// Narrow interface for the single API operation that [ProgressQueue] needs. /// /// Interface Segregation: [ProgressQueue] depends only on /// [batchUpdateProgress], not on the full [PlayerApiClient] surface. /// Production code passes a [PlayerApiClient] (which implements this); /// tests can provide a lightweight stub without subclassing the entire client. abstract class ProgressSyncClient { /// Submits a batch of progress updates to the server. /// /// Each map must include `media_id`, `position_seconds`, and `observed_at`. Future batchUpdateProgress(List> updates); } // --------------------------------------------------------------------------- // ProgressQueueBase — abstract lifecycle interface (LSP + DIP) // --------------------------------------------------------------------------- /// Abstract contract for an offline-capable progress queue. /// /// Callers (provider, player screens) depend on this interface rather than the /// concrete [ProgressQueue] class (Dependency Inversion). Alternative /// implementations (in-memory, no-op) are substitutable without breaking /// callers (Liskov Substitution). abstract class ProgressQueueBase { /// Opens the backing store and subscribes to connectivity changes. /// /// Must be called once before [enqueue]. Future init(); /// Persists a playback-progress update and, if online, flushes immediately. Future enqueue(int mediaId, double positionSeconds, {bool finished = false}); /// Cancels subscriptions and closes the backing store. Future dispose(); } // --------------------------------------------------------------------------- // ProgressUpdate value object // --------------------------------------------------------------------------- /// Immutable record of a single playback-progress update. /// /// Used both as a value object passed from the player screens and as an /// internal DTO deserialised from the SQLite row. Keeping it in this file /// avoids leaking a "models" dependency on the queue's persistence layer. class ProgressUpdate { const ProgressUpdate({ required this.mediaId, required this.positionSeconds, this.finished = false, required this.queuedAt, this.rowId, }); final int mediaId; final double positionSeconds; final bool finished; /// Wall-clock time the update was created (ISO-8601 UTC string stored in DB). /// Used as the `observed_at` field in the batch request so the server applies /// updates in chronological order. final String queuedAt; /// Non-null after the row has been persisted; null for newly constructed /// updates that have not been written to the DB yet. final int? rowId; /// Converts this update to the JSON shape expected by /// [ProgressSyncClient.batchUpdateProgress]. Map toBatchMap() => { 'media_id': mediaId, 'position_seconds': positionSeconds, 'observed_at': queuedAt, }; } // --------------------------------------------------------------------------- // ProgressQueue // --------------------------------------------------------------------------- /// Offline-capable progress queue backed by SQLite. /// /// Responsibilities (Single Responsibility: one per bullet): /// - Persist [enqueue] calls to a local SQLite table so updates survive /// process restarts while the device is offline. /// - Watch network connectivity via [Connectivity] and trigger a flush /// automatically when the device goes from offline to online. /// - Flush pending rows by calling [ProgressSyncClient.batchUpdateProgress]; /// remove successfully sent rows and retain any that fail (for retry). /// /// Design notes: /// - No Flutter imports — this is a pure-Dart service (can be unit tested /// without a widget tree). /// - [ProgressSyncClient] is injected (Interface Segregation + Dependency /// Inversion); [ProgressQueue] only depends on the one method it uses. /// - [databaseFactory] is injected so tests can supply an in-memory opener /// without touching the filesystem (Dependency Inversion). /// - [Database] may also be injected directly via [db] for tests that have /// already opened a connection. /// - Concurrent flush is prevented with [_isFlushing]; a second connectivity /// event while a flush is in progress is silently ignored — the flush will /// drain all rows anyway. class ProgressQueue implements ProgressQueueBase { /// Creates the queue. /// /// [apiClient] must implement [ProgressSyncClient]; in production this is /// a [PlayerApiClient]. Tests can pass a lightweight stub. /// /// [databaseFactory] is an optional factory for opening the SQLite database. /// When null, [init] calls [openDatabase] with the default on-disk path. /// Inject a custom factory in tests to get an in-memory database without /// touching the filesystem (Dependency Inversion). /// /// [db] is an already-opened [Database]; when non-null it takes precedence /// over [databaseFactory] and no additional open call is made. /// /// [connectivity] is optional; when null the default [Connectivity()] is /// used in production. Pass a fake in tests. ProgressQueue({ required ProgressSyncClient apiClient, Future Function()? databaseFactory, Database? db, Connectivity? connectivity, }) : _apiClient = apiClient, _databaseFactory = databaseFactory, _db = db, _connectivity = connectivity ?? Connectivity(); final ProgressSyncClient _apiClient; // Optional factory for opening the on-disk database; null means use the // built-in [_openDatabase] helper which calls sqflite's openDatabase(). final Future Function()? _databaseFactory; final Connectivity _connectivity; // Non-null after [init] has been called. Database? _db; // Guards against concurrent flush operations. bool _isFlushing = false; // Holds the in-flight flush future so [dispose] can await it before closing // the database, preventing "database_closed" errors on shutdown. Future? _flushFuture; // Subscription to connectivity changes; cancelled in [dispose]. StreamSubscription>? _connectivitySub; // --------------------------------------------------------------------------- // Lifecycle // --------------------------------------------------------------------------- /// Opens the SQLite database (if not already provided) and subscribes to /// connectivity changes. /// /// Must be called once before any other method. Safe to call multiple times /// (subsequent calls are no-ops if the DB is already open). /// /// The database is obtained from the injected [_databaseFactory] when /// supplied, falling back to [_openDatabase] which calls sqflite's /// [openDatabase] with the default on-disk path. @override Future init() async { _db ??= await (_databaseFactory?.call() ?? _openDatabase()); _subscribeToConnectivity(); } /// Cancels the connectivity subscription and closes the database. /// /// Awaits any in-flight flush before closing the DB so that a concurrent /// flush does not attempt to use the database after it has been closed /// (prevents "database_closed" errors during app shutdown or test teardown). @override Future dispose() async { await _connectivitySub?.cancel(); _connectivitySub = null; // Wait for any ongoing flush to finish before closing the database. // Ignore errors from the in-flight flush — they are already handled inside // [_flush] via try/finally; swallowing here avoids double-reporting. await _flushFuture?.catchError((_) {}); await _db?.close(); _db = null; } // --------------------------------------------------------------------------- // Public API // --------------------------------------------------------------------------- /// Persists a progress update locally and, if the device is currently online, /// triggers an immediate flush. /// /// Fire-and-forget in the player screens: any DB write failure is swallowed /// so a storage error never interrupts playback. @override Future enqueue( int mediaId, double positionSeconds, { bool finished = false, }) async { final db = _db; if (db == null) return; // Defensive: init not called. final now = DateTime.now().toUtc().toIso8601String(); await db.insert(_kTable, { _kColMediaId: mediaId, _kColPositionSeconds: positionSeconds, _kColFinished: finished ? 1 : 0, _kColQueuedAt: now, }); // Opportunistic online flush: attempt immediately on enqueue so that // updates sent while online bypass the DB round-trip latency. // Errors are swallowed — the row is already persisted so the next // connectivity event will retry. final results = await _connectivity.checkConnectivity(); if (_isOnline(results)) { await _flush().catchError((_) {}); } } // --------------------------------------------------------------------------- // Internal: flush // --------------------------------------------------------------------------- /// Sends all queued rows to the server via [batchUpdateProgress]. /// /// Rows that are successfully sent are deleted from the DB. Rows that fail /// (e.g., the server returns an error for a specific item) are retained for /// the next flush. The entire batch succeeds or fails atomically from the /// client perspective — if the call throws, no rows are deleted. /// /// [_isFlushing] prevents re-entrant flushes. The flag is cleared in a /// `finally` block so a thrown exception never permanently blocks flushing. /// /// The future is stored in [_flushFuture] so [dispose] can await it before /// closing the database, preventing use-after-close crashes on shutdown. Future _flush() { if (_isFlushing) return Future.value(); _isFlushing = true; _flushFuture = _flushPendingRows().whenComplete(() { _isFlushing = false; _flushFuture = null; }); return _flushFuture!; } /// Loads pending rows, sends them, and removes the ones that succeeded. /// /// Extracted from [_flush] to keep each method under ~30 lines and make the /// "load → send → delete" pipeline independently readable. Future _flushPendingRows() async { final db = _db; if (db == null) return; final rows = await db.query( _kTable, orderBy: '$_kColQueuedAt ASC', ); if (rows.isEmpty) return; final updates = rows.map(_rowToUpdate).toList(); // Build the batch payload for the server. final payload = updates.map((u) => u.toBatchMap()).toList(); // Send — if this throws (network error, server 5xx) we skip deletion and // let the next connectivity event retry. await _apiClient.batchUpdateProgress(payload); // Delete the rows that were just sent successfully. final ids = updates.map((u) => u.rowId!).toList(); await _deleteRows(db, ids); } // --------------------------------------------------------------------------- // Internal: connectivity // --------------------------------------------------------------------------- /// Subscribes to connectivity changes and flushes when online is detected. /// /// The subscription is only set up once; subsequent [init] calls are no-ops /// because [_connectivitySub] is already non-null. /// /// Errors from [_flush] are swallowed inside the listener — the flush /// already handles its own error recovery (rows retained on failure) and /// an unhandled stream error would tear down the subscription. void _subscribeToConnectivity() { _connectivitySub ??= _connectivity.onConnectivityChanged.listen( (results) async { if (_isOnline(results)) { await _flush().catchError((_) {}); } }, ); } // --------------------------------------------------------------------------- // Internal: helpers // --------------------------------------------------------------------------- /// Opens (or creates) the on-disk SQLite database and runs migrations. Future _openDatabase() { return openDatabase( _kDbName, version: _kDbVersion, onCreate: (db, version) => _createSchema(db), ); } /// Creates the progress_queue table on first run. Future _createSchema(Database db) { return db.execute(''' CREATE TABLE $_kTable ( $_kColId INTEGER PRIMARY KEY AUTOINCREMENT, $_kColMediaId INTEGER NOT NULL, $_kColPositionSeconds REAL NOT NULL, $_kColFinished INTEGER NOT NULL DEFAULT 0, $_kColQueuedAt TEXT NOT NULL ) '''); } /// Converts a raw SQLite row map into a [ProgressUpdate]. ProgressUpdate _rowToUpdate(Map row) { return ProgressUpdate( rowId: row[_kColId] as int, mediaId: row[_kColMediaId] as int, positionSeconds: (row[_kColPositionSeconds] as num).toDouble(), finished: (row[_kColFinished] as int) != 0, queuedAt: row[_kColQueuedAt] as String, ); } /// Deletes rows with the given [ids] from the queue table. /// /// Uses a single DELETE … WHERE id IN (…) statement for efficiency. Future _deleteRows(Database db, List ids) async { if (ids.isEmpty) return; final placeholders = List.filled(ids.length, '?').join(', '); await db.rawDelete( 'DELETE FROM $_kTable WHERE $_kColId IN ($placeholders)', ids, ); } /// Returns `true` when at least one connectivity result indicates an active /// network interface (WiFi, mobile, ethernet, or VPN). /// /// [ConnectivityResult.none] is the only value treated as offline. bool _isOnline(List results) { return results.any((r) => r != ConnectivityResult.none); } }