sequenceDiagram
    participant PIPELINE as Analysis Pipeline<br/>(ARQ Worker)
    participant CACHE as Redis Cache
    participant USER as User
    participant FE as Frontend
    participant API as FastAPI

    Note over PIPELINE: Analysis chạy 2-5 phút

    PIPELINE->>PIPELINE: Analysis result ready<br/>(English, canonical)

    PAR Pre-translation
        PIPELINE->>PIPELINE: Collect candidates (step 1-3)
        PIPELINE->>PIPELINE: LLM translate gpt-4o-mini<br/>(~2-5s)
        PIPELINE->>PIPELINE: Apply + validate
        PIPELINE->>CACHE: SET cache key<br/>localization:{id}:{locale}<br/>TTL=3600s
    end

    Note over USER: Vài phút sau, user mở analysis

    USER->>FE: Click "View Analysis"<br/>locale=ja
    FE->>API: GET /api/v1/analysis/{id}<br/>X-Scopelytics-Locale: ja
    API->>CACHE: Cache lookup
    CACHE-->>API: HIT 🚀
    API-->>FE: HTTP 200<br/>(translated, 10ms)

sequenceDiagram
    participant FE as Frontend
    participant API as FastAPI
    participant GUARD as Concurrency Guard
    participant LLM as OpenAI<br/>gpt-4o-mini
    participant CACHE as Redis Cache
    participant ARQ as ARQ Worker (retry)

    FE->>API: GET /analysis/{id}<br/>locale=ja
    API->>CACHE: Cache lookup
    CACHE-->>API: MISS

    API->>GUARD: Acquire semaphore<br/>(wait max 5s)

    alt Semaphore timeout
        API-->>FE: HTTP 200 English<br/>+ X-Localization-Status: skip_busy
        API->>ARQ: enqueue_job("localize_retry", id, ja)
        Note over ARQ: Translate background,<br/>lần sau cache hit

    else Guard acquired
        API->>LLM: chat_completions_parse()<br/>asyncio.wait_for(30s)

        alt LLM success
            LLM-->>API: decisions[]
            API->>API: Apply + validate
            API->>CACHE: SET cache key
            API-->>FE: HTTP 200<br/>(translated, ~3-6s)
        else LLM timeout / error
            LLM-->>API: ❌
            API-->>FE: HTTP 200 English<br/>+ X-Localization-Status: timeout
            API->>ARQ: enqueue_job("localize_retry", id, ja)
            Note over ARQ: Translate background,<br/>lần sau cache hit
        end
    end

flowchart TD
    REQ["Request vào FastAPI endpoint"]

    REQ --> ENGLISH["Build English response<br/>(Pydantic model)"]

    ENGLISH --> S0{"Step 0 - Early exit<br/>locale=en hoặc feature off?"}

    S0 -->|Có| EN_OUT["Return English<br/>0 token, 0 latency"]

    S0 -->|Không| CACHE{"Cache lookup<br/>localization:{surface}:{locale}:<br/>{sha256(payload)}"}

    CACHE -->|Hit| VALID{"Pydantic validate<br/>cached payload"}
    VALID -->|Pass| HIT["Return localized<br/>🚀 10-50ms"]
    VALID -->|Fail| STEP1

    CACHE -->|Miss| STEP1["Step 1 - Serialize<br/>model → JSON dict"]
    STEP1 --> STEP2["Step 2 - Collect candidates<br/>String leaf extraction"]
    STEP2 --> STEP3["Step 3 - Policy filter<br/>Denylist + surface allowlist"]

    STEP3 --> STEP4["Step 4 - Build LLM request"]

    STEP4 --> GUARD{"Acquire semaphore<br/>wait=5s"}

    GUARD -->|Timeout| GBUSY["Return English<br/>+ enqueue ARQ retry<br/>+ log guard_busy"]

    GUARD -->|OK| LLM_CALL["Step 5 - LLM translate<br/>asyncio.wait_for(30s)<br/>model=gpt-4o-mini"]

    LLM_CALL -->|Success| APPLY["Step 6 - Apply decisions<br/>Patch theo path"]
    LLM_CALL -->|Timeout| TOUT["Return English<br/>+ enqueue ARQ retry<br/>+ log llm_timeout"]
    LLM_CALL -->|Error| ERR["Return English<br/>+ enqueue ARQ retry<br/>+ log llm_error"]

    APPLY --> VALIDATE["Step 7 - Pydantic validate"]
    VALIDATE -->|Pass| WRITE["Write cache<br/>best-effort"]
    VALIDATE -->|Fail| VFAIL["Return English<br/>+ log validate_fail"]

    WRITE --> LOCAL_OUT["Return localized"]
    GBUSY --> EN_OUT
    TOUT --> EN_OUT
    ERR --> EN_OUT
    VFAIL --> EN_OUT

flowchart TD
    ANALYSIS["Analysis pipeline hoàn thành<br/>trong ARQ Worker"]

    ANALYSIS --> CHECK{"Locale support enabled?<br/>User có locale preference?"}

    CHECK -->|No supported locales| SKIP["Skip priming<br/>0 cost"]

    CHECK -->|Yes| LOAD["Load analysis result<br/>Pydantic model từ DB"]

    LOAD --> CACHE_CHECK{"Cache đã tồn tại?<br/>(analysis rerun)"}

    CACHE_CHECK -->|Đã có| SKIP
    CACHE_CHECK -->|Chưa| COLLECT["Collect candidates<br/>(steps 1-3)"]

    COLLECT --> EMPTY{"Candidates > 0?"}
    EMPTY -->|Không| SKIP

    EMPTY -->|Có| TRIGGER["Kiểm tra concurrency guard<br/>(nếu busy → skip, lần sau sync sẽ fill cache)"]

    TRIGGER -->|Busy| SKIP_PRIME["Skip priming<br/>+ log guard_busy"]

    TRIGGER -->|OK| PRIME["LLM translate<br/>gpt-4o-mini<br/>timeout=30s"]

    PRIME -->|Success| WRITE_CACHE["SET cache key<br/>TTL=3600s"]
    PRIME -->|Timeout/Lỗi| SKIP_PRIME

    WRITE_CACHE --> DONE["Priming done ✅"]

=== HAPPY PATH (Cache hit) ===
Cache lookup (Redis):            0.01-0.02s
Pydantic validate:               0.01-0.05s
Total:                           0.02-0.07s → ~10ms 🚀

=== EDGE CASE (Cache miss - sync fallback) ===
Build English response:          0.1-0.3s
Cache lookup (miss):             0.01s
Chờ semaphore (nếu busy):       0-5s (tối đa)
LLM translate (gpt-4o-mini):    2-5s (typical)
                                8-15s (worst, batch lớn 12k chars)
Apply + validate:               0.05-0.1s
Write cache:                    0.01s
Total:                           2-6s (typical)
                                8-20s (worst)

=== ERROR PATH (LLM timeout → English fallback) ===
Build English response:          0.1-0.3s
Cache lookup (miss):             0.01s
LLM timeout sau 30s:             30s (wait_for timeout)
→ return English:                ~30s (worst)
+ enqueue ARQ retry              0.01s

Copy

# Sau khi analysis result được persist vào DB
if settings.LOCALIZATION_ENABLED and settings.LOCALIZATION_PRIMING_ENABLED:
    await enqueue_job("localize_prime", analysis_id=analysis.id)

Copy

from ..services.localization import prime_localization_cache

# Trong worker, sau khi analysis xong:
await prime_localization_cache(analysis_result, supported_locales=["ja"])

Copy

async def prime_localization_cache(
    analysis: AnalysisDetailResponse,
    supported_locales: list[str],
) -> None:
    """Pre-translate analysis result cho tất cả supported locales (trừ canonical)."""
    for locale in supported_locales:
        if locale == LOCALIZATION_CANONICAL_LOCALE:
            continue
        try:
            # Giống hệt localize_response() nhưng không trả về, chỉ write cache
            decisions = await _call_translator(
                _prepare_candidates(analysis),
                locale=locale,
                context=LocalizationContext(surface="analysis_detail", ...),
            )
            merged = _apply_decisions(analysis.model_dump(mode="json"), decisions)
            validated = AnalysisDetailResponse.model_validate(merged)
            
            await _write_cache(
                key=_make_cache_key(analysis, locale),
                payload=validated.model_dump(mode="json"),
                ttl=LOCALIZATION_CACHE_TTL_SECONDS,
            )
        except Exception:
            logger.warning("localize_prime_failed", extra={"locale": locale})
            # Không fail pipeline — best-effort

Copy

# backend/src/app/workers/tasks/localization.py

async def localize_retry(ctx: dict, analysis_id: str, locale: str) -> None:
    """ARQ job: translate analysis trong background, write cache."""
    from ..services.localization import prime_localization_cache
    from ..crud.analysis import get_analysis
    
    analysis = await get_analysis(analysis_id)
    if not analysis:
        return
    
    await prime_localization_cache(analysis, supported_locales=[locale])

Copy

# backend/src/app/workers/settings.py — thêm vào WorkerSettings.functions
func(
    localize_retry,
    keep_result=0, 
    timeout=120,  # translation job timeout 120s
),

Copy

# backend/src/app/core/config.py — thêm vào AnalysisSettings

# ── Localization ──────────────────────────────────────────────
LOCALIZATION_ENABLED: bool = False
LOCALIZATION_PRIMING_ENABLED: bool = True       # Cache priming tại pipeline completion
LOCALIZATION_MODEL: str = "gpt-4o-mini"         # Model nhanh cho translation
LOCALIZATION_PROMPT_VERSION: str = "v1"
LOCALIZATION_POLICY_VERSION: str = "v1"
LOCALIZATION_SUPPORTED_LOCALES: str = "en,ja"
LOCALIZATION_CANONICAL_LOCALE: str = "en"
LOCALIZATION_TIMEOUT_SECONDS: float = 30.0      # Timeout cho 1 LLM call
LOCALIZATION_SEMAPHORE_WAIT_SECONDS: float = 5.0 # Timeout chờ semaphore
LOCALIZATION_MAX_CHARS_PER_BATCH: int = 12000
LOCALIZATION_MAX_CANDIDATES: int = 200
LOCALIZATION_MAX_OUTPUT_TOKENS: int = 4000
LOCALIZATION_CACHE_ENABLED: bool = True
LOCALIZATION_CACHE_TTL_SECONDS: int = 3600
LOCALIZATION_DEBUG_HEADERS: bool = False
LOCALIZATION_ENABLED_SURFACES: str = "analysis_detail"

Copy

Đầu ra: locale đã vào BE, localizer chưa wire → no-op.

Copy

Đầu ra: locale=ja → analysis detail trả Japanese (cache hit instant, miss ~3-6s).

Copy

Đầu ra: Analysis xong → tự động pre-translate → user mở page → cache hit 10ms 🚀

Copy

async def localize_response(
    model: BaseModel,
    locale: RequestLocale,
    *,
    context: LocalizationContext,
) -> BaseModel:
    """Localize response — guaranteed to return a valid model, never raises HTTP error."""

    # Step 0: Early exit
    if locale.code == LOCALIZATION_CANONICAL_LOCALE:
        return model
    if not settings.LOCALIZATION_ENABLED:
        return model

    try:
        payload = model.model_dump(mode="json")

        # Cache lookup
        if settings.LOCALIZATION_CACHE_ENABLED:
            cached = await _read_cache(_cache_key(payload, locale, context))
            if cached is not None:
                try:
                    return type(model).model_validate(cached)
                except Exception:
                    pass  # Cache corrupt → fall through to re-translate

        # Collect + filter candidates
        candidates = collect_strings(payload)
        translatable = policy_filter(candidates, context.surface)

        if not translatable:
            # Write empty cache to avoid re-checking
            await _write_cache(_cache_key(payload, locale, context), payload)
            return model

        # LLM translate with guard
        try:
            async with _localization_semaphore():
                decisions = await asyncio.wait_for(
                    _call_translator(translatable, locale, context),
                    timeout=settings.LOCALIZATION_TIMEOUT_SECONDS,
                )
        except (asyncio.TimeoutError, SemaphoreTimeout):
            # Fallback: English + enqueue ARQ retry
            await _enqueue_retry(context, locale)
            return model

        # Apply + validate
        merged = apply_decisions(payload, decisions)
        localized = type(model).model_validate(merged)

        # Write cache
        await _write_cache(_cache_key(payload, locale, context), merged)

        return localized

    except Exception:
        logger.exception("localization_fallback")
        return model  # English fallback — always HTTP 200

Copy

# Trong ARQ worker, sau khi analysis persist:
await prime_localization_cache(analysis, locales=["ja"])

Copy

backend/src/app/
├── core/
│   ├── config.py                          # + LocalizationSettings
│   └── localization.py                    # [MỚI] get_request_locale(), RequestLocale
│
├── services/
│   ├── analysis_pipeline.py               # [SỬA] + priming call sau persist
│   │
│   └── localization/                      # [MỚI] Package localization
│       ├── __init__.py                    # localize_response(), prime_localization_cache()
│       ├── types.py                       # Pydantic/dataclass types
│       ├── json_collector.py              # Duyệt JSON → StringCandidate[]
│       ├── denylist.py                    # Hard rules (code, không LLM)
│       ├── surface_policy.py             # Allowlist theo surface
│       ├── cache.py                       # Redis fingerprint cache
│       ├── prompts.py                     # System/user prompt templates
│       ├── ai_translator.py              # 1× OpenAI call via guard
│       └── response_localizer.py         # Orchestrator steps 0→7
│
└── workers/
    ├── settings.py                        # [SỬA] + localize_retry function
    └── tasks/
        └── localization.py               # [MỚI] ARQ task: localize_retry, localize_prime

Copy

-- Bảng shadow: lưu translation cho mọi string có thể dịch
CREATE TABLE localization_translations (
    id              BIGSERIAL PRIMARY KEY,
    
    -- Target
    surface         VARCHAR(64)     NOT NULL,   -- analysis_detail, bid_advisor_pack, ...
    row_id          INTEGER         NOT NULL,   -- PK của bảng gốc (vd: analysis_results.id)
    column_name     VARCHAR(128)    NOT NULL,   -- features, summary, task, ...
    json_path       VARCHAR(256)    NOT NULL,   -- JSON pointer: results[0].features[2].title
    locale          VARCHAR(8)      NOT NULL,   -- ja, vi, ko, ...
    
    -- Data
    source_text     TEXT            NOT NULL,   -- English gốc (để detect change)
    translated_text TEXT            NOT NULL,   -- Đã dịch
    prompt_hash     VARCHAR(64)     NOT NULL,   -- Phiên bản prompt đã dịch
    
    -- Metadata
    created_at      TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
    is_active       BOOLEAN         NOT NULL DEFAULT TRUE,
    
    -- Constraints
    UNIQUE (surface, row_id, json_path, locale, prompt_hash)
);

-- Index cho lookup nhanh theo locale
CREATE INDEX idx_localization_lookup 
    ON localization_translations (surface, row_id, locale, prompt_hash)
    WHERE is_active = TRUE;

-- Index cho backfill / migration
CREATE INDEX idx_localization_locale
    ON localization_translations (locale, surface)
    WHERE is_active = TRUE;

Copy

        ┌──────────────────┐
        │   User Request   │
        │  locale != en    │
        └────────┬─────────┘
                 │
        ┌────────▼─────────┐
        │   Redis Cache     │  ← Tier 1: Hot cache (10ms)
        │   TTL=3600s       │
        └────────┬─────────┘
                 │
    ┌────────────┴────────────┐
    │                         │
    ▼                         ▼
┌──────────┐           ┌──────────┐
│ Cache    │           │ Cache    │
│ HIT 🚀   │           │ MISS     │
└──────────┘           └────┬─────┘
                           │
                    ┌──────▼──────┐
                    │ PostgreSQL   │  ← Tier 2: Persistent (5ms)
                    │ Shadow Table │
                    └──────┬──────┘
                           │
               ┌───────────┴───────────┐
               │                       │
               ▼                       ▼
        ┌──────────┐           ┌──────────┐
        │ DB HIT   │           │ DB MISS  │
        │→ Redis   │           │→ LLM     │
        │  warm    │           │  translate│
        │→ return  │           │→ DB write │
        └──────────┘           │→ Redis   │
                               │  write   │
                               │→ return  │
                               └──────────┘

Copy

async def _get_translated_payload(
    analysis_id: int,
    locale: str,
    surface: str,
    prompt_hash: str,
) -> dict | None:
    
    # Tier 1: Redis
    cache_key = f"localization:{surface}:{analysis_id}:{locale}:{prompt_hash}"
    cached = await redis.get(cache_key)
    if cached is not None:
        return json.loads(cached)
    
    # Tier 2: PostgreSQL
    rows = await db.execute(
        select(LocalizationTranslation)
        .where(
            LocalizationTranslation.surface == surface,
            LocalizationTranslation.row_id == analysis_id,
            LocalizationTranslation.locale == locale,
            LocalizationTranslation.prompt_hash == prompt_hash,
            LocalizationTranslation.is_active == True,
        )
    )
    
    if rows:
        # Warm Redis
        payload = _rebuild_payload_from_rows(rows)
        await redis.setex(cache_key, 3600, json.dumps(payload))
        return payload
    
    return None  # Cache miss → LLM translate

Copy

async def _save_translations(
    analysis_id: int,
    locale: str,
    surface: str,
    prompt_hash: str,
    decisions: list[TranslationDecision],
):
    # PostgreSQL — batch insert
    rows = [
        LocalizationTranslation(
            surface=surface,
            row_id=analysis_id,
            column_name=_extract_column(d.path),
            json_path=d.path,
            locale=locale,
            source_text=d.source_text,
            translated_text=d.translated_text or d.source_text,
            prompt_hash=prompt_hash,
        )
        for d in decisions
        if d.action == "translate"
    ]
    await db.execute(
        insert(LocalizationTranslation)
        .on_conflict_do_update(
            constraint="...",
            set_=dict(translated_text=..., prompt_hash=..., updated_at=func.now()),
        )
    )
    
    # Redis — set cache
    cache_key = f"localization:{surface}:{analysis_id}:{locale}:{prompt_hash}"
    await redis.setex(cache_key, 3600, _build_payload(decisions))

Copy

-- 1. Tạo bảng
CREATE TABLE localization_translations (...);

-- 2. Index
CREATE INDEX ...;

-- 3. ARQ job: translate tất cả analysis cũ
-- Job chạy background, batch 50 analysis / lần

Copy

# backend/src/app/workers/tasks/localization_backfill.py

async def backfill_localization_cache(ctx: dict, locale: str):
    """Backfill: translate all existing analyses for a new locale."""
    analyses = await get_all_completed_analyses()
    for batch in chunked(analyses, 50):
        async with asyncio.TaskGroup() as tg:
            for analysis in batch:
                tg.create_task(
                    prime_single_analysis(analysis, locale)
                )

Copy

erDiagram
    analyses ||--o{ analysis_results : has
    analysis_results ||--o{ localization_translations : "translated as"
    analyses ||--o{ localization_translations : "translated as"
    
    analyses {
        int id PK
        text title "English canonical"
        text description "English canonical"
        jsonb transcript_metadata
    }
    
    analysis_results {
        int id PK
        int analysis_id FK
        text summary "English canonical"
        jsonb features "English canonical"
    }
    
    localization_translations {
        bigint id PK
        varchar surface "analysis_detail"
        int row_id "FK to analyses.id or analysis_results.id"
        varchar column_name "features, summary, title..."
        varchar json_path "results[0].features[2].title"
        varchar locale "ja, vi, ko..."
        text source_text "English gốc để detect change"
        text translated_text "Đã dịch"
        varchar prompt_hash "Phiên bản prompt"
        timestamp created_at
        timestamp updated_at
        boolean is_active
    }

# backend/src/app/main.py — startup event

@app.on_event("startup")
async def warm_localization_cache():
    """Warm Redis cache from PostgreSQL on startup."""
    if not settings.LOCALIZATION_ENABLED:
        return
    
    # Warm các translation gần đây nhất
    recent = await db.execute(
        select(LocalizationTranslation)
        .where(LocalizationTranslation.is_active == True)
        .order_by(LocalizationTranslation.updated_at.desc())
        .limit(1000)
    )
    
    for row in recent:
        cache_key = f"localization:{row.surface}:{row.row_id}:{row.locale}:{row.prompt_hash}"
        await redis.setex(cache_key, 3600, row.translated_text)

Copy

flowchart LR
    W["ARQ Worker<br/>(analysis xong)"] --> P["prime_localization_cache()<br/>for ALL locales"]
    P --> R["Redis SET<br/>localization:{id}:{locale}"]
    
    U["User request"] --> GET["Redis GET"]
    GET -->|Hit 🚀 10ms| TRANS["Trả translated"]
    GET -->|Miss| FALLBACK["Sync LLM gpt-4o-mini<br/>asyncio.wait_for(30s)"]
    FALLBACK -->|OK| TRANS
    FALLBACK -->|Timeout| EN["English + ARQ retry"]

flowchart LR
    REQ["User request"] --> RES["HTTP 200 English<br/>+ X-Job-Id"]
    RES --> POLL["FE polling every 2s<br/>GET /localization/status/{id}"]
    
    ARQ["ARQ Worker"] --> TRANSLLM["Translate gpt-4o-mini"]
    TRANSLLM --> CACHE["Redis SET"]
    CACHE --> POLL
    
    POLL -->|Done| RENDER["FE fetch translated<br/>→ render lại"]
    POLL -->|Pending| POLL

flowchart LR
    REQ["User request"] --> ACQ["Acquire semaphore<br/>wait max 5s"]
    ACQ -->|Busy| EN1["English fallback"]
    ACQ -->|OK| LLM["LLM gpt-4o-mini<br/>asyncio.wait_for 30s"]
    LLM -->|OK| TRANS["Trả translated<br/>+ Redis cache"]
    LLM -->|Timeout| EN2["English fallback"]

flowchart TD
    START["Bạn muốn implement translation?"]
    
    START --> A{"Translation task có<br/>thường xuyên > 30s không?"}
    
    A -->|Có (video/image gen)| B["Approach 2: Async Polling"]
    
    A -->|Không (text translate 2-5s)| C{"Backend đã có<br/>ARQ Worker + Redis?"}
    
    C -->|Không| D["Approach 3: Sync Optimized<br/>(đơn giản, dễ maintain)"]
    
    C -->|Có ✅| E["Approach 1: Cache Priming<br/>tận dụng ARQ + Redis"]
    
    E --> F{"Có analysis cũ<br/>cần backfill không?"}
    
    F -->|Có| G["Chạy backfill job 1 lần<br/>+ bật priming cho analysis mới"]
    F -->|Không| H["Bật priming ngay<br/>cho analysis mới"]
    
    G --> DONE["Done: Zero-latency 🚀"]
    H --> DONE

├── Primary path:  Cache Priming (ARQ worker prime ALL locales)
│   ├── Cache hit  → 10ms response  🚀
│   ├── Cache miss → sync LLM fallback (gpt-4o-mini, 30s timeout)
│   └── LLM error  → English + ARQ retry background
│
├── Backfill:      Migration job for existing analyses (1 time)
│
└── FE change:     Zero (không cần sửa gì)

Copy

localization:{surface}:{locale}:{prompt_version}:{policy_version}:{sha256(canonical_payload)}

Copy

localization:analysis_detail:ja:v1:v1:a1b2c3d4e5...

Copy