feat(subtitles): on-demand AI subtitle translation with live streaming

Add server-side AI subtitle translation backed by any OpenAI-compatible
chat endpoint (OpenAI, Groq, a local Ollama/llama.cpp server). A viewer
picks a source track and target language in the player; the server runs a
bounded, resumable job pipeline that translates SRT/VTT cues in batches and
streams them back over the realtime websocket so playback pauses, fills in
cues near the playhead, and resumes. The finished track is persisted as an
ordinary downloaded subtitle, so it reaches every client through the
existing subtitle pipeline with no client changes.

- Job lifecycle persisted in subtitle_ai_jobs (migration 168): enqueue with
  idempotency, bounded concurrency, progress/heartbeat, cancellation, and
  crash recovery.
- New realtime events (subtitle_ready + subtitle_translation_*) with a
  per-session notifier; the player renders a synthetic "live" track fed by
  websocket cues. Timestamps never leave the server, so timing can't drift.
- Admin settings card for endpoint / model / concurrency.

Player + lifecycle hardening (from the code review of this feature):
- Hand off from the live track to the persisted track on completion
  (selected by downloaded-subtitle id) and on the subtitle_ready broadcast,
  so the saved track survives a reload and a mid-stream socket drop.
- Never persist the synthetic live-track sentinel index as a subtitle
  preference; restore the prior selection on failure; only auto-resume
  playback if the viewer was actually playing.
- Resume promptly when the playhead is past the last cue; rebuild the live
  track on a new job; O(batch) live-cue ingestion instead of O(n^2).

Reliability:
- Root translation jobs in the application context so shutdown cancels them.
- Heartbeat-based stale-job reaper (safe across multiple instances) replaces
  the table-wide startup reset.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Quick
2026-05-29 23:59:12 -04:00
co-authored by Claude Opus 4.8
parent 9123cfb1f4
commit e441d2d6e9
34 changed files with 3173 additions and 59 deletions
+1
View File
@@ -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,
+188
View File
@@ -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
}
+1 -15
View File
@@ -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
+59 -2
View File
@@ -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)
})
+15
View File
@@ -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:"-"`
+25
View File
@@ -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 {
+150 -4
View File
@@ -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
@@ -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)
}
}
+181
View File
@@ -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
}
+20
View File
@@ -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 != ""
}
+41
View File
@@ -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
}
+87
View File
@@ -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[:])
}
+36
View File
@@ -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
}
+24
View File
@@ -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)
}
+156
View File
@@ -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
}
+25
View File
@@ -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)
}
+558
View File
@@ -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 midLLM-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 })
}
+65
View File
@@ -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)
}
}
+158
View File
@@ -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)
}
+79
View File
@@ -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")
}
}
+223
View File
@@ -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
}
+1
View File
@@ -0,0 +1 @@
DROP TABLE IF EXISTS public.subtitle_ai_jobs;
+39
View File
@@ -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');
@@ -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 (
<div className="border-border bg-surface max-w-2xl rounded-lg border px-5 py-4">
<div className="mb-2 flex items-center justify-between gap-3">
<div>
<h3 className="text-sm font-semibold">AI Subtitle Translation</h3>
<p className="text-muted-foreground text-xs">
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.
</p>
</div>
<SubtitleCredentialStatus configured={apiKeyConfigured} />
</div>
<SettingField
label="Enabled"
type="toggle"
value={enabled}
onChange={setEnabled}
hint="Show the “Translate with AI” action in the player."
/>
<SettingField
label="Base URL"
type="text"
value={baseUrl}
onChange={setBaseUrl}
hint="https://api.openai.com"
/>
<SettingField
label="Chat model"
type="text"
value={chatModel}
onChange={setChatModel}
hint="e.g. gpt-4o-mini, llama3.1"
/>
<SettingField
label="API Key"
type="password"
value={apiKey}
onChange={setApiKey}
sensitiveConfigured={apiKeyConfigured}
hint="Leave blank to keep current. Empty is fine for keyless local servers."
/>
<SettingField
label="Max concurrent jobs"
type="number"
value={maxConcurrent}
onChange={setMaxConcurrent}
hint="Caps simultaneous translations so they don't starve transcodes."
/>
<div className="pt-2">
<Button type="button" onClick={save} disabled={updateSetting.isPending}>
{updateSetting.isPending ? "Saving..." : "Save AI Translation Settings"}
</Button>
</div>
</div>
);
}
export default function IntegrationsSettings() {
return (
<div className="flex h-full flex-col">
@@ -461,6 +558,9 @@ export default function IntegrationsSettings() {
<div className="mb-8">
<IntroDBCredentialCard />
</div>
<div className="mb-8">
<AISubtitleTranslationCard />
</div>
<SubtitlesContent />
</div>
);
@@ -44,6 +44,8 @@ interface PlayerControlsProps {
mediaFileId?: number;
playerConfig?: PlayerConfig;
onRefreshSubtitles?: () => void;
sessionId?: string;
getSubtitleStartPosition?: () => number;
// Audio
audioTracks: PlayerAudioTrack[];
activeAudioIndex: number;
@@ -102,6 +104,8 @@ export function PlayerControls({
mediaFileId,
playerConfig,
onRefreshSubtitles,
sessionId,
getSubtitleStartPosition,
audioTracks,
activeAudioIndex,
onAudioSelect,
@@ -304,6 +308,8 @@ export function PlayerControls({
mediaFileId={mediaFileId}
playerConfig={playerConfig}
onRefreshSubtitles={onRefreshSubtitles}
sessionId={sessionId}
getSubtitleStartPosition={getSubtitleStartPosition}
/>
<QualityMenu
+58 -2
View File
@@ -1,10 +1,12 @@
import { useState, useCallback, useEffect, useRef, useMemo } from "react";
import { createPortal } from "react-dom";
import { Captions, CaptionsOff, Minus, Plus, SlidersHorizontal } from "lucide-react";
import { Captions, CaptionsOff, Languages, Minus, Plus, SlidersHorizontal } from "lucide-react";
import type { PlayerSubtitleInfo } from "../types";
import type { PlayerConfig } from "../context/PlayerConfigContext";
import { SubtitleSearchModal } from "./SubtitleSearchModal";
import { SubtitleTranslateModal } from "./SubtitleTranslateModal";
import { SubtitleAppearancePanel } from "./SubtitleAppearancePanel";
import { playerFetch } from "../player-fetch";
import { getLanguageName } from "../utils/languageNames";
import { sortSubtitlesBySource } from "../utils/subtitleSort";
import { getSubtitleFormatLabel } from "../utils/assSubtitles";
@@ -18,6 +20,8 @@ interface SubtitleMenuProps {
mediaFileId?: number;
playerConfig?: PlayerConfig;
onRefreshSubtitles?: () => void;
sessionId?: string;
getSubtitleStartPosition?: () => number;
}
const DELAY_STEP_MS = 100;
@@ -44,14 +48,37 @@ export function SubtitleMenu({
mediaFileId,
playerConfig,
onRefreshSubtitles,
sessionId,
getSubtitleStartPosition,
}: SubtitleMenuProps) {
const [open, setOpen] = useState(false);
const [searchOpen, setSearchOpen] = useState(false);
const [translateOpen, setTranslateOpen] = useState(false);
const [aiEnabled, setAiEnabled] = useState(false);
const [appearanceOpen, setAppearanceOpen] = useState(false);
const menuRef = useRef<HTMLDivElement>(null);
const sortedTracks = useMemo(() => sortSubtitlesBySource(tracks), [tracks]);
// Discover whether the server has AI subtitle translation configured, so we
// only surface the entry point when it can actually do something. This is a
// server-wide capability, so we fetch it once per session (keyed on the stable
// playerConfig) rather than re-checking on every file change.
useEffect(() => {
if (!playerConfig) return;
let cancelled = false;
playerFetch<{ enabled: boolean }>(playerConfig, "/subtitles/ai/status")
.then((res) => {
if (!cancelled) setAiEnabled(Boolean(res?.enabled));
})
.catch(() => {
if (!cancelled) setAiEnabled(false);
});
return () => {
cancelled = true;
};
}, [playerConfig]);
const clampedDelay = useCallback(
(ms: number) => Math.max(-DELAY_MAX_MS, Math.min(DELAY_MAX_MS, ms)),
[],
@@ -274,9 +301,26 @@ export function SubtitleMenu({
Search Online
</button>
)}
{aiEnabled && mediaFileId && playerConfig && tracks.length > 0 && (
<button
ref={(el) => {
menuItemsRef.current[menuItemIndex + 2] = el;
}}
role="menuitem"
type="button"
className="flex w-full items-center gap-2 px-3 py-2 text-left text-sm text-white/70 hover:bg-white/10 focus-visible:ring-2 focus-visible:ring-white/70 focus-visible:outline-none"
onClick={() => {
setTranslateOpen(true);
setOpen(false);
}}
>
<Languages className="h-3.5 w-3.5 text-white/50" />
Translate with AI
</button>
)}
<button
ref={(el) => {
menuItemsRef.current[menuItemIndex + 2] = el;
menuItemsRef.current[menuItemIndex + 3] = el;
}}
role="menuitem"
type="button"
@@ -311,6 +355,18 @@ export function SubtitleMenu({
/>,
document.body,
)}
{translateOpen && mediaFileId && playerConfig && (
<SubtitleTranslateModal
mediaFileId={mediaFileId}
playerConfig={playerConfig}
tracks={tracks}
isOpen={translateOpen}
sessionId={sessionId}
getStartPosition={getSubtitleStartPosition}
onClose={() => setTranslateOpen(false)}
/>
)}
</div>
);
}
@@ -0,0 +1,188 @@
import { useState, useEffect, useCallback, useMemo } from "react";
import { createPortal } from "react-dom";
import type { PlayerConfig } from "../context/PlayerConfigContext";
import type { PlayerSubtitleInfo } from "../types";
import { playerFetch } from "../player-fetch";
import { LANGUAGES, getLanguageName } from "../utils/languageNames";
interface SubtitleTranslateModalProps {
mediaFileId: number;
playerConfig: PlayerConfig;
tracks: PlayerSubtitleInfo[];
isOpen: boolean;
sessionId?: string;
getStartPosition?: () => number;
onClose: () => void;
}
function sourceLabel(track: PlayerSubtitleInfo): string {
const lang = getLanguageName(track.language) || track.language || "Unknown";
const origin = track.source ? ` · ${track.source}` : "";
return `${lang}${origin}`;
}
export function SubtitleTranslateModal({
mediaFileId,
playerConfig,
tracks,
isOpen,
sessionId,
getStartPosition,
onClose,
}: SubtitleTranslateModalProps) {
// Live (in-progress) tracks can't be a translation source.
const sourceTracks = useMemo(() => tracks.filter((t) => !t.live), [tracks]);
const [sourceIndex, setSourceIndex] = useState<number | null>(null);
const [targetLang, setTargetLang] = useState("en");
const [submitting, setSubmitting] = useState(false);
const [error, setError] = useState<string | null>(null);
const effectiveSourceIndex = sourceIndex ?? sourceTracks[0]?.index ?? null;
useEffect(() => {
if (!isOpen) return;
const onKey = (e: KeyboardEvent) => {
if (e.key === "Escape") onClose();
};
document.addEventListener("keydown", onKey);
return () => document.removeEventListener("keydown", onKey);
}, [isOpen, onClose]);
const handleTranslate = useCallback(async () => {
if (effectiveSourceIndex === null) return;
const source = sourceTracks.find((t) => t.index === effectiveSourceIndex);
setSubmitting(true);
setError(null);
try {
await playerFetch(playerConfig, "/subtitles/ai/translate", {
method: "POST",
body: JSON.stringify({
media_file_id: mediaFileId,
source_index: effectiveSourceIndex,
source_language: source?.language ?? "",
target_language: targetLang,
session_id: sessionId ?? "",
start_position: getStartPosition?.() ?? 0,
}),
});
// The player takes over from here: it pauses, streams cues in as they're
// translated, then resumes once your position is covered.
onClose();
} catch (err) {
setError(err instanceof Error ? err.message : "Couldn't start translation.");
} finally {
setSubmitting(false);
}
}, [
effectiveSourceIndex,
sourceTracks,
mediaFileId,
targetLang,
sessionId,
getStartPosition,
playerConfig,
onClose,
]);
if (!isOpen) return null;
const modal = (
<div
className="fixed inset-0 z-50 flex items-center justify-center bg-black/80"
onClick={onClose}
role="dialog"
aria-modal="true"
aria-label="Translate subtitles with AI"
>
<div
className="w-full max-w-[440px] rounded-lg bg-neutral-900 text-white shadow-xl"
onClick={(e) => e.stopPropagation()}
>
<div className="flex items-center justify-between border-b border-white/10 px-4 py-3">
<h2 className="text-sm font-semibold">Translate subtitles with AI</h2>
<button
type="button"
className="rounded text-white/60 hover:text-white focus-visible:ring-2 focus-visible:ring-white/70 focus-visible:outline-none"
onClick={onClose}
aria-label="Close"
>
</button>
</div>
<div className="space-y-3 px-4 py-4">
{sourceTracks.length === 0 ? (
<p className="py-4 text-center text-xs text-white/50">
No text subtitle track is available to translate. Add or download one first.
</p>
) : (
<>
<label className="block">
<span className="mb-1 block text-xs font-medium text-white/60">Translate from</span>
<select
className="w-full rounded bg-neutral-800 px-2 py-1.5 text-sm text-white focus-visible:ring-2 focus-visible:ring-white/70 focus-visible:outline-none disabled:opacity-50"
value={effectiveSourceIndex ?? ""}
onChange={(e) => setSourceIndex(Number(e.target.value))}
disabled={submitting}
>
{sourceTracks.map((track) => (
<option key={track.index} value={track.index}>
{sourceLabel(track)}
</option>
))}
</select>
</label>
<label className="block">
<span className="mb-1 block text-xs font-medium text-white/60">Translate to</span>
<select
className="w-full rounded bg-neutral-800 px-2 py-1.5 text-sm text-white focus-visible:ring-2 focus-visible:ring-white/70 focus-visible:outline-none disabled:opacity-50"
value={targetLang}
onChange={(e) => setTargetLang(e.target.value)}
disabled={submitting}
>
{LANGUAGES.map((lang) => (
<option key={lang.code} value={lang.code}>
{lang.label}
</option>
))}
</select>
</label>
{error && (
<div role="alert" className="rounded bg-red-900/40 px-3 py-2 text-xs text-red-300">
{error}
</div>
)}
<div className="flex justify-end gap-2 pt-1">
<button
type="button"
className="rounded px-3 py-1.5 text-sm text-white/60 hover:bg-white/10 focus-visible:ring-2 focus-visible:ring-white/70 focus-visible:outline-none"
onClick={onClose}
>
Cancel
</button>
<button
type="button"
className="rounded bg-white/10 px-3 py-1.5 text-sm font-medium hover:bg-white/20 focus-visible:ring-2 focus-visible:ring-white/70 focus-visible:outline-none disabled:opacity-50"
onClick={handleTranslate}
disabled={submitting || effectiveSourceIndex === null}
>
{submitting ? "Starting…" : "Translate"}
</button>
</div>
<p className="text-[11px] leading-relaxed text-white/35">
Playback pauses while the first lines are translated, then resumes with subtitles
streaming in. The finished track is saved for everyone.
</p>
</>
)}
</div>
</div>
</div>
);
return createPortal(modal, document.body);
}
+231 -8
View File
@@ -1,4 +1,5 @@
import { useCallback, useEffect, useRef, useState } from "react";
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import type { ParsedCue } from "../utils/parseVTT";
import { resolveSubtitleAutoSelect } from "../utils/subtitleSort";
import type HlsType from "hls.js";
import { PlayerControls } from "./PlayerControls";
@@ -50,6 +51,13 @@ import { toMediaTime, toPlayerTime } from "../utils/mediaTimeline";
import { buildWatchTogetherInviteUrl } from "@/lib/watchTogether";
import { toast } from "sonner";
// Reserved index for the in-progress live AI translation track. Sits well above
// any real subtitle index so it never collides.
const LIVE_SUBTITLE_INDEX = 1_000_000;
// Resume playback once translated cues cover at least this far ahead of the
// playhead; a hard cap also resumes so we never wait forever.
const TRANSLATION_RESUME_TIMEOUT_MS = 30_000;
interface VideoPlayerProps {
title: string;
year?: number;
@@ -244,6 +252,57 @@ export function VideoPlayer({
setSubtitleDelayMs(0);
}, [activeFileId]);
// -- Live AI subtitle translation (streamed over the realtime websocket) --
// While a translation runs, a synthetic "live" track is added to the list and
// selected; cues arrive over the websocket and the player pauses until the
// region near the playhead is covered, then resumes.
const [liveTranslation, setLiveTranslation] = useState<{
trackKey: string;
language: string;
label: string;
} | null>(null);
const [liveCues, setLiveCues] = useState<ParsedCue[]>([]);
const [translationBuffering, setTranslationBuffering] = useState(false);
const translationPauseRef = useRef(false);
const translationResumeTimerRef = useRef<number | null>(null);
// Whether playback should auto-resume once buffering ends. Captured at
// translation start: if the viewer had deliberately paused, we don't yank
// them back into playback.
const translationResumeOnFinishRef = useRef(false);
// The subtitle selection active before a translation hijacked it, so a failed
// translation can restore it instead of leaving subtitles off.
const preTranslationSubtitleIndexRef = useRef<number | null>(null);
// The persisted downloaded-subtitle id to switch to once a completed
// translation's track lands in the refreshed list.
const pendingTranslatedSubtitleIdRef = useRef<number | null>(null);
// Drop any live translation when the media changes so a stale track from the
// previous file never lingers.
useEffect(() => {
setLiveTranslation(null);
setLiveCues([]);
setTranslationBuffering(false);
translationPauseRef.current = false;
pendingTranslatedSubtitleIdRef.current = null;
}, [activeFileId]);
// Merge the live track into the track list the player + menu see.
const effectiveSubtitleTracks = useMemo(() => {
if (!liveTranslation) return subtitleUrls;
return [
...subtitleUrls,
{
index: LIVE_SUBTITLE_INDEX,
language: liveTranslation.language,
label: liveTranslation.label || "AI translation",
source: "downloaded" as const,
codec: "srt",
url: "",
live: true,
},
];
}, [subtitleUrls, liveTranslation]);
// -- Transcode quality switching --
// Remux also uses HLS (codec copy) via the transcode pipeline.
const transcodeQuality = useTranscodeQuality({
@@ -739,11 +798,161 @@ export function VideoPlayer({
(index: number | null) => {
subtitleSelectionWasManualRef.current = true;
setActiveSubtitleIndex(index);
// The in-progress live translation track is synthetic (a sentinel index
// that exists only in memory); never persist it as the saved preference or
// we'd store a nonexistent track and lose the real selection.
if (index === LIVE_SUBTITLE_INDEX) return;
onSubtitleChanged?.(index);
},
[onSubtitleChanged],
);
// The media-time playhead, sent with a translate request so the server starts
// where the viewer is watching.
const getSubtitleStartPosition = useCallback(
() => toMediaTime(videoRef.current?.currentTime ?? 0, streamOriginRef.current ?? 0),
[],
);
const resumeFromTranslationPause = useCallback(() => {
if (translationResumeTimerRef.current !== null) {
window.clearTimeout(translationResumeTimerRef.current);
translationResumeTimerRef.current = null;
}
if (translationPauseRef.current) {
translationPauseRef.current = false;
// Only resume if the viewer was playing when the translation began; if
// they had paused on purpose, leave them paused.
if (translationResumeOnFinishRef.current) {
void videoRef.current?.play().catch(() => {});
}
}
setTranslationBuffering(false);
}, []);
// Intercept live-translation events; forward everything else to the parent.
const handleRealtimeEvent = useCallback(
(event: PlaybackRealtimeEventEnvelope) => {
switch (event.name) {
case "subtitle_ready": {
// Broadcast to every viewer of the file when a generated track is
// persisted. Refresh the list so it appears (the requesting session
// also auto-selects it via the completed handler below).
if (event.payload.file_id === activeFileId) {
onRefreshSubtitles?.();
}
break;
}
case "subtitle_translation_started": {
// Remember the real selection we're displacing and whether we were
// playing, so completion/failure can restore the right state.
const wasPlaying = !(videoRef.current?.paused ?? true);
translationResumeOnFinishRef.current = wasPlaying;
setActiveSubtitleIndex((idx) => {
if (idx !== LIVE_SUBTITLE_INDEX) {
preTranslationSubtitleIndexRef.current = idx;
}
return LIVE_SUBTITLE_INDEX;
});
pendingTranslatedSubtitleIdRef.current = null;
setLiveCues([]);
setLiveTranslation({
trackKey: event.payload.track_key,
language: event.payload.language,
label: event.payload.label ?? "",
});
subtitleSelectionWasManualRef.current = true;
translationPauseRef.current = true;
setTranslationBuffering(true);
// Only pause if the viewer was playing; don't disturb a deliberate pause.
if (wasPlaying) videoRef.current?.pause();
if (translationResumeTimerRef.current !== null) {
window.clearTimeout(translationResumeTimerRef.current);
}
translationResumeTimerRef.current = window.setTimeout(
resumeFromTranslationPause,
TRANSLATION_RESUME_TIMEOUT_MS,
);
break;
}
case "subtitle_translation_cues": {
const cues = event.payload.cues.map((c) => ({
start: c.start,
end: c.end,
text: c.text,
}));
setLiveCues((prev) => [...prev, ...cues]);
break;
}
case "subtitle_translation_completed": {
resumeFromTranslationPause();
// Hand off from the ephemeral live track to the persisted downloaded
// track: refresh the list and let the effect below select it by id
// once it lands. Without a refresh callback we keep the live track
// (which already holds the full cue set) as a best-effort fallback.
if (onRefreshSubtitles) {
pendingTranslatedSubtitleIdRef.current = event.payload.subtitle_id;
onRefreshSubtitles();
}
break;
}
case "subtitle_translation_failed": {
resumeFromTranslationPause();
setLiveTranslation(null);
setLiveCues([]);
pendingTranslatedSubtitleIdRef.current = null;
// Restore the selection the translation displaced rather than leaving
// subtitles off.
const restore = preTranslationSubtitleIndexRef.current;
setActiveSubtitleIndex((idx) => (idx === LIVE_SUBTITLE_INDEX ? restore : idx));
toast.error(
event.payload.message
? `Translation failed: ${event.payload.message}`
: "Subtitle translation failed",
);
break;
}
default:
onRealtimeEvent?.(event);
}
},
[onRealtimeEvent, onRefreshSubtitles, activeFileId, resumeFromTranslationPause],
);
// Once a completed translation's persisted track lands in the refreshed list,
// switch to it (selecting by downloaded-subtitle id) and drop the live track,
// so the viewer ends up on the real saved subtitle rather than the synthetic
// one that would vanish on reload.
useEffect(() => {
const pendingId = pendingTranslatedSubtitleIdRef.current;
if (pendingId == null) return;
const match = subtitleUrls.find((t) => t.id === pendingId);
if (!match) return;
pendingTranslatedSubtitleIdRef.current = null;
setLiveTranslation(null);
setLiveCues([]);
handleSubtitleSelect(match.index);
}, [subtitleUrls, handleSubtitleSelect]);
// Resume as soon as the first translated cues arrive. Playhead-first
// translation means the cues covering the current position are delivered
// first, so the first batch is enough; and when the playhead is past the last
// cue (e.g. end credits) there is nothing at the playhead to wait for, so we
// still resume here rather than stalling until the 30s timeout.
useEffect(() => {
if (!translationPauseRef.current || liveCues.length === 0) return;
resumeFromTranslationPause();
}, [liveCues, resumeFromTranslationPause]);
useEffect(
() => () => {
if (translationResumeTimerRef.current !== null) {
window.clearTimeout(translationResumeTimerRef.current);
}
},
[],
);
// -- PiP toggle --
const handleTogglePiP = useCallback(async () => {
const video = videoRef.current;
@@ -1335,10 +1544,12 @@ export function VideoPlayer({
// (which has browser bugs with stale cues persisting after seek).
const activeCueTexts = useSubtitleTracks(
videoRef,
subtitleUrls,
effectiveSubtitleTracks,
activeSubtitleIndex,
streamOriginRef,
subtitleDelayMs,
liveCues,
liveTranslation?.trackKey ?? null,
);
// -- ASS/SSA subtitle rendering via JASSUB (client-side libass) --
@@ -1356,14 +1567,14 @@ export function VideoPlayer({
if (subtitleSelectionWasManualRef.current) {
const selectionStillExists =
activeSubtitleIndex === null ||
subtitleUrls.some((track) => track.index === activeSubtitleIndex);
effectiveSubtitleTracks.some((track) => track.index === activeSubtitleIndex);
if (selectionStillExists) {
return;
}
subtitleSelectionWasManualRef.current = false;
}
if (subtitleUrls.length === 0) {
if (effectiveSubtitleTracks.length === 0) {
setActiveSubtitleIndex(null);
lastSubtitleIndexRef.current = null;
return;
@@ -1376,7 +1587,7 @@ export function VideoPlayer({
const match = resolveSubtitleAutoSelect({
mode: effectiveMode,
tracks: subtitleUrls,
tracks: effectiveSubtitleTracks,
preferredLanguage: preferredSubtitleLanguage ?? null,
preferredTrackSignature: preferredSubtitleTrackSignature ?? null,
audioLanguage: audioLang,
@@ -1395,7 +1606,7 @@ export function VideoPlayer({
activeSubtitleIndex,
preferredSubtitleLanguage,
preferredSubtitleTrackSignature,
subtitleUrls,
effectiveSubtitleTracks,
subtitleMode,
showForcedSubtitles,
profileLanguage,
@@ -1719,7 +1930,7 @@ export function VideoPlayer({
const realtime = usePlaybackRealtime({
sessionId,
onCommand: executeRealtimeCommand,
onEvent: onRealtimeEvent,
onEvent: handleRealtimeEvent,
});
useEffect(() => {
@@ -2044,6 +2255,16 @@ export function VideoPlayer({
/>
)}
{/* Live translation buffering indicator */}
{translationBuffering && (
<div className="pointer-events-none absolute inset-0 z-30 flex items-center justify-center">
<div className="flex items-center gap-3 rounded-lg bg-black/80 px-4 py-3 text-sm text-white shadow-lg">
<span className="h-4 w-4 animate-spin rounded-full border-2 border-white/30 border-t-white" />
Preparing {liveTranslation?.label || "translated"} subtitles
</div>
</div>
)}
{/* Controls */}
{!isDetached && isPlayerReady && (
<PlayerControls
@@ -2058,7 +2279,7 @@ export function VideoPlayer({
volume={volume}
muted={muted}
isFullscreen={isFullscreen}
subtitleTracks={subtitleUrls}
subtitleTracks={effectiveSubtitleTracks}
activeSubtitleIndex={activeSubtitleIndex}
onSubtitleSelect={handleSubtitleSelect}
subtitleDelayMs={subtitleDelayMs}
@@ -2066,6 +2287,8 @@ export function VideoPlayer({
mediaFileId={activeFileId ?? undefined}
playerConfig={playerConfig}
onRefreshSubtitles={onRefreshSubtitles}
sessionId={sessionId}
getSubtitleStartPosition={getSubtitleStartPosition}
audioTracks={audioTracks}
activeAudioIndex={activeAudioIndex}
onAudioSelect={onAudioSelect}
+4
View File
@@ -172,6 +172,10 @@ export function WatchPage({
if (!seriesId) return;
const track = index !== null ? playableSubtitles.find((s) => s.index === index) : null;
// Never persist an index we can't resolve to a real track (e.g. the
// in-progress AI live track's sentinel index): it would store a
// nonexistent track with empty language and clobber the saved preference.
if (index !== null && !track) return;
const trackSignature: PlayerSubtitleTrackSignature | null = track
? {
source: track.source,
@@ -513,6 +513,7 @@ export function usePlaybackSession(
const token = config.getAccessToken();
const newTracks: PlayerSubtitleInfo[] = downloaded.map((dl, i) => ({
index: baseIndex + i,
id: dl.id,
language: dl.language,
codec: dl.format,
label: `${dl.release_name} (${dl.provider})`,
+91 -26
View File
@@ -23,6 +23,29 @@ function stripVTTTags(text: string): string {
return text.replace(/<[^>]+>/g, "");
}
/**
* Add parsed cues to a TextTrack, applying the stream origin and user delay and
* deduping against `seen`. Shared by the URL fetcher and the live-cue path.
*/
function addCuesToTrack(
track: TextTrack,
cues: ParsedCue[],
origin: number,
delaySec: number,
seen: Set<string>,
): void {
for (const parsed of cues) {
if (parsed.end <= parsed.start) continue;
const startTime = Math.max(0, parsed.start - origin + delaySec);
const endTime = parsed.end - origin + delaySec;
if (endTime <= 0) continue;
const key = `${startTime}|${endTime}|${parsed.text}`;
if (seen.has(key)) continue;
seen.add(key);
track.addCue(new VTTCue(startTime, endTime, parsed.text));
}
}
/** Append or replace the `position` query param on a subtitle URL. */
function appendPosition(url: string, position: number): string {
const sep = url.includes("?") ? "&" : "?";
@@ -55,6 +78,10 @@ export function useSubtitleTracks(
activeSubtitleIndex: number | null,
streamOriginRef: React.RefObject<number>,
subtitleDelayMs: number,
liveCues?: ParsedCue[] | null,
// Identifies the current live translation job. Changing it (a new job) rebuilds
// the track and resets the dedup set so cues from a prior run never linger.
liveTrackKey?: string | null,
): string[] {
const [activeCueTexts, setActiveCueTexts] = useState<string[]>([]);
@@ -65,12 +92,22 @@ export function useSubtitleTracks(
const activeUrl = activeSub?.url ?? null;
const activeCodec = activeSub?.codec;
const activeLang = activeSub?.language ?? "";
// A live track's cues arrive over the websocket (liveCues) instead of from a
// URL; the main effect builds the track but skips the sliding-window fetcher.
const activeIsLive = activeSub?.live === true;
// Track which delay is currently baked into the VTTCues on the active track,
// so the delay-update effect below can compute the exact shift to apply.
// Cue-add paths also read this to keep new cues aligned with existing ones.
const appliedDelayMsRef = useRef(0);
const trackRef = useRef<TextTrack | null>(null);
// Cue dedup set, held in a ref so the live-cue effect and the URL fetcher
// share it. Reset whenever a fresh track is built.
const seenCueKeysRef = useRef<Set<string>>(new Set());
// How many of `liveCues` have already been pushed onto the live track. Lets the
// live-cue effect add only the new tail each batch instead of rescanning the
// whole (growing) array. Reset on every track rebuild.
const processedLiveCuesRef = useRef(0);
useEffect(() => {
const video = videoRef.current;
@@ -80,7 +117,11 @@ export function useSubtitleTracks(
setActiveCueTexts([]);
// Skip entirely for ASS/SSA: JASSUB handles those via useASSSubtitles.
if (!activeUrl || isASSCodec(activeCodec)) {
if (isASSCodec(activeCodec)) {
return;
}
// Need either a URL to stream from or a live cue source.
if (!activeUrl && !activeIsLive) {
return;
}
@@ -91,6 +132,8 @@ export function useSubtitleTracks(
const track = videoEl.addTextTrack("subtitles", "Silo", activeLang || undefined);
track.mode = "hidden";
trackRef.current = track;
seenCueKeysRef.current = new Set();
processedLiveCuesRef.current = 0;
let cancelled = false;
let hasFetched = false;
@@ -101,10 +144,6 @@ export function useSubtitleTracks(
let atEOF = false;
let inflight: AbortController | null = null;
// Persistent dedup: cues from overlapping windows key to the same string.
// Cleared alongside track cues on backward-seek resets.
const seenCueKeys = new Set<string>();
function handleCueChange() {
const active = track.activeCues;
if (!active || active.length === 0) {
@@ -124,7 +163,7 @@ export function useSubtitleTracks(
for (const cue of Array.from(cues)) {
track.removeCue(cue);
}
seenCueKeys.clear();
seenCueKeysRef.current.clear();
}
function addParsedCues(newCues: ParsedCue[]) {
@@ -136,17 +175,7 @@ export function useSubtitleTracks(
// baked in here so new cues line up with existing ones.
const origin = streamOriginRef.current ?? 0;
const delaySec = appliedDelayMsRef.current / 1000;
for (const parsed of newCues) {
if (parsed.end <= parsed.start) continue;
const startTime = Math.max(0, parsed.start - origin + delaySec);
const endTime = parsed.end - origin + delaySec;
if (endTime <= 0) continue;
const key = `${startTime}|${endTime}|${parsed.text}`;
if (seenCueKeys.has(key)) continue;
seenCueKeys.add(key);
track.addCue(new VTTCue(startTime, endTime, parsed.text));
}
addCuesToTrack(track, newCues, origin, delaySec, seenCueKeysRef.current);
}
async function fetchWindow(seekStart: number, resetExisting: boolean) {
@@ -251,15 +280,19 @@ export function useSubtitleTracks(
}
}
// Kick off the first window before any player event fires so cues
// are already in flight for the current position.
maybeFetch();
// URL-backed tracks run the sliding-window fetcher; live tracks receive
// their cues from the liveCues effect below instead.
if (!activeIsLive) {
// Kick off the first window before any player event fires so cues
// are already in flight for the current position.
maybeFetch();
// Cue activation is driven by the browser via `cuechange`; these
// listeners exist only to keep the sliding-window fetcher scheduled.
videoEl.addEventListener("timeupdate", maybeFetch);
videoEl.addEventListener("seeking", maybeFetch);
videoEl.addEventListener("seeked", maybeFetch);
// Cue activation is driven by the browser via `cuechange`; these
// listeners exist only to keep the sliding-window fetcher scheduled.
videoEl.addEventListener("timeupdate", maybeFetch);
videoEl.addEventListener("seeking", maybeFetch);
videoEl.addEventListener("seeked", maybeFetch);
}
return () => {
cancelled = true;
@@ -281,7 +314,7 @@ export function useSubtitleTracks(
// `subtitleDelayMs` is intentionally excluded — nudging delay must not
// tear down and refetch the track. The delay-update effect below shifts
// existing cues in place instead.
}, [activeUrl, activeCodec, activeLang, streamOriginRef, videoRef]);
}, [activeUrl, activeCodec, activeLang, activeIsLive, liveTrackKey, streamOriginRef, videoRef]);
// Apply delay changes to already-loaded cues without rebuilding the track.
// Runs after the main effect, so trackRef is current.
@@ -303,5 +336,37 @@ export function useSubtitleTracks(
}
}, [subtitleDelayMs]);
// Feed websocket-pushed cues into the active live track as they arrive. Only
// the new tail since the last push is added (liveCues is append-only within a
// job), so ingestion stays O(batch) rather than O(total) per push. When the
// job restarts liveCues is replaced with a shorter array, which the length
// check below detects to start over (the track itself is rebuilt via
// liveTrackKey, so a fresh seen-set and pointer are already in place).
useEffect(() => {
if (!activeIsLive) return;
const track = trackRef.current;
if (!track || !liveCues) return;
if (liveCues.length < processedLiveCuesRef.current) {
processedLiveCuesRef.current = 0;
}
const fresh = liveCues.slice(processedLiveCuesRef.current);
if (fresh.length === 0) return;
processedLiveCuesRef.current = liveCues.length;
const origin = streamOriginRef.current ?? 0;
const delaySec = appliedDelayMsRef.current / 1000;
addCuesToTrack(track, fresh, origin, delaySec, seenCueKeysRef.current);
// While paused, adding a cue over the playhead doesn't reliably fire
// `cuechange`, so refresh the on-screen text by hand. While playing the
// browser drives `cuechange`, so skip the redundant state update.
if (videoRef.current?.paused) {
const active = track.activeCues;
setActiveCueTexts(
active && active.length > 0
? Array.from(active).map((c) => stripVTTTags((c as VTTCue).text))
: [],
);
}
}, [liveCues, activeIsLive, activeSubtitleIndex, liveTrackKey, streamOriginRef, videoRef]);
return activeCueTexts;
}
+43
View File
@@ -104,6 +104,49 @@ describe("realtime protocol", () => {
});
});
it("parses subtitle ready events", () => {
const event = parsePlaybackRealtimeMessage(
JSON.stringify({
type: "event",
session_id: "session-1",
name: "subtitle_ready",
payload: {
session_id: "session-1",
file_id: 42,
subtitle_id: 7,
language: "es",
label: "English → Spanish (AI)",
},
}),
);
expect(event).toEqual({
type: "event",
session_id: "session-1",
name: "subtitle_ready",
payload: {
session_id: "session-1",
file_id: 42,
subtitle_id: 7,
language: "es",
label: "English → Spanish (AI)",
},
});
});
it("rejects subtitle ready events missing the subtitle id", () => {
const event = parsePlaybackRealtimeMessage(
JSON.stringify({
type: "event",
session_id: "session-1",
name: "subtitle_ready",
payload: { session_id: "session-1", file_id: 42, language: "es" },
}),
);
expect(event).toBeNull();
});
it("builds hello, ack, and result envelopes", () => {
expect(buildPlaybackRealtimeHello("session-1")).toEqual({
type: "hello",
+208 -1
View File
@@ -17,7 +17,14 @@ export type PlaybackCommandName =
export type PlaybackRealtimeAckStatus = "accepted";
export type PlaybackRealtimeResultStatus = "completed" | "rejected";
export type PlaybackRealtimeEventName = "chapter_thumbnail_ready" | "markers_updated";
export type PlaybackRealtimeEventName =
| "chapter_thumbnail_ready"
| "markers_updated"
| "subtitle_ready"
| "subtitle_translation_started"
| "subtitle_translation_cues"
| "subtitle_translation_completed"
| "subtitle_translation_failed";
export interface PlaybackRealtimeCommandEnvelope {
type: "command";
@@ -66,6 +73,64 @@ export interface PlaybackMarkersUpdatedPayload {
preview?: PlaybackTimeRangePayload | null;
}
/**
* Broadcast to every session watching a file when a newly generated subtitle
* track (AI translation, later ASR) has been persisted, so players can refresh
* their track list and pick it up without a manual reload.
*/
export interface PlaybackSubtitleReadyPayload {
session_id: string;
file_id: number;
subtitle_id: number;
language: string;
label?: string;
}
/** One translated subtitle cue pushed during a live translation (media seconds). */
export interface PlaybackStreamCue {
start: number;
end: number;
text: string;
}
export interface PlaybackSubtitleTranslationStartedPayload {
session_id: string;
file_id: number;
job_id: number;
track_key: string;
language: string;
label?: string;
total_cues: number;
}
export interface PlaybackSubtitleTranslationCuesPayload {
session_id: string;
file_id: number;
job_id: number;
track_key: string;
cues: PlaybackStreamCue[];
done: number;
total: number;
}
export interface PlaybackSubtitleTranslationCompletedPayload {
session_id: string;
file_id: number;
job_id: number;
track_key: string;
subtitle_id: number;
language: string;
label?: string;
}
export interface PlaybackSubtitleTranslationFailedPayload {
session_id: string;
file_id: number;
job_id: number;
track_key: string;
message?: string;
}
export interface PlaybackRealtimeEventEnvelopeBase {
type: "event";
session_id: string;
@@ -79,6 +144,26 @@ export type PlaybackRealtimeEventEnvelope =
| (PlaybackRealtimeEventEnvelopeBase & {
name: "markers_updated";
payload: PlaybackMarkersUpdatedPayload;
})
| (PlaybackRealtimeEventEnvelopeBase & {
name: "subtitle_ready";
payload: PlaybackSubtitleReadyPayload;
})
| (PlaybackRealtimeEventEnvelopeBase & {
name: "subtitle_translation_started";
payload: PlaybackSubtitleTranslationStartedPayload;
})
| (PlaybackRealtimeEventEnvelopeBase & {
name: "subtitle_translation_cues";
payload: PlaybackSubtitleTranslationCuesPayload;
})
| (PlaybackRealtimeEventEnvelopeBase & {
name: "subtitle_translation_completed";
payload: PlaybackSubtitleTranslationCompletedPayload;
})
| (PlaybackRealtimeEventEnvelopeBase & {
name: "subtitle_translation_failed";
payload: PlaybackSubtitleTranslationFailedPayload;
});
export interface PlaybackRealtimeAckEnvelope {
@@ -164,6 +249,79 @@ function isMarkersUpdatedPayload(value: unknown): value is PlaybackMarkersUpdate
);
}
function isSubtitleReadyPayload(value: unknown): value is PlaybackSubtitleReadyPayload {
return (
isRecord(value) &&
typeof value.session_id === "string" &&
typeof value.file_id === "number" &&
typeof value.subtitle_id === "number" &&
typeof value.language === "string"
);
}
function isStreamCue(value: unknown): value is PlaybackStreamCue {
return (
isRecord(value) &&
typeof value.start === "number" &&
typeof value.end === "number" &&
typeof value.text === "string"
);
}
function isTranslationStartedPayload(
value: unknown,
): value is PlaybackSubtitleTranslationStartedPayload {
return (
isRecord(value) &&
typeof value.session_id === "string" &&
typeof value.file_id === "number" &&
typeof value.job_id === "number" &&
typeof value.track_key === "string" &&
typeof value.language === "string" &&
typeof value.total_cues === "number"
);
}
function isTranslationCuesPayload(value: unknown): value is PlaybackSubtitleTranslationCuesPayload {
return (
isRecord(value) &&
typeof value.session_id === "string" &&
typeof value.file_id === "number" &&
typeof value.job_id === "number" &&
typeof value.track_key === "string" &&
Array.isArray(value.cues) &&
value.cues.every(isStreamCue) &&
typeof value.done === "number" &&
typeof value.total === "number"
);
}
function isTranslationCompletedPayload(
value: unknown,
): value is PlaybackSubtitleTranslationCompletedPayload {
return (
isRecord(value) &&
typeof value.session_id === "string" &&
typeof value.file_id === "number" &&
typeof value.job_id === "number" &&
typeof value.track_key === "string" &&
typeof value.subtitle_id === "number" &&
typeof value.language === "string"
);
}
function isTranslationFailedPayload(
value: unknown,
): value is PlaybackSubtitleTranslationFailedPayload {
return (
isRecord(value) &&
typeof value.session_id === "string" &&
typeof value.file_id === "number" &&
typeof value.job_id === "number" &&
typeof value.track_key === "string"
);
}
export function parsePlaybackRealtimeMessage(
data: string,
): PlaybackRealtimeCommandEnvelope | PlaybackRealtimeEventEnvelope | null {
@@ -206,6 +364,14 @@ export function parsePlaybackRealtimeMessage(
payload: value.payload,
};
}
if (value.name === "subtitle_ready" && isSubtitleReadyPayload(value.payload)) {
return {
type: "event",
session_id: value.session_id,
name: value.name,
payload: value.payload,
};
}
if (value.name === "markers_updated" && isMarkersUpdatedPayload(value.payload)) {
return {
type: "event",
@@ -214,6 +380,47 @@ export function parsePlaybackRealtimeMessage(
payload: value.payload,
};
}
if (
value.name === "subtitle_translation_started" &&
isTranslationStartedPayload(value.payload)
) {
return {
type: "event",
session_id: value.session_id,
name: value.name,
payload: value.payload,
};
}
if (value.name === "subtitle_translation_cues" && isTranslationCuesPayload(value.payload)) {
return {
type: "event",
session_id: value.session_id,
name: value.name,
payload: value.payload,
};
}
if (
value.name === "subtitle_translation_completed" &&
isTranslationCompletedPayload(value.payload)
) {
return {
type: "event",
session_id: value.session_id,
name: value.name,
payload: value.payload,
};
}
if (
value.name === "subtitle_translation_failed" &&
isTranslationFailedPayload(value.payload)
) {
return {
type: "event",
session_id: value.session_id,
name: value.name,
payload: value.payload,
};
}
}
return null;
} catch {
+12
View File
@@ -110,6 +110,12 @@ export interface PlaybackSessionPlaybackInfo {
/** Subtitle track information. */
export interface PlayerSubtitleInfo {
index: number;
/**
* Downloaded-subtitle row id, when this track is a stored downloaded subtitle.
* Lets the player match a translation-completed / `subtitle_ready` event
* (which carries the DB id) to a track after a list refresh.
*/
id?: number;
language: string;
codec?: string;
label: string;
@@ -117,6 +123,12 @@ export interface PlayerSubtitleInfo {
forced?: boolean;
hearing_impaired?: boolean;
url: string;
/**
* When true, this is an in-progress AI translation whose cues arrive over the
* realtime websocket rather than from `url`. `useSubtitleTracks` feeds it from
* the `liveCues` source instead of fetching.
*/
live?: boolean;
}
export interface PlayerSubtitleTrackSignature {