cjones/immich
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: sync

Browse Source
This commit is contained in:
shenlong-tanwen
2024-09-15 20:02:09 +05:30
parent 37b15869d5
commit ded4481190
20 changed files with 307 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mobile-v2/lib/domain/interfaces/asset.interface.dart
+14 -3
View File
@@ -1,15 +1,26 @@
import 'dart:async';
import 'package:immich_mobile/domain/models/asset.model.dart';
import 'package:immich_mobile/domain/models/render_list.model.dart';
abstract class IAssetRepository {
/// Batch insert asset
Future<bool> addAll(Iterable<Asset> assets);
FutureOr<bool> addAll(Iterable<Asset> assets);
/// Removes assets with the [localIds]
FutureOr<List<Asset>> fetchLocalAssetsForIds(List<String> localIds);
/// Removes assets with the [remoteIds]
FutureOr<List<Asset>> fetchRemoteAssetsForIds(List<String> remoteIds);
/// Removes assets with the given [ids]
FutureOr<void> deleteAssetsForIds(List<int> ids);
/// Removes all assets
Future<bool> clearAll();
FutureOr<bool> clearAll();
/// Fetch assets from the [offset] with the [limit]
Future<List<Asset>> fetchAssets({int? offset, int? limit});
FutureOr<List<Asset>> fetchAssets({int? offset, int? limit});
/// Streams assets as groups grouped by the group type passed
Stream<RenderList> watchRenderList();
mobile-v2/lib/domain/models/asset.model.dart
+32 -4
View File
@@ -1,3 +1,5 @@
import 'package:flutter/material.dart';
import 'package:immich_mobile/utils/collection_util.dart';
import 'package:immich_mobile/utils/extensions/string.extension.dart';
import 'package:openapi/api.dart';
@@ -70,8 +72,8 @@ class Asset {
DateTime? createdTime,
DateTime? modifiedTime,
int? duration,
String? localId,
String? remoteId,
ValueGetter<String?>? localId,
ValueGetter<String?>? remoteId,
String? livePhotoVideoId,
}) {
return Asset(
@@ -84,12 +86,32 @@ class Asset {
createdTime: createdTime ?? this.createdTime,
modifiedTime: modifiedTime ?? this.modifiedTime,
duration: duration ?? this.duration,
localId: localId ?? this.localId,
remoteId: remoteId ?? this.remoteId,
localId: localId != null ? localId() : this.localId,
remoteId: remoteId != null ? remoteId() : this.remoteId,
livePhotoVideoId: livePhotoVideoId ?? this.livePhotoVideoId,
);
}
Asset merge(Asset newAsset) {
if (newAsset.modifiedTime.isAfter(modifiedTime)) {
return newAsset.copyWith(
height: newAsset.height ?? height,
width: newAsset.width ?? width,
localId: () => newAsset.localId ?? localId,
remoteId: () => newAsset.remoteId ?? remoteId,
livePhotoVideoId: newAsset.livePhotoVideoId ?? livePhotoVideoId,
);
}
return copyWith(
height: height ?? newAsset.height,
width: width ?? newAsset.width,
localId: () => localId ?? newAsset.localId,
remoteId: () => remoteId ?? newAsset.remoteId,
livePhotoVideoId: livePhotoVideoId ?? newAsset.livePhotoVideoId,
);
}
@override
String toString() => """
{
@@ -140,6 +162,12 @@ class Asset {
remoteId.hashCode ^
livePhotoVideoId.hashCode;
}
static int compareByRemoteId(Asset a, Asset b) =>
CollectionUtil.compareToNullable(a.remoteId, b.remoteId);
static int compareByLocalId(Asset a, Asset b) =>
CollectionUtil.compareToNullable(a.localId, b.localId);
}
AssetType _toAssetType(AssetTypeEnum type) => switch (type) {
mobile-v2/lib/domain/repositories/asset.repository.dart
+23
View File
@@ -85,6 +85,29 @@ class RemoteAssetDriftRepository with LogContext implements IAssetRepository {
.watch()
.map((elements) => RenderList(elements: elements));
}
@override
Future<List<Asset>> fetchLocalAssetsForIds(List<String> localIds) async {
final query = _db.asset.select()
..where((row) => row.localId.isIn(localIds))
..orderBy([(asset) => OrderingTerm.asc(asset.localId)]);
return (await query.get()).map(_toModel).toList();
}
@override
Future<List<Asset>> fetchRemoteAssetsForIds(List<String> remoteIds) async {
final query = _db.asset.select()
..where((row) => row.remoteId.isIn(remoteIds))
..orderBy([(asset) => OrderingTerm.asc(asset.remoteId)]);
return (await query.get()).map(_toModel).toList();
}
@override
FutureOr<void> deleteAssetsForIds(List<int> ids) async {
await _db.asset.deleteWhere((row) => row.id.isIn(ids));
}
}
AssetCompanion _toEntity(Asset asset) {
mobile-v2/lib/domain/repositories/store.repository.dart
+1 -1
View File
@@ -75,7 +75,7 @@ class StoreDriftRepository with LogContext implements IStoreRepository {
_ => null,
} as U?;
if (primitive != null) {
return key.converter.fromPrimitive(primitive);
return await key.converter.fromPrimitive(primitive);
}
return null;
}
mobile-v2/lib/domain/services/asset_sync.service.dart
+149
View File
@@ -0,0 +1,149 @@
import 'dart:async';
import 'package:collection/collection.dart';
import 'package:immich_mobile/domain/interfaces/asset.interface.dart';
import 'package:immich_mobile/domain/models/asset.model.dart';
import 'package:immich_mobile/domain/models/user.model.dart';
import 'package:immich_mobile/service_locator.dart';
import 'package:immich_mobile/utils/collection_util.dart';
import 'package:immich_mobile/utils/constants/globals.dart';
import 'package:immich_mobile/utils/immich_api_client.dart';
import 'package:immich_mobile/utils/isolate_helper.dart';
import 'package:immich_mobile/utils/mixins/log_context.mixin.dart';
import 'package:logging/logging.dart';
import 'package:openapi/api.dart';
class AssetSyncService with LogContext {
const AssetSyncService();
Future<bool> doFullRemoteSyncForUserDrift(
User user, {
DateTime? updatedUtil,
int? limit,
}) async {
return await IsolateHelper.run(() async {
try {
final logger = Logger("SyncService <Isolate>");
final syncClient = di<ImmichApiClient>().getSyncApi();
final chunkSize = limit ?? kFullSyncChunkSize;
final updatedTill = updatedUtil ?? DateTime.now().toUtc();
updatedUtil ??= DateTime.now().toUtc();
String? lastAssetId;
while (true) {
logger.info(
"Requesting more chunks from lastId - ${lastAssetId ?? "<initial_fetch>"}",
);
final assets = await syncClient.getFullSyncForUser(AssetFullSyncDto(
limit: chunkSize,
updatedUntil: updatedTill,
lastId: lastAssetId,
userId: user.id,
));
if (assets == null) {
break;
}
final assetsFromServer =
assets.map(Asset.remote).sorted(Asset.compareByRemoteId);
final assetsInDb =
await di<IAssetRepository>().fetchRemoteAssetsForIds(
assetsFromServer.map((a) => a.remoteId!).toList(),
);
await _syncAssetsToDbDrift(
assetsFromServer,
assetsInDb,
Asset.compareByRemoteId,
isRemoteSync: true,
);
lastAssetId = assets.lastOrNull?.id;
if (assets.length != chunkSize) break;
}
return true;
} catch (e, s) {
log.severe("Error performing full sync for user - ${user.name}", e, s);
}
return false;
});
}
Future<void> _syncAssetsToDbDrift(
List<Asset> newAssets,
List<Asset> existingAssets,
Comparator<Asset> compare, {
bool? isRemoteSync,
}) async {
final (toAdd, toUpdate, assetsToRemove) = _diffAssets(
newAssets,
existingAssets,
compare: compare,
isRemoteSync: isRemoteSync,
);
final assetsToAdd = toAdd.followedBy(toUpdate);
await di<IAssetRepository>().addAll(assetsToAdd);
await di<IAssetRepository>()
.deleteAssetsForIds(assetsToRemove.map((a) => a.id).toList());
}
/// Returns a triple (toAdd, toUpdate, toRemove)
(List<Asset>, List<Asset>, List<Asset>) _diffAssets(
List<Asset> newAssets,
List<Asset> inDb, {
bool? isRemoteSync,
required Comparator<Asset> compare,
}) {
// fast paths for trivial cases: reduces memory usage during initial sync etc.
if (newAssets.isEmpty && inDb.isEmpty) {
return const ([], [], []);
} else if (newAssets.isEmpty && isRemoteSync == null) {
// remove all from database
return (const [], const [], inDb);
} else if (inDb.isEmpty) {
// add all assets
return (newAssets, const [], const []);
}
final List<Asset> toAdd = [];
final List<Asset> toUpdate = [];
final List<Asset> toRemove = [];
CollectionUtil.diffSortedLists(
inDb,
newAssets,
compare: compare,
both: (Asset a, Asset b) {
if (a == b) {
toUpdate.add(a.merge(b));
return true;
}
return false;
},
// Only in DB (removed asset)
onlyFirst: (Asset a) {
// We are syncing remote assets, if asset only inDB, then it is removed from remote
if (isRemoteSync == true && a.isLocal) {
if (a.remoteId != null) {
toUpdate.add(a.copyWith(remoteId: () => null));
}
// We are syncing local assets, mark the asset inDB as local only
} else if (isRemoteSync == false && a.isRemote) {
if (a.isLocal) {
toUpdate.add(a.copyWith(localId: () => null));
}
} else {
toRemove.add(a);
}
},
// Only in remote (new asset)
onlySecond: (Asset b) => toAdd.add(b),
);
return (toAdd, toUpdate, toRemove);
}
}
mobile-v2/lib/domain/services/login.service.dart
+1
View File
@@ -132,6 +132,7 @@ class LoginService with LogContext {
return false;
}
ServiceLocator.registerCurrentUser(user);
return true;
}
}
mobile-v2/lib/domain/services/sync.service.dart
-63
View File
@@ -1,63 +0,0 @@
import 'dart:async';
import 'package:immich_mobile/domain/interfaces/asset.interface.dart';
import 'package:immich_mobile/domain/models/asset.model.dart';
import 'package:immich_mobile/domain/models/user.model.dart';
import 'package:immich_mobile/domain/repositories/database.repository.dart';
import 'package:immich_mobile/service_locator.dart';
import 'package:immich_mobile/utils/constants/globals.dart';
import 'package:immich_mobile/utils/immich_api_client.dart';
import 'package:immich_mobile/utils/isolate_helper.dart';
import 'package:immich_mobile/utils/mixins/log_context.mixin.dart';
import 'package:logging/logging.dart';
import 'package:openapi/api.dart';
class SyncService with LogContext {
SyncService();
Future<bool> doFullSyncForUserDrift(
User user, {
DateTime? updatedUtil,
int? limit,
}) async {
return await IsolateHelper.run(() async {
try {
final logger = Logger("SyncService <Isolate>");
final syncClient = di<ImmichApiClient>().getSyncApi();
final chunkSize = limit ?? kFullSyncChunkSize;
final updatedTill = updatedUtil ?? DateTime.now().toUtc();
updatedUtil ??= DateTime.now().toUtc();
String? lastAssetId;
while (true) {
logger.info(
"Requesting more chunks from lastId - ${lastAssetId ?? "<initial_fetch>"}",
);
final assets = await syncClient.getFullSyncForUser(AssetFullSyncDto(
limit: chunkSize,
updatedUntil: updatedTill,
lastId: lastAssetId,
userId: user.id,
));
if (assets == null) {
break;
}
await di<IAssetRepository>().addAll(assets.map(Asset.remote));
lastAssetId = assets.lastOrNull?.id;
if (assets.length != chunkSize) break;
}
return true;
} catch (e, s) {
log.severe("Error performing full sync for user - ${user.name}", e, s);
} finally {
await di<DriftDatabaseRepository>().close();
}
return false;
});
}
}