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

import 'dart:async';
import 'dart:math' as math;

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/entities/local_asset.entity.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? lastLocalId;
  final seenRemoteAssetIds = <String>{};

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

    final items = <AssetMetadataBulkUpsertItemDto>[];
    for (final mapping in mappings) {
      if (!seenRemoteAssetIds.add(mapping.remoteAssetId)) {
        logger.fine('Duplicate remote asset ID found: ${mapping.remoteAssetId}. Skipping duplicate entry.');
        continue;
      }

      items.add(
        .new(
          assetId: mapping.remoteAssetId,
          key: kMobileMetadataKey,
          value: Map<String, Object>.from(
            RemoteAssetMobileAppMetadata(
              cloudId: mapping.localAsset.cloudId,
              createdAt: mapping.localAsset.createdAt.toIso8601String(),
              adjustmentTime: mapping.localAsset.adjustmentTime?.toIso8601String(),
              latitude: mapping.localAsset.latitude?.toString(),
              longitude: mapping.localAsset.longitude?.toString(),
            ).toJson(),
          ),
        ),
      );
    }

    if (items.isNotEmpty) {
      if (canBulkUpdate) {
        for (int i = 0; i < items.length; i += _kUploadBatchSize) {
          if (cancellation.isCompleted) {
            break;
          }
          final end = math.min(i + _kUploadBatchSize, items.length);
          await _bulkUpdate(assetsApi, items.sublist(i, end), cancellation.future);
        }
      } else {
        await _sequentialUpdate(assetsApi, items, cancellation);
      }
    }

    lastLocalId = mappings.last.localAsset.id;
    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);
}

typedef _CloudIdMapping = ({String remoteAssetId, LocalAsset localAsset});

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

  if (lastLocalId != null) {
    query.where(drift.localAssetEntity.id.isBiggerThanValue(lastLocalId));
  }

  return query.map((row) {
    return (
      remoteAssetId: row.read(drift.remoteAssetEntity.id)!,
      localAsset: row.readTable(drift.localAssetEntity).toDto(),
    );
  }).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);
  }
}