import { eq } from "drizzle-orm"; import { schema, trackProductAnalytics } from "@app/database"; import { scanCompleted, scanFailed } from "@app/analytics"; import type { AamosResult, AamosTaskType, DetectedItem } from "@app/ai-contracts"; import type { WorkerContext } from "../context.js"; import { getLocaleContext } from "../locale.js"; import type { LocaleContext } from "@app/shared-types"; import { recordAiUsage } from "../lib/ai-usage.js"; import { captureTrainingSample, type CaptureConsentFlags } from "../lib/shadow-capture.js"; /** * Bild-/OCR-jobb (spec §54): hämtar scan_job, anropar AAMOS med kontraktvaliderad * input/output, sparar resultatet och sätter awaiting_confirmation. * Användaren bekräftar ALLTID innan lagret röres (spec §61.5). */ export async function processScanJob(ctx: WorkerContext, scanJobId: string): Promise { const [job] = await ctx.db .select() .from(schema.scanJobs) .where(eq(schema.scanJobs.id, scanJobId)) .limit(1); if (!job) throw new Error(`scan_job ${scanJobId} finns inte`); if (job.status === "completed" || job.status === "awaiting_confirmation") return; // idempotent await ctx.db .update(schema.scanJobs) .set({ status: "running", attempts: job.attempts + 1, updatedAt: new Date() }) .where(eq(schema.scanJobs.id, scanJobId)); const imageUrls = await Promise.all(job.s3Keys.map((k) => ctx.readUrl(k))); const localeContext = await getLocaleContext(ctx, job.userId); const consentFlags = await loadConsentFlags(ctx, job.userId); const started = Date.now(); const result = await runAamosForJob( ctx, job.jobType as AamosTaskType, job.scanType, imageUrls, job.context, localeContext, consentFlags, ); if (result.status === "failed" || result.output == null) { const latencyMs = Date.now() - started; await ctx.db .update(schema.scanJobs) .set({ status: "failed", error: result.error ?? "AI-analysen misslyckades. Försök igen eller registrera manuellt.", latencyMs, updatedAt: new Date(), }) .where(eq(schema.scanJobs.id, scanJobId)); await recordScanFailed(ctx, job, { ...result, latencyMs }); return; } let output = result.output as Record; // SKIVA 1: mappa lagrade bilders detekterade namn mot kanoniska ingredienser. if ( (job.jobType === "ANALYZE_FRIDGE_IMAGE" || job.jobType === "ANALYZE_PANTRY_IMAGE") && Array.isArray((output as { items?: unknown }).items) ) { output = await mapDetectedItemsToCanonical(ctx, output); } // Kvitto (READ_RECEIPT) ger 'lines', men hela bekräftelseflödet (app-vyn // scan-review + API:ts extractProposals/annotateScanDuplicates) förväntar sig // 'items' i samma format som kylskåps-/skafferiscanning. Utan mappning blev // varje kvitto TOMT trots att Gemini läste alla rader. Mappa icke-rabattrader // till items och kör samma kanoniska namnmatchning som för foton. if (job.jobType === "READ_RECEIPT" && Array.isArray((output as { lines?: unknown }).lines)) { const lines = (output as { lines: Array> }).lines; (output as Record).items = lines .filter((l) => l.isDiscount !== true && (l.normalizedName ?? l.rawText)) .map((l, idx) => ({ tempId: String(idx), detectedName: String(l.normalizedName ?? l.rawText ?? ""), canonicalIngredientId: (l.canonicalIngredientId as string | null) ?? null, brand: null, estimatedQuantity: typeof l.quantity === "number" ? l.quantity : null, unit: (l.unit as string | null) ?? null, bestBeforeDate: null, confidence: typeof l.confidence === "number" ? l.confidence : 0.5, requiresConfirmation: true, })); output = await mapDetectedItemsToCanonical(ctx, output); } await ctx.db .update(schema.scanJobs) .set({ status: "awaiting_confirmation", result: output, modelVersion: result.modelVersion ?? null, promptVersion: result.promptVersion ?? null, latencyMs: result.latencyMs ?? Date.now() - started, costUsd: result.costUsd ?? null, updatedAt: new Date(), }) .where(eq(schema.scanJobs.id, scanJobId)); await recordScanCompleted(ctx, job, result); // Logga token-telemetri för Gemini (best-effort, får aldrig faila scan-vägen). try { await ctx.db.insert(schema.geminiUsage).values({ taskType: job.jobType, promptTokens: result.inputTokens ?? null, outputTokens: result.outputTokens ?? null, totalTokens: result.totalTokens ?? null, costMicrocents: result.costUsd != null ? Math.round(result.costUsd * 100_000_000) : null, }); } catch (err) { console.error("[scan processor] gemini_usage logging misslyckades:", err); } // Shadow-capture: spara träningspar för framtida AAMOS-distillation (FAS 1). // Kör aldrig synkront på användarens kritiska väg; fel swallås. if (result.status === "ok" && result.output != null) { const capture = await captureTrainingSample({ taskType: job.jobType as AamosTaskType, inputS3Keys: job.s3Keys, output: result.output as Record, modelVersion: result.modelVersion, promptVersion: result.promptVersion, latencyMs: result.latencyMs, costUsd: result.costUsd, inputTokens: result.inputTokens, outputTokens: result.outputTokens, consentFlags, readUrl: ctx.readUrl, }); if (!capture.ok) { console.error("[scan processor] shadow-capture misslyckades:", capture.error); } } // Bokför verklig AI-kostnad/tokens utan PII (spec §45). Kvoten (aiScans) drogs // redan vid skapandet (consumeAiScan, per bild) – räkna INTE upp den igen här, // annars dubbeldebiteras varje lyckad skanning. if (result.costUsd != null || result.inputTokens || result.outputTokens) { await recordAiUsage(ctx.db, job.userId, { costUsd: result.costUsd ?? 0, inputTokens: result.inputTokens ?? 0, outputTokens: result.outputTokens ?? 0, aiScans: 0, }); } // MEAL_PHOTO_ANALYZED-event för tallriksfoton (spec §55) if (job.jobType === "ANALYZE_MEAL_IMAGE") { const mealOutput = output as { kcalRange?: { mostLikely: number } | null; matchesRecipeContext?: boolean | null; }; await ctx.db.insert(schema.domainEvents).values({ type: "MEAL_PHOTO_ANALYZED", userId: job.userId, householdId: job.householdId, payload: { scanJobId, matched: mealOutput.matchesRecipeContext ?? false, kcalMostLikely: mealOutput.kcalRange?.mostLikely ?? null, }, }); } } async function runAamosForJob( ctx: WorkerContext, jobType: AamosTaskType, scanType: string, imageUrls: string[], context: unknown, localeContext: LocaleContext, consentFlags: CaptureConsentFlags, ) { switch (jobType) { case "ANALYZE_FRIDGE_IMAGE": case "ANALYZE_PANTRY_IMAGE": return ctx.aamos.runTask( jobType, { imageUrls, locationType: scanType, marketLocale: localeContext.languageTag, knownItems: [], }, { localeContext, consentFlags }, ); case "ANALYZE_MEAL_IMAGE": { const recipeContext = await buildRecipeContext(ctx, context); return ctx.aamos.runTask( "ANALYZE_MEAL_IMAGE", { imageUrls, recipeContext, marketLocale: localeContext.languageTag }, { localeContext, consentFlags }, ); } case "READ_RECEIPT": return ctx.aamos.runTask( "READ_RECEIPT", { imageUrls, marketLocale: localeContext.languageTag }, { localeContext, consentFlags }, ); case "READ_NUTRITION_LABEL": return ctx.aamos.runTask( "READ_NUTRITION_LABEL", { imageUrls, marketLocale: localeContext.languageTag }, { localeContext, consentFlags }, ); case "READ_EXPIRY_DATE": return ctx.aamos.runTask( "READ_EXPIRY_DATE", { imageUrls: imageUrls.slice(0, 2) }, { localeContext, consentFlags }, ); default: throw new Error(`Jobbtypen ${jobType} hanteras inte av scan-processorn`); } } async function buildRecipeContext(ctx: WorkerContext, context: unknown) { if ( typeof context !== "object" || context === null || typeof (context as { recipeId?: unknown }).recipeId !== "string" ) { return null; } const recipeId = (context as { recipeId: string }).recipeId; const [recipe] = await ctx.db .select({ id: schema.recipes.id, titleSv: schema.recipes.titleSv, nutritionPerPortion: schema.recipes.nutritionPerPortion, portions: schema.recipes.portions, }) .from(schema.recipes) .where(eq(schema.recipes.id, recipeId)) .limit(1); if (!recipe) return null; return { recipeId: recipe.id, titleSv: recipe.titleSv, nutritionPerPortion: recipe.nutritionPerPortion as unknown as Record, portions: recipe.portions, }; } async function loadConsentFlags(ctx: WorkerContext, userId: string) { const consents = await ctx.db .select() .from(schema.userConsents) .where(eq(schema.userConsents.userId, userId)); const get = (kind: string) => consents.find((c) => c.kind === kind)?.status === "granted"; return { personalization: get("personalization"), anonymizedImprovement: get("anonymized_improvement"), imageTraining: get("image_training"), }; } // --------------------------------------------------------------------------- // Kanonisk ingrediensmappning (SKIVA 1). AI föreslår, vi matchar mjukt, // användaren bekräftar alltid innan commit. // --------------------------------------------------------------------------- interface CanonicalIndex { id: string; nameSv: string; nameEn: string; aliases: string[]; } async function mapDetectedItemsToCanonical( ctx: WorkerContext, output: Record, ): Promise> { const items = (output as { items: DetectedItem[] }).items; if (!items.length) return output; const index = await loadCanonicalIndex(ctx); const mapped = items.map((item) => { const match = findBestCanonicalMatch(item.detectedName, index); return { ...item, canonicalIngredientId: match?.id ?? null, requiresConfirmation: match == null || item.confidence < 0.92, }; }); return { ...output, items: mapped }; } async function loadCanonicalIndex(ctx: WorkerContext): Promise { return ctx.db .select({ id: schema.canonicalIngredients.id, nameSv: schema.canonicalIngredients.nameSv, nameEn: schema.canonicalIngredients.nameEn, aliases: schema.canonicalIngredients.aliases, }) .from(schema.canonicalIngredients); } function findBestCanonicalMatch( detectedName: string, index: CanonicalIndex[], ): CanonicalIndex | null { const query = detectedName.toLowerCase(); let best: { item: CanonicalIndex; score: number } | null = null; for (const item of index) { const score = scoreMatch(query, item); if (score > 0 && (!best || score > best.score)) { best = { item, score }; } } // Threshold: require a strong token overlap or exact substring. if (!best || best.score < 0.35) return null; return best.item; } function fold(s: string): string { // Vik bort accenter (crème -> creme) OCH å/ä/ö -> a/a/o så att plural-vokalskifte // (morot <-> morötter) och lånord (crème fraîche) matchar konsekvent. return s.toLowerCase().normalize("NFD").replace(/[\u0300-\u036f]/g, ""); } function scoreMatch(query: string, item: CanonicalIndex): number { const q = fold(query); const candidates = [item.nameSv, item.nameEn, ...item.aliases].map(fold); let max = 0; const queryTokens = tokenize(q); for (const cand of candidates) { if (cand === q) return 1; if (cand.includes(q) || q.includes(cand)) { max = Math.max(max, 0.85); continue; } // Delad prefix (vindruva <-> vindruvor): stark signal utan token-exakthet. const shorter = q.length <= cand.length ? q : cand; const longer = q.length <= cand.length ? cand : q; let p = 0; while (p < shorter.length && shorter[p] === longer[p]) p++; if (p >= 5 && p / shorter.length >= 0.7) max = Math.max(max, 0.7); const candTokens = tokenize(cand); const intersection = queryTokens.filter((t) => candTokens.includes(t)); if (intersection.length > 0) { const overlap = intersection.length / Math.max(queryTokens.length, candTokens.length); max = Math.max(max, overlap); } } return max; } function tokenize(text: string): string[] { return text .toLowerCase() .replace(/[^a-zåäö0-9\s]/g, " ") .split(/\s+/) .filter((t) => t.length > 1); } function classifyScanError(error?: string | null): string { if (!error) return "unknown"; const lower = error.toLowerCase(); if (lower.includes("budget")) return "budget_exhausted"; if (lower.includes("timeout")) return "timeout"; if (lower.includes("kunde inte hämta bild")) return "image_fetch_failed"; if (lower.includes("matchar inte schema") || lower.includes("inte giltig json")) return "parse_error"; if (lower.includes("inga items") || lower.includes("no items")) return "no_items_detected"; return "ai_provider_error"; } async function recordScanCompleted( ctx: WorkerContext, job: { userId: string; householdId: string | null; scanType: string; jobType: string }, result: AamosResult, ): Promise { await trackProductAnalytics(ctx.db, job.userId, { ...scanCompleted(), householdId: job.householdId ?? undefined, properties: { scanType: job.scanType, jobType: job.jobType, latencyMs: result.latencyMs ?? null, costUsd: result.costUsd ?? null, modelVersion: result.modelVersion ?? null, promptVersion: result.promptVersion ?? null, }, }); } async function recordScanFailed( ctx: WorkerContext, job: { userId: string; householdId: string | null; scanType: string; jobType: string }, result: AamosResult, ): Promise { await trackProductAnalytics(ctx.db, job.userId, { ...scanFailed(), householdId: job.householdId ?? undefined, properties: { scanType: job.scanType, jobType: job.jobType, errorCode: classifyScanError(result.error), latencyMs: result.latencyMs ?? null, }, }); }