1/**2 * Image-processing worker (§113 basics): validates and caches product images, records3 * status/dimensions/perceptual hash in `images`, and promotes a replacement hero image when the4 * current one is dead. Shares the fetch/cache core with the web `/img` proxy.5 */6import { and, eq, inArray, isNull, lt, or, sql } from 'drizzle-orm';7import { assets, images } from '@rareindex/database';8import { logger } from '@rareindex/shared';9import sharp from 'sharp';10import { db } from '../lib/db.ts';11import { ensureOriginal, ensureVariant, imageKey, perceptualHash } from './core.ts';1213export interface ImagesResult {14 registered: number;15 checked: number;16 ok: number;17 dead: number;18 blocked: number;19 error: number;20 promoted: number;21 placeholders: number;22}2324export interface ImagesOptions {25 limit?: number;26 recheck?: boolean;27 assetIds?: string[];28 concurrency?: number;29 /** also warm the 192/384 variants so first paint is instant */30 warm?: boolean;31}3233const RECHECK_AFTER_DAYS = 30;34const RETRY_ERROR_AFTER_HOURS = 6;3536export async function processImages(opts: ImagesOptions = {}): Promise<ImagesResult> {37 const log = logger.child({ component: 'images' });38 const limit = opts.limit ?? 2000;39 const res: ImagesResult = { registered: 0, checked: 0, ok: 0, dead: 0, blocked: 0, error: 0, promoted: 0, placeholders: 0 };4041 // 1. Register hero URLs that have no images row yet (catalog sources may set hero_image_url directly).42 const assetFilter = opts.assetIds?.length ? sql` and a.id = any(${opts.assetIds}::text[])` : sql``;43 const reg = (await db().execute(sql`44 insert into images (id, asset_id, source_id, url, role, status)45 select 'img_' || substr(md5(a.id || a.hero_image_url), 1, 20), a.id, null, a.hero_image_url, 'hero', 'unchecked'46 from assets a47 where a.hero_image_url is not null and a.hero_image_url ~ '^https?://'48 and not exists (select 1 from images i where i.url = a.hero_image_url)${assetFilter}49 limit ${limit}50 on conflict (url) do nothing51 returning id52 `)) as unknown as unknown[];53 res.registered = reg.length;5455 // 2. Pick rows to check.56 const staleCut = new Date(Date.now() - RECHECK_AFTER_DAYS * 86_400_000);57 const errorCut = new Date(Date.now() - RETRY_ERROR_AFTER_HOURS * 3600_000);58 const conds = [59 eq(images.status, 'unchecked'),60 and(eq(images.status, 'error'), lt(images.checkedAt, errorCut)),61 ...(opts.recheck ? [and(eq(images.status, 'ok'), or(isNull(images.checkedAt), lt(images.checkedAt, staleCut)))] : []),62 ];63 const rows = await db()64 .select({ id: images.id, url: images.url, assetId: images.assetId, role: images.role, width: images.width })65 .from(images)66 .where(and(or(...conds), opts.assetIds?.length ? inArray(images.assetId, opts.assetIds) : undefined))67 .orderBy(sql`case ${images.role} when 'hero' then 0 else 1 end`, images.createdAt)68 .limit(limit);6970 const deadHeroAssets = new Set<string>();71 await pool(rows, opts.concurrency ?? 6, async (row) => {72 res.checked++;73 const outcome = await checkOne(row.url, opts.warm ?? true);74 if (outcome.status === 'ok') res.ok++;75 else if (outcome.status === 'dead') res.dead++;76 else if (outcome.status === 'blocked') res.blocked++;77 else res.error++;78 await db()79 .update(images)80 .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 })81 .where(eq(images.id, row.id));82 if (outcome.status !== 'ok' && outcome.status !== 'error' && row.assetId && row.role === 'hero') deadHeroAssets.add(row.assetId);83 });8485 // 3. Promote a replacement hero for assets whose hero image is gone.86 for (const assetId of deadHeroAssets) {87 const promoted = await promoteHero(assetId);88 if (promoted) res.promoted++;89 else res.placeholders++;90 }91 log.info(res, 'images processed');92 return res;93}9495interface CheckOutcome {96 status: 'ok' | 'dead' | 'blocked' | 'error';97 bytes: number | null;98 contentType: string | null;99 width: number | null;100 height: number | null;101 phash: string | null;102 error: string | null;103}104105async function checkOne(url: string, warm: boolean): Promise<CheckOutcome> {106 const out = await ensureOriginal(url);107 if (!out.ok) return { status: out.status, bytes: null, contentType: null, width: null, height: null, phash: null, error: out.reason.slice(0, 200) };108 const info = out.info;109 let width = info.width;110 let height = info.height;111 if (width === null || height === null) {112 try {113 const meta = await sharp(info.path).metadata();114 width = meta.width ?? null;115 height = meta.height ?? null;116 } catch {117 /* keep nulls */118 }119 }120 const phash = await perceptualHash(info.path);121 if (warm) {122 try {123 await ensureVariant(info, 192);124 await ensureVariant(info, 384);125 } catch (err) {126 return { status: 'error', bytes: info.bytes, contentType: info.contentType, width, height, phash, error: `variant: ${err instanceof Error ? err.message : String(err)}`.slice(0, 200) };127 }128 }129 return { status: 'ok', bytes: info.bytes, contentType: info.contentType, width, height, phash, error: null };130}131132/** Pick the best surviving image of an asset and make it the hero; returns false when none exists. */133async function promoteHero(assetId: string): Promise<boolean> {134 const [asset] = await db().select({ hero: assets.heroImageUrl }).from(assets).where(eq(assets.id, assetId)).limit(1);135 if (!asset) return false;136 const candidates = await db()137 .select({ id: images.id, url: images.url, status: images.status })138 .from(images)139 .where(and(eq(images.assetId, assetId), inArray(images.status, ['ok', 'unchecked']), sql`${images.url} <> ${asset.hero ?? ''}`))140 .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)141 .limit(6);142 for (const c of candidates) {143 let status = c.status;144 if (status === 'unchecked') {145 const outcome = await checkOne(c.url, true);146 status = outcome.status;147 await db()148 .update(images)149 .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 })150 .where(eq(images.id, c.id));151 }152 if (status === 'ok') {153 await db().update(assets).set({ heroImageUrl: c.url, updatedAt: new Date() }).where(eq(assets.id, assetId));154 await db().update(images).set({ role: 'hero' }).where(eq(images.id, c.id));155 return true;156 }157 }158 return false;159}160161async function pool<T>(items: T[], concurrency: number, fn: (item: T) => Promise<void>): Promise<void> {162 let i = 0;163 const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => {164 while (i < items.length) {165 const item = items[i++]!;166 try {167 await fn(item);168 } catch (err) {169 logger.warn({ err }, 'image task failed');170 }171 }172 });173 await Promise.all(workers);174}175176/** Summary for the admin/health views. */177export async function imageStats(): Promise<Record<string, number>> {178 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 }>;179 const out: Record<string, number> = {};180 for (const r of rows) out[r.status] = r.n;181 return out;182}183