From 1b66b5f573c8de883947d712057ff85f5513e8ae Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Thu, 21 May 2026 23:50:26 +0300 Subject: Implement offline progress queue with sqflite and connectivity_plus (3b) Introduces ProgressQueue backed by SQLite for offline-capable progress tracking, with ProgressQueueBase (LSP+DIP), databaseFactory injection (DIP), and ProgressSyncClient narrow interface (ISP). Co-Authored-By: Claude Sonnet 4.6 --- player-android/lib/api/player_api_client.dart | 9 +- player-android/lib/main.dart | 19 +- .../lib/providers/progress_queue_provider.dart | 27 ++ .../lib/screens/audio_player_screen.dart | 37 +- .../lib/screens/video_player_screen.dart | 41 ++- player-android/lib/services/progress_queue.dart | 378 +++++++++++++++++++++ player-android/pubspec.lock | 42 ++- player-android/pubspec.yaml | 9 + .../test/screens/audio_player_screen_test.dart | 26 ++ .../test/screens/video_player_screen_test.dart | 30 ++ .../test/services/progress_queue_test.dart | 332 ++++++++++++++++++ 11 files changed, 921 insertions(+), 29 deletions(-) create mode 100644 player-android/lib/providers/progress_queue_provider.dart create mode 100644 player-android/lib/services/progress_queue.dart create mode 100644 player-android/test/services/progress_queue_test.dart diff --git a/player-android/lib/api/player_api_client.dart b/player-android/lib/api/player_api_client.dart index a1f4fe9..0094a90 100644 --- a/player-android/lib/api/player_api_client.dart +++ b/player-android/lib/api/player_api_client.dart @@ -3,6 +3,7 @@ import 'dart:typed_data'; import 'package:dio/dio.dart'; import '../models/models.dart'; +import '../services/progress_queue.dart' show ProgressSyncClient; /// High-level API surface that maps 1-to-1 with the player-server REST API /// (see player-server/docs/api.md for the authoritative contract). @@ -14,9 +15,12 @@ import '../models/models.dart'; /// In production, create the [Dio] via [DioClient] which wires up the auth /// and 401-redirect interceptors. In tests, pass a plain or mocked [Dio]. /// +/// Implements [ProgressSyncClient] so it can be injected directly into +/// [ProgressQueue] without exposing the full API surface (Interface Segregation). +/// /// Concrete implementations of the stub methods will be added incrementally as /// features are built. -class PlayerApiClient { +class PlayerApiClient implements ProgressSyncClient { /// Creates a client backed by [dio]. /// /// Prefer creating [dio] via [DioClient] in production to get bearer-token @@ -359,6 +363,9 @@ class PlayerApiClient { /// Designed for offline clients that accumulate updates while disconnected /// and sync on reconnect. Each entry must include [mediaId], /// [positionSeconds], and [observedAt] (ISO-8601 UTC string). + /// + /// Implements [ProgressSyncClient.batchUpdateProgress]. + @override Future batchUpdateProgress( List> updates, ) => diff --git a/player-android/lib/main.dart b/player-android/lib/main.dart index b32c8e7..99e85cb 100644 --- a/player-android/lib/main.dart +++ b/player-android/lib/main.dart @@ -4,6 +4,7 @@ import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:just_audio/just_audio.dart'; import 'providers/audio_handler_provider.dart'; +import 'providers/progress_queue_provider.dart'; import 'router.dart'; import 'services/audio_handler.dart'; @@ -42,12 +43,20 @@ void main() async { ), ); + // Create the ProviderScope first so we can read providers before runApp. + // The scope is then passed to PlayerAndroidApp so it is the single root. + final container = ProviderContainer( + overrides: [audioHandlerProvider.overrideWithValue(handler)], + ); + + // Initialise the offline progress queue (opens SQLite DB, subscribes to + // connectivity). Must be done before any player screen opens so that the + // queue is ready to accept enqueue calls immediately. + await container.read(progressQueueProvider).init(); + runApp( - ProviderScope( - // Override the provider with the concrete handler instance so that every - // widget and provider that reads [audioHandlerProvider] gets the same - // singleton without going through a global variable. - overrides: [audioHandlerProvider.overrideWithValue(handler)], + UncontrolledProviderScope( + container: container, child: const PlayerAndroidApp(), ), ); diff --git a/player-android/lib/providers/progress_queue_provider.dart b/player-android/lib/providers/progress_queue_provider.dart new file mode 100644 index 0000000..e138f20 --- /dev/null +++ b/player-android/lib/providers/progress_queue_provider.dart @@ -0,0 +1,27 @@ +import 'package:flutter_riverpod/flutter_riverpod.dart'; + +import '../services/progress_queue.dart'; +import 'api_client_provider.dart'; + +/// Provides the singleton [ProgressQueueBase] for the whole application. +/// +/// The provider return type is [ProgressQueueBase] (not the concrete +/// [ProgressQueue]) so that callers depend only on the abstract interface +/// (Dependency Inversion). Tests override this provider with a lightweight +/// [ProgressQueueBase] implementation that never opens a real database. +/// +/// [ProgressQueueBase.init] must be called before the queue is useful; this is +/// done once in `main()` after [ProviderScope] is set up so the DB is open +/// and the connectivity subscription is active before any player screen opens. +/// +/// Dependency Inversion: [PlayerApiClient] is injected from [apiClientProvider] +/// via its [ProgressSyncClient] interface so [ProgressQueue] has no knowledge +/// of Dio or concrete HTTP classes. The [databaseFactory] is left null so +/// [ProgressQueue] opens the default on-disk database; tests override the +/// entire provider to avoid touching the filesystem. +final progressQueueProvider = Provider((ref) { + final apiClient = ref.watch(apiClientProvider); + // databaseFactory is null → ProgressQueue.init() opens the on-disk DB. + // Connectivity() is created lazily inside ProgressQueue; no extra wiring needed. + return ProgressQueue(apiClient: apiClient); +}); diff --git a/player-android/lib/screens/audio_player_screen.dart b/player-android/lib/screens/audio_player_screen.dart index 58affa4..9e28062 100644 --- a/player-android/lib/screens/audio_player_screen.dart +++ b/player-android/lib/screens/audio_player_screen.dart @@ -8,7 +8,9 @@ import '../api/dio_client.dart'; import '../api/player_api_client.dart'; import '../providers/api_client_provider.dart'; import '../providers/audio_handler_provider.dart'; +import '../providers/progress_queue_provider.dart'; import '../services/audio_handler.dart'; +import '../services/progress_queue.dart'; // How often progress updates are emitted to the server while playing. // Mirrors VideoPlayerScreen._kProgressInterval exactly. @@ -153,8 +155,11 @@ class _AudioPlayerScreenState extends ConsumerState { setState(() => _isLoading = false); // Step 6: begin playback and start the periodic progress ticker. + // The queue is read once here so the timer callback does not access [ref] + // after the widget may have been disposed (mirrors the client capture). unawaited(handler.play()); - _startProgressTicker(mediaIdInt, client, player); + final queue = ref.read(progressQueueProvider); + _startProgressTicker(mediaIdInt, client, player, queue); } /// Reads the bearer token and returns the `Authorization` header map. @@ -222,9 +227,14 @@ class _AudioPlayerScreenState extends ConsumerState { /// Starts a periodic timer that emits progress updates every /// [_kProgressInterval] and marks the item finished at [_kFinishedThreshold]. /// - /// The [client] and [player] references are captured once here so we avoid - /// accessing [ref] or [_audioPlayer] inside the timer callback after the - /// widget may have been disposed. + /// Progress updates are routed through [queue] rather than calling + /// [client.updateProgress] directly so that offline buffering and + /// online batch-flush are handled transparently (Open-Closed: screens + /// need not change if the queue strategy changes). + /// + /// The [client], [player], and [queue] references are captured once here so + /// we avoid accessing [ref] inside the timer callback after the widget may + /// have been disposed. /// /// The timer intentionally lives in the screen — not in the handler — so /// that [PlayerApiClient] (an HTTP concern) is not imported into @@ -233,21 +243,22 @@ class _AudioPlayerScreenState extends ConsumerState { int mediaId, PlayerApiClient client, AudioPlayer player, + ProgressQueueBase queue, ) { _progressTimer = Timer.periodic(_kProgressInterval, (_) async { - // Skip network calls while paused — no progress to record and avoids - // unnecessary server traffic when the user has paused playback. + // Skip updates while paused — no progress to record and avoids + // unnecessary DB writes when the user has paused playback. if (player.playing == false) return; final position = player.position; final duration = player.duration; - // Emit raw position update — fire-and-forget so a transient network - // error never interrupts playback. + // Enqueue position update — fire-and-forget so a transient error + // never interrupts playback. The queue handles online/offline. try { - await client.updateProgress( - mediaId: mediaId, - positionSeconds: position.inMilliseconds / 1000.0, + await queue.enqueue( + mediaId, + position.inMilliseconds / 1000.0, ); } catch (_) {} @@ -260,6 +271,10 @@ class _AudioPlayerScreenState extends ConsumerState { _kFinishedThreshold) { _finishedEmitted = true; try { + // The finished status update is still sent directly to the API + // because it is a distinct endpoint and should not be queued with + // position updates (different semantics: idempotent status vs. + // position accumulation). await client.updateProgressStatus( mediaId: mediaId, status: 'finished', diff --git a/player-android/lib/screens/video_player_screen.dart b/player-android/lib/screens/video_player_screen.dart index b3247a3..763168d 100644 --- a/player-android/lib/screens/video_player_screen.dart +++ b/player-android/lib/screens/video_player_screen.dart @@ -7,6 +7,8 @@ import 'package:video_player/video_player.dart'; import '../api/player_api_client.dart'; import '../providers/api_client_provider.dart'; +import '../providers/progress_queue_provider.dart'; +import '../services/progress_queue.dart'; // How often progress updates are emitted to the server while playing. const _kProgressInterval = Duration(seconds: 5); @@ -190,7 +192,10 @@ class _VideoPlayerScreenState extends ConsumerState { }); // Start the periodic progress ticker now that playback is ready. - _startProgressTicker(mediaIdInt, client); + // The queue is read once here so the timer callback does not access [ref] + // after the widget may have been disposed (mirrors the client capture). + final queue = ref.read(progressQueueProvider); + _startProgressTicker(mediaIdInt, client, queue); } // --------------------------------------------------------------------------- @@ -200,26 +205,36 @@ class _VideoPlayerScreenState extends ConsumerState { /// Starts a periodic timer that emits progress updates every /// [_kProgressInterval] and marks the item finished at [_kFinishedThreshold]. /// - /// The [client] reference is captured once here so we avoid accessing [ref] - /// inside the timer callback after the widget may have been disposed. - void _startProgressTicker(int mediaId, PlayerApiClient client) { + /// Progress updates are routed through [queue] rather than calling + /// [client.updateProgress] directly so that offline buffering and + /// online batch-flush are handled transparently (Open-Closed: screens + /// need not change if the queue strategy changes). + /// + /// The [client] and [queue] references are captured once here so we avoid + /// accessing [ref] inside the timer callback after the widget may have been + /// disposed. + void _startProgressTicker( + int mediaId, + PlayerApiClient client, + ProgressQueueBase queue, + ) { _progressTimer = Timer.periodic(_kProgressInterval, (_) async { final vc = _videoController; if (vc == null) return; - // Skip network calls while paused — no progress to record and avoids - // unnecessary server traffic when the user has paused playback. + // Skip updates while paused — no progress to record and avoids + // unnecessary DB writes when the user has paused playback. if (!vc.value.isPlaying) return; final position = vc.value.position; final duration = vc.value.duration; - // Emit raw position update — fire-and-forget so a transient network - // error never interrupts playback. + // Enqueue position update — fire-and-forget so a transient error + // never interrupts playback. The queue handles online/offline. try { - await client.updateProgress( - mediaId: mediaId, - positionSeconds: position.inMilliseconds / 1000.0, + await queue.enqueue( + mediaId, + position.inMilliseconds / 1000.0, ); } catch (_) {} @@ -231,6 +246,10 @@ class _VideoPlayerScreenState extends ConsumerState { _kFinishedThreshold) { _finishedEmitted = true; try { + // The finished status update is still sent directly to the API + // because it is a distinct endpoint and should not be queued with + // position updates (different semantics: idempotent status vs. + // position accumulation). await client.updateProgressStatus( mediaId: mediaId, status: 'finished', diff --git a/player-android/lib/services/progress_queue.dart b/player-android/lib/services/progress_queue.dart new file mode 100644 index 0000000..74b8e5e --- /dev/null +++ b/player-android/lib/services/progress_queue.dart @@ -0,0 +1,378 @@ +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); + } +} diff --git a/player-android/pubspec.lock b/player-android/pubspec.lock index 47f6b23..0ae7f78 100644 --- a/player-android/pubspec.lock +++ b/player-android/pubspec.lock @@ -121,6 +121,22 @@ packages: url: "https://pub.dev" source: hosted version: "1.19.1" + connectivity_plus: + dependency: "direct main" + description: + name: connectivity_plus + sha256: b5e72753cf63becce2c61fd04dfe0f1c430cc5278b53a1342dc5ad839eab29ec + url: "https://pub.dev" + source: hosted + version: "6.1.5" + connectivity_plus_platform_interface: + dependency: transitive + description: + name: connectivity_plus_platform_interface + sha256: "3c09627c536d22fd24691a905cdd8b14520de69da52c7a97499c8be5284a32ed" + url: "https://pub.dev" + source: hosted + version: "2.1.0" crypto: dependency: transitive description: @@ -488,6 +504,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.0.0" + nm: + dependency: transitive + description: + name: nm + sha256: "2c9aae4127bdc8993206464fcc063611e0e36e72018696cd9631023a31b24254" + url: "https://pub.dev" + source: hosted + version: "0.5.0" objective_c: dependency: transitive description: @@ -718,7 +742,7 @@ packages: source: hosted version: "1.10.2" sqflite: - dependency: transitive + dependency: "direct main" description: name: sqflite sha256: "564cfed0746fe53140c23b70b308e045c3b31f17778f2f326ccb7d804ea0250a" @@ -741,6 +765,14 @@ packages: url: "https://pub.dev" source: hosted version: "2.5.8" + sqflite_common_ffi: + dependency: "direct dev" + description: + name: sqflite_common_ffi + sha256: cd0c7f7de39a08f2d54ef144d9058c46eca8461879aaa648025643455c1e5a20 + url: "https://pub.dev" + source: hosted + version: "2.4.0+3" sqflite_darwin: dependency: transitive description: @@ -757,6 +789,14 @@ packages: url: "https://pub.dev" source: hosted version: "2.4.0" + sqlite3: + dependency: transitive + description: + name: sqlite3 + sha256: "56da3e13ed7d28a66f930aa2b2b29db6736a233f08283326e96321dd812030f5" + url: "https://pub.dev" + source: hosted + version: "3.3.1" stack_trace: dependency: transitive description: diff --git a/player-android/pubspec.yaml b/player-android/pubspec.yaml index 2fcc290..305a788 100644 --- a/player-android/pubspec.yaml +++ b/player-android/pubspec.yaml @@ -35,6 +35,12 @@ dependencies: # audio_service: runs audio in the background as a foreground service, # integrates with Android media notifications and lock-screen controls. audio_service: ^0.18.15 + # sqflite: SQLite plugin for Flutter; used by ProgressQueue to persist pending + # progress updates locally so they survive process restarts while offline. + sqflite: ^2.4.1 + # connectivity_plus: monitors network reachability changes; ProgressQueue + # subscribes to its stream to trigger a flush when connectivity is restored. + connectivity_plus: ^6.1.4 dev_dependencies: flutter_test: @@ -43,6 +49,9 @@ dev_dependencies: # http_mock_adapter: intercepts Dio requests in unit tests, returning canned # responses without a real network connection. http_mock_adapter: ^0.6.1 + # sqflite_common_ffi: in-memory SQLite backend for pure-Dart tests; allows + # ProgressQueue tests to run without a real Android device or platform channel. + sqflite_common_ffi: ^2.3.4 flutter: uses-material-design: true diff --git a/player-android/test/screens/audio_player_screen_test.dart b/player-android/test/screens/audio_player_screen_test.dart index a7af1fb..0c10177 100644 --- a/player-android/test/screens/audio_player_screen_test.dart +++ b/player-android/test/screens/audio_player_screen_test.dart @@ -39,8 +39,10 @@ import 'package:player_android/api/dio_client.dart'; import 'package:player_android/api/player_api_client.dart'; import 'package:player_android/providers/api_client_provider.dart'; import 'package:player_android/providers/audio_handler_provider.dart'; +import 'package:player_android/providers/progress_queue_provider.dart'; import 'package:player_android/screens/audio_player_screen.dart'; import 'package:player_android/services/audio_handler.dart'; +import 'package:player_android/services/progress_queue.dart'; // --------------------------------------------------------------------------- // Fakes @@ -99,6 +101,27 @@ class _FakeApiClient extends PlayerApiClient { 'http://localhost:8080/api/v1/media/$mediaId/stream'; } +/// No-op [ProgressQueueBase] stub for widget tests. +/// +/// Implements [ProgressQueueBase] directly rather than extending [ProgressQueue] +/// so no real SQLite database is opened and no connectivity subscription is +/// created in the test harness (Liskov Substitution — any [ProgressQueueBase] +/// can be injected wherever the interface is required). +class _FakeProgressQueue implements ProgressQueueBase { + @override + Future init() async {} // no-op — no DB needed in widget tests + + @override + Future enqueue( + int mediaId, + double positionSeconds, { + bool finished = false, + }) async {} // no-op — prevent SQLite calls in widget tests + + @override + Future dispose() async {} // no-op +} + /// A [PlayerAudioHandler] subclass that wraps a real [AudioPlayer] but /// overrides [setMediaItem] and playback methods to be no-ops so that no /// platform channels are invoked during widget tests. @@ -185,6 +208,9 @@ Future _pumpScreen( // Override audioHandlerProvider so no real AudioService or AudioPlayer // platform channels are invoked during widget tests. audioHandlerProvider.overrideWithValue(handler), + // Override progressQueueProvider so no real SQLite DB is opened and + // no connectivity subscription is created during widget tests. + progressQueueProvider.overrideWithValue(_FakeProgressQueue()), ], child: MaterialApp.router(routerConfig: router), ), diff --git a/player-android/test/screens/video_player_screen_test.dart b/player-android/test/screens/video_player_screen_test.dart index 8f24817..e2b34c0 100644 --- a/player-android/test/screens/video_player_screen_test.dart +++ b/player-android/test/screens/video_player_screen_test.dart @@ -31,7 +31,9 @@ import 'package:go_router/go_router.dart'; import 'package:player_android/api/dio_client.dart'; import 'package:player_android/api/player_api_client.dart'; import 'package:player_android/providers/api_client_provider.dart'; +import 'package:player_android/providers/progress_queue_provider.dart'; import 'package:player_android/screens/video_player_screen.dart'; +import 'package:player_android/services/progress_queue.dart'; // --------------------------------------------------------------------------- // Fakes @@ -101,6 +103,31 @@ class _FakeApiClient extends PlayerApiClient { 'http://localhost:8080/api/v1/media/$mediaId/stream'; } +// --------------------------------------------------------------------------- +// Fakes +// --------------------------------------------------------------------------- + +/// No-op [ProgressQueueBase] stub for widget tests. +/// +/// Implements [ProgressQueueBase] directly rather than extending [ProgressQueue] +/// so no real SQLite database is opened and no connectivity subscription is +/// created in the test harness (Liskov Substitution — any [ProgressQueueBase] +/// can be injected wherever the interface is required). +class _FakeProgressQueue implements ProgressQueueBase { + @override + Future init() async {} + + @override + Future enqueue( + int mediaId, + double positionSeconds, { + bool finished = false, + }) async {} + + @override + Future dispose() async {} +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- @@ -137,6 +164,9 @@ Future _pumpScreen( overrides: [ tokenStorageProvider.overrideWithValue(const _FakeTokenStorage()), apiClientProvider.overrideWithValue(fakeClient), + // Override progressQueueProvider so no real SQLite DB is opened and + // no connectivity subscription is created during widget tests. + progressQueueProvider.overrideWithValue(_FakeProgressQueue()), ], child: MaterialApp.router(routerConfig: router), ), diff --git a/player-android/test/services/progress_queue_test.dart b/player-android/test/services/progress_queue_test.dart new file mode 100644 index 0000000..2be65a8 --- /dev/null +++ b/player-android/test/services/progress_queue_test.dart @@ -0,0 +1,332 @@ +// Unit tests for ProgressQueue (lib/services/progress_queue.dart). +// +// Tests cover: +// 1. enqueue stores a row to the SQLite database (no flush while offline). +// 2. flush sends all queued rows via batchUpdateProgress. +// 3. flush clears rows after a successful send. +// 4. flush retains rows on server error. +// 5. offline items are flushed when connectivity is restored. +// 6. concurrent flush calls do not double-send. +// +// The SQLite backend is replaced with sqflite_common_ffi's in-memory factory so +// the tests run on Linux/macOS CI without a real Android device. Connectivity +// is simulated by injecting a [_FakeConnectivity] whose stream is controlled by +// a [StreamController]. +// +// Timing note: after emitting a connectivity event the listener is async. +// [_pump] drains the Dart microtask and timer queues by issuing several +// [Future.delayed(Duration.zero)] calls to give the async chain +// enough event-loop turns to complete. +// +// Run with: flutter test test/services/progress_queue_test.dart + +import 'dart:async'; + +import 'package:connectivity_plus/connectivity_plus.dart'; +import 'package:dio/dio.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:sqflite_common_ffi/sqflite_ffi.dart'; + +import 'package:player_android/api/player_api_client.dart'; +import 'package:player_android/services/progress_queue.dart'; + +// --------------------------------------------------------------------------- +// Fakes +// --------------------------------------------------------------------------- + +/// Fake [PlayerApiClient] that records [batchUpdateProgress] calls. +/// +/// [shouldThrowOnNextCall] makes the next call throw an [Exception] to +/// simulate a server error. +class _FakeApiClient extends PlayerApiClient { + _FakeApiClient() : super(dio: Dio()); + + // Accumulated payloads (each entry = one batchUpdateProgress call). + final List>> calls = []; + + // When true, the next call throws instead of recording. + bool shouldThrowOnNextCall = false; + + @override + Future batchUpdateProgress( + List> updates, + ) async { + if (shouldThrowOnNextCall) { + shouldThrowOnNextCall = false; + throw Exception('simulated server error'); + } + calls.add(List.unmodifiable(updates)); + } +} + +/// Fake [Connectivity] driven by the test via [emitStatus]. +/// +/// Defaults to offline ([ConnectivityResult.none]) so enqueue tests do not +/// trigger accidental auto-flush. +class _FakeConnectivity implements Connectivity { + _FakeConnectivity() { + _controller = StreamController>.broadcast(); + } + + late final StreamController> _controller; + List _current = [ConnectivityResult.none]; + + /// Pushes [results] to the stream and updates [checkConnectivity] state. + void emitStatus(List results) { + _current = results; + _controller.add(results); + } + + @override + Stream> get onConnectivityChanged => + _controller.stream; + + @override + Future> checkConnectivity() async => _current; + + void close() => _controller.close(); + + @override + dynamic noSuchMethod(Invocation i) => super.noSuchMethod(i); +} + +// --------------------------------------------------------------------------- +// Fixture factory +// --------------------------------------------------------------------------- + +/// Creates an in-memory [Database] with the production schema. +Future _openInMemoryDb() async { + return databaseFactoryFfi.openDatabase( + inMemoryDatabasePath, + options: OpenDatabaseOptions( + version: 1, + onCreate: (db, _) => db.execute(''' + CREATE TABLE progress_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + media_id INTEGER NOT NULL, + position_seconds REAL NOT NULL, + finished INTEGER NOT NULL DEFAULT 0, + queued_at TEXT NOT NULL + ) + '''), + ), + ); +} + +/// Builds a [ProgressQueue] with in-memory DB and fake connectivity. +Future<({ProgressQueue queue, _FakeApiClient client, _FakeConnectivity conn})> + _makeQueue() async { + final db = await _openInMemoryDb(); + final client = _FakeApiClient(); + final conn = _FakeConnectivity(); + final queue = ProgressQueue(apiClient: client, db: db, connectivity: conn); + await queue.init(); + return (queue: queue, client: client, conn: conn); +} + +/// Drains the Dart event loop enough times for async stream listeners and DB +/// operations to complete. +/// +/// A single [Future.delayed(Duration.zero)] is not sufficient because +/// stream listeners schedule their work one microtask turn later; the DB calls +/// inside the listener add further async hops. Five round-trips covers the +/// full async chain (stream delivery → listener body → DB query → DB delete). +Future _pump() async { + for (var i = 0; i < 5; i++) { + await Future.delayed(Duration.zero); + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +void main() { + setUpAll(() { + sqfliteFfiInit(); + databaseFactory = databaseFactoryFfi; + }); + + // -------------------------------------------------------------------------- + // 1. enqueue stores to the database (no flush while offline) + // -------------------------------------------------------------------------- + + group('enqueue stores to DB', () { + test('does not call the API when device is offline', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + await queue.enqueue(42, 12.5); + + expect(client.calls, isEmpty, + reason: 'No API call expected while offline'); + await queue.dispose(); + }); + + test('stores multiple items while offline without calling the API', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + await queue.enqueue(1, 5.0); + await queue.enqueue(2, 10.0); + + expect(client.calls, isEmpty, + reason: 'No API call expected while offline'); + await queue.dispose(); + }); + + test('flushes immediately when online at enqueue time', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + // Report WiFi so checkConnectivity() returns online during enqueue. + conn.emitStatus([ConnectivityResult.wifi]); + await _pump(); + + await queue.enqueue(7, 30.0); + await _pump(); + + expect(client.calls, hasLength(1), + reason: 'Should flush immediately when already online'); + expect(client.calls.first.first['media_id'], equals(7)); + await queue.dispose(); + }); + }); + + // -------------------------------------------------------------------------- + // 2. flush sends via batchUpdateProgress + // -------------------------------------------------------------------------- + + group('flush sends batch', () { + test('sends all queued rows in one batchUpdateProgress call', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + await queue.enqueue(10, 15.0); + await queue.enqueue(11, 25.0); + + conn.emitStatus([ConnectivityResult.mobile]); + await _pump(); + + expect(client.calls, hasLength(1), + reason: 'Exactly one batch call expected'); + expect(client.calls.first, hasLength(2), + reason: 'Both rows must be in the batch'); + final mediaIds = client.calls.first.map((m) => m['media_id']).toSet(); + expect(mediaIds, equals({10, 11})); + await queue.dispose(); + }); + + test('payload contains media_id, position_seconds, and observed_at', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + await queue.enqueue(5, 99.5); + conn.emitStatus([ConnectivityResult.wifi]); + await _pump(); + + expect(client.calls, hasLength(1)); + final item = client.calls.first.first; + expect(item['media_id'], equals(5)); + expect(item['position_seconds'], equals(99.5)); + expect(item.containsKey('observed_at'), isTrue, + reason: 'observed_at is required by the server for ordering'); + await queue.dispose(); + }); + }); + + // -------------------------------------------------------------------------- + // 3. flush clears rows after successful send + // -------------------------------------------------------------------------- + + group('flush clears rows on success', () { + test('rows are removed from the DB so a second flush is a no-op', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + await queue.enqueue(20, 1.0); + await queue.enqueue(21, 2.0); + + conn.emitStatus([ConnectivityResult.wifi]); + await _pump(); + + // Second flush on an empty table must not trigger another API call. + conn.emitStatus([ConnectivityResult.wifi]); + await _pump(); + + expect(client.calls, hasLength(1), + reason: 'Second flush must be a no-op after rows are cleared'); + await queue.dispose(); + }); + }); + + // -------------------------------------------------------------------------- + // 4. flush retains rows on server error + // -------------------------------------------------------------------------- + + group('flush retains rows on server error', () { + test('rows survive a failed flush and are sent on the next attempt', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + await queue.enqueue(30, 5.0); + + // First flush: API throws. + client.shouldThrowOnNextCall = true; + conn.emitStatus([ConnectivityResult.wifi]); + await _pump(); + + expect(client.calls, isEmpty, + reason: 'No successful call should have been recorded after throw'); + + // Second flush: API succeeds; rows should still be present. + conn.emitStatus([ConnectivityResult.wifi]); + await _pump(); + + expect(client.calls, hasLength(1), + reason: 'Row should be retried on the second flush'); + expect(client.calls.first.first['media_id'], equals(30)); + await queue.dispose(); + }); + }); + + // -------------------------------------------------------------------------- + // 5. offline items flushed on reconnect + // -------------------------------------------------------------------------- + + group('offline items flushed on reconnect', () { + test('all queued items are sent when connectivity is restored', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + for (var i = 0; i < 3; i++) { + await queue.enqueue(100 + i, i * 10.0); + } + expect(client.calls, isEmpty, reason: 'No flush while offline'); + + conn.emitStatus([ConnectivityResult.wifi]); + await _pump(); + + expect(client.calls, hasLength(1), + reason: 'One batch call expected after reconnect'); + expect(client.calls.first, hasLength(3), + reason: 'All three rows must be in the batch'); + await queue.dispose(); + }); + }); + + // -------------------------------------------------------------------------- + // 6. concurrent flush guard + // -------------------------------------------------------------------------- + + group('concurrent flush guard', () { + test('two rapid connectivity events result in at most one API call', () async { + final (:queue, :client, :conn) = await _makeQueue(); + + await queue.enqueue(50, 1.0); + + // Emit two events in rapid succession before any async work can run. + conn.emitStatus([ConnectivityResult.wifi]); + conn.emitStatus([ConnectivityResult.wifi]); + await _pump(); + + // The _isFlushing guard prevents a second concurrent flush, so at most + // one successful API call should have been made. + expect(client.calls.length, lessThanOrEqualTo(1), + reason: '_isFlushing should prevent double-flush'); + await queue.dispose(); + }); + }); +} -- cgit v1.2.3