TypeScript 61.9%
HTML 37.2%
SQL 0.7%
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 // drizzle serialises a JS array parameter as a row (record), so pass the ids as JSON and expand them server-side.43 const assetFilter = opts.assetIds?.length ? sql` and a.id in (select jsonb_array_elements_text(${JSON.stringify(opts.assetIds)}::jsonb))` : sql``;44 const reg = (await db().execute(sql`45 insert into images (id, asset_id, source_id, url, role, status)46 select 'img_' || substr(md5(a.id || a.hero_image_url), 1, 20), a.id, null, a.hero_image_url, 'hero', 'unchecked'47 from assets a48 where a.hero_image_url is not null and a.hero_image_url ~ '^https?://'49 and not exists (select 1 from images i where i.url = a.hero_image_url)${assetFilter}50 limit ${limit}51 on conflict (url) do nothing52 returning id53 `)) as unknown as unknown[];54 res.registered = reg.length;5556 // 2. Pick rows to check.57 const staleCut = new Date(Date.now() - RECHECK_AFTER_DAYS * 86_400_000);58 const errorCut = new Date(Date.now() - RETRY_ERROR_AFTER_HOURS * 3600_000);59 const conds = [60 eq(images.status, 'unchecked'),61 and(eq(images.status, 'error'), lt(images.checkedAt, errorCut)),62 ...(opts.recheck ? [and(eq(images.status, 'ok'), or(isNull(images.checkedAt), lt(images.checkedAt, staleCut)))] : []),63 ];64 const rows = await db()65 .select({ id: images.id, url: images.url, assetId: images.assetId, role: images.role, width: images.width })66 .from(images)67 .where(and(or(...conds), opts.assetIds?.length ? inArray(images.assetId, opts.assetIds) : undefined))68 .orderBy(sql`case ${images.role} when 'hero' then 0 else 1 end`, images.createdAt)69 .limit(limit);7071 const deadHeroAssets = new Set<string>();72 await pool(rows, opts.concurrency ?? 6, async (row) => {73 res.checked++;74 const outcome = await checkOne(row.url, opts.warm ?? true);75 if (outcome.status === 'ok') res.ok++;76 else if (outcome.status === 'dead') res.dead++;77 else if (outcome.status === 'blocked') res.blocked++;78 else res.error++;79 await db()80 .update(images)81 .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 })82 .where(eq(images.id, row.id));83 if (outcome.status !== 'ok' && outcome.status !== 'error' && row.assetId && row.role === 'hero') deadHeroAssets.add(row.assetId);84 });8586 // 3. Promote a replacement hero for assets whose hero image is gone.87 for (const assetId of deadHeroAssets) {88 const promoted = await promoteHero(assetId);89 if (promoted) res.promoted++;90 else res.placeholders++;91 }92 log.info(res, 'images processed');93 return res;94}9596interface CheckOutcome {97 status: 'ok' | 'dead' | 'blocked' | 'error';98 bytes: number | null;99 contentType: string | null;100 width: number | null;101 height: number | null;102 phash: string | null;103 error: string | null;104}105106async function checkOne(url: string, warm: boolean): Promise<CheckOutcome> {107 const out = await ensureOriginal(url);108 if (!out.ok) return { status: out.status, bytes: null, contentType: null, width: null, height: null, phash: null, error: out.reason.slice(0, 200) };109 const info = out.info;110 let width = info.width;111 let height = info.height;112 if (width === null || height === null) {113 try {114 const meta = await sharp(info.path).metadata();115 width = meta.width ?? null;116 height = meta.height ?? null;117 } catch {118 /* keep nulls */119 }120 }121 const phash = await perceptualHash(info.path);122 if (warm) {123 try {124 await ensureVariant(info, 192);125 await ensureVariant(info, 384);126 } catch (err) {127 return { status: 'error', bytes: info.bytes, contentType: info.contentType, width, height, phash, error: `variant: ${err instanceof Error ? err.message : String(err)}`.slice(0, 200) };128 }129 }130 return { status: 'ok', bytes: info.bytes, contentType: info.contentType, width, height, phash, error: null };131}132133/** Pick the best surviving image of an asset and make it the hero; returns false when none exists. */134async function promoteHero(assetId: string): Promise<boolean> {135 const [asset] = await db().select({ hero: assets.heroImageUrl }).from(assets).where(eq(assets.id, assetId)).limit(1);136 if (!asset) return false;137 const candidates = await db()138 .select({ id: images.id, url: images.url, status: images.status })139 .from(images)140 .where(and(eq(images.assetId, assetId), inArray(images.status, ['ok', 'unchecked']), sql`${images.url} <> ${asset.hero ?? ''}`))141 .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)142 .limit(6);143 for (const c of candidates) {144 let status = c.status;145 if (status === 'unchecked') {146 const outcome = await checkOne(c.url, true);147 status = outcome.status;148 await db()149 .update(images)150 .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 })151 .where(eq(images.id, c.id));152 }153 if (status === 'ok') {154 await db().update(assets).set({ heroImageUrl: c.url, updatedAt: new Date() }).where(eq(assets.id, assetId));155 await db().update(images).set({ role: 'hero' }).where(eq(images.id, c.id));156 return true;157 }158 }159 return false;160}161162async function pool<T>(items: T[], concurrency: number, fn: (item: T) => Promise<void>): Promise<void> {163 let i = 0;164 const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => {165 while (i < items.length) {166 const item = items[i++]!;167 try {168 await fn(item);169 } catch (err) {170 logger.warn({ err }, 'image task failed');171 }172 }173 });174 await Promise.all(workers);175}176177/** Summary for the admin/health views. */178export async function imageStats(): Promise<Record<string, number>> {179 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 }>;180 const out: Record<string, number> = {};181 for (const r of rows) out[r.status] = r.n;182 return out;183}184