package service import ( "bufio" "context" "database/sql" "errors" "fmt" "log" "os" "os/exec" "path/filepath" "strconv" "strings" "sync" "syscall" "time" "vidarchive/internal/config" "vidarchive/internal/models" "vidarchive/internal/repository" ) type DownloadService struct { repo *repository.DownloadRepository librarySvc *LibraryService presetSvc *PresetService settingsSvc *SettingsService subscriptionSvc *SubscriptionService cfg *config.Config cache *ProgressCache activeMu sync.Mutex active map[int64]context.CancelFunc } func NewDownloadService(repo *repository.DownloadRepository, librarySvc *LibraryService, presetSvc *PresetService, settingsSvc *SettingsService, subscriptionSvc *SubscriptionService, cfg *config.Config) *DownloadService { return &DownloadService{ repo: repo, librarySvc: librarySvc, presetSvc: presetSvc, settingsSvc: settingsSvc, subscriptionSvc: subscriptionSvc, cfg: cfg, cache: NewProgressCache(), active: make(map[int64]context.CancelFunc), } } func (s *DownloadService) Create(url string, presetID *int64, formatOverride, customFlags, outputDir string) (*models.Download, error) { d := &models.Download{ URL: url, Status: "queued", FormatOverride: formatOverride, CustomFlags: customFlags, OutputDir: sql.NullString{String: outputDir, Valid: outputDir != ""}, } if presetID != nil { d.PresetID = sqlNullInt64(*presetID) } if err := s.repo.Create(d); err != nil { return nil, err } return d, nil } // CreateForSubscription queues a download for a subscription run, copying its // download options and tagging it with the subscription id so ExecuteDownload // applies the right refresh mode and pruning. func (s *DownloadService) CreateForSubscription(sub *models.Subscription) (*models.Download, error) { d := &models.Download{ URL: sub.URL, Status: "queued", FormatOverride: sub.FormatOverride, CustomFlags: sub.CustomFlags, OutputDir: sql.NullString{String: sub.OutputDir, Valid: sub.OutputDir != ""}, PresetID: sub.PresetID, SubscriptionID: sqlNullInt64(sub.ID), } if err := s.repo.Create(d); err != nil { return nil, err } return d, nil } func (s *DownloadService) GetByID(id int64) (*models.Download, error) { d, err := s.repo.GetByID(id) if err != nil { return nil, err } if logs := s.cache.Snapshot(id); logs != "" { d.Logs = sql.NullString{String: logs, Valid: true} } return d, nil } func (s *DownloadService) GetAll(status, sortBy string) ([]*models.Download, error) { downloads, err := s.repo.GetAll(status, sortBy) if err != nil { return nil, err } for _, d := range downloads { if logs := s.cache.Snapshot(d.ID); logs != "" { d.Logs = sql.NullString{String: logs, Valid: true} } } return downloads, nil } func (s *DownloadService) GetQueued(limit int) ([]*models.Download, error) { return s.repo.GetQueued(limit) } // HasActiveForSubscription reports whether the subscription already has a queued // or in-progress download, so the scheduler can skip stacking another run. func (s *DownloadService) HasActiveForSubscription(subID int64) (bool, error) { return s.repo.HasActiveForSubscription(subID) } func (s *DownloadService) Delete(id int64) error { s.cancelDownload(id) s.cache.Delete(id) return s.repo.Delete(id) } // registerActive records the cancel func for a claimed download and returns a // release func. Registration happens at claim time rather than after the process // spawns, so a delete arriving during setup, between yt-dlp and the import, or // mid-import still stops the work instead of silently letting it finish. func (s *DownloadService) registerActive(id int64, cancel context.CancelFunc) func() { s.activeMu.Lock() s.active[id] = cancel s.activeMu.Unlock() return func() { s.activeMu.Lock() delete(s.active, id) s.activeMu.Unlock() } } func (s *DownloadService) cancelDownload(id int64) { s.activeMu.Lock() cancel, ok := s.active[id] delete(s.active, id) s.activeMu.Unlock() if ok { cancel() } } // CancelAll stops every download currently in flight. Used on shutdown and when // clearing the queue, so no yt-dlp child outlives the rows that described it. func (s *DownloadService) CancelAll() { s.activeMu.Lock() cancels := make([]context.CancelFunc, 0, len(s.active)) for id, cancel := range s.active { cancels = append(cancels, cancel) delete(s.active, id) } s.activeMu.Unlock() for _, cancel := range cancels { cancel() } } func (s *DownloadService) DeleteAll() error { // Clearing the queue must also stop what is running; otherwise yt-dlp keeps // going and imports into the library after its row is gone. s.CancelAll() return s.repo.DeleteAll() } func (s *DownloadService) Ping(ctx context.Context) error { return s.repo.Ping(ctx) } // ResetStalledDownloads re-queues downloads left mid-flight by a previous run and // discards their temp directories. Without the cleanup the re-run imports into a // fresh uniqueDir and the library ends up with a duplicate of the same item. func (s *DownloadService) ResetStalledDownloads() error { ids, err := s.repo.IDsByStatus("downloading") if err != nil { return err } for _, id := range ids { for _, dir := range s.tempDirsFor(id) { if err := os.RemoveAll(dir); err != nil { log.Printf("warning: failed to remove stale temp dir %s: %v", dir, err) } } } return s.repo.UpdateStatusWhere("downloading", "queued") } // tempDirFor returns the scratch directory a download writes into. func (s *DownloadService) tempDirFor(id int64) string { return filepath.Join(s.cfg.TempDir, strconv.FormatInt(id, 10)) } // tempNewDirFor returns the second-pass scratch directory used by metadata mode. func (s *DownloadService) tempNewDirFor(id int64) string { return s.tempDirFor(id) + "-new" } // tempDirsFor returns every scratch directory a download owns. ResetStalledDownloads // clears these, so the two builders above must stay the only places that name them. func (s *DownloadService) tempDirsFor(id int64) []string { return []string{s.tempDirFor(id), s.tempNewDirFor(id)} } // ExecuteDownload runs the download for d. The bool reports whether this call // actually processed it: false means another worker already claimed it (Submit // and the queue checker can both enqueue the same row within the 2s poll window), // so the caller should not log it as completed. A cancelled run returns // ErrCancelled. // // parent belongs to the worker pool: deriving from it means a shutdown cancels // the download even if it lands before this call registers its own cancel func. func (s *DownloadService) ExecuteDownload(parent context.Context, d *models.Download) (bool, error) { claimed, err := s.repo.MarkStarted(d.ID) if err != nil { return false, err } if !claimed { return false, nil } // Registered before any work starts so Delete/CancelAll can interrupt every // phase, not just the window where yt-dlp happens to be running. ctx, cancel := context.WithCancel(parent) defer cancel() defer s.registerActive(d.ID, cancel)() s.cache.Set(d.ID, &LiveDownload{}) defer s.cache.Delete(d.ID) var preset *models.Preset if d.PresetID.Valid { preset, err = s.presetSvc.GetByID(d.PresetID.Int64) if err != nil { log.Printf("download %d: preset %d lookup failed (%v); falling back to default", d.ID, d.PresetID.Int64, err) preset = nil } } if preset == nil { var derr error if preset, derr = s.presetSvc.GetDefault(); derr != nil { log.Printf("download %d: no default preset available (%v); using built-in defaults", d.ID, derr) preset = &models.Preset{} } } var sub *models.Subscription if d.SubscriptionID.Valid && s.subscriptionSvc != nil { var serr error if sub, serr = s.subscriptionSvc.GetByID(d.SubscriptionID.Int64); serr != nil { log.Printf("download %d: subscription %d lookup failed: %v", d.ID, d.SubscriptionID.Int64, serr) } } // Reject custom flags that clash with options VidArchive sets itself, before // spending any work — the download fails with a message naming the offender. isSubscription := d.SubscriptionID.Valid for _, flags := range []string{d.CustomFlags, preset.CustomFlags} { if err := checkReservedFlags(flags, isSubscription); err != nil { s.finalizeError(d, err) return false, err } } // From here the run is really under way, so a subscription shows "downloading" // instead of the "queued" the scheduler recorded. s.recordSubscriptionStatus(d, "downloading") tempDownloadDir := s.tempDirFor(d.ID) if err := os.MkdirAll(tempDownloadDir, 0755); err != nil { return false, fmt.Errorf("create temp download dir: %w", err) } // Own the temp dir's lifetime here, where it's created, so it's removed on // every exit path — including a failed yt-dlp run or an early return that // crashes mid-import. The import helpers below no longer clean it up. defer os.RemoveAll(tempDownloadDir) args := s.presetSvc.BuildArgs(preset, d.FormatOverride, d.CustomFlags) // Record the meaningful flags (format/audio/subs/custom) that shaped this // download, before the internal plumbing (cookies, -P/-o, URL) is appended, // so each imported item can show how it was fetched. ytdlpFlags := strings.Join(args, " ") var cookieCleanup func() args, cookieCleanup = s.appendCookies(args) defer cookieCleanup() if sub != nil { // Always write info.json so the import step can read the stable identity // (yt-dlp's video id) used to match/replace existing items. args = append(args, "--write-info-json") switch sub.RefreshMode { case "skip": // Let yt-dlp skip entries already recorded — no re-download. archive := s.subscriptionSvc.ArchivePath(sub.ID) if err := os.MkdirAll(filepath.Dir(archive), 0755); err == nil { args = append(args, "--download-archive", archive) } case "metadata": // Refresh metadata only; don't fetch media. args = append(args, "--skip-download") } } args = append(args, "-P", tempDownloadDir) args = append(args, "-o", "item-%(autonumber)05d/%(title)s.%(ext)s") args = append(args, d.URL) runErr := s.runYTDLP(ctx, d, args) // Cancellation wins over both the run error and the import: a cancel that // lands just after yt-dlp exited 0 leaves runErr nil, and the item must not // reach the library after the user removed it. if ctx.Err() != nil { return false, s.finalizeCancelled(parent, d) } if runErr != nil { s.finalizeError(d, runErr) return false, runErr } mode := "" if sub != nil { mode = sub.RefreshMode } // Post-process before marking completed, so the download stays "downloading" // until everything is really done — including metadata mode's second pass, // which downloads any genuinely new entries as full items. var postErr error if mode == "metadata" { // The main pass ran with --skip-download, so the temp dir holds only // info.json files: refresh existing items in place and fetch new ones. postErr = s.refreshAndAddNew(ctx, d, preset, tempDownloadDir, ytdlpFlags) } else { imported, err := s.importDownloadedItems(ctx, d, tempDownloadDir, mode, ytdlpFlags) switch { case err != nil: postErr = err // A plain (non-subscription) download that yields nothing is a failure, not // a silent "completed". Subscription modes legitimately import zero (skip // mode, or a metadata refresh with no new entries), so only enforce this for // plain runs. case sub == nil && imported == 0: postErr = fmt.Errorf("yt-dlp finished but no media files were downloaded") } } if postErr == nil && sub != nil && sub.PruneRemoved { s.pruneSubscription(ctx, d, sub) } // Same cancellation check as above: a stop during post-processing must not // be recorded as "completed". if ctx.Err() != nil { return false, s.finalizeCancelled(parent, d) } if postErr != nil { s.finalizeError(d, postErr) return false, postErr } if err := s.repo.MarkCompleted(d.ID, "completed"); err != nil { return false, err } s.recordSubscriptionStatus(d, "completed") return true, nil } // runYTDLP executes yt-dlp with args, streaming combined output into the live // progress cache and periodically flushing it to the download's persisted log. // Cancelling ctx kills the whole process group and makes this return. func (s *DownloadService) runYTDLP(ctx context.Context, d *models.Download, args []string) error { // --newline forces yt-dlp to emit each progress update on its own line. Without // it, progress is rewritten in place with carriage returns, so a long download // becomes one ever-growing line that overflows the reader's buffer and stalls // the pipe — hanging the download. See the hardened scanner below. fullArgs := append([]string{"--newline"}, args...) cmd := exec.CommandContext(ctx, s.cfg.YTDLPPath, fullArgs...) cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} // yt-dlp spawns helpers (ffmpeg, external downloaders). Kill the whole group // rather than just the parent, which would leave those orphaned. cmd.Cancel = func() error { return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) } stdout, err := cmd.StdoutPipe() if err != nil { return err } cmd.Stderr = cmd.Stdout if err := cmd.Start(); err != nil { return err } ticker := time.NewTicker(10 * time.Second) defer ticker.Stop() done := make(chan struct{}) go func() { for { select { case <-ticker.C: s.flushLogs(d.ID) case <-done: return } } }() scanner := bufio.NewScanner(stdout) // Allow long lines (a single yt-dlp message can exceed the 64 KiB default) // rather than letting the scanner abort and leave the pipe unread. scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) for scanner.Scan() { s.cache.AppendLog(d.ID, scanner.Text()) } if err := scanner.Err(); err != nil { log.Printf("download %d: error reading yt-dlp output: %v", d.ID, err) } close(done) s.flushLogs(d.ID) return cmd.Wait() } // appendCookies writes the saved cookies (if any) to a temp file and appends a // --cookies flag. The returned cleanup removes the temp file and is always safe // to call, even when no cookies were configured. func (s *DownloadService) appendCookies(args []string) ([]string, func()) { cookies, err := s.settingsSvc.GetCookies() if err != nil || strings.TrimSpace(cookies) == "" { return args, func() {} } path, err := s.writeCookiesFile(cookies) if err != nil { return args, func() {} } return append(args, "--cookies", path), func() { os.Remove(path) } } // writeCookiesFile writes cookies to a temp file in the app's own temp dir (the // same volume the rest of the run uses). On any failure the partial file is // removed — a truncated cookies file must not be handed to yt-dlp. func (s *DownloadService) writeCookiesFile(cookies string) (string, error) { if err := os.MkdirAll(s.cfg.TempDir, 0755); err != nil { return "", err } tmpFile, err := os.CreateTemp(s.cfg.TempDir, "cookies-*.txt") if err != nil { return "", err } if _, err := tmpFile.WriteString(cookies); err != nil { tmpFile.Close() os.Remove(tmpFile.Name()) return "", err } if err := tmpFile.Close(); err != nil { os.Remove(tmpFile.Name()) return "", err } return tmpFile.Name(), nil } func (s *DownloadService) finalizeError(d *models.Download, err error) { s.flushLogs(d.ID) if markErr := s.repo.MarkError(d.ID, err.Error()); markErr != nil { log.Printf("download %d: failed to record error: %v", d.ID, markErr) } s.recordSubscriptionStatus(d, "error") } // recordSubscriptionStatus mirrors a subscription download's state onto the // subscription row. Without it the row keeps the status it had when the // scheduler queued it, so the subscriptions page reports "queued" long after the // run finished — or failed. func (s *DownloadService) recordSubscriptionStatus(d *models.Download, status string) { if !d.SubscriptionID.Valid || s.subscriptionSvc == nil { return } if err := s.subscriptionSvc.SetLastStatus(d.SubscriptionID.Int64, status); err != nil { log.Printf("download %d: failed to record subscription %d status %q: %v", d.ID, d.SubscriptionID.Int64, status, err) } } // ErrCancelled reports that a download was deliberately stopped (deleted, queue // cleared, or shutdown) rather than having failed. Callers distinguish it so a // cancellation isn't logged as an error. var ErrCancelled = errors.New("download cancelled") // finalizeCancelled records a stopped download. // // A shutdown (parent already cancelled) deliberately leaves the row // "downloading": ResetStalledDownloads re-queues it on the next start, so // stopping the server resumes the download instead of losing it. Only a // user-initiated cancel is terminal. The row may already be deleted in that // case — cancellation usually arrives via Delete — so a missing row is fine. func (s *DownloadService) finalizeCancelled(parent context.Context, d *models.Download) error { s.flushLogs(d.ID) // A shutdown leaves the subscription status alone too: the run resumes on the // next start, so it is still in progress rather than cancelled. if parent.Err() != nil { return ErrCancelled } if err := s.repo.MarkCompleted(d.ID, "cancelled"); err != nil { log.Printf("download %d: failed to record cancellation: %v", d.ID, err) } s.recordSubscriptionStatus(d, "cancelled") return ErrCancelled } // flushLogs persists whatever output has accumulated for a download. func (s *DownloadService) flushLogs(id int64) { logs := s.cache.FlushLogs(id) if logs == "" { return } if err := s.repo.AppendLogs(id, logs); err != nil { log.Printf("download %d: failed to persist logs: %v", id, err) } } func sqlNullInt64(v int64) sql.NullInt64 { return sql.NullInt64{Int64: v, Valid: true} }