immich/mobile/lib/domain/utils/migrate_cloud_ids.dart

import 'dart:async';

import 'package:collection/collection.dart';
import 'package:drift/drift.dart';
import 'package:flutter/foundation.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'package:immich_mobile/constants/constants.dart';
import 'package:immich_mobile/domain/models/asset/asset_metadata.model.dart';
import 'package:immich_mobile/domain/models/asset/base_asset.model.dart';
import 'package:immich_mobile/domain/utils/cloud_id_resolver.dart';
import 'package:immich_mobile/extensions/platform_extensions.dart';
import 'package:immich_mobile/infrastructure/repositories/db.repository.dart';
import 'package:immich_mobile/infrastructure/repositories/local_album.repository.dart';
import 'package:immich_mobile/platform/native_sync_api.g.dart';
import 'package:immich_mobile/providers/api.provider.dart';
import 'package:immich_mobile/providers/infrastructure/cancel.provider.dart';
import 'package:immich_mobile/providers/infrastructure/db.provider.dart';
import 'package:immich_mobile/providers/infrastructure/platform.provider.dart';
import 'package:immich_mobile/providers/infrastructure/sync.provider.dart';
import 'package:immich_mobile/providers/server_info.provider.dart';
import 'package:immich_mobile/providers/user.provider.dart';
import 'package:logging/logging.dart';
// ignore: import_rule_openapi
import 'package:openapi/api.dart' hide AssetVisibility;

const _kDbPageSize = 20000;

const _kUploadBatchSize = 5000;

Future<void> syncCloudIds(ProviderContainer ref) async {
  if (!CurrentPlatform.isIOS) {
    return;
  }
  final logger = Logger('migrateCloudIds');

  final db = ref.read(driftProvider);
  final cancellation = ref.read(cancellationProvider);

  await populateMissingCloudIds(db, ref.read(nativeSyncApiProvider), cancellation);

  final serverInfo = await ref.read(serverInfoProvider.notifier).getServerInfo();
  final canUpdateMetadata = serverInfo.serverVersion.isAtLeast(major: 2, minor: 4);
  if (!canUpdateMetadata) {
    logger.fine('Server version does not support asset metadata updates. Skipping cloudId migration.');
    return;
  }
  final canBulkUpdateMetadata = serverInfo.serverVersion.isAtLeast(major: 2, minor: 5);

  // Wait for remote sync to complete, so we have up-to-date asset metadata entries
  try {
    await ref.read(syncStreamServiceProvider).sync();
  } catch (e, s) {
    logger.fine('Failed to complete remote sync before cloudId migration.', e, s);
    return;
  }

  // Fetch the mapping for backed up assets that have a cloud ID locally but do not have a cloud ID on the server
  final currentUser = ref.read(currentUserProvider);
  if (currentUser == null) {
    logger.warning('Current user is null. Aborting cloudId migration.');
    return;
  }

  final assetApi = ref.read(apiServiceProvider).assetsApi;

  // Process cloud IDs in paginated batches
  await _processCloudIdMappingsInBatches(db, currentUser.id, assetApi, canBulkUpdateMetadata, logger, cancellation);
}

Future<void> _processCloudIdMappingsInBatches(
  Drift drift,
  String userId,
  AssetsApi assetsApi,
  bool canBulkUpdate,
  Logger logger,
  Completer<void> cancellation,
) async {
  String? lastRemoteId;

  while (true) {
    if (cancellation.isCompleted) {
      logger.warning('Cloud ID migration cancelled. Stopping batch processing.');
      break;
    }
    final mappings = await fetchMapping(drift, userId, _kDbPageSize, lastRemoteId);
    if (mappings.isEmpty) {
      break;
    }

    final items = <AssetMetadataBulkUpsertItemDto>[];
    for (final mapping in mappings) {
      items.add(
        .new(
          assetId: mapping.remoteAssetId,
          key: kMobileMetadataKey,
          value: Map<String, Object>.from(
            RemoteAssetMobileAppMetadata(
              cloudId: mapping.cloudId,
              createdAt: mapping.createdAt.toIso8601String(),
              adjustmentTime: mapping.adjustmentTime?.toIso8601String(),
              latitude: mapping.latitude?.toString(),
              longitude: mapping.longitude?.toString(),
            ).toJson(),
          ),
        ),
      );
    }

    if (items.isNotEmpty) {
      if (canBulkUpdate) {
        for (final batch in items.slices(_kUploadBatchSize)) {
          if (cancellation.isCompleted) {
            break;
          }

          await _bulkUpdate(assetsApi, batch, cancellation.future);
        }
      } else {
        await _sequentialUpdate(assetsApi, items, cancellation);
      }
    }

    lastRemoteId = mappings.last.remoteAssetId;
    if (mappings.length < _kDbPageSize) {
      break;
    }
  }
}

@visibleForTesting
Future<void> populateMissingCloudIds(Drift drift, NativeSyncApi nativeSyncApi, Completer<void> cancellation) async {
  final query = drift.localAssetEntity.selectOnly()
    ..addColumns([drift.localAssetEntity.id])
    ..where(drift.localAssetEntity.iCloudId.isNull());
  final ids = await query.map((row) => row.read(drift.localAssetEntity.id)!).get();

  await resolveCloudIds(nativeSyncApi, DriftLocalAlbumRepository(drift), ids, cancellation: cancellation);
}

@visibleForTesting
typedef CloudIdMapping = ({
  String remoteAssetId,
  String cloudId,
  DateTime createdAt,
  DateTime? adjustmentTime,
  double? latitude,
  double? longitude,
});

@visibleForTesting
Future<List<CloudIdMapping>> fetchMapping(Drift db, String userId, int limit, String? lastRemoteId) async {
  final query = db.remoteAssetEntity.selectOnly()
    ..addColumns([
      db.remoteAssetEntity.id,
      db.localAssetEntity.iCloudId,
      db.localAssetEntity.createdAt,
      db.localAssetEntity.adjustmentTime,
      db.localAssetEntity.latitude,
      db.localAssetEntity.longitude,
    ])
    ..join([
      innerJoin(
        db.localAssetEntity,
        db.localAssetEntity.id.isInQuery(
          db.localAssetEntity.selectOnly()
            ..addColumns([db.localAssetEntity.id.min()])
            ..where(db.localAssetEntity.checksum.equalsExp(db.remoteAssetEntity.checksum)),
        ),
        useColumns: false,
      ),
      leftOuterJoin(
        db.remoteAssetCloudIdEntity,
        db.remoteAssetEntity.id.equalsExp(db.remoteAssetCloudIdEntity.assetId),
        useColumns: false,
      ),
    ])
    ..where(
      db.remoteAssetEntity.ownerId.equals(userId) &
          // Skip locked assets as we cannot update them without unlocking first
          db.remoteAssetEntity.visibility.isNotValue(AssetVisibility.locked.index) &
          db.localAssetEntity.iCloudId.isNotNull() &
          // Only select assets that have a local cloud ID but either no remote cloud ID or a mismatched eTag
          (db.remoteAssetCloudIdEntity.cloudId.isNull() |
              db.remoteAssetCloudIdEntity.adjustmentTime.isNotExp(db.localAssetEntity.adjustmentTime) |
              db.remoteAssetCloudIdEntity.latitude.isNotExp(db.localAssetEntity.latitude) |
              db.remoteAssetCloudIdEntity.longitude.isNotExp(db.localAssetEntity.longitude) |
              db.remoteAssetCloudIdEntity.createdAt.isNotExp(db.localAssetEntity.createdAt)),
    )
    ..orderBy([.asc(db.remoteAssetEntity.id)])
    ..limit(limit);

  if (lastRemoteId != null) {
    query.where(db.remoteAssetEntity.id.isBiggerThanValue(lastRemoteId));
  }

  return query.map((row) {
    return (
      remoteAssetId: row.read(db.remoteAssetEntity.id)!,
      cloudId: row.read(db.localAssetEntity.iCloudId)!,
      createdAt: row.read(db.localAssetEntity.createdAt)!,
      adjustmentTime: row.read(db.localAssetEntity.adjustmentTime),
      latitude: row.read(db.localAssetEntity.latitude),
      longitude: row.read(db.localAssetEntity.longitude),
    );
  }).get();
}

Future<void> _sequentialUpdate(
  AssetsApi assetsApi,
  List<AssetMetadataBulkUpsertItemDto> items,
  Completer<void> cancellation,
) async {
  for (final item in items) {
    if (cancellation.isCompleted) {
      break;
    }
    try {
      await assetsApi.updateAssetMetadata(
        item.assetId,
        .new(
          items: [.new(key: item.key, value: item.value)],
        ),
        abortTrigger: cancellation.future,
      );
    } catch (error, stack) {
      Logger('migrateCloudIds').warning('Failed to update metadata for asset ${item.assetId}', error, stack);
    }
  }
}

Future<void> _bulkUpdate(
  AssetsApi assetsApi,
  List<AssetMetadataBulkUpsertItemDto> items,
  Future<void> abortTrigger,
) async {
  try {
    await assetsApi.updateBulkAssetMetadata(.new(items: items), abortTrigger: abortTrigger);
  } catch (error, stack) {
    Logger('migrateCloudIds').warning('Failed to bulk update metadata', error, stack);
  }
}