package playback import ( "context" "fmt" "io/fs" "log/slog" "net/url" "os" "os/exec" "path/filepath" "sort" "strconv" "strings" "sync" "time" "github.com/google/uuid" "pipu/api/internal/platform/sign" ) // StreamTTL — срок жизни ссылки на стрим (ADR-006). const StreamTTL = 12 * time.Hour // StreamURL подписывает ссылку на стрим трека для пользователя. Качество клиент добавляет сам (&q=). func StreamURL(s *sign.Signer, userID, trackID uuid.UUID, now time.Time) string { exp := now.Add(StreamTTL).Unix() e := strconv.FormatInt(exp, 10) return "/api/v1/stream/" + trackID.String() + "?u=" + userID.String() + "&exp=" + e + "&sig=" + s.Sign("stream", userID.String(), trackID.String(), e) } // VerifyStream проверяет подпись и срок. func VerifyStream(s *sign.Signer, sig string, userID, trackID uuid.UUID, exp int64, now time.Time) bool { if now.Unix() > exp { return false } return s.Verify(sig, "stream", userID.String(), trackID.String(), strconv.FormatInt(exp, 10)) } // MIME-типы оригиналов. var mimeTypes = map[string]string{ "flac": "audio/flac", "mp3": "audio/mpeg", "m4a": "audio/mp4", "aac": "audio/aac", "ogg": "audio/ogg", "opus": "audio/ogg; codecs=opus", "wav": "audio/wav", "aiff": "audio/aiff", "aif": "audio/aiff", "wv": "audio/x-wavpack", "ape": "audio/x-ape", } // ContentType — MIME оригинала по формату. func ContentType(format string) string { if m, ok := mimeTypes[format]; ok { return m } return "application/octet-stream" } // Transcoder перекодирует в Opus и хранит результат в кэше. Один файл в одном качестве // кодируется не больше одного раза одновременно; общее число ffmpeg ограничено. type Transcoder struct { FFmpeg string Dir string // /cache/transcode LimitGB int log *slog.Logger sem chan struct{} mu sync.Mutex inflight map[string]*flight } type flight struct { done chan struct{} err error } // NewTranscoder создаёт транскодер с concurrency параллельными ffmpeg. func NewTranscoder(ffmpeg, dir string, concurrency, limitGB int, log *slog.Logger) *Transcoder { return &Transcoder{ FFmpeg: ffmpeg, Dir: dir, LimitGB: limitGB, log: log, sem: make(chan struct{}, max(1, concurrency)), inflight: map[string]*flight{}, } } // CacheName — имя файла транскода: зависит от файла и его размера, чтобы изменённый файл // не отдавался из устаревшего кэша. func CacheName(fileID uuid.UUID, size int64, quality string) string { return fmt.Sprintf("%s-%d-%s.opus", fileID, size, quality) } // Ensure возвращает путь к готовому транскоду, кодируя его при необходимости. func (t *Transcoder) Ensure(ctx context.Context, src string, name, quality string) (string, error) { dst := filepath.Join(t.Dir, name) if _, err := os.Stat(dst); err == nil { now := time.Now() _ = os.Chtimes(dst, now, now) // LRU: отмечаем использование return dst, nil } t.mu.Lock() if f, ok := t.inflight[name]; ok { t.mu.Unlock() select { case <-f.done: return dst, f.err case <-ctx.Done(): return "", ctx.Err() } } f := &flight{done: make(chan struct{})} t.inflight[name] = f t.mu.Unlock() f.err = t.encode(src, dst, quality) close(f.done) t.mu.Lock() delete(t.inflight, name) t.mu.Unlock() return dst, f.err } // encode не зависит от контекста запроса: если слушатель ушёл, транскод всё равно пригодится. func (t *Transcoder) encode(src, dst, quality string) error { t.sem <- struct{}{} defer func() { <-t.sem }() if err := os.MkdirAll(t.Dir, 0o755); err != nil { return err } tmp := dst + ".tmp" ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) defer cancel() out, err := exec.CommandContext(ctx, t.FFmpeg, "-v", "error", "-y", "-i", src, "-map", "0:a:0", "-c:a", "libopus", "-b:a", fmt.Sprintf("%dk", opusKbps[quality]), "-vbr", "on", "-f", "ogg", tmp).CombinedOutput() if err != nil { _ = os.Remove(tmp) return fmt.Errorf("ffmpeg opus: %s", strings.TrimSpace(string(out))) } return os.Rename(tmp, dst) } // RunCleanup раз в час удаляет давно не использованные транскоды сверх лимита. func (t *Transcoder) RunCleanup(ctx context.Context) { tick := time.NewTicker(time.Hour) defer tick.Stop() for { t.cleanup() select { case <-ctx.Done(): return case <-tick.C: } } } func (t *Transcoder) cleanup() { type entry struct { path string size int64 at time.Time } var ( files []entry total int64 ) _ = filepath.WalkDir(t.Dir, func(p string, d fs.DirEntry, err error) error { if err != nil || d.IsDir() || !strings.HasSuffix(p, ".opus") { return nil } if info, err := d.Info(); err == nil { files = append(files, entry{p, info.Size(), info.ModTime()}) total += info.Size() } return nil }) limit := int64(t.LimitGB) << 30 if total <= limit { return } sort.Slice(files, func(i, j int) bool { return files[i].at.Before(files[j].at) }) removed := 0 for _, f := range files { if total <= limit { break } if os.Remove(f.path) == nil { total -= f.size removed++ } } t.log.Info("transcode cache trimmed", "removed", removed) } // accelPath — внутренний путь nginx к оригиналу (сегменты экранированы). func accelPath(f StreamFile) string { segs := strings.Split(f.RelPath, "/") if f.SourceType == "local" && f.LocalPath != nil && *f.LocalPath != "" { segs = append(strings.Split(*f.LocalPath, "/"), segs...) } for i, s := range segs { segs[i] = url.PathEscape(s) } if f.SourceType == "local" { return "/_internal/library-host/" + strings.Join(segs, "/") } return "/_internal/library/" + f.SourceID.String() + "/" + strings.Join(segs, "/") }