/**
 * Image-processing worker (§113 basics): validates and caches product images, records
 * status/dimensions/perceptual hash in `images`, and promotes a replacement hero image when the
 * current one is dead. Shares the fetch/cache core with the web `/img` proxy.
 */
import { and, eq, inArray, isNull, lt, or, sql } from 'drizzle-orm';
import { assets, images } from '@rareindex/database';
import { logger } from '@rareindex/shared';
import sharp from 'sharp';
import { db } from '../lib/db.ts';
import { ensureOriginal, ensureVariant, imageKey, perceptualHash } from './core.ts';

export interface ImagesResult {
  registered: number;
  checked: number;
  ok: number;
  dead: number;
  blocked: number;
  error: number;
  promoted: number;
  placeholders: number;
}

export interface ImagesOptions {
  limit?: number;
  recheck?: boolean;
  assetIds?: string[];
  concurrency?: number;
  /** also warm the 192/384 variants so first paint is instant */
  warm?: boolean;
}

const RECHECK_AFTER_DAYS = 30;
const RETRY_ERROR_AFTER_HOURS = 6;

export async function processImages(opts: ImagesOptions = {}): Promise<ImagesResult> {
  const log = logger.child({ component: 'images' });
  const limit = opts.limit ?? 2000;
  const res: ImagesResult = { registered: 0, checked: 0, ok: 0, dead: 0, blocked: 0, error: 0, promoted: 0, placeholders: 0 };

  // 1. Register hero URLs that have no images row yet (catalog sources may set hero_image_url directly).
  // drizzle serialises a JS array parameter as a row (record), so pass the ids as JSON and expand them server-side.
  const assetFilter = opts.assetIds?.length ? sql` and a.id in (select jsonb_array_elements_text(${JSON.stringify(opts.assetIds)}::jsonb))` : sql``;
  const reg = (await db().execute(sql`
    insert into images (id, asset_id, source_id, url, role, status)
    select 'img_' || substr(md5(a.id || a.hero_image_url), 1, 20), a.id, null, a.hero_image_url, 'hero', 'unchecked'
    from assets a
    where a.hero_image_url is not null and a.hero_image_url ~ '^https?://'
      and not exists (select 1 from images i where i.url = a.hero_image_url)${assetFilter}
    limit ${limit}
    on conflict (url) do nothing
    returning id
  `)) as unknown as unknown[];
  res.registered = reg.length;

  // 2. Pick rows to check.
  const staleCut = new Date(Date.now() - RECHECK_AFTER_DAYS * 86_400_000);
  const errorCut = new Date(Date.now() - RETRY_ERROR_AFTER_HOURS * 3600_000);
  const conds = [
    eq(images.status, 'unchecked'),
    and(eq(images.status, 'error'), lt(images.checkedAt, errorCut)),
    ...(opts.recheck ? [and(eq(images.status, 'ok'), or(isNull(images.checkedAt), lt(images.checkedAt, staleCut)))] : []),
  ];
  const rows = await db()
    .select({ id: images.id, url: images.url, assetId: images.assetId, role: images.role, width: images.width })
    .from(images)
    .where(and(or(...conds), opts.assetIds?.length ? inArray(images.assetId, opts.assetIds) : undefined))
    .orderBy(sql`case ${images.role} when 'hero' then 0 else 1 end`, images.createdAt)
    .limit(limit);

  const deadHeroAssets = new Set<string>();
  await pool(rows, opts.concurrency ?? 6, async (row) => {
    res.checked++;
    const outcome = await checkOne(row.url, opts.warm ?? true);
    if (outcome.status === 'ok') res.ok++;
    else if (outcome.status === 'dead') res.dead++;
    else if (outcome.status === 'blocked') res.blocked++;
    else res.error++;
    await db()
      .update(images)
      .set({ status: outcome.status, checkedAt: new Date(), cacheKey: imageKey(row.url), bytes: outcome.bytes, contentType: outcome.contentType, width: outcome.width ?? row.width ?? null, height: outcome.height ?? null, phash: outcome.phash ?? undefined, error: outcome.error })
      .where(eq(images.id, row.id));
    if (outcome.status !== 'ok' && outcome.status !== 'error' && row.assetId && row.role === 'hero') deadHeroAssets.add(row.assetId);
  });

  // 3. Promote a replacement hero for assets whose hero image is gone.
  for (const assetId of deadHeroAssets) {
    const promoted = await promoteHero(assetId);
    if (promoted) res.promoted++;
    else res.placeholders++;
  }
  log.info(res, 'images processed');
  return res;
}

interface CheckOutcome {
  status: 'ok' | 'dead' | 'blocked' | 'error';
  bytes: number | null;
  contentType: string | null;
  width: number | null;
  height: number | null;
  phash: string | null;
  error: string | null;
}

async function checkOne(url: string, warm: boolean): Promise<CheckOutcome> {
  const out = await ensureOriginal(url);
  if (!out.ok) return { status: out.status, bytes: null, contentType: null, width: null, height: null, phash: null, error: out.reason.slice(0, 200) };
  const info = out.info;
  let width = info.width;
  let height = info.height;
  if (width === null || height === null) {
    try {
      const meta = await sharp(info.path).metadata();
      width = meta.width ?? null;
      height = meta.height ?? null;
    } catch {
      /* keep nulls */
    }
  }
  const phash = await perceptualHash(info.path);
  if (warm) {
    try {
      await ensureVariant(info, 192);
      await ensureVariant(info, 384);
    } catch (err) {
      return { status: 'error', bytes: info.bytes, contentType: info.contentType, width, height, phash, error: `variant: ${err instanceof Error ? err.message : String(err)}`.slice(0, 200) };
    }
  }
  return { status: 'ok', bytes: info.bytes, contentType: info.contentType, width, height, phash, error: null };
}

/** Pick the best surviving image of an asset and make it the hero; returns false when none exists. */
async function promoteHero(assetId: string): Promise<boolean> {
  const [asset] = await db().select({ hero: assets.heroImageUrl }).from(assets).where(eq(assets.id, assetId)).limit(1);
  if (!asset) return false;
  const candidates = await db()
    .select({ id: images.id, url: images.url, status: images.status })
    .from(images)
    .where(and(eq(images.assetId, assetId), inArray(images.status, ['ok', 'unchecked']), sql`${images.url} <> ${asset.hero ?? ''}`))
    .orderBy(sql`case ${images.status} when 'ok' then 0 else 1 end`, sql`case ${images.role} when 'hero' then 0 when 'gallery' then 1 else 2 end`, images.createdAt)
    .limit(6);
  for (const c of candidates) {
    let status = c.status;
    if (status === 'unchecked') {
      const outcome = await checkOne(c.url, true);
      status = outcome.status;
      await db()
        .update(images)
        .set({ status: outcome.status, checkedAt: new Date(), cacheKey: imageKey(c.url), bytes: outcome.bytes, contentType: outcome.contentType, width: outcome.width, height: outcome.height, phash: outcome.phash ?? undefined, error: outcome.error })
        .where(eq(images.id, c.id));
    }
    if (status === 'ok') {
      await db().update(assets).set({ heroImageUrl: c.url, updatedAt: new Date() }).where(eq(assets.id, assetId));
      await db().update(images).set({ role: 'hero' }).where(eq(images.id, c.id));
      return true;
    }
  }
  return false;
}

async function pool<T>(items: T[], concurrency: number, fn: (item: T) => Promise<void>): Promise<void> {
  let i = 0;
  const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => {
    while (i < items.length) {
      const item = items[i++]!;
      try {
        await fn(item);
      } catch (err) {
        logger.warn({ err }, 'image task failed');
      }
    }
  });
  await Promise.all(workers);
}

/** Summary for the admin/health views. */
export async function imageStats(): Promise<Record<string, number>> {
  const rows = (await db().execute(sql`select status, count(*)::int as n from images group by status`)) as unknown as Array<{ status: string; n: number }>;
  const out: Record<string, number> = {};
  for (const r of rows) out[r.status] = r.n;
  return out;
}