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 copies the subscription's download options. The
69// subscription id is what makes ExecuteDownload apply refresh mode and pruning.
70func (s *DownloadService) CreateForSubscription(sub *models.Subscription) (*models.Download, error) {
71 d := &models.Download{
72 URL: sub.URL,
73 Status: "queued",
74 FormatOverride: sub.FormatOverride,
75 CustomFlags: sub.CustomFlags,
76 OutputDir: sql.NullString{String: sub.OutputDir, Valid: sub.OutputDir != ""},
77 PresetID: sub.PresetID,
78 SubscriptionID: sqlNullInt64(sub.ID),
79 }
80 if err := s.repo.Create(d); err != nil {
81 return nil, err
82 }
83 return d, nil
84}
85
86func (s *DownloadService) GetByID(id int64) (*models.Download, error) {
87 d, err := s.repo.GetByID(id)
88 if err != nil {
89 return nil, err
90 }
91 if logs := s.cache.Snapshot(id); logs != "" {
92 d.Logs = sql.NullString{String: logs, Valid: true}
93 }
94 return d, nil
95}
96
97func (s *DownloadService) GetAll(status, sortBy string, limit, offset int) ([]*models.Download, error) {
98 downloads, err := s.repo.GetAll(status, sortBy, limit, offset)
99 if err != nil {
100 return nil, err
101 }
102
103 for _, d := range downloads {
104 if logs := s.cache.Snapshot(d.ID); logs != "" {
105 d.Logs = sql.NullString{String: logs, Valid: true}
106 }
107 }
108
109 return downloads, nil
110}
111
112func (s *DownloadService) GetQueued(limit int) ([]*models.Download, error) {
113 return s.repo.GetQueued(limit)
114}
115
116func (s *DownloadService) HasActiveForSubscription(subID int64) (bool, error) {
117 return s.repo.HasActiveForSubscription(subID)
118}
119
120func (s *DownloadService) Delete(id int64) error {
121 s.cancelDownload(id)
122 s.cache.Delete(id)
123 return s.repo.Delete(id)
124}
125
126// registerActive returns a release func. Registration happens at claim time
127// rather than after the process spawns, so a delete arriving during setup or
128// mid-import still stops the work.
129func (s *DownloadService) registerActive(id int64, cancel context.CancelFunc) func() {
130 s.activeMu.Lock()
131 s.active[id] = cancel
132 s.activeMu.Unlock()
133
134 return func() {
135 s.activeMu.Lock()
136 delete(s.active, id)
137 s.activeMu.Unlock()
138 }
139}
140
141// cancelDownload stops the download if it is in flight, reporting whether there
142// was one to stop.
143func (s *DownloadService) cancelDownload(id int64) bool {
144 s.activeMu.Lock()
145 cancel, ok := s.active[id]
146 delete(s.active, id)
147 s.activeMu.Unlock()
148
149 if ok {
150 cancel()
151 }
152 return ok
153}
154
155// CancelAll stops every in-flight download, so no yt-dlp child outlives the row
156// that described it.
157func (s *DownloadService) CancelAll() {
158 s.activeMu.Lock()
159 cancels := make([]context.CancelFunc, 0, len(s.active))
160 for id, cancel := range s.active {
161 cancels = append(cancels, cancel)
162 delete(s.active, id)
163 }
164 s.activeMu.Unlock()
165
166 for _, cancel := range cancels {
167 cancel()
168 }
169}
170
171// Retry re-queues a download. The pool's queue checker picks it up on its next
172// tick, so nothing is submitted here.
173func (s *DownloadService) Retry(id int64) error {
174 ok, err := s.repo.Requeue(id)
175 if err != nil {
176 return err
177 }
178 if !ok {
179 return fmt.Errorf("download %d is not in a retryable state", id)
180 }
181 s.cache.Delete(id)
182 return nil
183}
184
185// Cancel stops a download without deleting its row, so it stays visible and
186// retryable. Marking the row first means a worker racing to claim it finds it no
187// longer 'queued' and skips it.
188func (s *DownloadService) Cancel(id int64) error {
189 wasQueued, err := s.repo.CancelQueued(id)
190 if err != nil {
191 return err
192 }
193 // Neither queued nor running: already finished, or a stale 'downloading' row
194 // no worker owns.
195 if !s.cancelDownload(id) && !wasQueued {
196 return fmt.Errorf("download %d is not running", id)
197 }
198 return nil
199}
200
201// DeleteByStatus needs no cancelling: only finished statuses are offered in the UI.
202func (s *DownloadService) DeleteByStatus(status string) error {
203 return s.repo.DeleteByStatus(status)
204}
205
206func (s *DownloadService) DeleteAll() error {
207 // Clearing the queue must stop what is running; otherwise yt-dlp imports into
208 // the library after its row is gone.
209 s.CancelAll()
210 return s.repo.DeleteAll()
211}
212
213func (s *DownloadService) Ping(ctx context.Context) error {
214 return s.repo.Ping(ctx)
215}
216
217// ResetStalledDownloads re-queues downloads left mid-flight by a previous run.
218// The temp dirs go too: otherwise the re-run imports into a fresh uniqueDir and
219// the library ends up with a duplicate.
220func (s *DownloadService) ResetStalledDownloads() error {
221 ids, err := s.repo.IDsByStatus("downloading")
222 if err != nil {
223 return err
224 }
225
226 for _, id := range ids {
227 for _, dir := range s.tempDirsFor(id) {
228 if err := os.RemoveAll(dir); err != nil {
229 slog.Warn("failed to remove stale temp dir", "dir", dir, "err", err)
230 }
231 }
232 }
233
234 return s.repo.UpdateStatusWhere("downloading", "queued")
235}
236
237func (s *DownloadService) tempDirFor(id int64) string {
238 return filepath.Join(s.cfg.TempDir, strconv.FormatInt(id, 10))
239}
240
241// tempNewDirFor is the second-pass scratch directory, used by metadata mode.
242func (s *DownloadService) tempNewDirFor(id int64) string {
243 return s.tempDirFor(id) + "-new"
244}
245
246// tempDirsFor must list every scratch directory a download owns:
247// ResetStalledDownloads relies on it, so keep the two builders above the only
248// places that name one.
249func (s *DownloadService) tempDirsFor(id int64) []string {
250 return []string{s.tempDirFor(id), s.tempNewDirFor(id)}
251}
252
253// ExecuteDownload runs the download for d. The bool reports whether this call
254// processed it: false means another worker already claimed it (Submit and the
255// queue checker can both enqueue the same row within the 2s poll window). A
256// cancelled run returns ErrCancelled.
257//
258// parent belongs to the worker pool, so a shutdown cancels the download even if
259// it lands before this call registers its own cancel func.
260func (s *DownloadService) ExecuteDownload(parent context.Context, d *models.Download) (bool, error) {
261 claimed, err := s.repo.MarkStarted(d.ID)
262 if err != nil {
263 return false, err
264 }
265 if !claimed {
266 return false, nil
267 }
268
269 ctx, cancel := context.WithCancel(parent)
270 defer cancel()
271 defer s.registerActive(d.ID, cancel)()
272
273 s.cache.Start(d.ID)
274 defer s.cache.Delete(d.ID)
275
276 var preset *models.Preset
277 if d.PresetID.Valid {
278 preset, err = s.presetSvc.GetByID(d.PresetID.Int64)
279 if err != nil {
280 slog.Warn("preset lookup failed, falling back to default", "download_id", d.ID, "preset_id", d.PresetID.Int64, "err", err)
281 preset = nil
282 }
283 }
284 if preset == nil {
285 var derr error
286 if preset, derr = s.presetSvc.GetDefault(); derr != nil {
287 slog.Warn("no default preset available, using built-in defaults", "download_id", d.ID, "err", derr)
288 preset = &models.Preset{}
289 }
290 }
291
292 var sub *models.Subscription
293 if d.SubscriptionID.Valid && s.subscriptionSvc != nil {
294 var serr error
295 if sub, serr = s.subscriptionSvc.GetByID(d.SubscriptionID.Int64); serr != nil {
296 slog.Error("subscription lookup failed", "download_id", d.ID, "subscription_id", d.SubscriptionID.Int64, "err", serr)
297 }
298 }
299
300 // Reject flags that clash with options VidArchive sets itself before spending
301 // any work.
302 isSubscription := d.SubscriptionID.Valid
303 for _, flags := range []string{d.CustomFlags, preset.CustomFlags} {
304 if err := checkReservedFlags(flags, isSubscription); err != nil {
305 s.finalizeError(d, err)
306 return false, err
307 }
308 }
309
310 // The run is really under way now; the scheduler recorded "queued".
311 s.recordSubscriptionStatus(d, "downloading")
312
313 tempDownloadDir := s.tempDirFor(d.ID)
314 if err := os.MkdirAll(tempDownloadDir, 0o755); err != nil {
315 // MarkStarted already moved the row to "downloading"; returning without
316 // finalizing strands it there until the next restart.
317 err = fmt.Errorf("create temp download dir: %w", err)
318 s.finalizeError(d, err)
319 return false, err
320 }
321 // The temp dir's lifetime is owned here, not by the import helpers below, so
322 // every exit path removes it.
323 defer os.RemoveAll(tempDownloadDir)
324
325 args := s.presetSvc.BuildArgs(preset, d.FormatOverride, d.CustomFlags)
326
327 // Snapshot the flags that shaped this download before the plumbing (cookies,
328 // -P/-o, URL) is appended, so each imported item can show how it was fetched.
329 ytdlpFlags := strings.Join(args, " ")
330
331 var cookieCleanup func()
332 args, cookieCleanup = s.appendCookies(args)
333 defer cookieCleanup()
334
335 if sub != nil {
336 // info.json carries yt-dlp's video id, which the import uses to match and
337 // replace existing items.
338 if !slices.Contains(args, "--write-info-json") {
339 args = append(args, "--write-info-json")
340 }
341 switch sub.RefreshMode {
342 case "skip":
343 archive := s.subscriptionSvc.ArchivePath(sub.ID)
344 if err := os.MkdirAll(filepath.Dir(archive), 0o755); err == nil {
345 args = append(args, "--download-archive", archive)
346 }
347 case "metadata":
348 args = append(args, "--skip-download")
349 }
350 }
351
352 args = append(args, "-P", tempDownloadDir)
353 args = append(args, "-o", "item-%(autonumber)05d/%(title)s.%(ext)s")
354 args = append(args, d.URL)
355
356 runErr := s.runYTDLP(ctx, d, args)
357
358 // Cancellation wins over runErr and the import: a cancel landing just after
359 // yt-dlp exits 0 leaves runErr nil, and the item must not reach the library
360 // after the user removed it.
361 if ctx.Err() != nil {
362 return false, s.finalizeCancelled(parent, d)
363 }
364
365 mode := ""
366 if sub != nil {
367 mode = sub.RefreshMode
368 }
369
370 // Post-processing runs before the row is marked completed, and runs even when
371 // runErr is set: yt-dlp exits non-zero if a single playlist entry fails, and
372 // skipping the import would throw the successful entries away with the temp
373 // dir. The outcome is decided below, once the imported count is known.
374 var postErr error
375 partial := ""
376 if mode == "metadata" {
377 // The main pass ran with --skip-download, so the temp dir holds only
378 // info.json files.
379 postErr = s.refreshAndAddNew(ctx, d, preset, tempDownloadDir, ytdlpFlags)
380 if postErr == nil && runErr != nil {
381 partial = fmt.Sprintf("VidArchive: yt-dlp exited with an error (%v); the metadata refresh finished anyway. Check the log above for failed entries.", runErr)
382 }
383 } else {
384 imported, err := s.importDownloadedItems(ctx, d, tempDownloadDir, mode, ytdlpFlags)
385 switch {
386 case err != nil:
387 postErr = err
388 case imported == 0 && runErr != nil:
389 postErr = runErr
390 // Subscription modes legitimately import zero (skip mode, or a metadata
391 // refresh with no new entries); a plain download that yields nothing failed.
392 case imported == 0 && sub == nil:
393 postErr = fmt.Errorf("yt-dlp finished but no media files were downloaded")
394 case runErr != nil:
395 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)
396 }
397 }
398
399 if postErr == nil && sub != nil && sub.PruneRemoved {
400 s.pruneSubscription(ctx, d, sub)
401 }
402
403 // A stop during post-processing must not be recorded as "completed" either.
404 if ctx.Err() != nil {
405 return false, s.finalizeCancelled(parent, d)
406 }
407 if postErr != nil {
408 s.finalizeError(d, postErr)
409 return false, postErr
410 }
411
412 // The run counts as completed, so the log is the only place that can tell the
413 // user some entries failed.
414 if partial != "" {
415 s.cache.AppendLog(d.ID, partial)
416 s.flushLogs(d.ID)
417 }
418
419 if err := s.repo.MarkCompleted(d.ID, "completed"); err != nil {
420 return false, err
421 }
422 s.recordSubscriptionStatus(d, "completed")
423
424 return true, nil
425}
426
427// runYTDLP streams yt-dlp's combined output into the progress cache, flushing it
428// to the persisted log as it goes. Cancelling ctx kills the whole process group.
429func (s *DownloadService) runYTDLP(ctx context.Context, d *models.Download, args []string) error {
430 // --newline: without it progress is rewritten in place with carriage returns,
431 // so a long download becomes one ever-growing line that overflows the reader's
432 // buffer and stalls the pipe.
433 // --no-write-playlist-metafiles: the playlist-level info.json lands in the
434 // first item dir and would be imported as if it were an item.
435 // --socket-timeout: the only bound on a stalled connection. Without it a dead
436 // socket pins a worker forever.
437 fullArgs := append([]string{"--newline", "--no-write-playlist-metafiles", "--socket-timeout", "30"}, args...)
438 cmd := util.KillableCommand(ctx, s.cfg.YTDLPPath, fullArgs...)
439
440 stdout, err := cmd.StdoutPipe()
441 if err != nil {
442 return err
443 }
444 cmd.Stderr = cmd.Stdout
445
446 if err := cmd.Start(); err != nil {
447 return err
448 }
449
450 ticker := time.NewTicker(10 * time.Second)
451 defer ticker.Stop()
452 done := make(chan struct{})
453 go func() {
454 for {
455 select {
456 case <-ticker.C:
457 s.flushLogs(d.ID)
458 case <-done:
459 return
460 }
461 }
462 }()
463
464 scanner := bufio.NewScanner(stdout)
465 // A single yt-dlp message can exceed the 64 KiB default, and an aborted
466 // scanner leaves the pipe unread.
467 scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
468 for scanner.Scan() {
469 s.cache.AppendLog(d.ID, scanner.Text())
470 }
471 if err := scanner.Err(); err != nil {
472 slog.Error("error reading yt-dlp output", "download_id", d.ID, "err", err)
473 }
474 close(done)
475
476 s.flushLogs(d.ID)
477
478 return cmd.Wait()
479}
480
481// appendCookies writes the saved cookies, if any, to a temp file. The returned
482// cleanup saves back what yt-dlp left behind and is always safe to call.
483func (s *DownloadService) appendCookies(args []string) ([]string, func()) {
484 cookies, err := s.settingsSvc.GetCookies()
485 if err != nil || strings.TrimSpace(cookies) == "" {
486 return args, func() {}
487 }
488 path, err := s.writeCookiesFile(cookies)
489 if err != nil {
490 return args, func() {}
491 }
492 return append(args, "--cookies", path), func() {
493 s.saveRefreshedCookies(path, cookies)
494 os.Remove(path)
495 }
496}
497
498// saveRefreshedCookies stores back what yt-dlp wrote to the cookie file. YouTube
499// rotates session cookies on use, so replaying the snapshot we sent invalidates
500// the session and the user has to export cookies again. sent is that snapshot,
501// so an untouched file saves nothing.
502func (s *DownloadService) saveRefreshedCookies(path, sent string) {
503 data, err := os.ReadFile(path)
504 // A missing file means yt-dlp never got that far; empty content would wipe
505 // working cookies.
506 if err != nil || strings.TrimSpace(string(data)) == "" || string(data) == sent {
507 return
508 }
509 if err := s.settingsSvc.SetCookies(string(data)); err != nil {
510 slog.Error("failed to save refreshed cookies", "err", err)
511 }
512}
513
514// writeCookiesFile removes the partial file on any failure: a truncated cookies
515// file must not be handed to yt-dlp.
516func (s *DownloadService) writeCookiesFile(cookies string) (string, error) {
517 if err := os.MkdirAll(s.cfg.TempDir, 0o755); err != nil {
518 return "", err
519 }
520 tmpFile, err := os.CreateTemp(s.cfg.TempDir, "cookies-*.txt")
521 if err != nil {
522 return "", err
523 }
524 if _, err := tmpFile.WriteString(cookies); err != nil {
525 tmpFile.Close()
526 os.Remove(tmpFile.Name())
527 return "", err
528 }
529 if err := tmpFile.Close(); err != nil {
530 os.Remove(tmpFile.Name())
531 return "", err
532 }
533 return tmpFile.Name(), nil
534}
535
536func (s *DownloadService) finalizeError(d *models.Download, err error) {
537 s.flushLogs(d.ID)
538 if markErr := s.repo.MarkError(d.ID, err.Error()); markErr != nil {
539 slog.Error("failed to record download error", "download_id", d.ID, "err", markErr)
540 }
541 s.recordSubscriptionStatus(d, "error")
542}
543
544// recordSubscriptionStatus mirrors the download's state onto the subscription
545// row, which otherwise keeps the "queued" the scheduler wrote.
546func (s *DownloadService) recordSubscriptionStatus(d *models.Download, status string) {
547 if !d.SubscriptionID.Valid || s.subscriptionSvc == nil {
548 return
549 }
550 if err := s.subscriptionSvc.SetLastStatus(d.SubscriptionID.Int64, status); err != nil {
551 slog.Error("failed to record subscription status", "download_id", d.ID,
552 "subscription_id", d.SubscriptionID.Int64, "status", status, "err", err)
553 }
554}
555
556// ErrCancelled reports a deliberate stop (deleted, queue cleared, or shutdown)
557// rather than a failure, so callers don't log it as an error.
558var ErrCancelled = errors.New("download cancelled")
559
560// finalizeCancelled leaves the row "downloading" on shutdown (parent already
561// cancelled), so ResetStalledDownloads resumes it on the next start. Only a
562// user-initiated cancel is terminal, and its row is often already deleted, so a
563// missing row is fine.
564func (s *DownloadService) finalizeCancelled(parent context.Context, d *models.Download) error {
565 s.flushLogs(d.ID)
566
567 // The subscription status is left alone too: the run resumes, so it is still
568 // in progress rather than cancelled.
569 if parent.Err() != nil {
570 return ErrCancelled
571 }
572
573 if err := s.repo.MarkCompleted(d.ID, "cancelled"); err != nil {
574 slog.Error("failed to record cancellation", "download_id", d.ID, "err", err)
575 }
576 s.recordSubscriptionStatus(d, "cancelled")
577 return ErrCancelled
578}
579
580func (s *DownloadService) flushLogs(id int64) {
581 logs := s.cache.FlushLogs(id)
582 if logs == "" {
583 return
584 }
585 if err := s.repo.AppendLogs(id, logs); err != nil {
586 slog.Error("failed to persist logs", "download_id", id, "err", err)
587 }
588}
589
590func sqlNullInt64(v int64) sql.NullInt64 {
591 return sql.NullInt64{Int64: v, Valid: true}
592}
593