docs/20260506_번역언어감지효율화구상.md

@@ -0,0 +1,338 @@
+# 번역/언어감지 효율화 구상
+
+## 배경
+- 현재 구조는 도메인별 번역 테이블과 이벤트 기반 Papago 호출이 혼재되어 있다.
+- 생성/수정 이벤트 번역과 조회 시 캐시 미스 번역이 동시에 존재해 동일 원문이 중복 번역될 수 있다.
+- 번역 저장 로직은 `ContentTranslation`, `SeriesTranslation`, `AiCharacterTranslation`, `OriginalWorkTranslation`, `CategoryTranslation` 등으로 분산되어 있으며, 조회/upsert 패턴이 반복된다.
+- Papago 호출은 텍스트와 대상 locale 단위로 순차 실행되어 대상 하나가 여러 HTTP 요청으로 확장된다.
+
+## 목표
+- Papago 호출 수와 비용을 줄인다.
+- 조회 API에서 Papago 호출로 인한 응답 지연을 줄인다.
+- 번역 저장/조회 로직의 중복을 줄이고, 향후 번역 provider 교체가 가능하도록 한다.
+- 기존 API 응답 스키마는 유지하고, DB 변경은 단계적으로 적용한다.
+
+## 구현 항목
+- [x] `TranslationProvider` 인터페이스와 Papago provider 구현을 분리한다.
+- [x] `translation_memory`, `translation_job`, `language_detection_result` 테이블을 추가한다.
+- [x] 조회 fallback의 직접 Papago 호출을 제거하고 누락 번역 job 예약으로 전환한다.
+- [x] 생성/수정 이벤트 번역을 `translation_job` 기반 워커 처리로 전환한다.
+- [x] `TranslationMemory` 결과를 기존 JSON/scalar read model로 materialize한다.
+- [ ] 목록 조회 병목 측정 후 필요한 도메인에만 hot column을 추가한다.
+
+## 접근안 비교
+
+### 접근안 A: 현재 구조 유지 + 중복 방지 보강
+- 도메인별 번역 테이블은 유지한다.
+- 번역 전 기존 번역 조회와 unique 제약을 강화한다.
+- 조회 fallback에서 번역 생성 대신 번역 작업만 예약하도록 바꾼다.
+
+장점:
+- 변경 범위가 가장 작다.
+- 기존 코드와 데이터 마이그레이션 위험이 낮다.
+
+단점:
+- 도메인별 중복 저장/로직 구조는 그대로 남는다.
+- 원문이 같아도 도메인별로 재번역될 수 있다.
+
+### 접근안 B: 원문 해시 기반 Translation Memory + 비동기 작업 큐 추가
+- 원문 텍스트를 정규화하고 `source_hash`로 중복을 제거한다.
+- `translation_memory`에 원문/언어쌍/번역 결과를 저장한다.
+- 도메인 엔티티는 `translation_job`을 생성하고, 워커가 Papago를 호출해 결과를 저장한다.
+- 조회 API는 Papago를 직접 호출하지 않고 기존 번역을 반환하거나 미번역 상태를 유지한다.
+
+장점:
+- 동일 원문 재번역을 줄일 수 있다.
+- 조회 응답 지연을 안정적으로 제거할 수 있다.
+- 기존 도메인별 번역 테이블을 당장 제거하지 않아도 단계 적용이 가능하다.
+
+단점:
+- 작업 큐, 상태 관리, 재시도 정책이 필요하다.
+- 생성 직후 번역이 아직 없을 수 있으므로 UX/응답 정책을 정해야 한다.
+
+### 접근안 C: 도메인별 번역 테이블을 공통 번역 저장소로 통합
+- 모든 번역 결과를 `translation_entry` 같은 공통 테이블에 저장한다.
+- `resource_type`, `resource_id`, `field_key`, `locale`, `translated_text` 또는 JSON payload로 도메인별 필드를 표현한다.
+- 기존 도메인별 번역 테이블은 읽기 호환 기간 후 제거한다.
+
+장점:
+- 저장/조회/upsert 로직을 크게 단순화할 수 있다.
+- 새 번역 대상 추가 비용이 줄어든다.
+
+단점:
+- 마이그레이션 위험이 가장 크다.
+- 도메인별 payload 검증과 타입 안정성이 약해질 수 있다.
+- 기존 조회 쿼리와 응답 조립 로직 영향 범위가 넓다.
+
+## 권장 방향
+- 최종 구현방식은 접근안 B를 중심으로 하되, 조회 성능을 위해 도메인별 read model을 유지하는 하이브리드 구조로 한다.
+- 기존 번역 데이터를 배제할 수 있어도 모든 번역 결과를 정규화 row만으로 저장하는 방식은 채택하지 않는다.
+- 이유는 Papago 호출 수 절감은 원문 해시 기반 `TranslationMemory`가 담당하고, API 조회 성능은 `entityId + locale` 단위 read model이 담당하는 분리가 가장 효율적이기 때문이다.
+- 2차로 운영 안정화 후 도메인별 read model의 중복을 줄이되, 조회 경로가 복잡해지는 전면 통합은 실제 병목이 확인될 때만 진행한다.
+
+## 최종 구현방식
+
+### 설계 원칙
+- 번역 원장은 정규화한다.
+- 조회 결과는 도메인별로 materialize한다.
+- Papago 호출은 사용자 요청 스레드에서 수행하지 않는다.
+- JSON payload는 원장이 아니라 조회 최적화용 read model로만 사용한다.
+
+### 저장 구조
+
+#### `translation_memory`
+- 동일 원문을 반복 번역하지 않기 위한 공통 번역 캐시이다.
+- 원문은 의미가 바뀌지 않는 선에서 공백, 줄바꿈, Unicode 표현을 정규화한다.
+- 주요 컬럼:
+  - `id`
+  - `source_hash`
+  - `source_text`
+  - `source_language`
+  - `target_language`
+  - `translated_text`
+  - `provider`
+  - `provider_version`
+  - `normalization_version`
+  - `created_at`
+  - `updated_at`
+- unique 제약:
+  - `(source_hash, source_language, target_language, provider, normalization_version)`
+- 역할:
+  - Papago 호출 수 절감의 핵심 원장이다.
+  - 도메인이 달라도 원문과 언어쌍이 같으면 같은 번역을 재사용한다.
+
+#### `translation_job`
+- 번역 실행 상태를 관리하는 durable job 테이블이다.
+- 현재 `@Async` 이벤트만으로 처리하던 작업을 명시적인 상태 모델로 바꾼다.
+- 주요 컬럼:
+  - `id`
+  - `resource_type`
+  - `resource_id`
+  - `field_key`
+  - `source_hash`
+  - `source_language`
+  - `target_language`
+  - `status`
+  - `retry_count`
+  - `last_error_message`
+  - `next_retry_at`
+  - `created_at`
+  - `updated_at`
+- 중복 방지:
+  - 활성 상태 기준 `(resource_type, resource_id, field_key, target_language, source_hash)` 중복 생성을 막는다.
+- 역할:
+  - 중복 작업 방지, 재시도, 실패 추적, 운영자 재처리를 담당한다.
+
+#### `language_detection_result`
+- 언어 감지 결과를 원본 엔티티에만 저장하지 않고 별도 캐시/이력으로 관리한다.
+- 주요 컬럼:
+  - `id`
+  - `source_hash`
+  - `source_text_sample`
+  - `detected_language`
+  - `provider`
+  - `confidence`
+  - `normalization_version`
+  - `created_at`
+- 역할:
+  - 짧은 문구나 반복 문구의 감지 호출을 줄인다.
+  - 원본 엔티티의 `languageCode`는 기존 호환 필드로 유지할 수 있다.
+
+#### 도메인별 번역 read model
+- `ContentTranslation`, `SeriesTranslation`, `AiCharacterTranslation`, `OriginalWorkTranslation`처럼 여러 필드가 묶인 응답은 JSON payload read model을 유지한다.
+- `CategoryTranslation`, `ContentThemeTranslation`, `SeriesGenreTranslation`처럼 단일 문자열만 필요한 대상은 scalar column을 유지한다.
+- JSON payload는 `translation_memory`를 조립한 결과를 저장하는 materialized cache로 본다.
+- provider, retry, source hash 같은 실행/원장 메타데이터는 JSON payload에 넣지 않는다.
+
+### JSON 유지 여부
+- JSON payload 저장은 유지한다.
+- 단, JSON을 번역의 최종 원장으로 보지 않고 API 응답을 빠르게 만들기 위한 read model로 격하한다.
+- 현재 조회 패턴은 JSON 내부 검색이 아니라 `resourceId + locale`로 1-row 조인 후 payload를 읽는 방식이므로, 정규화 row를 매번 pivot하는 것보다 조회 경로가 단순하다.
+- 목록 API에서 제목이나 이름 하나만 자주 읽는데 payload가 커지는 도메인은 `translated_title`, `translated_name` 같은 hot column을 선택적으로 추가한다.
+- 단일 문자열 번역 대상은 JSON으로 바꾸지 않는다.
+
+### 처리 흐름
+- 생성/수정 시 번역 대상 필드를 세그먼트로 추출한다.
+- 세그먼트별 `source_hash`를 계산한다.
+- 언어 정보가 없으면 `language_detection_result`를 조회하고, 없을 때만 감지 작업을 수행한다.
+- 대상 언어별로 `translation_memory`를 먼저 조회한다.
+- cache hit이면 Papago를 호출하지 않고 read model materialize 단계로 넘어간다.
+- cache miss이면 `translation_job`을 `PENDING`으로 생성한다.
+- 워커가 job을 가져와 Papago를 호출하고 `translation_memory`에 저장한다.
+- 대상 resource의 모든 필드 번역이 준비되면 도메인별 read model JSON 또는 scalar row를 갱신한다.
+
+### 조회 정책
+- 조회 API에서는 Papago를 직접 호출하지 않는다.
+- 번역 read model이 있으면 해당 번역을 반환한다.
+- 번역 read model이 없으면 원문 또는 기존 fallback locale을 반환한다.
+- 누락된 번역은 조회 요청에서 job만 예약할 수 있으며, 외부 API 응답을 기다리지 않는다.
+
+### 최종 선택 요약
+| 영역 | 최종 선택 | 이유 |
+|---|---|---|
+| Papago 중복 호출 방지 | `translation_memory` 정규화 | 동일 원문/언어쌍 재사용이 가능하다. |
+| 번역 실행 | `translation_job` 비동기 큐 | 재시도와 실패 추적이 가능하고 조회 응답을 지연시키지 않는다. |
+| 다중 필드 응답 저장 | JSON read model 유지 | `entityId + locale` 1-row 조회가 가능해 API 조립이 단순하다. |
+| 단일 문자열 번역 저장 | scalar column 유지 | JSON보다 단순하고 불필요한 변환 비용이 없다. |
+| 자주 읽는 일부 필드 | 선택적 hot column 추가 | 큰 JSON payload 전체 로딩 비용을 줄일 수 있다. |
+| provider 확장 | `TranslationProvider` 인터페이스 | Papago 의존을 낮추고 향후 교체를 쉽게 한다. |
+
+## 상세 정책
+
+### 운영자 재처리
+- 관리자/운영 API에서는 번역 상태를 확인하고 재번역을 요청할 수 있게 한다.
+- 실패한 job은 원문, 대상 언어, provider, 실패 사유, 재시도 횟수를 함께 보여준다.
+
+### 중복 방지 정책
+- `TranslationMemory`는 `(source_hash, source_language, target_language, provider, normalization_version)`에 unique 제약을 둔다.
+- `TranslationJob`은 활성 상태 기준 `(resource_type, resource_id, field_key, target_language, source_hash)` 중복 생성을 막는다.
+- 동일 작업이 동시에 들어오면 기존 `PENDING` 또는 `RUNNING` 작업을 재사용한다.
+
+### 오류와 재시도
+- Papago 실패 시 `TranslationJob.status = FAILED`와 실패 사유를 저장한다.
+- 일시 오류는 지수 백오프로 재시도한다.
+- 영구 오류나 빈 원문은 재시도하지 않는다.
+- 재시도 횟수를 초과하면 운영자가 재시도할 수 있도록 별도 상태로 남긴다.
+
+### 운영 안정화 보완 작업
+- 현재 `TranslationJobWorker.claimNextJob()`는 `PENDING` job을 조회한 뒤 `RUNNING`으로 변경한다.
+- 다중 애플리케이션 인스턴스에서 같은 job을 동시에 조회할 수 있으므로 운영 반영 전 claim을 원자화한다.
+- 권장 방식은 MySQL 기준 `SELECT ... FOR UPDATE SKIP LOCKED` 또는 `UPDATE ... WHERE status = 'PENDING' ... LIMIT 1` 기반 원자적 claim이다.
+- job row를 여러 인스턴스가 나눠 처리하는 목적에는 `FOR UPDATE SKIP LOCKED` 또는 atomic update claim을 우선한다.
+- `ShedLock`은 스케줄러 실행 자체의 중복을 막는 용도로는 적합하지만, job row 단위 분산 claim을 대체하지는 않는다.
+- Papago 호출은 DB lock을 잡은 트랜잭션 밖에서 수행하고, claim/완료/실패 상태 변경만 짧은 트랜잭션으로 처리한다.
+- `FAILED`로 즉시 종료하는 최소 구현에서 지수 백오프 기반 재시도 정책으로 보완한다.
+- 재시도 정책은 `retry_count`, `next_retry_at`, `last_error_message`를 함께 갱신하고, 최대 재시도 초과 상태를 운영자가 확인할 수 있게 한다.
+- worker 처리량과 부하를 운영 설정으로 제어할 수 있도록 `fixed-delay-ms`, tick당 처리 group 수, 최대 재시도 횟수, Papago 호출 rate limit을 설정화한다.
+- tick당 처리 단위는 단순 job row 수가 아니라 `(resource_type, resource_id, target_language)` group으로 잡는다.
+- 1차 운영 기준은 tick당 최대 5개 group 처리로 둔다.
+- group 내부의 field job은 순차 처리하고, 같은 resource/locale의 모든 필드가 `translation_memory`에 준비된 뒤 read model을 materialize한다.
+- 콘텐츠 기준으로는 1개 group이 `title`, `detail`, `tags` 3개 field job이므로 tick당 최대 15개 field job이 된다.
+- 캐릭터처럼 필드가 많은 리소스도 group 5개 제한 안에서 처리해 Papago 호출 burst를 완화한다.
+- 운영 관측을 위해 pending/running/failed/completed count, oldest pending age, 처리 성공/실패 수, Papago 호출 시간, materialize 실패 수를 로그 또는 metric으로 남긴다.
+
+### 번역 job 실행 주기 조정 검토
+- 현재 구현은 `sodalive.translation-job.fixed-delay-ms` 설정이 없으면 기본 `5000ms` fixed delay로 실행된다.
+- 콘텐츠가 지속적으로 올라오는 서비스 형태가 아니므로 기본 주기를 10분(`600000ms`)으로 늘리는 방향을 검토한다.
+- `fixedDelay`는 작업 종료 후 다음 실행까지의 지연 시간이므로 실제 실행 간격은 `처리 시간 + 10분`이 된다.
+- 정확히 벽시계 기준 10분마다 실행해야 한다면 `cron + 스케줄 중복 방지 lock`을 검토하고, 처리 후 10분 쉬는 정책이면 `fixedDelay`를 사용한다.
+- 10분 주기의 장점은 불필요한 DB polling과 Papago 호출 burst 가능성을 줄이고, 낮은 트래픽 환경에서 백그라운드 작업 부하를 완화하는 것이다.
+- 10분 주기의 단점은 번역 read model 반영 지연이 최대 10분 이상으로 늘어날 수 있다는 점이다.
+- 조회 정책이 원문 즉시 반환 + job 예약 방식이므로 API 응답 실패로 이어지지는 않지만, 사용자는 첫 조회 후 최대 다음 worker 실행까지 원문을 볼 수 있다.
+- 기존 tick당 20 job row 처리 방식은 같은 resource/locale의 일부 필드만 처리하고 다음 tick으로 넘어갈 수 있다.
+- 10분 주기에서는 부분 처리된 resource의 read model 반영이 다음 tick까지 지연될 수 있으므로 `(resource_type, resource_id, target_language)` group 단위 처리로 보완한다.
+- 10분 주기를 적용하려면 tick당 처리 group 수를 운영 설정으로 조정하거나, pending backlog가 특정 기준을 넘을 때 수동/운영자 재처리 또는 임시 짧은 주기 전환이 가능해야 한다.
+- 생성 직후 번역 노출이 중요한 리소스가 발견되면 해당 리소스만 별도 즉시 처리 정책을 두고, 일반 조회 fallback은 10분 주기를 유지한다.
+- 1차 운영 기준은 `fixed-delay-ms = 600000`, tick당 최대 5개 group, 원문 fallback 허용, backlog/oldest pending age 모니터링으로 둔다.
+
+### 단계별 적용
+- 1단계: `TranslationProvider` 인터페이스를 만들고 기존 `PapagoTranslationService`를 provider 구현으로 감싼다.
+- 2단계: `translation_memory`, `translation_job`, `language_detection_result` 테이블을 추가한다.
+- 3단계: 조회 fallback의 직접 Papago 호출을 제거하고, 누락 번역 job 예약 방식으로 전환한다.
+- 4단계: 생성/수정 이벤트 번역을 `translation_job` 기반 워커로 전환한다.
+- 5단계: `TranslationMemory` 결과를 기존 JSON/scalar read model로 materialize한다.
+- 6단계: 목록 조회에서 큰 JSON payload 로딩이 병목이면 hot column을 선택적으로 추가한다.
+
+## 검증 관점
+- 동일 원문을 여러 도메인에서 번역해도 Papago 호출이 한 번만 발생하는지 확인한다.
+- 조회 API에서 Papago 호출이 발생하지 않는지 확인한다.
+- 생성/수정 이벤트 후 번역 작업이 중복 생성되지 않는지 확인한다.
+- Papago 장애 시 원 API 응답이 실패하지 않고 작업 상태만 실패로 남는지 확인한다.
+- 기존 API 응답 스키마가 바뀌지 않는지 확인한다.
+
+## 남은 결정사항
+- 번역이 없는 조회 응답은 원문을 반환한다.
+- 실시간 번역이 꼭 필요한 엔드포인트는 1차 구현 범위에 포함하지 않는다. 단, 생성 직후 번역 노출이 중요한 리소스가 확인되면 해당 리소스만 별도 즉시 처리 정책을 검토한다.
+- JSON read model을 장기 유지하되, hot column이 필요한 목록 API를 측정으로 결정해야 한다.
+- Papago 외 provider는 1차 구현에서는 인터페이스만 준비하고 실제 추가 provider는 범위에서 제외한다.
+
+## 검증 기록
+- 2026-05-06: 코드 탐색 결과를 바탕으로 현재 번역 저장 구조, Papago 호출 흐름, 외부 아키텍처 패턴을 비교해 설계 초안을 작성했다.
+- 2026-05-06: 문서 내 미완성 placeholder, 상충되는 범위, 구현 전제 누락 여부를 점검했다.
+- 2026-05-06: 기존 번역 데이터 보존 제약을 배제해도 `TranslationMemory + TranslationJob + JSON read model` 하이브리드가 최종 구현방식으로 적합한지 재평가하고 문서에 반영했다.
+- 2026-05-06: TDD RED 확인: `./gradlew test --tests 'kr.co.vividnext.sodalive.i18n.translation.SourceTextNormalizerTest' --tests 'kr.co.vividnext.sodalive.i18n.translation.TranslationJobSchedulerTest' --tests 'kr.co.vividnext.sodalive.content.LanguageDetectionCacheServiceTest'` 실행 시 `SourceTextNormalizer`, `TranslationJobRepository`, `TranslationJobScheduler`, `LanguageDetectionResultRepository`, `LanguageDetectionCacheService` 미구현으로 `compileTestKotlin` 실패를 확인했다.
+- 2026-05-06: 구현 후 동일 targeted test 명령을 재실행해 `BUILD SUCCESSFUL`을 확인했다. 정규화/해시, 누락 번역 job 생성, 언어 감지 캐시 hit 시 provider 미호출 동작을 검증했다.
+- 2026-05-06: 전체 회귀 검증으로 `./gradlew test`와 `./gradlew build`를 실행해 모두 `BUILD SUCCESSFUL`을 확인했다. `build`에는 ktlint main/test/sourceSet check가 포함되어 스타일도 함께 검증됐다.
+- 2026-05-06: `PapagoTranslationService|translationService\.translate|papagoTranslationService\.translate|TranslateRequest` 검색으로 직접 번역 호출이 `PapagoTranslationService`, `TranslationProvider`, `TranslationJobWorker`, DTO, 지원 언어 목록 참조에만 남아 있음을 확인했다. 조회 fallback과 `LanguageTranslationListener`에는 직접 Papago 번역 호출이 남아 있지 않다.
+- 2026-05-06: MySQL unique 제약은 활성 상태 partial unique를 표현할 수 없으므로 완료 job이 있는 동일 key 재예약 시 중복 insert가 발생하지 않도록 repository 파생 쿼리 기반 회귀 테스트를 추가했다. RED는 중복 job 조회 메서드 미구현으로 `compileTestKotlin` 실패, GREEN은 동일 테스트 `BUILD SUCCESSFUL`로 확인했다.
+- 2026-05-06: 최종 확인에서 Kotlin LSP 서버가 설정되어 있지 않아 `lsp_diagnostics`는 `No LSP server configured for extension: .kt`로 실행할 수 없었다. 대체 검증으로 `./gradlew test`, `./gradlew build`, 신규 focused test `--rerun-tasks`를 실행했고 모두 `BUILD SUCCESSFUL`을 확인했다.
+- 2026-05-06: 운영 안정화 보완 구현으로 `TranslationJobWorker` 기본 fixed delay를 10분(`600000ms`)으로 변경하고, MySQL `FOR UPDATE SKIP LOCKED` 기반 job id claim, 실패 시 `PENDING` 재전환 + `next_retry_at` backoff + 최대 재시도 후 `FAILED` 전환을 적용했다. RED는 `TranslationJobWorkerTest`에서 원자 claim 메서드 미구현으로 `compileTestKotlin` 실패, GREEN은 동일 focused test `BUILD SUCCESSFUL`로 확인했다.
+- 2026-05-06: 문서에 반영한 group 단위 처리 정책을 구현했다. `TranslationJobWorker`는 tick당 최대 5개 `(resource_type, resource_id, target_language)` group을 처리하고, group 내부 pending field job을 `RUNNING`으로 claim한 뒤 모두 성공한 경우 한 번만 read model을 materialize한다. RED는 `findPendingJobIdsForGroupForUpdate`, `processNextGroup` 미구현으로 `TranslationJobWorkerTest`의 `compileTestKotlin` 실패를 확인했고, GREEN은 동일 focused test `BUILD SUCCESSFUL`로 확인했다.
+- 2026-05-06: group 처리 리뷰에서 materialize 실패 후 재시도 불가 가능성과 seed row 기반 claim의 group 분리 가능성을 확인했다. 보완 구현으로 단일 native query `findNextPendingGroupJobIdsForUpdate`에서 다음 pending group의 job id들을 `FOR UPDATE SKIP LOCKED`로 함께 claim하고, materialize 실패 시 group job들을 backoff 재시도 대상으로 되돌리도록 수정했다. RED는 새 group claim 메서드 미구현으로 `TranslationJobWorkerTest`의 `compileTestKotlin` 실패를 확인했고, GREEN은 동일 focused test `BUILD SUCCESSFUL`로 확인했다.
+- 2026-05-06: group 처리 전환 후 남은 이전 row 단위 claim 함수 사용처를 `rg`, AST 검색, explore/librarian 병렬 탐색으로 확인했다. production 경로가 `processNextGroup` + `findNextPendingGroupJobIdsForUpdate`로 수렴되어 `processNextJob`, `findFirstByStatusAndNextRetryAtLessThanEqualOrderByCreatedAtAsc`, `findNextPendingJobIdForUpdate`, `findPendingJobIdsForGroupForUpdate`를 제거했다.
+
+## 2026-05-06 구현 DDL
+
+운영 MySQL은 `spring.jpa.hibernate.ddl-auto=validate`이므로 아래 DDL을 선반영해야 한다. `created_at`, `updated_at`은 `BaseEntity`의 `createdAt`, `updatedAt`과 매핑된다.
+
+```sql
+CREATE TABLE translation_memory (
+    id BIGINT NOT NULL AUTO_INCREMENT COMMENT '번역 메모리 ID',
+    source_hash VARCHAR(64) NOT NULL COMMENT '정규화 원문 SHA-256 해시',
+    source_text TEXT NOT NULL COMMENT '정규화 원문 텍스트',
+    source_language VARCHAR(10) NOT NULL COMMENT '원문 언어 코드',
+    target_language VARCHAR(10) NOT NULL COMMENT '대상 언어 코드',
+    translated_text TEXT NOT NULL COMMENT '번역 결과 텍스트',
+    provider VARCHAR(50) NOT NULL COMMENT '번역 provider 이름',
+    provider_version VARCHAR(50) NOT NULL COMMENT '번역 provider 버전',
+    normalization_version VARCHAR(20) NOT NULL COMMENT '정규화 규칙 버전',
+    created_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP COMMENT '생성 시각',
+    updated_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '수정 시각',
+    PRIMARY KEY (id),
+    UNIQUE KEY uk_translation_memory_source_target_provider (
+        source_hash,
+        source_language,
+        target_language,
+        provider,
+        normalization_version
+    ),
+    KEY idx_translation_memory_source_hash (source_hash),
+    KEY idx_translation_memory_target_language (target_language)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE translation_job (
+    id BIGINT NOT NULL AUTO_INCREMENT COMMENT '번역 작업 ID',
+    resource_type VARCHAR(50) NOT NULL COMMENT '번역 대상 리소스 타입',
+    resource_id BIGINT NOT NULL COMMENT '번역 대상 리소스 ID',
+    field_key VARCHAR(80) NOT NULL COMMENT '번역 대상 필드 키',
+    source_hash VARCHAR(64) NOT NULL COMMENT '정규화 원문 SHA-256 해시',
+    source_text TEXT NOT NULL COMMENT '정규화 원문 텍스트',
+    source_language VARCHAR(10) NOT NULL COMMENT '원문 언어 코드',
+    target_language VARCHAR(10) NOT NULL COMMENT '대상 언어 코드',
+    status VARCHAR(20) NOT NULL COMMENT '번역 작업 상태',
+    retry_count INT NOT NULL COMMENT '재시도 횟수',
+    last_error_message TEXT DEFAULT NULL COMMENT '마지막 오류 메시지',
+    next_retry_at TIMESTAMP NOT NULL COMMENT '다음 재시도 가능 시각',
+    created_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP COMMENT '생성 시각',
+    updated_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '수정 시각',
+    PRIMARY KEY (id),
+    UNIQUE KEY uk_translation_job_resource_field_target_hash (
+        resource_type,
+        resource_id,
+        field_key,
+        target_language,
+        source_hash
+    ),
+    KEY idx_translation_job_status_retry (status, next_retry_at),
+    KEY idx_translation_job_resource (resource_type, resource_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE language_detection_result (
+    id BIGINT NOT NULL AUTO_INCREMENT COMMENT '언어 감지 결과 ID',
+    source_hash VARCHAR(64) NOT NULL COMMENT '정규화 원문 SHA-256 해시',
+    source_text_sample VARCHAR(500) NOT NULL COMMENT '정규화 원문 샘플 텍스트',
+    detected_language VARCHAR(10) NOT NULL COMMENT '감지된 언어 코드',
+    provider VARCHAR(50) NOT NULL COMMENT '언어 감지 provider 이름',
+    confidence DOUBLE DEFAULT NULL COMMENT '언어 감지 신뢰도',
+    normalization_version VARCHAR(20) NOT NULL COMMENT '정규화 규칙 버전',
+    created_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP COMMENT '생성 시각',
+    updated_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '수정 시각',
+    PRIMARY KEY (id),
+    UNIQUE KEY uk_language_detection_result_hash_provider_version (
+        source_hash,
+        provider,
+        normalization_version
+    ),
+    KEY idx_language_detection_result_source_hash (source_hash)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+```

src/main/kotlin/kr/co/vividnext/sodalive/chat/character/controller/ChatCharacterController.kt

@@ -14,8 +14,6 @@ import kr.co.vividnext.sodalive.chat.character.dto.RecentCharacter
 import kr.co.vividnext.sodalive.chat.character.dto.RecentCharactersResponse
 import kr.co.vividnext.sodalive.chat.character.service.ChatCharacterBannerService
 import kr.co.vividnext.sodalive.chat.character.service.ChatCharacterService
-import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslation
-import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslationRenderedPayload
 import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslationRepository
 import kr.co.vividnext.sodalive.chat.character.translate.TranslatedAiCharacterBackground
 import kr.co.vividnext.sodalive.chat.character.translate.TranslatedAiCharacterDetail
@@ -24,8 +22,8 @@ import kr.co.vividnext.sodalive.chat.room.service.ChatRoomService
 import kr.co.vividnext.sodalive.common.ApiResponse
 import kr.co.vividnext.sodalive.common.SodaException
 import kr.co.vividnext.sodalive.i18n.LangContext
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService
-import kr.co.vividnext.sodalive.i18n.translation.TranslateRequest
+import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationTargetType
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
 import kr.co.vividnext.sodalive.member.Member
 import kr.co.vividnext.sodalive.member.contentpreference.MemberContentPreferenceService
 import org.springframework.beans.factory.annotation.Value
@@ -46,7 +44,7 @@ class ChatCharacterController(
     private val characterCommentService: CharacterCommentService,
     private val curationQueryService: CharacterCurationQueryService,
 
-    private val translationService: PapagoTranslationService,
+    private val resourceTranslationJobScheduler: ResourceTranslationJobScheduler,
     private val aiCharacterTranslationRepository: AiCharacterTranslationRepository,
 
     private val langContext: LangContext,
@@ -212,89 +210,11 @@ class ChatCharacterController(
                     tags = payload.tags
                 )
             } else {
-                val texts = mutableListOf<String>()
-                texts.add(character.name)
-                texts.add(character.description)
-                texts.add(character.gender ?: "")
-
-                val hasPersonality = personality != null
-                if (hasPersonality) {
-                    texts.add(personality!!.trait)
-                    texts.add(personality.description)
-                }
-
-                val hasBackground = background != null
-                if (hasBackground) {
-                    texts.add(background!!.topic)
-                    texts.add(background.description)
-                }
-
-                texts.add(tags)
-
-                val sourceLanguage = character.languageCode ?: "ko"
-
-                val response = translationService.translate(
-                    request = TranslateRequest(
-                        texts = texts,
-                        sourceLanguage = sourceLanguage,
-                        targetLanguage = langContext.lang.code
-                    )
+                resourceTranslationJobScheduler.scheduleResourceTranslation(
+                    resourceType = LanguageTranslationTargetType.CHARACTER,
+                    resourceId = character.id!!,
+                    targetLanguage = langContext.lang.code
                 )
-
-                val translatedTexts = response.translatedText
-                if (translatedTexts.size == texts.size) {
-                    var index = 0
-
-                    val translatedName = translatedTexts[index++]
-                    val translatedDescription = translatedTexts[index++]
-                    val translatedGender = translatedTexts[index++]
-
-                    var translatedPersonality: TranslatedAiCharacterPersonality? = null
-                    if (hasPersonality) {
-                        translatedPersonality = TranslatedAiCharacterPersonality(
-                            trait = translatedTexts[index++],
-                            description = translatedTexts[index++]
-                        )
-                    }
-
-                    var translatedBackground: TranslatedAiCharacterBackground? = null
-                    if (hasBackground) {
-                        translatedBackground = TranslatedAiCharacterBackground(
-                            topic = translatedTexts[index++],
-                            description = translatedTexts[index++]
-                        )
-                    }
-
-                    val translatedTags = translatedTexts[index]
-
-                    val payload = AiCharacterTranslationRenderedPayload(
-                        name = translatedName,
-                        description = translatedDescription,
-                        gender = translatedGender,
-                        personalityTrait = translatedPersonality?.trait ?: "",
-                        personalityDescription = translatedPersonality?.description ?: "",
-                        backgroundTopic = translatedBackground?.topic ?: "",
-                        backgroundDescription = translatedBackground?.description ?: "",
-                        tags = translatedTags
-                    )
-
-                    val entity = AiCharacterTranslation(
-                        characterId = character.id!!,
-                        locale = langContext.lang.code,
-                        renderedPayload = payload
-                    )
-
-                    aiCharacterTranslationRepository.save(entity)
-
-                    translated = TranslatedAiCharacterDetail(
-                        name = translatedName,
-                        description = translatedDescription,
-                        gender = translatedGender,
-                        personality = translatedPersonality,
-                        background = translatedBackground,
-                        tags = translatedTags
-                    )
-                }
             }
         }
 

src/main/kotlin/kr/co/vividnext/sodalive/chat/original/service/OriginalWorkTranslationService.kt

@@ -1,28 +1,22 @@
 package kr.co.vividnext.sodalive.chat.original.service
 
 import kr.co.vividnext.sodalive.chat.original.OriginalWork
-import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslation
-import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslationPayload
 import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslationRepository
 import kr.co.vividnext.sodalive.chat.original.translation.TranslatedOriginalWork
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService
-import kr.co.vividnext.sodalive.i18n.translation.TranslateRequest
-import org.slf4j.LoggerFactory
+import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationTargetType
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
 
 @Service
 class OriginalWorkTranslationService(
     private val translationRepository: OriginalWorkTranslationRepository,
-    private val papagoTranslationService: PapagoTranslationService
+    private val resourceTranslationJobScheduler: ResourceTranslationJobScheduler
 ) {
-
-    private val log = LoggerFactory.getLogger(javaClass)
-
     /**
      * 원작의 언어와 요청 언어가 다를 때 번역 데이터를 확보하고 반환한다.
      * - 기존 번역이 있으면 그대로 사용
-     * - 없으면 파파고 번역 수행 후 저장
+     * - 없으면 누락 번역 job 예약 후 null 반환
      * - 실패/불필요 시 null 반환
      */
     @Transactional
@@ -55,70 +49,11 @@ class OriginalWorkTranslationService(
             }
         }
 
-        // 파파고 번역 수행
-        return try {
-            val tags = originalWork.tagMappings.map { it.tag.tag }.filter { it.isNotBlank() }
-            val texts = buildList {
-                add(originalWork.title)
-                add(originalWork.contentType)
-                add(originalWork.category)
-                add(originalWork.description)
-                addAll(tags)
-            }
-
-            val response = papagoTranslationService.translate(
-                TranslateRequest(
-                    texts = texts,
-                    sourceLanguage = source,
-                    targetLanguage = target
-                )
-            )
-
-            val out = response.translatedText
-            if (out.isEmpty()) return null
-
-            // 앞 4개는 필드, 나머지는 태그
-            val title = out.getOrNull(0)?.trim().orEmpty()
-            val contentType = out.getOrNull(1)?.trim().orEmpty()
-            val category = out.getOrNull(2)?.trim().orEmpty()
-            val description = out.getOrNull(3)?.trim().orEmpty()
-            val translatedTags = if (out.size > 4) {
-                out.drop(4).map { it.trim() }.filter { it.isNotEmpty() }
-            } else {
-                emptyList()
-            }
-
-            val hasAny = title.isNotEmpty() || contentType.isNotEmpty() ||
-                category.isNotEmpty() || description.isNotEmpty() || translatedTags.isNotEmpty()
-            if (!hasAny) return null
-
-            val payload = OriginalWorkTranslationPayload(
-                title = title,
-                contentType = contentType,
-                category = category,
-                description = description,
-                tags = translatedTags
-            )
-
-            val entity = existed?.apply { this.renderedPayload = payload }
-                ?: OriginalWorkTranslation(
-                    originalWorkId = originalWork.id!!,
-                    locale = target,
-                    renderedPayload = payload
-                )
-
-            translationRepository.save(entity)
-
-            TranslatedOriginalWork(
-                title = title,
-                contentType = contentType,
-                category = category,
-                description = description,
-                tags = translatedTags
-            )
-        } catch (e: Exception) {
-            log.warn("Failed to translate OriginalWork(id={}) from {} to {}: {}", originalWork.id, source, target, e.message)
-            null
-        }
+        resourceTranslationJobScheduler.scheduleResourceTranslation(
+            resourceType = LanguageTranslationTargetType.ORIGINAL_WORK,
+            resourceId = originalWork.id!!,
+            targetLanguage = target
+        )
+        return null
     }
 }

src/main/kotlin/kr/co/vividnext/sodalive/content/AudioContentService.kt

@@ -22,8 +22,6 @@ import kr.co.vividnext.sodalive.content.pin.PinContent
 import kr.co.vividnext.sodalive.content.pin.PinContentRepository
 import kr.co.vividnext.sodalive.content.theme.AudioContentThemeQueryRepository
 import kr.co.vividnext.sodalive.content.theme.translation.ContentThemeTranslationRepository
-import kr.co.vividnext.sodalive.content.translation.ContentTranslation
-import kr.co.vividnext.sodalive.content.translation.ContentTranslationPayload
 import kr.co.vividnext.sodalive.content.translation.ContentTranslationRepository
 import kr.co.vividnext.sodalive.content.translation.TranslatedContent
 import kr.co.vividnext.sodalive.explorer.ExplorerQueryRepository
@@ -36,8 +34,7 @@ import kr.co.vividnext.sodalive.i18n.LangContext
 import kr.co.vividnext.sodalive.i18n.SodaMessageSource
 import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationEvent
 import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationTargetType
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService
-import kr.co.vividnext.sodalive.i18n.translation.TranslateRequest
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
 import kr.co.vividnext.sodalive.member.Member
 import kr.co.vividnext.sodalive.member.block.BlockMemberRepository
 import kr.co.vividnext.sodalive.member.contentpreference.isAdultVisibleByPolicy
@@ -70,7 +67,7 @@ class AudioContentService(
     private val audioContentLikeRepository: AudioContentLikeRepository,
     private val pinContentRepository: PinContentRepository,
 
-    private val translationService: PapagoTranslationService,
+    private val resourceTranslationJobScheduler: ResourceTranslationJobScheduler,
     private val contentTranslationRepository: ContentTranslationRepository,
 
     private val s3Uploader: S3Uploader,
@@ -770,7 +767,7 @@ class AudioContentService(
          * TranslatedContent로 가공한다
          *
          * 번역 콘텐츠가 없으면
-         * 파파고 API를 통해 번역한 후 저장한다.
+         * 누락 번역 job을 예약하고 원문 응답을 유지한다.
          *
          * 번역 대상: title, detail, tags
          *
@@ -792,49 +789,11 @@ class AudioContentService(
                     tags = payload.tags
                 )
             } else {
-                val texts = mutableListOf<String>()
-                texts.add(audioContent.title)
-                texts.add(audioContent.detail)
-                texts.add(tag)
-
-                val sourceLanguage = audioContent.languageCode ?: "ko"
-
-                val response = translationService.translate(
-                    request = TranslateRequest(
-                        texts = texts,
-                        sourceLanguage = sourceLanguage,
-                        targetLanguage = langContext.lang.code
-                    )
+                resourceTranslationJobScheduler.scheduleResourceTranslation(
+                    resourceType = LanguageTranslationTargetType.CONTENT,
+                    resourceId = audioContent.id!!,
+                    targetLanguage = langContext.lang.code
                 )
-
-                val translatedTexts = response.translatedText
-                if (translatedTexts.size == texts.size) {
-                    var index = 0
-
-                    val translatedTitle = translatedTexts[index++]
-                    val translatedDetail = translatedTexts[index++]
-                    val translatedTags = translatedTexts[index]
-
-                    val payload = ContentTranslationPayload(
-                        title = translatedTitle,
-                        detail = translatedDetail,
-                        tags = translatedTags
-                    )
-
-                    contentTranslationRepository.save(
-                        ContentTranslation(
-                            contentId = audioContent.id!!,
-                            locale = langContext.lang.code,
-                            renderedPayload = payload
-                        )
-                    )
-
-                    translated = TranslatedContent(
-                        title = translatedTitle,
-                        detail = translatedDetail,
-                        tags = translatedTags
-                    )
-                }
             }
         }
 

src/main/kotlin/kr/co/vividnext/sodalive/content/LanguageDetectEvent.kt

@@ -60,6 +60,7 @@ class LanguageDetectListener(
     private val seriesRepository: ContentSeriesRepository,
     private val originalWorkRepository: OriginalWorkRepository,
     private val categoryRepository: CategoryRepository,
+    private val languageDetectionCacheService: LanguageDetectionCacheService,
 
     private val applicationEventPublisher: ApplicationEventPublisher,
 
@@ -116,7 +117,7 @@ class LanguageDetectListener(
             return
         }
 
-        val langCode = requestPapagoLanguageCode(event.query, characterId) ?: return
+        val langCode = detectLanguageCode(event, characterId) ?: return
 
         character.languageCode = langCode
         chatCharacterRepository.save(character)
@@ -154,7 +155,7 @@ class LanguageDetectListener(
             return
         }
 
-        val langCode = requestPapagoLanguageCode(event.query, contentId) ?: return
+        val langCode = detectLanguageCode(event, contentId) ?: return
 
         audioContent.languageCode = langCode
 
@@ -194,7 +195,7 @@ class LanguageDetectListener(
             return
         }
 
-        val langCode = requestPapagoLanguageCode(event.query, commentId) ?: return
+        val langCode = detectLanguageCode(event, commentId) ?: return
 
         comment.languageCode = langCode
         audioContentCommentRepository.save(comment)
@@ -226,7 +227,7 @@ class LanguageDetectListener(
             return
         }
 
-        val langCode = requestPapagoLanguageCode(event.query, commentId) ?: return
+        val langCode = detectLanguageCode(event, commentId) ?: return
 
         comment.languageCode = langCode
         characterCommentRepository.save(comment)
@@ -257,7 +258,7 @@ class LanguageDetectListener(
             return
         }
 
-        val langCode = requestPapagoLanguageCode(event.query, cheersId) ?: return
+        val langCode = detectLanguageCode(event, cheersId) ?: return
 
         cheers.languageCode = langCode
         creatorCheersRepository.save(cheers)
@@ -288,7 +289,7 @@ class LanguageDetectListener(
             return
         }
 
-        val langCode = requestPapagoLanguageCode(event.query, seriesId) ?: return
+        val langCode = detectLanguageCode(event, seriesId) ?: return
 
         series.languageCode = langCode
         seriesRepository.save(series)
@@ -326,7 +327,7 @@ class LanguageDetectListener(
             return
         }
 
-        val langCode = requestPapagoLanguageCode(event.query, originalWorkId) ?: return
+        val langCode = detectLanguageCode(event, originalWorkId) ?: return
 
         originalWork.languageCode = langCode
         originalWorkRepository.save(originalWork)
@@ -352,7 +353,7 @@ class LanguageDetectListener(
         val category = categoryRepository.findByIdOrNull(categoryId) ?: return
         if (!category.languageCode.isNullOrBlank()) return
 
-        val langCode = requestPapagoLanguageCode(event.query, categoryId) ?: return
+        val langCode = detectLanguageCode(event, categoryId) ?: return
 
         category.languageCode = langCode
         categoryRepository.save(category)
@@ -365,6 +366,12 @@ class LanguageDetectListener(
         )
     }
 
+    private fun detectLanguageCode(event: LanguageDetectEvent, targetIdForLog: Long): String? {
+        return languageDetectionCacheService.detectWithCache(event.query) {
+            requestPapagoLanguageCode(event.query, targetIdForLog)
+        }
+    }
+
     private fun requestPapagoLanguageCode(query: String, targetIdForLog: Long): String? {
         return try {
             val headers = HttpHeaders().apply {

src/main/kotlin/kr/co/vividnext/sodalive/content/LanguageDetectionCacheService.kt

@@ -0,0 +1,46 @@
+package kr.co.vividnext.sodalive.content
+
+import kr.co.vividnext.sodalive.i18n.translation.SourceTextNormalizer
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+
+@Service
+class LanguageDetectionCacheService(
+    private val languageDetectionResultRepository: LanguageDetectionResultRepository
+) {
+    @Transactional
+    fun detectWithCache(
+        query: String,
+        provider: String = DEFAULT_PROVIDER,
+        detector: () -> String?
+    ): String? {
+        val normalizedQuery = SourceTextNormalizer.normalize(query)
+        if (normalizedQuery.isBlank()) return null
+
+        val sourceHash = SourceTextNormalizer.hash(normalizedQuery)
+        val cached = languageDetectionResultRepository.findBySourceHashAndProviderAndNormalizationVersion(
+            sourceHash = sourceHash,
+            provider = provider,
+            normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+        )
+        if (cached != null) return cached.detectedLanguage
+
+        val detectedLanguage = detector()?.takeIf { it.isNotBlank() } ?: return null
+        languageDetectionResultRepository.save(
+            LanguageDetectionResult(
+                sourceHash = sourceHash,
+                sourceTextSample = normalizedQuery.take(MAX_SAMPLE_LENGTH),
+                detectedLanguage = detectedLanguage.lowercase(),
+                provider = provider,
+                confidence = null,
+                normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+            )
+        )
+        return detectedLanguage.lowercase()
+    }
+
+    companion object {
+        const val DEFAULT_PROVIDER = "papago"
+        private const val MAX_SAMPLE_LENGTH = 500
+    }
+}

src/main/kotlin/kr/co/vividnext/sodalive/content/LanguageDetectionResult.kt

@@ -0,0 +1,38 @@
+package kr.co.vividnext.sodalive.content
+
+import kr.co.vividnext.sodalive.common.BaseEntity
+import kr.co.vividnext.sodalive.i18n.translation.SourceTextNormalizer
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.Table
+import javax.persistence.UniqueConstraint
+
+@Entity
+@Table(
+    name = "language_detection_result",
+    uniqueConstraints = [
+        UniqueConstraint(
+            name = "uk_language_detection_result_hash_provider_version",
+            columnNames = ["source_hash", "provider", "normalization_version"]
+        )
+    ]
+)
+class LanguageDetectionResult(
+    @Column(name = "source_hash", nullable = false, length = 64)
+    val sourceHash: String,
+
+    @Column(name = "source_text_sample", nullable = false, length = 500)
+    val sourceTextSample: String,
+
+    @Column(name = "detected_language", nullable = false, length = 10)
+    val detectedLanguage: String,
+
+    @Column(name = "provider", nullable = false, length = 50)
+    val provider: String,
+
+    @Column(name = "confidence")
+    val confidence: Double? = null,
+
+    @Column(name = "normalization_version", nullable = false, length = 20)
+    val normalizationVersion: String = SourceTextNormalizer.NORMALIZATION_VERSION
+) : BaseEntity()

src/main/kotlin/kr/co/vividnext/sodalive/content/LanguageDetectionResultRepository.kt

@@ -0,0 +1,11 @@
+package kr.co.vividnext.sodalive.content
+
+import org.springframework.data.jpa.repository.JpaRepository
+
+interface LanguageDetectionResultRepository : JpaRepository<LanguageDetectionResult, Long> {
+    fun findBySourceHashAndProviderAndNormalizationVersion(
+        sourceHash: String,
+        provider: String,
+        normalizationVersion: String
+    ): LanguageDetectionResult?
+}

src/main/kotlin/kr/co/vividnext/sodalive/content/category/CategoryService.kt

@@ -7,8 +7,7 @@ import kr.co.vividnext.sodalive.content.LanguageDetectTargetType
 import kr.co.vividnext.sodalive.i18n.LangContext
 import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationEvent
 import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationTargetType
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService
-import kr.co.vividnext.sodalive.i18n.translation.TranslateRequest
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
 import kr.co.vividnext.sodalive.member.Member
 import kr.co.vividnext.sodalive.member.block.BlockMemberRepository
 import org.springframework.context.ApplicationEventPublisher
@@ -25,7 +24,7 @@ class CategoryService(
 
     private val langContext: LangContext,
     private val applicationEventPublisher: ApplicationEventPublisher,
-    private val translationService: PapagoTranslationService
+    private val resourceTranslationJobScheduler: ResourceTranslationJobScheduler
 ) {
     @Transactional
     fun createCategory(request: CreateCategoryRequest, member: Member) {
@@ -148,7 +147,7 @@ class CategoryService(
             .findByCategoryIdInAndLocale(categoryIds, locale)
             .associateBy { it.categoryId }
 
-        // 각 항목에 대해 번역 적용. 없으면 Papago로 번역 저장 후 적용
+        // 각 항목에 대해 번역 적용. 없으면 누락 번역 job만 예약하고 원문을 반환한다.
         val result = mutableListOf<GetCategoryListResponse>()
         for (item in baseList) {
             val entity = entityMap[item.categoryId]
@@ -165,38 +164,11 @@ class CategoryService(
                     continue
                 }
 
-                // 번역본이 없으면 Papago 번역 후 저장
-                val texts = listOf(entity.title)
-                val response = translationService.translate(
-                    request = TranslateRequest(
-                        texts = texts,
-                        sourceLanguage = sourceLang,
-                        targetLanguage = locale
-                    )
+                resourceTranslationJobScheduler.scheduleResourceTranslation(
+                    resourceType = LanguageTranslationTargetType.CREATOR_CONTENT_CATEGORY,
+                    resourceId = entity.id!!,
+                    targetLanguage = locale
                 )
-
-                val translatedTexts = response.translatedText
-                if (translatedTexts.size == texts.size) {
-                    val translatedCategory = translatedTexts[0]
-
-                    val existingOne = categoryTranslationRepository
-                        .findByCategoryIdAndLocale(entity.id!!, locale)
-                    if (existingOne == null) {
-                        categoryTranslationRepository.save(
-                            CategoryTranslation(
-                                categoryId = entity.id!!,
-                                locale = locale,
-                                category = translatedCategory
-                            )
-                        )
-                    } else {
-                        existingOne.category = translatedCategory
-                        categoryTranslationRepository.save(existingOne)
-                    }
-
-                    result.add(GetCategoryListResponse(categoryId = item.categoryId, category = translatedCategory))
-                    continue
-                }
             }
 
             // 번역이 필요 없거나 실패한 경우 원본 사용

src/main/kotlin/kr/co/vividnext/sodalive/content/series/ContentSeriesService.kt

@@ -7,8 +7,6 @@ import kr.co.vividnext.sodalive.content.order.OrderType
 import kr.co.vividnext.sodalive.content.series.content.ContentSeriesContentRepository
 import kr.co.vividnext.sodalive.content.series.content.GetSeriesContentListResponse
 import kr.co.vividnext.sodalive.content.series.translation.SeriesGenreTranslationRepository
-import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslation
-import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslationPayload
 import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslationRepository
 import kr.co.vividnext.sodalive.content.series.translation.TranslatedSeries
 import kr.co.vividnext.sodalive.content.translation.ContentTranslationRepository
@@ -17,8 +15,8 @@ import kr.co.vividnext.sodalive.creator.admin.content.series.SeriesSortType
 import kr.co.vividnext.sodalive.explorer.ExplorerQueryRepository
 import kr.co.vividnext.sodalive.i18n.Lang
 import kr.co.vividnext.sodalive.i18n.LangContext
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService
-import kr.co.vividnext.sodalive.i18n.translation.TranslateRequest
+import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationTargetType
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
 import kr.co.vividnext.sodalive.member.Member
 import kr.co.vividnext.sodalive.member.block.BlockMemberRepository
 import kr.co.vividnext.sodalive.member.contentpreference.isAdultVisibleByPolicy
@@ -41,7 +39,7 @@ class ContentSeriesService(
     private val seriesTranslationRepository: SeriesTranslationRepository,
     private val seriesGenreTranslationRepository: SeriesGenreTranslationRepository,
     private val contentTranslationRepository: ContentTranslationRepository,
-    private val translationService: PapagoTranslationService,
+    private val resourceTranslationJobScheduler: ResourceTranslationJobScheduler,
 
     @Value("\${cloud.aws.cloud-front.host}")
     private val coverImageHost: String
@@ -91,7 +89,7 @@ class ContentSeriesService(
     fun getGenreList(memberId: Long, isAdult: Boolean, contentType: ContentType): List<GetSeriesGenreListResponse> {
         /**
          * langContext.lang == Lang.EN || Lang.JA 일 때 번역된 콘텐츠 테마 반환
-         * 번역이 없으면 번역 API 호출 후 저장하고 반환
+         * 번역이 없으면 누락 번역 job만 예약하고 원문을 반환
          */
         val genres = repository.getGenreList(memberId = memberId, isAdult = isAdult, contentType = contentType)
 
@@ -120,32 +118,12 @@ class ContentSeriesService(
 
             // 미번역 항목 수집
             val untranslated = genres.filter { existingMap[it.id] == null }
-            if (untranslated.isNotEmpty()) {
-                val texts = untranslated.map { it.genre }
-                val response = translationService.translate(
-                    request = TranslateRequest(
-                        texts = texts,
-                        sourceLanguage = "ko",
-                        targetLanguage = targetLocale
-                    )
+            untranslated.forEach { item ->
+                resourceTranslationJobScheduler.scheduleResourceTranslation(
+                    resourceType = LanguageTranslationTargetType.SERIES_GENRE,
+                    resourceId = item.id,
+                    targetLanguage = targetLocale
                 )
-
-                val translatedTexts = response.translatedText
-                val toSave = mutableListOf<kr.co.vividnext.sodalive.content.series.translation.SeriesGenreTranslation>()
-                untranslated.forEachIndexed { index, item ->
-                    val translated = translatedTexts.getOrNull(index) ?: item.genre
-                    toSave.add(
-                        kr.co.vividnext.sodalive.content.series.translation.SeriesGenreTranslation(
-                            seriesGenreId = item.id,
-                            locale = targetLocale,
-                            genre = translated
-                        )
-                    )
-                }
-                if (toSave.isNotEmpty()) {
-                    seriesGenreTranslationRepository.saveAll(toSave)
-                    toSave.forEach { saved -> existingMap[saved.seriesGenreId] = saved }
-                }
             }
 
             // 원래 순서 보존하여 결과 조립
@@ -283,7 +261,7 @@ class ContentSeriesService(
          * TranslatedSeries로 가공한다
          *
          * 번역 콘텐츠가 없으면
-         * 파파고 API를 통해 번역한 후 저장한다.
+         * 누락 번역 job을 예약하고 원문 응답을 유지한다.
          *
          * 번역 대상: title, introduction, keywordList
          *
@@ -309,54 +287,11 @@ class ContentSeriesService(
                         keywords = kws
                     )
                 } else {
-                    val texts = mutableListOf<String>()
-                    texts.add(series.title)
-                    texts.add(series.introduction)
-                    // 키워드는 개별 항목으로 번역 요청하여 N회 호출을 방지한다.
-                    val keywordListForTranslate = keywordList
-                    texts.addAll(keywordListForTranslate)
-
-                    val response = translationService.translate(
-                        request = TranslateRequest(
-                            texts = texts,
-                            sourceLanguage = languageCode,
-                            targetLanguage = locale
-                        )
+                    resourceTranslationJobScheduler.scheduleResourceTranslation(
+                        resourceType = LanguageTranslationTargetType.SERIES,
+                        resourceId = seriesId,
+                        targetLanguage = locale
                     )
-
-                    val translatedTexts = response.translatedText
-                    if (translatedTexts.size == texts.size) {
-                        var index = 0
-                        val translatedTitle = translatedTexts[index++]
-                        val translatedIntroduction = translatedTexts[index++]
-                        val translatedKeywords = if (keywordListForTranslate.isNotEmpty()) {
-                            translatedTexts.subList(index, translatedTexts.size)
-                        } else {
-                            // 번역할 키워드가 없으면 원본 키워드 반환 정책 적용
-                            keywordList
-                        }
-
-                        val payload = SeriesTranslationPayload(
-                            title = translatedTitle,
-                            introduction = translatedIntroduction,
-                            keywords = translatedKeywords
-                        )
-
-                        seriesTranslationRepository.save(
-                            SeriesTranslation(
-                                seriesId = seriesId,
-                                locale = locale,
-                                renderedPayload = payload
-                            )
-                        )
-
-                        val kws = translatedKeywords.ifEmpty { keywordList }
-                        translated = TranslatedSeries(
-                            title = translatedTitle,
-                            introduction = translatedIntroduction,
-                            keywords = kws
-                        )
-                    }
                 }
             }
         }

src/main/kotlin/kr/co/vividnext/sodalive/content/theme/AudioContentThemeService.kt

@@ -5,12 +5,11 @@ import kr.co.vividnext.sodalive.content.AudioContentRepository
 import kr.co.vividnext.sodalive.content.ContentType
 import kr.co.vividnext.sodalive.content.SortType
 import kr.co.vividnext.sodalive.content.theme.content.GetContentByThemeResponse
-import kr.co.vividnext.sodalive.content.theme.translation.ContentThemeTranslation
 import kr.co.vividnext.sodalive.content.theme.translation.ContentThemeTranslationRepository
 import kr.co.vividnext.sodalive.i18n.Lang
 import kr.co.vividnext.sodalive.i18n.LangContext
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService
-import kr.co.vividnext.sodalive.i18n.translation.TranslateRequest
+import kr.co.vividnext.sodalive.i18n.translation.LanguageTranslationTargetType
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
 import kr.co.vividnext.sodalive.member.Member
 import kr.co.vividnext.sodalive.member.contentpreference.isAdultVisibleByPolicy
 import org.springframework.stereotype.Service
@@ -22,7 +21,7 @@ class AudioContentThemeService(
     private val contentRepository: AudioContentRepository,
     private val contentThemeTranslationRepository: ContentThemeTranslationRepository,
 
-    private val papagoTranslationService: PapagoTranslationService,
+    private val resourceTranslationJobScheduler: ResourceTranslationJobScheduler,
     private val langContext: LangContext
 ) {
     @Transactional(readOnly = true)
@@ -51,7 +50,7 @@ class AudioContentThemeService(
 
         /**
          * langContext.lang == Lang.EN || Lang.JA 일 때 번역된 콘텐츠 테마 반환
-         * 번역이 없으면 번역 API 호출 후 저장하고 반환
+         * 번역이 없으면 누락 번역 job만 예약하고 원문을 반환
          */
         val currentLang = langContext.lang
         if (currentLang == Lang.EN || currentLang == Lang.JA) {
@@ -66,43 +65,14 @@ class AudioContentThemeService(
 
             val existingMap = existingTranslations.associateBy { it.contentThemeId }
 
-            // 2) 미번역 항목만 수집하여 한 번에 번역 요청
+            // 2) 미번역 항목은 조회 스레드에서 번역하지 않고 job만 예약
             val untranslatedPairs = themesWithIds.filter { existingMap[it.id] == null }
-
-            if (untranslatedPairs.isNotEmpty()) {
-                val texts = untranslatedPairs.map { it.theme }
-
-                val response = papagoTranslationService.translate(
-                    request = TranslateRequest(
-                        texts = texts,
-                        sourceLanguage = "ko",
-                        targetLanguage = targetLocale
-                    )
+            untranslatedPairs.forEach { pair ->
+                resourceTranslationJobScheduler.scheduleResourceTranslation(
+                    resourceType = LanguageTranslationTargetType.CONTENT_THEME,
+                    resourceId = pair.id,
+                    targetLanguage = targetLocale
                 )
-
-                val translatedTexts = response.translatedText
-                val entitiesToSave = mutableListOf<ContentThemeTranslation>()
-
-                // translatedTexts 크기가 다르면 안전하게 원문으로 대체
-                untranslatedPairs.forEachIndexed { index, pair ->
-                    val translated = translatedTexts.getOrNull(index) ?: pair.theme
-                    entitiesToSave.add(
-                        ContentThemeTranslation(
-                            contentThemeId = pair.id,
-                            locale = targetLocale,
-                            theme = translated
-                        )
-                    )
-                }
-
-                if (entitiesToSave.isNotEmpty()) {
-                    contentThemeTranslationRepository.saveAll(entitiesToSave)
-                }
-
-                // 저장 후 맵을 갱신
-                entitiesToSave.forEach { entity ->
-                    (existingMap as MutableMap)[entity.contentThemeId] = entity
-                }
             }
 
             // 3) 원래 순서대로 결과 조립 (번역 없으면 원문 fallback)

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/LanguageTranslationEvent.kt

@@ -1,35 +1,6 @@
 package kr.co.vividnext.sodalive.i18n.translation
 
-import kr.co.vividnext.sodalive.admin.content.series.AdminContentSeriesRepository
-import kr.co.vividnext.sodalive.admin.content.series.genre.AdminContentSeriesGenreRepository
-import kr.co.vividnext.sodalive.chat.character.repository.ChatCharacterRepository
-import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslation
-import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslationRenderedPayload
-import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslationRepository
-import kr.co.vividnext.sodalive.chat.character.translate.TranslatedAiCharacterBackground
-import kr.co.vividnext.sodalive.chat.character.translate.TranslatedAiCharacterPersonality
-import kr.co.vividnext.sodalive.chat.original.OriginalWorkRepository
-import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslation
-import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslationPayload
-import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslationRepository
-import kr.co.vividnext.sodalive.content.AudioContentRepository
-import kr.co.vividnext.sodalive.content.category.CategoryRepository
-import kr.co.vividnext.sodalive.content.category.CategoryTranslation
-import kr.co.vividnext.sodalive.content.category.CategoryTranslationRepository
-import kr.co.vividnext.sodalive.content.series.translation.SeriesGenreTranslation
-import kr.co.vividnext.sodalive.content.series.translation.SeriesGenreTranslationRepository
-import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslation
-import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslationPayload
-import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslationRepository
-import kr.co.vividnext.sodalive.content.theme.AudioContentThemeQueryRepository
-import kr.co.vividnext.sodalive.content.theme.translation.ContentThemeTranslation
-import kr.co.vividnext.sodalive.content.theme.translation.ContentThemeTranslationRepository
-import kr.co.vividnext.sodalive.content.translation.ContentTranslation
-import kr.co.vividnext.sodalive.content.translation.ContentTranslationPayload
-import kr.co.vividnext.sodalive.content.translation.ContentTranslationRepository
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService.Companion.getTranslatableLanguageCodes
 import org.springframework.context.event.EventListener
-import org.springframework.data.repository.findByIdOrNull
 import org.springframework.scheduling.annotation.Async
 import org.springframework.stereotype.Component
 import org.springframework.transaction.annotation.Propagation
@@ -58,24 +29,7 @@ class LanguageTranslationEvent(
 
 @Component
 class LanguageTranslationListener(
-    private val audioContentRepository: AudioContentRepository,
-    private val chatCharacterRepository: ChatCharacterRepository,
-    private val audioContentThemeRepository: AudioContentThemeQueryRepository,
-    private val seriesRepository: AdminContentSeriesRepository,
-    private val seriesGenreRepository: AdminContentSeriesGenreRepository,
-    private val originalWorkRepository: OriginalWorkRepository,
-
-    private val contentTranslationRepository: ContentTranslationRepository,
-    private val aiCharacterTranslationRepository: AiCharacterTranslationRepository,
-    private val contentThemeTranslationRepository: ContentThemeTranslationRepository,
-    private val seriesTranslationRepository: SeriesTranslationRepository,
-    private val seriesGenreTranslationRepository: SeriesGenreTranslationRepository,
-    private val originalWorkTranslationRepository: OriginalWorkTranslationRepository,
-
-    private val categoryRepository: CategoryRepository,
-    private val categoryTranslationRepository: CategoryTranslationRepository,
-
-    private val translationService: PapagoTranslationService
+    private val resourceTranslationJobScheduler: ResourceTranslationJobScheduler
 ) {
     @Async
     @EventListener(condition = "!#event.waitTransactionCommit")
@@ -92,424 +46,6 @@ class LanguageTranslationListener(
     }
 
     private fun processTranslation(event: LanguageTranslationEvent) {
-        when (event.targetType) {
-            LanguageTranslationTargetType.CONTENT -> handleContentLanguageTranslation(event)
-            LanguageTranslationTargetType.CHARACTER -> handleCharacterLanguageTranslation(event)
-            LanguageTranslationTargetType.CONTENT_THEME -> handleContentThemeLanguageTranslation(event)
-            LanguageTranslationTargetType.SERIES -> handleSeriesLanguageTranslation(event)
-            LanguageTranslationTargetType.SERIES_GENRE -> handleSeriesGenreLanguageTranslation(event)
-            LanguageTranslationTargetType.ORIGINAL_WORK -> handleOriginalWorkLanguageTranslation(event)
-            LanguageTranslationTargetType.CREATOR_CONTENT_CATEGORY -> handleCreatorContentCategoryLanguageTranslation(event)
-        }
-    }
-
-    private fun handleContentLanguageTranslation(event: LanguageTranslationEvent) {
-        val audioContent = audioContentRepository.findByIdOrNull(event.id) ?: return
-        val languageCode = audioContent.languageCode ?: return
-
-        getTranslatableLanguageCodes(languageCode).forEach { locale ->
-            val tags = audioContent.audioContentHashTags
-                .mapNotNull { it.hashTag?.tag }
-                .joinToString(",")
-
-            val texts = mutableListOf<String>()
-            texts.add(audioContent.title)
-            texts.add(audioContent.detail)
-            texts.add(tags)
-
-            val sourceLanguage = audioContent.languageCode ?: "ko"
-
-            val response = translationService.translate(
-                request = TranslateRequest(
-                    texts = texts,
-                    sourceLanguage = sourceLanguage,
-                    targetLanguage = locale
-                )
-            )
-
-            val translatedTexts = response.translatedText
-            if (translatedTexts.size == texts.size) {
-                var index = 0
-
-                val translatedTitle = translatedTexts[index++]
-                val translatedDetail = translatedTexts[index++]
-                val translatedTags = translatedTexts[index]
-
-                val payload = ContentTranslationPayload(
-                    title = translatedTitle,
-                    detail = translatedDetail,
-                    tags = translatedTags
-                )
-
-                val existing = contentTranslationRepository
-                    .findByContentIdAndLocale(audioContent.id!!, locale)
-
-                if (existing == null) {
-                    contentTranslationRepository.save(
-                        ContentTranslation(
-                            contentId = audioContent.id!!,
-                            locale = locale,
-                            renderedPayload = payload
-                        )
-                    )
-                } else {
-                    existing.renderedPayload = payload
-                    contentTranslationRepository.save(existing)
-                }
-            }
-        }
-    }
-
-    private fun handleCharacterLanguageTranslation(event: LanguageTranslationEvent) {
-        val character = chatCharacterRepository.findByIdOrNull(event.id) ?: return
-        val languageCode = character.languageCode ?: return
-
-        getTranslatableLanguageCodes(languageCode).forEach { locale ->
-            val personality = character.personalities.firstOrNull()
-            val background = character.backgrounds.firstOrNull()
-
-            val tags = character.tagMappings.joinToString(",") { it.tag.tag }
-
-            val texts = mutableListOf<String>()
-            texts.add(character.name)
-            texts.add(character.description)
-            texts.add(character.gender ?: "")
-
-            val hasPersonality = personality != null
-            if (hasPersonality) {
-                texts.add(personality!!.trait)
-                texts.add(personality.description)
-            }
-
-            val hasBackground = background != null
-            if (hasBackground) {
-                texts.add(background!!.topic)
-                texts.add(background.description)
-            }
-
-            texts.add(tags)
-
-            val sourceLanguage = character.languageCode ?: "ko"
-
-            val response = translationService.translate(
-                request = TranslateRequest(
-                    texts = texts,
-                    sourceLanguage = sourceLanguage,
-                    targetLanguage = locale
-                )
-            )
-
-            val translatedTexts = response.translatedText
-            if (translatedTexts.size == texts.size) {
-                var index = 0
-
-                val translatedName = translatedTexts[index++]
-                val translatedDescription = translatedTexts[index++]
-                val translatedGender = translatedTexts[index++]
-
-                var translatedPersonality: TranslatedAiCharacterPersonality? = null
-                if (hasPersonality) {
-                    translatedPersonality = TranslatedAiCharacterPersonality(
-                        trait = translatedTexts[index++],
-                        description = translatedTexts[index++]
-                    )
-                }
-
-                var translatedBackground: TranslatedAiCharacterBackground? = null
-                if (hasBackground) {
-                    translatedBackground = TranslatedAiCharacterBackground(
-                        topic = translatedTexts[index++],
-                        description = translatedTexts[index++]
-                    )
-                }
-
-                val translatedTags = translatedTexts[index]
-
-                val payload = AiCharacterTranslationRenderedPayload(
-                    name = translatedName,
-                    description = translatedDescription,
-                    gender = translatedGender,
-                    personalityTrait = translatedPersonality?.trait ?: "",
-                    personalityDescription = translatedPersonality?.description ?: "",
-                    backgroundTopic = translatedBackground?.topic ?: "",
-                    backgroundDescription = translatedBackground?.description ?: "",
-                    tags = translatedTags
-                )
-
-                val existing = aiCharacterTranslationRepository
-                    .findByCharacterIdAndLocale(character.id!!, locale)
-
-                if (existing == null) {
-                    val entity = AiCharacterTranslation(
-                        characterId = character.id!!,
-                        locale = locale,
-                        renderedPayload = payload
-                    )
-
-                    aiCharacterTranslationRepository.save(entity)
-                } else {
-                    existing.renderedPayload = payload
-                    aiCharacterTranslationRepository.save(existing)
-                }
-            }
-        }
-    }
-
-    private fun handleContentThemeLanguageTranslation(event: LanguageTranslationEvent) {
-        val contentTheme = audioContentThemeRepository.findThemeByIdAndActive(event.id) ?: return
-
-        val sourceLanguage = "ko"
-        getTranslatableLanguageCodes(sourceLanguage).forEach { locale ->
-            val texts = mutableListOf<String>()
-            texts.add(contentTheme.theme)
-
-            val response = translationService.translate(
-                request = TranslateRequest(
-                    texts = texts,
-                    sourceLanguage = sourceLanguage,
-                    targetLanguage = locale
-                )
-            )
-
-            val translatedTexts = response.translatedText
-            if (translatedTexts.size == texts.size) {
-                val translatedTheme = translatedTexts[0]
-
-                val existing = contentThemeTranslationRepository
-                    .findByContentThemeIdAndLocale(contentTheme.id!!, locale)
-
-                if (existing == null) {
-                    contentThemeTranslationRepository.save(
-                        ContentThemeTranslation(
-                            contentThemeId = contentTheme.id!!,
-                            locale = locale,
-                            theme = translatedTheme
-                        )
-                    )
-                } else {
-                    existing.theme = translatedTheme
-                    contentThemeTranslationRepository.save(existing)
-                }
-            }
-        }
-    }
-
-    private fun handleSeriesLanguageTranslation(event: LanguageTranslationEvent) {
-        val series = seriesRepository.findByIdOrNull(event.id) ?: return
-        val languageCode = series.languageCode ?: return
-
-        getTranslatableLanguageCodes(languageCode).forEach { locale ->
-            val keywords = series.keywordList
-                .mapNotNull { it.keyword?.tag }
-                .joinToString(", ")
-            val texts = mutableListOf<String>()
-            texts.add(series.title)
-            texts.add(series.introduction)
-            texts.add(keywords)
-
-            val sourceLanguage = series.languageCode ?: "ko"
-
-            val response = translationService.translate(
-                request = TranslateRequest(
-                    texts = texts,
-                    sourceLanguage = sourceLanguage,
-                    targetLanguage = locale
-                )
-            )
-
-            val translatedTexts = response.translatedText
-            if (translatedTexts.size == texts.size) {
-                var index = 0
-                val translatedTitle = translatedTexts[index++]
-                val translatedIntroduction = translatedTexts[index++]
-                val translatedKeywordsJoined = translatedTexts[index]
-
-                val translatedKeywords = translatedKeywordsJoined
-                    .split(",")
-                    .map { it.trim() }
-                    .filter { it.isNotBlank() }
-
-                val payload = SeriesTranslationPayload(
-                    title = translatedTitle,
-                    introduction = translatedIntroduction,
-                    keywords = translatedKeywords
-                )
-
-                val existing = seriesTranslationRepository
-                    .findBySeriesIdAndLocale(series.id!!, locale)
-
-                if (existing == null) {
-                    seriesTranslationRepository.save(
-                        SeriesTranslation(
-                            seriesId = series.id!!,
-                            locale = locale,
-                            renderedPayload = payload
-                        )
-                    )
-                } else {
-                    existing.renderedPayload = payload
-                    seriesTranslationRepository.save(existing)
-                }
-            }
-        }
-    }
-
-    private fun handleSeriesGenreLanguageTranslation(event: LanguageTranslationEvent) {
-        val seriesGenre = seriesGenreRepository.findActiveSeriesGenreById(event.id) ?: return
-
-        val sourceLanguage = "ko"
-        getTranslatableLanguageCodes(sourceLanguage).forEach { locale ->
-            val texts = mutableListOf<String>()
-            texts.add(seriesGenre.genre)
-
-            val response = translationService.translate(
-                request = TranslateRequest(
-                    texts = texts,
-                    sourceLanguage = sourceLanguage,
-                    targetLanguage = locale
-                )
-            )
-
-            val translatedTexts = response.translatedText
-            if (translatedTexts.size == texts.size) {
-                val translatedGenre = translatedTexts[0]
-
-                val existing = seriesGenreTranslationRepository
-                    .findBySeriesGenreIdAndLocale(seriesGenre.id!!, locale)
-
-                if (existing == null) {
-                    seriesGenreTranslationRepository.save(
-                        SeriesGenreTranslation(
-                            seriesGenreId = seriesGenre.id!!,
-                            locale = locale,
-                            genre = translatedGenre
-                        )
-                    )
-                } else {
-                    existing.genre = translatedGenre
-                    seriesGenreTranslationRepository.save(existing)
-                }
-            }
-        }
-    }
-
-    private fun handleOriginalWorkLanguageTranslation(event: LanguageTranslationEvent) {
-        val originalWork = originalWorkRepository.findByIdOrNull(event.id) ?: return
-        val languageCode = originalWork.languageCode ?: return
-
-        /**
-         * handleSeriesLanguageTranslation 참조하여 원작 번역 구현
-         *
-         * originalWorkTranslationRepository
-         *
-         * 번역대상
-         * - title
-         * - contentType
-         * - category
-         * - description
-         * - tags
-         */
-        getTranslatableLanguageCodes(languageCode).forEach { locale ->
-            val tagsJoined = originalWork.tagMappings
-                .mapNotNull { it.tag.tag }
-                .joinToString(", ")
-
-            val texts = mutableListOf<String>()
-            texts.add(originalWork.title)
-            texts.add(originalWork.contentType)
-            texts.add(originalWork.category)
-            texts.add(originalWork.description)
-            texts.add(tagsJoined)
-
-            val sourceLanguage = originalWork.languageCode ?: "ko"
-
-            val response = translationService.translate(
-                request = TranslateRequest(
-                    texts = texts,
-                    sourceLanguage = sourceLanguage,
-                    targetLanguage = locale
-                )
-            )
-
-            val translatedTexts = response.translatedText
-            if (translatedTexts.size == texts.size) {
-                var index = 0
-                val translatedTitle = translatedTexts[index++]
-                val translatedContentType = translatedTexts[index++]
-                val translatedCategory = translatedTexts[index++]
-                val translatedDescription = translatedTexts[index++]
-                val translatedTagsJoined = translatedTexts[index]
-
-                val translatedTags = translatedTagsJoined
-                    .split(",")
-                    .map { it.trim() }
-                    .filter { it.isNotBlank() }
-
-                val payload = OriginalWorkTranslationPayload(
-                    title = translatedTitle,
-                    contentType = translatedContentType,
-                    category = translatedCategory,
-                    description = translatedDescription,
-                    tags = translatedTags
-                )
-
-                val existing = originalWorkTranslationRepository
-                    .findByOriginalWorkIdAndLocale(originalWork.id!!, locale)
-
-                if (existing == null) {
-                    originalWorkTranslationRepository.save(
-                        OriginalWorkTranslation(
-                            originalWorkId = originalWork.id!!,
-                            locale = locale,
-                            renderedPayload = payload
-                        )
-                    )
-                } else {
-                    existing.renderedPayload = payload
-                    originalWorkTranslationRepository.save(existing)
-                }
-            }
-        }
-    }
-
-    private fun handleCreatorContentCategoryLanguageTranslation(event: LanguageTranslationEvent) {
-        val category = categoryRepository.findByIdOrNull(event.id)
-
-        if (category == null || !category.isActive || category.languageCode.isNullOrBlank()) return
-
-        val sourceLanguage = category.languageCode ?: "ko"
-        getTranslatableLanguageCodes(sourceLanguage).forEach { locale ->
-            val texts = mutableListOf<String>()
-            texts.add(category.title)
-
-            val response = translationService.translate(
-                request = TranslateRequest(
-                    texts = texts,
-                    sourceLanguage = sourceLanguage,
-                    targetLanguage = locale
-                )
-            )
-
-            val translatedTexts = response.translatedText
-            if (translatedTexts.size == texts.size) {
-                val translatedCategory = translatedTexts[0]
-
-                val existing = categoryTranslationRepository
-                    .findByCategoryIdAndLocale(category.id!!, locale)
-
-                if (existing == null) {
-                    categoryTranslationRepository.save(
-                        CategoryTranslation(
-                            categoryId = category.id!!,
-                            locale = locale,
-                            category = translatedCategory
-                        )
-                    )
-                } else {
-                    existing.category = translatedCategory
-                    categoryTranslationRepository.save(existing)
-                }
-            }
-        }
+        resourceTranslationJobScheduler.scheduleResourceTranslations(event.targetType, event.id)
     }
 }

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/PapagoTranslationService.kt

@@ -15,12 +15,16 @@ class PapagoTranslationService(
 
     @Value("\${cloud.naver.papago-client-secret}")
     private val papagoClientSecret: String
-) {
+) : TranslationProvider {
     private val restTemplate: RestTemplate = RestTemplate()
 
     private val papagoTranslateUrl = "https://papago.apigw.ntruss.com/nmt/v1/translation"
 
-    fun translate(request: TranslateRequest): TranslateResult {
+    override val providerName: String = "papago"
+
+    override val providerVersion: String = "nmt-v1"
+
+    override fun translate(request: TranslateRequest): TranslateResult {
         if (request.texts.isEmpty() || request.sourceLanguage == request.targetLanguage) {
             return TranslateResult(emptyList())
         }

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/ResourceTranslationJobScheduler.kt

@@ -0,0 +1,39 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService.Companion.getTranslatableLanguageCodes
+import org.springframework.stereotype.Service
+
+@Service
+class ResourceTranslationJobScheduler(
+    private val sourceExtractor: TranslationSourceExtractor,
+    private val translationJobScheduler: TranslationJobScheduler
+) {
+    fun scheduleResourceTranslations(resourceType: LanguageTranslationTargetType, resourceId: Long) {
+        val source = sourceExtractor.extract(resourceType, resourceId) ?: return
+        getTranslatableLanguageCodes(source.sourceLanguage).forEach { targetLanguage ->
+            scheduleSource(source, targetLanguage)
+        }
+    }
+
+    fun scheduleResourceTranslation(
+        resourceType: LanguageTranslationTargetType,
+        resourceId: Long,
+        targetLanguage: String
+    ) {
+        val source = sourceExtractor.extract(resourceType, resourceId) ?: return
+        scheduleSource(source, targetLanguage)
+    }
+
+    private fun scheduleSource(source: TranslationSource, targetLanguage: String) {
+        source.fields.forEach { field ->
+            translationJobScheduler.scheduleMissingTranslation(
+                resourceType = source.resourceType,
+                resourceId = source.resourceId,
+                fieldKey = field.fieldKey,
+                sourceText = field.sourceText,
+                sourceLanguage = source.sourceLanguage,
+                targetLanguage = targetLanguage
+            )
+        }
+    }
+}

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/SourceTextNormalizer.kt

@@ -0,0 +1,23 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import java.security.MessageDigest
+import java.text.Normalizer
+
+object SourceTextNormalizer {
+    const val NORMALIZATION_VERSION = "v1"
+
+    private val whitespaceRegex = Regex("\\s+")
+
+    fun normalize(sourceText: String): String {
+        return Normalizer.normalize(sourceText, Normalizer.Form.NFC)
+            .replace(whitespaceRegex, " ")
+            .trim()
+    }
+
+    fun hash(sourceText: String): String {
+        val normalized = normalize(sourceText)
+        val digest = MessageDigest.getInstance("SHA-256")
+            .digest(normalized.toByteArray(Charsets.UTF_8))
+        return digest.joinToString("") { "%02x".format(it) }
+    }
+}

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationJob.kt

@@ -0,0 +1,64 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import kr.co.vividnext.sodalive.common.BaseEntity
+import java.time.LocalDateTime
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.EnumType
+import javax.persistence.Enumerated
+import javax.persistence.Table
+import javax.persistence.UniqueConstraint
+
+enum class TranslationJobStatus {
+    PENDING,
+    RUNNING,
+    COMPLETED,
+    FAILED
+}
+
+@Entity
+@Table(
+    name = "translation_job",
+    uniqueConstraints = [
+        UniqueConstraint(
+            name = "uk_translation_job_resource_field_target_hash",
+            columnNames = ["resource_type", "resource_id", "field_key", "target_language", "source_hash"]
+        )
+    ]
+)
+class TranslationJob(
+    @Enumerated(EnumType.STRING)
+    @Column(name = "resource_type", nullable = false, length = 50)
+    val resourceType: LanguageTranslationTargetType,
+
+    @Column(name = "resource_id", nullable = false)
+    val resourceId: Long,
+
+    @Column(name = "field_key", nullable = false, length = 80)
+    val fieldKey: String,
+
+    @Column(name = "source_hash", nullable = false, length = 64)
+    val sourceHash: String,
+
+    @Column(name = "source_text", nullable = false, columnDefinition = "text")
+    val sourceText: String,
+
+    @Column(name = "source_language", nullable = false, length = 10)
+    val sourceLanguage: String,
+
+    @Column(name = "target_language", nullable = false, length = 10)
+    val targetLanguage: String,
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "status", nullable = false, length = 20)
+    var status: TranslationJobStatus = TranslationJobStatus.PENDING,
+
+    @Column(name = "retry_count", nullable = false)
+    var retryCount: Int = 0,
+
+    @Column(name = "last_error_message", columnDefinition = "text")
+    var lastErrorMessage: String? = null,
+
+    @Column(name = "next_retry_at", nullable = false)
+    var nextRetryAt: LocalDateTime = LocalDateTime.now()
+) : BaseEntity()

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationJobRepository.kt

@@ -0,0 +1,58 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Query
+import org.springframework.data.repository.query.Param
+import java.time.LocalDateTime
+
+interface TranslationJobRepository : JpaRepository<TranslationJob, Long> {
+    fun findByResourceTypeAndResourceIdAndFieldKeyAndTargetLanguageAndSourceHash(
+        resourceType: LanguageTranslationTargetType,
+        resourceId: Long,
+        fieldKey: String,
+        targetLanguage: String,
+        sourceHash: String
+    ): TranslationJob?
+
+    @Query(
+        """
+        select j from TranslationJob j
+        where j.resourceType = :resourceType
+          and j.resourceId = :resourceId
+          and j.fieldKey = :fieldKey
+          and j.targetLanguage = :targetLanguage
+          and j.sourceHash = :sourceHash
+          and j.status in (kr.co.vividnext.sodalive.i18n.translation.TranslationJobStatus.PENDING, kr.co.vividnext.sodalive.i18n.translation.TranslationJobStatus.RUNNING)
+        """
+    )
+    fun findActiveJob(
+        @Param("resourceType") resourceType: LanguageTranslationTargetType,
+        @Param("resourceId") resourceId: Long,
+        @Param("fieldKey") fieldKey: String,
+        @Param("targetLanguage") targetLanguage: String,
+        @Param("sourceHash") sourceHash: String
+    ): TranslationJob?
+
+    @Query(
+        value = """
+        select j.id
+        from translation_job j
+        join (
+            select resource_type, resource_id, target_language
+            from translation_job
+            where status = 'PENDING'
+              and next_retry_at <= :now
+            order by created_at asc
+            limit 1
+        ) g on j.resource_type = g.resource_type
+           and j.resource_id = g.resource_id
+           and j.target_language = g.target_language
+        where j.status = 'PENDING'
+          and j.next_retry_at <= :now
+        order by j.created_at asc
+        for update skip locked
+        """,
+        nativeQuery = true
+    )
+    fun findNextPendingGroupJobIdsForUpdate(@Param("now") now: LocalDateTime): List<Long>
+}

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationJobScheduler.kt

@@ -0,0 +1,50 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+import java.time.LocalDateTime
+
+@Service
+class TranslationJobScheduler(
+    private val translationJobRepository: TranslationJobRepository
+) {
+    @Transactional
+    fun scheduleMissingTranslation(
+        resourceType: LanguageTranslationTargetType,
+        resourceId: Long,
+        fieldKey: String,
+        sourceText: String,
+        sourceLanguage: String,
+        targetLanguage: String
+    ) {
+        val normalizedText = SourceTextNormalizer.normalize(sourceText)
+        if (normalizedText.isBlank()) return
+
+        val normalizedSourceLanguage = sourceLanguage.lowercase()
+        val normalizedTargetLanguage = targetLanguage.lowercase()
+        if (normalizedSourceLanguage == normalizedTargetLanguage) return
+
+        val sourceHash = SourceTextNormalizer.hash(normalizedText)
+        val existingJob = translationJobRepository.findByResourceTypeAndResourceIdAndFieldKeyAndTargetLanguageAndSourceHash(
+            resourceType = resourceType,
+            resourceId = resourceId,
+            fieldKey = fieldKey,
+            targetLanguage = normalizedTargetLanguage,
+            sourceHash = sourceHash
+        )
+        if (existingJob != null) return
+
+        translationJobRepository.save(
+            TranslationJob(
+                resourceType = resourceType,
+                resourceId = resourceId,
+                fieldKey = fieldKey,
+                sourceHash = sourceHash,
+                sourceText = normalizedText,
+                sourceLanguage = normalizedSourceLanguage,
+                targetLanguage = normalizedTargetLanguage,
+                nextRetryAt = LocalDateTime.now()
+            )
+        )
+    }
+}

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationJobWorker.kt

@@ -0,0 +1,155 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import org.slf4j.LoggerFactory
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.support.TransactionTemplate
+import java.time.LocalDateTime
+
+@Component
+class TranslationJobWorker(
+    private val translationJobRepository: TranslationJobRepository,
+    private val translationMemoryRepository: TranslationMemoryRepository,
+    private val translationProvider: TranslationProvider,
+    private val materializer: TranslationReadModelMaterializer,
+    transactionManager: PlatformTransactionManager
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+    private val transactionTemplate = TransactionTemplate(transactionManager)
+
+    @Scheduled(fixedDelayString = "\${sodalive.translation-job.fixed-delay-ms:600000}")
+    fun runPendingJobs() {
+        repeat(MAX_GROUPS_PER_TICK) {
+            if (!processNextGroup()) return
+        }
+    }
+
+    fun processNextGroup(): Boolean {
+        val jobs = claimNextGroup()
+        if (jobs.isEmpty()) return false
+
+        val firstJob = jobs.first()
+        val succeededJobs = mutableListOf<TranslationJob>()
+        val failedJobs = mutableListOf<Pair<TranslationJob, Exception>>()
+        jobs.forEach { job ->
+            try {
+                ensureMemory(job)
+                succeededJobs.add(job)
+            } catch (ex: Exception) {
+                failedJobs.add(job to ex)
+            }
+        }
+
+        if (failedJobs.isNotEmpty()) {
+            succeededJobs.forEach { completeJob(it.id!!) }
+            failedJobs.forEach { (job, ex) -> failJob(job.id!!, ex) }
+            return true
+        }
+
+        try {
+            materializer.materialize(firstJob.resourceType, firstJob.resourceId, firstJob.targetLanguage)
+            succeededJobs.forEach { completeJob(it.id!!) }
+        } catch (ex: Exception) {
+            succeededJobs.forEach { failJob(it.id!!, ex) }
+        }
+        return true
+    }
+
+    private fun claimNextGroup(): List<TranslationJob> {
+        return transactionTemplate.execute {
+            val jobIds = translationJobRepository.findNextPendingGroupJobIdsForUpdate(LocalDateTime.now())
+            jobIds.mapNotNull { jobId ->
+                val job = translationJobRepository.findById(jobId).orElse(null) ?: return@mapNotNull null
+                job.status = TranslationJobStatus.RUNNING
+                translationJobRepository.save(job)
+                job
+            }
+        }.orEmpty()
+    }
+
+    private fun ensureMemory(job: TranslationJob) {
+        val existing = translationMemoryRepository
+            .findBySourceHashAndSourceLanguageAndTargetLanguageAndProviderAndNormalizationVersion(
+                sourceHash = job.sourceHash,
+                sourceLanguage = job.sourceLanguage,
+                targetLanguage = job.targetLanguage,
+                provider = translationProvider.providerName,
+                normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+            )
+        if (existing != null) return
+
+        val response = translationProvider.translate(
+            TranslateRequest(
+                texts = listOf(job.sourceText),
+                sourceLanguage = job.sourceLanguage,
+                targetLanguage = job.targetLanguage
+            )
+        )
+        val translated = response.translatedText.firstOrNull()?.takeIf { it.isNotBlank() }
+            ?: throw IllegalStateException("empty translation result")
+
+        transactionTemplate.executeWithoutResult {
+            val memory = translationMemoryRepository
+                .findBySourceHashAndSourceLanguageAndTargetLanguageAndProviderAndNormalizationVersion(
+                    sourceHash = job.sourceHash,
+                    sourceLanguage = job.sourceLanguage,
+                    targetLanguage = job.targetLanguage,
+                    provider = translationProvider.providerName,
+                    normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+                )
+            if (memory == null) {
+                translationMemoryRepository.save(
+                    TranslationMemory(
+                        sourceHash = job.sourceHash,
+                        sourceText = job.sourceText,
+                        sourceLanguage = job.sourceLanguage,
+                        targetLanguage = job.targetLanguage,
+                        translatedText = translated,
+                        provider = translationProvider.providerName,
+                        providerVersion = translationProvider.providerVersion
+                    )
+                )
+            }
+        }
+    }
+
+    private fun completeJob(jobId: Long) {
+        transactionTemplate.executeWithoutResult {
+            val job = translationJobRepository.findById(jobId).orElse(null) ?: return@executeWithoutResult
+            job.status = TranslationJobStatus.COMPLETED
+            job.lastErrorMessage = null
+            translationJobRepository.save(job)
+        }
+    }
+
+    private fun failJob(jobId: Long, ex: Exception) {
+        log.warn("Failed to process translation job. jobId={}, error={}", jobId, ex.message)
+        transactionTemplate.executeWithoutResult {
+            val job = translationJobRepository.findById(jobId).orElse(null) ?: return@executeWithoutResult
+            job.retryCount += 1
+            job.lastErrorMessage = ex.message?.take(MAX_ERROR_LENGTH)
+            if (job.retryCount >= MAX_RETRY_COUNT) {
+                job.status = TranslationJobStatus.FAILED
+            } else {
+                job.status = TranslationJobStatus.PENDING
+                job.nextRetryAt = LocalDateTime.now().plusMinutes(backoffMinutes(job.retryCount))
+            }
+            translationJobRepository.save(job)
+        }
+    }
+
+    private fun backoffMinutes(retryCount: Int): Long {
+        return when (retryCount) {
+            1 -> 1L
+            2 -> 5L
+            else -> 15L
+        }
+    }
+
+    companion object {
+        private const val MAX_GROUPS_PER_TICK = 5
+        private const val MAX_ERROR_LENGTH = 1000
+        private const val MAX_RETRY_COUNT = 3
+    }
+}

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationMemory.kt

@@ -0,0 +1,43 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import kr.co.vividnext.sodalive.common.BaseEntity
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.Table
+import javax.persistence.UniqueConstraint
+
+@Entity
+@Table(
+    name = "translation_memory",
+    uniqueConstraints = [
+        UniqueConstraint(
+            name = "uk_translation_memory_source_target_provider",
+            columnNames = ["source_hash", "source_language", "target_language", "provider", "normalization_version"]
+        )
+    ]
+)
+class TranslationMemory(
+    @Column(name = "source_hash", nullable = false, length = 64)
+    val sourceHash: String,
+
+    @Column(name = "source_text", nullable = false, columnDefinition = "text")
+    val sourceText: String,
+
+    @Column(name = "source_language", nullable = false, length = 10)
+    val sourceLanguage: String,
+
+    @Column(name = "target_language", nullable = false, length = 10)
+    val targetLanguage: String,
+
+    @Column(name = "translated_text", nullable = false, columnDefinition = "text")
+    val translatedText: String,
+
+    @Column(name = "provider", nullable = false, length = 50)
+    val provider: String,
+
+    @Column(name = "provider_version", nullable = false, length = 50)
+    val providerVersion: String,
+
+    @Column(name = "normalization_version", nullable = false, length = 20)
+    val normalizationVersion: String = SourceTextNormalizer.NORMALIZATION_VERSION
+) : BaseEntity()

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationMemoryRepository.kt

@@ -0,0 +1,13 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import org.springframework.data.jpa.repository.JpaRepository
+
+interface TranslationMemoryRepository : JpaRepository<TranslationMemory, Long> {
+    fun findBySourceHashAndSourceLanguageAndTargetLanguageAndProviderAndNormalizationVersion(
+        sourceHash: String,
+        sourceLanguage: String,
+        targetLanguage: String,
+        provider: String,
+        normalizationVersion: String
+    ): TranslationMemory?
+}

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationProvider.kt

@@ -0,0 +1,8 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+interface TranslationProvider {
+    val providerName: String
+    val providerVersion: String
+
+    fun translate(request: TranslateRequest): TranslateResult
+}

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationReadModelMaterializer.kt

@@ -0,0 +1,186 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslation
+import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslationRenderedPayload
+import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslationRepository
+import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslation
+import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslationPayload
+import kr.co.vividnext.sodalive.chat.original.translation.OriginalWorkTranslationRepository
+import kr.co.vividnext.sodalive.content.category.CategoryTranslation
+import kr.co.vividnext.sodalive.content.category.CategoryTranslationRepository
+import kr.co.vividnext.sodalive.content.series.translation.SeriesGenreTranslation
+import kr.co.vividnext.sodalive.content.series.translation.SeriesGenreTranslationRepository
+import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslation
+import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslationPayload
+import kr.co.vividnext.sodalive.content.series.translation.SeriesTranslationRepository
+import kr.co.vividnext.sodalive.content.theme.translation.ContentThemeTranslation
+import kr.co.vividnext.sodalive.content.theme.translation.ContentThemeTranslationRepository
+import kr.co.vividnext.sodalive.content.translation.ContentTranslation
+import kr.co.vividnext.sodalive.content.translation.ContentTranslationPayload
+import kr.co.vividnext.sodalive.content.translation.ContentTranslationRepository
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+
+@Service
+class TranslationReadModelMaterializer(
+    private val sourceExtractor: TranslationSourceExtractor,
+    private val translationMemoryRepository: TranslationMemoryRepository,
+    private val contentTranslationRepository: ContentTranslationRepository,
+    private val aiCharacterTranslationRepository: AiCharacterTranslationRepository,
+    private val contentThemeTranslationRepository: ContentThemeTranslationRepository,
+    private val seriesTranslationRepository: SeriesTranslationRepository,
+    private val seriesGenreTranslationRepository: SeriesGenreTranslationRepository,
+    private val originalWorkTranslationRepository: OriginalWorkTranslationRepository,
+    private val categoryTranslationRepository: CategoryTranslationRepository
+) {
+    @Transactional
+    fun materialize(resourceType: LanguageTranslationTargetType, resourceId: Long, targetLanguage: String): Boolean {
+        val source = sourceExtractor.extract(resourceType, resourceId) ?: return false
+        val translations = resolveTranslatedFields(source, targetLanguage.lowercase()) ?: return false
+
+        when (resourceType) {
+            LanguageTranslationTargetType.CONTENT -> upsertContent(resourceId, targetLanguage, translations)
+            LanguageTranslationTargetType.CHARACTER -> upsertCharacter(resourceId, targetLanguage, translations)
+            LanguageTranslationTargetType.CONTENT_THEME -> upsertContentTheme(resourceId, targetLanguage, translations)
+            LanguageTranslationTargetType.SERIES -> upsertSeries(resourceId, targetLanguage, translations)
+            LanguageTranslationTargetType.SERIES_GENRE -> upsertSeriesGenre(resourceId, targetLanguage, translations)
+            LanguageTranslationTargetType.ORIGINAL_WORK -> upsertOriginalWork(resourceId, targetLanguage, translations)
+            LanguageTranslationTargetType.CREATOR_CONTENT_CATEGORY -> upsertCategory(resourceId, targetLanguage, translations)
+        }
+
+        return true
+    }
+
+    private fun resolveTranslatedFields(source: TranslationSource, targetLanguage: String): Map<String, String>? {
+        val result = mutableMapOf<String, String>()
+        source.fields.forEach { field ->
+            val normalizedText = SourceTextNormalizer.normalize(field.sourceText)
+            if (normalizedText.isBlank()) {
+                result[field.fieldKey] = ""
+                return@forEach
+            }
+
+            val memory = translationMemoryRepository
+                .findBySourceHashAndSourceLanguageAndTargetLanguageAndProviderAndNormalizationVersion(
+                    sourceHash = SourceTextNormalizer.hash(normalizedText),
+                    sourceLanguage = source.sourceLanguage.lowercase(),
+                    targetLanguage = targetLanguage,
+                    provider = DEFAULT_PROVIDER,
+                    normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+                ) ?: return null
+
+            result[field.fieldKey] = memory.translatedText
+        }
+        return result
+    }
+
+    private fun upsertContent(resourceId: Long, targetLanguage: String, translations: Map<String, String>) {
+        val payload = ContentTranslationPayload(
+            title = translations["title"].orEmpty(),
+            detail = translations["detail"].orEmpty(),
+            tags = translations["tags"].orEmpty()
+        )
+        val existing = contentTranslationRepository.findByContentIdAndLocale(resourceId, targetLanguage)
+        if (existing == null) {
+            contentTranslationRepository.save(ContentTranslation(resourceId, targetLanguage, payload))
+        } else {
+            existing.renderedPayload = payload
+            contentTranslationRepository.save(existing)
+        }
+    }
+
+    private fun upsertCharacter(resourceId: Long, targetLanguage: String, translations: Map<String, String>) {
+        val payload = AiCharacterTranslationRenderedPayload(
+            name = translations["name"].orEmpty(),
+            description = translations["description"].orEmpty(),
+            gender = translations["gender"].orEmpty(),
+            personalityTrait = translations["personalityTrait"].orEmpty(),
+            personalityDescription = translations["personalityDescription"].orEmpty(),
+            backgroundTopic = translations["backgroundTopic"].orEmpty(),
+            backgroundDescription = translations["backgroundDescription"].orEmpty(),
+            tags = translations["tags"].orEmpty()
+        )
+        val existing = aiCharacterTranslationRepository.findByCharacterIdAndLocale(resourceId, targetLanguage)
+        if (existing == null) {
+            aiCharacterTranslationRepository.save(AiCharacterTranslation(resourceId, targetLanguage, payload))
+        } else {
+            existing.renderedPayload = payload
+            aiCharacterTranslationRepository.save(existing)
+        }
+    }
+
+    private fun upsertContentTheme(resourceId: Long, targetLanguage: String, translations: Map<String, String>) {
+        val theme = translations["theme"].orEmpty()
+        val existing = contentThemeTranslationRepository.findByContentThemeIdAndLocale(resourceId, targetLanguage)
+        if (existing == null) {
+            contentThemeTranslationRepository.save(ContentThemeTranslation(resourceId, targetLanguage, theme))
+        } else {
+            existing.theme = theme
+            contentThemeTranslationRepository.save(existing)
+        }
+    }
+
+    private fun upsertSeries(resourceId: Long, targetLanguage: String, translations: Map<String, String>) {
+        val payload = SeriesTranslationPayload(
+            title = translations["title"].orEmpty(),
+            introduction = translations["introduction"].orEmpty(),
+            keywords = translations["keywords"].orEmpty()
+                .split(",")
+                .map { it.trim() }
+                .filter { it.isNotBlank() }
+        )
+        val existing = seriesTranslationRepository.findBySeriesIdAndLocale(resourceId, targetLanguage)
+        if (existing == null) {
+            seriesTranslationRepository.save(SeriesTranslation(resourceId, targetLanguage, payload))
+        } else {
+            existing.renderedPayload = payload
+            seriesTranslationRepository.save(existing)
+        }
+    }
+
+    private fun upsertSeriesGenre(resourceId: Long, targetLanguage: String, translations: Map<String, String>) {
+        val genre = translations["genre"].orEmpty()
+        val existing = seriesGenreTranslationRepository.findBySeriesGenreIdAndLocale(resourceId, targetLanguage)
+        if (existing == null) {
+            seriesGenreTranslationRepository.save(SeriesGenreTranslation(resourceId, targetLanguage, genre))
+        } else {
+            existing.genre = genre
+            seriesGenreTranslationRepository.save(existing)
+        }
+    }
+
+    private fun upsertOriginalWork(resourceId: Long, targetLanguage: String, translations: Map<String, String>) {
+        val payload = OriginalWorkTranslationPayload(
+            title = translations["title"].orEmpty(),
+            contentType = translations["contentType"].orEmpty(),
+            category = translations["category"].orEmpty(),
+            description = translations["description"].orEmpty(),
+            tags = translations["tags"].orEmpty()
+                .split(",")
+                .map { it.trim() }
+                .filter { it.isNotBlank() }
+        )
+        val existing = originalWorkTranslationRepository.findByOriginalWorkIdAndLocale(resourceId, targetLanguage)
+        if (existing == null) {
+            originalWorkTranslationRepository.save(OriginalWorkTranslation(resourceId, targetLanguage, payload))
+        } else {
+            existing.renderedPayload = payload
+            originalWorkTranslationRepository.save(existing)
+        }
+    }
+
+    private fun upsertCategory(resourceId: Long, targetLanguage: String, translations: Map<String, String>) {
+        val category = translations["category"].orEmpty()
+        val existing = categoryTranslationRepository.findByCategoryIdAndLocale(resourceId, targetLanguage)
+        if (existing == null) {
+            categoryTranslationRepository.save(CategoryTranslation(resourceId, targetLanguage, category))
+        } else {
+            existing.category = category
+            categoryTranslationRepository.save(existing)
+        }
+    }
+
+    companion object {
+        const val DEFAULT_PROVIDER = "papago"
+    }
+}

src/main/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationSourceExtractor.kt

@@ -0,0 +1,155 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import kr.co.vividnext.sodalive.admin.content.series.AdminContentSeriesRepository
+import kr.co.vividnext.sodalive.admin.content.series.genre.AdminContentSeriesGenreRepository
+import kr.co.vividnext.sodalive.chat.character.repository.ChatCharacterRepository
+import kr.co.vividnext.sodalive.chat.original.OriginalWorkRepository
+import kr.co.vividnext.sodalive.content.AudioContentRepository
+import kr.co.vividnext.sodalive.content.category.CategoryRepository
+import kr.co.vividnext.sodalive.content.theme.AudioContentThemeQueryRepository
+import org.springframework.data.repository.findByIdOrNull
+import org.springframework.stereotype.Component
+
+data class TranslationSourceField(
+    val fieldKey: String,
+    val sourceText: String
+)
+
+data class TranslationSource(
+    val resourceType: LanguageTranslationTargetType,
+    val resourceId: Long,
+    val sourceLanguage: String,
+    val fields: List<TranslationSourceField>
+)
+
+@Component
+class TranslationSourceExtractor(
+    private val audioContentRepository: AudioContentRepository,
+    private val chatCharacterRepository: ChatCharacterRepository,
+    private val audioContentThemeRepository: AudioContentThemeQueryRepository,
+    private val seriesRepository: AdminContentSeriesRepository,
+    private val seriesGenreRepository: AdminContentSeriesGenreRepository,
+    private val originalWorkRepository: OriginalWorkRepository,
+    private val categoryRepository: CategoryRepository
+) {
+    fun extract(resourceType: LanguageTranslationTargetType, resourceId: Long): TranslationSource? {
+        return when (resourceType) {
+            LanguageTranslationTargetType.CONTENT -> extractContent(resourceId)
+            LanguageTranslationTargetType.CHARACTER -> extractCharacter(resourceId)
+            LanguageTranslationTargetType.CONTENT_THEME -> extractContentTheme(resourceId)
+            LanguageTranslationTargetType.SERIES -> extractSeries(resourceId)
+            LanguageTranslationTargetType.SERIES_GENRE -> extractSeriesGenre(resourceId)
+            LanguageTranslationTargetType.ORIGINAL_WORK -> extractOriginalWork(resourceId)
+            LanguageTranslationTargetType.CREATOR_CONTENT_CATEGORY -> extractCategory(resourceId)
+        }
+    }
+
+    private fun extractContent(resourceId: Long): TranslationSource? {
+        val content = audioContentRepository.findByIdOrNull(resourceId) ?: return null
+        val sourceLanguage = content.languageCode?.takeIf { it.isNotBlank() } ?: return null
+        val tags = content.audioContentHashTags
+            .mapNotNull { it.hashTag?.tag }
+            .joinToString(",")
+        return TranslationSource(
+            resourceType = LanguageTranslationTargetType.CONTENT,
+            resourceId = resourceId,
+            sourceLanguage = sourceLanguage,
+            fields = listOf(
+                TranslationSourceField("title", content.title),
+                TranslationSourceField("detail", content.detail),
+                TranslationSourceField("tags", tags)
+            )
+        )
+    }
+
+    private fun extractCharacter(resourceId: Long): TranslationSource? {
+        val character = chatCharacterRepository.findByIdOrNull(resourceId) ?: return null
+        val sourceLanguage = character.languageCode?.takeIf { it.isNotBlank() } ?: return null
+        val personality = character.personalities.firstOrNull()
+        val background = character.backgrounds.firstOrNull()
+        val tags = character.tagMappings.joinToString(",") { it.tag.tag }
+        return TranslationSource(
+            resourceType = LanguageTranslationTargetType.CHARACTER,
+            resourceId = resourceId,
+            sourceLanguage = sourceLanguage,
+            fields = listOf(
+                TranslationSourceField("name", character.name),
+                TranslationSourceField("description", character.description),
+                TranslationSourceField("gender", character.gender ?: ""),
+                TranslationSourceField("personalityTrait", personality?.trait ?: ""),
+                TranslationSourceField("personalityDescription", personality?.description ?: ""),
+                TranslationSourceField("backgroundTopic", background?.topic ?: ""),
+                TranslationSourceField("backgroundDescription", background?.description ?: ""),
+                TranslationSourceField("tags", tags)
+            )
+        )
+    }
+
+    private fun extractContentTheme(resourceId: Long): TranslationSource? {
+        val contentTheme = audioContentThemeRepository.findThemeByIdAndActive(resourceId) ?: return null
+        return TranslationSource(
+            resourceType = LanguageTranslationTargetType.CONTENT_THEME,
+            resourceId = resourceId,
+            sourceLanguage = "ko",
+            fields = listOf(TranslationSourceField("theme", contentTheme.theme))
+        )
+    }
+
+    private fun extractSeries(resourceId: Long): TranslationSource? {
+        val series = seriesRepository.findByIdOrNull(resourceId) ?: return null
+        val sourceLanguage = series.languageCode?.takeIf { it.isNotBlank() } ?: return null
+        val keywords = series.keywordList
+            .mapNotNull { it.keyword?.tag }
+            .joinToString(", ")
+        return TranslationSource(
+            resourceType = LanguageTranslationTargetType.SERIES,
+            resourceId = resourceId,
+            sourceLanguage = sourceLanguage,
+            fields = listOf(
+                TranslationSourceField("title", series.title),
+                TranslationSourceField("introduction", series.introduction),
+                TranslationSourceField("keywords", keywords)
+            )
+        )
+    }
+
+    private fun extractSeriesGenre(resourceId: Long): TranslationSource? {
+        val seriesGenre = seriesGenreRepository.findActiveSeriesGenreById(resourceId) ?: return null
+        return TranslationSource(
+            resourceType = LanguageTranslationTargetType.SERIES_GENRE,
+            resourceId = resourceId,
+            sourceLanguage = "ko",
+            fields = listOf(TranslationSourceField("genre", seriesGenre.genre))
+        )
+    }
+
+    private fun extractOriginalWork(resourceId: Long): TranslationSource? {
+        val originalWork = originalWorkRepository.findByIdOrNull(resourceId) ?: return null
+        val sourceLanguage = originalWork.languageCode?.takeIf { it.isNotBlank() } ?: return null
+        val tags = originalWork.tagMappings.joinToString(", ") { it.tag.tag }
+        return TranslationSource(
+            resourceType = LanguageTranslationTargetType.ORIGINAL_WORK,
+            resourceId = resourceId,
+            sourceLanguage = sourceLanguage,
+            fields = listOf(
+                TranslationSourceField("title", originalWork.title),
+                TranslationSourceField("contentType", originalWork.contentType),
+                TranslationSourceField("category", originalWork.category),
+                TranslationSourceField("description", originalWork.description),
+                TranslationSourceField("tags", tags)
+            )
+        )
+    }
+
+    private fun extractCategory(resourceId: Long): TranslationSource? {
+        val category = categoryRepository.findByIdOrNull(resourceId) ?: return null
+        val sourceLanguage = category.languageCode?.takeIf { it.isNotBlank() } ?: return null
+        if (!category.isActive) return null
+        return TranslationSource(
+            resourceType = LanguageTranslationTargetType.CREATOR_CONTENT_CATEGORY,
+            resourceId = resourceId,
+            sourceLanguage = sourceLanguage,
+            fields = listOf(TranslationSourceField("category", category.title))
+        )
+    }
+}

src/test/kotlin/kr/co/vividnext/sodalive/chat/character/controller/ChatCharacterControllerTest.kt

@@ -11,7 +11,7 @@ import kr.co.vividnext.sodalive.chat.character.translate.AiCharacterTranslationR
 import kr.co.vividnext.sodalive.chat.room.service.ChatRoomService
 import kr.co.vividnext.sodalive.i18n.Lang
 import kr.co.vividnext.sodalive.i18n.LangContext
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
 import kr.co.vividnext.sodalive.member.contentpreference.MemberContentPreferenceService
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertTrue
@@ -26,7 +26,7 @@ class ChatCharacterControllerTest {
     private val chatRoomService = Mockito.mock(ChatRoomService::class.java)
     private val characterCommentService = Mockito.mock(CharacterCommentService::class.java)
     private val curationQueryService = Mockito.mock(CharacterCurationQueryService::class.java)
-    private val translationService = Mockito.mock(PapagoTranslationService::class.java)
+    private val resourceTranslationJobScheduler = Mockito.mock(ResourceTranslationJobScheduler::class.java)
     private val aiCharacterTranslationRepository = Mockito.mock(AiCharacterTranslationRepository::class.java)
     private val langContext = LangContext().apply { setLang(Lang.JA) }
     private val memberContentPreferenceService = Mockito.mock(MemberContentPreferenceService::class.java)
@@ -36,7 +36,7 @@ class ChatCharacterControllerTest {
         chatRoomService = chatRoomService,
         characterCommentService = characterCommentService,
         curationQueryService = curationQueryService,
-        translationService = translationService,
+        resourceTranslationJobScheduler = resourceTranslationJobScheduler,
         aiCharacterTranslationRepository = aiCharacterTranslationRepository,
         langContext = langContext,
         memberContentPreferenceService = memberContentPreferenceService,
@@ -73,7 +73,7 @@ class ChatCharacterControllerTest {
             chatRoomService = chatRoomService,
             characterCommentService = characterCommentService,
             curationQueryService = curationQueryService,
-            translationService = translationService,
+            resourceTranslationJobScheduler = resourceTranslationJobScheduler,
             aiCharacterTranslationRepository = aiCharacterTranslationRepository,
             langContext = LangContext().apply { setLang(Lang.EN) },
             memberContentPreferenceService = memberContentPreferenceService,

src/test/kotlin/kr/co/vividnext/sodalive/content/AudioContentServiceTest.kt

@@ -18,7 +18,7 @@ import kr.co.vividnext.sodalive.content.translation.ContentTranslationRepository
 import kr.co.vividnext.sodalive.explorer.ExplorerQueryRepository
 import kr.co.vividnext.sodalive.i18n.LangContext
 import kr.co.vividnext.sodalive.i18n.SodaMessageSource
-import kr.co.vividnext.sodalive.i18n.translation.PapagoTranslationService
+import kr.co.vividnext.sodalive.i18n.translation.ResourceTranslationJobScheduler
 import kr.co.vividnext.sodalive.member.Member
 import kr.co.vividnext.sodalive.member.block.BlockMemberRepository
 import org.junit.jupiter.api.Assertions.assertEquals
@@ -44,7 +44,7 @@ class AudioContentServiceTest {
     private lateinit var commentRepository: AudioContentCommentRepository
     private lateinit var audioContentLikeRepository: AudioContentLikeRepository
     private lateinit var pinContentRepository: PinContentRepository
-    private lateinit var translationService: PapagoTranslationService
+    private lateinit var resourceTranslationJobScheduler: ResourceTranslationJobScheduler
     private lateinit var contentTranslationRepository: ContentTranslationRepository
     private lateinit var s3Uploader: S3Uploader
     private lateinit var audioContentCloudFront: AudioContentCloudFront
@@ -66,7 +66,7 @@ class AudioContentServiceTest {
         commentRepository = Mockito.mock(AudioContentCommentRepository::class.java)
         audioContentLikeRepository = Mockito.mock(AudioContentLikeRepository::class.java)
         pinContentRepository = Mockito.mock(PinContentRepository::class.java)
-        translationService = Mockito.mock(PapagoTranslationService::class.java)
+        resourceTranslationJobScheduler = Mockito.mock(ResourceTranslationJobScheduler::class.java)
         contentTranslationRepository = Mockito.mock(ContentTranslationRepository::class.java)
         s3Uploader = Mockito.mock(S3Uploader::class.java)
         audioContentCloudFront = Mockito.mock(AudioContentCloudFront::class.java)
@@ -85,7 +85,7 @@ class AudioContentServiceTest {
             commentRepository = commentRepository,
             audioContentLikeRepository = audioContentLikeRepository,
             pinContentRepository = pinContentRepository,
-            translationService = translationService,
+            resourceTranslationJobScheduler = resourceTranslationJobScheduler,
             contentTranslationRepository = contentTranslationRepository,
             s3Uploader = s3Uploader,
             objectMapper = ObjectMapper(),

src/test/kotlin/kr/co/vividnext/sodalive/content/LanguageDetectionCacheServiceTest.kt

@@ -0,0 +1,42 @@
+package kr.co.vividnext.sodalive.content
+
+import kr.co.vividnext.sodalive.i18n.translation.SourceTextNormalizer
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito
+
+class LanguageDetectionCacheServiceTest {
+    @Test
+    fun shouldReuseCachedLanguageDetectionForSameNormalizedText() {
+        val repository = Mockito.mock(LanguageDetectionResultRepository::class.java)
+        val service = LanguageDetectionCacheService(repository)
+        val sourceHash = SourceTextNormalizer.hash("Hello world")
+
+        Mockito.`when`(
+            repository.findBySourceHashAndProviderAndNormalizationVersion(
+                sourceHash = sourceHash,
+                provider = "papago",
+                normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+            )
+        ).thenReturn(
+            LanguageDetectionResult(
+                sourceHash = sourceHash,
+                sourceTextSample = "Hello world",
+                detectedLanguage = "en",
+                provider = "papago",
+                confidence = null,
+                normalizationVersion = SourceTextNormalizer.NORMALIZATION_VERSION
+            )
+        )
+
+        var providerCalls = 0
+        val detected = service.detectWithCache("Hello   world") {
+            providerCalls++
+            "ko"
+        }
+
+        assertEquals("en", detected)
+        assertEquals(0, providerCalls)
+        Mockito.verify(repository, Mockito.never()).save(Mockito.any(LanguageDetectionResult::class.java))
+    }
+}

src/test/kotlin/kr/co/vividnext/sodalive/i18n/translation/SourceTextNormalizerTest.kt

@@ -0,0 +1,15 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+class SourceTextNormalizerTest {
+    @Test
+    fun shouldNormalizeWhitespaceAndUnicodeBeforeHashing() {
+        val composed = "카페\n\t소개"
+        val decomposed = "카페   소개"
+
+        assertEquals("카페 소개", SourceTextNormalizer.normalize(composed))
+        assertEquals(SourceTextNormalizer.hash(composed), SourceTextNormalizer.hash(decomposed))
+    }
+}

src/test/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationJobSchedulerTest.kt

@@ -0,0 +1,87 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Test
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito
+
+class TranslationJobSchedulerTest {
+    @Test
+    fun shouldCreateOnePendingJobForMissingNormalizedText() {
+        val jobRepository = Mockito.mock(TranslationJobRepository::class.java)
+        val scheduler = TranslationJobScheduler(jobRepository)
+
+        Mockito.`when`(
+            jobRepository.findActiveJob(
+                resourceType = LanguageTranslationTargetType.CONTENT,
+                resourceId = 10L,
+                fieldKey = "title",
+                targetLanguage = "en",
+                sourceHash = SourceTextNormalizer.hash("제목")
+            )
+        ).thenReturn(null)
+
+        scheduler.scheduleMissingTranslation(
+            resourceType = LanguageTranslationTargetType.CONTENT,
+            resourceId = 10L,
+            fieldKey = "title",
+            sourceText = "  제목\n",
+            sourceLanguage = "ko",
+            targetLanguage = "EN"
+        )
+
+        val captor = ArgumentCaptor.forClass(TranslationJob::class.java)
+        Mockito.verify(jobRepository).save(captor.capture())
+
+        val saved = captor.value
+        assertEquals(LanguageTranslationTargetType.CONTENT, saved.resourceType)
+        assertEquals(10L, saved.resourceId)
+        assertEquals("title", saved.fieldKey)
+        assertEquals("제목", saved.sourceText)
+        assertEquals(SourceTextNormalizer.hash("제목"), saved.sourceHash)
+        assertEquals("ko", saved.sourceLanguage)
+        assertEquals("en", saved.targetLanguage)
+        assertEquals(TranslationJobStatus.PENDING, saved.status)
+        assertNotNull(saved.nextRetryAt)
+    }
+
+    @Test
+    fun shouldNotCreateDuplicateJobWhenSameCompletedJobAlreadyExists() {
+        val jobRepository = Mockito.mock(TranslationJobRepository::class.java)
+        val scheduler = TranslationJobScheduler(jobRepository)
+        val sourceHash = SourceTextNormalizer.hash("제목")
+
+        Mockito.`when`(
+            jobRepository.findByResourceTypeAndResourceIdAndFieldKeyAndTargetLanguageAndSourceHash(
+                resourceType = LanguageTranslationTargetType.CONTENT,
+                resourceId = 10L,
+                fieldKey = "title",
+                targetLanguage = "en",
+                sourceHash = sourceHash
+            )
+        ).thenReturn(
+            TranslationJob(
+                resourceType = LanguageTranslationTargetType.CONTENT,
+                resourceId = 10L,
+                fieldKey = "title",
+                sourceHash = sourceHash,
+                sourceText = "제목",
+                sourceLanguage = "ko",
+                targetLanguage = "en",
+                status = TranslationJobStatus.COMPLETED
+            )
+        )
+
+        scheduler.scheduleMissingTranslation(
+            resourceType = LanguageTranslationTargetType.CONTENT,
+            resourceId = 10L,
+            fieldKey = "title",
+            sourceText = "제목",
+            sourceLanguage = "ko",
+            targetLanguage = "en"
+        )
+
+        Mockito.verify(jobRepository, Mockito.never()).save(Mockito.any(TranslationJob::class.java))
+    }
+}

src/test/kotlin/kr/co/vividnext/sodalive/i18n/translation/TranslationJobWorkerTest.kt

@@ -0,0 +1,231 @@
+package kr.co.vividnext.sodalive.i18n.translation
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.transaction.support.AbstractPlatformTransactionManager
+import org.springframework.transaction.support.DefaultTransactionStatus
+import java.time.LocalDateTime
+import java.util.Optional
+
+class TranslationJobWorkerTest {
+    @Test
+    fun shouldRunEveryTenMinutesByDefault() {
+        val scheduled = TranslationJobWorker::class.java
+            .getDeclaredMethod("runPendingJobs")
+            .getAnnotation(Scheduled::class.java)
+
+        assertEquals("\${sodalive.translation-job.fixed-delay-ms:600000}", scheduled.fixedDelayString)
+    }
+
+    @Test
+    fun shouldClaimPendingJobGroupByLockedRepositoryMethod() {
+        val jobRepository = Mockito.mock(TranslationJobRepository::class.java)
+        val memoryRepository = Mockito.mock(TranslationMemoryRepository::class.java)
+        val provider = successfulProvider()
+        val materializer = Mockito.mock(TranslationReadModelMaterializer::class.java)
+        val worker = TranslationJobWorker(
+            translationJobRepository = jobRepository,
+            translationMemoryRepository = memoryRepository,
+            translationProvider = provider,
+            materializer = materializer,
+            transactionManager = TestTransactionManager()
+        )
+        val job = translationJob()
+        job.id = 100L
+
+        Mockito.`when`(jobRepository.findNextPendingGroupJobIdsForUpdate(anyLocalDateTime())).thenReturn(listOf(100L))
+        Mockito.`when`(jobRepository.findById(100L)).thenReturn(Optional.of(job))
+
+        worker.processNextGroup()
+
+        Mockito.verify(jobRepository).findNextPendingGroupJobIdsForUpdate(anyLocalDateTime())
+    }
+
+    @Test
+    fun shouldProcessAllJobsInClaimedGroupBeforeMaterializing() {
+        val jobRepository = Mockito.mock(TranslationJobRepository::class.java)
+        val memoryRepository = Mockito.mock(TranslationMemoryRepository::class.java)
+        val provider = successfulProvider()
+        val materializer = Mockito.mock(TranslationReadModelMaterializer::class.java)
+        val worker = TranslationJobWorker(
+            translationJobRepository = jobRepository,
+            translationMemoryRepository = memoryRepository,
+            translationProvider = provider,
+            materializer = materializer,
+            transactionManager = TestTransactionManager()
+        )
+        val titleJob = translationJob(fieldKey = "title", sourceText = "제목")
+        titleJob.id = 100L
+        val detailJob = translationJob(fieldKey = "detail", sourceText = "설명")
+        detailJob.id = 101L
+
+        Mockito.`when`(jobRepository.findNextPendingGroupJobIdsForUpdate(anyLocalDateTime()))
+            .thenReturn(listOf(100L, 101L))
+        Mockito.`when`(jobRepository.findById(100L)).thenReturn(Optional.of(titleJob))
+        Mockito.`when`(jobRepository.findById(101L)).thenReturn(Optional.of(detailJob))
+
+        worker.processNextGroup()
+
+        assertEquals(TranslationJobStatus.COMPLETED, titleJob.status)
+        assertEquals(TranslationJobStatus.COMPLETED, detailJob.status)
+        Mockito.verify(materializer, Mockito.times(1)).materialize(LanguageTranslationTargetType.CONTENT, 10L, "en")
+    }
+
+    @Test
+    fun shouldRetryGroupWhenMaterializationFails() {
+        val jobRepository = Mockito.mock(TranslationJobRepository::class.java)
+        val memoryRepository = Mockito.mock(TranslationMemoryRepository::class.java)
+        val provider = successfulProvider()
+        val materializer = Mockito.mock(TranslationReadModelMaterializer::class.java)
+        val worker = TranslationJobWorker(
+            translationJobRepository = jobRepository,
+            translationMemoryRepository = memoryRepository,
+            translationProvider = provider,
+            materializer = materializer,
+            transactionManager = TestTransactionManager()
+        )
+        val titleJob = translationJob(fieldKey = "title", sourceText = "제목")
+        titleJob.id = 100L
+        val detailJob = translationJob(fieldKey = "detail", sourceText = "설명")
+        detailJob.id = 101L
+        val beforeRetryAt = titleJob.nextRetryAt
+
+        Mockito.`when`(jobRepository.findNextPendingGroupJobIdsForUpdate(anyLocalDateTime()))
+            .thenReturn(listOf(100L, 101L))
+        Mockito.`when`(jobRepository.findById(100L)).thenReturn(Optional.of(titleJob))
+        Mockito.`when`(jobRepository.findById(101L)).thenReturn(Optional.of(detailJob))
+        Mockito.`when`(materializer.materialize(LanguageTranslationTargetType.CONTENT, 10L, "en"))
+            .thenThrow(IllegalStateException("materialize down"))
+
+        worker.processNextGroup()
+
+        assertEquals(TranslationJobStatus.PENDING, titleJob.status)
+        assertEquals(TranslationJobStatus.PENDING, detailJob.status)
+        assertEquals(1, titleJob.retryCount)
+        assertEquals(1, detailJob.retryCount)
+        assertTrue(titleJob.nextRetryAt.isAfter(beforeRetryAt))
+        assertTrue(detailJob.nextRetryAt.isAfter(beforeRetryAt))
+    }
+
+    @Test
+    fun shouldLimitRunToFiveGroupsPerTick() {
+        val jobRepository = Mockito.mock(TranslationJobRepository::class.java)
+        val memoryRepository = Mockito.mock(TranslationMemoryRepository::class.java)
+        val provider = successfulProvider()
+        val materializer = Mockito.mock(TranslationReadModelMaterializer::class.java)
+        val worker = TranslationJobWorker(
+            translationJobRepository = jobRepository,
+            translationMemoryRepository = memoryRepository,
+            translationProvider = provider,
+            materializer = materializer,
+            transactionManager = TestTransactionManager()
+        )
+        val jobs = (1L..6L).map { id ->
+            translationJob(resourceId = id).also { it.id = id }
+        }
+        Mockito.`when`(jobRepository.findNextPendingGroupJobIdsForUpdate(anyLocalDateTime()))
+            .thenReturn(listOf(1L), listOf(2L), listOf(3L), listOf(4L), listOf(5L), listOf(6L))
+        jobs.forEach { job ->
+            Mockito.`when`(jobRepository.findById(job.id!!)).thenReturn(Optional.of(job))
+        }
+
+        worker.runPendingJobs()
+
+        val savedJobCaptor = ArgumentCaptor.forClass(TranslationJob::class.java)
+        Mockito.verify(jobRepository, Mockito.atLeastOnce()).save(savedJobCaptor.capture())
+        val completedResourceIds = savedJobCaptor.allValues
+            .filter { it.status == TranslationJobStatus.COMPLETED }
+            .map { it.resourceId }
+            .toSet()
+        assertEquals(setOf(1L, 2L, 3L, 4L, 5L), completedResourceIds)
+    }
+
+    @Test
+    fun shouldRetryFailedJobWithBackoff() {
+        val jobRepository = Mockito.mock(TranslationJobRepository::class.java)
+        val memoryRepository = Mockito.mock(TranslationMemoryRepository::class.java)
+        val provider = failingProvider()
+        val materializer = Mockito.mock(TranslationReadModelMaterializer::class.java)
+        val worker = TranslationJobWorker(
+            translationJobRepository = jobRepository,
+            translationMemoryRepository = memoryRepository,
+            translationProvider = provider,
+            materializer = materializer,
+            transactionManager = TestTransactionManager()
+        )
+        val job = translationJob()
+        job.id = 200L
+        val beforeRetryAt = job.nextRetryAt
+
+        Mockito.`when`(jobRepository.findNextPendingGroupJobIdsForUpdate(anyLocalDateTime())).thenReturn(listOf(200L))
+        Mockito.`when`(jobRepository.findById(200L)).thenReturn(Optional.of(job))
+
+        worker.processNextGroup()
+
+        assertEquals(TranslationJobStatus.PENDING, job.status)
+        assertEquals(1, job.retryCount)
+        assertEquals("provider down", job.lastErrorMessage)
+        assertTrue(job.nextRetryAt.isAfter(beforeRetryAt))
+    }
+
+    private fun translationJob(
+        resourceId: Long = 10L,
+        fieldKey: String = "title",
+        sourceText: String = "제목"
+    ): TranslationJob {
+        return TranslationJob(
+            resourceType = LanguageTranslationTargetType.CONTENT,
+            resourceId = resourceId,
+            fieldKey = fieldKey,
+            sourceHash = SourceTextNormalizer.hash(sourceText),
+            sourceText = sourceText,
+            sourceLanguage = "ko",
+            targetLanguage = "en"
+        )
+    }
+
+    private fun successfulProvider(): TranslationProvider {
+        return object : TranslationProvider {
+            override val providerName: String = "papago"
+            override val providerVersion: String = "nmt-v1"
+
+            override fun translate(request: TranslateRequest): TranslateResult {
+                return TranslateResult(listOf("title"))
+            }
+        }
+    }
+
+    private fun failingProvider(): TranslationProvider {
+        return object : TranslationProvider {
+            override val providerName: String = "papago"
+            override val providerVersion: String = "nmt-v1"
+
+            override fun translate(request: TranslateRequest): TranslateResult {
+                throw IllegalStateException("provider down")
+            }
+        }
+    }
+
+    private fun anyLocalDateTime(): LocalDateTime {
+        return Mockito.any(LocalDateTime::class.java) ?: LocalDateTime.now()
+    }
+}
+
+private class TestTransactionManager : AbstractPlatformTransactionManager() {
+    override fun doGetTransaction(): Any {
+        return Any()
+    }
+
+    override fun doBegin(transaction: Any, definition: org.springframework.transaction.TransactionDefinition) {
+    }
+
+    override fun doCommit(status: DefaultTransactionStatus) {
+    }
+
+    override fun doRollback(status: DefaultTransactionStatus) {
+    }
+}