vidarchive

vidarchive/internal/service/download.go
download.go
⎇
Raw
1package service
2
3import (
4 "bufio"
5 "context"
6 "database/sql"
7 "errors"
8 "fmt"
9 "log"
10 "os"
11 "os/exec"
12 "path/filepath"
13 "strconv"
14 "strings"
15 "sync"
16 "syscall"
17 "time"
18
19 "vidarchive/internal/config"
20 "vidarchive/internal/models"
21 "vidarchive/internal/repository"
22)
23
24type DownloadService struct {
25 repo *repository.DownloadRepository
26 librarySvc *LibraryService
27 presetSvc *PresetService
28 settingsSvc *SettingsService
29 subscriptionSvc *SubscriptionService
30 cfg *config.Config
31 cache *ProgressCache
32 activeMu sync.Mutex
33 active map[int64]context.CancelFunc
34}
35
36func NewDownloadService(repo *repository.DownloadRepository, librarySvc *LibraryService, presetSvc *PresetService, settingsSvc *SettingsService, subscriptionSvc *SubscriptionService, cfg *config.Config) *DownloadService {
37 return &DownloadService{
38 repo: repo,
39 librarySvc: librarySvc,
40 presetSvc: presetSvc,
41 settingsSvc: settingsSvc,
42 subscriptionSvc: subscriptionSvc,
43 cfg: cfg,
44 cache: NewProgressCache(),
45 active: make(map[int64]context.CancelFunc),
46 }
47}
48
49func (s *DownloadService) Create(url string, presetID *int64, formatOverride, customFlags, outputDir string) (*models.Download, error) {
50 d := &models.Download{
51 URL: url,
52 Status: "queued",
53 FormatOverride: formatOverride,
54 CustomFlags: customFlags,
55 OutputDir: sql.NullString{String: outputDir, Valid: outputDir != ""},
56 }
57
58 if presetID != nil {
59 d.PresetID = sqlNullInt64(*presetID)
60 }
61
62 if err := s.repo.Create(d); err != nil {
63 return nil, err
64 }
65 return d, nil
66}
67
68// CreateForSubscription queues a download for a subscription run, copying its
69// download options and tagging it with the subscription id so ExecuteDownload
70// applies the right refresh mode and pruning.
71func (s *DownloadService) CreateForSubscription(sub *models.Subscription) (*models.Download, error) {
72 d := &models.Download{
73 URL: sub.URL,
74 Status: "queued",
75 FormatOverride: sub.FormatOverride,
76 CustomFlags: sub.CustomFlags,
77 OutputDir: sql.NullString{String: sub.OutputDir, Valid: sub.OutputDir != ""},
78 PresetID: sub.PresetID,
79 SubscriptionID: sqlNullInt64(sub.ID),
80 }
81 if err := s.repo.Create(d); err != nil {
82 return nil, err
83 }
84 return d, nil
85}
86
87func (s *DownloadService) GetByID(id int64) (*models.Download, error) {
88 d, err := s.repo.GetByID(id)
89 if err != nil {
90 return nil, err
91 }
92 if logs := s.cache.Snapshot(id); logs != "" {
93 d.Logs = sql.NullString{String: logs, Valid: true}
94 }
95 return d, nil
96}
97
98func (s *DownloadService) GetAll(status, sortBy string) ([]*models.Download, error) {
99 downloads, err := s.repo.GetAll(status, sortBy)
100 if err != nil {
101 return nil, err
102 }
103
104 for _, d := range downloads {
105 if logs := s.cache.Snapshot(d.ID); logs != "" {
106 d.Logs = sql.NullString{String: logs, Valid: true}
107 }
108 }
109
110 return downloads, nil
111}
112
113func (s *DownloadService) GetQueued(limit int) ([]*models.Download, error) {
114 return s.repo.GetQueued(limit)
115}
116
117// HasActiveForSubscription reports whether the subscription already has a queued
118// or in-progress download, so the scheduler can skip stacking another run.
119func (s *DownloadService) HasActiveForSubscription(subID int64) (bool, error) {
120 return s.repo.HasActiveForSubscription(subID)
121}
122
123func (s *DownloadService) Delete(id int64) error {
124 s.cancelDownload(id)
125 s.cache.Delete(id)
126 return s.repo.Delete(id)
127}
128
129// registerActive records the cancel func for a claimed download and returns a
130// release func. Registration happens at claim time rather than after the process
131// spawns, so a delete arriving during setup, between yt-dlp and the import, or
132// mid-import still stops the work instead of silently letting it finish.
133func (s *DownloadService) registerActive(id int64, cancel context.CancelFunc) func() {
134 s.activeMu.Lock()
135 s.active[id] = cancel
136 s.activeMu.Unlock()
137
138 return func() {
139 s.activeMu.Lock()
140 delete(s.active, id)
141 s.activeMu.Unlock()
142 }
143}
144
145func (s *DownloadService) cancelDownload(id int64) {
146 s.activeMu.Lock()
147 cancel, ok := s.active[id]
148 delete(s.active, id)
149 s.activeMu.Unlock()
150
151 if ok {
152 cancel()
153 }
154}
155
156// CancelAll stops every download currently in flight. Used on shutdown and when
157// clearing the queue, so no yt-dlp child outlives the rows that described it.
158func (s *DownloadService) CancelAll() {
159 s.activeMu.Lock()
160 cancels := make([]context.CancelFunc, 0, len(s.active))
161 for id, cancel := range s.active {
162 cancels = append(cancels, cancel)
163 delete(s.active, id)
164 }
165 s.activeMu.Unlock()
166
167 for _, cancel := range cancels {
168 cancel()
169 }
170}
171
172func (s *DownloadService) DeleteAll() error {
173 // Clearing the queue must also stop what is running; otherwise yt-dlp keeps
174 // going and imports into the library after its row is gone.
175 s.CancelAll()
176 return s.repo.DeleteAll()
177}
178
179func (s *DownloadService) Ping(ctx context.Context) error {
180 return s.repo.Ping(ctx)
181}
182
183// ResetStalledDownloads re-queues downloads left mid-flight by a previous run and
184// discards their temp directories. Without the cleanup the re-run imports into a
185// fresh uniqueDir and the library ends up with a duplicate of the same item.
186func (s *DownloadService) ResetStalledDownloads() error {
187 ids, err := s.repo.IDsByStatus("downloading")
188 if err != nil {
189 return err
190 }
191
192 for _, id := range ids {
193 for _, dir := range s.tempDirsFor(id) {
194 if err := os.RemoveAll(dir); err != nil {
195 log.Printf("warning: failed to remove stale temp dir %s: %v", dir, err)
196 }
197 }
198 }
199
200 return s.repo.UpdateStatusWhere("downloading", "queued")
201}
202
203// tempDirFor returns the scratch directory a download writes into.
204func (s *DownloadService) tempDirFor(id int64) string {
205 return filepath.Join(s.cfg.TempDir, strconv.FormatInt(id, 10))
206}
207
208// tempNewDirFor returns the second-pass scratch directory used by metadata mode.
209func (s *DownloadService) tempNewDirFor(id int64) string {
210 return s.tempDirFor(id) + "-new"
211}
212
213// tempDirsFor returns every scratch directory a download owns. ResetStalledDownloads
214// clears these, so the two builders above must stay the only places that name them.
215func (s *DownloadService) tempDirsFor(id int64) []string {
216 return []string{s.tempDirFor(id), s.tempNewDirFor(id)}
217}
218
219// ExecuteDownload runs the download for d. The bool reports whether this call
220// actually processed it: false means another worker already claimed it (Submit
221// and the queue checker can both enqueue the same row within the 2s poll window),
222// so the caller should not log it as completed. A cancelled run returns
223// ErrCancelled.
224//
225// parent belongs to the worker pool: deriving from it means a shutdown cancels
226// the download even if it lands before this call registers its own cancel func.
227func (s *DownloadService) ExecuteDownload(parent context.Context, d *models.Download) (bool, error) {
228 claimed, err := s.repo.MarkStarted(d.ID)
229 if err != nil {
230 return false, err
231 }
232 if !claimed {
233 return false, nil
234 }
235
236 // Registered before any work starts so Delete/CancelAll can interrupt every
237 // phase, not just the window where yt-dlp happens to be running.
238 ctx, cancel := context.WithCancel(parent)
239 defer cancel()
240 defer s.registerActive(d.ID, cancel)()
241
242 s.cache.Set(d.ID, &LiveDownload{})
243 defer s.cache.Delete(d.ID)
244
245 var preset *models.Preset
246 if d.PresetID.Valid {
247 preset, err = s.presetSvc.GetByID(d.PresetID.Int64)
248 if err != nil {
249 log.Printf("download %d: preset %d lookup failed (%v); falling back to default", d.ID, d.PresetID.Int64, err)
250 preset = nil
251 }
252 }
253 if preset == nil {
254 var derr error
255 if preset, derr = s.presetSvc.GetDefault(); derr != nil {
256 log.Printf("download %d: no default preset available (%v); using built-in defaults", d.ID, derr)
257 preset = &models.Preset{}
258 }
259 }
260
261 var sub *models.Subscription
262 if d.SubscriptionID.Valid && s.subscriptionSvc != nil {
263 var serr error
264 if sub, serr = s.subscriptionSvc.GetByID(d.SubscriptionID.Int64); serr != nil {
265 log.Printf("download %d: subscription %d lookup failed: %v", d.ID, d.SubscriptionID.Int64, serr)
266 }
267 }
268
269 // Reject custom flags that clash with options VidArchive sets itself, before
270 // spending any work — the download fails with a message naming the offender.
271 isSubscription := d.SubscriptionID.Valid
272 for _, flags := range []string{d.CustomFlags, preset.CustomFlags} {
273 if err := checkReservedFlags(flags, isSubscription); err != nil {
274 s.finalizeError(d, err)
275 return false, err
276 }
277 }
278
279 // From here the run is really under way, so a subscription shows "downloading"
280 // instead of the "queued" the scheduler recorded.
281 s.recordSubscriptionStatus(d, "downloading")
282
283 tempDownloadDir := s.tempDirFor(d.ID)
284 if err := os.MkdirAll(tempDownloadDir, 0755); err != nil {
285 return false, fmt.Errorf("create temp download dir: %w", err)
286 }
287 // Own the temp dir's lifetime here, where it's created, so it's removed on
288 // every exit path — including a failed yt-dlp run or an early return that
289 // crashes mid-import. The import helpers below no longer clean it up.
290 defer os.RemoveAll(tempDownloadDir)
291
292 args := s.presetSvc.BuildArgs(preset, d.FormatOverride, d.CustomFlags)
293
294 // Record the meaningful flags (format/audio/subs/custom) that shaped this
295 // download, before the internal plumbing (cookies, -P/-o, URL) is appended,
296 // so each imported item can show how it was fetched.
297 ytdlpFlags := strings.Join(args, " ")
298
299 var cookieCleanup func()
300 args, cookieCleanup = s.appendCookies(args)
301 defer cookieCleanup()
302
303 if sub != nil {
304 // Always write info.json so the import step can read the stable identity
305 // (yt-dlp's video id) used to match/replace existing items.
306 args = append(args, "--write-info-json")
307 switch sub.RefreshMode {
308 case "skip":
309 // Let yt-dlp skip entries already recorded — no re-download.
310 archive := s.subscriptionSvc.ArchivePath(sub.ID)
311 if err := os.MkdirAll(filepath.Dir(archive), 0755); err == nil {
312 args = append(args, "--download-archive", archive)
313 }
314 case "metadata":
315 // Refresh metadata only; don't fetch media.
316 args = append(args, "--skip-download")
317 }
318 }
319
320 args = append(args, "-P", tempDownloadDir)
321 args = append(args, "-o", "item-%(autonumber)05d/%(title)s.%(ext)s")
322 args = append(args, d.URL)
323
324 runErr := s.runYTDLP(ctx, d, args)
325
326 // Cancellation wins over both the run error and the import: a cancel that
327 // lands just after yt-dlp exited 0 leaves runErr nil, and the item must not
328 // reach the library after the user removed it.
329 if ctx.Err() != nil {
330 return false, s.finalizeCancelled(parent, d)
331 }
332 if runErr != nil {
333 s.finalizeError(d, runErr)
334 return false, runErr
335 }
336
337 mode := ""
338 if sub != nil {
339 mode = sub.RefreshMode
340 }
341
342 // Post-process before marking completed, so the download stays "downloading"
343 // until everything is really done — including metadata mode's second pass,
344 // which downloads any genuinely new entries as full items.
345 var postErr error
346 if mode == "metadata" {
347 // The main pass ran with --skip-download, so the temp dir holds only
348 // info.json files: refresh existing items in place and fetch new ones.
349 postErr = s.refreshAndAddNew(ctx, d, preset, tempDownloadDir, ytdlpFlags)
350 } else {
351 imported, err := s.importDownloadedItems(ctx, d, tempDownloadDir, mode, ytdlpFlags)
352 switch {
353 case err != nil:
354 postErr = err
355 // A plain (non-subscription) download that yields nothing is a failure, not
356 // a silent "completed". Subscription modes legitimately import zero (skip
357 // mode, or a metadata refresh with no new entries), so only enforce this for
358 // plain runs.
359 case sub == nil && imported == 0:
360 postErr = fmt.Errorf("yt-dlp finished but no media files were downloaded")
361 }
362 }
363
364 if postErr == nil && sub != nil && sub.PruneRemoved {
365 s.pruneSubscription(ctx, d, sub)
366 }
367
368 // Same cancellation check as above: a stop during post-processing must not
369 // be recorded as "completed".
370 if ctx.Err() != nil {
371 return false, s.finalizeCancelled(parent, d)
372 }
373 if postErr != nil {
374 s.finalizeError(d, postErr)
375 return false, postErr
376 }
377
378 if err := s.repo.MarkCompleted(d.ID, "completed"); err != nil {
379 return false, err
380 }
381 s.recordSubscriptionStatus(d, "completed")
382
383 return true, nil
384}
385
386// runYTDLP executes yt-dlp with args, streaming combined output into the live
387// progress cache and periodically flushing it to the download's persisted log.
388// Cancelling ctx kills the whole process group and makes this return.
389func (s *DownloadService) runYTDLP(ctx context.Context, d *models.Download, args []string) error {
390 // --newline forces yt-dlp to emit each progress update on its own line. Without
391 // it, progress is rewritten in place with carriage returns, so a long download
392 // becomes one ever-growing line that overflows the reader's buffer and stalls
393 // the pipe — hanging the download. See the hardened scanner below.
394 fullArgs := append([]string{"--newline"}, args...)
395 cmd := exec.CommandContext(ctx, s.cfg.YTDLPPath, fullArgs...)
396 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
397 // yt-dlp spawns helpers (ffmpeg, external downloaders). Kill the whole group
398 // rather than just the parent, which would leave those orphaned.
399 cmd.Cancel = func() error {
400 return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
401 }
402
403 stdout, err := cmd.StdoutPipe()
404 if err != nil {
405 return err
406 }
407 cmd.Stderr = cmd.Stdout
408
409 if err := cmd.Start(); err != nil {
410 return err
411 }
412
413 ticker := time.NewTicker(10 * time.Second)
414 defer ticker.Stop()
415 done := make(chan struct{})
416 go func() {
417 for {
418 select {
419 case <-ticker.C:
420 s.flushLogs(d.ID)
421 case <-done:
422 return
423 }
424 }
425 }()
426
427 scanner := bufio.NewScanner(stdout)
428 // Allow long lines (a single yt-dlp message can exceed the 64 KiB default)
429 // rather than letting the scanner abort and leave the pipe unread.
430 scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
431 for scanner.Scan() {
432 s.cache.AppendLog(d.ID, scanner.Text())
433 }
434 if err := scanner.Err(); err != nil {
435 log.Printf("download %d: error reading yt-dlp output: %v", d.ID, err)
436 }
437 close(done)
438
439 s.flushLogs(d.ID)
440
441 return cmd.Wait()
442}
443
444// appendCookies writes the saved cookies (if any) to a temp file and appends a
445// --cookies flag. The returned cleanup removes the temp file and is always safe
446// to call, even when no cookies were configured.
447func (s *DownloadService) appendCookies(args []string) ([]string, func()) {
448 cookies, err := s.settingsSvc.GetCookies()
449 if err != nil || strings.TrimSpace(cookies) == "" {
450 return args, func() {}
451 }
452 path, err := s.writeCookiesFile(cookies)
453 if err != nil {
454 return args, func() {}
455 }
456 return append(args, "--cookies", path), func() { os.Remove(path) }
457}
458
459// writeCookiesFile writes cookies to a temp file in the app's own temp dir (the
460// same volume the rest of the run uses). On any failure the partial file is
461// removed — a truncated cookies file must not be handed to yt-dlp.
462func (s *DownloadService) writeCookiesFile(cookies string) (string, error) {
463 if err := os.MkdirAll(s.cfg.TempDir, 0755); err != nil {
464 return "", err
465 }
466 tmpFile, err := os.CreateTemp(s.cfg.TempDir, "cookies-*.txt")
467 if err != nil {
468 return "", err
469 }
470 if _, err := tmpFile.WriteString(cookies); err != nil {
471 tmpFile.Close()
472 os.Remove(tmpFile.Name())
473 return "", err
474 }
475 if err := tmpFile.Close(); err != nil {
476 os.Remove(tmpFile.Name())
477 return "", err
478 }
479 return tmpFile.Name(), nil
480}
481
482func (s *DownloadService) finalizeError(d *models.Download, err error) {
483 s.flushLogs(d.ID)
484 if markErr := s.repo.MarkError(d.ID, err.Error()); markErr != nil {
485 log.Printf("download %d: failed to record error: %v", d.ID, markErr)
486 }
487 s.recordSubscriptionStatus(d, "error")
488}
489
490// recordSubscriptionStatus mirrors a subscription download's state onto the
491// subscription row. Without it the row keeps the status it had when the
492// scheduler queued it, so the subscriptions page reports "queued" long after the
493// run finished — or failed.
494func (s *DownloadService) recordSubscriptionStatus(d *models.Download, status string) {
495 if !d.SubscriptionID.Valid || s.subscriptionSvc == nil {
496 return
497 }
498 if err := s.subscriptionSvc.SetLastStatus(d.SubscriptionID.Int64, status); err != nil {
499 log.Printf("download %d: failed to record subscription %d status %q: %v",
500 d.ID, d.SubscriptionID.Int64, status, err)
501 }
502}
503
504// ErrCancelled reports that a download was deliberately stopped (deleted, queue
505// cleared, or shutdown) rather than having failed. Callers distinguish it so a
506// cancellation isn't logged as an error.
507var ErrCancelled = errors.New("download cancelled")
508
509// finalizeCancelled records a stopped download.
510//
511// A shutdown (parent already cancelled) deliberately leaves the row
512// "downloading": ResetStalledDownloads re-queues it on the next start, so
513// stopping the server resumes the download instead of losing it. Only a
514// user-initiated cancel is terminal. The row may already be deleted in that
515// case — cancellation usually arrives via Delete — so a missing row is fine.
516func (s *DownloadService) finalizeCancelled(parent context.Context, d *models.Download) error {
517 s.flushLogs(d.ID)
518
519 // A shutdown leaves the subscription status alone too: the run resumes on the
520 // next start, so it is still in progress rather than cancelled.
521 if parent.Err() != nil {
522 return ErrCancelled
523 }
524
525 if err := s.repo.MarkCompleted(d.ID, "cancelled"); err != nil {
526 log.Printf("download %d: failed to record cancellation: %v", d.ID, err)
527 }
528 s.recordSubscriptionStatus(d, "cancelled")
529 return ErrCancelled
530}
531
532// flushLogs persists whatever output has accumulated for a download.
533func (s *DownloadService) flushLogs(id int64) {
534 logs := s.cache.FlushLogs(id)
535 if logs == "" {
536 return
537 }
538 if err := s.repo.AppendLogs(id, logs); err != nil {
539 log.Printf("download %d: failed to persist logs: %v", id, err)
540 }
541}
542
543func sqlNullInt64(v int64) sql.NullInt64 {
544 return sql.NullInt64{Int64: v, Valid: true}
545}
546