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/slog"
10 "os"
11 "path/filepath"
12 "slices"
13 "strconv"
14 "strings"
15 "sync"
16 "time"
17
18 "vidarchive/internal/config"
19 "vidarchive/internal/models"
20 "vidarchive/internal/repository"
21 "vidarchive/internal/util"
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, limit, offset int) ([]*models.Download, error) {
99 downloads, err := s.repo.GetAll(status, sortBy, limit, offset)
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
145// cancelDownload stops the download if it is in flight, reporting whether there
146// was one to stop.
147func (s *DownloadService) cancelDownload(id int64) bool {
148 s.activeMu.Lock()
149 cancel, ok := s.active[id]
150 delete(s.active, id)
151 s.activeMu.Unlock()
152
153 if ok {
154 cancel()
155 }
156 return ok
157}
158
159// CancelAll stops every download currently in flight. Used on shutdown and when
160// clearing the queue, so no yt-dlp child outlives the rows that described it.
161func (s *DownloadService) CancelAll() {
162 s.activeMu.Lock()
163 cancels := make([]context.CancelFunc, 0, len(s.active))
164 for id, cancel := range s.active {
165 cancels = append(cancels, cancel)
166 delete(s.active, id)
167 }
168 s.activeMu.Unlock()
169
170 for _, cancel := range cancels {
171 cancel()
172 }
173}
174
175// Retry puts a failed or cancelled download back in the queue. The worker pool's
176// queue checker picks it up on its next tick, so nothing is submitted here.
177func (s *DownloadService) Retry(id int64) error {
178 ok, err := s.repo.Requeue(id)
179 if err != nil {
180 return err
181 }
182 if !ok {
183 return fmt.Errorf("download %d is not in a retryable state", id)
184 }
185 s.cache.Delete(id)
186 return nil
187}
188
189// Cancel stops a download without deleting its row, so it stays visible and
190// retryable. The queued case is marked here; a running one is stopped by
191// cancelling its context, and ExecuteDownload records the 'cancelled' status.
192//
193// The order matters: marking first means a worker racing to claim the row finds
194// it no longer 'queued' and skips it, instead of starting work we just cancelled.
195func (s *DownloadService) Cancel(id int64) error {
196 wasQueued, err := s.repo.CancelQueued(id)
197 if err != nil {
198 return err
199 }
200 // Neither queued nor running: the row is already finished, or it is a stale
201 // 'downloading' row no worker owns. Say so instead of reporting success.
202 if !s.cancelDownload(id) && !wasQueued {
203 return fmt.Errorf("download %d is not running", id)
204 }
205 return nil
206}
207
208// DeleteByStatus clears one status' worth of rows. Only finished statuses are
209// offered in the UI, so nothing it removes needs cancelling first.
210func (s *DownloadService) DeleteByStatus(status string) error {
211 return s.repo.DeleteByStatus(status)
212}
213
214func (s *DownloadService) DeleteAll() error {
215 // Clearing the queue must also stop what is running; otherwise yt-dlp keeps
216 // going and imports into the library after its row is gone.
217 s.CancelAll()
218 return s.repo.DeleteAll()
219}
220
221func (s *DownloadService) Ping(ctx context.Context) error {
222 return s.repo.Ping(ctx)
223}
224
225// ResetStalledDownloads re-queues downloads left mid-flight by a previous run and
226// discards their temp directories. Without the cleanup the re-run imports into a
227// fresh uniqueDir and the library ends up with a duplicate of the same item.
228func (s *DownloadService) ResetStalledDownloads() error {
229 ids, err := s.repo.IDsByStatus("downloading")
230 if err != nil {
231 return err
232 }
233
234 for _, id := range ids {
235 for _, dir := range s.tempDirsFor(id) {
236 if err := os.RemoveAll(dir); err != nil {
237 slog.Warn("failed to remove stale temp dir", "dir", dir, "err", err)
238 }
239 }
240 }
241
242 return s.repo.UpdateStatusWhere("downloading", "queued")
243}
244
245// tempDirFor returns the scratch directory a download writes into.
246func (s *DownloadService) tempDirFor(id int64) string {
247 return filepath.Join(s.cfg.TempDir, strconv.FormatInt(id, 10))
248}
249
250// tempNewDirFor returns the second-pass scratch directory used by metadata mode.
251func (s *DownloadService) tempNewDirFor(id int64) string {
252 return s.tempDirFor(id) + "-new"
253}
254
255// tempDirsFor returns every scratch directory a download owns. ResetStalledDownloads
256// clears these, so the two builders above must stay the only places that name them.
257func (s *DownloadService) tempDirsFor(id int64) []string {
258 return []string{s.tempDirFor(id), s.tempNewDirFor(id)}
259}
260
261// ExecuteDownload runs the download for d. The bool reports whether this call
262// actually processed it: false means another worker already claimed it (Submit
263// and the queue checker can both enqueue the same row within the 2s poll window),
264// so the caller should not log it as completed. A cancelled run returns
265// ErrCancelled.
266//
267// parent belongs to the worker pool: deriving from it means a shutdown cancels
268// the download even if it lands before this call registers its own cancel func.
269func (s *DownloadService) ExecuteDownload(parent context.Context, d *models.Download) (bool, error) {
270 claimed, err := s.repo.MarkStarted(d.ID)
271 if err != nil {
272 return false, err
273 }
274 if !claimed {
275 return false, nil
276 }
277
278 // Registered before any work starts so Delete/CancelAll can interrupt every
279 // phase, not just the window where yt-dlp happens to be running.
280 ctx, cancel := context.WithCancel(parent)
281 defer cancel()
282 defer s.registerActive(d.ID, cancel)()
283
284 s.cache.Start(d.ID)
285 defer s.cache.Delete(d.ID)
286
287 var preset *models.Preset
288 if d.PresetID.Valid {
289 preset, err = s.presetSvc.GetByID(d.PresetID.Int64)
290 if err != nil {
291 slog.Warn("preset lookup failed, falling back to default", "download_id", d.ID, "preset_id", d.PresetID.Int64, "err", err)
292 preset = nil
293 }
294 }
295 if preset == nil {
296 var derr error
297 if preset, derr = s.presetSvc.GetDefault(); derr != nil {
298 slog.Warn("no default preset available, using built-in defaults", "download_id", d.ID, "err", derr)
299 preset = &models.Preset{}
300 }
301 }
302
303 var sub *models.Subscription
304 if d.SubscriptionID.Valid && s.subscriptionSvc != nil {
305 var serr error
306 if sub, serr = s.subscriptionSvc.GetByID(d.SubscriptionID.Int64); serr != nil {
307 slog.Error("subscription lookup failed", "download_id", d.ID, "subscription_id", d.SubscriptionID.Int64, "err", serr)
308 }
309 }
310
311 // Reject custom flags that clash with options VidArchive sets itself, before
312 // spending any work — the download fails with a message naming the offender.
313 isSubscription := d.SubscriptionID.Valid
314 for _, flags := range []string{d.CustomFlags, preset.CustomFlags} {
315 if err := checkReservedFlags(flags, isSubscription); err != nil {
316 s.finalizeError(d, err)
317 return false, err
318 }
319 }
320
321 // From here the run is really under way, so a subscription shows "downloading"
322 // instead of the "queued" the scheduler recorded.
323 s.recordSubscriptionStatus(d, "downloading")
324
325 tempDownloadDir := s.tempDirFor(d.ID)
326 if err := os.MkdirAll(tempDownloadDir, 0o755); err != nil {
327 // MarkStarted already moved the row to "downloading"; returning without
328 // finalizing would strand it there until the next restart.
329 err = fmt.Errorf("create temp download dir: %w", err)
330 s.finalizeError(d, err)
331 return false, err
332 }
333 // Own the temp dir's lifetime here, where it's created, so it's removed on
334 // every exit path — including a failed yt-dlp run or an early return that
335 // crashes mid-import. The import helpers below no longer clean it up.
336 defer os.RemoveAll(tempDownloadDir)
337
338 args := s.presetSvc.BuildArgs(preset, d.FormatOverride, d.CustomFlags)
339
340 // Record the meaningful flags (format/audio/subs/custom) that shaped this
341 // download, before the internal plumbing (cookies, -P/-o, URL) is appended,
342 // so each imported item can show how it was fetched.
343 ytdlpFlags := strings.Join(args, " ")
344
345 var cookieCleanup func()
346 args, cookieCleanup = s.appendCookies(args)
347 defer cookieCleanup()
348
349 if sub != nil {
350 // Always write info.json so the import step can read the stable identity
351 // (yt-dlp's video id) used to match/replace existing items.
352 if !slices.Contains(args, "--write-info-json") {
353 args = append(args, "--write-info-json")
354 }
355 switch sub.RefreshMode {
356 case "skip":
357 // Let yt-dlp skip entries already recorded — no re-download.
358 archive := s.subscriptionSvc.ArchivePath(sub.ID)
359 if err := os.MkdirAll(filepath.Dir(archive), 0o755); err == nil {
360 args = append(args, "--download-archive", archive)
361 }
362 case "metadata":
363 // Refresh metadata only; don't fetch media.
364 args = append(args, "--skip-download")
365 }
366 }
367
368 args = append(args, "-P", tempDownloadDir)
369 args = append(args, "-o", "item-%(autonumber)05d/%(title)s.%(ext)s")
370 args = append(args, d.URL)
371
372 runErr := s.runYTDLP(ctx, d, args)
373
374 // Cancellation wins over both the run error and the import: a cancel that
375 // lands just after yt-dlp exited 0 leaves runErr nil, and the item must not
376 // reach the library after the user removed it.
377 if ctx.Err() != nil {
378 return false, s.finalizeCancelled(parent, d)
379 }
380
381 mode := ""
382 if sub != nil {
383 mode = sub.RefreshMode
384 }
385
386 // Post-process before marking completed, so the download stays "downloading"
387 // until everything is really done — including metadata mode's second pass,
388 // which downloads any genuinely new entries as full items.
389 //
390 // This runs even when runErr is set: yt-dlp exits non-zero if a single
391 // playlist entry fails, while the other entries downloaded fine. Skipping the
392 // import would throw those away with the temp dir. The outcome is decided
393 // below, once the imported count is known.
394 var postErr error
395 partial := ""
396 if mode == "metadata" {
397 // The main pass ran with --skip-download, so the temp dir holds only
398 // info.json files: refresh existing items in place and fetch new ones.
399 postErr = s.refreshAndAddNew(ctx, d, preset, tempDownloadDir, ytdlpFlags)
400 if postErr == nil && runErr != nil {
401 partial = fmt.Sprintf("VidArchive: yt-dlp exited with an error (%v); the metadata refresh finished anyway. Check the log above for failed entries.", runErr)
402 }
403 } else {
404 imported, err := s.importDownloadedItems(ctx, d, tempDownloadDir, mode, ytdlpFlags)
405 switch {
406 case err != nil:
407 postErr = err
408 case imported == 0 && runErr != nil:
409 postErr = runErr
410 // A plain (non-subscription) download that yields nothing is a failure, not
411 // a silent "completed". Subscription modes legitimately import zero (skip
412 // mode, or a metadata refresh with no new entries), so only enforce this for
413 // plain runs.
414 case imported == 0 && sub == nil:
415 postErr = fmt.Errorf("yt-dlp finished but no media files were downloaded")
416 case runErr != nil:
417 partial = fmt.Sprintf("VidArchive: yt-dlp exited with an error (%v); %d item(s) were imported anyway. Check the log above for failed entries.", runErr, imported)
418 }
419 }
420
421 if postErr == nil && sub != nil && sub.PruneRemoved {
422 s.pruneSubscription(ctx, d, sub)
423 }
424
425 // Same cancellation check as above: a stop during post-processing must not
426 // be recorded as "completed".
427 if ctx.Err() != nil {
428 return false, s.finalizeCancelled(parent, d)
429 }
430 if postErr != nil {
431 s.finalizeError(d, postErr)
432 return false, postErr
433 }
434
435 // Record the partial failure in the download's own log. The run counts as
436 // completed, so nothing else would tell the user some entries failed.
437 if partial != "" {
438 s.cache.AppendLog(d.ID, partial)
439 s.flushLogs(d.ID)
440 }
441
442 if err := s.repo.MarkCompleted(d.ID, "completed"); err != nil {
443 return false, err
444 }
445 s.recordSubscriptionStatus(d, "completed")
446
447 return true, nil
448}
449
450// runYTDLP executes yt-dlp with args, streaming combined output into the live
451// progress cache and periodically flushing it to the download's persisted log.
452// Cancelling ctx kills the whole process group and makes this return.
453func (s *DownloadService) runYTDLP(ctx context.Context, d *models.Download, args []string) error {
454 // --newline forces yt-dlp to emit each progress update on its own line. Without
455 // it, progress is rewritten in place with carriage returns, so a long download
456 // becomes one ever-growing line that overflows the reader's buffer and stalls
457 // the pipe — hanging the download. See the hardened scanner below.
458 //
459 // --no-write-playlist-metafiles suppresses the playlist-level info.json that
460 // --write-info-json would also produce. It lands in the first item dir and
461 // would be imported as if it were an item.
462 //
463 // --socket-timeout bounds a stalled connection. Without it a dead socket pins
464 // a worker forever. There is no inactivity killer beyond this.
465 fullArgs := append([]string{"--newline", "--no-write-playlist-metafiles", "--socket-timeout", "30"}, args...)
466 cmd := util.KillableCommand(ctx, s.cfg.YTDLPPath, fullArgs...)
467
468 stdout, err := cmd.StdoutPipe()
469 if err != nil {
470 return err
471 }
472 cmd.Stderr = cmd.Stdout
473
474 if err := cmd.Start(); err != nil {
475 return err
476 }
477
478 ticker := time.NewTicker(10 * time.Second)
479 defer ticker.Stop()
480 done := make(chan struct{})
481 go func() {
482 for {
483 select {
484 case <-ticker.C:
485 s.flushLogs(d.ID)
486 case <-done:
487 return
488 }
489 }
490 }()
491
492 scanner := bufio.NewScanner(stdout)
493 // Allow long lines (a single yt-dlp message can exceed the 64 KiB default)
494 // rather than letting the scanner abort and leave the pipe unread.
495 scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
496 for scanner.Scan() {
497 s.cache.AppendLog(d.ID, scanner.Text())
498 }
499 if err := scanner.Err(); err != nil {
500 slog.Error("error reading yt-dlp output", "download_id", d.ID, "err", err)
501 }
502 close(done)
503
504 s.flushLogs(d.ID)
505
506 return cmd.Wait()
507}
508
509// appendCookies writes the saved cookies (if any) to a temp file and appends a
510// --cookies flag. The returned cleanup saves the cookies yt-dlp left behind and
511// removes the temp file. It is always safe to call, even when no cookies were
512// configured.
513func (s *DownloadService) appendCookies(args []string) ([]string, func()) {
514 cookies, err := s.settingsSvc.GetCookies()
515 if err != nil || strings.TrimSpace(cookies) == "" {
516 return args, func() {}
517 }
518 path, err := s.writeCookiesFile(cookies)
519 if err != nil {
520 return args, func() {}
521 }
522 return append(args, "--cookies", path), func() {
523 s.saveRefreshedCookies(path, cookies)
524 os.Remove(path)
525 }
526}
527
528// saveRefreshedCookies stores back what yt-dlp wrote to the cookie file.
529//
530// yt-dlp rewrites the jar on exit. YouTube rotates session cookies on use, so
531// the snapshot we sent is stale once the run ends. Keeping the old snapshot and
532// replaying it later gets the session invalidated, and the user has to export
533// cookies again. sent is what we wrote, so an untouched file saves nothing.
534func (s *DownloadService) saveRefreshedCookies(path, sent string) {
535 data, err := os.ReadFile(path)
536 // A missing file means yt-dlp never got that far. Empty content would wipe
537 // working cookies, so treat it as nothing to do.
538 if err != nil || strings.TrimSpace(string(data)) == "" || string(data) == sent {
539 return
540 }
541 if err := s.settingsSvc.SetCookies(string(data)); err != nil {
542 slog.Error("failed to save refreshed cookies", "err", err)
543 }
544}
545
546// writeCookiesFile writes cookies to a temp file in the app's own temp dir (the
547// same volume the rest of the run uses). On any failure the partial file is
548// removed — a truncated cookies file must not be handed to yt-dlp.
549func (s *DownloadService) writeCookiesFile(cookies string) (string, error) {
550 if err := os.MkdirAll(s.cfg.TempDir, 0o755); err != nil {
551 return "", err
552 }
553 tmpFile, err := os.CreateTemp(s.cfg.TempDir, "cookies-*.txt")
554 if err != nil {
555 return "", err
556 }
557 if _, err := tmpFile.WriteString(cookies); err != nil {
558 tmpFile.Close()
559 os.Remove(tmpFile.Name())
560 return "", err
561 }
562 if err := tmpFile.Close(); err != nil {
563 os.Remove(tmpFile.Name())
564 return "", err
565 }
566 return tmpFile.Name(), nil
567}
568
569func (s *DownloadService) finalizeError(d *models.Download, err error) {
570 s.flushLogs(d.ID)
571 if markErr := s.repo.MarkError(d.ID, err.Error()); markErr != nil {
572 slog.Error("failed to record download error", "download_id", d.ID, "err", markErr)
573 }
574 s.recordSubscriptionStatus(d, "error")
575}
576
577// recordSubscriptionStatus mirrors a subscription download's state onto the
578// subscription row. Without it the row keeps the status it had when the
579// scheduler queued it, so the subscriptions page reports "queued" long after the
580// run finished — or failed.
581func (s *DownloadService) recordSubscriptionStatus(d *models.Download, status string) {
582 if !d.SubscriptionID.Valid || s.subscriptionSvc == nil {
583 return
584 }
585 if err := s.subscriptionSvc.SetLastStatus(d.SubscriptionID.Int64, status); err != nil {
586 slog.Error("failed to record subscription status", "download_id", d.ID,
587 "subscription_id", d.SubscriptionID.Int64, "status", status, "err", err)
588 }
589}
590
591// ErrCancelled reports that a download was deliberately stopped (deleted, queue
592// cleared, or shutdown) rather than having failed. Callers distinguish it so a
593// cancellation isn't logged as an error.
594var ErrCancelled = errors.New("download cancelled")
595
596// finalizeCancelled records a stopped download.
597//
598// A shutdown (parent already cancelled) deliberately leaves the row
599// "downloading": ResetStalledDownloads re-queues it on the next start, so
600// stopping the server resumes the download instead of losing it. Only a
601// user-initiated cancel is terminal. The row may already be deleted in that
602// case — cancellation usually arrives via Delete — so a missing row is fine.
603func (s *DownloadService) finalizeCancelled(parent context.Context, d *models.Download) error {
604 s.flushLogs(d.ID)
605
606 // A shutdown leaves the subscription status alone too: the run resumes on the
607 // next start, so it is still in progress rather than cancelled.
608 if parent.Err() != nil {
609 return ErrCancelled
610 }
611
612 if err := s.repo.MarkCompleted(d.ID, "cancelled"); err != nil {
613 slog.Error("failed to record cancellation", "download_id", d.ID, "err", err)
614 }
615 s.recordSubscriptionStatus(d, "cancelled")
616 return ErrCancelled
617}
618
619// flushLogs persists whatever output has accumulated for a download.
620func (s *DownloadService) flushLogs(id int64) {
621 logs := s.cache.FlushLogs(id)
622 if logs == "" {
623 return
624 }
625 if err := s.repo.AppendLogs(id, logs); err != nil {
626 slog.Error("failed to persist logs", "download_id", id, "err", err)
627 }
628}
629
630func sqlNullInt64(v int64) sql.NullInt64 {
631 return sql.NullInt64{Int64: v, Valid: true}
632}
633