feat(translations): async LLM translation pipeline for EMEX/PCAT terms

Add BullMQ-backed cold-path translation so first-time category and part
names from EMEX/PCAT scrapers no longer stay English. translateMany now
returns the original on DB miss but enqueues a translation job; the
worker calls DeepSeek V3 via OpenRouter, persists to
emex_category_translations, and updates already-stored rows where
name = name_original. Redis NX flag dedupes concurrent enqueues so the
same term is not translated 100x when many users hit it at once.

Removes the dictionary fallback from TranslationsService — its
word-by-word replacement produced half-translated strings (e.g.
"Body frame" → "Kaporta frame") that polluted the DB.

Bootstrap and backfill scripts cover EMEX and PCAT sources together.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

apps/api/src/jobs/bull.config.ts

@@ -25,4 +25,5 @@ export const QUEUE_NAMES = {
   SUBSCRIPTION_EXPIRY: "subscription-expiry",
   QUERY_CLEANUP: "query-cleanup",
   CATALOG_PREFETCH: "catalog-prefetch",
+  TRANSLATION: "translation",
 } as const;

apps/api/src/jobs/processors/translation.processor.ts

@@ -0,0 +1,142 @@
+import { Job } from "bullmq";
+import { sql as drizzleSql, inArray } from "drizzle-orm";
+import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
+import OpenAI from "openai";
+import Redis from "ioredis";
+import { emexCategoryTranslations } from "../../database/schema/core";
+
+// Schema-loose local alias (matches the shape used by worker.ts which builds
+// drizzle without schema generics). Other processors do the same.
+type Database = PostgresJsDatabase<Record<string, unknown>>;
+
+export interface TranslationJobData {
+  /** Raw original names (from EMEX/PCAT scraper output) to translate. */
+  terms: string[];
+}
+
+const MODEL = "deepseek/deepseek-chat";
+const SYSTEM_PROMPT = `Sen bir Türk otomotiv çevirmenisin. Görevin: araç yedek parça kataloğundan gelen kategori ve parça isimlerini İngilizce'den (zaman zaman Rusça'dan) Türkçe'ye çevirmek.
+
+Kurallar:
+1. Türkiye yedek parça sektöründe kullanılan terminolojiyi kullan ("Brake Pad" → "Fren Balatası", "Spark Plug" → "Buji").
+2. Türkçe karakterleri (ş, ı, ğ, ü, ö, ç) doğru kullan.
+3. Marka isimleri (BMW, VW, Toyota), model kodları, OEM parça kodları ve teknik kısaltmalar (ABS, ESP, OBD, ECU, R.H., L.H.) çevirmeden olduğu gibi kalır.
+4. Belirsiz terim için en yakın TR karşılığını yaz; çok belirsizse orijinali koru.
+5. Kısa, UI'da gösterilebilir (1-5 kelime ideal).
+6. "Boot" otomotiv bağlamında "Bagaj" demektir.
+7. Çıktı: girdi listesinin AYNI sırasında, eşit uzunlukta JSON dizisi.
+
+Yanıt KESİN olarak şu JSON formatında olmalı:
+{"translations": ["çeviri1", "çeviri2", ...]}`;
+
+async function callLLM(ai: OpenAI, terms: string[]): Promise<string[]> {
+  let lastErr: unknown = null;
+  for (let attempt = 0; attempt < 3; attempt++) {
+    try {
+      const response = await ai.chat.completions.create({
+        model: MODEL,
+        max_tokens: 4096,
+        response_format: { type: "json_object" },
+        messages: [
+          { role: "system", content: SYSTEM_PROMPT },
+          { role: "user", content: JSON.stringify({ terms }) },
+        ],
+      });
+      const content = response.choices[0]?.message?.content;
+      if (!content) throw new Error("Empty response");
+      const match = content.match(/\{[\s\S]*?"translations"[\s\S]*?\}/);
+      if (!match) throw new Error("No JSON in response");
+      const parsed = JSON.parse(match[0]) as { translations: string[] };
+      if (!Array.isArray(parsed.translations) || parsed.translations.length !== terms.length) {
+        throw new Error(`Length mismatch: ${parsed.translations?.length} vs ${terms.length}`);
+      }
+      return parsed.translations;
+    } catch (err) {
+      lastErr = err;
+      const status =
+        (err as { status?: number })?.status ??
+        (err as { response?: { status?: number } })?.response?.status;
+      if (status === 429 || status === 529 || (status && status >= 500)) {
+        await new Promise((r) => setTimeout(r, 2 ** attempt * 1000));
+        continue;
+      }
+      throw err;
+    }
+  }
+  throw lastErr;
+}
+
+/**
+ * Process a translation job: translate `terms` via LLM, persist to
+ * emex_category_translations, then UPDATE categories/parts rows where
+ * `name = name_original` (still raw English) so already-fetched data
+ * shows up Turkish on the next refresh. Cache invalidation flushes
+ * Redis tree caches.
+ */
+export async function processTranslation(
+  job: Job<TranslationJobData>,
+  db: Database,
+  ai: OpenAI,
+  redis: Redis,
+): Promise<{ translated: number; skipped: number }> {
+  const { terms } = job.data;
+  if (!terms?.length) return { translated: 0, skipped: 0 };
+
+  // Skip terms already in DB (another worker may have raced ahead)
+  const existing = await db
+    .select({ originalName: emexCategoryTranslations.originalName })
+    .from(emexCategoryTranslations)
+    .where(inArray(emexCategoryTranslations.originalName, terms));
+  const existingSet = new Set(existing.map((r) => r.originalName));
+  const todo = terms.filter((t) => !existingSet.has(t));
+  if (!todo.length) return { translated: 0, skipped: terms.length };
+
+  // LLM call (single batch)
+  const translations = await callLLM(ai, todo);
+
+  // Persist translations
+  await db
+    .insert(emexCategoryTranslations)
+    .values(
+      todo.map((orig, i) => ({
+        originalName: orig,
+        translatedName: translations[i] || orig,
+        isManual: false,
+      })),
+    )
+    .onConflictDoNothing();
+
+  // Update existing categories/parts rows where name was left as English.
+  // Run as a single statement per term — small N (≤50), and we keep it
+  // scoped to the two scraper sources.
+  for (let i = 0; i < todo.length; i++) {
+    const orig = todo[i];
+    const tr = translations[i];
+    if (!tr || tr === orig) continue;
+    await db.execute(drizzleSql`
+      UPDATE categories
+         SET name = ${tr}
+       WHERE source IN ('emex', 'parts-catalogs')
+         AND name_original = ${orig}
+         AND name = name_original
+    `);
+    await db.execute(drizzleSql`
+      UPDATE parts
+         SET name = ${tr}
+       WHERE source IN ('emex', 'parts-catalogs')
+         AND name_original = ${orig}
+         AND name = name_original
+    `);
+  }
+
+  // Invalidate translation lookup cache so next read sees DB value.
+  // Tree cache (cat:tree:*) is intentionally NOT flushed wholesale here
+  // because that would punish unrelated vehicles; a stale tree just
+  // expires within 1h and the underlying name is already updated in DB.
+  const trKeys = todo.map((t) => `tr:${t}`);
+  if (trKeys.length) {
+    await redis.del(...trKeys).catch(() => undefined);
+  }
+
+  return { translated: todo.length, skipped: terms.length - todo.length };
+}

apps/api/src/jobs/queues/translation.queue.ts

@@ -0,0 +1,22 @@
+import { Provider } from "@nestjs/common";
+import { Queue } from "bullmq";
+import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config";
+
+export const TRANSLATION_QUEUE = "TRANSLATION_QUEUE";
+
+export const TranslationQueueProvider: Provider = {
+  provide: TRANSLATION_QUEUE,
+  useFactory: () => {
+    const telemetry = getBullTelemetry();
+    return new Queue(QUEUE_NAMES.TRANSLATION, {
+      connection: getBullConnection(),
+      ...(telemetry ? { telemetry } : {}),
+      defaultJobOptions: {
+        attempts: 3,
+        backoff: { type: "exponential", delay: 5000 },
+        removeOnComplete: { count: 500 },
+        removeOnFail: { count: 1000 },
+      },
+    });
+  },
+};

apps/api/src/translations/translations.module.ts

@@ -1,10 +1,11 @@
 import { Module } from "@nestjs/common";
+import { TranslationQueueProvider } from "../jobs/queues/translation.queue";
 import { TranslationsController } from "./translations.controller";
 import { TranslationsService } from "./translations.service";
 
 @Module({
   controllers: [TranslationsController],
-  providers: [TranslationsService],
+  providers: [TranslationsService, TranslationQueueProvider],
   exports: [TranslationsService],
 })
 export class TranslationsModule {}

apps/api/src/translations/translations.service.spec.ts

@@ -31,7 +31,8 @@ function createMockDeps() {
 }
 
 function createService(db: unknown, redis: unknown): TranslationsService {
-  return new TranslationsService(db as any, redis as any);
+  const queue = { add: vi.fn().mockResolvedValue(undefined) };
+  return new TranslationsService(db as any, redis as any, queue as any);
 }
 
 describe("TranslationsService", () => {

apps/api/src/translations/translations.service.ts

@@ -1,7 +1,9 @@
 import { Inject, Injectable, Logger } from "@nestjs/common";
+import { Queue } from "bullmq";
 import { eq, ilike, inArray, or } from "drizzle-orm";
 import { DATABASE, type Database } from "../database/database.provider";
 import { emexCategoryTranslations } from "../database/schema/core";
+import { TRANSLATION_QUEUE } from "../jobs/queues/translation.queue";
 import { RedisService } from "../redis/redis.service";
 
 /** 30 days in seconds */
@@ -259,8 +261,42 @@ export class TranslationsService {
   constructor(
     @Inject(DATABASE) private db: Database,
     private redis: RedisService,
+    @Inject(TRANSLATION_QUEUE) private translationQueue: Queue,
   ) {}
 
+  /**
+   * Enqueue terms for async LLM translation. Skips terms already known to
+   * be queued (Redis NX flag, 5min TTL) to avoid 100x duplicate jobs when
+   * many users hit the same fresh term concurrently.
+   */
+  private async enqueueTranslation(terms: string[]): Promise<void> {
+    if (!terms.length) return;
+    const redis = this.redis.getClient();
+    // Redis SET NX EX dedup: only queue terms whose flag was just set
+    const fresh: string[] = [];
+    const pipeline = redis.pipeline();
+    for (const t of terms) pipeline.set(`tr:queued:${t}`, "1", "EX", 300, "NX");
+    const results = await pipeline.exec();
+    if (results) {
+      results.forEach((r, i) => {
+        if (r && r[1] === "OK") fresh.push(terms[i]);
+      });
+    }
+    if (!fresh.length) return;
+
+    // Chunk into 50-term batches (matches bootstrap LLM throughput sweet spot)
+    for (let i = 0; i < fresh.length; i += 50) {
+      const chunk = fresh.slice(i, i + 50);
+      await this.translationQueue.add(
+        "translate",
+        { terms: chunk },
+        // jobId hashes the chunk so identical retries collapse
+        { jobId: `tr:${Buffer.from(chunk.sort().join("|")).toString("base64").slice(0, 32)}` },
+      );
+    }
+    this.logger.debug(`Enqueued ${fresh.length} term(s) for async translation`);
+  }
+
   /**
    * Translate a single key: Redis cache → DB → dictionary → original
    */
@@ -291,9 +327,9 @@ export class TranslationsService {
       return result;
     }
 
-    // 3. No DB hit — return original. Dictionary fallback removed: its
-    //    word-by-word replacement produces half-translated strings and
-    //    would poison the DB. Bootstrap script handles new terms via LLM.
+    // 3. No DB hit — return original now, async-translate in background.
+    //    Dictionary fallback was removed because its word-by-word replacement
+    //    produced half-translated strings and would poison the DB.
     const result: TranslationResult = {
       key,
       sourceText,
@@ -302,6 +338,9 @@ export class TranslationsService {
       isAutoTranslated: false,
     };
     await this.redis.setJson(cacheKey, result, CACHE_MISS_TTL);
+    this.enqueueTranslation([sourceText]).catch((err) =>
+      this.logger.warn(`Translation enqueue failed: ${(err as Error).message}`),
+    );
     return result;
   }
 
@@ -359,14 +398,12 @@ export class TranslationsService {
       dbMap.set(row.originalName, { translated: row.translatedName, isManual: row.isManual });
     }
 
-    // Phase 3: Build result map + populate Redis pipeline.
-    // We deliberately do NOT use translateWithDictionary's word-by-word
-    // replacement here — it produces half-translated strings (e.g.
-    // "Body frame" → "Kaporta frame") and would persist them to the DB,
-    // poisoning future lookups. Anything missing falls through to a
-    // short-TTL cache miss; the next emex-translate-bootstrap.ts run
-    // picks it up and writes the proper LLM translation.
+    // Phase 3: Build result map + populate Redis pipeline. DB miss falls
+    // through to a short-TTL cache placeholder + async BullMQ job — the
+    // worker translates via LLM and updates DB so the next refresh shows
+    // Turkish without blocking this request.
     const pipeline = redis.pipeline();
+    const toEnqueue: string[] = [];
 
     for (const text of missing) {
       const dbHit = dbMap.get(text);
@@ -390,10 +427,18 @@ export class TranslationsService {
           isAutoTranslated: false,
         };
         pipeline.set(`${CACHE_PREFIX}${text}`, JSON.stringify(payload), "EX", CACHE_MISS_TTL);
+        toEnqueue.push(text);
       }
     }
 
     await pipeline.exec();
+
+    // Fire-and-forget — failures here must not break the hot path
+    if (toEnqueue.length) {
+      this.enqueueTranslation(toEnqueue).catch((err) =>
+        this.logger.warn(`Translation enqueue failed: ${(err as Error).message}`),
+      );
+    }
     return result;
   }
 

apps/api/src/worker.ts

@@ -4,10 +4,13 @@ import "dotenv/config";
 import { Worker } from "bullmq";
 import { drizzle } from "drizzle-orm/postgres-js";
 import postgres from "postgres";
+import Redis from "ioredis";
+import OpenAI from "openai";
 import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "./jobs/bull.config";
 import { processEmexScrape } from "./jobs/processors/emex-scrape.processor";
 import { processQueryCleanup } from "./jobs/processors/query-cleanup.processor";
 import { processSubscriptionExpiry } from "./jobs/processors/subscription-expiry.processor";
+import { processTranslation } from "./jobs/processors/translation.processor";
 import { isOtelEnabled } from "./telemetry";
 import { sdk } from "./telemetry/worker-tracing";
 
@@ -119,6 +122,52 @@ queryCleanupWorker.on("failed", (job, err) => {
 
 workers.push(queryCleanupWorker);
 
+// Translation Worker (async LLM translation for new EMEX/PCAT terms)
+const openrouterApiKey = process.env.OPENROUTER_API_KEY;
+if (openrouterApiKey) {
+  const ai = new OpenAI({
+    apiKey: openrouterApiKey,
+    baseURL: "https://openrouter.ai/api/v1",
+    defaultHeaders: {
+      "HTTP-Referer": "https://sase.tr",
+      "X-Title": "Sase Translation Worker",
+    },
+  });
+  const translationRedis = new Redis({
+    host: process.env.REDIS_HOST || "localhost",
+    port: Number(process.env.REDIS_PORT) || 6379,
+    password: process.env.REDIS_PASSWORD || undefined,
+    maxRetriesPerRequest: null,
+  });
+
+  const translationWorker = new Worker(
+    QUEUE_NAMES.TRANSLATION,
+    async (job) => {
+      return processTranslation(job, db, ai, translationRedis);
+    },
+    {
+      connection,
+      concurrency: 3,
+      ...(telemetry ? { telemetry } : {}),
+      // Light rate limit — DeepSeek via OpenRouter handles ~10 req/s comfortably
+      limiter: { max: 30, duration: 60_000 },
+    },
+  );
+
+  translationWorker.on("completed", (job) => {
+    console.log(`[worker] translation job ${job.id} completed`);
+  });
+  translationWorker.on("failed", (job, err) => {
+    console.error(`[worker] translation job ${job?.id} failed: ${err.message}`);
+  });
+
+  workers.push(translationWorker);
+} else {
+  console.warn(
+    "[worker] OPENROUTER_API_KEY not set — translation worker disabled (terms will stay English until next manual bootstrap)",
+  );
+}
+
 // ─── Startup ──────────────────────────────────────────
 console.log("[worker] BullMQ workers started");
 console.log(`[worker] Listening on queues: ${Object.values(QUEUE_NAMES).join(", ")}`);