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

feat: full local assets / album sync

Browse Source
This commit is contained in:
shenlong-tanwen
2024-10-17 23:33:00 +05:30
parent a09710ec7b
commit c91a2878dc
87 changed files with 2417 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mobile-v2/lib/domain/services/album_sync.service.dart
+131
View File
@@ -0,0 +1,131 @@
import 'package:immich_mobile/domain/interfaces/album.interface.dart';
import 'package:immich_mobile/domain/interfaces/album_asset.interface.dart';
import 'package:immich_mobile/domain/interfaces/album_etag.interface.dart';
import 'package:immich_mobile/domain/interfaces/asset.interface.dart';
import 'package:immich_mobile/domain/interfaces/database.interface.dart';
import 'package:immich_mobile/domain/interfaces/device_album.interface.dart';
import 'package:immich_mobile/domain/models/album.model.dart';
import 'package:immich_mobile/domain/models/album_etag.model.dart';
import 'package:immich_mobile/domain/services/asset_sync.service.dart';
import 'package:immich_mobile/domain/services/hash.service.dart';
import 'package:immich_mobile/service_locator.dart';
import 'package:immich_mobile/utils/collection_util.dart';
import 'package:immich_mobile/utils/isolate_helper.dart';
import 'package:immich_mobile/utils/mixins/log.mixin.dart';
class AlbumSyncService with LogMixin {
const AlbumSyncService();
Future<bool> performFullDeviceSyncIsolate() async {
return await IsolateHelper.run(performFullDeviceSync);
}
Future<bool> performFullDeviceSync() async {
try {
final deviceAlbums = await di<IDeviceAlbumRepository>().getAll();
final dbAlbums = await di<IAlbumRepository>().getAll(localOnly: true);
final hasChange = await CollectionUtil.diffLists(
dbAlbums,
deviceAlbums,
compare: Album.compareByLocalId,
both: _syncDeviceAlbum,
// Album is in DB but not anymore in device. Remove album and album specific assets
onlyFirst: _removeDeviceAlbum,
onlySecond: _addDeviceAlbum,
);
return hasChange;
} catch (e, s) {
log.e("Error performing full device sync", e, s);
}
return false;
}
Future<bool> _syncDeviceAlbum(
Album dbAlbum,
Album deviceAlbum, {
DateTime? modifiedUntil,
}) async {
assert(dbAlbum.id != null, "Album ID from DB is null");
final albumEtag =
await di<IAlbumETagRepository>().get(dbAlbum.id!) ?? AlbumETag.empty();
final assetCountInDevice =
await di<IDeviceAlbumRepository>().getAssetCount(deviceAlbum.localId!);
final albumNotUpdated = deviceAlbum.name == dbAlbum.name &&
dbAlbum.modifiedTime.isAtSameMomentAs(deviceAlbum.modifiedTime) &&
assetCountInDevice == albumEtag.assetCount;
if (albumNotUpdated) {
log.i("Device Album ${deviceAlbum.name} not updated. Skipping sync.");
return false;
}
await _addDeviceAlbum(dbAlbum, modifiedUntil: modifiedUntil);
return true;
}
Future<void> _addDeviceAlbum(Album album, {DateTime? modifiedUntil}) async {
try {
final albumId = (await di<IAlbumRepository>().upsert(album))?.id;
// break fast if we cannot add an album
if (albumId == null) {
log.d("Failed creating device album. Skipped assets from album");
return;
}
final assets = await di<HashService>().getHashedAssetsForAlbum(
album.localId!,
modifiedUntil: modifiedUntil,
);
await di<IDatabaseRepository>().txn(() async {
final albumAssetsInDB =
await di<IAlbumToAssetRepository>().getAssetsForAlbum(albumId);
await di<AssetSyncService>().upsertAssetsToDb(
assets,
albumAssetsInDB,
isRemoteSync: false,
);
// This is needed to get the updated assets for device album with valid db id field
final albumAssets = await di<IAssetRepository>()
.getForLocalIds(assets.map((a) => a.localId!));
await di<IAlbumToAssetRepository>().addAssetIds(
albumId,
albumAssets.map((a) => a.id!),
);
await di<IAlbumRepository>().upsert(
album.copyWith(thumbnailAssetId: albumAssets.firstOrNull?.id),
);
// Update ETag
final albumETag = AlbumETag(
albumId: albumId,
assetCount: assets.length,
modifiedTime: album.modifiedTime,
);
await di<IAlbumETagRepository>().upsert(albumETag);
});
} catch (e, s) {
log.w("Error while adding device album", e, s);
}
}
Future<void> _removeDeviceAlbum(Album album) async {
assert(album.id != null, "Album ID from DB is null");
final albumId = album.id!;
try {
await di<IDatabaseRepository>().txn(() async {
final toRemove =
await di<IAlbumToAssetRepository>().getAssetIdsOnlyInAlbum(albumId);
await di<IAlbumRepository>().deleteId(albumId);
await di<IAlbumToAssetRepository>().deleteAlbumId(albumId);
await di<IAssetRepository>().deleteIds(toRemove);
});
} catch (e, s) {
log.w("Error while removing device album", e, s);
}
}
}
mobile-v2/lib/domain/services/asset_sync.service.dart
+77 -66
View File
@@ -2,6 +2,7 @@ import 'dart:async';
import 'package:collection/collection.dart';
import 'package:immich_mobile/domain/interfaces/asset.interface.dart';
import 'package:immich_mobile/domain/interfaces/database.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';
@@ -9,76 +10,87 @@ 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/log_manager.dart';
import 'package:immich_mobile/utils/mixins/log.mixin.dart';
import 'package:openapi/api.dart';
class AssetSyncService with LogMixin {
const AssetSyncService();
Future<bool> performFullRemoteSyncForUser(
Future<bool> performFullRemoteSyncIsolate(
User user, {
DateTime? updatedUtil,
int? limit,
}) async {
return await IsolateHelper.run(() async {
try {
final logger = LogManager.I.get("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.d(
"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>().getForRemoteIds(
assetsFromServer.map((a) => a.remoteId!).toList(),
);
await _syncAssetsToDb(
assetsFromServer,
assetsInDb,
Asset.compareByRemoteId,
isRemoteSync: true,
);
lastAssetId = assets.lastOrNull?.id;
if (assets.length != chunkSize) break;
}
return true;
} catch (e, s) {
log.e("Error performing full sync for user - ${user.name}", e, s);
}
return false;
return await performFullRemoteSync(
user,
updatedUtil: updatedUtil,
limit: limit,
);
});
}
Future<void> _syncAssetsToDb(
List<Asset> newAssets,
List<Asset> existingAssets,
Comparator<Asset> compare, {
bool? isRemoteSync,
Future<bool> performFullRemoteSync(
User user, {
DateTime? updatedUtil,
int? limit,
}) async {
final (toAdd, toUpdate, assetsToRemove) = _diffAssets(
try {
final syncClient = di<ImApiClient>().getSyncApi();
final db = di<IDatabaseRepository>();
final assetRepo = di<IAssetRepository>();
final chunkSize = limit ?? kFullSyncChunkSize;
final updatedTill = updatedUtil ?? DateTime.now().toUtc();
updatedUtil ??= DateTime.now().toUtc();
String? lastAssetId;
while (true) {
log.d(
"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).toList();
await db.txn(() async {
final assetsInDb =
await assetRepo.getForHashes(assetsFromServer.map((a) => a.hash));
await upsertAssetsToDb(
assetsFromServer,
assetsInDb,
isRemoteSync: true,
);
});
lastAssetId = assets.lastOrNull?.id;
if (assets.length != chunkSize) break;
}
return true;
} catch (e, s) {
log.e("Error performing full remote sync for user - ${user.name}", e, s);
}
return false;
}
Future<void> upsertAssetsToDb(
List<Asset> newAssets,
List<Asset> existingAssets, {
bool? isRemoteSync,
Comparator<Asset> compare = Asset.compareByHash,
}) async {
final (toAdd, toUpdate, toRemove) = await _diffAssets(
newAssets,
existingAssets,
compare: compare,
@@ -88,37 +100,36 @@ class AssetSyncService with LogMixin {
final assetsToAdd = toAdd.followedBy(toUpdate);
await di<IAssetRepository>().upsertAll(assetsToAdd);
await di<IAssetRepository>()
.deleteIds(assetsToRemove.map((a) => a.id).toList());
await di<IAssetRepository>().deleteIds(toRemove.map((a) => a.id!).toList());
}
/// Returns a triple (toAdd, toUpdate, toRemove)
(List<Asset>, List<Asset>, List<Asset>) _diffAssets(
FutureOr<(List<Asset>, List<Asset>, List<Asset>)> _diffAssets(
List<Asset> newAssets,
List<Asset> inDb, {
bool? isRemoteSync,
required Comparator<Asset> compare,
}) {
Comparator<Asset> compare = Asset.compareByHash,
}) async {
// fast paths for trivial cases: reduces memory usage during initial sync etc.
if (newAssets.isEmpty && inDb.isEmpty) {
return const ([], [], []);
return const (<Asset>[], <Asset>[], <Asset>[]);
} else if (newAssets.isEmpty && isRemoteSync == null) {
// remove all from database
return (const [], const [], inDb);
return (const <Asset>[], const <Asset>[], inDb);
} else if (inDb.isEmpty) {
// add all assets
return (newAssets, const [], const []);
return (newAssets, const <Asset>[], const <Asset>[]);
}
final List<Asset> toAdd = [];
final List<Asset> toUpdate = [];
final List<Asset> toRemove = [];
CollectionUtil.diffSortedLists(
await CollectionUtil.diffLists(
inDb,
newAssets,
compare: compare,
both: (Asset a, Asset b) {
if (a == b) {
if (a != b) {
toUpdate.add(a.merge(b));
return true;
}
@@ -140,7 +151,7 @@ class AssetSyncService with LogMixin {
toRemove.add(a);
}
},
// Only in remote (new asset)
// Only in new assets
onlySecond: (Asset b) => toAdd.add(b),
);
return (toAdd, toUpdate, toRemove);
mobile-v2/lib/domain/services/hash.service.dart
+149
View File
@@ -0,0 +1,149 @@
import 'dart:convert';
import 'dart:io';
import 'dart:typed_data';
import 'package:flutter/foundation.dart';
import 'package:immich_mobile/domain/interfaces/device_album.interface.dart';
import 'package:immich_mobile/domain/interfaces/device_asset.interface.dart';
import 'package:immich_mobile/domain/interfaces/device_asset_hash.interface.dart';
import 'package:immich_mobile/domain/models/asset.model.dart';
import 'package:immich_mobile/domain/models/device_asset_hash.model.dart';
import 'package:immich_mobile/platform/messages.g.dart';
import 'package:immich_mobile/utils/constants/globals.dart';
import 'package:immich_mobile/utils/extensions/file.extension.dart';
import 'package:immich_mobile/utils/mixins/log.mixin.dart';
class HashService with LogMixin {
final ImHostService _hostService;
final IDeviceAssetRepository _deviceAssetRepository;
final IDeviceAlbumRepository _deviceAlbumRepository;
final IDeviceAssetToHashRepository _assetHashRepository;
const HashService({
required ImHostService hostService,
required IDeviceAssetRepository deviceAssetRepo,
required IDeviceAlbumRepository deviceAlbumRepo,
required IDeviceAssetToHashRepository assetToHashRepo,
}) : _hostService = hostService,
_deviceAssetRepository = deviceAssetRepo,
_deviceAlbumRepository = deviceAlbumRepo,
_assetHashRepository = assetToHashRepo;
Future<List<Asset>> getHashedAssetsForAlbum(
String albumId, {
DateTime? modifiedUntil,
}) async {
final assets = await _deviceAlbumRepository.getAssetsForAlbum(
albumId,
modifiedUntil: modifiedUntil,
);
assets.sort(Asset.compareByLocalId);
final assetIds = assets.map((a) => a.localId!);
final hashesInDB = await _assetHashRepository.getForIds(assetIds);
hashesInDB.sort(DeviceAssetToHash.compareByLocalId);
final hashedAssets = <Asset>[];
final orphanedHashes = <DeviceAssetToHash>[];
int bytesToBeProcessed = 0;
final filesToBeCleaned = <File>[];
final toBeHashed = <_AssetPath>[];
for (final asset in assets) {
if (hashesInDB.isNotEmpty && hashesInDB.first.localId == asset.localId) {
final hashed = hashesInDB.removeAt(0);
if (hashed.modifiedTime.isAtSameMomentAs(asset.modifiedTime)) {
hashedAssets.add(asset.copyWith(hash: hashed.hash));
continue;
}
// localID is matching, but the asset is modified. Discard the DeviceAssetToHash row
orphanedHashes.add(hashed);
}
final file = await _deviceAssetRepository.getOriginalFile(asset.localId!);
if (file == null) {
log.w("Cannot obtain file for localId ${asset.localId!}. Skipping");
continue;
}
filesToBeCleaned.add(file);
bytesToBeProcessed += await file.length();
toBeHashed.add(_AssetPath(asset: asset, path: file.path));
if (toBeHashed.length == kHashAssetsFileLimit ||
bytesToBeProcessed >= kHashAssetsSizeLimit) {
hashedAssets.addAll(await _processAssetBatch(toBeHashed));
// Clear file cache
await Future.wait(filesToBeCleaned.map((f) => f.deleteDarwinCache()));
toBeHashed.clear();
filesToBeCleaned.clear();
bytesToBeProcessed = 0;
}
}
if (toBeHashed.isNotEmpty) {
hashedAssets.addAll(await _processAssetBatch(toBeHashed));
// Clear file cache
await Future.wait(filesToBeCleaned.map((f) => f.deleteDarwinCache()));
}
assert(hashesInDB.isEmpty, "All hashes should be processed at this point");
_assetHashRepository.deleteIds(orphanedHashes.map((e) => e.id!).toList());
return hashedAssets;
}
/// Processes a batch of files and returns a list of successfully hashed assets after saving
/// them in [DeviceAssetToHash] for future retrieval
Future<List<Asset>> _processAssetBatch(List<_AssetPath> toBeHashed) async {
final hashes = await _hashFiles(toBeHashed.map((e) => e.path).toList());
assert(hashes.length == toBeHashed.length,
"Number of Hashes returned from platform should be the same as the input");
final hashedAssets = <Asset>[];
for (final (index, hash) in hashes.indexed) {
// ignore: avoid-unsafe-collection-methods
final asset = toBeHashed.elementAt(index).asset;
if (hash?.length == 20) {
hashedAssets.add(asset.copyWith(hash: base64.encode(hash!)));
} else {
log.w("Failed to hash file ${asset.localId ?? '<null>'}, skipping");
}
}
// Store the cache for future retrieval
_assetHashRepository.upsertAll(hashedAssets.map((a) => DeviceAssetToHash(
localId: a.localId!,
hash: a.hash,
modifiedTime: a.modifiedTime,
)));
log.v("Hashed ${hashedAssets.length}/${toBeHashed.length} assets");
return hashedAssets;
}
/// Hashes the given files and returns a list of the same length.
/// Files that could not be hashed will have a `null` value
Future<List<Uint8List?>> _hashFiles(List<String> paths) async {
try {
final hashes = await _hostService.digestFiles(paths);
return hashes;
} catch (e, s) {
log.e("Error occured while hashing assets", e, s);
}
return paths.map((p) => null).toList();
}
}
class _AssetPath {
final Asset asset;
final String path;
const _AssetPath({required this.asset, required this.path});
_AssetPath copyWith({Asset? asset, String? path}) {
return _AssetPath(asset: asset ?? this.asset, path: path ?? this.path);
}
}
mobile-v2/lib/domain/services/login.service.dart
+7 -7
View File
@@ -15,15 +15,15 @@ import 'package:openapi/api.dart';
class LoginService with LogMixin {
const LoginService();
Future<bool> isEndpointAvailable(Uri uri, {ImmichApiClient? client}) async {
Future<bool> isEndpointAvailable(Uri uri, {ImApiClient? client}) async {
String baseUrl = uri.toString();
if (!baseUrl.endsWith('/api')) {
baseUrl += '/api';
}
final serverAPI = client?.getServerApi() ??
ImmichApiClient(endpoint: baseUrl).getServerApi();
final serverAPI =
client?.getServerApi() ?? ImApiClient(endpoint: baseUrl).getServerApi();
try {
await serverAPI.pingServer();
} catch (e) {
@@ -35,7 +35,7 @@ class LoginService with LogMixin {
Future<String> resolveEndpoint(Uri uri, {Client? client}) async {
String baseUrl = uri.toString();
final d = client ?? ImmichApiClient(endpoint: baseUrl).client;
final d = client ?? ImApiClient(endpoint: baseUrl).client;
try {
// Check for well-known endpoint
@@ -62,7 +62,7 @@ class LoginService with LogMixin {
Future<String?> passwordLogin(String email, String password) async {
try {
final loginResponse =
await di<ImmichApiClient>().getAuthenticationApi().login(
await di<ImApiClient>().getAuthenticationApi().login(
LoginCredentialDto(email: email, password: password),
);
@@ -76,7 +76,7 @@ class LoginService with LogMixin {
Future<String?> oAuthLogin() async {
const String oAuthCallbackSchema = 'app.immich';
final oAuthApi = di<ImmichApiClient>().getOAuthApi();
final oAuthApi = di<ImApiClient>().getOAuthApi();
try {
final oAuthUrl = await oAuthApi.startOAuth(
@@ -125,7 +125,7 @@ class LoginService with LogMixin {
}
/// Set token to interceptor
await di<ImmichApiClient>().init(accessToken: accessToken);
await di<ImApiClient>().init(accessToken: accessToken);
final user = await di<UserService>().getMyUser().timeout(
const Duration(seconds: 10),