// Package service — HLS on-demand transcoder. // // TranscoderService spawns ffmpeg processes that segment a source media file // into HLS (.m3u8 + .ts). The output lives under cache.cache_dir/hls/. // The HTTP layer serves these files directly with a normal http.FileServer. // // Encoder selection (read once at startup from the config): // // transcoder.encoder = "" | "nvenc" | "qsv" | "vaapi" // // "" software libx264 (default; runs anywhere) // nvenc h264_nvenc (NVIDIA GPU, requires --gpus all on Docker) // qsv h264_qsv (Intel iGPU, requires /dev/dri:/dev/dri) // vaapi h264_vaapi (Mesa/Intel VAAPI, requires /dev/dri:/dev/dri // plus the kernel module loaded) // // Concurrency model: // - Each Media has at most one active ffmpeg job. // - jobs[mediaID] tracks the running goroutine + cancel func. // - Calling Start while a job already exists is a no-op. // - When the playlist file appears on disk we consider the job "ready" // and unblock the HTTP handler that was waiting on it. package service import ( "context" "errors" "fmt" "os" "os/exec" "path/filepath" "strings" "sync" "time" "go.uber.org/zap" "github.com/ShukeBta/MediaStationGo/internal/config" "github.com/ShukeBta/MediaStationGo/internal/repository" ) // TranscoderService orchestrates background ffmpeg transcodes. type TranscoderService struct { cfg *config.Config log *zap.Logger repo *repository.Container hub *Hub mu sync.Mutex jobs map[string]*hlsJob } // hlsJob holds the live state of one ffmpeg run. type hlsJob struct { mediaID string outputDir string cancel context.CancelFunc startedAt time.Time lastAccess time.Time playlistOK bool encoder string } var ( // ErrTranscodeDisabled is returned when HLS transcoding is globally disabled. ErrTranscodeDisabled = errors.New("transcode disabled") // ErrTranscodeBusy is returned when the server has reached its configured // ffmpeg concurrency limit. ErrTranscodeBusy = errors.New("transcode concurrency limit reached") ) // NewTranscoderService is the constructor. func NewTranscoderService(cfg *config.Config, log *zap.Logger, repo *repository.Container, hub *Hub) *TranscoderService { return &TranscoderService{ cfg: cfg, log: log, repo: repo, hub: hub, jobs: make(map[string]*hlsJob), } } // HLSDir is the per-media directory that holds index.m3u8 + segment files. func (t *TranscoderService) HLSDir(mediaID string) string { return filepath.Join(t.cfg.Cache.CacheDir, "hls", mediaID) } // PlaylistPath returns the absolute path of the m3u8 playlist for a media. func (t *TranscoderService) PlaylistPath(mediaID string) string { return filepath.Join(t.HLSDir(mediaID), "index.m3u8") } // EnsureJob makes sure a transcode is running for mediaID. The function is // non-blocking: it returns the playlist path immediately. The caller is // expected to poll until WaitReady reports true. func (t *TranscoderService) EnsureJob(ctx context.Context, mediaID string) (string, error) { if !t.cfg.Transcoder.Enabled { return "", ErrTranscodeDisabled } m, err := t.repo.Media.FindByID(ctx, mediaID) if err != nil { return "", err } if m == nil { return "", ErrMediaNotFound } if _, err := os.Stat(m.Path); err != nil { return "", ErrMediaNotFound } if _, err := t.resolveFFmpegPath(); err != nil { return "", err } t.mu.Lock() if _, ok := t.jobs[mediaID]; ok { t.touchJobLocked(mediaID) t.mu.Unlock() return t.PlaylistPath(mediaID), nil } if max := t.maxConcurrent(); max > 0 && len(t.jobs) >= max { t.mu.Unlock() return "", ErrTranscodeBusy } outDir := t.HLSDir(mediaID) if err := os.MkdirAll(outDir, 0o750); err != nil { t.mu.Unlock() return "", err } jobCtx, cancel := context.WithCancel(context.Background()) job := &hlsJob{ mediaID: mediaID, outputDir: outDir, cancel: cancel, startedAt: time.Now(), lastAccess: time.Now(), encoder: t.effectiveEncoder(), } t.jobs[mediaID] = job t.mu.Unlock() go t.monitorIdle(jobCtx, job) go t.runFFmpeg(jobCtx, job, m.Path) return t.PlaylistPath(mediaID), nil } // WaitReady blocks (with a deadline) until the playlist file shows up on // disk. Returns true on success. func (t *TranscoderService) WaitReady(ctx context.Context, mediaID string, timeout time.Duration) bool { deadline := time.Now().Add(timeout) for { if _, err := os.Stat(t.PlaylistPath(mediaID)); err == nil { t.mu.Lock() if j, ok := t.jobs[mediaID]; ok { j.playlistOK = true } t.mu.Unlock() return true } if time.Now().After(deadline) || ctx.Err() != nil { return false } select { case <-ctx.Done(): return false case <-time.After(250 * time.Millisecond): } } } // StopJob cancels a running ffmpeg process for mediaID, if any. func (t *TranscoderService) StopJob(mediaID string) { t.mu.Lock() defer t.mu.Unlock() if j, ok := t.jobs[mediaID]; ok { j.cancel() delete(t.jobs, mediaID) } } // TouchJob records client activity for the HLS playlist or segment. The idle // watchdog uses it to stop ffmpeg soon after the player is closed or switches // back to direct play. func (t *TranscoderService) TouchJob(mediaID string) { t.mu.Lock() defer t.mu.Unlock() t.touchJobLocked(mediaID) } func (t *TranscoderService) touchJobLocked(mediaID string) { if j, ok := t.jobs[mediaID]; ok { j.lastAccess = time.Now() } } // StopAll terminates every running transcode (called on graceful shutdown). func (t *TranscoderService) StopAll() { t.mu.Lock() defer t.mu.Unlock() for id, j := range t.jobs { j.cancel() delete(t.jobs, id) } } // ActiveJob is the JSON shape exposed to the React Tasks panel. type ActiveJob struct { MediaID string `json:"media_id"` Encoder string `json:"encoder"` StartedAt time.Time `json:"started_at"` PlaylistOK bool `json:"playlist_ok"` } // Active returns a snapshot of the currently running transcode jobs. func (t *TranscoderService) Active() []ActiveJob { t.mu.Lock() defer t.mu.Unlock() out := make([]ActiveJob, 0, len(t.jobs)) for _, j := range t.jobs { out = append(out, ActiveJob{ MediaID: j.mediaID, Encoder: j.encoder, StartedAt: j.startedAt, PlaylistOK: j.playlistOK, }) } return out } func (t *TranscoderService) maxConcurrent() int { if t.cfg.Transcoder.MaxConcurrent <= 0 { return 1 } return t.cfg.Transcoder.MaxConcurrent } func (t *TranscoderService) idleTimeout() time.Duration { if t.cfg.Transcoder.IdleTimeoutSeconds <= 0 { return 120 * time.Second } return time.Duration(t.cfg.Transcoder.IdleTimeoutSeconds) * time.Second } func (t *TranscoderService) monitorIdle(ctx context.Context, job *hlsJob) { timeout := t.idleTimeout() ticker := time.NewTicker(15 * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: t.mu.Lock() current, ok := t.jobs[job.mediaID] if !ok { t.mu.Unlock() return } idleFor := time.Since(current.lastAccess) t.mu.Unlock() if idleFor >= timeout { t.log.Info("transcode idle timeout", zap.String("media_id", job.mediaID), zap.Duration("idle_for", idleFor), zap.Duration("timeout", timeout), ) t.StopJob(job.mediaID) return } } } } func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, source string) { bin, err := t.resolveFFmpegPath() if err != nil { t.log.Warn("ffmpeg unavailable", zap.String("media_id", job.mediaID), zap.Error(err)) t.mu.Lock() delete(t.jobs, job.mediaID) t.mu.Unlock() t.hub.Publish("transcode", map[string]any{ "media_id": job.mediaID, "status": "error", "error": err.Error(), }) return } playlist := filepath.Join(job.outputDir, "index.m3u8") segments := filepath.Join(job.outputDir, "seg_%05d.ts") args := buildFFmpegArgs(t.cfg, source, playlist, segments) cmd := exec.CommandContext(ctx, bin, args...) // #nosec G204 -- bin is resolved by resolveFFmpegPath and args are passed without a shell. cmd.Stderr = os.Stderr t.log.Info("transcode started", zap.String("media_id", job.mediaID), zap.String("encoder", job.encoder), zap.String("source", source), ) t.hub.Publish("transcode", map[string]any{ "media_id": job.mediaID, "encoder": job.encoder, "status": "started", }) if err := cmd.Run(); err != nil && !errors.Is(ctx.Err(), context.Canceled) { t.log.Warn("ffmpeg exited", zap.String("media_id", job.mediaID), zap.Error(err), ) } t.mu.Lock() delete(t.jobs, job.mediaID) t.mu.Unlock() t.hub.Publish("transcode", map[string]any{ "media_id": job.mediaID, "status": "stopped", "duration": time.Since(job.startedAt).Seconds(), }) } func (t *TranscoderService) resolveFFmpegPath() (string, error) { var lastErr error for _, bin := range executableCandidates(strings.TrimSpace(t.cfg.App.FFmpegPath), "ffmpeg") { if err := validateFFmpegForTranscode(context.Background(), bin, t.effectiveEncoder()); err != nil { lastErr = err continue } t.cfg.App.FFmpegPath = bin return bin, nil } if lastErr != nil { return "", fmt.Errorf("no usable ffmpeg found for HLS transcode: %w", lastErr) } return "", fmt.Errorf("ffmpeg not found in PATH or common local app directories; configure app.ffmpeg_path to an existing local ffmpeg") } func (t *TranscoderService) effectiveEncoder() string { if !t.cfg.Transcoder.HardwareAccel { return "" } return normalizedHardwareEncoder(t.cfg.Transcoder.Encoder) } func normalizedHardwareEncoder(encoder string) string { switch strings.ToLower(strings.TrimSpace(encoder)) { case "nvenc", "qsv", "vaapi": return strings.ToLower(strings.TrimSpace(encoder)) default: return "" } } func validateFFmpegForTranscode(ctx context.Context, bin, encoder string) error { required := requiredVideoEncoder(encoder) out, err := commandOutput(ctx, 8*time.Second, bin, "-hide_banner", "-encoders") if err != nil { return fmt.Errorf("%s cannot list encoders: %w", bin, err) } if !hasFFmpegListEntry(string(out), required) { return fmt.Errorf("%s does not provide required encoder %s", bin, required) } out, err = commandOutput(ctx, 8*time.Second, bin, "-hide_banner", "-muxers") if err != nil || !hasFFmpegListEntry(string(out), "hls") { out, err = commandOutput(ctx, 8*time.Second, bin, "-hide_banner", "-formats") if err != nil { return fmt.Errorf("%s cannot list muxers/formats: %w", bin, err) } } if !hasFFmpegListEntry(string(out), "hls") { return fmt.Errorf("%s does not provide hls muxer", bin) } return nil } func requiredVideoEncoder(encoder string) string { switch encoder { case "nvenc": return "h264_nvenc" case "qsv": return "h264_qsv" case "vaapi": return "h264_vaapi" default: return "libx264" } } func hasFFmpegListEntry(output, name string) bool { for _, line := range strings.Split(output, "\n") { fields := strings.Fields(line) for _, field := range fields { if field == name { return true } } } return false } // HumanFFmpegProfile is exposed for the admin UI / settings view. func (t *TranscoderService) HumanFFmpegProfile() string { return fmt.Sprintf("ffmpeg=%s, encoder=%s, output=%s", t.cfg.App.FFmpegPath, t.cfg.Transcoder.Encoder, filepath.Join(t.cfg.Cache.CacheDir, "hls")) }