diff --git a/internal/api/handlers/admin.go b/internal/api/handlers/admin.go
index e2449b2b..10e27250 100644
--- a/internal/api/handlers/admin.go
+++ b/internal/api/handlers/admin.go
@@ -1082,6 +1082,7 @@ var sensitiveSettingKeys = map[string]bool{
"mdblist.api_key": true,
"requests.radarr.api_key": true,
"requests.sonarr.api_key": true,
+ "subtitle_ai.api_key": true,
"watchsync.trakt.client_id": true,
"watchsync.trakt.client_secret": true,
"watchsync.simkl.client_id": true,
diff --git a/internal/api/handlers/subtitle_ai.go b/internal/api/handlers/subtitle_ai.go
new file mode 100644
index 00000000..72d0bf4e
--- /dev/null
+++ b/internal/api/handlers/subtitle_ai.go
@@ -0,0 +1,188 @@
+package handlers
+
+import (
+ "encoding/json"
+ "errors"
+ "log/slog"
+ "net/http"
+ "strconv"
+
+ "github.com/go-chi/chi/v5"
+
+ apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
+ "github.com/Silo-Server/silo-server/internal/catalog"
+ "github.com/Silo-Server/silo-server/internal/subtitles/ai"
+)
+
+// SubtitleAIHandler exposes on-demand AI subtitle translation backed by the
+// configured OpenAI-compatible engine. Generated tracks are stored as ordinary
+// downloaded subtitles, so they reach every client through the existing
+// subtitle pipeline.
+type SubtitleAIHandler struct {
+ service *ai.Service
+ FileAuthorizer *MediaFileAuthorizer
+}
+
+// NewSubtitleAIHandler creates a handler backed by the given service.
+func NewSubtitleAIHandler(service *ai.Service) *SubtitleAIHandler {
+ return &SubtitleAIHandler{service: service}
+}
+
+// authorizeMediaFileAccess verifies the caller may access the given media file.
+// Shared by the subtitle handlers so authorization stays in one place.
+func authorizeMediaFileAccess(w http.ResponseWriter, r *http.Request, authorizer *MediaFileAuthorizer, fileID int) bool {
+ if authorizer == nil {
+ writeError(w, http.StatusInternalServerError, "internal_error", "Media file authorization is not configured")
+ return false
+ }
+ if _, err := authorizer.Authorize(r, fileID); err != nil {
+ switch {
+ case errors.Is(err, catalog.ErrItemNotFound), errors.Is(err, catalog.ErrEpisodeNotFound):
+ writeError(w, http.StatusNotFound, "not_found", "Media file not found")
+ default:
+ writeError(w, http.StatusInternalServerError, "internal_error", "Failed to authorize media file")
+ }
+ return false
+ }
+ return true
+}
+
+type translateSubtitleRequest struct {
+ MediaFileID int `json:"media_file_id"`
+ SourceIndex int `json:"source_index"`
+ SourceLanguage string `json:"source_language"`
+ TargetLanguage string `json:"target_language"`
+ SessionID string `json:"session_id"`
+ StartPosition float64 `json:"start_position"`
+}
+
+// HandleStatus reports whether AI subtitle translation is available, so the
+// player can show or hide the entry point. GET /api/v1/subtitles/ai/status
+func (h *SubtitleAIHandler) HandleStatus(w http.ResponseWriter, r *http.Request) {
+ writeJSON(w, http.StatusOK, map[string]any{"enabled": h.service.Enabled()})
+}
+
+// WriteSubtitleAIDisabledStatus answers the AI status capability probe with a
+// 200 {"enabled": false} when no AI handler is wired, so the client gets a clean
+// negative instead of a 404 (the 2-segment /ai/status path is not shadowed by the
+// 1-segment /{media_file_id} route — they never compete in chi's router).
+func WriteSubtitleAIDisabledStatus(w http.ResponseWriter, _ *http.Request) {
+ writeJSON(w, http.StatusOK, map[string]any{"enabled": false})
+}
+
+// HandleTranslate enqueues a translation job. POST /api/v1/subtitles/ai/translate
+func (h *SubtitleAIHandler) HandleTranslate(w http.ResponseWriter, r *http.Request) {
+ var req translateSubtitleRequest
+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+ writeError(w, http.StatusBadRequest, "invalid_request", "Invalid request body")
+ return
+ }
+ if req.MediaFileID <= 0 {
+ writeError(w, http.StatusBadRequest, "bad_request", "media_file_id is required")
+ return
+ }
+ if req.TargetLanguage == "" {
+ writeError(w, http.StatusBadRequest, "bad_request", "target_language is required")
+ return
+ }
+
+ if !authorizeMediaFileAccess(w, r, h.FileAuthorizer, req.MediaFileID) {
+ return
+ }
+
+ var requestedBy *int
+ if userID := apimw.GetUserID(r.Context()); userID != 0 {
+ requestedBy = &userID
+ }
+
+ job, err := h.service.Enqueue(r.Context(), ai.JobRequest{
+ MediaFileID: req.MediaFileID,
+ Kind: ai.JobKindTranslate,
+ SourceIndex: req.SourceIndex,
+ SourceLanguage: req.SourceLanguage,
+ TargetLanguage: req.TargetLanguage,
+ RequestedBy: requestedBy,
+ SessionID: req.SessionID,
+ StartPosition: req.StartPosition,
+ })
+ if err != nil {
+ switch {
+ case errors.Is(err, ai.ErrEngineNotConfigured):
+ writeError(w, http.StatusServiceUnavailable, "not_configured",
+ "AI subtitle translation is not configured on this server")
+ case errors.Is(err, ai.ErrInvalidRequest):
+ writeError(w, http.StatusBadRequest, "bad_request", err.Error())
+ default:
+ slog.Error("failed to enqueue subtitle translation",
+ "media_file_id", req.MediaFileID, "error", err)
+ writeError(w, http.StatusInternalServerError, "internal_error", "Failed to start translation")
+ }
+ return
+ }
+
+ writeJSON(w, http.StatusAccepted, map[string]any{"job": job})
+}
+
+// HandleGetJob returns a job's current state. GET /api/v1/subtitles/ai/jobs/{job_id}
+func (h *SubtitleAIHandler) HandleGetJob(w http.ResponseWriter, r *http.Request) {
+ job, ok := h.loadAuthorizedJob(w, r)
+ if !ok {
+ return
+ }
+ writeJSON(w, http.StatusOK, map[string]any{"job": job})
+}
+
+// HandleCancelJob cancels a job. POST /api/v1/subtitles/ai/jobs/{job_id}/cancel
+func (h *SubtitleAIHandler) HandleCancelJob(w http.ResponseWriter, r *http.Request) {
+ job, ok := h.loadAuthorizedJob(w, r)
+ if !ok {
+ return
+ }
+ if err := h.service.Cancel(r.Context(), job.ID); err != nil {
+ writeError(w, http.StatusInternalServerError, "internal_error", "Failed to cancel job")
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+}
+
+// HandleListJobs lists recent jobs for a media file.
+// GET /api/v1/subtitles/ai/jobs?media_file_id=N
+func (h *SubtitleAIHandler) HandleListJobs(w http.ResponseWriter, r *http.Request) {
+ mediaFileID, err := strconv.Atoi(r.URL.Query().Get("media_file_id"))
+ if err != nil || mediaFileID <= 0 {
+ writeError(w, http.StatusBadRequest, "invalid_id", "Invalid or missing media_file_id")
+ return
+ }
+ if !authorizeMediaFileAccess(w, r, h.FileAuthorizer, mediaFileID) {
+ return
+ }
+ jobs, err := h.service.ListJobs(r.Context(), mediaFileID)
+ if err != nil {
+ writeError(w, http.StatusInternalServerError, "list_error", "Failed to list jobs")
+ return
+ }
+ writeJSON(w, http.StatusOK, map[string]any{"jobs": jobs})
+}
+
+// loadAuthorizedJob parses the job_id param, loads the job, and authorizes
+// access against its media file. It writes the error response on failure.
+func (h *SubtitleAIHandler) loadAuthorizedJob(w http.ResponseWriter, r *http.Request) (*ai.Job, bool) {
+ id, err := strconv.ParseInt(chi.URLParam(r, "job_id"), 10, 64)
+ if err != nil {
+ writeError(w, http.StatusBadRequest, "invalid_id", "Invalid job ID")
+ return nil, false
+ }
+ job, err := h.service.GetJob(r.Context(), id)
+ if err != nil {
+ if errors.Is(err, ai.ErrJobNotFound) {
+ writeError(w, http.StatusNotFound, "not_found", "Job not found")
+ return nil, false
+ }
+ writeError(w, http.StatusInternalServerError, "internal_error", "Failed to load job")
+ return nil, false
+ }
+ if !authorizeMediaFileAccess(w, r, h.FileAuthorizer, job.MediaFileID) {
+ return nil, false
+ }
+ return job, true
+}
diff --git a/internal/api/handlers/subtitle_search.go b/internal/api/handlers/subtitle_search.go
index f4054550..430ffcfb 100644
--- a/internal/api/handlers/subtitle_search.go
+++ b/internal/api/handlers/subtitle_search.go
@@ -12,7 +12,6 @@ import (
"strings"
apimw "github.com/Silo-Server/silo-server/internal/api/middleware"
- "github.com/Silo-Server/silo-server/internal/catalog"
"github.com/Silo-Server/silo-server/internal/subtitles"
"github.com/go-chi/chi/v5"
)
@@ -96,20 +95,7 @@ type downloadSubtitleRequest struct {
}
func (h *SubtitleSearchHandler) authorizeMediaFile(w http.ResponseWriter, r *http.Request, fileID int) bool {
- if h.FileAuthorizer == nil {
- writeError(w, http.StatusInternalServerError, "internal_error", "Media file authorization is not configured")
- return false
- }
- if _, err := h.FileAuthorizer.Authorize(r, fileID); err != nil {
- switch {
- case errors.Is(err, catalog.ErrItemNotFound), errors.Is(err, catalog.ErrEpisodeNotFound):
- writeError(w, http.StatusNotFound, "not_found", "Media file not found")
- default:
- writeError(w, http.StatusInternalServerError, "internal_error", "Failed to authorize media file")
- }
- return false
- }
- return true
+ return authorizeMediaFileAccess(w, r, h.FileAuthorizer, fileID)
}
// HandleSearch handles POST /api/v1/subtitles/search
diff --git a/internal/api/router.go b/internal/api/router.go
index d988241b..62e8cd80 100644
--- a/internal/api/router.go
+++ b/internal/api/router.go
@@ -54,6 +54,7 @@ import (
"github.com/Silo-Server/silo-server/internal/scanqueue"
"github.com/Silo-Server/silo-server/internal/sections"
"github.com/Silo-Server/silo-server/internal/subtitles"
+ subtitleai "github.com/Silo-Server/silo-server/internal/subtitles/ai"
"github.com/Silo-Server/silo-server/internal/subtitles/opensubtitles"
"github.com/Silo-Server/silo-server/internal/subtitles/subdl"
"github.com/Silo-Server/silo-server/internal/subtitles/subsource"
@@ -515,6 +516,11 @@ func NewRouter(deps Dependencies) chi.Router {
subtitleRepo = subtitles.NewPgRepository(deps.DB)
}
+ // Notifier that pushes "subtitle ready" events to active sessions when an AI
+ // translation completes. Assigned inside the playback handler block where the
+ // realtime hub and session manager are in scope; nil when playback is off.
+ var subtitleAINotifier *playback.SubtitleReadyNotifier
+
// Build playback handler if session manager is available.
var playbackHandler *handlers.PlaybackHandler
var adminPlaybackControlHandler *handlers.AdminPlaybackControlHandler
@@ -609,6 +615,7 @@ func NewRouter(deps Dependencies) chi.Router {
playbackHandler.MarkerUpserter = deps.FileRepo
}
playbackHandler.MarkerUpdateNotifier = playback.NewMarkerUpdateNotifier(deps.SessionMgr, realtimeHub)
+ subtitleAINotifier = playback.NewSubtitleReadyNotifier(deps.SessionMgr, realtimeHub)
adminPlaybackControlHandler = handlers.NewAdminPlaybackControlHandler(playbackHandler)
if deps.DB != nil && deps.FileRepo != nil && viewerResolver != nil && deps.Config != nil && detailSvc != nil {
@@ -772,6 +779,40 @@ func NewRouter(deps Dependencies) chi.Router {
adminSubtitleHandler.SetDownloadedSubtitleDeps(deps.DB, subtitleManager)
}
+ // Build the AI subtitle handler (on-demand translation). Generated tracks are
+ // stored as ordinary downloaded subtitles, so they reach every client through
+ // the existing subtitle pipeline with no client changes.
+ var subtitleAIHandler *handlers.SubtitleAIHandler
+ if subtitleManager != nil && subtitleRepo != nil && deps.FileRepo != nil && deps.DB != nil && deps.Config != nil {
+ aiCfg := subtitleai.Config{
+ Enabled: deps.Config.SubtitleAI.Enabled,
+ BaseURL: deps.Config.SubtitleAI.BaseURL,
+ APIKey: deps.Config.SubtitleAI.APIKey,
+ ChatModel: deps.Config.SubtitleAI.ChatModel,
+ MaxConcurrentJobs: deps.Config.SubtitleAI.MaxConcurrentJobs,
+ BatchSize: deps.Config.SubtitleAI.BatchSize,
+ ContextNeighbors: deps.Config.SubtitleAI.ContextNeighbors,
+ }
+ var aiNotifier subtitleai.Notifier
+ if subtitleAINotifier != nil {
+ aiNotifier = subtitleAINotifier
+ }
+ aiService := subtitleai.NewService(
+ deps.AppContext,
+ aiCfg,
+ subtitleai.NewPgJobRepository(deps.DB),
+ subtitleai.NewLLMTranslator(subtitleai.NewClient(aiCfg), aiCfg.BatchSize, aiCfg.ContextNeighbors),
+ subtitleManager,
+ subtitleRepo,
+ deps.FileRepo,
+ aiNotifier,
+ deps.Config.Playback.FFmpegPath,
+ slog.Default(),
+ )
+ aiService.Recover()
+ subtitleAIHandler = handlers.NewSubtitleAIHandler(aiService)
+ }
+
// Build section handler if DB is available.
var sectionHandler *handlers.SectionHandler
var sectionSettingsHandler *handlers.SectionSettingsHandler
@@ -1541,20 +1582,36 @@ func NewRouter(deps Dependencies) chi.Router {
})
}
- // Subtitle search routes.
+ // Subtitle search + AI translation routes.
if subtitleSearchHandler != nil {
if deps.FileRepo != nil && itemRepo != nil {
- subtitleSearchHandler.FileAuthorizer = &handlers.MediaFileAuthorizer{
+ fileAuthorizer := &handlers.MediaFileAuthorizer{
FileResolver: deps.FileRepo,
ItemAccess: itemRepo,
EpisodeLookup: episodeRepo,
}
+ subtitleSearchHandler.FileAuthorizer = fileAuthorizer
+ if subtitleAIHandler != nil {
+ subtitleAIHandler.FileAuthorizer = fileAuthorizer
+ }
}
r.Route("/subtitles", func(r chi.Router) {
r.Post("/search", subtitleSearchHandler.HandleSearch)
r.Post("/download", subtitleSearchHandler.HandleDownload)
r.Post("/upload", subtitleSearchHandler.HandleUpload)
r.Post("/detect-language", subtitleSearchHandler.HandleDetectLanguage)
+ if subtitleAIHandler != nil {
+ r.Get("/ai/status", subtitleAIHandler.HandleStatus)
+ r.Post("/ai/translate", subtitleAIHandler.HandleTranslate)
+ r.Get("/ai/jobs", subtitleAIHandler.HandleListJobs)
+ r.Get("/ai/jobs/{job_id}", subtitleAIHandler.HandleGetJob)
+ r.Post("/ai/jobs/{job_id}/cancel", subtitleAIHandler.HandleCancelJob)
+ } else {
+ // Answer the capability probe with 200 {"enabled": false}
+ // when AI translation isn't wired, so the client gets a
+ // clean negative instead of a 404.
+ r.Get("/ai/status", handlers.WriteSubtitleAIDisabledStatus)
+ }
r.Get("/{media_file_id}", subtitleSearchHandler.HandleList)
r.Delete("/{id}", subtitleSearchHandler.HandleDelete)
})
diff --git a/internal/config/config.go b/internal/config/config.go
index b63c4ae2..a0fa7b7d 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -238,6 +238,20 @@ type RecommendationsConfig struct {
CowatchCron string `yaml:"-"`
}
+// SubtitleAIConfig holds settings for on-demand AI subtitle translation (and,
+// in a follow-up, Whisper ASR generation) via a single OpenAI-compatible
+// endpoint — the operator can point it at OpenAI, Groq, a local Ollama server,
+// etc. Mirrors the recommendations embedding client's configuration style.
+type SubtitleAIConfig struct {
+ Enabled bool `yaml:"-"`
+ BaseURL string `yaml:"-"`
+ APIKey string `yaml:"-"`
+ ChatModel string `yaml:"-"`
+ MaxConcurrentJobs int `yaml:"-"`
+ BatchSize int `yaml:"-"`
+ ContextNeighbors int `yaml:"-"`
+}
+
// DownloadConfig holds server-wide download policy settings.
type DownloadConfig struct {
Enabled bool `yaml:"-"`
@@ -268,6 +282,7 @@ type Config struct {
Auth AuthConfig `yaml:"-"`
JellyfinCompat JellyfinCompatConfig `yaml:"-"`
Recommendations RecommendationsConfig `yaml:"-"`
+ SubtitleAI SubtitleAIConfig `yaml:"-"`
Download DownloadConfig `yaml:"-"`
TMDBAPIKey string `yaml:"-"`
MDBListAPIKey string `yaml:"-"`
diff --git a/internal/config/db_loader.go b/internal/config/db_loader.go
index 978ec47b..940ca641 100644
--- a/internal/config/db_loader.go
+++ b/internal/config/db_loader.go
@@ -412,6 +412,31 @@ func LoadFromDB(m map[string]string) (*Config, error) {
cfg.Recommendations.DiversityLambda = diversityLambda
cfg.Recommendations.CowatchCron = stringOr(m, "recommendations.cowatch_cron", "30 4 * * *")
+ // Subtitle AI (on-demand translation; Whisper ASR generation in a follow-up)
+ subtitleAIEnabled, err := boolOr(m, "subtitle_ai.enabled", false)
+ if err != nil {
+ return nil, err
+ }
+ cfg.SubtitleAI.Enabled = subtitleAIEnabled
+ cfg.SubtitleAI.BaseURL = stringOr(m, "subtitle_ai.base_url", "https://api.openai.com")
+ cfg.SubtitleAI.APIKey = stringOr(m, "subtitle_ai.api_key", "")
+ cfg.SubtitleAI.ChatModel = stringOr(m, "subtitle_ai.chat_model", "gpt-4o-mini")
+ subtitleAIMaxConcurrent, err := intOr(m, "subtitle_ai.max_concurrent_jobs", 2)
+ if err != nil {
+ return nil, err
+ }
+ cfg.SubtitleAI.MaxConcurrentJobs = subtitleAIMaxConcurrent
+ subtitleAIBatchSize, err := intOr(m, "subtitle_ai.batch_size", 40)
+ if err != nil {
+ return nil, err
+ }
+ cfg.SubtitleAI.BatchSize = subtitleAIBatchSize
+ subtitleAIContextNeighbors, err := intOr(m, "subtitle_ai.context_neighbors", 2)
+ if err != nil {
+ return nil, err
+ }
+ cfg.SubtitleAI.ContextNeighbors = subtitleAIContextNeighbors
+
// Download
downloadEnabled, err := boolOr(m, "download.enabled", false)
if err != nil {
diff --git a/internal/playback/realtime.go b/internal/playback/realtime.go
index d070f147..5ff0dba8 100644
--- a/internal/playback/realtime.go
+++ b/internal/playback/realtime.go
@@ -21,13 +21,23 @@ const (
type RealtimeEventName string
const (
- RealtimeEventChapterThumbnailReady RealtimeEventName = "chapter_thumbnail_ready"
- RealtimeEventMarkersUpdated RealtimeEventName = "markers_updated"
+ RealtimeEventChapterThumbnailReady RealtimeEventName = "chapter_thumbnail_ready"
+ RealtimeEventMarkersUpdated RealtimeEventName = "markers_updated"
+ RealtimeEventSubtitleReady RealtimeEventName = "subtitle_ready"
+ RealtimeEventSubtitleTranslationStart RealtimeEventName = "subtitle_translation_started"
+ RealtimeEventSubtitleTranslationCues RealtimeEventName = "subtitle_translation_cues"
+ RealtimeEventSubtitleTranslationDone RealtimeEventName = "subtitle_translation_completed"
+ RealtimeEventSubtitleTranslationFail RealtimeEventName = "subtitle_translation_failed"
)
var supportedRealtimeEventNameSet = map[RealtimeEventName]struct{}{
- RealtimeEventChapterThumbnailReady: {},
- RealtimeEventMarkersUpdated: {},
+ RealtimeEventChapterThumbnailReady: {},
+ RealtimeEventMarkersUpdated: {},
+ RealtimeEventSubtitleReady: {},
+ RealtimeEventSubtitleTranslationStart: {},
+ RealtimeEventSubtitleTranslationCues: {},
+ RealtimeEventSubtitleTranslationDone: {},
+ RealtimeEventSubtitleTranslationFail: {},
}
// CommandName identifies a supported realtime command.
@@ -125,6 +135,74 @@ type MarkersUpdatedPayload struct {
Preview *TimeRangePayload `json:"preview"`
}
+// SubtitleReadyPayload announces that a newly generated subtitle track (AI
+// translation, and later ASR) is available for the file, so the player can
+// refresh its track list and optionally select it.
+type SubtitleReadyPayload struct {
+ SessionID string `json:"session_id"`
+ FileID int `json:"file_id"`
+ SubtitleID int `json:"subtitle_id"`
+ Language string `json:"language"`
+ Label string `json:"label,omitempty"`
+}
+
+// StreamCue is one translated subtitle cue pushed to the player during a live
+// translation. Start/End are absolute media-time seconds; Text may contain
+// embedded newlines for multi-line cues.
+type StreamCue struct {
+ Start float64 `json:"start"`
+ End float64 `json:"end"`
+ Text string `json:"text"`
+}
+
+// SubtitleTranslationStartedPayload tells the player a live translation has
+// begun, so it can create a placeholder track, select it, and pause until the
+// first cues near the playhead arrive. TrackKey identifies the live track for
+// subsequent cue/completion events.
+type SubtitleTranslationStartedPayload struct {
+ SessionID string `json:"session_id"`
+ FileID int `json:"file_id"`
+ JobID int64 `json:"job_id"`
+ TrackKey string `json:"track_key"`
+ Language string `json:"language"`
+ Label string `json:"label,omitempty"`
+ TotalCues int `json:"total_cues"`
+}
+
+// SubtitleTranslationCuesPayload delivers a batch of translated cues for a live
+// track as it is produced. Done/Total track overall progress.
+type SubtitleTranslationCuesPayload struct {
+ SessionID string `json:"session_id"`
+ FileID int `json:"file_id"`
+ JobID int64 `json:"job_id"`
+ TrackKey string `json:"track_key"`
+ Cues []StreamCue `json:"cues"`
+ Done int `json:"done"`
+ Total int `json:"total"`
+}
+
+// SubtitleTranslationCompletedPayload signals the live translation finished and
+// the full track is persisted as a downloaded subtitle (SubtitleID).
+type SubtitleTranslationCompletedPayload struct {
+ SessionID string `json:"session_id"`
+ FileID int `json:"file_id"`
+ JobID int64 `json:"job_id"`
+ TrackKey string `json:"track_key"`
+ SubtitleID int `json:"subtitle_id"`
+ Language string `json:"language"`
+ Label string `json:"label,omitempty"`
+}
+
+// SubtitleTranslationFailedPayload signals a live translation failed, so the
+// player can drop the placeholder track and resume playback.
+type SubtitleTranslationFailedPayload struct {
+ SessionID string `json:"session_id"`
+ FileID int `json:"file_id"`
+ JobID int64 `json:"job_id"`
+ TrackKey string `json:"track_key"`
+ Message string `json:"message,omitempty"`
+}
+
// NewEventEnvelope creates a validated realtime event envelope.
func NewEventEnvelope(sessionID string, name RealtimeEventName, payload json.RawMessage) (EventEnvelope, error) {
normalizedPayload, err := normalizeJSONPayload(payload)
@@ -186,6 +264,74 @@ func NewMarkersUpdatedEvent(
return NewEventEnvelope(sessionID, RealtimeEventMarkersUpdated, payload)
}
+// NewSubtitleReadyEvent creates a validated subtitle-ready event.
+func NewSubtitleReadyEvent(
+ sessionID string,
+ fileID int,
+ subtitleID int,
+ language string,
+ label string,
+) (EventEnvelope, error) {
+ payload, err := json.Marshal(SubtitleReadyPayload{
+ SessionID: sessionID,
+ FileID: fileID,
+ SubtitleID: subtitleID,
+ Language: language,
+ Label: label,
+ })
+ if err != nil {
+ return EventEnvelope{}, err
+ }
+ return NewEventEnvelope(sessionID, RealtimeEventSubtitleReady, payload)
+}
+
+// NewSubtitleTranslationStartedEvent creates a validated translation-started event.
+func NewSubtitleTranslationStartedEvent(sessionID string, fileID int, jobID int64, trackKey, language, label string, totalCues int) (EventEnvelope, error) {
+ payload, err := json.Marshal(SubtitleTranslationStartedPayload{
+ SessionID: sessionID, FileID: fileID, JobID: jobID,
+ TrackKey: trackKey, Language: language, Label: label, TotalCues: totalCues,
+ })
+ if err != nil {
+ return EventEnvelope{}, err
+ }
+ return NewEventEnvelope(sessionID, RealtimeEventSubtitleTranslationStart, payload)
+}
+
+// NewSubtitleTranslationCuesEvent creates a validated translation-cues event.
+func NewSubtitleTranslationCuesEvent(sessionID string, fileID int, jobID int64, trackKey string, cues []StreamCue, done, total int) (EventEnvelope, error) {
+ payload, err := json.Marshal(SubtitleTranslationCuesPayload{
+ SessionID: sessionID, FileID: fileID, JobID: jobID,
+ TrackKey: trackKey, Cues: cues, Done: done, Total: total,
+ })
+ if err != nil {
+ return EventEnvelope{}, err
+ }
+ return NewEventEnvelope(sessionID, RealtimeEventSubtitleTranslationCues, payload)
+}
+
+// NewSubtitleTranslationCompletedEvent creates a validated translation-completed event.
+func NewSubtitleTranslationCompletedEvent(sessionID string, fileID int, jobID int64, trackKey string, subtitleID int, language, label string) (EventEnvelope, error) {
+ payload, err := json.Marshal(SubtitleTranslationCompletedPayload{
+ SessionID: sessionID, FileID: fileID, JobID: jobID,
+ TrackKey: trackKey, SubtitleID: subtitleID, Language: language, Label: label,
+ })
+ if err != nil {
+ return EventEnvelope{}, err
+ }
+ return NewEventEnvelope(sessionID, RealtimeEventSubtitleTranslationDone, payload)
+}
+
+// NewSubtitleTranslationFailedEvent creates a validated translation-failed event.
+func NewSubtitleTranslationFailedEvent(sessionID string, fileID int, jobID int64, trackKey, message string) (EventEnvelope, error) {
+ payload, err := json.Marshal(SubtitleTranslationFailedPayload{
+ SessionID: sessionID, FileID: fileID, JobID: jobID, TrackKey: trackKey, Message: message,
+ })
+ if err != nil {
+ return EventEnvelope{}, err
+ }
+ return NewEventEnvelope(sessionID, RealtimeEventSubtitleTranslationFail, payload)
+}
+
// ParseEventEnvelope decodes and validates a realtime event envelope.
func ParseEventEnvelope(data []byte) (EventEnvelope, error) {
var env EventEnvelope
diff --git a/internal/playback/subtitle_ready_notifier.go b/internal/playback/subtitle_ready_notifier.go
new file mode 100644
index 00000000..7ca0a324
--- /dev/null
+++ b/internal/playback/subtitle_ready_notifier.go
@@ -0,0 +1,98 @@
+package playback
+
+import (
+ "context"
+ "errors"
+ "log/slog"
+)
+
+type subtitleReadySessionLookup interface {
+ GetSessionsByMediaFileID(fileID int) []*Session
+}
+
+// SubtitleReadyNotifier pushes "subtitle ready" events to active playback
+// sessions when a generated subtitle track (AI translation, later ASR) becomes
+// available, so the player can refresh and select it without a manual reload.
+//
+// It satisfies the subtitles/ai Notifier interface structurally, keeping the ai
+// package free of any playback dependency.
+type SubtitleReadyNotifier struct {
+ sessions subtitleReadySessionLookup
+ hub *RealtimeHub
+}
+
+// NewSubtitleReadyNotifier returns a notifier, or nil if its dependencies are
+// missing (callers treat a nil notifier as a no-op).
+func NewSubtitleReadyNotifier(sessions subtitleReadySessionLookup, hub *RealtimeHub) *SubtitleReadyNotifier {
+ if sessions == nil || hub == nil {
+ return nil
+ }
+ return &SubtitleReadyNotifier{sessions: sessions, hub: hub}
+}
+
+// SubtitleReady notifies active sessions for the file that a new subtitle track
+// with the given downloaded-subtitle ID is available.
+func (n *SubtitleReadyNotifier) SubtitleReady(_ context.Context, mediaFileID, subtitleID int, language, label string) {
+ if n == nil || mediaFileID <= 0 || subtitleID <= 0 {
+ return
+ }
+
+ for _, session := range n.sessions.GetSessionsByMediaFileID(mediaFileID) {
+ if session == nil || session.ID == "" || !session.HasRealtimeConnection {
+ continue
+ }
+ event, err := NewSubtitleReadyEvent(session.ID, mediaFileID, subtitleID, language, label)
+ if err != nil {
+ slog.Warn("failed to encode subtitle ready realtime event",
+ "session_id", session.ID, "file_id", mediaFileID, "subtitle_id", subtitleID, "error", err)
+ continue
+ }
+ if err := n.hub.Send(session.ID, event); err != nil && !errors.Is(err, ErrRealtimeConnectionNotFound) {
+ slog.Warn("failed to deliver subtitle ready realtime event",
+ "session_id", session.ID, "file_id", mediaFileID, "subtitle_id", subtitleID, "error", err)
+ }
+ }
+}
+
+// TranslationStarted tells one session a live translation has begun.
+func (n *SubtitleReadyNotifier) TranslationStarted(_ context.Context, sessionID string, fileID int, jobID int64, trackKey, language, label string, totalCues int) {
+ n.sendTranslation(sessionID, func() (EventEnvelope, error) {
+ return NewSubtitleTranslationStartedEvent(sessionID, fileID, jobID, trackKey, language, label, totalCues)
+ })
+}
+
+// TranslationCues pushes a batch of translated cues to one session.
+func (n *SubtitleReadyNotifier) TranslationCues(_ context.Context, sessionID string, fileID int, jobID int64, trackKey string, cues []StreamCue, done, total int) {
+ n.sendTranslation(sessionID, func() (EventEnvelope, error) {
+ return NewSubtitleTranslationCuesEvent(sessionID, fileID, jobID, trackKey, cues, done, total)
+ })
+}
+
+// TranslationCompleted tells one session a live translation finished.
+func (n *SubtitleReadyNotifier) TranslationCompleted(_ context.Context, sessionID string, fileID int, jobID int64, trackKey string, subtitleID int, language, label string) {
+ n.sendTranslation(sessionID, func() (EventEnvelope, error) {
+ return NewSubtitleTranslationCompletedEvent(sessionID, fileID, jobID, trackKey, subtitleID, language, label)
+ })
+}
+
+// TranslationFailed tells one session a live translation failed.
+func (n *SubtitleReadyNotifier) TranslationFailed(_ context.Context, sessionID string, fileID int, jobID int64, trackKey, message string) {
+ n.sendTranslation(sessionID, func() (EventEnvelope, error) {
+ return NewSubtitleTranslationFailedEvent(sessionID, fileID, jobID, trackKey, message)
+ })
+}
+
+// sendTranslation builds and delivers a translation event to a single session.
+func (n *SubtitleReadyNotifier) sendTranslation(sessionID string, build func() (EventEnvelope, error)) {
+ if n == nil || n.hub == nil || sessionID == "" {
+ return
+ }
+ event, err := build()
+ if err != nil {
+ slog.Warn("failed to encode subtitle translation realtime event", "session_id", sessionID, "error", err)
+ return
+ }
+ if err := n.hub.Send(sessionID, event); err != nil && !errors.Is(err, ErrRealtimeConnectionNotFound) {
+ slog.Warn("failed to deliver subtitle translation realtime event", "session_id", sessionID, "error", err)
+ }
+}
diff --git a/internal/subtitles/ai/client.go b/internal/subtitles/ai/client.go
new file mode 100644
index 00000000..1343c5b0
--- /dev/null
+++ b/internal/subtitles/ai/client.go
@@ -0,0 +1,181 @@
+package ai
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "io"
+ "log/slog"
+ "net/http"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// Client is a minimal OpenAI-compatible chat-completions client. It follows the
+// same retry/backoff conventions as the recommendations embedding client so the
+// two behave consistently against OpenAI, Groq, Ollama, llama.cpp servers, etc.
+type Client struct {
+ cfg Config
+ httpClient *http.Client
+}
+
+// NewClient builds a client from the engine config.
+func NewClient(cfg Config) *Client {
+ return &Client{
+ cfg: cfg,
+ httpClient: &http.Client{Timeout: 10 * time.Minute},
+ }
+}
+
+type chatMessage struct {
+ Role string `json:"role"`
+ Content string `json:"content"`
+}
+
+type chatResponseFormat struct {
+ Type string `json:"type"`
+}
+
+type chatCompletionRequest struct {
+ Model string `json:"model"`
+ Messages []chatMessage `json:"messages"`
+ Temperature float32 `json:"temperature"`
+ ResponseFormat *chatResponseFormat `json:"response_format,omitempty"`
+}
+
+type chatCompletionResponse struct {
+ Choices []struct {
+ Message chatMessage `json:"message"`
+ } `json:"choices"`
+ // Some OpenAI-compatible gateways (e.g. OpenRouter) return a 200 with an
+ // error object instead of an HTTP error status when an upstream provider
+ // fails. We surface and retry on it.
+ Error *struct {
+ Message string `json:"message"`
+ } `json:"error"`
+}
+
+// chat performs one chat completion and returns the first choice's content.
+// When jsonObject is true it requests response_format=json_object; providers
+// that ignore the field still work because the prompt itself demands JSON.
+func (c *Client) chat(ctx context.Context, messages []chatMessage, jsonObject bool) (string, error) {
+ reqBody := chatCompletionRequest{
+ Model: c.cfg.ChatModel,
+ Messages: messages,
+ Temperature: 0.2,
+ }
+ if jsonObject {
+ reqBody.ResponseFormat = &chatResponseFormat{Type: "json_object"}
+ }
+
+ body, err := json.Marshal(reqBody)
+ if err != nil {
+ return "", fmt.Errorf("marshal chat request: %w", err)
+ }
+
+ url := strings.TrimRight(c.cfg.BaseURL, "/") + "/v1/chat/completions"
+
+ const maxAttempts = 6
+ var lastErr error
+ for attempt := 0; attempt < maxAttempts; attempt++ {
+ httpReq, reqErr := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
+ if reqErr != nil {
+ return "", fmt.Errorf("create request: %w", reqErr)
+ }
+ httpReq.Header.Set("Content-Type", "application/json")
+ if c.cfg.APIKey != "" {
+ httpReq.Header.Set("Authorization", "Bearer "+c.cfg.APIKey)
+ }
+
+ resp, doErr := c.httpClient.Do(httpReq)
+ if doErr != nil {
+ lastErr = fmt.Errorf("chat request failed: %w", doErr)
+ if waitErr := sleepCtx(ctx, time.Duration(attempt+1)*time.Second); waitErr != nil {
+ return "", waitErr
+ }
+ continue
+ }
+
+ respBody, _ := io.ReadAll(resp.Body)
+ resp.Body.Close()
+
+ switch {
+ case resp.StatusCode == http.StatusTooManyRequests:
+ wait := rateLimitBackoff(resp, attempt)
+ slog.Warn("rate limited by subtitle AI chat API, waiting", "attempt", attempt+1, "wait", wait)
+ lastErr = fmt.Errorf("chat API returned 429: %s", truncate(string(respBody), 300))
+ if waitErr := sleepCtx(ctx, wait); waitErr != nil {
+ return "", waitErr
+ }
+ continue
+ case resp.StatusCode >= 500:
+ lastErr = fmt.Errorf("chat API returned %d: %s", resp.StatusCode, truncate(string(respBody), 300))
+ if waitErr := sleepCtx(ctx, time.Duration(attempt+1)*time.Second); waitErr != nil {
+ return "", waitErr
+ }
+ continue
+ case resp.StatusCode != http.StatusOK:
+ // 4xx other than 429: not retryable.
+ return "", fmt.Errorf("chat API returned %d: %s", resp.StatusCode, truncate(string(respBody), 300))
+ }
+
+ var parsed chatCompletionResponse
+ if err := json.Unmarshal(respBody, &parsed); err != nil {
+ lastErr = fmt.Errorf("decode chat response: %w", err)
+ } else if parsed.Error != nil && parsed.Error.Message != "" {
+ // 200 with an upstream error object — transient on gateways.
+ lastErr = fmt.Errorf("chat API error: %s", parsed.Error.Message)
+ } else if len(parsed.Choices) == 0 || parsed.Choices[0].Message.Content == "" {
+ lastErr = fmt.Errorf("chat API returned no choices")
+ slog.Warn("subtitle AI chat returned no choices, retrying",
+ "attempt", attempt+1, "model", c.cfg.ChatModel, "body", truncate(string(respBody), 300))
+ } else {
+ return parsed.Choices[0].Message.Content, nil
+ }
+
+ // Empty-choices / 200-error / decode failure: retry with backoff.
+ if waitErr := sleepCtx(ctx, time.Duration(attempt+1)*time.Second); waitErr != nil {
+ return "", waitErr
+ }
+ }
+
+ if lastErr == nil {
+ lastErr = fmt.Errorf("chat API: retries exhausted")
+ }
+ return "", lastErr
+}
+
+// truncate caps a string for inclusion in an error or log line.
+func truncate(s string, maxLen int) string {
+ if len(s) > maxLen {
+ return s[:maxLen] + "..."
+ }
+ return s
+}
+
+// sleepCtx waits for d or until ctx is cancelled.
+func sleepCtx(ctx context.Context, d time.Duration) error {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-time.After(d):
+ return nil
+ }
+}
+
+// rateLimitBackoff returns how long to wait after a 429, honoring Retry-After
+// and otherwise backing off exponentially (capped at 60s).
+func rateLimitBackoff(resp *http.Response, attempt int) time.Duration {
+ if ra := resp.Header.Get("Retry-After"); ra != "" {
+ if secs, err := strconv.Atoi(ra); err == nil && secs > 0 {
+ return time.Duration(secs) * time.Second
+ }
+ }
+ wait := 10 * time.Second * (1 << attempt)
+ if wait > 60*time.Second {
+ wait = 60 * time.Second
+ }
+ return wait
+}
diff --git a/internal/subtitles/ai/config.go b/internal/subtitles/ai/config.go
new file mode 100644
index 00000000..dafc15b8
--- /dev/null
+++ b/internal/subtitles/ai/config.go
@@ -0,0 +1,20 @@
+package ai
+
+// Config holds the runtime configuration for the AI subtitle engine. Values are
+// loaded from server_settings (see internal/config) and mirror the existing
+// recommendations embedding client: one OpenAI-compatible endpoint that the
+// operator can point at OpenAI, Groq, a local Ollama/llama.cpp server, etc.
+type Config struct {
+ Enabled bool
+ BaseURL string // e.g. "https://api.openai.com" (no trailing /v1)
+ APIKey string // empty for keyless local servers
+ ChatModel string // chat-completions model used for translation
+ MaxConcurrentJobs int // semaphore bound so jobs never starve transcodes
+ BatchSize int // cues per translation request
+ ContextNeighbors int // preceding source cues sent as untranslated context
+}
+
+// Ready reports whether the engine is enabled and minimally configured.
+func (c Config) Ready() bool {
+ return c.Enabled && c.BaseURL != "" && c.ChatModel != ""
+}
diff --git a/internal/subtitles/ai/engine.go b/internal/subtitles/ai/engine.go
new file mode 100644
index 00000000..45666100
--- /dev/null
+++ b/internal/subtitles/ai/engine.go
@@ -0,0 +1,41 @@
+// Package ai provides on-the-fly subtitle translation (and, in a follow-up,
+// Whisper ASR generation) backed by a single OpenAI-compatible API. Generated
+// tracks are stored as ordinary downloaded subtitles and served to every
+// client through the existing subtitle pipeline.
+package ai
+
+import (
+ "context"
+ "time"
+)
+
+// SubtitleCue is one timed subtitle entry. Times are absolute media positions.
+// Translation only rewrites Lines; Start/End are preserved verbatim so timing
+// can never drift.
+type SubtitleCue struct {
+ Start time.Duration
+ End time.Duration
+ Lines []string
+}
+
+// Translator converts subtitle cues from one language to another, preserving
+// cue count, order, and timing. The built-in implementation is LLMTranslator
+// (OpenAI-compatible chat completions); the interface is the seam through which
+// a future translation plugin can be substituted without touching the job
+// pipeline.
+type Translator interface {
+ // Translate returns translated cues with the same count and order as the
+ // input. onBatch, when non-nil, is called after each batch with that batch's
+ // translated cues and overall progress (done/total cues), so callers can both
+ // report progress and stream cues as they land. Implementations must honor
+ // ctx cancellation.
+ Translate(ctx context.Context, req TranslateRequest, onBatch func(batch []SubtitleCue, done, total int)) ([]SubtitleCue, error)
+}
+
+// TranslateRequest is the input to a Translator.
+type TranslateRequest struct {
+ Cues []SubtitleCue
+ SourceLanguage string // ISO/BCP-47 code; "" lets the model infer the source
+ TargetLanguage string // ISO/BCP-47 code (required)
+ MediaTitle string // optional context hint for the model
+}
diff --git a/internal/subtitles/ai/job.go b/internal/subtitles/ai/job.go
new file mode 100644
index 00000000..62a6ec2b
--- /dev/null
+++ b/internal/subtitles/ai/job.go
@@ -0,0 +1,87 @@
+package ai
+
+import (
+ "crypto/sha256"
+ "encoding/hex"
+ "fmt"
+ "time"
+)
+
+// JobKind identifies what an AI subtitle job does. Only translation ships in
+// the first iteration; transcription kinds are reserved for the Whisper ASR
+// follow-up so the schema and API do not change again.
+type JobKind string
+
+const (
+ JobKindTranslate JobKind = "translate"
+)
+
+// JobStatus is the lifecycle state of a job.
+type JobStatus string
+
+const (
+ JobStatusPending JobStatus = "pending"
+ JobStatusRunning JobStatus = "running"
+ JobStatusCompleted JobStatus = "completed"
+ JobStatusFailed JobStatus = "failed"
+ JobStatusCancelled JobStatus = "cancelled"
+)
+
+// Terminal reports whether a status is final.
+func (s JobStatus) Terminal() bool {
+ switch s {
+ case JobStatusCompleted, JobStatusFailed, JobStatusCancelled:
+ return true
+ default:
+ return false
+ }
+}
+
+// Job is a persisted AI subtitle job. It is serialized to the API as-is.
+type Job struct {
+ ID int64 `json:"id"`
+ MediaFileID int `json:"media_file_id"`
+ Kind JobKind `json:"kind"`
+ SourceIndex int `json:"source_index"`
+ SourceLanguage string `json:"source_language"`
+ TargetLanguage string `json:"target_language"`
+ Engine string `json:"engine"`
+ Model string `json:"model"`
+ Status JobStatus `json:"status"`
+ Progress float64 `json:"progress"`
+ ProgressMessage string `json:"progress_message"`
+ ResultSubtitleID *int `json:"result_subtitle_id"`
+ ErrorMessage string `json:"error_message,omitempty"`
+ IdempotencyKey string `json:"-"`
+ RequestedBy *int `json:"-"`
+ CreatedAt time.Time `json:"created_at"`
+ UpdatedAt time.Time `json:"updated_at"`
+ HeartbeatAt time.Time `json:"-"`
+
+ // Transient, set from the request and used only while the job runs (not
+ // persisted): the realtime session to stream live cues to, and the playhead
+ // position so translation starts where the viewer is watching.
+ SessionID string `json:"-"`
+ StartPosition float64 `json:"-"`
+}
+
+// JobRequest is the input to Service.Enqueue.
+type JobRequest struct {
+ MediaFileID int
+ Kind JobKind
+ SourceIndex int
+ SourceLanguage string
+ TargetLanguage string
+ RequestedBy *int
+ // SessionID, when set, streams live cues to that playback session.
+ SessionID string
+ // StartPosition (seconds) makes translation start at the viewer's playhead.
+ StartPosition float64
+}
+
+// idempotencyKey derives the dedup key for a job. Two requests for the same
+// source track, target language, and model collapse to one in-flight job.
+func idempotencyKey(mediaFileID int, kind JobKind, sourceIndex int, targetLang, model string) string {
+ sum := sha256.Sum256([]byte(fmt.Sprintf("%d|%s|%d|%s|%s", mediaFileID, kind, sourceIndex, targetLang, model)))
+ return hex.EncodeToString(sum[:])
+}
diff --git a/internal/subtitles/ai/languages.go b/internal/subtitles/ai/languages.go
new file mode 100644
index 00000000..6e561b3a
--- /dev/null
+++ b/internal/subtitles/ai/languages.go
@@ -0,0 +1,36 @@
+package ai
+
+import "strings"
+
+// languageNames maps common ISO 639-1 codes to English names. The model handles
+// bare codes acceptably, but full names noticeably improve translation quality,
+// so we resolve the common cases and fall back to the raw code otherwise.
+var languageNames = map[string]string{
+ "ar": "Arabic", "bg": "Bulgarian", "bn": "Bengali", "cs": "Czech",
+ "da": "Danish", "de": "German", "el": "Greek", "en": "English",
+ "es": "Spanish", "et": "Estonian", "fa": "Persian", "fi": "Finnish",
+ "fr": "French", "he": "Hebrew", "hi": "Hindi", "hr": "Croatian",
+ "hu": "Hungarian", "id": "Indonesian", "it": "Italian", "ja": "Japanese",
+ "ko": "Korean", "lt": "Lithuanian", "lv": "Latvian", "ms": "Malay",
+ "nl": "Dutch", "no": "Norwegian", "pl": "Polish", "pt": "Portuguese",
+ "ro": "Romanian", "ru": "Russian", "sk": "Slovak", "sl": "Slovenian",
+ "sr": "Serbian", "sv": "Swedish", "ta": "Tamil", "th": "Thai",
+ "tr": "Turkish", "uk": "Ukrainian", "vi": "Vietnamese", "zh": "Chinese",
+}
+
+// languageDisplayName returns a human-readable language name for a code, or the
+// trimmed code itself when unknown. An empty code yields an empty string.
+func languageDisplayName(code string) string {
+ code = strings.TrimSpace(code)
+ if code == "" {
+ return ""
+ }
+ base := strings.ToLower(code)
+ if i := strings.IndexAny(base, "-_"); i >= 0 {
+ base = base[:i]
+ }
+ if name, ok := languageNames[base]; ok {
+ return name
+ }
+ return code
+}
diff --git a/internal/subtitles/ai/notifier.go b/internal/subtitles/ai/notifier.go
new file mode 100644
index 00000000..9ec77876
--- /dev/null
+++ b/internal/subtitles/ai/notifier.go
@@ -0,0 +1,24 @@
+package ai
+
+import (
+ "context"
+
+ "github.com/Silo-Server/silo-server/internal/playback"
+)
+
+// Notifier surfaces translation progress to clients. It is optional: without
+// one, the pipeline still stores the finished track and clients pick it up on
+// their next subtitle-list refresh. The playback layer implements it over the
+// per-session realtime hub.
+//
+// SubtitleReady broadcasts a finished track to every session watching the file.
+// The Translation* methods stream a single requesting session's live job so the
+// player can pause, fill in cues as they arrive, and resume.
+type Notifier interface {
+ SubtitleReady(ctx context.Context, mediaFileID, subtitleID int, language, label string)
+
+ TranslationStarted(ctx context.Context, sessionID string, fileID int, jobID int64, trackKey, language, label string, totalCues int)
+ TranslationCues(ctx context.Context, sessionID string, fileID int, jobID int64, trackKey string, cues []playback.StreamCue, done, total int)
+ TranslationCompleted(ctx context.Context, sessionID string, fileID int, jobID int64, trackKey string, subtitleID int, language, label string)
+ TranslationFailed(ctx context.Context, sessionID string, fileID int, jobID int64, trackKey, message string)
+}
diff --git a/internal/subtitles/ai/pgrepo.go b/internal/subtitles/ai/pgrepo.go
new file mode 100644
index 00000000..3960a3d6
--- /dev/null
+++ b/internal/subtitles/ai/pgrepo.go
@@ -0,0 +1,156 @@
+package ai
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "time"
+
+ "github.com/jackc/pgx/v5"
+ "github.com/jackc/pgx/v5/pgxpool"
+)
+
+// PgJobRepository implements JobRepository on PostgreSQL.
+type PgJobRepository struct {
+ pool *pgxpool.Pool
+}
+
+// NewPgJobRepository creates a Postgres-backed job repository.
+func NewPgJobRepository(pool *pgxpool.Pool) *PgJobRepository {
+ return &PgJobRepository{pool: pool}
+}
+
+const jobColumns = `id, media_file_id, kind, source_index, source_language, target_language,
+ engine, model, status, progress, progress_message, result_subtitle_id,
+ error_message, idempotency_key, requested_by, created_at, updated_at, heartbeat_at`
+
+func scanJob(row pgx.Row) (*Job, error) {
+ var j Job
+ err := row.Scan(
+ &j.ID, &j.MediaFileID, &j.Kind, &j.SourceIndex, &j.SourceLanguage, &j.TargetLanguage,
+ &j.Engine, &j.Model, &j.Status, &j.Progress, &j.ProgressMessage, &j.ResultSubtitleID,
+ &j.ErrorMessage, &j.IdempotencyKey, &j.RequestedBy, &j.CreatedAt, &j.UpdatedAt, &j.HeartbeatAt,
+ )
+ if err != nil {
+ return nil, err
+ }
+ return &j, nil
+}
+
+func (r *PgJobRepository) InsertJob(ctx context.Context, job *Job) error {
+ if job.Engine == "" {
+ job.Engine = "openai"
+ }
+ if job.Status == "" {
+ job.Status = JobStatusPending
+ }
+ return r.pool.QueryRow(ctx,
+ `INSERT INTO subtitle_ai_jobs
+ (media_file_id, kind, source_index, source_language, target_language,
+ engine, model, status, progress, progress_message, idempotency_key, requested_by)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
+ RETURNING id, created_at, updated_at, heartbeat_at`,
+ job.MediaFileID, job.Kind, job.SourceIndex, job.SourceLanguage, job.TargetLanguage,
+ job.Engine, job.Model, job.Status, job.Progress, job.ProgressMessage, job.IdempotencyKey, job.RequestedBy,
+ ).Scan(&job.ID, &job.CreatedAt, &job.UpdatedAt, &job.HeartbeatAt)
+}
+
+func (r *PgJobRepository) GetJob(ctx context.Context, id int64) (*Job, error) {
+ job, err := scanJob(r.pool.QueryRow(ctx,
+ `SELECT `+jobColumns+` FROM subtitle_ai_jobs WHERE id = $1`, id))
+ if errors.Is(err, pgx.ErrNoRows) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, fmt.Errorf("get subtitle ai job: %w", err)
+ }
+ return job, nil
+}
+
+func (r *PgJobRepository) GetActiveJobByIdempotencyKey(ctx context.Context, key string) (*Job, error) {
+ job, err := scanJob(r.pool.QueryRow(ctx,
+ `SELECT `+jobColumns+` FROM subtitle_ai_jobs
+ WHERE idempotency_key = $1 AND status IN ('pending', 'running')
+ ORDER BY created_at DESC LIMIT 1`, key))
+ if errors.Is(err, pgx.ErrNoRows) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, fmt.Errorf("get active subtitle ai job: %w", err)
+ }
+ return job, nil
+}
+
+func (r *PgJobRepository) ListJobsByMediaFile(ctx context.Context, mediaFileID int) ([]Job, error) {
+ rows, err := r.pool.Query(ctx,
+ `SELECT `+jobColumns+` FROM subtitle_ai_jobs
+ WHERE media_file_id = $1 ORDER BY created_at DESC LIMIT 50`, mediaFileID)
+ if err != nil {
+ return nil, fmt.Errorf("list subtitle ai jobs: %w", err)
+ }
+ defer rows.Close()
+
+ var jobs []Job
+ for rows.Next() {
+ job, err := scanJob(rows)
+ if err != nil {
+ return nil, fmt.Errorf("scan subtitle ai job: %w", err)
+ }
+ jobs = append(jobs, *job)
+ }
+ return jobs, rows.Err()
+}
+
+func (r *PgJobRepository) UpdateProgress(ctx context.Context, id int64, status JobStatus, progress float64, message string) error {
+ _, err := r.pool.Exec(ctx,
+ `UPDATE subtitle_ai_jobs
+ SET status = $2, progress = $3, progress_message = $4, updated_at = now(), heartbeat_at = now()
+ WHERE id = $1`, id, status, progress, message)
+ if err != nil {
+ return fmt.Errorf("update subtitle ai job progress: %w", err)
+ }
+ return nil
+}
+
+func (r *PgJobRepository) CompleteJob(ctx context.Context, id int64, subtitleID int) error {
+ _, err := r.pool.Exec(ctx,
+ `UPDATE subtitle_ai_jobs
+ SET status = 'completed', progress = 1, result_subtitle_id = $2,
+ error_message = '', updated_at = now(), heartbeat_at = now()
+ WHERE id = $1`, id, subtitleID)
+ if err != nil {
+ return fmt.Errorf("complete subtitle ai job: %w", err)
+ }
+ return nil
+}
+
+func (r *PgJobRepository) FailJob(ctx context.Context, id int64, status JobStatus, message string) error {
+ _, err := r.pool.Exec(ctx,
+ `UPDATE subtitle_ai_jobs
+ SET status = $2, error_message = $3, updated_at = now(), heartbeat_at = now()
+ WHERE id = $1`, id, status, message)
+ if err != nil {
+ return fmt.Errorf("fail subtitle ai job: %w", err)
+ }
+ return nil
+}
+
+func (r *PgJobRepository) Heartbeat(ctx context.Context, id int64) error {
+ _, err := r.pool.Exec(ctx,
+ `UPDATE subtitle_ai_jobs SET heartbeat_at = now() WHERE id = $1`, id)
+ if err != nil {
+ return fmt.Errorf("heartbeat subtitle ai job: %w", err)
+ }
+ return nil
+}
+
+func (r *PgJobRepository) ResetStaleJobs(ctx context.Context, before time.Time, message string) (int64, error) {
+ tag, err := r.pool.Exec(ctx,
+ `UPDATE subtitle_ai_jobs
+ SET status = 'failed', error_message = $1, updated_at = now()
+ WHERE status IN ('pending', 'running') AND heartbeat_at < $2`, message, before)
+ if err != nil {
+ return 0, fmt.Errorf("reset stale subtitle ai jobs: %w", err)
+ }
+ return tag.RowsAffected(), nil
+}
diff --git a/internal/subtitles/ai/repo.go b/internal/subtitles/ai/repo.go
new file mode 100644
index 00000000..7ae8d192
--- /dev/null
+++ b/internal/subtitles/ai/repo.go
@@ -0,0 +1,25 @@
+package ai
+
+import (
+ "context"
+ "time"
+)
+
+// JobRepository persists AI subtitle jobs.
+type JobRepository interface {
+ InsertJob(ctx context.Context, job *Job) error
+ GetJob(ctx context.Context, id int64) (*Job, error)
+ // GetActiveJobByIdempotencyKey returns a pending/running job with the given
+ // key, or nil if none exists.
+ GetActiveJobByIdempotencyKey(ctx context.Context, key string) (*Job, error)
+ ListJobsByMediaFile(ctx context.Context, mediaFileID int) ([]Job, error)
+ UpdateProgress(ctx context.Context, id int64, status JobStatus, progress float64, message string) error
+ CompleteJob(ctx context.Context, id int64, subtitleID int) error
+ FailJob(ctx context.Context, id int64, status JobStatus, message string) error
+ Heartbeat(ctx context.Context, id int64) error
+ // ResetStaleJobs marks pending/running jobs whose heartbeat predates `before`
+ // as failed with the given message, clearing jobs orphaned by a crashed
+ // worker while leaving alone a job a live worker is still heartbeating.
+ // Returns the number of rows reset.
+ ResetStaleJobs(ctx context.Context, before time.Time, message string) (int64, error)
+}
diff --git a/internal/subtitles/ai/service.go b/internal/subtitles/ai/service.go
new file mode 100644
index 00000000..e7d773f9
--- /dev/null
+++ b/internal/subtitles/ai/service.go
@@ -0,0 +1,558 @@
+package ai
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log/slog"
+ "os"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/Silo-Server/silo-server/internal/models"
+ "github.com/Silo-Server/silo-server/internal/playback"
+ "github.com/Silo-Server/silo-server/internal/subtitles"
+)
+
+const providerTranslated = "translated"
+
+const (
+ // A running job refreshes its heartbeat every heartbeatInterval; one whose
+ // heartbeat has not advanced for staleJobThreshold is treated as orphaned by
+ // a crashed worker and reaped. The margin over heartbeatInterval avoids
+ // reaping a job that is merely mid–LLM-call.
+ heartbeatInterval = 30 * time.Second
+ staleJobThreshold = 2 * time.Minute
+ // How often the background reaper scans for orphaned jobs.
+ reaperInterval = time.Minute
+)
+
+var (
+ // ErrEngineNotConfigured is returned when translation is requested but the
+ // engine is disabled or missing required settings.
+ ErrEngineNotConfigured = errors.New("subtitle AI engine is not configured")
+ // ErrInvalidRequest wraps caller-input validation failures (e.g. an invalid
+ // target language) so handlers can map them to 400 rather than 500.
+ ErrInvalidRequest = errors.New("invalid translation request")
+ // ErrJobNotFound is returned for unknown job IDs.
+ ErrJobNotFound = errors.New("subtitle ai job not found")
+ // ErrSourceUnsupported is returned when the chosen source track cannot be
+ // translated (bitmap track, or a styled/unsupported format in this version).
+ ErrSourceUnsupported = errors.New("subtitle source is not supported for translation")
+)
+
+// MediaFileResolver loads a media file (path + subtitle metadata) by ID.
+type MediaFileResolver interface {
+ GetByID(ctx context.Context, id int) (*models.MediaFile, error)
+}
+
+// SubtitleStore stores a generated subtitle and reads existing stored ones.
+// Satisfied by *subtitles.Manager.
+type SubtitleStore interface {
+ StoreSubtitle(ctx context.Context, req subtitles.StoreSubtitleRequest) (*subtitles.DownloadedSubtitle, error)
+ GetSubtitleContent(ctx context.Context, id int) (*subtitles.DownloadedSubtitle, []byte, error)
+}
+
+// SubtitleLister lists stored subtitles for a media file, used to resolve a
+// downloaded-source track by its position. Satisfied by subtitles.Repository.
+type SubtitleLister interface {
+ ListDownloadedSubtitles(ctx context.Context, mediaFileID int) ([]subtitles.DownloadedSubtitle, error)
+}
+
+// Service owns the AI subtitle job lifecycle: enqueue, bounded concurrent
+// execution, progress/heartbeat, cancellation, and restart recovery.
+type Service struct {
+ // baseCtx is the application context; dispatched jobs and the reaper derive
+ // from it so they stop when the server shuts down.
+ baseCtx context.Context
+ cfg Config
+ repo JobRepository
+ translator Translator
+ store SubtitleStore
+ lister SubtitleLister
+ files MediaFileResolver
+ notifier Notifier // optional
+ ffmpegPath string
+ logger *slog.Logger
+
+ sem chan struct{}
+ mu sync.Mutex
+ cancels map[int64]context.CancelFunc
+ wg sync.WaitGroup
+}
+
+// NewService wires a translation service. notifier may be nil. appCtx is the
+// application lifecycle context; jobs and the reaper derive from it so they stop
+// on shutdown. A nil appCtx falls back to context.Background().
+func NewService(
+ appCtx context.Context,
+ cfg Config,
+ repo JobRepository,
+ translator Translator,
+ store SubtitleStore,
+ lister SubtitleLister,
+ files MediaFileResolver,
+ notifier Notifier,
+ ffmpegPath string,
+ logger *slog.Logger,
+) *Service {
+ maxConcurrent := cfg.MaxConcurrentJobs
+ if maxConcurrent <= 0 {
+ maxConcurrent = 2
+ }
+ if logger == nil {
+ logger = slog.Default()
+ }
+ if appCtx == nil {
+ appCtx = context.Background()
+ }
+ return &Service{
+ baseCtx: appCtx,
+ cfg: cfg,
+ repo: repo,
+ translator: translator,
+ store: store,
+ lister: lister,
+ files: files,
+ notifier: notifier,
+ ffmpegPath: ffmpegPath,
+ logger: logger,
+ sem: make(chan struct{}, maxConcurrent),
+ cancels: make(map[int64]context.CancelFunc),
+ }
+}
+
+// Enabled reports whether translation can currently run.
+func (s *Service) Enabled() bool { return s.cfg.Ready() }
+
+// Recover clears jobs orphaned by a crashed worker and starts a background
+// reaper that keeps doing so. Reaping is heartbeat-based (not "every active
+// job"), so it is safe when multiple instances share one database: a job still
+// being heartbeat-updated by a live worker is never reset. Call once at startup;
+// jobs and the reaper derive from the application context passed to NewService,
+// so they stop on shutdown.
+func (s *Service) Recover() {
+ s.reapStaleJobs()
+ go s.reaperLoop()
+}
+
+// reaperLoop periodically reaps orphaned jobs until the application context is
+// cancelled (server shutdown).
+func (s *Service) reaperLoop() {
+ ticker := time.NewTicker(reaperInterval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-s.baseCtx.Done():
+ return
+ case <-ticker.C:
+ s.reapStaleJobs()
+ }
+ }
+}
+
+// reapStaleJobs fails any pending/running job whose heartbeat has not advanced
+// within staleJobThreshold.
+func (s *Service) reapStaleJobs() {
+ before := time.Now().Add(-staleJobThreshold)
+ n, err := s.repo.ResetStaleJobs(context.WithoutCancel(s.baseCtx), before, "interrupted by server restart")
+ if err != nil {
+ s.logger.Warn("failed to reset stale subtitle ai jobs", "error", err)
+ return
+ }
+ if n > 0 {
+ s.logger.Info("reset stale subtitle ai jobs", "count", n)
+ }
+}
+
+// Enqueue validates and queues a job, returning immediately. If an identical
+// job is already pending or running, that job is returned instead of a new one.
+func (s *Service) Enqueue(ctx context.Context, req JobRequest) (*Job, error) {
+ if !s.cfg.Ready() {
+ return nil, ErrEngineNotConfigured
+ }
+ if req.Kind == "" {
+ req.Kind = JobKindTranslate
+ }
+
+ target, err := subtitles.NormalizeLanguageCode(req.TargetLanguage)
+ if err != nil {
+ return nil, fmt.Errorf("%w: invalid target language %q", ErrInvalidRequest, req.TargetLanguage)
+ }
+ req.TargetLanguage = target
+
+ key := idempotencyKey(req.MediaFileID, req.Kind, req.SourceIndex, req.TargetLanguage, s.cfg.ChatModel)
+ if existing, err := s.repo.GetActiveJobByIdempotencyKey(ctx, key); err != nil {
+ return nil, err
+ } else if existing != nil {
+ return existing, nil
+ }
+
+ job := &Job{
+ MediaFileID: req.MediaFileID,
+ Kind: req.Kind,
+ SourceIndex: req.SourceIndex,
+ SourceLanguage: req.SourceLanguage,
+ TargetLanguage: req.TargetLanguage,
+ Engine: "openai",
+ Model: s.cfg.ChatModel,
+ Status: JobStatusPending,
+ ProgressMessage: "Queued",
+ IdempotencyKey: key,
+ RequestedBy: req.RequestedBy,
+ SessionID: req.SessionID,
+ StartPosition: req.StartPosition,
+ }
+ if err := s.repo.InsertJob(ctx, job); err != nil {
+ // A racing duplicate trips the partial unique index; return the winner.
+ if existing, lookupErr := s.repo.GetActiveJobByIdempotencyKey(ctx, key); lookupErr == nil && existing != nil {
+ return existing, nil
+ }
+ return nil, err
+ }
+
+ s.dispatch(*job)
+ return job, nil
+}
+
+// GetJob returns a job by ID.
+func (s *Service) GetJob(ctx context.Context, id int64) (*Job, error) {
+ job, err := s.repo.GetJob(ctx, id)
+ if err != nil {
+ return nil, err
+ }
+ if job == nil {
+ return nil, ErrJobNotFound
+ }
+ return job, nil
+}
+
+// ListJobs returns recent jobs for a media file.
+func (s *Service) ListJobs(ctx context.Context, mediaFileID int) ([]Job, error) {
+ return s.repo.ListJobsByMediaFile(ctx, mediaFileID)
+}
+
+// Cancel requests cancellation of a job.
+func (s *Service) Cancel(ctx context.Context, id int64) error {
+ job, err := s.repo.GetJob(ctx, id)
+ if err != nil {
+ return err
+ }
+ if job == nil {
+ return ErrJobNotFound
+ }
+
+ s.mu.Lock()
+ cancel := s.cancels[id]
+ s.mu.Unlock()
+
+ if cancel != nil {
+ cancel()
+ return nil
+ }
+ // No in-flight goroutine (e.g. another node, or never started): best-effort
+ // terminal transition if it is still active.
+ if !job.Status.Terminal() {
+ return s.repo.FailJob(ctx, id, JobStatusCancelled, "cancelled")
+ }
+ return nil
+}
+
+// dispatch launches a bounded background goroutine to run the job.
+func (s *Service) dispatch(job Job) {
+ // Derive from the application context so a server shutdown cancels in-flight
+ // translations (the per-job cancel still allows user-initiated cancellation).
+ runCtx, cancel := context.WithCancel(s.baseCtx)
+ s.mu.Lock()
+ s.cancels[job.ID] = cancel
+ s.mu.Unlock()
+
+ s.wg.Add(1)
+ go func() {
+ defer s.wg.Done()
+ defer func() {
+ s.mu.Lock()
+ delete(s.cancels, job.ID)
+ s.mu.Unlock()
+ cancel()
+ }()
+
+ // Bound concurrency so translation never starves transcodes.
+ select {
+ case s.sem <- struct{}{}:
+ case <-runCtx.Done():
+ _ = s.repo.FailJob(context.Background(), job.ID, JobStatusCancelled, "cancelled before start")
+ return
+ }
+ defer func() { <-s.sem }()
+
+ s.run(runCtx, &job)
+ }()
+}
+
+func (s *Service) run(ctx context.Context, job *Job) {
+ // Keep heartbeat_at fresh while the job runs (progress updates also bump it),
+ // so the stale-job reaper never resets a job that is still alive during a long
+ // single LLM call.
+ stopHeartbeat := make(chan struct{})
+ defer close(stopHeartbeat)
+ go func() {
+ ticker := time.NewTicker(heartbeatInterval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-stopHeartbeat:
+ return
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ _ = s.repo.Heartbeat(context.WithoutCancel(ctx), job.ID)
+ }
+ }
+ }()
+
+ if err := s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, 0, "Loading subtitle"); err != nil {
+ s.logger.Warn("failed to mark subtitle ai job running", "job", job.ID, "error", err)
+ }
+
+ cues, sourceLang, err := s.loadSource(ctx, job)
+ if err != nil {
+ s.finishWithError(ctx, job, err)
+ return
+ }
+ if job.SourceLanguage == "" {
+ job.SourceLanguage = sourceLang
+ }
+
+ releaseName := translatedReleaseName(job.SourceLanguage, job.TargetLanguage)
+ streaming := job.SessionID != "" && s.notifier != nil
+ trackKey := liveTrackKey(job.ID)
+ if streaming {
+ s.notifier.TranslationStarted(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey,
+ job.TargetLanguage, releaseName, len(cues))
+ }
+
+ // Translate from the viewer's playhead forward (then wrap to the start), so
+ // the region they're watching fills first. Cues carry absolute timing, so
+ // the player places streamed cues correctly regardless of arrival order.
+ ordered := reorderFromPosition(cues, job.StartPosition)
+
+ translated, err := s.translator.Translate(ctx, TranslateRequest{
+ Cues: ordered,
+ SourceLanguage: job.SourceLanguage,
+ TargetLanguage: job.TargetLanguage,
+ }, func(batch []SubtitleCue, done, total int) {
+ // Map cue progress into the 5%..95% band; push the batch live.
+ _ = s.repo.UpdateProgress(ctx, job.ID, JobStatusRunning, 0.05+0.9*float64(done)/float64(total), "Translating")
+ if streaming {
+ s.notifier.TranslationCues(ctx, job.SessionID, job.MediaFileID, job.ID, trackKey,
+ toStreamCues(batch), done, total)
+ }
+ })
+ if err != nil {
+ s.finishWithError(ctx, job, err)
+ return
+ }
+
+ // A finished translation should not be thrown away by a last-moment cancel.
+ storeCtx := context.WithoutCancel(ctx)
+
+ // Persist in chronological order regardless of the playhead-first order used
+ // for translation/streaming.
+ sortCuesByStart(translated)
+ sub, err := s.store.StoreSubtitle(storeCtx, subtitles.StoreSubtitleRequest{
+ MediaFileID: job.MediaFileID,
+ UserID: job.RequestedBy,
+ Provider: providerTranslated,
+ Language: job.TargetLanguage,
+ Format: subtitles.FormatSRT,
+ ReleaseName: releaseName,
+ Data: SerializeSRT(translated),
+ })
+ if err != nil {
+ s.finishWithError(ctx, job, fmt.Errorf("store translated subtitle: %w", err))
+ return
+ }
+
+ if err := s.repo.CompleteJob(storeCtx, job.ID, sub.ID); err != nil {
+ s.logger.Warn("failed to mark subtitle ai job complete", "job", job.ID, "error", err)
+ }
+
+ if s.notifier != nil {
+ if streaming {
+ s.notifier.TranslationCompleted(storeCtx, job.SessionID, job.MediaFileID, job.ID, trackKey,
+ sub.ID, job.TargetLanguage, releaseName)
+ }
+ // Broadcast to other sessions watching this file (a no-op for the
+ // streaming requester, which already tracks its live track).
+ s.notifier.SubtitleReady(storeCtx, job.MediaFileID, sub.ID, job.TargetLanguage, releaseName)
+ }
+}
+
+func (s *Service) finishWithError(ctx context.Context, job *Job, err error) {
+ status := JobStatusFailed
+ msg := truncate(err.Error(), 500)
+ // Only a genuine cancellation (user cancel, or shutdown via cancel) becomes
+ // "cancelled". A deadline/timeout (context.DeadlineExceeded) stays "failed".
+ if errors.Is(err, context.Canceled) || errors.Is(ctx.Err(), context.Canceled) {
+ status = JobStatusCancelled
+ msg = "cancelled"
+ }
+ if dbErr := s.repo.FailJob(context.WithoutCancel(ctx), job.ID, status, msg); dbErr != nil {
+ s.logger.Warn("failed to record subtitle ai job failure", "job", job.ID, "error", dbErr)
+ }
+ if job.SessionID != "" && s.notifier != nil {
+ s.notifier.TranslationFailed(context.WithoutCancel(ctx), job.SessionID, job.MediaFileID, job.ID,
+ liveTrackKey(job.ID), msg)
+ }
+ if status == JobStatusFailed {
+ s.logger.Warn("subtitle ai job failed", "job", job.ID, "media_file", job.MediaFileID, "error", err)
+ }
+}
+
+// loadSource resolves the combined player subtitle index to translatable cues,
+// mirroring the index space used by the playback subtitle endpoints
+// (external → embedded → downloaded). Embedded tracks are extracted to SRT via
+// ffmpeg (so any non-bitmap codec works); external/downloaded sources must be a
+// text format (SRT/VTT) in this version.
+func (s *Service) loadSource(ctx context.Context, job *Job) ([]SubtitleCue, string, error) {
+ file, err := s.files.GetByID(ctx, job.MediaFileID)
+ if err != nil {
+ return nil, "", fmt.Errorf("load media file: %w", err)
+ }
+ if file == nil {
+ return nil, "", fmt.Errorf("media file not found")
+ }
+
+ idx := job.SourceIndex
+ externalCount := len(file.ExternalSubtitles)
+
+ switch {
+ case idx < 0:
+ return nil, "", fmt.Errorf("invalid source subtitle index")
+
+ case idx < externalCount:
+ ext := file.ExternalSubtitles[idx]
+ if !isParsableTextFormat(ext.Format) {
+ return nil, "", fmt.Errorf("%w: external %s", ErrSourceUnsupported, ext.Format)
+ }
+ data, err := os.ReadFile(ext.Path)
+ if err != nil {
+ return nil, "", fmt.Errorf("read external subtitle: %w", err)
+ }
+ cues, err := ParseCues(data)
+ if err != nil {
+ return nil, "", err
+ }
+ return cues, ext.Language, nil
+
+ case idx < externalCount+len(file.SubtitleTracks):
+ embeddedIndex := idx - externalCount
+ track := file.SubtitleTracks[embeddedIndex]
+ if playback.NeedsBurnIn(track.Codec) {
+ return nil, "", fmt.Errorf("%w: bitmap track", ErrSourceUnsupported)
+ }
+ data, _, err := playback.ExtractSubtitle(ctx, file.FilePath, embeddedIndex, s.ffmpegPath)
+ if err != nil {
+ return nil, "", fmt.Errorf("extract embedded subtitle: %w", err)
+ }
+ cues, err := ParseCues(data)
+ if err != nil {
+ return nil, "", err
+ }
+ return cues, track.Language, nil
+
+ default:
+ downloadedIndex := idx - externalCount - len(file.SubtitleTracks)
+ list, err := s.lister.ListDownloadedSubtitles(ctx, file.ID)
+ if err != nil {
+ return nil, "", fmt.Errorf("list downloaded subtitles: %w", err)
+ }
+ if downloadedIndex < 0 || downloadedIndex >= len(list) {
+ return nil, "", fmt.Errorf("source subtitle index out of range")
+ }
+ dl := list[downloadedIndex]
+ if !isParsableTextFormat(string(dl.Format)) {
+ return nil, "", fmt.Errorf("%w: downloaded %s", ErrSourceUnsupported, dl.Format)
+ }
+ _, data, err := s.store.GetSubtitleContent(ctx, dl.ID)
+ if err != nil {
+ return nil, "", fmt.Errorf("fetch source subtitle: %w", err)
+ }
+ cues, err := ParseCues(data)
+ if err != nil {
+ return nil, "", err
+ }
+ return cues, dl.Language, nil
+ }
+}
+
+func isParsableTextFormat(format string) bool {
+ switch strings.ToLower(strings.TrimSpace(format)) {
+ case "srt", "subrip", "vtt", "webvtt":
+ return true
+ default:
+ return false
+ }
+}
+
+func translatedReleaseName(sourceLang, targetLang string) string {
+ src := languageDisplayName(sourceLang)
+ if src == "" {
+ src = "Original"
+ }
+ tgt := languageDisplayName(targetLang)
+ if tgt == "" {
+ tgt = targetLang
+ }
+ return fmt.Sprintf("%s → %s (AI)", src, tgt)
+}
+
+// liveTrackKey is the stable client-side identifier for a job's live track.
+func liveTrackKey(jobID int64) string {
+ return fmt.Sprintf("ai-%d", jobID)
+}
+
+// reorderFromPosition rotates chronological cues so the first cue still visible
+// at startSeconds leads, with earlier cues appended after the end. This makes a
+// live translation fill the viewer's current region first. Returns the input
+// unchanged when there's no useful pivot.
+func reorderFromPosition(cues []SubtitleCue, startSeconds float64) []SubtitleCue {
+ if startSeconds <= 0 || len(cues) < 2 {
+ return cues
+ }
+ start := time.Duration(startSeconds * float64(time.Second))
+ pivot := -1
+ for i, c := range cues {
+ if c.End >= start {
+ pivot = i
+ break
+ }
+ }
+ if pivot <= 0 {
+ return cues
+ }
+ out := make([]SubtitleCue, 0, len(cues))
+ out = append(out, cues[pivot:]...)
+ out = append(out, cues[:pivot]...)
+ return out
+}
+
+// toStreamCues converts cues to the realtime wire form (absolute seconds).
+func toStreamCues(cues []SubtitleCue) []playback.StreamCue {
+ out := make([]playback.StreamCue, 0, len(cues))
+ for _, c := range cues {
+ out = append(out, playback.StreamCue{
+ Start: c.Start.Seconds(),
+ End: c.End.Seconds(),
+ Text: strings.Join(c.Lines, "\n"),
+ })
+ }
+ return out
+}
+
+func sortCuesByStart(cues []SubtitleCue) {
+ sort.SliceStable(cues, func(i, j int) bool { return cues[i].Start < cues[j].Start })
+}
diff --git a/internal/subtitles/ai/service_test.go b/internal/subtitles/ai/service_test.go
new file mode 100644
index 00000000..2fceefb0
--- /dev/null
+++ b/internal/subtitles/ai/service_test.go
@@ -0,0 +1,65 @@
+package ai
+
+import (
+ "context"
+ "sync"
+ "testing"
+ "time"
+)
+
+// recordingRepo is a JobRepository that records ResetStaleJobs calls and no-ops
+// everything else, so the recovery/reaper behavior can be tested in isolation.
+type recordingRepo struct {
+ mu sync.Mutex
+ resets int
+ lastBefore time.Time
+}
+
+func (r *recordingRepo) InsertJob(context.Context, *Job) error { return nil }
+func (r *recordingRepo) GetJob(context.Context, int64) (*Job, error) { return nil, nil }
+func (r *recordingRepo) GetActiveJobByIdempotencyKey(context.Context, string) (*Job, error) {
+ return nil, nil
+}
+func (r *recordingRepo) ListJobsByMediaFile(context.Context, int) ([]Job, error) { return nil, nil }
+func (r *recordingRepo) UpdateProgress(context.Context, int64, JobStatus, float64, string) error {
+ return nil
+}
+func (r *recordingRepo) CompleteJob(context.Context, int64, int) error { return nil }
+func (r *recordingRepo) FailJob(context.Context, int64, JobStatus, string) error { return nil }
+func (r *recordingRepo) Heartbeat(context.Context, int64) error { return nil }
+
+func (r *recordingRepo) ResetStaleJobs(_ context.Context, before time.Time, _ string) (int64, error) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ r.resets++
+ r.lastBefore = before
+ return 0, nil
+}
+
+func (r *recordingRepo) snapshot() (int, time.Time) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.resets, r.lastBefore
+}
+
+// Recover reaps immediately using a heartbeat cutoff of now-staleJobThreshold,
+// not "every active job", so a live worker's jobs survive a peer's startup.
+func TestRecoverReapsStaleJobsImmediately(t *testing.T) {
+ repo := &recordingRepo{}
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel() // stops the reaper goroutine started by Recover
+
+ svc := NewService(ctx, Config{}, repo, nil, nil, nil, nil, nil, "", nil)
+
+ approxNow := time.Now()
+ svc.Recover()
+
+ resets, before := repo.snapshot()
+ if resets < 1 {
+ t.Fatalf("Recover did not reap immediately: resets=%d", resets)
+ }
+ want := approxNow.Add(-staleJobThreshold)
+ if diff := before.Sub(want); diff > 2*time.Second || diff < -2*time.Second {
+ t.Errorf("stale cutoff = %v, want ~%v (now-staleJobThreshold)", before, want)
+ }
+}
diff --git a/internal/subtitles/ai/srt.go b/internal/subtitles/ai/srt.go
new file mode 100644
index 00000000..b8d4d834
--- /dev/null
+++ b/internal/subtitles/ai/srt.go
@@ -0,0 +1,158 @@
+package ai
+
+import (
+ "bytes"
+ "fmt"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// ParseCues parses SRT or WebVTT subtitle bytes into cues. It tolerates both
+// comma (SRT) and period (VTT) millisecond separators, a leading WEBVTT header,
+// optional cue identifiers, VTT cue settings after the end timestamp, a UTF-8
+// BOM, and CRLF line endings. Malformed individual cues are skipped rather than
+// failing the whole document.
+func ParseCues(data []byte) ([]SubtitleCue, error) {
+ text := string(data)
+ text = strings.TrimPrefix(text, "\ufeff") // strip UTF-8 BOM
+ text = strings.ReplaceAll(text, "\r\n", "\n")
+ text = strings.ReplaceAll(text, "\r", "\n")
+
+ var cues []SubtitleCue
+ for _, block := range strings.Split(text, "\n\n") {
+ block = strings.Trim(block, "\n")
+ if strings.TrimSpace(block) == "" {
+ continue
+ }
+ lines := strings.Split(block, "\n")
+
+ timingIdx := -1
+ for i, l := range lines {
+ if strings.Contains(l, "-->") {
+ timingIdx = i
+ break
+ }
+ }
+ if timingIdx < 0 {
+ continue // WEBVTT header, NOTE/STYLE blocks, etc.
+ }
+
+ start, end, err := parseTimingLine(lines[timingIdx])
+ if err != nil {
+ continue
+ }
+
+ textLines := append([]string(nil), lines[timingIdx+1:]...)
+ for len(textLines) > 0 && strings.TrimSpace(textLines[len(textLines)-1]) == "" {
+ textLines = textLines[:len(textLines)-1]
+ }
+ if len(textLines) == 0 {
+ continue
+ }
+ cues = append(cues, SubtitleCue{Start: start, End: end, Lines: textLines})
+ }
+
+ if len(cues) == 0 {
+ return nil, fmt.Errorf("no subtitle cues found")
+ }
+ return cues, nil
+}
+
+// SerializeSRT renders cues as a well-formed SRT document. Output always uses
+// comma millisecond separators; the serving layer converts SRT to VTT on read.
+func SerializeSRT(cues []SubtitleCue) []byte {
+ var buf bytes.Buffer
+ for i, cue := range cues {
+ fmt.Fprintf(&buf, "%d\n%s --> %s\n", i+1, formatSRTTimestamp(cue.Start), formatSRTTimestamp(cue.End))
+ for _, line := range cue.Lines {
+ buf.WriteString(line)
+ buf.WriteByte('\n')
+ }
+ buf.WriteByte('\n')
+ }
+ return buf.Bytes()
+}
+
+func parseTimingLine(line string) (time.Duration, time.Duration, error) {
+ parts := strings.SplitN(line, "-->", 2)
+ if len(parts) != 2 {
+ return 0, 0, fmt.Errorf("invalid timing line")
+ }
+ start, err := parseTimestamp(strings.TrimSpace(parts[0]))
+ if err != nil {
+ return 0, 0, err
+ }
+ // The end timestamp may be followed by VTT cue settings (e.g. "line:0%").
+ endField := strings.TrimSpace(parts[1])
+ if sp := strings.IndexAny(endField, " \t"); sp >= 0 {
+ endField = endField[:sp]
+ }
+ end, err := parseTimestamp(endField)
+ if err != nil {
+ return 0, 0, err
+ }
+ return start, end, nil
+}
+
+// parseTimestamp accepts HH:MM:SS,mmm / HH:MM:SS.mmm / MM:SS.mmm forms.
+func parseTimestamp(s string) (time.Duration, error) {
+ s = strings.TrimSpace(strings.Replace(s, ",", ".", 1))
+ colon := strings.Split(s, ":")
+ if len(colon) < 2 || len(colon) > 3 {
+ return 0, fmt.Errorf("invalid timestamp %q", s)
+ }
+
+ var hours int
+ idx := 0
+ if len(colon) == 3 {
+ h, err := strconv.Atoi(colon[0])
+ if err != nil {
+ return 0, fmt.Errorf("invalid hours in %q", s)
+ }
+ hours = h
+ idx = 1
+ }
+
+ minutes, err := strconv.Atoi(colon[idx])
+ if err != nil {
+ return 0, fmt.Errorf("invalid minutes in %q", s)
+ }
+
+ secFrac := strings.SplitN(colon[idx+1], ".", 2)
+ seconds, err := strconv.Atoi(secFrac[0])
+ if err != nil {
+ return 0, fmt.Errorf("invalid seconds in %q", s)
+ }
+
+ millis := 0
+ if len(secFrac) == 2 {
+ frac := secFrac[1]
+ for len(frac) < 3 {
+ frac += "0"
+ }
+ millis, err = strconv.Atoi(frac[:3])
+ if err != nil {
+ return 0, fmt.Errorf("invalid milliseconds in %q", s)
+ }
+ }
+
+ return time.Duration(hours)*time.Hour +
+ time.Duration(minutes)*time.Minute +
+ time.Duration(seconds)*time.Second +
+ time.Duration(millis)*time.Millisecond, nil
+}
+
+func formatSRTTimestamp(d time.Duration) string {
+ if d < 0 {
+ d = 0
+ }
+ totalMS := int64(d / time.Millisecond)
+ ms := totalMS % 1000
+ totalSec := totalMS / 1000
+ s := totalSec % 60
+ totalMin := totalSec / 60
+ m := totalMin % 60
+ h := totalMin / 60
+ return fmt.Sprintf("%02d:%02d:%02d,%03d", h, m, s, ms)
+}
diff --git a/internal/subtitles/ai/srt_test.go b/internal/subtitles/ai/srt_test.go
new file mode 100644
index 00000000..fd4f8dbd
--- /dev/null
+++ b/internal/subtitles/ai/srt_test.go
@@ -0,0 +1,79 @@
+package ai
+
+import (
+ "testing"
+ "time"
+)
+
+func TestParseCuesSRT(t *testing.T) {
+ in := "1\n00:00:01,000 --> 00:00:02,500\nHello world\n\n" +
+ "2\n00:00:03,000 --> 00:00:04,000\nLine one\nLine two\n"
+
+ cues, err := ParseCues([]byte(in))
+ if err != nil {
+ t.Fatalf("ParseCues: %v", err)
+ }
+ if len(cues) != 2 {
+ t.Fatalf("got %d cues, want 2", len(cues))
+ }
+ if cues[0].Start != time.Second || cues[0].End != 2500*time.Millisecond {
+ t.Errorf("cue 0 timing = %v..%v", cues[0].Start, cues[0].End)
+ }
+ if len(cues[0].Lines) != 1 || cues[0].Lines[0] != "Hello world" {
+ t.Errorf("cue 0 lines = %#v", cues[0].Lines)
+ }
+ if len(cues[1].Lines) != 2 {
+ t.Errorf("cue 1 lines = %#v", cues[1].Lines)
+ }
+}
+
+func TestParseCuesVTT(t *testing.T) {
+ in := "WEBVTT\n\n" +
+ "00:00:01.000 --> 00:00:02.500 line:80%\nHello\n\n" +
+ "NOTE a comment block\n\n" +
+ "00:01:00.000 --> 00:01:01.000\nWorld\n"
+
+ cues, err := ParseCues([]byte(in))
+ if err != nil {
+ t.Fatalf("ParseCues: %v", err)
+ }
+ if len(cues) != 2 {
+ t.Fatalf("got %d cues, want 2 (NOTE block must be skipped)", len(cues))
+ }
+ if cues[0].Start != time.Second || cues[0].End != 2500*time.Millisecond {
+ t.Errorf("cue 0 timing = %v..%v", cues[0].Start, cues[0].End)
+ }
+ if cues[1].Start != time.Minute {
+ t.Errorf("cue 1 start = %v, want 1m", cues[1].Start)
+ }
+}
+
+func TestSerializeRoundTrip(t *testing.T) {
+ orig := []SubtitleCue{
+ {Start: time.Second, End: 2500 * time.Millisecond, Lines: []string{"Hello world"}},
+ {Start: 3 * time.Second, End: 4 * time.Second, Lines: []string{"Line one", "Line two"}},
+ }
+
+ reparsed, err := ParseCues(SerializeSRT(orig))
+ if err != nil {
+ t.Fatalf("reparse: %v", err)
+ }
+ if len(reparsed) != len(orig) {
+ t.Fatalf("round-trip changed cue count: got %d, want %d", len(reparsed), len(orig))
+ }
+ for i := range orig {
+ if reparsed[i].Start != orig[i].Start || reparsed[i].End != orig[i].End {
+ t.Errorf("cue %d timing drifted: %v..%v vs %v..%v",
+ i, reparsed[i].Start, reparsed[i].End, orig[i].Start, orig[i].End)
+ }
+ if len(reparsed[i].Lines) != len(orig[i].Lines) {
+ t.Errorf("cue %d line count changed: %#v", i, reparsed[i].Lines)
+ }
+ }
+}
+
+func TestParseCuesEmpty(t *testing.T) {
+ if _, err := ParseCues([]byte("not a subtitle file")); err == nil {
+ t.Error("expected error for input with no cues")
+ }
+}
diff --git a/internal/subtitles/ai/translator.go b/internal/subtitles/ai/translator.go
new file mode 100644
index 00000000..adda62a7
--- /dev/null
+++ b/internal/subtitles/ai/translator.go
@@ -0,0 +1,223 @@
+package ai
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "strconv"
+ "strings"
+)
+
+// LLMTranslator translates subtitle cues with an OpenAI-compatible chat model.
+//
+// Cues are translated in batches. Each batch is sent as a JSON object keyed by
+// cue number; the model must return the same keys with translated values. Only
+// the text is sent to the model — timestamps never leave the server — so timing
+// alignment is structurally guaranteed. A few preceding source cues are
+// included as untranslated context so the model can keep scene continuity
+// across batch boundaries.
+type LLMTranslator struct {
+ client *Client
+ batchSize int
+ contextNeighbors int
+ maxRetries int
+}
+
+// NewLLMTranslator builds a translator. batchSize and contextNeighbors fall back
+// to sane defaults when non-positive.
+func NewLLMTranslator(client *Client, batchSize, contextNeighbors int) *LLMTranslator {
+ if batchSize <= 0 {
+ batchSize = 40
+ }
+ if contextNeighbors < 0 {
+ contextNeighbors = 0
+ }
+ return &LLMTranslator{
+ client: client,
+ batchSize: batchSize,
+ contextNeighbors: contextNeighbors,
+ maxRetries: 2,
+ }
+}
+
+// Translate implements Translator.
+func (t *LLMTranslator) Translate(ctx context.Context, req TranslateRequest, onBatch func(batch []SubtitleCue, done, total int)) ([]SubtitleCue, error) {
+ total := len(req.Cues)
+ if total == 0 {
+ return nil, fmt.Errorf("no cues to translate")
+ }
+
+ // Preserve timing by copying input cues and only replacing Lines.
+ out := make([]SubtitleCue, total)
+ copy(out, req.Cues)
+
+ srcName := languageDisplayName(req.SourceLanguage)
+ tgtName := languageDisplayName(req.TargetLanguage)
+ system := translationSystemPrompt(srcName, tgtName)
+
+ for start := 0; start < total; start += t.batchSize {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ end := min(start+t.batchSize, total)
+
+ contextStart := max(0, start-t.contextNeighbors)
+ translated, err := t.translateBatch(ctx, system, tgtName, req.Cues[contextStart:start], req.Cues[start:end])
+ if err != nil {
+ return nil, fmt.Errorf("translate cues %d-%d: %w", start+1, end, err)
+ }
+ for i, lines := range translated {
+ out[start+i].Lines = lines
+ }
+ if onBatch != nil {
+ onBatch(out[start:end], end, total)
+ }
+ }
+
+ return out, nil
+}
+
+func (t *LLMTranslator) translateBatch(ctx context.Context, system, targetName string, contextCues, batch []SubtitleCue) ([][]string, error) {
+ texts := make([]string, len(batch))
+ for i, c := range batch {
+ texts[i] = strings.Join(c.Lines, "\n")
+ }
+ payload, err := buildIndexedJSON(texts)
+ if err != nil {
+ return nil, err
+ }
+
+ var user strings.Builder
+ if len(contextCues) > 0 {
+ user.WriteString("Preceding lines for context only — do not translate or include them in your output:\n")
+ for _, c := range contextCues {
+ user.WriteString(strings.Join(c.Lines, " "))
+ user.WriteByte('\n')
+ }
+ user.WriteByte('\n')
+ }
+ fmt.Fprintf(&user, "Translate these %d cues into %s. Respond with only the JSON object:\n%s", len(batch), targetName, payload)
+
+ messages := []chatMessage{
+ {Role: "system", Content: system},
+ {Role: "user", Content: user.String()},
+ }
+
+ var lastErr error
+ for attempt := 0; attempt <= t.maxRetries; attempt++ {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ content, err := t.client.chat(ctx, messages, true)
+ if err != nil {
+ return nil, err // transport/API errors are already retried inside chat
+ }
+
+ obj, err := extractJSONObject(content)
+ if err != nil {
+ lastErr = err
+ continue
+ }
+ var m map[string]string
+ if err := json.Unmarshal([]byte(obj), &m); err != nil {
+ lastErr = fmt.Errorf("decode translation JSON: %w", err)
+ continue
+ }
+
+ out := make([][]string, len(batch))
+ complete := true
+ for i := range batch {
+ v, ok := m[strconv.Itoa(i+1)]
+ if !ok {
+ complete = false
+ break
+ }
+ out[i] = splitCueLines(v)
+ }
+ if !complete {
+ lastErr = fmt.Errorf("model omitted one or more cues")
+ continue
+ }
+ return out, nil
+ }
+
+ return nil, fmt.Errorf("invalid model response after %d attempts: %w", t.maxRetries+1, lastErr)
+}
+
+func translationSystemPrompt(srcName, tgtName string) string {
+ src := srcName
+ if src == "" {
+ src = "the source language"
+ }
+ return fmt.Sprintf(
+ "You are a professional subtitle translator. Translate subtitle cues from %s into %s. "+
+ "Produce natural, idiomatic %s that preserves meaning, tone, register, and proper nouns. "+
+ "You receive a JSON object whose keys are cue numbers and whose values are the source text. "+
+ "Respond with ONLY a JSON object using the exact same keys, where each value is the translation of that cue. "+
+ "Preserve line breaks within a cue as \\n. Do not add, remove, merge, split, reorder, or renumber cues, "+
+ "and do not output anything except the JSON object.",
+ src, tgtName, tgtName,
+ )
+}
+
+// buildIndexedJSON renders texts as a JSON object {"1":..., "2":...} keyed by
+// 1-based cue number, escaping each value safely. It is built by hand rather
+// than json.Marshal'ing a map so the keys stay in numeric order — that reads
+// more naturally for the model than the lexicographic order Go emits for maps
+// ("1","10","11",...,"2"). Correctness doesn't depend on order (results are
+// mapped back by key), but ordered input gives the model better scene context.
+func buildIndexedJSON(texts []string) (string, error) {
+ var b strings.Builder
+ b.WriteByte('{')
+ for i, text := range texts {
+ if i > 0 {
+ b.WriteByte(',')
+ }
+ key, err := json.Marshal(strconv.Itoa(i + 1))
+ if err != nil {
+ return "", err
+ }
+ val, err := json.Marshal(text)
+ if err != nil {
+ return "", err
+ }
+ b.Write(key)
+ b.WriteByte(':')
+ b.Write(val)
+ }
+ b.WriteByte('}')
+ return b.String(), nil
+}
+
+// extractJSONObject pulls the first balanced-looking JSON object out of a model
+// reply, tolerating ``` code fences and surrounding prose.
+func extractJSONObject(s string) (string, error) {
+ s = strings.TrimSpace(s)
+ if strings.HasPrefix(s, "```") {
+ s = strings.TrimPrefix(s, "```")
+ if nl := strings.IndexByte(s, '\n'); nl >= 0 {
+ s = s[nl+1:]
+ }
+ if idx := strings.LastIndex(s, "```"); idx >= 0 {
+ s = s[:idx]
+ }
+ s = strings.TrimSpace(s)
+ }
+ start := strings.IndexByte(s, '{')
+ end := strings.LastIndexByte(s, '}')
+ if start < 0 || end < 0 || end < start {
+ return "", fmt.Errorf("no JSON object found in model response")
+ }
+ return s[start : end+1], nil
+}
+
+func splitCueLines(v string) []string {
+ lines := strings.Split(v, "\n")
+ for len(lines) > 0 && strings.TrimSpace(lines[len(lines)-1]) == "" {
+ lines = lines[:len(lines)-1]
+ }
+ if len(lines) == 0 {
+ return []string{""}
+ }
+ return lines
+}
diff --git a/migrations/168_subtitle_ai_jobs.down.sql b/migrations/168_subtitle_ai_jobs.down.sql
new file mode 100644
index 00000000..868b62fd
--- /dev/null
+++ b/migrations/168_subtitle_ai_jobs.down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS public.subtitle_ai_jobs;
diff --git a/migrations/168_subtitle_ai_jobs.up.sql b/migrations/168_subtitle_ai_jobs.up.sql
new file mode 100644
index 00000000..d5aec09c
--- /dev/null
+++ b/migrations/168_subtitle_ai_jobs.up.sql
@@ -0,0 +1,39 @@
+-- AI subtitle jobs: on-demand translation (and, later, Whisper ASR generation)
+-- of subtitle tracks. Each row is one user-triggered job. The worker runs the
+-- configured OpenAI-compatible engine and stores the result as an ordinary
+-- downloaded_subtitles row, so the generated track is served to every client
+-- through the existing subtitle pipeline with no client changes.
+CREATE TABLE public.subtitle_ai_jobs (
+ id bigserial PRIMARY KEY,
+ media_file_id integer NOT NULL,
+ kind text NOT NULL, -- 'translate' (future: 'transcribe', 'transcribe_translate')
+ source_index integer NOT NULL DEFAULT -1, -- combined player subtitle index of the source track
+ source_language text NOT NULL DEFAULT '',
+ target_language text NOT NULL,
+ engine text NOT NULL DEFAULT 'openai',
+ model text NOT NULL DEFAULT '', -- snapshot of the model used, for provenance
+ status text NOT NULL DEFAULT 'pending', -- pending|running|completed|failed|cancelled
+ progress double precision NOT NULL DEFAULT 0, -- 0..1
+ progress_message text NOT NULL DEFAULT '',
+ result_subtitle_id integer, -- downloaded_subtitles.id on success
+ error_message text NOT NULL DEFAULT '',
+ idempotency_key text NOT NULL,
+ requested_by integer,
+ created_at timestamptz NOT NULL DEFAULT now(),
+ updated_at timestamptz NOT NULL DEFAULT now(),
+ heartbeat_at timestamptz NOT NULL DEFAULT now()
+);
+
+-- Prevent duplicate in-flight jobs for the same work. Completed/failed/cancelled
+-- rows do not block a re-run, so a user can retry after a failure.
+CREATE UNIQUE INDEX subtitle_ai_jobs_active_idempotency_idx
+ ON public.subtitle_ai_jobs (idempotency_key)
+ WHERE status IN ('pending', 'running');
+
+-- Listing recent jobs for a media file (player "Translate" panel, admin views).
+CREATE INDEX subtitle_ai_jobs_media_file_idx
+ ON public.subtitle_ai_jobs (media_file_id, created_at DESC);
+
+-- Startup recovery scan for jobs left running by a crashed process.
+CREATE INDEX subtitle_ai_jobs_status_idx
+ ON public.subtitle_ai_jobs (status) WHERE status IN ('pending', 'running');
diff --git a/web/src/pages/admin-settings/IntegrationsSettings.tsx b/web/src/pages/admin-settings/IntegrationsSettings.tsx
index 37ded799..2088e3e7 100644
--- a/web/src/pages/admin-settings/IntegrationsSettings.tsx
+++ b/web/src/pages/admin-settings/IntegrationsSettings.tsx
@@ -4,7 +4,11 @@ import {
useUpdateSubtitleProvider,
useTestSubtitleProvider,
} from "@/hooks/queries/admin/subtitles";
-import { useAdminSensitiveStatus, useUpdateServerSetting } from "@/hooks/queries/admin/settings";
+import {
+ useAdminSensitiveStatus,
+ useAdminServerSettings,
+ useUpdateServerSetting,
+} from "@/hooks/queries/admin/settings";
import type { SubtitleProviderConfig } from "@/api/types";
import { Button } from "@/components/ui/button";
@@ -444,6 +448,99 @@ function IntroDBCredentialCard() {
);
}
+function AISubtitleTranslationCard() {
+ const { data: settings } = useAdminServerSettings();
+ const { data: sensitive } = useAdminSensitiveStatus();
+ const updateSetting = useUpdateServerSetting();
+
+ const apiKeyConfigured = new Set(sensitive?.configured ?? []).has("subtitle_ai.api_key");
+
+ const [enabled, setEnabled] = useState("false");
+ const [baseUrl, setBaseUrl] = useState("");
+ const [chatModel, setChatModel] = useState("");
+ const [maxConcurrent, setMaxConcurrent] = useState("2");
+ const [apiKey, setApiKey] = useState("");
+
+ // Hydrate the form from current server settings once loaded.
+ useEffect(() => {
+ if (!settings) return;
+ setEnabled(settings["subtitle_ai.enabled"] ?? "false");
+ setBaseUrl(settings["subtitle_ai.base_url"] ?? "https://api.openai.com");
+ setChatModel(settings["subtitle_ai.chat_model"] ?? "gpt-4o-mini");
+ setMaxConcurrent(settings["subtitle_ai.max_concurrent_jobs"] ?? "2");
+ }, [settings]);
+
+ function save() {
+ const updates = [
+ updateSetting.mutateAsync({ key: "subtitle_ai.enabled", value: enabled }),
+ updateSetting.mutateAsync({ key: "subtitle_ai.base_url", value: baseUrl }),
+ updateSetting.mutateAsync({ key: "subtitle_ai.chat_model", value: chatModel }),
+ updateSetting.mutateAsync({ key: "subtitle_ai.max_concurrent_jobs", value: maxConcurrent }),
+ ];
+ if (apiKey.trim() !== "") {
+ updates.push(updateSetting.mutateAsync({ key: "subtitle_ai.api_key", value: apiKey }));
+ }
+ void Promise.all(updates).then(() => setApiKey(""));
+ }
+
+ return (
+
+
+
+
AI Subtitle Translation
+
+ On-demand subtitle translation via any OpenAI-compatible chat API (OpenAI, Groq, a local
+ Ollama server, …). Translated tracks are generated once on the server and served to
+ every client.
+