#!/usr/bin/env bun
/**
 * One-time ETL: imajin-media-gallery -> @mac-sync.
 *
 * Copies devices, albums, photos (completed only), album_photos, and
 * identities from imajin's Postgres + MinIO into @mac-sync's. Device UUIDs
 * are re-minted; storage keys are remapped to point at the new device IDs.
 *
 * Idempotent / resumable via a checkpoint file at ./data/etl-imajin-state.json.
 *
 * Buckets must allow anonymous read/write for the duration of the run; the
 * default storage helpers do not sign requests.
 */

import { Pool } from 'pg';
import { randomUUIDv7 } from 'bun';

import { openDb, getDb, pgAll } from '@/shared/db';
import { upsertDeviceByHardwareId } from '@/entities/device';
import { upsertAlbum, setAlbumCoverPhoto } from '@/entities/album';
import { upsertPhoto } from '@/entities/photo';
import { attachPhotoToAlbumsBulk } from '@/entities/photo-album';
import type { PhotoMediaKind } from '@/entities/photo';
import { logger } from '@/shared/logger';
import { headBlob, readBlob, writeBlob, type MinioOverride } from '@/features/iphoto/storage';

import {
  loadCheckpoint,
  saveCheckpoint,
  markStep,
  isStepCompleted,
  type CheckpointState,
} from './lib/etl-checkpoint';

interface Args {
  imajinPgUrl: string;
  imajinMinioUrl: string;
  imajinMinioBucket: string;
  imajinMinioKey: string;
  imajinMinioSecret: string;
  dryRun: boolean;
  skipBlobs: boolean;
  concurrency: number;
  checkpoint: string;
}

function parseArgs(): Args {
  const a: Partial<Args> = { dryRun: false, skipBlobs: false, concurrency: 8 };
  for (const arg of process.argv.slice(2)) {
    if (arg === '--dry-run') a.dryRun = true;
    else if (arg === '--skip-blobs') a.skipBlobs = true;
    else if (arg.startsWith('--imajin-pg-url=')) a.imajinPgUrl = arg.slice('--imajin-pg-url='.length);
    else if (arg.startsWith('--imajin-minio-url=')) a.imajinMinioUrl = arg.slice('--imajin-minio-url='.length);
    else if (arg.startsWith('--imajin-minio-bucket=')) a.imajinMinioBucket = arg.slice('--imajin-minio-bucket='.length);
    else if (arg.startsWith('--imajin-minio-key=')) a.imajinMinioKey = arg.slice('--imajin-minio-key='.length);
    else if (arg.startsWith('--imajin-minio-secret=')) a.imajinMinioSecret = arg.slice('--imajin-minio-secret='.length);
    else if (arg.startsWith('--concurrency=')) a.concurrency = Number(arg.slice('--concurrency='.length));
    else if (arg.startsWith('--checkpoint=')) a.checkpoint = arg.slice('--checkpoint='.length);
    else throw new Error(`unknown arg: ${arg}`);
  }
  const required: Array<keyof Args> = [
    'imajinPgUrl',
    'imajinMinioUrl',
    'imajinMinioBucket',
    'imajinMinioKey',
    'imajinMinioSecret',
  ];
  for (const k of required) {
    if (!a[k]) throw new Error(`missing required arg: --${k.replace(/[A-Z]/g, (m) => `-${m.toLowerCase()}`)}`);
  }
  a.checkpoint = a.checkpoint ?? './data/etl-imajin-state.json';
  return a as Args;
}

interface ImajinDeviceRow {
  id: string;
  name: string;
  hardware_id: string;
  platform: string;
  os_version: string | null;
  is_active: boolean;
}

interface ImajinAlbumRow {
  id: string;
  device_id: string;
  local_identifier: string;
  title: string;
  album_type: string;
  cover_photo_id: string | null;
  start_date: string | null;
  end_date: string | null;
  sort_order: number;
}

interface ImajinPhotoRow {
  id: string;
  device_id: string;
  local_identifier: string;
  media_type: string;
  width: number;
  height: number;
  file_size: number | null;
  duration_seconds: number | null;
  captured_at: string;
  modified_at: string | null;
  imported_at: string;
  storage_key: string | null;
  thumbnail_key: string | null;
  preview_key: string | null;
  original_filename: string | null;
  mime_type: string | null;
  latitude: number | null;
  longitude: number | null;
  location_name: string | null;
  is_favorite: boolean;
  is_hidden: boolean;
  is_screenshot: boolean;
  is_selfie: boolean;
  is_burst: boolean;
  burst_identifier: string | null;
  exif: Record<string, unknown> | null;
  content_hash: string | null;
  processing_status: string;
  category: string | null;
  classification_status: string;
  semantic_tags: unknown;
  face_extraction_status: string;
  face_count: number | null;
}

interface ImajinAlbumPhotoRow {
  album_id: string;
  photo_id: string;
}

interface ImajinIdentityRow {
  id: string;
  name: string | null;
  photo_count: number;
  cover_photo_id: string | null;
  centroid_status: string;
  centroid_photo_count: number;
}

const PHOTO_BATCH = 200;
const JUNCTION_BATCH = 1000;

async function withRetries<T>(label: string, n: number, fn: () => Promise<T>): Promise<T> {
  let lastErr: unknown;
  for (let i = 0; i < n; i++) {
    try {
      return await fn();
    } catch (err) {
      lastErr = err;
      const wait = Math.min(8_000, 250 * 2 ** i);
      logger.warn(`etl: ${label} retry ${i + 1}/${n} in ${wait}ms`, {
        err: err instanceof Error ? err.message : String(err),
      });
      await new Promise((r) => setTimeout(r, wait));
    }
  }
  throw lastErr;
}

function remapStorageKey(oldKey: string, oldDeviceId: string, newDeviceId: string): string {
  return oldKey.split(`/${oldDeviceId}/`).join(`/${newDeviceId}/`);
}

function mediaKindFromImajin(imajinMediaType: string): PhotoMediaKind {
  if (imajinMediaType === 'video') return 'video';
  if (imajinMediaType === 'live_photo') return 'live_photo';
  if (imajinMediaType === 'audio') return 'audio';
  return 'image';
}

async function copyDevices(source: Pool, state: CheckpointState, dryRun: boolean): Promise<void> {
  if (isStepCompleted(state, 'devices')) {
    logger.info('etl: devices step already completed', { count: Object.keys(state.deviceMap).length });
    return;
  }
  markStep(state, 'devices', { status: 'in_progress' });

  const rows = await pgAll<ImajinDeviceRow>(
    source,
    'SELECT id, name, hardware_id, platform, os_version, is_active FROM devices ORDER BY created_at',
  );
  logger.info('etl: devices', { count: rows.length, dryRun });

  if (dryRun) {
    for (const r of rows) {
      state.deviceMap[r.id] = state.deviceMap[r.id] ?? `(dry-run-${r.hardware_id})`;
    }
    markStep(state, 'devices', { status: 'completed', n: rows.length });
    return;
  }

  const target = getDb();
  for (const r of rows) {
    if (state.deviceMap[r.id]) continue;
    // Synthetic token — the imajin source has none usable here, and the
    // devices table has UNIQUE(token), so empty strings would collide.
    const token = `etl-imajin-${r.hardware_id}-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
    const { device } = await upsertDeviceByHardwareId(target, {
      id: randomUUIDv7(),
      name: r.name,
      hardwareId: r.hardware_id,
      token,
      platform: r.platform,
    });
    state.deviceMap[r.id] = device.id;
  }
  markStep(state, 'devices', { status: 'completed', n: rows.length });
}

async function copyAlbums(source: Pool, state: CheckpointState, dryRun: boolean): Promise<void> {
  if (isStepCompleted(state, 'albums')) {
    logger.info('etl: albums step already completed');
    return;
  }
  markStep(state, 'albums', { status: 'in_progress' });

  const rows = await pgAll<ImajinAlbumRow>(
    source,
    `SELECT id, device_id, local_identifier, title, album_type, cover_photo_id, start_date, end_date, sort_order
     FROM albums ORDER BY device_id, sort_order, title`,
  );
  logger.info('etl: albums', { count: rows.length, dryRun });

  if (dryRun) {
    for (const r of rows) state.albumMap[r.id] = state.albumMap[r.id] ?? `(dry-run)`;
    markStep(state, 'albums', { status: 'completed', n: rows.length });
    return;
  }

  const target = getDb();
  for (const r of rows) {
    if (state.albumMap[r.id]) continue;
    const newDeviceId = state.deviceMap[r.device_id];
    if (!newDeviceId) throw new Error(`unmapped device ${r.device_id} for album ${r.id}`);
    const newId = randomUUIDv7();
    await upsertAlbum(target, {
      id: newId,
      deviceId: newDeviceId,
      externalId: r.local_identifier,
      name: r.title,
      albumType: (r.album_type as 'user' | 'smart' | 'shared' | 'system') ?? 'user',
      startDate: r.start_date,
      endDate: r.end_date,
      coverPhotoId: null,
      sortOrder: r.sort_order,
    });
    state.albumMap[r.id] = newId;
  }
  markStep(state, 'albums', { status: 'completed', n: rows.length });
}

async function copyPhotos(
  source: Pool,
  state: CheckpointState,
  dryRun: boolean,
  checkpointPath: string,
): Promise<void> {
  if (isStepCompleted(state, 'photos')) {
    logger.info('etl: photos step already completed', { mapped: Object.keys(state.photoMap).length });
    return;
  }
  markStep(state, 'photos', { status: 'in_progress' });

  const total =
    (await pgAll<{ n: number }>(source, `SELECT COUNT(*)::int AS n FROM photos WHERE processing_status = 'completed'`))[0]
      ?.n ?? 0;
  logger.info('etl: photos to copy', { total, dryRun });

  let offset = 0;
  let processed = state.steps['photos']?.n ?? 0;
  const startTs = Date.now();

  while (true) {
    const rows = await pgAll<ImajinPhotoRow>(
      source,
      `SELECT id, device_id, local_identifier, media_type, width, height, file_size, duration_seconds,
              captured_at, modified_at, imported_at, storage_key, thumbnail_key, preview_key,
              original_filename, mime_type, latitude, longitude, location_name,
              is_favorite, is_hidden, is_screenshot, is_selfie, is_burst, burst_identifier,
              exif, content_hash, processing_status, category, classification_status, semantic_tags,
              face_extraction_status, face_count
         FROM photos
         WHERE processing_status = 'completed'
         ORDER BY id
         LIMIT $1 OFFSET $2`,
      [PHOTO_BATCH, offset],
    );
    if (rows.length === 0) break;

    if (!dryRun) {
      const target = getDb();
      for (const r of rows) {
        if (state.photoMap[r.id]) continue;
        const newDeviceId = state.deviceMap[r.device_id];
        if (!newDeviceId) throw new Error(`unmapped device ${r.device_id} for photo ${r.id}`);
        const newId = randomUUIDv7();
        const newStorageKey = r.storage_key ? remapStorageKey(r.storage_key, r.device_id, newDeviceId) : null;
        const newThumbKey = r.thumbnail_key ? remapStorageKey(r.thumbnail_key, r.device_id, newDeviceId) : null;
        const newPreviewKey = r.preview_key ? remapStorageKey(r.preview_key, r.device_id, newDeviceId) : null;

        await upsertPhoto(target, {
          id: newId,
          deviceId: newDeviceId,
          externalId: r.local_identifier,
          filename: r.original_filename ?? '',
          mediaType: r.mime_type ?? 'application/octet-stream',
          mediaKind: mediaKindFromImajin(r.media_type),
          mimeType: r.mime_type,
          width: r.width,
          height: r.height,
          duration: r.duration_seconds,
          takenAt: r.captured_at,
          modifiedAt: r.modified_at,
          importedAt: r.imported_at,
          storageBucket: 'mac-sync',
          storageKey: newStorageKey,
          thumbnailKey: newThumbKey,
          previewKey: newPreviewKey,
          latitude: r.latitude,
          longitude: r.longitude,
          locationName: r.location_name,
          isFavorite: r.is_favorite,
          isHidden: r.is_hidden,
          isScreenshot: r.is_screenshot,
          isSelfie: r.is_selfie,
          isBurst: r.is_burst,
          burstIdentifier: r.burst_identifier,
          exif: r.exif,
          contentHash: r.content_hash,
          processingStatus: 'completed',
          classificationStatus:
            (r.classification_status as 'pending' | 'completed' | 'processing' | 'failed') ?? 'pending',
          faceExtractionStatus: 'pending',
          faceCount: r.face_count ?? 0,
          category: r.category,
          semanticTags: Array.isArray(r.semantic_tags) ? (r.semantic_tags as string[]) : null,
        });
        state.photoMap[r.id] = newId;
      }
    }
    processed += rows.length;
    offset += rows.length;

    const elapsed = (Date.now() - startTs) / 1000;
    const rate = processed / Math.max(1, elapsed);
    logger.info('etl: photos progress', {
      processed,
      total,
      pct: Math.round((processed / Math.max(1, total)) * 100),
      ratePerSec: rate.toFixed(1),
    });
    markStep(state, 'photos', { status: 'in_progress', n: processed });
    await saveCheckpoint(checkpointPath, state);
  }

  markStep(state, 'photos', { status: 'completed', n: processed });
}

async function backfillAlbumCovers(source: Pool, state: CheckpointState, dryRun: boolean): Promise<void> {
  if (isStepCompleted(state, 'cover_backfill')) return;
  markStep(state, 'cover_backfill', { status: 'in_progress' });
  if (dryRun) {
    markStep(state, 'cover_backfill', { status: 'completed' });
    return;
  }
  const rows = await pgAll<{ id: string; cover_photo_id: string | null }>(
    source,
    'SELECT id, cover_photo_id FROM albums WHERE cover_photo_id IS NOT NULL',
  );
  const target = getDb();
  for (const r of rows) {
    const newAlbumId = state.albumMap[r.id];
    if (!newAlbumId) continue;
    const newCoverId = r.cover_photo_id ? state.photoMap[r.cover_photo_id] : null;
    if (!newCoverId) continue;
    await setAlbumCoverPhoto(target, newAlbumId, newCoverId);
  }
  markStep(state, 'cover_backfill', { status: 'completed', n: rows.length });
}

async function copyAlbumPhotos(
  source: Pool,
  state: CheckpointState,
  dryRun: boolean,
  checkpointPath: string,
): Promise<void> {
  if (isStepCompleted(state, 'album_photos')) return;
  markStep(state, 'album_photos', { status: 'in_progress' });

  const total = (await pgAll<{ n: number }>(source, 'SELECT COUNT(*)::int AS n FROM album_photos'))[0]?.n ?? 0;
  logger.info('etl: album_photos to copy', { total, dryRun });

  if (dryRun) {
    markStep(state, 'album_photos', { status: 'completed', n: total });
    return;
  }

  let offset = 0;
  let inserted = 0;
  let dropped = 0;
  const target = getDb();
  while (true) {
    const rows = await pgAll<ImajinAlbumPhotoRow>(
      source,
      'SELECT album_id, photo_id FROM album_photos ORDER BY album_id, photo_id LIMIT $1 OFFSET $2',
      [JUNCTION_BATCH, offset],
    );
    if (rows.length === 0) break;

    const pairs: Array<{ photoId: string; albumId: string }> = [];
    for (const r of rows) {
      const newPhotoId = state.photoMap[r.photo_id];
      const newAlbumId = state.albumMap[r.album_id];
      if (!newPhotoId || !newAlbumId) {
        dropped++;
        continue;
      }
      pairs.push({ photoId: newPhotoId, albumId: newAlbumId });
    }
    if (pairs.length > 0) await attachPhotoToAlbumsBulk(target, pairs);
    inserted += pairs.length;
    offset += rows.length;

    logger.info('etl: album_photos progress', { inserted, dropped, total });
    markStep(state, 'album_photos', { status: 'in_progress', n: inserted });
    await saveCheckpoint(checkpointPath, state);
  }
  markStep(state, 'album_photos', { status: 'completed', n: inserted });
}

async function copyIdentities(source: Pool, state: CheckpointState, dryRun: boolean): Promise<void> {
  if (isStepCompleted(state, 'identities')) return;
  markStep(state, 'identities', { status: 'in_progress' });

  const rows = await pgAll<ImajinIdentityRow>(
    source,
    'SELECT id, name, photo_count, cover_photo_id, centroid_status, centroid_photo_count FROM identities ORDER BY created_at',
  );
  logger.info('etl: identities', { count: rows.length, dryRun });

  if (dryRun) {
    markStep(state, 'identities', { status: 'completed', n: rows.length });
    return;
  }

  const firstDeviceId = Object.values(state.deviceMap)[0];
  if (!firstDeviceId) throw new Error('no devices migrated; cannot anchor identities');

  const target = getDb();
  for (const r of rows) {
    const id = randomUUIDv7();
    await target.query(
      `INSERT INTO macsync.identities (id, device_id, name, face_count, centroid_embedding, centroid_updated_at)
       VALUES ($1, $2, $3, $4, NULL, NULL)`,
      [id, firstDeviceId, r.name, r.photo_count],
    );
  }
  markStep(state, 'identities', { status: 'completed', n: rows.length });
}

async function copyBlobs(
  source: Pool,
  state: CheckpointState,
  src: MinioOverride,
  dryRun: boolean,
  concurrency: number,
  checkpointPath: string,
): Promise<void> {
  if (isStepCompleted(state, 'blobs')) return;
  markStep(state, 'blobs', { status: 'in_progress' });

  const photoRows = await pgAll<{
    id: string;
    device_id: string;
    storage_key: string | null;
    thumbnail_key: string | null;
    preview_key: string | null;
    mime_type: string | null;
  }>(
    source,
    `SELECT id, device_id, storage_key, thumbnail_key, preview_key, mime_type
       FROM photos WHERE processing_status = 'completed' ORDER BY id`,
  );

  interface BlobTask {
    oldKey: string;
    newKey: string;
    contentType: string;
  }
  const tasks: BlobTask[] = [];
  for (const r of photoRows) {
    if (!state.photoMap[r.id]) continue;
    const newDevice = state.deviceMap[r.device_id];
    if (!newDevice) continue;
    if (r.storage_key) {
      tasks.push({
        oldKey: r.storage_key,
        newKey: remapStorageKey(r.storage_key, r.device_id, newDevice),
        contentType: r.mime_type ?? 'application/octet-stream',
      });
    }
    if (r.thumbnail_key) {
      tasks.push({
        oldKey: r.thumbnail_key,
        newKey: remapStorageKey(r.thumbnail_key, r.device_id, newDevice),
        contentType: 'image/webp',
      });
    }
    if (r.preview_key) {
      tasks.push({
        oldKey: r.preview_key,
        newKey: remapStorageKey(r.preview_key, r.device_id, newDevice),
        contentType: 'image/webp',
      });
    }
  }

  logger.info('etl: blobs to copy', { tasks: tasks.length, dryRun, concurrency });
  if (dryRun) {
    markStep(state, 'blobs', { status: 'completed', n: tasks.length });
    return;
  }

  let done = 0;
  let bytes = 0;
  const startTs = Date.now();

  const worker = async (slot: number): Promise<void> => {
    while (true) {
      const i = done + slot;
      if (i >= tasks.length) return;
      const task = tasks[i];
      if (!task) return;
      done++;
      try {
        if (await headBlob(task.newKey)) {
          continue;
        }
        const buf = await withRetries(`read ${task.oldKey}`, 3, () => readBlob(task.oldKey, src));
        await withRetries(`write ${task.newKey}`, 3, () => writeBlob(task.newKey, task.contentType, buf));
        bytes += buf.byteLength;
      } catch (err) {
        logger.error('etl: blob copy failed', { task, err: err instanceof Error ? err.message : String(err) });
        throw err;
      }
      if (done % 100 === 0) {
        const elapsed = (Date.now() - startTs) / 1000;
        logger.info('etl: blobs progress', {
          done,
          total: tasks.length,
          pct: Math.round((done / tasks.length) * 100),
          mbps: (bytes / 1024 / 1024 / Math.max(1, elapsed)).toFixed(2),
        });
        // Defer checkpoint writes to the final save in main()'s finally block.
        // Concurrent workers used to race here and saveCheckpoint's tmp→rename
        // fails for the losers. headBlob already provides per-object resume.
        markStep(state, 'blobs', { status: 'in_progress', n: done });
      }
    }
  };

  await Promise.all(Array.from({ length: concurrency }, (_, i) => worker(i)));
  markStep(state, 'blobs', { status: 'completed', n: tasks.length });
}

async function verify(source: Pool): Promise<void> {
  const target = getDb();
  const srcPhotos =
    (await pgAll<{ n: number }>(source, `SELECT COUNT(*)::int AS n FROM photos WHERE processing_status='completed'`))[0]
      ?.n ?? 0;
  const dstPhotos =
    (
      await pgAll<{ n: number }>(target, `SELECT COUNT(*)::int AS n FROM macsync.photos WHERE processing_status='completed'`)
    )[0]?.n ?? 0;
  const srcAlbums = (await pgAll<{ n: number }>(source, `SELECT COUNT(*)::int AS n FROM albums`))[0]?.n ?? 0;
  const dstAlbums = (await pgAll<{ n: number }>(target, `SELECT COUNT(*)::int AS n FROM macsync.albums`))[0]?.n ?? 0;
  const srcJunction = (await pgAll<{ n: number }>(source, `SELECT COUNT(*)::int AS n FROM album_photos`))[0]?.n ?? 0;
  const dstJunction =
    (await pgAll<{ n: number }>(target, `SELECT COUNT(*)::int AS n FROM macsync.photo_albums`))[0]?.n ?? 0;
  logger.info('etl: verification', {
    photos: { source: srcPhotos, target: dstPhotos, ok: dstPhotos >= srcPhotos },
    albums: { source: srcAlbums, target: dstAlbums, ok: dstAlbums >= srcAlbums },
    junction: { source: srcJunction, target: dstJunction, ok: dstJunction <= srcJunction },
  });
}

async function main(): Promise<void> {
  const args = parseArgs();
  logger.info('etl: starting', { dryRun: args.dryRun, skipBlobs: args.skipBlobs, checkpoint: args.checkpoint });

  const source = new Pool({ connectionString: args.imajinPgUrl, max: 2 });
  await openDb('macsync');

  const src: MinioOverride = {
    endpoint: args.imajinMinioUrl,
    accessKey: args.imajinMinioKey,
    secretKey: args.imajinMinioSecret,
    bucket: args.imajinMinioBucket,
  };

  const state = await loadCheckpoint(args.checkpoint);
  logger.info('etl: checkpoint loaded', { steps: state.steps });

  try {
    await copyDevices(source, state, args.dryRun);
    await saveCheckpoint(args.checkpoint, state);

    await copyAlbums(source, state, args.dryRun);
    await saveCheckpoint(args.checkpoint, state);

    await copyPhotos(source, state, args.dryRun, args.checkpoint);
    await saveCheckpoint(args.checkpoint, state);

    await backfillAlbumCovers(source, state, args.dryRun);
    await saveCheckpoint(args.checkpoint, state);

    await copyAlbumPhotos(source, state, args.dryRun, args.checkpoint);
    await saveCheckpoint(args.checkpoint, state);

    await copyIdentities(source, state, args.dryRun);
    await saveCheckpoint(args.checkpoint, state);

    if (!args.skipBlobs) {
      await copyBlobs(source, state, src, args.dryRun, args.concurrency, args.checkpoint);
      await saveCheckpoint(args.checkpoint, state);
    }

    if (!args.dryRun) {
      await verify(source);
    }
  } finally {
    await source.end();
    await saveCheckpoint(args.checkpoint, state);
  }

  logger.info('etl: done');
}

main().catch((err) => {
  logger.error('etl: fatal', { err: err instanceof Error ? err.stack ?? err.message : String(err) });
  process.exit(1);
});
