mirror of
https://github.com/immich-app/immich.git
synced 2026-05-18 03:10:24 +03:00
merge: remote-tracking branch 'origin/main' into feat/integrity-checks-izzy
This commit is contained in:
izzy
@@ -0,0 +1,33 @@
|
||||
import 'package:cancellation_token_http/http.dart';
|
||||
import 'package:hooks_riverpod/hooks_riverpod.dart';
|
||||
|
||||
/// Tracks per-asset upload progress.
|
||||
/// Key: local asset ID, Value: upload progress 0.0 to 1.0, or -1.0 for error
|
||||
class AssetUploadProgressNotifier extends Notifier<Map<String, double>> {
|
||||
static const double errorValue = -1.0;
|
||||
|
||||
@override
|
||||
Map<String, double> build() => {};
|
||||
|
||||
void setProgress(String localAssetId, double progress) {
|
||||
state = {...state, localAssetId: progress};
|
||||
}
|
||||
|
||||
void setError(String localAssetId) {
|
||||
state = {...state, localAssetId: errorValue};
|
||||
}
|
||||
|
||||
void remove(String localAssetId) {
|
||||
state = Map.from(state)..remove(localAssetId);
|
||||
}
|
||||
|
||||
void clear() {
|
||||
state = {};
|
||||
}
|
||||
}
|
||||
|
||||
final assetUploadProgressProvider = NotifierProvider<AssetUploadProgressNotifier, Map<String, double>>(
|
||||
AssetUploadProgressNotifier.new,
|
||||
);
|
||||
|
||||
final manualUploadCancelTokenProvider = StateProvider<CancellationToken?>((ref) => null);
|
||||
@@ -1,19 +1,18 @@
|
||||
// ignore_for_file: public_member_api_docs, sort_constructors_first
|
||||
import 'dart:async';
|
||||
|
||||
import 'package:background_downloader/background_downloader.dart';
|
||||
import 'package:cancellation_token_http/http.dart';
|
||||
import 'package:collection/collection.dart';
|
||||
import 'package:hooks_riverpod/hooks_riverpod.dart';
|
||||
import 'package:logging/logging.dart';
|
||||
|
||||
import 'package:immich_mobile/constants/constants.dart';
|
||||
import 'package:immich_mobile/domain/models/album/local_album.model.dart';
|
||||
import 'package:immich_mobile/domain/models/asset/base_asset.model.dart';
|
||||
import 'package:immich_mobile/extensions/string_extensions.dart';
|
||||
import 'package:immich_mobile/infrastructure/repositories/backup.repository.dart';
|
||||
import 'package:immich_mobile/utils/upload_speed_calculator.dart';
|
||||
import 'package:immich_mobile/providers/infrastructure/asset.provider.dart';
|
||||
import 'package:immich_mobile/providers/user.provider.dart';
|
||||
import 'package:immich_mobile/services/upload.service.dart';
|
||||
import 'package:immich_mobile/utils/debug_print.dart';
|
||||
import 'package:logging/logging.dart';
|
||||
import 'package:immich_mobile/services/foreground_upload.service.dart';
|
||||
import 'package:immich_mobile/services/background_upload.service.dart';
|
||||
|
||||
class EnqueueStatus {
|
||||
final int enqueueCount;
|
||||
@@ -106,26 +105,24 @@ class DriftBackupState {
|
||||
final int remainderCount;
|
||||
final int processingCount;
|
||||
|
||||
final int enqueueCount;
|
||||
final int enqueueTotalCount;
|
||||
|
||||
final bool isSyncing;
|
||||
final bool isCanceling;
|
||||
final BackupError error;
|
||||
|
||||
final Map<String, DriftUploadStatus> uploadItems;
|
||||
final CancellationToken? cancelToken;
|
||||
|
||||
final Map<String, double> iCloudDownloadProgress;
|
||||
|
||||
const DriftBackupState({
|
||||
required this.totalCount,
|
||||
required this.backupCount,
|
||||
required this.remainderCount,
|
||||
required this.processingCount,
|
||||
required this.enqueueCount,
|
||||
required this.enqueueTotalCount,
|
||||
required this.isCanceling,
|
||||
required this.isSyncing,
|
||||
required this.uploadItems,
|
||||
this.error = BackupError.none,
|
||||
required this.uploadItems,
|
||||
this.cancelToken,
|
||||
this.iCloudDownloadProgress = const {},
|
||||
});
|
||||
|
||||
DriftBackupState copyWith({
|
||||
@@ -133,30 +130,28 @@ class DriftBackupState {
|
||||
int? backupCount,
|
||||
int? remainderCount,
|
||||
int? processingCount,
|
||||
int? enqueueCount,
|
||||
int? enqueueTotalCount,
|
||||
bool? isCanceling,
|
||||
bool? isSyncing,
|
||||
Map<String, DriftUploadStatus>? uploadItems,
|
||||
BackupError? error,
|
||||
Map<String, DriftUploadStatus>? uploadItems,
|
||||
CancellationToken? cancelToken,
|
||||
Map<String, double>? iCloudDownloadProgress,
|
||||
}) {
|
||||
return DriftBackupState(
|
||||
totalCount: totalCount ?? this.totalCount,
|
||||
backupCount: backupCount ?? this.backupCount,
|
||||
remainderCount: remainderCount ?? this.remainderCount,
|
||||
processingCount: processingCount ?? this.processingCount,
|
||||
enqueueCount: enqueueCount ?? this.enqueueCount,
|
||||
enqueueTotalCount: enqueueTotalCount ?? this.enqueueTotalCount,
|
||||
isCanceling: isCanceling ?? this.isCanceling,
|
||||
isSyncing: isSyncing ?? this.isSyncing,
|
||||
uploadItems: uploadItems ?? this.uploadItems,
|
||||
error: error ?? this.error,
|
||||
uploadItems: uploadItems ?? this.uploadItems,
|
||||
cancelToken: cancelToken ?? this.cancelToken,
|
||||
iCloudDownloadProgress: iCloudDownloadProgress ?? this.iCloudDownloadProgress,
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
String toString() {
|
||||
return 'DriftBackupState(totalCount: $totalCount, backupCount: $backupCount, remainderCount: $remainderCount, processingCount: $processingCount, enqueueCount: $enqueueCount, enqueueTotalCount: $enqueueTotalCount, isCanceling: $isCanceling, isSyncing: $isSyncing, uploadItems: $uploadItems, error: $error)';
|
||||
return 'DriftBackupState(totalCount: $totalCount, backupCount: $backupCount, remainderCount: $remainderCount, processingCount: $processingCount, isSyncing: $isSyncing, error: $error, uploadItems: $uploadItems, cancelToken: $cancelToken, iCloudDownloadProgress: $iCloudDownloadProgress)';
|
||||
}
|
||||
|
||||
@override
|
||||
@@ -168,12 +163,11 @@ class DriftBackupState {
|
||||
other.backupCount == backupCount &&
|
||||
other.remainderCount == remainderCount &&
|
||||
other.processingCount == processingCount &&
|
||||
other.enqueueCount == enqueueCount &&
|
||||
other.enqueueTotalCount == enqueueTotalCount &&
|
||||
other.isCanceling == isCanceling &&
|
||||
other.isSyncing == isSyncing &&
|
||||
other.error == error &&
|
||||
mapEquals(other.iCloudDownloadProgress, iCloudDownloadProgress) &&
|
||||
mapEquals(other.uploadItems, uploadItems) &&
|
||||
other.error == error;
|
||||
other.cancelToken == cancelToken;
|
||||
}
|
||||
|
||||
@override
|
||||
@@ -182,44 +176,40 @@ class DriftBackupState {
|
||||
backupCount.hashCode ^
|
||||
remainderCount.hashCode ^
|
||||
processingCount.hashCode ^
|
||||
enqueueCount.hashCode ^
|
||||
enqueueTotalCount.hashCode ^
|
||||
isCanceling.hashCode ^
|
||||
isSyncing.hashCode ^
|
||||
error.hashCode ^
|
||||
uploadItems.hashCode ^
|
||||
error.hashCode;
|
||||
cancelToken.hashCode ^
|
||||
iCloudDownloadProgress.hashCode;
|
||||
}
|
||||
}
|
||||
|
||||
final driftBackupProvider = StateNotifierProvider<DriftBackupNotifier, DriftBackupState>((ref) {
|
||||
return DriftBackupNotifier(ref.watch(uploadServiceProvider));
|
||||
return DriftBackupNotifier(
|
||||
ref.watch(foregroundUploadServiceProvider),
|
||||
ref.watch(backgroundUploadServiceProvider),
|
||||
UploadSpeedManager(),
|
||||
);
|
||||
});
|
||||
|
||||
class DriftBackupNotifier extends StateNotifier<DriftBackupState> {
|
||||
DriftBackupNotifier(this._uploadService)
|
||||
DriftBackupNotifier(this._foregroundUploadService, this._backgroundUploadService, this._uploadSpeedManager)
|
||||
: super(
|
||||
const DriftBackupState(
|
||||
totalCount: 0,
|
||||
backupCount: 0,
|
||||
remainderCount: 0,
|
||||
processingCount: 0,
|
||||
enqueueCount: 0,
|
||||
enqueueTotalCount: 0,
|
||||
isCanceling: false,
|
||||
isSyncing: false,
|
||||
uploadItems: {},
|
||||
error: BackupError.none,
|
||||
),
|
||||
) {
|
||||
{
|
||||
_statusSubscription = _uploadService.taskStatusStream.listen(_handleTaskStatusUpdate);
|
||||
_progressSubscription = _uploadService.taskProgressStream.listen(_handleTaskProgressUpdate);
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
final ForegroundUploadService _foregroundUploadService;
|
||||
final BackgroundUploadService _backgroundUploadService;
|
||||
final UploadSpeedManager _uploadSpeedManager;
|
||||
|
||||
final UploadService _uploadService;
|
||||
StreamSubscription<TaskStatusUpdate>? _statusSubscription;
|
||||
StreamSubscription<TaskProgressUpdate>? _progressSubscription;
|
||||
final _logger = Logger("DriftBackupNotifier");
|
||||
|
||||
/// Remove upload item from state
|
||||
@@ -235,120 +225,12 @@ class DriftBackupNotifier extends StateNotifier<DriftBackupState> {
|
||||
}
|
||||
}
|
||||
|
||||
void _handleTaskStatusUpdate(TaskStatusUpdate update) {
|
||||
if (!mounted) {
|
||||
_logger.warning("Skip _handleTaskStatusUpdate: notifier disposed");
|
||||
return;
|
||||
}
|
||||
final taskId = update.task.taskId;
|
||||
|
||||
switch (update.status) {
|
||||
case TaskStatus.complete:
|
||||
if (update.task.group == kBackupGroup) {
|
||||
if (update.responseStatusCode == 201) {
|
||||
state = state.copyWith(backupCount: state.backupCount + 1, remainderCount: state.remainderCount - 1);
|
||||
}
|
||||
}
|
||||
|
||||
// Remove the completed task from the upload items
|
||||
if (state.uploadItems.containsKey(taskId)) {
|
||||
Future.delayed(const Duration(milliseconds: 1000), () {
|
||||
_removeUploadItem(taskId);
|
||||
});
|
||||
}
|
||||
|
||||
case TaskStatus.failed:
|
||||
// Ignore retry errors to avoid confusing users
|
||||
if (update.exception?.description == 'Delayed or retried enqueue failed') {
|
||||
_removeUploadItem(taskId);
|
||||
return;
|
||||
}
|
||||
|
||||
final currentItem = state.uploadItems[taskId];
|
||||
if (currentItem == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
String? error;
|
||||
final exception = update.exception;
|
||||
if (exception != null && exception is TaskHttpException) {
|
||||
final message = tryJsonDecode(exception.description)?['message'] as String?;
|
||||
if (message != null) {
|
||||
final responseCode = exception.httpResponseCode;
|
||||
error = "${exception.exceptionType}, response code $responseCode: $message";
|
||||
}
|
||||
}
|
||||
error ??= update.exception?.toString();
|
||||
|
||||
state = state.copyWith(
|
||||
uploadItems: {
|
||||
...state.uploadItems,
|
||||
taskId: currentItem.copyWith(isFailed: true, error: error),
|
||||
},
|
||||
);
|
||||
_logger.fine("Upload failed for taskId: $taskId, exception: ${update.exception}");
|
||||
break;
|
||||
|
||||
case TaskStatus.canceled:
|
||||
_removeUploadItem(update.task.taskId);
|
||||
break;
|
||||
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
void _handleTaskProgressUpdate(TaskProgressUpdate update) {
|
||||
if (!mounted) {
|
||||
_logger.warning("Skip _handleTaskProgressUpdate: notifier disposed");
|
||||
return;
|
||||
}
|
||||
final taskId = update.task.taskId;
|
||||
final filename = update.task.displayName;
|
||||
final progress = update.progress;
|
||||
final currentItem = state.uploadItems[taskId];
|
||||
if (currentItem != null) {
|
||||
if (progress == kUploadStatusCanceled) {
|
||||
_removeUploadItem(update.task.taskId);
|
||||
return;
|
||||
}
|
||||
|
||||
state = state.copyWith(
|
||||
uploadItems: {
|
||||
...state.uploadItems,
|
||||
taskId: update.hasExpectedFileSize
|
||||
? currentItem.copyWith(
|
||||
progress: progress,
|
||||
fileSize: update.expectedFileSize,
|
||||
networkSpeedAsString: update.networkSpeedAsString,
|
||||
)
|
||||
: currentItem.copyWith(progress: progress),
|
||||
},
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
state = state.copyWith(
|
||||
uploadItems: {
|
||||
...state.uploadItems,
|
||||
taskId: DriftUploadStatus(
|
||||
taskId: taskId,
|
||||
filename: filename,
|
||||
progress: progress,
|
||||
fileSize: update.expectedFileSize,
|
||||
networkSpeedAsString: update.networkSpeedAsString,
|
||||
),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> getBackupStatus(String userId) async {
|
||||
if (!mounted) {
|
||||
_logger.warning("Skip getBackupStatus (pre-call): notifier disposed");
|
||||
return;
|
||||
}
|
||||
final counts = await _uploadService.getBackupCounts(userId);
|
||||
final counts = await _foregroundUploadService.getBackupCounts(userId);
|
||||
if (!mounted) {
|
||||
_logger.warning("Skip getBackupStatus (post-call): notifier disposed");
|
||||
return;
|
||||
@@ -374,47 +256,126 @@ class DriftBackupNotifier extends StateNotifier<DriftBackupState> {
|
||||
state = state.copyWith(isSyncing: isSyncing);
|
||||
}
|
||||
|
||||
Future<void> startBackup(String userId) {
|
||||
Future<void> startForegroundBackup(String userId) async {
|
||||
state = state.copyWith(error: BackupError.none);
|
||||
return _uploadService.startBackup(userId, _updateEnqueueCount);
|
||||
|
||||
final cancelToken = CancellationToken();
|
||||
state = state.copyWith(cancelToken: cancelToken);
|
||||
|
||||
return _foregroundUploadService.uploadCandidates(
|
||||
userId,
|
||||
cancelToken,
|
||||
callbacks: UploadCallbacks(
|
||||
onProgress: _handleForegroundBackupProgress,
|
||||
onSuccess: _handleForegroundBackupSuccess,
|
||||
onError: _handleForegroundBackupError,
|
||||
onICloudProgress: _handleICloudProgress,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
void _updateEnqueueCount(EnqueueStatus status) {
|
||||
state = state.copyWith(enqueueCount: status.enqueueCount, enqueueTotalCount: status.totalCount);
|
||||
Future<void> stopForegroundBackup() async {
|
||||
state.cancelToken?.cancel();
|
||||
_uploadSpeedManager.clear();
|
||||
state = state.copyWith(cancelToken: null, uploadItems: {}, iCloudDownloadProgress: {});
|
||||
}
|
||||
|
||||
Future<void> cancel() async {
|
||||
if (!mounted) {
|
||||
_logger.warning("Skip cancel (pre-call): notifier disposed");
|
||||
return;
|
||||
void _handleICloudProgress(String localAssetId, double progress) {
|
||||
state = state.copyWith(iCloudDownloadProgress: {...state.iCloudDownloadProgress, localAssetId: progress});
|
||||
|
||||
if (progress >= 1.0) {
|
||||
Future.delayed(const Duration(milliseconds: 250), () {
|
||||
final updatedProgress = Map<String, double>.from(state.iCloudDownloadProgress);
|
||||
updatedProgress.remove(localAssetId);
|
||||
state = state.copyWith(iCloudDownloadProgress: updatedProgress);
|
||||
});
|
||||
}
|
||||
dPrint(() => "Canceling backup tasks...");
|
||||
state = state.copyWith(enqueueCount: 0, enqueueTotalCount: 0, isCanceling: true, error: BackupError.none);
|
||||
}
|
||||
|
||||
final activeTaskCount = await _uploadService.cancelBackup();
|
||||
if (!mounted) {
|
||||
_logger.warning("Skip cancel (post-call): notifier disposed");
|
||||
void _handleForegroundBackupProgress(String localAssetId, String filename, int bytes, int totalBytes) {
|
||||
if (state.cancelToken == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (activeTaskCount > 0) {
|
||||
dPrint(() => "$activeTaskCount tasks left, continuing to cancel...");
|
||||
await cancel();
|
||||
final progress = totalBytes > 0 ? bytes / totalBytes : 0.0;
|
||||
final networkSpeedAsString = _uploadSpeedManager.updateProgress(localAssetId, bytes, totalBytes);
|
||||
final currentItem = state.uploadItems[localAssetId];
|
||||
if (currentItem != null) {
|
||||
state = state.copyWith(
|
||||
uploadItems: {
|
||||
...state.uploadItems,
|
||||
localAssetId: currentItem.copyWith(
|
||||
filename: filename,
|
||||
progress: progress,
|
||||
fileSize: totalBytes,
|
||||
networkSpeedAsString: networkSpeedAsString,
|
||||
),
|
||||
},
|
||||
);
|
||||
} else {
|
||||
dPrint(() => "All tasks canceled successfully.");
|
||||
// Clear all upload items when cancellation is complete
|
||||
state = state.copyWith(isCanceling: false, uploadItems: {});
|
||||
state = state.copyWith(
|
||||
uploadItems: {
|
||||
...state.uploadItems,
|
||||
localAssetId: DriftUploadStatus(
|
||||
taskId: localAssetId,
|
||||
filename: filename,
|
||||
progress: progress,
|
||||
fileSize: totalBytes,
|
||||
networkSpeedAsString: networkSpeedAsString,
|
||||
),
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> handleBackupResume(String userId) async {
|
||||
void _handleForegroundBackupSuccess(String localAssetId, String remoteAssetId) {
|
||||
state = state.copyWith(backupCount: state.backupCount + 1, remainderCount: state.remainderCount - 1);
|
||||
_uploadSpeedManager.removeTask(localAssetId);
|
||||
|
||||
Future.delayed(const Duration(milliseconds: 1000), () {
|
||||
_removeUploadItem(localAssetId);
|
||||
});
|
||||
}
|
||||
|
||||
void _handleForegroundBackupError(String localAssetId, String errorMessage) {
|
||||
_logger.severe("Upload failed for $localAssetId: $errorMessage");
|
||||
|
||||
final currentItem = state.uploadItems[localAssetId];
|
||||
if (currentItem != null) {
|
||||
state = state.copyWith(
|
||||
uploadItems: {
|
||||
...state.uploadItems,
|
||||
localAssetId: currentItem.copyWith(isFailed: true, error: errorMessage),
|
||||
},
|
||||
);
|
||||
} else {
|
||||
state = state.copyWith(
|
||||
uploadItems: {
|
||||
...state.uploadItems,
|
||||
localAssetId: DriftUploadStatus(
|
||||
taskId: localAssetId,
|
||||
filename: 'Unknown',
|
||||
progress: 0,
|
||||
fileSize: 0,
|
||||
networkSpeedAsString: '',
|
||||
isFailed: true,
|
||||
error: errorMessage,
|
||||
),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
_uploadSpeedManager.removeTask(localAssetId);
|
||||
}
|
||||
|
||||
Future<void> startBackupWithURLSession(String userId) async {
|
||||
if (!mounted) {
|
||||
_logger.warning("Skip handleBackupResume (pre-call): notifier disposed");
|
||||
return;
|
||||
}
|
||||
_logger.info("Resuming backup tasks...");
|
||||
state = state.copyWith(error: BackupError.none);
|
||||
final tasks = await _uploadService.getActiveTasks(kBackupGroup);
|
||||
final tasks = await _backgroundUploadService.getActiveTasks(kBackupGroup);
|
||||
if (!mounted) {
|
||||
_logger.warning("Skip handleBackupResume (post-call): notifier disposed");
|
||||
return;
|
||||
@@ -422,20 +383,12 @@ class DriftBackupNotifier extends StateNotifier<DriftBackupState> {
|
||||
_logger.info("Found ${tasks.length} tasks");
|
||||
|
||||
if (tasks.isEmpty) {
|
||||
// Start a new backup queue
|
||||
_logger.info("Start a new backup queue");
|
||||
return startBackup(userId);
|
||||
_logger.info("Start backup with URLSession");
|
||||
return _backgroundUploadService.uploadBackupCandidates(userId);
|
||||
}
|
||||
|
||||
_logger.info("Tasks to resume: ${tasks.length}");
|
||||
return _uploadService.resumeBackup();
|
||||
}
|
||||
|
||||
@override
|
||||
void dispose() {
|
||||
_statusSubscription?.cancel();
|
||||
_progressSubscription?.cancel();
|
||||
super.dispose();
|
||||
return _backgroundUploadService.resume();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -445,7 +398,7 @@ final driftBackupCandidateProvider = FutureProvider.autoDispose<List<LocalAsset>
|
||||
return [];
|
||||
}
|
||||
|
||||
return ref.read(backupRepositoryProvider).getCandidates(user.id, onlyHashed: false);
|
||||
return ref.read(foregroundUploadServiceProvider).getBackupCandidates(user.id, onlyHashed: false);
|
||||
});
|
||||
|
||||
final driftCandidateBackupAlbumInfoProvider = FutureProvider.autoDispose.family<List<LocalAlbum>, String>((
|
||||
|
||||
Reference in New Issue
Block a user