1package dvr23import (4 "cmp"5 "context"6 "errors"7 "fmt"8 "log"9 "os"10 "path/filepath"11 "slices"12 "strings"13 "sync"14 "time"15 "unicode/utf8"1617 "airwaves/internal/guide"18 "airwaves/internal/lineup"19 "airwaves/internal/tuner"20)2122// Why a recording ended before its time.23var (24 errStopped = errors.New("stopped")25 errDiscard = errors.New("discarded")26 errShutdown = errors.New("the server stopped")27)2829// retry is how long a recording waits after losing its tuner or signal30// before trying again.31var retry = 5 * time.Second3233// job is a recording in progress.34type job struct {35 id string36 ctx context.Context37 cancel context.CancelCauseFunc38 extend chan struct{}39 done chan struct{}4041 mu sync.Mutex42 written int6443 began time.Time // first data44 last time.Time // latest data45 // covered is how much time the stretches so far recorded, and cur the46 // stretch now.47 covered, cur time.Duration48 trouble string // the latest problem49}5051func (j *job) size() int64 {52 j.mu.Lock()53 defer j.mu.Unlock()54 return j.written55}5657func (j *job) problem() string {58 j.mu.Lock()59 defer j.mu.Unlock()60 return j.trouble61}6263func (j *job) note(err error) {64 j.mu.Lock()65 j.trouble = err.Error()66 j.mu.Unlock()67}6869// Run starts recordings as their airings begin, until ctx ends. It first70// picks up recordings the server was making when it last stopped.71func (m *Manager) Run(ctx context.Context, listings Listings) {72 m.resume(time.Now())73 tick := time.NewTicker(10 * time.Second)74 defer tick.Stop()75 for {76 if g, chans, err := listings(ctx); err == nil {77 m.Start(g, chans, time.Now())78 }79 select {80 case <-ctx.Done():81 return82 case <-tick.C:83 }84 }85}8687// Close stops the recordings in progress, keeping them to pick up again.88func (m *Manager) Close() {89 m.cancel(errShutdown)90 m.mu.Lock()91 var jobs []*job92 for _, j := range m.jobs {93 jobs = append(jobs, j)94 }95 m.mu.Unlock()96 for _, j := range jobs {97 <-j.done98 }99}100101// Start begins recording every airing the rules want that's on at now.102func (m *Manager) Start(g *guide.Guide, chans []lineup.Channel, now time.Time) {103 m.mu.Lock()104 defer m.mu.Unlock()105 if m.ctx.Err() != nil {106 return107 }108 n := len(m.st.Rules)109 // Once rules whose airing is over are done.110 m.st.Rules = slices.DeleteFunc(m.st.Rules, func(r Rule) bool { return r.Kind == Once && !r.End.Add(lateSports).After(now) })111 changed := len(m.st.Rules) != n112 for _, a := range m.desired(g, chans, now, m.recorded()) {113 from, until := a.window()114 if now.Before(from) || m.has(a.key) {115 continue116 }117 r := Rec{118 ID: newID(), Key: a.key, RuleID: a.rule.ID, Title: a.p.Title, Subtitle: a.p.EpisodeTitle,119 Description: a.p.Description, Image: cmp.Or(a.p.Image, a.rule.Image), ProgramID: a.p.ProgramID,120 Channel: a.ch.Number, CallSign: a.ch.CallSign, Start: a.p.Start, End: a.p.End, From: from, Until: until,121 Status: Recording,122 }123 r.File = m.fileName(r)124 m.st.Recordings = append(m.st.Recordings, r)125 m.run(r)126 changed = true127 log.Printf("recording %s on %s until %s", r.Title, r.Channel, until.Format(time.Kitchen))128 }129 if changed {130 m.saveLogged()131 }132}133134// resume picks up the recordings that were in progress when the server135// stopped: those whose time isn't up carry on, appending to their files;136// the rest are finished as they are.137func (m *Manager) resume(now time.Time) {138 m.mu.Lock()139 defer m.mu.Unlock()140 changed := false141 for i := range m.st.Recordings {142 r := &m.st.Recordings[i]143 if r.Status != Recording || m.jobs[r.ID] != nil {144 continue145 }146 changed = true147 fi, err := os.Stat(filepath.Join(m.dir, r.File))148 if err == nil {149 r.Size = fi.Size()150 }151 if now.Before(r.Until) {152 log.Printf("recording %s again, after the server stopped", r.Title)153 m.run(*r)154 continue155 }156 if r.Size == 0 {157 r.Status, r.Detail = Failed, "The server stopped before it recorded anything"158 continue159 }160 r.Status, r.Ended = Completed, fi.ModTime()161 r.Detail = describe(*r, "the server stopped", false)162 }163 if changed {164 m.saveLogged()165 }166}167168// run starts recording r. Called with m.mu held.169func (m *Manager) run(r Rec) {170 ctx, cancel := context.WithCancelCause(m.ctx)171 j := &job{id: r.ID, ctx: ctx, cancel: cancel, extend: make(chan struct{}, 1), done: make(chan struct{}),172 written: r.Size, began: r.Began, last: r.Ended, covered: r.Covered}173 m.jobs[r.ID] = j174 go m.record(j, r)175}176177// until is when the recording id stops.178func (m *Manager) until(id string) time.Time {179 m.mu.Lock()180 defer m.mu.Unlock()181 for _, r := range m.st.Recordings {182 if r.ID == id {183 return r.Until184 }185 }186 return time.Time{}187}188189// record records r until its time is up or it's stopped. Losing the tuner190// or the signal makes a gap, not an end: it tries again until its time is191// up.192func (m *Manager) record(j *job, r Rec) {193 defer close(j.done)194 path := filepath.Join(m.dir, r.File)195 if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {196 j.note(err)197 } else {198 for j.ctx.Err() == nil {199 left := time.Until(m.until(j.id))200 if left <= 0 {201 break202 }203 if err := m.stretch(j, r, path); err != nil && j.ctx.Err() == nil {204 j.note(err)205 log.Printf("recording %s: %v", r.Title, err)206 select {207 case <-j.ctx.Done():208 case <-j.extend:209 case <-time.After(min(retry, left)):210 }211 }212 }213 }214 m.finish(j)215}216217// stretch records r on one tuner for as long as it can: to its end, or218// until the tuner or the signal is lost. Extending the recording moves219// the end while it records.220func (m *Manager) stretch(j *job, r Rec, path string) error {221 ctx := j.ctx222 freq, program, err := m.locate(ctx, r.Channel)223 if err != nil {224 return err225 }226 sub, err := m.tuners.Open(ctx, tuner.Request{FrequencyHz: freq, Program: program, Priority: tuner.Recording, Who: r.Title})227 if err != nil {228 return err229 }230 defer sub.Close()231 wait, cancel := context.WithTimeout(ctx, 15*time.Second)232 _, err = sub.Wait(wait)233 cancel()234 if errors.Is(err, context.DeadlineExceeded) {235 return fmt.Errorf("no signal on %s", r.Channel)236 }237 if err != nil {238 return err239 }240 f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)241 if err != nil {242 return err243 }244 w := &stretchWriter{f: f, j: j, began: func(at time.Time) { m.began(j.id, at) }}245 defer func() {246 w.end()247 f.Close()248 }()249 for {250 until := m.until(j.id)251 part, stop := context.WithDeadline(ctx, until)252 go func() {253 select {254 case <-j.extend:255 stop()256 case <-part.Done():257 }258 }()259 err := sub.Copy(part, w)260 stop()261 switch {262 case ctx.Err() != nil:263 return nil // stopped264 case !time.Now().Before(m.until(j.id)):265 return nil // done266 case errors.Is(err, context.Canceled) && sub.Err() == nil:267 continue // extended268 }269 if err == nil {270 err = errors.New("the tuner stopped")271 }272 return err273 }274}275276// began notes when a recording's first data arrived.277func (m *Manager) began(id string, at time.Time) {278 m.mu.Lock()279 defer m.mu.Unlock()280 for i := range m.st.Recordings {281 if r := &m.st.Recordings[i]; r.ID == id && r.Began.IsZero() {282 r.Began = at283 m.saveLogged()284 }285 }286}287288// stretchWriter writes a recording's file and counts what it wrote.289type stretchWriter struct {290 f *os.File291 j *job292 first time.Time293 began func(time.Time)294}295296func (w *stretchWriter) Write(p []byte) (int, error) {297 n, err := w.f.Write(p)298 now := time.Now()299 w.j.mu.Lock()300 first := w.j.began.IsZero()301 if first {302 w.j.began = now303 }304 if w.first.IsZero() {305 w.first = now306 }307 w.j.written += int64(n)308 w.j.last, w.j.cur = now, now.Sub(w.first)309 w.j.mu.Unlock()310 if first {311 w.began(now)312 }313 return n, err314}315316// end adds the stretch to what's covered.317func (w *stretchWriter) end() {318 w.j.mu.Lock()319 w.j.covered += w.j.cur320 w.j.cur = 0321 w.j.mu.Unlock()322}323324// finish records how a recording went, once it has ended.325func (m *Manager) finish(j *job) {326 cause := context.Cause(j.ctx)327 m.mu.Lock()328 defer m.mu.Unlock()329 delete(m.jobs, j.id)330 i := slices.IndexFunc(m.st.Recordings, func(r Rec) bool { return r.ID == j.id })331 if i < 0 || errors.Is(cause, errDiscard) {332 return // DeleteEntry removes it333 }334 r := &m.st.Recordings[i]335 j.mu.Lock()336 r.Size, r.Began, r.Ended, r.Covered = j.written, j.began, j.last, j.covered+j.cur337 trouble := j.trouble338 j.mu.Unlock()339 switch {340 case errors.Is(cause, errShutdown):341 // Still recording, as far as the next start is concerned.342 case r.Size == 0:343 r.Status, r.Detail = Failed, "Nothing was recorded"344 if trouble != "" {345 r.Detail += ": " + trouble346 }347 log.Printf("recording %s failed: %s", r.Title, r.Detail)348 default:349 r.Status = Completed350 r.Detail = describe(*r, trouble, errors.Is(cause, errStopped))351 log.Printf("recorded %s (%d MB)%s", r.Title, r.Size>>20, cmp.Or(": "+r.Detail, ""))352 }353 m.saveLogged()354}355356// describe says what a finished recording is missing, if anything: a late357// start, an early stop, gaps. trouble is the latest problem it had.358func describe(r Rec, trouble string, stopped bool) string {359 mins := func(d time.Duration) string {360 n := int(d.Round(time.Minute).Minutes())361 if n == 1 {362 return "1 minute"363 }364 return fmt.Sprintf("%d minutes", n)365 }366 var out []string367 if d := r.Began.Sub(r.Start); !r.Began.IsZero() && d >= time.Minute {368 out = append(out, "Began "+mins(d)+" into the program")369 }370 if d := r.End.Sub(r.Ended); !r.Ended.IsZero() && d >= time.Minute {371 if stopped {372 out = append(out, "Stopped "+mins(d)+" before the end")373 } else {374 out = append(out, "Ends "+mins(d)+" before the program did")375 }376 }377 if gap := r.Ended.Sub(r.Began) - r.Covered; r.Covered > 0 && gap >= time.Minute {378 out = append(out, "About "+mins(gap)+" are missing in between")379 }380 if len(out) > 0 && trouble != "" && !stopped {381 out[len(out)-1] += ": " + trouble382 }383 return strings.Join(out, ". ")384}385386// Stop ends a recording in progress early, keeping what it recorded.387func (m *Manager) Stop(id string) error {388 m.mu.Lock()389 j := m.jobs[id]390 m.mu.Unlock()391 if j == nil {392 return fmt.Errorf("%s isn't recording", id)393 }394 j.cancel(errStopped)395 <-j.done396 return nil397}398399// Extend makes a recording in progress run d longer.400func (m *Manager) Extend(id string, d time.Duration) (time.Time, error) {401 if d <= 0 || d > 6*time.Hour {402 return time.Time{}, fmt.Errorf("can't extend a recording by %s", d)403 }404 m.mu.Lock()405 defer m.mu.Unlock()406 j := m.jobs[id]407 i := slices.IndexFunc(m.st.Recordings, func(r Rec) bool { return r.ID == id })408 if j == nil || i < 0 {409 return time.Time{}, fmt.Errorf("%s isn't recording", id)410 }411 r := &m.st.Recordings[i]412 r.Until = r.Until.Add(d)413 select {414 case j.extend <- struct{}{}:415 default:416 }417 return r.Until, m.save()418}419420// fileName picks r's file in the recordings folder: a folder per show,421// named by date and episode, and never one already there. Called with422// m.mu held.423func (m *Manager) fileName(r Rec) string {424 name := r.Start.Local().Format("2006-01-02 15.04")425 if s := clean(r.Subtitle); s != "" {426 name += " " + s427 }428 base := filepath.Join(cmp.Or(clean(r.Title), "Recording"), name)429 for n := 1; ; n++ {430 f := base + ".ts"431 if n > 1 {432 f = fmt.Sprintf("%s (%d).ts", base, n)433 }434 _, err := os.Stat(filepath.Join(m.dir, f))435 taken := slices.ContainsFunc(m.st.Recordings, func(x Rec) bool { return x.File == f })436 if errors.Is(err, os.ErrNotExist) && !taken {437 return f438 }439 }440}441442// clean makes s fit for a file name: no separators or control characters,443// not hidden, and not too long.444func clean(s string) string {445 s = strings.Map(func(r rune) rune {446 switch {447 case r == '/' || r == '\\':448 return '-'449 case r < 0x20 || r == 0x7f:450 return -1451 }452 return r453 }, s)454 s = strings.TrimLeft(strings.TrimSpace(s), ".")455 for len(s) > 80 {456 _, n := utf8.DecodeLastRuneInString(s)457 s = s[:len(s)-n]458 }459 return strings.TrimSpace(s)460}