feat(flutter/cache): SyncController — token-based delta sync
Drives /api/library/sync against drift: - reads cursor from SyncMetadata (default 0 = full snapshot) - 204 → just bump lastSyncAt, return zeroes - 410 → wipe all cached entities + retry from cursor=0 - 200 → apply upserts + deletes per entity type, advance cursor Handles all 8 entity types (artist/album/track + like_track/like_album/ like_artist + playlist/playlist_track) for both upsert and delete paths. Composite-key entities use the "<a>:<b>" string format the server emits. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
+293
@@ -0,0 +1,293 @@
|
||||
import 'package:drift/drift.dart' as drift;
|
||||
import 'package:flutter_riverpod/flutter_riverpod.dart';
|
||||
|
||||
import '../library/library_providers.dart' show dioProvider;
|
||||
import 'audio_cache_manager.dart' show appDbProvider;
|
||||
import 'db.dart';
|
||||
|
||||
/// Counts returned from a sync operation. Surfaced to the operator-
|
||||
/// facing "Sync now" button + Settings card.
|
||||
class SyncResult {
|
||||
const SyncResult({
|
||||
required this.upserts,
|
||||
required this.deletes,
|
||||
required this.cursor,
|
||||
});
|
||||
final int upserts;
|
||||
final int deletes;
|
||||
final int cursor;
|
||||
}
|
||||
|
||||
/// Drives the delta-sync against the server (#357 Plan A endpoint).
|
||||
/// Reads cursor from drift, calls /api/library/sync?since=<cursor>,
|
||||
/// applies upserts + deletes, advances cursor.
|
||||
class SyncController extends AsyncNotifier<SyncResult?> {
|
||||
@override
|
||||
Future<SyncResult?> build() async => null;
|
||||
|
||||
Future<SyncResult?> sync() async {
|
||||
state = const AsyncLoading();
|
||||
try {
|
||||
final db = ref.read(appDbProvider);
|
||||
final dio = await ref.read(dioProvider.future);
|
||||
|
||||
final meta = await db.select(db.syncMetadata).getSingleOrNull();
|
||||
final cursor = meta?.cursor ?? 0;
|
||||
|
||||
final resp = await dio.get<dynamic>(
|
||||
'/api/library/sync',
|
||||
queryParameters: {'since': cursor},
|
||||
);
|
||||
|
||||
// 204 No Content — no changes since cursor.
|
||||
if (resp.statusCode == 204) {
|
||||
await db.into(db.syncMetadata).insertOnConflictUpdate(
|
||||
SyncMetadataCompanion.insert(
|
||||
lastSyncAt: drift.Value(DateTime.now()),
|
||||
),
|
||||
);
|
||||
final r = SyncResult(upserts: 0, deletes: 0, cursor: cursor);
|
||||
state = AsyncData(r);
|
||||
return r;
|
||||
}
|
||||
|
||||
// 410 Gone — cursor too old, server has compacted past it. Reset
|
||||
// local state and retry once with cursor=0.
|
||||
if (resp.statusCode == 410) {
|
||||
await db.transaction(() async {
|
||||
for (final t in [
|
||||
db.cachedArtists,
|
||||
db.cachedAlbums,
|
||||
db.cachedTracks,
|
||||
db.cachedLikes,
|
||||
db.cachedPlaylists,
|
||||
db.cachedPlaylistTracks,
|
||||
]) {
|
||||
await db.delete(t).go();
|
||||
}
|
||||
await db.into(db.syncMetadata).insertOnConflictUpdate(
|
||||
SyncMetadataCompanion.insert(cursor: const drift.Value(0)),
|
||||
);
|
||||
});
|
||||
return sync();
|
||||
}
|
||||
|
||||
final body = resp.data as Map<String, dynamic>;
|
||||
final newCursor = (body['cursor'] as num).toInt();
|
||||
final upserts = body['upserts'] as Map<String, dynamic>? ?? {};
|
||||
final deletes = body['deletes'] as Map<String, dynamic>? ?? {};
|
||||
|
||||
var upsertCount = 0;
|
||||
var deleteCount = 0;
|
||||
|
||||
await db.transaction(() async {
|
||||
// ---- Upserts ----
|
||||
for (final a in (upserts['artist'] as List? ?? const [])) {
|
||||
await db.into(db.cachedArtists).insertOnConflictUpdate(
|
||||
_artistFromJson(a as Map<String, dynamic>),
|
||||
);
|
||||
upsertCount++;
|
||||
}
|
||||
for (final a in (upserts['album'] as List? ?? const [])) {
|
||||
await db.into(db.cachedAlbums).insertOnConflictUpdate(
|
||||
_albumFromJson(a as Map<String, dynamic>),
|
||||
);
|
||||
upsertCount++;
|
||||
}
|
||||
for (final t in (upserts['track'] as List? ?? const [])) {
|
||||
await db.into(db.cachedTracks).insertOnConflictUpdate(
|
||||
_trackFromJson(t as Map<String, dynamic>),
|
||||
);
|
||||
upsertCount++;
|
||||
}
|
||||
for (final l in (upserts['like_track'] as List? ?? const [])) {
|
||||
final m = l as Map<String, dynamic>;
|
||||
await db.into(db.cachedLikes).insertOnConflictUpdate(
|
||||
CachedLikesCompanion.insert(
|
||||
userId: m['user_id'] as String,
|
||||
entityType: 'track',
|
||||
entityId: m['track_id'] as String,
|
||||
),
|
||||
);
|
||||
upsertCount++;
|
||||
}
|
||||
for (final l in (upserts['like_album'] as List? ?? const [])) {
|
||||
final m = l as Map<String, dynamic>;
|
||||
await db.into(db.cachedLikes).insertOnConflictUpdate(
|
||||
CachedLikesCompanion.insert(
|
||||
userId: m['user_id'] as String,
|
||||
entityType: 'album',
|
||||
entityId: m['album_id'] as String,
|
||||
),
|
||||
);
|
||||
upsertCount++;
|
||||
}
|
||||
for (final l in (upserts['like_artist'] as List? ?? const [])) {
|
||||
final m = l as Map<String, dynamic>;
|
||||
await db.into(db.cachedLikes).insertOnConflictUpdate(
|
||||
CachedLikesCompanion.insert(
|
||||
userId: m['user_id'] as String,
|
||||
entityType: 'artist',
|
||||
entityId: m['artist_id'] as String,
|
||||
),
|
||||
);
|
||||
upsertCount++;
|
||||
}
|
||||
for (final p in (upserts['playlist'] as List? ?? const [])) {
|
||||
await db.into(db.cachedPlaylists).insertOnConflictUpdate(
|
||||
_playlistFromJson(p as Map<String, dynamic>),
|
||||
);
|
||||
upsertCount++;
|
||||
}
|
||||
for (final pt in (upserts['playlist_track'] as List? ?? const [])) {
|
||||
final m = pt as Map<String, dynamic>;
|
||||
await db.into(db.cachedPlaylistTracks).insertOnConflictUpdate(
|
||||
CachedPlaylistTracksCompanion.insert(
|
||||
playlistId: m['playlist_id'] as String,
|
||||
trackId: m['track_id'] as String,
|
||||
),
|
||||
);
|
||||
upsertCount++;
|
||||
}
|
||||
|
||||
// ---- Deletes ----
|
||||
for (final id in (deletes['artist'] as List? ?? const [])) {
|
||||
await (db.delete(db.cachedArtists)
|
||||
..where((t) => t.id.equals(id as String)))
|
||||
.go();
|
||||
deleteCount++;
|
||||
}
|
||||
for (final id in (deletes['album'] as List? ?? const [])) {
|
||||
await (db.delete(db.cachedAlbums)
|
||||
..where((t) => t.id.equals(id as String)))
|
||||
.go();
|
||||
deleteCount++;
|
||||
}
|
||||
for (final id in (deletes['track'] as List? ?? const [])) {
|
||||
await (db.delete(db.cachedTracks)
|
||||
..where((t) => t.id.equals(id as String)))
|
||||
.go();
|
||||
deleteCount++;
|
||||
}
|
||||
for (final id in (deletes['like_track'] as List? ?? const [])) {
|
||||
final parts = (id as String).split(':');
|
||||
if (parts.length != 2) continue;
|
||||
await (db.delete(db.cachedLikes)
|
||||
..where((t) =>
|
||||
t.userId.equals(parts[0]) &
|
||||
t.entityType.equals('track') &
|
||||
t.entityId.equals(parts[1])))
|
||||
.go();
|
||||
deleteCount++;
|
||||
}
|
||||
for (final id in (deletes['like_album'] as List? ?? const [])) {
|
||||
final parts = (id as String).split(':');
|
||||
if (parts.length != 2) continue;
|
||||
await (db.delete(db.cachedLikes)
|
||||
..where((t) =>
|
||||
t.userId.equals(parts[0]) &
|
||||
t.entityType.equals('album') &
|
||||
t.entityId.equals(parts[1])))
|
||||
.go();
|
||||
deleteCount++;
|
||||
}
|
||||
for (final id in (deletes['like_artist'] as List? ?? const [])) {
|
||||
final parts = (id as String).split(':');
|
||||
if (parts.length != 2) continue;
|
||||
await (db.delete(db.cachedLikes)
|
||||
..where((t) =>
|
||||
t.userId.equals(parts[0]) &
|
||||
t.entityType.equals('artist') &
|
||||
t.entityId.equals(parts[1])))
|
||||
.go();
|
||||
deleteCount++;
|
||||
}
|
||||
for (final id in (deletes['playlist'] as List? ?? const [])) {
|
||||
await (db.delete(db.cachedPlaylists)
|
||||
..where((t) => t.id.equals(id as String)))
|
||||
.go();
|
||||
deleteCount++;
|
||||
}
|
||||
for (final id in (deletes['playlist_track'] as List? ?? const [])) {
|
||||
final parts = (id as String).split(':');
|
||||
if (parts.length != 2) continue;
|
||||
await (db.delete(db.cachedPlaylistTracks)
|
||||
..where((t) =>
|
||||
t.playlistId.equals(parts[0]) &
|
||||
t.trackId.equals(parts[1])))
|
||||
.go();
|
||||
deleteCount++;
|
||||
}
|
||||
|
||||
// ---- Cursor + lastSyncAt ----
|
||||
await db.into(db.syncMetadata).insertOnConflictUpdate(
|
||||
SyncMetadataCompanion.insert(
|
||||
cursor: drift.Value(newCursor),
|
||||
lastSyncAt: drift.Value(DateTime.now()),
|
||||
),
|
||||
);
|
||||
});
|
||||
|
||||
final result = SyncResult(
|
||||
upserts: upsertCount,
|
||||
deletes: deleteCount,
|
||||
cursor: newCursor,
|
||||
);
|
||||
state = AsyncData(result);
|
||||
return result;
|
||||
} catch (e, st) {
|
||||
state = AsyncError(e, st);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
CachedArtistsCompanion _artistFromJson(Map<String, dynamic> j) =>
|
||||
CachedArtistsCompanion.insert(
|
||||
id: j['id'] as String,
|
||||
name: (j['name'] as String?) ?? '',
|
||||
sortName: (j['sort_name'] as String?) ?? '',
|
||||
mbid: drift.Value(j['mbid'] as String?),
|
||||
artistThumbPath: drift.Value(j['artist_thumb_path'] as String?),
|
||||
artistFanartPath: drift.Value(j['artist_fanart_path'] as String?),
|
||||
);
|
||||
|
||||
CachedAlbumsCompanion _albumFromJson(Map<String, dynamic> j) =>
|
||||
CachedAlbumsCompanion.insert(
|
||||
id: j['id'] as String,
|
||||
artistId: j['artist_id'] as String,
|
||||
title: (j['title'] as String?) ?? '',
|
||||
sortTitle: (j['sort_title'] as String?) ?? '',
|
||||
releaseDate: drift.Value(j['release_date'] as String?),
|
||||
coverPath: drift.Value(j['cover_art_path'] as String?),
|
||||
mbid: drift.Value(j['mbid'] as String?),
|
||||
);
|
||||
|
||||
CachedTracksCompanion _trackFromJson(Map<String, dynamic> j) =>
|
||||
CachedTracksCompanion.insert(
|
||||
id: j['id'] as String,
|
||||
albumId: j['album_id'] as String,
|
||||
artistId: j['artist_id'] as String,
|
||||
title: (j['title'] as String?) ?? '',
|
||||
durationMs: drift.Value((j['duration_ms'] as num?)?.toInt() ?? 0),
|
||||
trackNumber: drift.Value((j['track_number'] as num?)?.toInt()),
|
||||
discNumber: drift.Value((j['disc_number'] as num?)?.toInt()),
|
||||
filePath: drift.Value(j['file_path'] as String?),
|
||||
fileFormat: drift.Value(j['file_format'] as String?),
|
||||
genre: drift.Value(j['genre'] as String?),
|
||||
);
|
||||
|
||||
CachedPlaylistsCompanion _playlistFromJson(Map<String, dynamic> j) =>
|
||||
CachedPlaylistsCompanion.insert(
|
||||
id: j['id'] as String,
|
||||
userId: j['user_id'] as String,
|
||||
name: (j['name'] as String?) ?? '',
|
||||
description: drift.Value((j['description'] as String?) ?? ''),
|
||||
isPublic: drift.Value((j['is_public'] as bool?) ?? false),
|
||||
coverPath: drift.Value(j['cover_path'] as String?),
|
||||
trackCount: drift.Value((j['track_count'] as num?)?.toInt() ?? 0),
|
||||
durationSec: drift.Value((j['duration_sec'] as num?)?.toInt() ?? 0),
|
||||
);
|
||||
}
|
||||
|
||||
final syncControllerProvider =
|
||||
AsyncNotifierProvider<SyncController, SyncResult?>(SyncController.new);
|
||||
+131
@@ -0,0 +1,131 @@
|
||||
import 'package:dio/dio.dart';
|
||||
import 'package:drift/native.dart' show NativeDatabase;
|
||||
import 'package:flutter_riverpod/flutter_riverpod.dart';
|
||||
import 'package:flutter_test/flutter_test.dart';
|
||||
|
||||
import 'package:minstrel/cache/audio_cache_manager.dart' show appDbProvider;
|
||||
import 'package:minstrel/cache/db.dart';
|
||||
import 'package:minstrel/cache/sync_controller.dart';
|
||||
import 'package:minstrel/library/library_providers.dart' show dioProvider;
|
||||
|
||||
/// Builds a Dio whose adapter resolves every request to the supplied
|
||||
/// status code + body. Avoids touching the network in tests.
|
||||
Dio _stubDio({required int status, dynamic body}) {
|
||||
final dio = Dio();
|
||||
dio.interceptors.add(InterceptorsWrapper(onRequest: (req, h) {
|
||||
h.resolve(Response<dynamic>(
|
||||
requestOptions: req,
|
||||
statusCode: status,
|
||||
data: body,
|
||||
));
|
||||
}));
|
||||
return dio;
|
||||
}
|
||||
|
||||
ProviderContainer _container({required AppDb db, required Dio dio}) {
|
||||
return ProviderContainer(overrides: [
|
||||
appDbProvider.overrideWithValue(db),
|
||||
dioProvider.overrideWith((ref) async => dio),
|
||||
]);
|
||||
}
|
||||
|
||||
void main() {
|
||||
test('204 advances lastSyncAt without changing cursor', () async {
|
||||
final db = AppDb(NativeDatabase.memory());
|
||||
addTearDown(db.close);
|
||||
|
||||
final container = _container(db: db, dio: _stubDio(status: 204));
|
||||
addTearDown(container.dispose);
|
||||
|
||||
final result = await container.read(syncControllerProvider.notifier).sync();
|
||||
expect(result?.upserts, 0);
|
||||
expect(result?.deletes, 0);
|
||||
final meta = await db.select(db.syncMetadata).getSingleOrNull();
|
||||
expect(meta?.lastSyncAt, isNotNull);
|
||||
});
|
||||
|
||||
test('200 with artist upsert writes drift row + advances cursor', () async {
|
||||
final db = AppDb(NativeDatabase.memory());
|
||||
addTearDown(db.close);
|
||||
|
||||
final container = _container(
|
||||
db: db,
|
||||
dio: _stubDio(status: 200, body: {
|
||||
'cursor': 7,
|
||||
'upserts': {
|
||||
'artist': [
|
||||
{'id': 'a1', 'name': 'A', 'sort_name': 'A'},
|
||||
],
|
||||
},
|
||||
'deletes': {},
|
||||
}),
|
||||
);
|
||||
addTearDown(container.dispose);
|
||||
|
||||
final result = await container.read(syncControllerProvider.notifier).sync();
|
||||
expect(result?.upserts, 1);
|
||||
expect(result?.cursor, 7);
|
||||
final artist = await (db.select(db.cachedArtists)
|
||||
..where((t) => t.id.equals('a1')))
|
||||
.getSingleOrNull();
|
||||
expect(artist?.name, 'A');
|
||||
final meta = await db.select(db.syncMetadata).getSingleOrNull();
|
||||
expect(meta?.cursor, 7);
|
||||
});
|
||||
|
||||
test('200 with track delete removes the row', () async {
|
||||
final db = AppDb(NativeDatabase.memory());
|
||||
addTearDown(db.close);
|
||||
|
||||
// Seed an existing cached track
|
||||
await db.into(db.cachedTracks).insertOnConflictUpdate(
|
||||
CachedTracksCompanion.insert(
|
||||
id: 't1', albumId: 'al1', artistId: 'ar1', title: 'song'),
|
||||
);
|
||||
|
||||
final container = _container(
|
||||
db: db,
|
||||
dio: _stubDio(status: 200, body: {
|
||||
'cursor': 3,
|
||||
'upserts': {},
|
||||
'deletes': {
|
||||
'track': ['t1'],
|
||||
},
|
||||
}),
|
||||
);
|
||||
addTearDown(container.dispose);
|
||||
|
||||
final result = await container.read(syncControllerProvider.notifier).sync();
|
||||
expect(result?.deletes, 1);
|
||||
final track = await (db.select(db.cachedTracks)
|
||||
..where((t) => t.id.equals('t1')))
|
||||
.getSingleOrNull();
|
||||
expect(track, isNull);
|
||||
});
|
||||
|
||||
test('like_track upsert + delete round-trip', () async {
|
||||
final db = AppDb(NativeDatabase.memory());
|
||||
addTearDown(db.close);
|
||||
|
||||
final container = _container(
|
||||
db: db,
|
||||
dio: _stubDio(status: 200, body: {
|
||||
'cursor': 1,
|
||||
'upserts': {
|
||||
'like_track': [
|
||||
{'user_id': 'u1', 'track_id': 't1'},
|
||||
],
|
||||
},
|
||||
'deletes': {},
|
||||
}),
|
||||
);
|
||||
addTearDown(container.dispose);
|
||||
|
||||
await container.read(syncControllerProvider.notifier).sync();
|
||||
final liked = await db.select(db.cachedLikes).get();
|
||||
expect(liked.length, 1);
|
||||
expect(liked.first.userId, 'u1');
|
||||
expect(liked.first.entityType, 'track');
|
||||
expect(liked.first.entityId, 't1');
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user