1// Package stream turns a tuner input or a recording into HLS with ffmpeg.2// Each client (a desktop app, a browser) gets its own session; starting a3// new stream for a client replaces its previous one.4package stream56import (7 "bufio"8 "bytes"9 "context"10 "crypto/rand"11 "encoding/hex"12 "encoding/json"13 "errors"14 "fmt"15 "io"16 "log"17 "net"18 "net/http"19 "os"20 "os/exec"21 "path/filepath"22 "strconv"23 "strings"24 "sync"25 "sync/atomic"26 "time"2728 "airwaves/internal/phase"29 "airwaves/internal/tuner"30)3132// maxSessions bounds concurrent transcodes; the oldest is evicted.33const maxSessions = 43435// IdleAfter is how long a stream lasts without a request for its files or36// a Touch: a player gone (an app quit, a TV switched off) frees its tuner.37const IdleAfter = 2 * time.Minute3839// Playback locates a running stream.40type Playback struct {41 ID string `json:"id"`42 // URL is absolute when the server can tell (loopback mode, or rewritten43 // by a remote client).44 URL string `json:"url"`45 // Path is the playlist relative to the server's origin.46 Path string `json:"path"`47 Note string `json:"note,omitempty"`48 // Offset is where a recording's playback starts, in seconds, so the49 // player can show positions in the whole recording.50 Offset float64 `json:"offset,omitempty"`51 // Timing lists when each step of starting the stream happened, from52 // the tune call (see phase).53 Timing []phase.Mark `json:"timing,omitempty"`54}5556// Server owns the ffmpeg sessions and serves their files under /live/.57type Server struct {58 ffmpeg string59 root string60 ln net.Listener61 srv *http.Server6263 mu sync.Mutex64 sessions map[string]*session // by client65 idle time.Duration // IdleAfter, but in tests66 quit chan struct{} // closed by Close, ending the reaper67}6869type session struct {70 id string71 client string72 dir string73 started time.Time74 used atomic.Int64 // when last used, in Unix nanoseconds75 cancel context.CancelFunc76 done chan struct{}77 stderr *tail78}7980func (sess *session) touch(t time.Time) { sess.used.Store(t.UnixNano()) }8182func (sess *session) idleFor(now time.Time) time.Duration {83 return now.Sub(time.Unix(0, sess.used.Load()))84}8586// FindFFmpeg locates the ffmpeg binary.87func FindFFmpeg() (string, error) {88 if bin, err := exec.LookPath("ffmpeg"); err == nil {89 return bin, nil90 }91 for _, p := range []string{"/opt/homebrew/bin/ffmpeg", "/usr/local/bin/ffmpeg", "/usr/bin/ffmpeg"} {92 if _, err := os.Stat(p); err == nil {93 return p, nil94 }95 }96 return "", errors.New("ffmpeg not found; install it with `brew install ffmpeg`")97}9899// NewServer prepares a stream server. With loopback set it also listens on100// 127.0.0.1 so a local player can fetch streams directly; otherwise mount101// it on an existing HTTP server.102func NewServer(loopback bool) (*Server, error) {103 bin, err := FindFFmpeg()104 if err != nil {105 return nil, err106 }107 root, err := os.MkdirTemp("", "airwaves-hls-")108 if err != nil {109 return nil, err110 }111 s := &Server{ffmpeg: bin, root: root, sessions: make(map[string]*session), idle: IdleAfter, quit: make(chan struct{})}112 go s.reap()113 if loopback {114 ln, err := net.Listen("tcp", "127.0.0.1:0")115 if err != nil {116 return nil, err117 }118 s.ln = ln119 s.srv = &http.Server{Handler: s, ReadHeaderTimeout: 5 * time.Second}120 go func() { _ = s.srv.Serve(ln) }()121 }122 return s, nil123}124125// FFmpeg returns the ffmpeg binary in use.126func (s *Server) FFmpeg() string { return s.ffmpeg }127128// FFprobe returns the ffprobe binary beside ffmpeg.129func (s *Server) FFprobe() string { return filepath.Join(filepath.Dir(s.ffmpeg), "ffprobe") }130131// Start replaces client's stream with one from in. It returns once the132// first segments are ready, or with ffmpeg's error output. When calls for133// the same client overlap (fast channel surfing) the last one wins and134// earlier ones fail; no ffmpeg process outlives its replacement.135func (s *Server) Start(ctx context.Context, client string, in tuner.Input) (*Playback, error) {136 // The steps are timed from the tune call when its context has a timer.137 t := phase.FromContext(ctx)138 if t == nil {139 t = phase.New("stream", "")140 }141 // Stop first so a network tuner frees its tuner before the next tune.142 if s.stop(client) {143 t.Mark("previous stopped")144 }145146 id := newID()147 dir := filepath.Join(s.root, id)148 if err := os.MkdirAll(dir, 0o755); err != nil {149 return nil, err150 }151 runCtx, cancel := context.WithCancel(phase.NewContext(context.Background(), t))152 // ffmpeg reports its progress on fd 3, which tells when it has153 // encoded its first frame.154 progress, progressW, err := os.Pipe()155 if err != nil {156 cancel()157 return nil, err158 }159 args := append([]string{"-progress", "pipe:3", "-stats_period", "0.1"}, ffmpegArgs(in, dir)...)160 cmd := exec.CommandContext(runCtx, s.ffmpeg, args...)161 cmd.ExtraFiles = []*os.File{progressW}162 sess := &session{163 id: id, client: client, dir: dir, started: time.Now(),164 cancel: cancel, done: make(chan struct{}), stderr: &tail{max: 4096},165 }166 sess.touch(sess.started)167 cmd.Stderr = sess.stderr168 var source io.WriteCloser169 if in.Source != nil {170 if source, err = cmd.StdinPipe(); err != nil {171 progress.Close()172 progressW.Close()173 cancel()174 return nil, err175 }176 }177 err = cmd.Start()178 progressW.Close()179 if err != nil {180 progress.Close()181 cancel()182 return nil, fmt.Errorf("start ffmpeg: %w", err)183 }184 t.Mark("ffmpeg started")185 go watchProgress(progress, func() { t.Mark("first frame") })186 if source != nil {187 go func() {188 defer source.Close()189 w := &firstWrite{w: source, mark: func() { t.Mark("first data") }}190 if err := in.Source(runCtx, w); err != nil && runCtx.Err() == nil {191 fmt.Fprintf(sess.stderr, "%v\n", err)192 }193 }()194 }195 if subtitled(in) {196 sub := newSubtitler(dir, in.Subtitles)197 if in.Copy {198 sub.byAudio()199 }200 go sub.run(runCtx)201 }202 go func() {203 _ = cmd.Wait()204 close(sess.done)205 }()206207 var evict []*session208 s.mu.Lock()209 if prev := s.sessions[client]; prev != nil {210 evict = append(evict, prev)211 }212 s.sessions[client] = sess213 for len(s.sessions) > maxSessions {214 oldest := sess215 for _, x := range s.sessions {216 if x.started.Before(oldest.started) {217 oldest = x218 }219 }220 delete(s.sessions, oldest.client)221 evict = append(evict, oldest)222 }223 s.mu.Unlock()224 for _, x := range evict {225 x.stop()226 }227228 if err := waitReady(ctx, sess, playlists(in), t); err != nil {229 s.mu.Lock()230 if s.sessions[client] == sess {231 delete(s.sessions, client)232 }233 s.mu.Unlock()234 sess.stop()235 log.Printf("%s failed after %.1fs: %s", t.Label, t.Since().Seconds(), t)236 return nil, err237 }238 if err := os.WriteFile(filepath.Join(dir, "master.m3u8"), []byte(masterPlaylist(in)), 0o644); err != nil {239 s.Stop(client)240 return nil, err241 }242 t.Mark("ready")243 pb := &Playback{ID: id, Path: "/live/" + id + "/master.m3u8", Note: in.Note, Offset: in.Offset, Timing: t.Marks()}244 if s.ln != nil {245 pb.URL = "http://" + s.ln.Addr().String() + pb.Path246 }247 log.Printf("%s: %s", t.Label, t)248 phase.Tunes.Add(t.Kind, phase.Ready, t.Since())249 return pb, nil250}251252// firstWrite marks the first write to w.253type firstWrite struct {254 w io.Writer255 once sync.Once256 mark func()257}258259func (f *firstWrite) Write(p []byte) (int, error) {260 f.once.Do(f.mark)261 return f.w.Write(p)262}263264// waitReady blocks until every media playlist lists two segments, marking265// on t when the first segment is done.266func waitReady(ctx context.Context, sess *session, lists []string, t *phase.Timer) error {267 ctx, cancel := context.WithTimeout(ctx, 25*time.Second)268 defer cancel()269 tick := time.NewTicker(50 * time.Millisecond)270 defer tick.Stop()271 for {272 select {273 case <-ctx.Done():274 return fmt.Errorf("stream did not start in time: %s", sess.stderr.String())275 case <-sess.done:276 return fmt.Errorf("ffmpeg exited: %s", sess.stderr.String())277 case <-tick.C:278 ready := 0279 for i, name := range lists {280 b, err := os.ReadFile(filepath.Join(sess.dir, name))281 n := bytes.Count(b, []byte("#EXTINF"))282 if err == nil && i == 0 && n >= 1 {283 t.Mark("first segment")284 }285 if err == nil && n >= 2 {286 ready++287 }288 }289 if ready == len(lists) {290 return nil291 }292 }293 }294}295296// watchProgress reads ffmpeg's progress reports until it exits, calling297// encoded once the first frame is out.298func watchProgress(r io.ReadCloser, encoded func()) {299 defer r.Close()300 sc := bufio.NewScanner(r)301 done := false302 for sc.Scan() {303 if n, ok := strings.CutPrefix(sc.Text(), "frame="); ok && !done {304 if frames, _ := strconv.Atoi(strings.TrimSpace(n)); frames > 0 {305 encoded()306 done = true307 }308 }309 }310 // ffmpeg must never wait on a report nobody reads.311 _, _ = io.Copy(io.Discard, r)312}313314// liveWindow is how far back a viewer can pause or rewind live TV.315const liveWindow = 30 * time.Minute316317// variants is how many media playlists ffmpeg writes: the video (with318// its audio when there is a single track), plus one per audio track when319// there are several.320func variants(in tuner.Input) int {321 if len(in.Audio) > 1 {322 return 1 + len(in.Audio)323 }324 return 1325}326327// playlists names a session's media playlists: ffmpeg's, and the328// subtitles' when it has them.329func playlists(in tuner.Input) []string {330 var out []string331 for i := range variants(in) {332 out = append(out, fmt.Sprintf("stream_%d.m3u8", i))333 }334 if subtitled(in) {335 out = append(out, subsPlaylist)336 }337 return out338}339340// subtitled reports whether a session has a WebVTT subtitle rendition:341// live streams with their captions as text, and podcasts' episodes (a342// recording's are in its picture).343func subtitled(in tuner.Input) bool { return in.Subtitles != nil && (!in.VOD || in.Copy) }344345// ffmpegArgs transcodes to H.264 and AAC in HLS, or copies a picture346// that's H.264 already (Copy). Several audio tracks become separate347// renditions ("v:0 a:0 a:1"); closed captions (CEA-608 in the picture of a348// broadcast or a custom channel) are carried into the H.264 stream by349// libx264.350func ffmpegArgs(in tuner.Input, dir string) []string {351 args := []string{"-hide_banner", "-loglevel", "error", "-nostdin"}352 args = append(args, probeArgs(in)...)353 args = append(args, in.Args...)354 args = append(args, "-map", in.Video)355 for _, a := range in.Audio {356 args = append(args, "-map", a.Map)357 }358 switch {359 case in.Copy:360 // Segments then start at the picture's own key frames.361 args = append(args, "-c:v", "copy")362 case in.Broadcast:363 // Deinterlace only frames flagged interlaced (1080i, 480i).364 args = append(args, "-vf", "bwdif=mode=send_frame:deint=interlaced")365 }366 if !in.Copy {367 args = append(args,368 "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-pix_fmt", "yuv420p",369 "-profile:v", "high", "-crf", "21", "-maxrate", "8M", "-bufsize", "12M",370 "-g", "60", "-keyint_min", "60", "-sc_threshold", "0",371 "-force_key_frames", "expr:gte(t,n_forced*2)", "-a53cc", "1",372 )373 }374 streamMap := "v:0"375 switch len(in.Audio) {376 case 0:377 case 1:378 streamMap += ",a:0"379 default:380 for i := range in.Audio {381 streamMap += fmt.Sprintf(" a:%d", i)382 }383 }384 if len(in.Audio) > 0 {385 args = append(args, "-c:a", "aac", "-b:a", "160k", "-ac", "2", "-ar", "48000")386 }387 args = append(args, "-f", "hls", "-var_stream_map", streamMap)388 if in.VOD {389 // Keep every segment so the player can seek back to the start.390 args = append(args, "-hls_time", "4", "-hls_list_size", "0",391 "-hls_playlist_type", "event", "-hls_flags", "independent_segments+temp_file")392 } else {393 // A sliding window long enough to pause and rewind live TV.394 segments := int(liveWindow / (2 * time.Second))395 args = append(args, "-hls_time", "2", "-hls_list_size", strconv.Itoa(segments),396 "-hls_flags", "delete_segments+independent_segments+omit_endlist+temp_file")397 }398 return append(args,399 "-hls_segment_filename", filepath.Join(dir, "seg_%v_%05d.ts"),400 filepath.Join(dir, "stream_%v.m3u8"),401 )402}403404// probeArgs keep ffmpeg's first look at a live input short. Before it405// starts, ffmpeg reads an MPEG-TS input for as long as it may analyze it406// (5 seconds by default) even once it knows every stream, and for a live407// input that is as long a wait. Half a second finds a broadcast's audio408// tracks and a custom channel's picture and sound. A broadcast's picture409// size may only come with its first full frame, later, which ffmpeg takes410// in stride. A recording is read as fast as it can be, so it keeps the411// default.412func probeArgs(in tuner.Input) []string {413 if in.VOD {414 return nil415 }416 return []string{"-probesize", "2000000", "-analyzeduration", "500000"}417}418419// masterPlaylist describes the session's streams to the player: the video,420// each audio track as a selectable rendition, and closed captions.421func masterPlaylist(in tuner.Input) string {422 var b strings.Builder423 b.WriteString("#EXTM3U\n#EXT-X-VERSION:6\n#EXT-X-INDEPENDENT-SEGMENTS\n")424 codecs, attrs := "avc1.640028", ""425 if len(in.Audio) > 0 {426 codecs += ",mp4a.40.2"427 }428 if len(in.Audio) > 1 {429 for i, a := range in.Audio {430 name, lang := TrackName(a, i)431 def := "NO"432 if i == 0 {433 def = "YES"434 }435 extra := ""436 if a.Described {437 extra = `,CHARACTERISTICS="public.accessibility.describes-video"`438 }439 fmt.Fprintf(&b, "#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"aud\",NAME=%q,LANGUAGE=%q,DEFAULT=%s,AUTOSELECT=YES%s,URI=\"stream_%d.m3u8\"\n",440 name, lang, def, extra, i+1)441 }442 attrs += `,AUDIO="aud"`443 }444 // Captions are never picked by the player (AUTOSELECT=NO): the app turns445 // them on and off, as the viewer asked or the program needs.446 switch {447 case subtitled(in):448 // Text rather than the same captions in the picture: one track.449 fmt.Fprintf(&b, "#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID=\"subs\",NAME=\"English\",LANGUAGE=\"en\",DEFAULT=NO,AUTOSELECT=NO,FORCED=NO,URI=%q\n", subsPlaylist)450 attrs += `,SUBTITLES="subs",CLOSED-CAPTIONS=NONE`451 case in.Broadcast || in.Captions:452 b.WriteString(`#EXT-X-MEDIA:TYPE=CLOSED-CAPTIONS,GROUP-ID="cc",NAME="English",LANGUAGE="en",INSTREAM-ID="CC1",DEFAULT=NO,AUTOSELECT=NO` + "\n")453 attrs += `,CLOSED-CAPTIONS="cc"`454 default:455 attrs += ",CLOSED-CAPTIONS=NONE"456 }457 fmt.Fprintf(&b, "#EXT-X-STREAM-INF:BANDWIDTH=8000000,CODECS=%q%s\nstream_0.m3u8\n", codecs, attrs)458 return b.String()459}460461// languages maps the ISO 639-2 codes broadcasters use to BCP 47 tags and462// display names.463var languages = map[string][2]string{464 "eng": {"en", "English"}, "spa": {"es", "Spanish"}, "fre": {"fr", "French"}, "fra": {"fr", "French"},465 "kor": {"ko", "Korean"}, "vie": {"vi", "Vietnamese"}, "chi": {"zh", "Chinese"}, "zho": {"zh", "Chinese"},466 "rus": {"ru", "Russian"}, "por": {"pt", "Portuguese"}, "ger": {"de", "German"}, "deu": {"de", "German"},467 "ita": {"it", "Italian"}, "jpn": {"ja", "Japanese"}, "tgl": {"tl", "Tagalog"}, "ara": {"ar", "Arabic"},468 "hin": {"hi", "Hindi"},469}470471// TrackName labels an audio track for the player: "Spanish",472// "English (Described)", or "Audio 2" when the broadcast gives no language.473func TrackName(a tuner.AudioTrack, i int) (name, lang string) {474 l, ok := languages[strings.ToLower(a.Lang)]475 switch {476 case ok:477 name, lang = l[1], l[0]478 case a.Lang != "":479 name, lang = strings.ToUpper(a.Lang), strings.ToLower(a.Lang)480 default:481 name, lang = fmt.Sprintf("Audio %d", i+1), "und"482 }483 if a.Described {484 name += " (Described)"485 }486 return name, lang487}488489// Stop ends client's stream and removes its files.490func (s *Server) Stop(client string) { s.stop(client) }491492// Touch marks client's stream as in use, for a player holding one without493// fetching from it (paused, say), and reports whether client has one.494func (s *Server) Touch(client string) bool {495 s.mu.Lock()496 defer s.mu.Unlock()497 sess := s.sessions[client]498 if sess != nil {499 sess.touch(time.Now())500 }501 return sess != nil502}503504// reap stops streams left idle, until Close.505func (s *Server) reap() {506 for {507 s.mu.Lock()508 every := s.idle / 4509 s.mu.Unlock()510 select {511 case <-s.quit:512 return513 case now := <-time.After(every):514 s.reapIdle(now)515 }516 }517}518519// reapIdle stops the streams idle longer than the server allows at now.520func (s *Server) reapIdle(now time.Time) {521 var idle []*session522 s.mu.Lock()523 for client, sess := range s.sessions {524 if sess.idleFor(now) > s.idle {525 delete(s.sessions, client)526 idle = append(idle, sess)527 }528 }529 s.mu.Unlock()530 for _, sess := range idle {531 log.Printf("stream: %s idle, stopped", sess.client)532 sess.stop()533 }534}535536// stop ends client's stream, reporting whether it had one.537func (s *Server) stop(client string) bool {538 s.mu.Lock()539 sess := s.sessions[client]540 delete(s.sessions, client)541 s.mu.Unlock()542 if sess != nil {543 sess.stop()544 }545 return sess != nil546}547548func (sess *session) stop() {549 sess.cancel()550 select {551 case <-sess.done:552 case <-time.After(3 * time.Second):553 }554 _ = os.RemoveAll(sess.dir)555}556557// Close stops every stream and the loopback listener.558func (s *Server) Close() {559 s.mu.Lock()560 all := make([]*session, 0, len(s.sessions))561 for _, x := range s.sessions {562 all = append(all, x)563 }564 s.sessions = make(map[string]*session)565 select {566 case <-s.quit:567 default:568 close(s.quit)569 }570 s.mu.Unlock()571 for _, x := range all {572 x.stop()573 }574 if s.srv != nil {575 _ = s.srv.Close()576 }577 _ = os.RemoveAll(s.root)578}579580// ServeHTTP serves /live/<id>/<file> for live sessions.581func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {582 w.Header().Set("Access-Control-Allow-Origin", "*")583 rest, ok := strings.CutPrefix(r.URL.Path, "/live/")584 if !ok {585 http.NotFound(w, r)586 return587 }588 id, file, ok := strings.Cut(rest, "/")589 var sess *session590 s.mu.Lock()591 for _, x := range s.sessions {592 if x.id == id {593 sess = x594 }595 }596 s.mu.Unlock()597 if !ok || sess == nil || file != filepath.Base(file) {598 http.NotFound(w, r)599 return600 }601 sess.touch(time.Now())602 switch filepath.Ext(file) {603 case ".m3u8":604 w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")605 w.Header().Set("Cache-Control", "no-cache")606 case ".ts":607 w.Header().Set("Content-Type", "video/mp2t")608 case ".vtt":609 w.Header().Set("Content-Type", "text/vtt; charset=utf-8")610 default:611 http.NotFound(w, r)612 return613 }614 http.ServeFile(w, r, filepath.Join(sess.dir, file))615}616617func newID() string {618 b := make([]byte, 8)619 _, _ = rand.Read(b)620 return hex.EncodeToString(b)621}622623// tail keeps the last max bytes written, for error messages.624type tail struct {625 mu sync.Mutex626 max int627 buf []byte628}629630func (t *tail) Write(p []byte) (int, error) {631 t.mu.Lock()632 defer t.mu.Unlock()633 t.buf = append(t.buf, p...)634 if len(t.buf) > t.max {635 t.buf = t.buf[len(t.buf)-t.max:]636 }637 return len(p), nil638}639640func (t *tail) String() string {641 t.mu.Lock()642 defer t.mu.Unlock()643 s := strings.TrimSpace(string(t.buf))644 if s == "" {645 return "no output"646 }647 return s648}649650// ProbeAudio lists the audio tracks of a recording with ffprobe, in stream651// order. AC-4 tracks (ATSC 3.0) cannot be decoded and are skipped.652func (s *Server) ProbeAudio(ctx context.Context, url string) ([]tuner.AudioTrack, error) {653 ctx, cancel := context.WithTimeout(ctx, 20*time.Second)654 defer cancel()655 out, err := exec.CommandContext(ctx, s.FFprobe(), "-v", "error", "-seekable", "0",656 "-probesize", "8M", "-analyzeduration", "5M", "-select_streams", "a",657 "-show_entries", "stream=codec_name:stream_tags=language:stream_disposition=visual_impaired",658 "-of", "json", url).Output()659 if err != nil {660 return nil, fmt.Errorf("probe recording: %w", err)661 }662 var body struct {663 Streams []struct {664 Codec string `json:"codec_name"`665 Tags map[string]string `json:"tags"`666 Disposition map[string]int `json:"disposition"`667 } `json:"streams"`668 }669 if err := json.Unmarshal(out, &body); err != nil {670 return nil, fmt.Errorf("probe recording: %w", err)671 }672 var tracks []tuner.AudioTrack673 for i, st := range body.Streams {674 if st.Codec == "" || st.Codec == "ac4" {675 continue676 }677 tracks = append(tracks, tuner.AudioTrack{678 Map: fmt.Sprintf("0:a:%d", i), Lang: st.Tags["language"], Described: st.Disposition["visual_impaired"] == 1,679 })680 }681 return tracks, nil682}