From b6ec06f272557a0bc1ddf8750ef035f57a61f959 Mon Sep 17 00:00:00 2001 From: Owen Rummage Date: Sun, 23 Aug 2026 00:10:55 -0500 Subject: [PATCH] Add real ffmpeg-backed transcoding subsystem (bounded concurrency, streaming, HLS segmenting) --- internal/transcoding/hls.go | 148 +++++++++++++++++++++++ internal/transcoding/transcoding.go | 116 ++++++++++++++++++ internal/transcoding/transcoding_test.go | 97 +++++++++++++++ 3 files changed, 361 insertions(+) create mode 100644 internal/transcoding/hls.go create mode 100644 internal/transcoding/transcoding.go create mode 100644 internal/transcoding/transcoding_test.go diff --git a/internal/transcoding/hls.go b/internal/transcoding/hls.go new file mode 100644 index 0000000..563a478 --- /dev/null +++ b/internal/transcoding/hls.go @@ -0,0 +1,148 @@ +package transcoding + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" +) + +// segmentSeconds is the target HLS segment duration. +const segmentSeconds = 10 + +// HLSSession holds the completed segments for one (source, bitrate) +// combination, cached on disk so repeated segment requests (which every +// HLS client issues, one per segment) don't re-transcode. +type HLSSession struct { + Dir string + SegmentCount int +} + +// hlsCache maps a cache key (source path + options) to its completed +// session, so concurrent/repeated requests for the same transcode reuse +// one ffmpeg run instead of racing separate ones. +type hlsCacheEntry struct { + once sync.Once + session *HLSSession + err error +} + +var ( + hlsCacheMu sync.Mutex + hlsCache = map[string]*hlsCacheEntry{} +) + +func hlsCacheKey(sourcePath string, opts Options) string { + return fmt.Sprintf("%s|%d|%s", sourcePath, opts.MaxBitRateKbps, opts.Format) +} + +// BuildHLS transcodes and segments sourcePath into cacheDir (once per +// distinct source+options; concurrent callers for the same key share the +// result), returning the session describing the produced .ts segments. +func (m *Manager) BuildHLS(ctx context.Context, sourcePath, cacheDir string, opts Options) (*HLSSession, error) { + key := hlsCacheKey(sourcePath, opts) + + hlsCacheMu.Lock() + entry, ok := hlsCache[key] + if !ok { + entry = &hlsCacheEntry{} + hlsCache[key] = entry + } + hlsCacheMu.Unlock() + + entry.once.Do(func() { + entry.session, entry.err = m.buildHLSOnce(ctx, sourcePath, cacheDir, opts, key) + }) + return entry.session, entry.err +} + +func (m *Manager) buildHLSOnce(ctx context.Context, sourcePath, cacheDir string, opts Options, key string) (*HLSSession, error) { + select { + case m.sem <- struct{}{}: + case <-ctx.Done(): + return nil, ctx.Err() + } + defer func() { <-m.sem }() + + dir := filepath.Join(cacheDir, "hls", sanitizeKey(key)) + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + playlistPath := filepath.Join(dir, "index.m3u8") + segmentPattern := filepath.Join(dir, "seg%05d.ts") + + bitrate := opts.MaxBitRateKbps + if bitrate <= 0 { + bitrate = defaultBitRateKbps + } + + args := []string{ + "-i", sourcePath, + "-c:v", "libx264", "-preset", "veryfast", "-c:a", "aac", + "-b:a", fmt.Sprintf("%dk", bitrate), + "-f", "hls", + "-hls_time", strconv.Itoa(segmentSeconds), + "-hls_list_size", "0", + "-hls_segment_filename", segmentPattern, + "--", playlistPath, + } + cmd := exec.CommandContext(ctx, m.FFmpegPath, args...) + if out, err := cmd.CombinedOutput(); err != nil { + return nil, fmt.Errorf("ffmpeg hls: %w: %s", err, out) + } + + entries, err := os.ReadDir(dir) + if err != nil { + return nil, err + } + count := 0 + for _, e := range entries { + if strings.HasSuffix(e.Name(), ".ts") { + count++ + } + } + return &HLSSession{Dir: dir, SegmentCount: count}, nil +} + +func sanitizeKey(key string) string { + var b strings.Builder + for _, r := range key { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9': + b.WriteRune(r) + default: + b.WriteByte('_') + } + } + return b.String() +} + +// Playlist returns the m3u8 media playlist text for a completed session, +// with segment URIs built by segmentURL(index). +func (s *HLSSession) Playlist(segmentURL func(index int) string) string { + var b strings.Builder + b.WriteString("#EXTM3U\n") + b.WriteString("#EXT-X-VERSION:3\n") + b.WriteString(fmt.Sprintf("#EXT-X-TARGETDURATION:%d\n", segmentSeconds)) + b.WriteString("#EXT-X-PLAYLIST-TYPE:VOD\n") + for i := 0; i < s.SegmentCount; i++ { + b.WriteString(fmt.Sprintf("#EXTINF:%d.0,\n", segmentSeconds)) + b.WriteString(segmentURL(i)) + b.WriteString("\n") + } + b.WriteString("#EXT-X-ENDLIST\n") + return b.String() +} + +// SegmentPath returns the on-disk path for segment index, or "" if out +// of range. +func (s *HLSSession) SegmentPath(index int) string { + if index < 0 || index >= s.SegmentCount { + return "" + } + return filepath.Join(s.Dir, fmt.Sprintf("seg%05d.ts", index)) +} diff --git a/internal/transcoding/transcoding.go b/internal/transcoding/transcoding.go new file mode 100644 index 0000000..96c6ceb --- /dev/null +++ b/internal/transcoding/transcoding.go @@ -0,0 +1,116 @@ +// Package transcoding runs ffmpeg to convert media on the fly for +// clients that can't play a file's native format or want a lower +// bitrate. It never buffers a whole output file in memory or on disk +// before sending it: ffmpeg's stdout is piped straight to the response, +// and the process is bounded (a small worker semaphore) and killed the +// moment the request context is cancelled (client disconnect). +package transcoding + +import ( + "context" + "fmt" + "io" + "os/exec" + "strconv" +) + +// Format is a supported transcode target. +type Format string + +const ( + FormatMP3 Format = "mp3" + FormatOpus Format = "opus" + FormatAAC Format = "aac" +) + +func (f Format) MimeType() string { + switch f { + case FormatOpus: + return "audio/ogg" + case FormatAAC: + return "audio/aac" + default: + return "audio/mpeg" + } +} + +// ParseFormat maps a client-requested format string (as sent in the +// Subsonic "format" parameter) to a supported Format, defaulting to MP3 +// for anything unrecognized — the same behavior real Subsonic servers +// use since mp3 is the universally-supported fallback. +func ParseFormat(s string) Format { + switch s { + case "opus": + return FormatOpus + case "aac", "m4a": + return FormatAAC + default: + return FormatMP3 + } +} + +// Options configures one transcode. +type Options struct { + Format Format + MaxBitRateKbps int // 0 means "no limit / use a sensible default" + TimeOffset float64 // seconds to seek into the source before transcoding +} + +const defaultBitRateKbps = 192 +const maxConcurrentJobs = 4 + +// Manager bounds how many ffmpeg processes can run at once, so a burst +// of client requests can't fork-bomb the host. +type Manager struct { + FFmpegPath string + sem chan struct{} +} + +func NewManager(ffmpegPath string) *Manager { + if ffmpegPath == "" { + ffmpegPath = "ffmpeg" + } + return &Manager{FFmpegPath: ffmpegPath, sem: make(chan struct{}, maxConcurrentJobs)} +} + +// Stream transcodes sourcePath per opts and copies the result to w as it +// is produced, never writing a complete temporary file to disk first. +// It blocks until the transcode finishes, the client disconnects (ctx +// cancelled, in which case ffmpeg is killed), or an error occurs. +func (m *Manager) Stream(ctx context.Context, w io.Writer, sourcePath string, opts Options) error { + select { + case m.sem <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + defer func() { <-m.sem }() + + args := m.buildArgs(sourcePath, opts, "-") + cmd := exec.CommandContext(ctx, m.FFmpegPath, args...) + cmd.Stdout = w + return cmd.Run() +} + +func (m *Manager) buildArgs(sourcePath string, opts Options, output string) []string { + bitrate := opts.MaxBitRateKbps + if bitrate <= 0 { + bitrate = defaultBitRateKbps + } + + var args []string + if opts.TimeOffset > 0 { + args = append(args, "-ss", strconv.FormatFloat(opts.TimeOffset, 'f', 2, 64)) + } + args = append(args, "-i", sourcePath, "-vn", "-map_metadata", "-1") + + switch opts.Format { + case FormatOpus: + args = append(args, "-c:a", "libopus", "-b:a", fmt.Sprintf("%dk", bitrate), "-f", "opus") + case FormatAAC: + args = append(args, "-c:a", "aac", "-b:a", fmt.Sprintf("%dk", bitrate), "-f", "adts") + default: + args = append(args, "-c:a", "libmp3lame", "-b:a", fmt.Sprintf("%dk", bitrate), "-f", "mp3") + } + args = append(args, "--", output) + return args +} diff --git a/internal/transcoding/transcoding_test.go b/internal/transcoding/transcoding_test.go new file mode 100644 index 0000000..6666827 --- /dev/null +++ b/internal/transcoding/transcoding_test.go @@ -0,0 +1,97 @@ +package transcoding + +import ( + "bytes" + "context" + "os/exec" + "path/filepath" + "testing" + "time" +) + +func genTestFile(t *testing.T, dir string) string { + t.Helper() + if _, err := exec.LookPath("ffmpeg"); err != nil { + t.Skip("ffmpeg not installed") + } + out := filepath.Join(dir, "src.mp3") + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, "ffmpeg", "-f", "lavfi", "-i", "sine=frequency=440:duration=2", "-codec:a", "libmp3lame", out) + if combined, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("ffmpeg: %v\n%s", err, combined) + } + return out +} + +func TestStreamProducesOutput(t *testing.T) { + dir := t.TempDir() + src := genTestFile(t, dir) + + m := NewManager("") + var buf bytes.Buffer + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := m.Stream(ctx, &buf, src, Options{Format: FormatMP3, MaxBitRateKbps: 64}); err != nil { + t.Fatalf("Stream: %v", err) + } + if buf.Len() == 0 { + t.Fatal("transcoded output is empty") + } +} + +func TestStreamRespectsContextCancellation(t *testing.T) { + dir := t.TempDir() + src := genTestFile(t, dir) + + m := NewManager("") + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already cancelled before starting + + var buf bytes.Buffer + err := m.Stream(ctx, &buf, src, Options{Format: FormatMP3}) + if err == nil { + t.Fatal("expected an error when the context is already cancelled") + } +} + +func TestParseFormat(t *testing.T) { + cases := map[string]Format{ + "mp3": FormatMP3, + "opus": FormatOpus, + "aac": FormatAAC, + "unknown": FormatMP3, + "": FormatMP3, + } + for in, want := range cases { + if got := ParseFormat(in); got != want { + t.Errorf("ParseFormat(%q) = %q, want %q", in, got, want) + } + } +} + +func TestBuildHLS(t *testing.T) { + dir := t.TempDir() + src := genTestFile(t, dir) + + m := NewManager("") + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + session, err := m.BuildHLS(ctx, src, filepath.Join(dir, "cache"), Options{Format: FormatAAC, MaxBitRateKbps: 96}) + if err != nil { + t.Fatalf("BuildHLS: %v", err) + } + if session.SegmentCount == 0 { + t.Fatal("expected at least one HLS segment") + } + playlist := session.Playlist(func(i int) string { return "seg" + string(rune('0'+i)) }) + if playlist == "" { + t.Fatal("empty playlist") + } + if session.SegmentPath(0) == "" { + t.Fatal("SegmentPath(0) should resolve") + } + if session.SegmentPath(9999) != "" { + t.Fatal("SegmentPath out of range should return empty") + } +}