various fixes
A.dockerignore
@@ -0,0 +1,10 @@
# The build context is copied wholesale by "COPY . ." — keep the runtime data
# (live database, WAL, and the entire downloaded library) out of the image.
data/
.git/
.gitignore
*.md
Containerfile
compose.yml
.dockerignore
MContainerfile
@@ -9,7 +9,11 @@ RUN CGO_ENABLED=0 go build -o vidarchive ./cmd/vidarchive
FROM alpine:3.24
RUN apk --no-cache add ca-certificates ffmpeg yt-dlp
RUN apk --no-cache add ca-certificates ffmpeg python3 py3-pip \
&& python3 -m venv /opt/ytdlp \
&& /opt/ytdlp/bin/pip install --no-cache-dir --upgrade yt-dlp
ENV PATH="/opt/ytdlp/bin:${PATH}"
WORKDIR /app
@@ -23,4 +27,11 @@ VOLUME ["/data"]
EXPOSE 8080
# /healthz reports 503 when the database is unreachable. start-period covers
# migrations on first boot so they don't count as failures.
HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \
CMD wget -q -O /dev/null "http://127.0.0.1:${VIDARCHIVE_PORT}/healthz" || exit 1
# exec form, so the app is PID 1 and gets SIGTERM directly — it needs that for a
# graceful shutdown (drain requests, kill yt-dlp children, checkpoint the DB).
ENTRYPOINT ["/app/vidarchive"]
Mcmd/vidarchive/main.go
@@ -1,11 +1,14 @@
package main
import (
"context"
"fmt"
"log"
"os"
"os/exec"
"os/signal"
"path/filepath"
"syscall"
"time"
"vidarchive/internal/config"
@@ -18,6 +21,11 @@ import (
)
func main() {
// Cancelled on SIGINT/SIGTERM so the server drains, workers stop, running
// yt-dlp children are killed and the database is checkpointed and closed.
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
cfg := config.New()
checkDependencies(cfg)
@@ -39,7 +47,6 @@ func main() {
if err != nil {
log.Fatalf("Failed to initialize database: %v", err)
}
defer db.Close()
presetRepo := repository.NewPresetRepository(db)
downloadRepo := repository.NewDownloadRepository(db)
@@ -54,11 +61,9 @@ func main() {
workerPool := worker.New(downloadSvc, cfg.Workers)
workerPool.Start()
defer workerPool.Stop()
scheduler := worker.NewScheduler(subscriptionSvc, downloadSvc, workerPool, time.Duration(cfg.SchedulerInterval)*time.Second)
scheduler.Start()
defer scheduler.Stop()
h, err := handler.New(cfg, presetSvc, downloadSvc, librarySvc, settingsSvc, subscriptionSvc, workerPool)
if err != nil {
@@ -72,8 +77,24 @@ func main() {
fmt.Printf("Library directory: %s\n", cfg.LibraryDir)
fmt.Printf("Temp directory: %s\n", cfg.TempDir)
if err := srv.Start(); err != nil {
log.Fatalf("Server error: %v", err)
srvErr := srv.Start(ctx)
// Shut down in dependency order: stop scheduling new runs, then stop the
// workers (killing any running yt-dlp), then checkpoint and close the DB.
// This runs on both a signal and a server error, so nothing is left orphaned.
log.Printf("Shutting down...")
scheduler.Stop()
workerPool.Stop()
if err := database.Checkpoint(db); err != nil {
log.Printf("Warning: %v", err)
}
if err := db.Close(); err != nil {
log.Printf("Warning: failed to close database: %v", err)
}
log.Printf("Shutdown complete")
if srvErr != nil {
log.Fatalf("Server error: %v", srvErr)
}
}
Mcompose.yml
@@ -1,5 +1,7 @@
services:
vidarchive:
# yt-dlp is installed when the image is built. Rebuild the image to get a
# newer one: `docker compose build --pull --no-cache vidarchive`.
build:
context: .
dockerfile: Containerfile
@@ -18,3 +20,6 @@ services:
# Uncomment to use a custom yt-dlp path:
# - VIDARCHIVE_YTDLP_PATH=/usr/bin/yt-dlp
restart: unless-stopped
# Give in-flight downloads and the database checkpoint time to finish before
# the container is killed; the default 10s can cut a shutdown short.
stop_grace_period: 60s
Minternal/config/config.go
@@ -36,9 +36,9 @@ func New() *Config {
FFmpegPath: getEnv("VIDARCHIVE_FFMPEG_PATH", "ffmpeg"),
FFprobePath: getEnv("VIDARCHIVE_FFPROBE_PATH", "ffprobe"),
BaseURL: getEnv("VIDARCHIVE_BASE_URL", ""),
Workers: getEnvInt("VIDARCHIVE_WORKERS", 2),
RefreshInterval: getEnvInt("VIDARCHIVE_REFRESH_INTERVAL", 5),
SchedulerInterval: getEnvInt("VIDARCHIVE_SCHEDULER_INTERVAL", 60),
Workers: getEnvIntMin("VIDARCHIVE_WORKERS", 2, 1),
RefreshInterval: getEnvIntMin("VIDARCHIVE_REFRESH_INTERVAL", 5, 1),
SchedulerInterval: getEnvIntMin("VIDARCHIVE_SCHEDULER_INTERVAL", 60, 1),
}
}
@@ -65,3 +65,15 @@ func getEnvInt(key string, defaultVal int) int {
}
return i
}
// getEnvIntMin is getEnvInt with a lower bound. A value below min is rejected in
// favour of the default: a zero or negative worker count or tick interval would
// otherwise stall downloads or spin the scheduler.
func getEnvIntMin(key string, defaultVal, min int) int {
i := getEnvInt(key, defaultVal)
if i < min {
fmt.Fprintf(os.Stderr, "invalid %s: %d is below the minimum %d, using default %d\n", key, i, min, defaultVal)
return defaultVal
}
return i
}
Minternal/database/database.go
@@ -30,14 +30,30 @@ func New(cfg *config.Config) (*sql.DB, error) {
if _, err := db.Exec("PRAGMA busy_timeout=5000"); err != nil {
return nil, fmt.Errorf("set busy timeout: %w", err)
}
// Migrations run with foreign keys off: the table rebuilds below drop and
// rename tables, which enforcement would reject mid-flight.
if err := migrate(db); err != nil {
return nil, fmt.Errorf("migrate: %w", err)
}
// SQLite ignores FOREIGN KEY clauses unless this is enabled per connection,
// so without it the declared constraints are decorative.
if _, err := db.Exec("PRAGMA foreign_keys=ON"); err != nil {
return nil, fmt.Errorf("enable foreign keys: %w", err)
}
return db, nil
}
// Checkpoint flushes the write-ahead log into the main database file. Call it
// before closing so a killed process doesn't leave a large uncheckpointed WAL.
func Checkpoint(db *sql.DB) error {
if _, err := db.Exec("PRAGMA wal_checkpoint(TRUNCATE)"); err != nil {
return fmt.Errorf("wal checkpoint: %w", err)
}
return nil
}
func migrate(db *sql.DB) error {
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY)`); err != nil {
return fmt.Errorf("create migrations table: %w", err)
@@ -116,18 +132,97 @@ func migrate(db *sql.DB) error {
FOREIGN KEY (preset_id) REFERENCES presets(id)
)`},
{14, `ALTER TABLE downloads ADD COLUMN subscription_id INTEGER`},
// Rebuild downloads so its foreign keys survive enforcement: deleting a
// preset or subscription now nulls the reference instead of failing, which
// matches how ExecuteDownload already degrades to the default preset.
// subscription_id gains the foreign key it never had.
{15, `CREATE TABLE downloads_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT NOT NULL,
status TEXT DEFAULT 'queued',
logs TEXT,
error_message TEXT,
preset_id INTEGER,
format_override TEXT,
custom_flags TEXT,
output_dir TEXT,
subscription_id INTEGER,
started_at DATETIME,
completed_at DATETIME,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (preset_id) REFERENCES presets(id) ON DELETE SET NULL,
FOREIGN KEY (subscription_id) REFERENCES subscriptions(id) ON DELETE SET NULL
);
INSERT INTO downloads_new (id, url, status, logs, error_message, preset_id,
format_override, custom_flags, output_dir, subscription_id, started_at,
completed_at, created_at)
SELECT id, url, status, logs, error_message, preset_id, format_override,
custom_flags, output_dir, subscription_id, started_at, completed_at,
created_at FROM downloads;
DROP TABLE downloads;
ALTER TABLE downloads_new RENAME TO downloads;`},
{16, `CREATE TABLE subscriptions_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
url TEXT NOT NULL,
enabled INTEGER DEFAULT 1,
refresh_mode TEXT DEFAULT 'overwrite',
schedule_kind TEXT DEFAULT 'daily',
cron_expr TEXT DEFAULT '0 3 * * *',
preset_id INTEGER,
format_override TEXT,
custom_flags TEXT,
output_dir TEXT NOT NULL,
prune_removed INTEGER DEFAULT 0,
last_run_at DATETIME,
next_run_at DATETIME,
last_status TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (preset_id) REFERENCES presets(id) ON DELETE SET NULL
);
INSERT INTO subscriptions_new SELECT id, name, url, enabled, refresh_mode,
schedule_kind, cron_expr, preset_id, format_override, custom_flags,
output_dir, prune_removed, last_run_at, next_run_at, last_status,
created_at FROM subscriptions;
DROP TABLE subscriptions;
ALTER TABLE subscriptions_new RENAME TO subscriptions;`},
// Indexes for the hot predicates: the queue poll and claim filter on
// status, the scheduler's dedup check filters on subscription_id, and
// GetDue filters on enabled + next_run_at.
{17, `CREATE INDEX IF NOT EXISTS idx_downloads_status ON downloads(status);
CREATE INDEX IF NOT EXISTS idx_downloads_subscription_id ON downloads(subscription_id);
CREATE INDEX IF NOT EXISTS idx_downloads_created_at ON downloads(created_at);
CREATE INDEX IF NOT EXISTS idx_subscriptions_due ON subscriptions(enabled, next_run_at);`},
}
for _, m := range migrations {
if m.version <= currentVersion {
continue
}
if _, err := db.Exec(m.sql); err != nil {
return fmt.Errorf("migration %d: %w", m.version, err)
}
if _, err := db.Exec(`INSERT INTO schema_migrations (version) VALUES (?)`, m.version); err != nil {
return fmt.Errorf("record migration %d: %w", m.version, err)
// Each migration runs in its own transaction so a multi-statement failure
// can't leave the schema half-applied with no version recorded.
if err := applyMigration(db, m.version, m.sql); err != nil {
return err
}
}
return nil
}
func applyMigration(db *sql.DB, version int, stmt string) error {
tx, err := db.Begin()
if err != nil {
return fmt.Errorf("migration %d: begin: %w", version, err)
}
defer tx.Rollback()
if _, err := tx.Exec(stmt); err != nil {
return fmt.Errorf("migration %d: %w", version, err)
}
if _, err := tx.Exec(`INSERT INTO schema_migrations (version) VALUES (?)`, version); err != nil {
return fmt.Errorf("record migration %d: %w", version, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("migration %d: commit: %w", version, err)
}
return nil
}
Ainternal/database/database_test.go
@@ -0,0 +1,111 @@
package database
import (
"database/sql"
"path/filepath"
"testing"
"vidarchive/internal/config"
)
func newTestDB(t *testing.T) *sql.DB {
t.Helper()
db, err := New(&config.Config{DBPath: filepath.Join(t.TempDir(), "test.db")})
if err != nil {
t.Fatalf("New: %v", err)
}
t.Cleanup(func() { db.Close() })
return db
}
func TestMigrateIsIdempotent(t *testing.T) {
path := filepath.Join(t.TempDir(), "test.db")
db, err := New(&config.Config{DBPath: path})
if err != nil {
t.Fatalf("first migrate: %v", err)
}
var first int
if err := db.QueryRow(`SELECT MAX(version) FROM schema_migrations`).Scan(&first); err != nil {
t.Fatalf("read version: %v", err)
}
db.Close()
db, err = New(&config.Config{DBPath: path})
if err != nil {
t.Fatalf("second migrate: %v", err)
}
defer db.Close()
var second int
if err := db.QueryRow(`SELECT MAX(version) FROM schema_migrations`).Scan(&second); err != nil {
t.Fatalf("read version: %v", err)
}
if first != second {
t.Errorf("version changed on re-open: %d -> %d", first, second)
}
}
func TestForeignKeysAreEnforced(t *testing.T) {
db := newTestDB(t)
var on int
if err := db.QueryRow(`PRAGMA foreign_keys`).Scan(&on); err != nil {
t.Fatalf("read pragma: %v", err)
}
if on != 1 {
t.Fatal("foreign_keys pragma is off; declared constraints would be ignored")
}
if _, err := db.Exec(`INSERT INTO downloads (url, preset_id) VALUES ('u', 99999)`); err == nil {
t.Error("expected insert with a nonexistent preset_id to be rejected")
}
}
// Deleting a preset must not fail just because downloads reference it: the
// reference is nulled and ExecuteDownload falls back to the default preset.
func TestDeletingPresetNullsReferences(t *testing.T) {
db := newTestDB(t)
res, err := db.Exec(`INSERT INTO presets (name) VALUES ('temp')`)
if err != nil {
t.Fatalf("insert preset: %v", err)
}
presetID, err := res.LastInsertId()
if err != nil {
t.Fatal(err)
}
if _, err := db.Exec(`INSERT INTO downloads (url, preset_id) VALUES ('u', ?)`, presetID); err != nil {
t.Fatalf("insert download: %v", err)
}
if _, err := db.Exec(`DELETE FROM presets WHERE id = ?`, presetID); err != nil {
t.Fatalf("delete preset: %v", err)
}
var nulls int
if err := db.QueryRow(`SELECT COUNT(*) FROM downloads WHERE preset_id IS NULL`).Scan(&nulls); err != nil {
t.Fatal(err)
}
if nulls != 1 {
t.Errorf("preset_id not nulled after preset delete: got %d rows with NULL", nulls)
}
}
func TestExpectedIndexesExist(t *testing.T) {
db := newTestDB(t)
want := []string{
"idx_downloads_status",
"idx_downloads_subscription_id",
"idx_subscriptions_due",
}
for _, name := range want {
var got string
err := db.QueryRow(`SELECT name FROM sqlite_master WHERE type='index' AND name = ?`, name).Scan(&got)
if err != nil {
t.Errorf("missing index %s: %v", name, err)
}
}
}
Minternal/handler/handler.go
@@ -33,8 +33,7 @@ type Handler struct {
}
func New(cfg *config.Config, presetSvc *service.PresetService, downloadSvc *service.DownloadService, librarySvc *service.LibraryService, settingsSvc *service.SettingsService, subscriptionSvc *service.SubscriptionService, workerPool *worker.Pool) (*Handler, error) {
presetSvcGlobal = presetSvc
tmpl, err := loadTemplates()
tmpl, err := loadTemplates(presetSvc)
if err != nil {
return nil, fmt.Errorf("load templates: %w", err)
}
@@ -51,7 +50,10 @@ func New(cfg *config.Config, presetSvc *service.PresetService, downloadSvc *serv
}, nil
}
func loadTemplates() (*template.Template, error) {
// loadTemplates builds the template set. The preset service is passed in and
// captured by the presetFlags func rather than reached through a package-level
// variable, so the handler owns its dependencies.
func loadTemplates(presetSvc *service.PresetService) (*template.Template, error) {
tmpl := template.New("").Funcs(template.FuncMap{
"formatDuration": formatDuration,
"formatFileSize": formatFileSize,
@@ -59,7 +61,7 @@ func loadTemplates() (*template.Template, error) {
"baseName": filepath.Base,
"ext": filepath.Ext,
"mimeType": mimeTypeFromFile,
"presetFlags": func(p *models.Preset) string { return presetSvcGlobal.EffectiveFlags(p, "", "") },
"presetFlags": func(p *models.Preset) string { return presetSvc.EffectiveFlags(p, "", "") },
"urlEncode": template.URLQueryEscaper,
"urlEncodePath": func(p string) string {
parts := strings.Split(strings.Trim(p, "/"), "/")
@@ -104,8 +106,6 @@ func loadTemplates() (*template.Template, error) {
return tmpl.ParseFS(vidarchive.TemplatesFS, "web/templates/*.html")
}
var presetSvcGlobal *service.PresetService
func formatDuration(seconds int) string {
if seconds <= 0 {
return "--:--"
@@ -272,18 +272,12 @@ func (h *Handler) settingsOrDefault() *models.Settings {
func (h *Handler) Library(w http.ResponseWriter, r *http.Request) {
path := r.URL.Query().Get("path")
sortBy := r.URL.Query().Get("sort")
filter := r.URL.Query().Get("filter")
if sortBy == "" {
sortBy = getCookie(r, "library_sort", "date")
} else {
setCookie(w, "library_sort", sortBy)
}
sortBy := sortFromRequest(w, r, "library_sort", "date")
items, folders, err := h.librarySvc.GetAll(path, sortBy, filter)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
h.serverError(w, r, "list library", err)
return
}
@@ -335,7 +329,6 @@ func (h *Handler) LibraryItem(w http.ResponseWriter, r *http.Request) {
}
func (h *Handler) libraryDetail(relPath string, w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
item, err := h.librarySvc.GetByRelPath(ctx, relPath)
if err != nil {
@@ -392,7 +385,6 @@ func (h *Handler) ServeMediaItem(w http.ResponseWriter, r *http.Request) {
}
func (h *Handler) serveMedia(relPath string, w http.ResponseWriter, r *http.Request) {
filename := r.URL.Query().Get("file")
if filename == "" {
http.Error(w, "Missing file", http.StatusBadRequest)
@@ -449,7 +441,6 @@ func isAudioFile(item *models.LibraryItem, filename string) bool {
}
func (h *Handler) serveSubtitles(relPath, lang string, w http.ResponseWriter, r *http.Request) {
if lang == "" {
http.Error(w, "Missing language", http.StatusBadRequest)
return
@@ -460,35 +451,32 @@ func (h *Handler) serveSubtitles(relPath, lang string, w http.ResponseWriter, r
return
}
subtitlePath := h.librarySvc.GetSubtitlePath(relPath, lang)
subtitlePath, err := h.librarySvc.GetSubtitlePath(relPath, lang)
if err != nil {
http.Error(w, "Not found", http.StatusNotFound)
return
}
w.Header().Set("Content-Type", "text/vtt")
http.ServeFile(w, r, subtitlePath)
}
func (h *Handler) deleteMedia(relPath string, w http.ResponseWriter, r *http.Request) {
if err := h.librarySvc.Delete(relPath); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
redirectWithError(w, r, "/library", "Couldn't delete this item.", err)
return
}
http.Redirect(w, r, "/library", http.StatusSeeOther)
redirectWithSuccess(w, r, "/library", "Item deleted.")
}
func (h *Handler) Downloads(w http.ResponseWriter, r *http.Request) {
status := r.URL.Query().Get("status")
sortBy := r.URL.Query().Get("sort")
if sortBy == "" {
sortBy = getCookie(r, "queue_sort", "date")
} else {
setCookie(w, "queue_sort", sortBy)
}
sortBy := sortFromRequest(w, r, "queue_sort", "date")
downloads, err := h.downloadSvc.GetAll(status, sortBy)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
h.serverError(w, r, "list downloads", err)
return
}
@@ -512,10 +500,8 @@ func (h *Handler) Downloads(w http.ResponseWriter, r *http.Request) {
}
func (h *Handler) DownloadDetail(w http.ResponseWriter, r *http.Request) {
idStr := chi.URLParam(r, "id")
id, err := strconv.ParseInt(idStr, 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
id, ok := parseID(w, r)
if !ok {
return
}
@@ -559,82 +545,48 @@ func (h *Handler) CreateDownload(w http.ResponseWriter, r *http.Request) {
download, err := h.downloadSvc.Create(url, presetID, formatOverride, customFlags, outputDir)
if err != nil {
flashError(w, "Couldn't queue download: "+err.Error())
http.Redirect(w, r, "/download", http.StatusSeeOther)
redirectWithError(w, r, "/download", "Couldn't queue this download.", err)
return
}
h.workerPool.Submit(download)
flashSuccess(w, "Download queued.")
http.Redirect(w, r, "/queue", http.StatusSeeOther)
}
func (h *Handler) ListFormats(w http.ResponseWriter, r *http.Request) {
url := r.URL.Query().Get("url")
if url == "" {
http.Error(w, "URL is required", http.StatusBadRequest)
return
}
formats, err := h.downloadSvc.ListFormats(url)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
h.renderWithRequest(w, r, "formats", PageData{
Title: "Available Formats",
ActiveTab: "download",
Data: struct {
URL string
Formats []*models.FormatInfo
}{
URL: url,
Formats: formats,
},
})
redirectWithSuccess(w, r, "/queue", "Download queued.")
}
func (h *Handler) DeleteDownload(w http.ResponseWriter, r *http.Request) {
idStr := chi.URLParam(r, "id")
id, err := strconv.ParseInt(idStr, 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
id, ok := parseID(w, r)
if !ok {
return
}
if err := h.downloadSvc.Delete(id); err != nil {
flashError(w, "Couldn't remove download: "+err.Error())
http.Redirect(w, r, "/queue", http.StatusSeeOther)
redirectWithError(w, r, "/queue", "Couldn't remove this download.", err)
return
}
flashSuccess(w, "Download removed.")
http.Redirect(w, r, "/queue", http.StatusSeeOther)
redirectWithSuccess(w, r, "/queue", "Download removed.")
}
func (h *Handler) ClearAllDownloads(w http.ResponseWriter, r *http.Request) {
if err := h.downloadSvc.DeleteAll(); err != nil {
flashError(w, "Couldn't clear the queue: "+err.Error())
http.Redirect(w, r, "/queue", http.StatusSeeOther)
redirectWithError(w, r, "/queue", "Couldn't clear the queue.", err)
return
}
flashSuccess(w, "Queue cleared.")
http.Redirect(w, r, "/queue", http.StatusSeeOther)
redirectWithSuccess(w, r, "/queue", "Queue cleared.")
}
func (h *Handler) Settings(w http.ResponseWriter, r *http.Request) {
presets, err := h.presetSvc.GetAll()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
h.serverError(w, r, "list presets", err)
return
}
settings, err := h.settingsSvc.GetAll()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
h.serverError(w, r, "load settings", err)
return
}
@@ -657,40 +609,59 @@ func (h *Handler) CreatePreset(w http.ResponseWriter, r *http.Request) {
return
}
preset := &models.Preset{
Name: r.FormValue("name"),
Description: r.FormValue("description"),
FormatMode: r.FormValue("format_mode"),
Format: r.FormValue("format"),
Quality: r.FormValue("quality"),
CustomFormat: r.FormValue("custom_format"),
AudioFormat: r.FormValue("audio_format"),
SubLangs: r.FormValue("sub_langs"),
CustomFlags: r.FormValue("custom_flags"),
preset := &models.Preset{}
if err := applyPresetForm(preset, r); err != nil {
redirectWithError(w, r, "/settings", err.Error(), nil)
return
}
preset.IsDefault = r.FormValue("is_default") == "1"
preset.ExtractAudio = r.FormValue("extract_audio") == "1"
preset.EmbedSubs = r.FormValue("embed_subs") == "1"
preset.EmbedThumbnail = r.FormValue("embed_thumbnail") == "1"
preset.EmbedMetadata = r.FormValue("embed_metadata") == "1"
preset.WriteInfoJSON = r.FormValue("write_info_json") == "1"
if err := h.presetSvc.Create(preset); err != nil {
flashError(w, "Couldn't create preset: "+err.Error())
http.Redirect(w, r, "/settings", http.StatusSeeOther)
redirectWithError(w, r, "/settings", "Couldn't create this preset.", err)
return
}
flashSuccess(w, "Preset created.")
http.Redirect(w, r, "/settings", http.StatusSeeOther)
redirectWithSuccess(w, r, "/settings", "Preset created.")
}
// applyPresetForm copies the preset form fields onto p and validates them. It is
// shared by create and update so the two can't drift apart as fields are added.
func applyPresetForm(p *models.Preset, r *http.Request) error {
name := strings.TrimSpace(r.FormValue("name"))
if name == "" {
return fmt.Errorf("A preset needs a name.")
}
// Mirrors the radio options on the settings form; empty means "unspecified"
// and BuildArgs applies its own default.
formatMode := r.FormValue("format_mode")
switch formatMode {
case "", "default", "preset", "custom":
default:
return fmt.Errorf("Unknown format mode %q.", formatMode)
}
p.Name = name
p.Description = r.FormValue("description")
p.FormatMode = formatMode
p.Format = r.FormValue("format")
p.Quality = r.FormValue("quality")
p.CustomFormat = r.FormValue("custom_format")
p.AudioFormat = r.FormValue("audio_format")
p.SubLangs = r.FormValue("sub_langs")
p.CustomFlags = r.FormValue("custom_flags")
p.IsDefault = r.FormValue("is_default") == "1"
p.ExtractAudio = r.FormValue("extract_audio") == "1"
p.EmbedSubs = r.FormValue("embed_subs") == "1"
p.EmbedThumbnail = r.FormValue("embed_thumbnail") == "1"
p.EmbedMetadata = r.FormValue("embed_metadata") == "1"
p.WriteInfoJSON = r.FormValue("write_info_json") == "1"
return nil
}
func (h *Handler) UpdatePreset(w http.ResponseWriter, r *http.Request) {
idStr := chi.URLParam(r, "id")
id, err := strconv.ParseInt(idStr, 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
id, ok := parseID(w, r)
if !ok {
return
}
@@ -705,48 +676,31 @@ func (h *Handler) UpdatePreset(w http.ResponseWriter, r *http.Request) {
return
}
preset.Name = r.FormValue("name")
preset.Description = r.FormValue("description")
preset.FormatMode = r.FormValue("format_mode")
preset.Format = r.FormValue("format")
preset.Quality = r.FormValue("quality")
preset.CustomFormat = r.FormValue("custom_format")
preset.AudioFormat = r.FormValue("audio_format")
preset.SubLangs = r.FormValue("sub_langs")
preset.CustomFlags = r.FormValue("custom_flags")
preset.IsDefault = r.FormValue("is_default") == "1"
preset.ExtractAudio = r.FormValue("extract_audio") == "1"
preset.EmbedSubs = r.FormValue("embed_subs") == "1"
preset.EmbedThumbnail = r.FormValue("embed_thumbnail") == "1"
preset.EmbedMetadata = r.FormValue("embed_metadata") == "1"
preset.WriteInfoJSON = r.FormValue("write_info_json") == "1"
if err := applyPresetForm(preset, r); err != nil {
redirectWithError(w, r, "/settings", err.Error(), nil)
return
}
if err := h.presetSvc.Update(preset); err != nil {
flashError(w, "Couldn't update preset: "+err.Error())
http.Redirect(w, r, "/settings", http.StatusSeeOther)
redirectWithError(w, r, "/settings", "Couldn't update this preset.", err)
return
}
flashSuccess(w, "Preset updated.")
http.Redirect(w, r, "/settings", http.StatusSeeOther)
redirectWithSuccess(w, r, "/settings", "Preset updated.")
}
func (h *Handler) DeletePreset(w http.ResponseWriter, r *http.Request) {
idStr := chi.URLParam(r, "id")
id, err := strconv.ParseInt(idStr, 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
id, ok := parseID(w, r)
if !ok {
return
}
if err := h.presetSvc.Delete(id); err != nil {
flashError(w, "Couldn't delete preset: "+err.Error())
http.Redirect(w, r, "/settings", http.StatusSeeOther)
redirectWithError(w, r, "/settings", "Couldn't delete this preset.", err)
return
}
flashSuccess(w, "Preset deleted.")
http.Redirect(w, r, "/settings", http.StatusSeeOther)
redirectWithSuccess(w, r, "/settings", "Preset deleted.")
}
func (h *Handler) UpdateSettings(w http.ResponseWriter, r *http.Request) {
@@ -781,7 +735,7 @@ func (h *Handler) UpdateSettings(w http.ResponseWriter, r *http.Request) {
func (h *Handler) DownloadForm(w http.ResponseWriter, r *http.Request) {
presets, err := h.presetSvc.GetAll()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
h.serverError(w, r, "list presets", err)
return
}
Ainternal/handler/health.go
@@ -0,0 +1,84 @@
package handler
import (
"context"
"encoding/json"
"net/http"
"os/exec"
"strings"
"syscall"
"time"
)
// healthProbeTimeout caps the external version lookups so the endpoint always
// answers promptly, even when a tool is wedged.
const healthProbeTimeout = 3 * time.Second
type healthResponse struct {
Status string `json:"status"`
Dependencies map[string]string `json:"dependencies"`
}
// Health reports whether the app can reach its database and external tools. It
// returns 503 when the database is unreachable, so a container healthcheck or
// orchestrator can act on it. A missing yt-dlp/ffmpeg is reported but does not
// fail the check: the UI still works, only downloads are affected.
func (h *Handler) Health(w http.ResponseWriter, r *http.Request) {
// Bound the whole probe: a health check that can hang is worse than useless,
// since an orchestrator reads a stuck request as "still starting".
ctx, cancel := context.WithTimeout(r.Context(), healthProbeTimeout)
defer cancel()
resp := healthResponse{Status: "ok", Dependencies: make(map[string]string, 4)}
// The database is probed first because it alone decides the status code, and
// it must get the budget before a wedged tool can spend it — otherwise a slow
// yt-dlp reports the database as unreachable and the container is restarted
// for the wrong reason.
status := http.StatusOK
if _, err := h.downloadSvc.CountByStatus(ctx); err != nil {
resp.Status = "unhealthy"
resp.Dependencies["database"] = "unreachable"
status = http.StatusServiceUnavailable
} else {
resp.Dependencies["database"] = "ok"
}
resp.Dependencies["yt-dlp"] = toolVersion(ctx, h.cfg.YTDLPPath, "--version")
resp.Dependencies["ffmpeg"] = toolVersion(ctx, h.cfg.FFmpegPath, "-version")
resp.Dependencies["ffprobe"] = toolVersion(ctx, h.cfg.FFprobePath, "-version")
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(resp)
}
// toolVersion returns the first line of the tool's version output. It reports
// "not installed" when the binary is missing and "timed out" when it doesn't
// answer within ctx.
func toolVersion(ctx context.Context, path string, versionArg string) string {
cmd := exec.CommandContext(ctx, path, versionArg)
// Kill the whole process group, not just the tool: yt-dlp and ffmpeg spawn
// helpers that inherit the output pipe, and Output() blocks reading it until
// every holder exits — which would defeat the timeout. WaitDelay is the
// backstop for anything that still survives the signal.
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
cmd.Cancel = func() error {
return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
}
cmd.WaitDelay = time.Second
out, err := cmd.Output()
if err != nil {
if ctx.Err() != nil {
return "timed out"
}
return "not installed"
}
line, _, _ := strings.Cut(strings.TrimSpace(string(out)), "\n")
// ffmpeg prints "ffmpeg version N-x ..."; keep the version token only.
if fields := strings.Fields(line); len(fields) >= 3 && fields[1] == "version" {
return fields[2]
}
return line
}
Ainternal/handler/helpers.go
@@ -0,0 +1,57 @@
package handler
import (
"log"
"net/http"
"strconv"
"github.com/go-chi/chi/v5"
)
// parseID reads the {id} route parameter. It writes a 400 and reports false when
// the value isn't a valid id, so callers can simply return.
func parseID(w http.ResponseWriter, r *http.Request) (int64, bool) {
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
return 0, false
}
return id, true
}
// sortFromRequest returns the sort order for a listing page. An explicit ?sort=
// is remembered in a cookie; without one the last choice is restored, so the
// order survives the auto-refresh that reloads these pages.
func sortFromRequest(w http.ResponseWriter, r *http.Request, cookie, defaultSort string) string {
sortBy := r.URL.Query().Get("sort")
if sortBy == "" {
return getCookie(r, cookie, defaultSort)
}
setCookie(w, cookie, sortBy)
return sortBy
}
// serverError logs the underlying failure and shows the user a generic message.
// Internal error strings can carry filesystem paths and SQL text, so they are
// kept out of the response.
func (h *Handler) serverError(w http.ResponseWriter, r *http.Request, context string, err error) {
log.Printf("%s %s: %s: %v", r.Method, r.URL.Path, context, err)
http.Error(w, "Something went wrong. Please try again.", http.StatusInternalServerError)
}
// redirectWithError flashes a message and redirects, the standard
// Post/Redirect/Get failure path for form submissions. The underlying error is
// logged rather than shown.
func redirectWithError(w http.ResponseWriter, r *http.Request, path, message string, err error) {
if err != nil {
log.Printf("%s %s: %s: %v", r.Method, r.URL.Path, message, err)
}
flashError(w, message)
http.Redirect(w, r, path, http.StatusSeeOther)
}
// redirectWithSuccess flashes a confirmation and redirects.
func redirectWithSuccess(w http.ResponseWriter, r *http.Request, path, message string) {
flashSuccess(w, message)
http.Redirect(w, r, path, http.StatusSeeOther)
}
Minternal/handler/subscription.go
@@ -7,8 +7,6 @@ import (
"strings"
"time"
"github.com/go-chi/chi/v5"
"vidarchive/internal/models"
)
@@ -96,8 +94,7 @@ func (h *Handler) CreateSubscription(w http.ResponseWriter, r *http.Request) {
sub, err := h.subscriptionFromForm(r)
if err != nil {
flashError(w, err.Error())
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithError(w, r, "/subscriptions", err.Error(), nil)
return
}
@@ -106,18 +103,15 @@ func (h *Handler) CreateSubscription(w http.ResponseWriter, r *http.Request) {
}
if err := h.subscriptionSvc.Create(sub); err != nil {
flashError(w, "Couldn't create subscription: "+err.Error())
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithError(w, r, "/subscriptions", "Couldn't create this subscription.", err)
return
}
flashSuccess(w, "Subscription created.")
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithSuccess(w, r, "/subscriptions", "Subscription created.")
}
func (h *Handler) UpdateSubscription(w http.ResponseWriter, r *http.Request) {
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
id, ok := parseID(w, r)
if !ok {
return
}
if err := r.ParseForm(); err != nil {
@@ -133,8 +127,7 @@ func (h *Handler) UpdateSubscription(w http.ResponseWriter, r *http.Request) {
sub, err := h.subscriptionFromForm(r)
if err != nil {
flashError(w, err.Error())
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithError(w, r, "/subscriptions", err.Error(), nil)
return
}
sub.ID = id
@@ -145,18 +138,15 @@ func (h *Handler) UpdateSubscription(w http.ResponseWriter, r *http.Request) {
}
if err := h.subscriptionSvc.Update(sub); err != nil {
flashError(w, "Couldn't update subscription: "+err.Error())
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithError(w, r, "/subscriptions", "Couldn't update this subscription.", err)
return
}
flashSuccess(w, "Subscription updated.")
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithSuccess(w, r, "/subscriptions", "Subscription updated.")
}
func (h *Handler) ToggleSubscription(w http.ResponseWriter, r *http.Request) {
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
id, ok := parseID(w, r)
if !ok {
return
}
sub, err := h.subscriptionSvc.GetByID(id)
@@ -165,22 +155,19 @@ func (h *Handler) ToggleSubscription(w http.ResponseWriter, r *http.Request) {
return
}
if err := h.subscriptionSvc.SetEnabled(id, !sub.Enabled); err != nil {
flashError(w, "Couldn't change subscription: "+err.Error())
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithError(w, r, "/subscriptions", "Couldn't change this subscription.", err)
return
}
message := "Subscription resumed."
if sub.Enabled {
flashSuccess(w, "Subscription paused.")
} else {
flashSuccess(w, "Subscription resumed.")
message = "Subscription paused."
}
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithSuccess(w, r, "/subscriptions", message)
}
func (h *Handler) RunSubscription(w http.ResponseWriter, r *http.Request) {
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
id, ok := parseID(w, r)
if !ok {
return
}
sub, err := h.subscriptionSvc.GetByID(id)
@@ -190,28 +177,23 @@ func (h *Handler) RunSubscription(w http.ResponseWriter, r *http.Request) {
}
download, err := h.downloadSvc.CreateForSubscription(sub)
if err != nil {
flashError(w, "Couldn't start subscription run: "+err.Error())
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithError(w, r, "/subscriptions", "Couldn't start this subscription run.", err)
return
}
h.workerPool.Submit(download)
flashSuccess(w, "Subscription run queued.")
http.Redirect(w, r, "/queue", http.StatusSeeOther)
redirectWithSuccess(w, r, "/queue", "Subscription run queued.")
}
func (h *Handler) DeleteSubscription(w http.ResponseWriter, r *http.Request) {
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
if err != nil {
http.Error(w, "Invalid ID", http.StatusBadRequest)
id, ok := parseID(w, r)
if !ok {
return
}
if err := h.subscriptionSvc.Delete(id); err != nil {
flashError(w, "Couldn't delete subscription: "+err.Error())
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithError(w, r, "/subscriptions", "Couldn't delete this subscription.", err)
return
}
flashSuccess(w, "Subscription deleted.")
http.Redirect(w, r, "/subscriptions", http.StatusSeeOther)
redirectWithSuccess(w, r, "/subscriptions", "Subscription deleted.")
}
// errBadRequest is a small sentinel-style error carrying a user-facing message.
Minternal/repository/download.go
@@ -1,12 +1,18 @@
package repository
import (
"context"
"database/sql"
"fmt"
"vidarchive/internal/models"
)
// downloadColumns is the canonical select list, kept in the order scanDownload
// expects so the two can't drift apart.
const downloadColumns = `id, url, status, logs, error_message, preset_id, format_override,
custom_flags, output_dir, subscription_id, started_at, completed_at, created_at`
type DownloadRepository struct {
db *sql.DB
}
@@ -33,16 +39,12 @@ func (r *DownloadRepository) Create(d *models.Download) error {
}
func (r *DownloadRepository) GetByID(id int64) (*models.Download, error) {
row := r.db.QueryRow(
`SELECT id, url, status, logs, error_message, preset_id, format_override, custom_flags, output_dir, subscription_id, started_at, completed_at, created_at
FROM downloads WHERE id = ?`, id,
)
row := r.db.QueryRow(`SELECT `+downloadColumns+` FROM downloads WHERE id = ?`, id)
return scanDownload(row)
}
func (r *DownloadRepository) GetAll(status, sortBy string) ([]*models.Download, error) {
query := `SELECT id, url, status, logs, error_message, preset_id, format_override, custom_flags, output_dir, subscription_id, started_at, completed_at, created_at
FROM downloads WHERE 1=1`
query := `SELECT ` + downloadColumns + ` FROM downloads WHERE 1=1`
var args []interface{}
if status != "" && status != "all" {
@@ -51,37 +53,66 @@ func (r *DownloadRepository) GetAll(status, sortBy string) ([]*models.Download,
}
switch sortBy {
case "date":
query += ` ORDER BY created_at DESC`
case "status":
query += ` ORDER BY status, created_at DESC`
default:
query += ` ORDER BY created_at DESC`
}
rows, err := r.db.Query(query, args...)
return r.queryDownloads(query, args...)
}
func (r *DownloadRepository) GetQueued(limit int) ([]*models.Download, error) {
return r.queryDownloads(
`SELECT `+downloadColumns+` FROM downloads WHERE status = 'queued' ORDER BY created_at ASC LIMIT ?`,
limit,
)
}
// CountByStatus returns the number of downloads per status, aggregated in SQL so
// callers that only need totals don't load every row's logs. It honours ctx: the
// pool is limited to one connection, so a caller with a deadline (the health
// probe) must be able to give up while a long write holds it.
func (r *DownloadRepository) CountByStatus(ctx context.Context) (map[string]int, error) {
rows, err := r.db.QueryContext(ctx, `SELECT status, COUNT(*) FROM downloads GROUP BY status`)
if err != nil {
return nil, err
}
defer rows.Close()
var downloads []*models.Download
counts := make(map[string]int)
for rows.Next() {
d, err := scanDownload(rows)
if err != nil {
var status string
var n int
if err := rows.Scan(&status, &n); err != nil {
return nil, err
}
downloads = append(downloads, d)
counts[status] = n
}
return downloads, rows.Err()
return counts, rows.Err()
}
func (r *DownloadRepository) GetQueued(limit int) ([]*models.Download, error) {
rows, err := r.db.Query(
`SELECT id, url, status, logs, error_message, preset_id, format_override, custom_flags, output_dir, subscription_id, started_at, completed_at, created_at
FROM downloads WHERE status = 'queued' ORDER BY created_at ASC LIMIT ?`,
limit,
)
// IDsByStatus returns the ids of downloads in the given status.
func (r *DownloadRepository) IDsByStatus(status string) ([]int64, error) {
rows, err := r.db.Query(`SELECT id FROM downloads WHERE status = ?`, status)
if err != nil {
return nil, err
}
defer rows.Close()
var ids []int64
for rows.Next() {
var id int64
if err := rows.Scan(&id); err != nil {
return nil, err
}
ids = append(ids, id)
}
return ids, rows.Err()
}
func (r *DownloadRepository) queryDownloads(query string, args ...interface{}) ([]*models.Download, error) {
rows, err := r.db.Query(query, args...)
if err != nil {
return nil, err
}
@@ -98,11 +129,6 @@ func (r *DownloadRepository) GetQueued(limit int) ([]*models.Download, error) {
return downloads, rows.Err()
}
func (r *DownloadRepository) UpdateStatus(id int64, status string) error {
_, err := r.db.Exec(`UPDATE downloads SET status = ? WHERE id = ?`, status, id)
return err
}
func (r *DownloadRepository) AppendLogs(id int64, logs string) error {
_, err := r.db.Exec(
`UPDATE downloads SET logs = COALESCE(logs, '') || ? WHERE id = ?`,
Minternal/repository/preset.go
@@ -86,10 +86,6 @@ func (r *PresetRepository) Save(p *models.Preset) error {
return tx.Commit()
}
func (r *PresetRepository) Create(p *models.Preset) error {
return r.Save(p)
}
func (r *PresetRepository) GetByID(id int64) (*models.Preset, error) {
row := r.db.QueryRow(
`SELECT id, name, description, is_default, format_mode, format, quality, custom_format, extract_audio, audio_format, embed_subs, sub_langs, embed_thumbnail, embed_metadata, write_info_json, custom_flags, created_at
@@ -127,10 +123,6 @@ func (r *PresetRepository) GetDefault() (*models.Preset, error) {
return scanPreset(row)
}
func (r *PresetRepository) Update(p *models.Preset) error {
return r.Save(p)
}
func (r *PresetRepository) Delete(id int64) error {
_, err := r.db.Exec(`DELETE FROM presets WHERE id = ?`, id)
return err
Minternal/repository/preset_test.go
@@ -33,7 +33,7 @@ func TestPresetRepository(t *testing.T) {
Format: "best",
Quality: "1080",
}
if err := repo.Create(preset); err != nil {
if err := repo.Save(preset); err != nil {
t.Fatalf("failed to create preset: %v", err)
}
if preset.ID == 0 {
@@ -60,7 +60,7 @@ func TestPresetRepository(t *testing.T) {
// Test Update
preset.Name = "Updated Preset"
if err := repo.Update(preset); err != nil {
if err := repo.Save(preset); err != nil {
t.Fatalf("failed to update preset: %v", err)
}
updated, _ := repo.GetByID(preset.ID)
@@ -92,7 +92,7 @@ func TestPresetRepositoryDefault(t *testing.T) {
Name: "TestDefault",
IsDefault: true,
}
if err := repo.Create(preset); err != nil {
if err := repo.Save(preset); err != nil {
t.Fatalf("failed to create preset: %v", err)
}
Minternal/server/handler_test.go
@@ -231,3 +231,72 @@ func TestSubscriptionEditFormRenders(t *testing.T) {
}
}
}
// findCookie returns the value of a cookie set on the response.
func findCookie(w *httptest.ResponseRecorder, name string) (string, bool) {
for _, c := range w.Result().Cookies() {
if c.Name == name {
return c.Value, true
}
}
return "", false
}
// An explicit ?sort= is remembered, and a later request without one restores it.
// The pages auto-refresh, so a sort order that did not persist would be reset
// every few seconds.
func TestSortOrderPersistsInCookie(t *testing.T) {
srv, _, cleanup := setupTestServer(t)
defer cleanup()
router := srv.Router()
cases := []struct {
name, path, cookie, sort, selected string
}{
{"queue", "/queue", "queue_sort", "status", `<option value="status" selected>`},
{"library", "/library", "library_sort", "title", `<option value="title" selected>`},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
w := getWith(router, tc.path+"?sort="+tc.sort, nil)
if w.Code != http.StatusOK {
t.Fatalf("GET %s: %d", tc.path, w.Code)
}
got, ok := findCookie(w, tc.cookie)
if !ok || got != tc.sort {
t.Fatalf("cookie %s = %q (set=%v), want %q", tc.cookie, got, ok, tc.sort)
}
// No ?sort= this time: the remembered order must still apply.
page := getWith(router, tc.path, w.Result().Cookies())
if !strings.Contains(page.Body.String(), tc.selected) {
t.Errorf("remembered sort not applied: %s not found in the rendered page", tc.selected)
}
// Nothing to remember, so the page must not rewrite the cookie.
if _, rewritten := findCookie(page, tc.cookie); rewritten {
t.Errorf("cookie %s was rewritten on a request without ?sort=", tc.cookie)
}
})
}
}
// The cancelled status has its own queue filter, so a stopped download stays
// reachable instead of disappearing from every view.
func TestQueueFilterOffersCancelled(t *testing.T) {
srv, _, cleanup := setupTestServer(t)
defer cleanup()
body := getWith(srv.Router(), "/queue", nil).Body.String()
if !strings.Contains(body, `<option value="cancelled"`) {
t.Error("queue filter has no cancelled option")
}
w := getWith(srv.Router(), "/queue?status=cancelled", nil)
if w.Code != http.StatusOK {
t.Errorf("filtering by cancelled: %d, want 200", w.Code)
}
if !strings.Contains(w.Body.String(), `<option value="cancelled" selected>`) {
t.Error("cancelled filter not marked selected after being applied")
}
}
Ainternal/server/health_test.go
@@ -0,0 +1,137 @@
package server
import (
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"testing"
"time"
)
// fakeTool writes an executable stand-in for one of the external tools and
// points the matching environment variable at it. Must be called before
// setupTestServer, which is what reads the configuration.
func fakeTool(t *testing.T, envVar, name, body string) {
t.Helper()
path := filepath.Join(t.TempDir(), name)
if err := os.WriteFile(path, []byte("#!/bin/sh\n"+body+"\n"), 0755); err != nil {
t.Fatal(err)
}
t.Setenv(envVar, path)
}
func getHealth(t *testing.T, router http.Handler) (*httptest.ResponseRecorder, map[string]any) {
t.Helper()
w := getWith(router, "/healthz", nil)
var body map[string]any
if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
t.Fatalf("decode /healthz body %q: %v", w.Body.String(), err)
}
return w, body
}
func dependencies(t *testing.T, body map[string]any) map[string]any {
t.Helper()
deps, ok := body["dependencies"].(map[string]any)
if !ok {
t.Fatalf("no dependencies object in %v", body)
}
return deps
}
func TestHealthReportsToolVersions(t *testing.T) {
fakeTool(t, "VIDARCHIVE_YTDLP_PATH", "yt-dlp", `echo 2026.01.01`)
fakeTool(t, "VIDARCHIVE_FFMPEG_PATH", "ffmpeg", `echo "ffmpeg version 7.1 Copyright (c) 2000-2024"`)
fakeTool(t, "VIDARCHIVE_FFPROBE_PATH", "ffprobe", `echo "ffprobe version 7.1 Copyright (c) 2000-2024"`)
srv, _, cleanup := setupTestServer(t)
defer cleanup()
w, body := getHealth(t, srv.Router())
if w.Code != http.StatusOK {
t.Errorf("status = %d, want 200", w.Code)
}
if got := body["status"]; got != "ok" {
t.Errorf("status field = %v, want ok", got)
}
deps := dependencies(t, body)
if got := deps["database"]; got != "ok" {
t.Errorf("database = %v, want ok", got)
}
if got := deps["yt-dlp"]; got != "2026.01.01" {
t.Errorf("yt-dlp = %v, want 2026.01.01", got)
}
// "ffmpeg version 7.1 ..." is reduced to the version token.
if got := deps["ffmpeg"]; got != "7.1" {
t.Errorf("ffmpeg = %v, want 7.1", got)
}
}
// A missing external tool is reported but does not fail the check: the UI still
// works, only downloads are affected.
func TestHealthMissingToolIsNotFatal(t *testing.T) {
t.Setenv("VIDARCHIVE_YTDLP_PATH", "/nonexistent/yt-dlp")
srv, _, cleanup := setupTestServer(t)
defer cleanup()
w, body := getHealth(t, srv.Router())
if w.Code != http.StatusOK {
t.Errorf("status = %d, want 200 for a missing tool", w.Code)
}
if got := body["status"]; got != "ok" {
t.Errorf("status field = %v, want ok", got)
}
if got := dependencies(t, body)["yt-dlp"]; got != "not installed" {
t.Errorf("yt-dlp = %v, want \"not installed\"", got)
}
}
// An unreachable database is the one condition that must fail the check, so an
// orchestrator can act on it.
func TestHealthUnreachableDatabaseIs503(t *testing.T) {
srv, _, db, cleanup := setupTestServerDB(t)
defer cleanup()
if err := db.Close(); err != nil {
t.Fatalf("close db: %v", err)
}
w, body := getHealth(t, srv.Router())
if w.Code != http.StatusServiceUnavailable {
t.Errorf("status = %d, want 503", w.Code)
}
if got := body["status"]; got != "unhealthy" {
t.Errorf("status field = %v, want unhealthy", got)
}
if got := dependencies(t, body)["database"]; got != "unreachable" {
t.Errorf("database = %v, want unreachable", got)
}
}
// A wedged tool must not wedge the probe: an orchestrator reads a stuck request
// as "still starting" and never restarts the container.
func TestHealthBoundsAHangingTool(t *testing.T) {
fakeTool(t, "VIDARCHIVE_YTDLP_PATH", "yt-dlp", `sleep 60`)
srv, _, cleanup := setupTestServer(t)
defer cleanup()
start := time.Now()
w, body := getHealth(t, srv.Router())
elapsed := time.Since(start)
// The handler allows 3s; anything near the 60s sleep means it is unbounded.
if elapsed > 15*time.Second {
t.Errorf("probe took %v, want it bounded by the handler timeout", elapsed)
}
if w.Code != http.StatusOK {
t.Errorf("status = %d, want 200", w.Code)
}
if got := dependencies(t, body)["yt-dlp"]; got != "timed out" {
t.Errorf("yt-dlp = %v, want \"timed out\"", got)
}
}
Minternal/server/server.go
@@ -1,9 +1,12 @@
package server
import (
"context"
"errors"
"fmt"
"io/fs"
"net/http"
"time"
"github.com/go-chi/chi/v5"
"github.com/go-chi/chi/v5/middleware"
@@ -13,6 +16,10 @@ import (
"vidarchive/internal/handler"
)
// shutdownTimeout bounds how long in-flight HTTP requests may take to finish
// once shutdown starts. Media streams are the long pole here.
const shutdownTimeout = 20 * time.Second
type Server struct {
router *chi.Mux
handler *handler.Handler
@@ -39,6 +46,8 @@ func (s *Server) setupRoutes() {
staticFS, _ := fs.Sub(vidarchive.StaticFS, "web/static")
s.router.Handle("/static/*", http.StripPrefix("/static/", http.FileServer(http.FS(staticFS))))
s.router.Get("/healthz", s.handler.Health)
s.router.Get("/media/item/*", s.handler.ServeMediaItem)
s.router.Get("/", func(w http.ResponseWriter, r *http.Request) {
@@ -81,9 +90,13 @@ func (s *Server) securityHeaders(next http.Handler) http.Handler {
w.Header().Set("X-XSS-Protection", "1; mode=block")
w.Header().Set("Referrer-Policy", "strict-origin-when-cross-origin")
// The app ships no JavaScript, so the strict policy costs nothing and is
// sent regardless of scheme — plain-HTTP deployments were previously left
// with no CSP at all.
w.Header().Set("Content-Security-Policy", "default-src 'self'; script-src 'none'; style-src 'self' 'unsafe-inline'; media-src 'self' blob:;")
if s.cfg.IsHTTPS() {
w.Header().Set("Strict-Transport-Security", "max-age=31536000; includeSubDomains")
w.Header().Set("Content-Security-Policy", "default-src 'self'; script-src 'none'; style-src 'self' 'unsafe-inline'; media-src 'self' blob:;")
}
next.ServeHTTP(w, r)
@@ -94,12 +107,39 @@ func (s *Server) Router() http.Handler {
return s.router
}
func (s *Server) Start() error {
addr := fmt.Sprintf(":%d", s.cfg.Port)
fmt.Printf("Starting server on %s\n", addr)
// Start serves until ctx is cancelled, then drains in-flight requests within
// shutdownTimeout. Media streaming rules out a WriteTimeout, but a header
// deadline still bounds a client that connects and never completes a request.
func (s *Server) Start(ctx context.Context) error {
srv := &http.Server{
Addr: fmt.Sprintf(":%d", s.cfg.Port),
Handler: s.router,
ReadHeaderTimeout: 15 * time.Second,
IdleTimeout: 120 * time.Second,
}
fmt.Printf("Starting server on %s\n", srv.Addr)
if s.cfg.BaseURL != "" {
fmt.Printf("Base URL: %s\n", s.cfg.BaseURL)
fmt.Printf("HTTPS mode: %v\n", s.cfg.IsHTTPS())
}
return http.ListenAndServe(addr, s.router)
errCh := make(chan error, 1)
go func() {
err := srv.ListenAndServe()
// A shutdown is the expected way this returns, not a startup failure.
if errors.Is(err, http.ErrServerClosed) {
err = nil
}
errCh <- err
}()
select {
case err := <-errCh:
return err
case <-ctx.Done():
shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
defer cancel()
return srv.Shutdown(shutdownCtx)
}
}
Minternal/server/server_test.go
@@ -1,6 +1,7 @@
package server
import (
"database/sql"
"html"
"io"
"net/http"
@@ -22,6 +23,13 @@ import (
)
func setupTestServer(t *testing.T) (*Server, *config.Config, func()) {
srv, cfg, _, cleanup := setupTestServerDB(t)
return srv, cfg, cleanup
}
// setupTestServerDB is setupTestServer with the database handle exposed, for
// tests that need to break the database on purpose.
func setupTestServerDB(t *testing.T) (*Server, *config.Config, *sql.DB, func()) {
t.Helper()
dataDir := t.TempDir()
t.Setenv("VIDARCHIVE_DATA_DIR", dataDir)
@@ -61,7 +69,7 @@ func setupTestServer(t *testing.T) (*Server, *config.Config, func()) {
workerPool.Stop()
db.Close()
}
return srv, cfg, cleanup
return srv, cfg, db, cleanup
}
func createItem(t *testing.T, libraryDir, relPath, name string, files map[string]string) {
Minternal/service/download.go
@@ -3,13 +3,18 @@ package service
import (
"bufio"
"bytes"
"cmp"
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"os"
"os/exec"
"path/filepath"
"slices"
"sort"
"strconv"
"strings"
@@ -33,8 +38,8 @@ type DownloadService struct {
subscriptionSvc *SubscriptionService
cfg *config.Config
cache *ProgressCache
processMu sync.Mutex
processes map[int64]*os.Process
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 {
@@ -46,7 +51,7 @@ func NewDownloadService(repo *repository.DownloadRepository, librarySvc *Library
subscriptionSvc: subscriptionSvc,
cfg: cfg,
cache: NewProgressCache(),
processes: make(map[int64]*os.Process),
active: make(map[int64]context.CancelFunc),
}
}
@@ -125,32 +130,101 @@ func (s *DownloadService) HasActiveForSubscription(subID int64) (bool, error) {
}
func (s *DownloadService) Delete(id int64) error {
s.killProcess(id)
s.cancelDownload(id)
s.cache.Delete(id)
return s.repo.Delete(id)
}
func (s *DownloadService) killProcess(id int64) {
s.processMu.Lock()
proc, ok := s.processes[id]
delete(s.processes, id)
s.processMu.Unlock()
// 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()
if !ok || proc == nil {
return
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()
_ = syscall.Kill(-proc.Pid, syscall.SIGKILL)
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) CountByStatus(ctx context.Context) (map[string]int, error) {
return s.repo.CountByStatus(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)}
}
func (s *DownloadService) ListFormats(url string) ([]*models.FormatInfo, error) {
// Use machine-readable JSON (-J) rather than scraping the human "-F" table,
// whose columns/separators shift between yt-dlp versions. stderr is captured
@@ -169,8 +243,12 @@ func (s *DownloadService) ListFormats(url string) ([]*models.FormatInfo, error)
// 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.
func (s *DownloadService) ExecuteDownload(d *models.Download) (bool, error) {
// 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) {
// Atomically claim the download. If it's no longer queued, another worker
// already took it — bail rather than download it twice.
claimed, err := s.repo.MarkStarted(d.ID)
@@ -181,6 +259,12 @@ func (s *DownloadService) ExecuteDownload(d *models.Download) (bool, error) {
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{LastUpdate: time.Now()})
defer s.cache.Delete(d.ID)
@@ -211,16 +295,14 @@ func (s *DownloadService) ExecuteDownload(d *models.Download) (bool, error) {
// 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
if err := checkReservedFlags(d.CustomFlags, isSubscription); err != nil {
s.finalizeError(d.ID, err)
return false, err
}
if err := checkReservedFlags(preset.CustomFlags, isSubscription); err != nil {
s.finalizeError(d.ID, err)
return false, err
for _, flags := range []string{d.CustomFlags, preset.CustomFlags} {
if err := checkReservedFlags(flags, isSubscription); err != nil {
s.finalizeError(d.ID, err)
return false, err
}
}
tempDownloadDir := filepath.Join(s.cfg.TempDir, fmt.Sprintf("%d", d.ID))
tempDownloadDir := s.tempDirFor(d.ID)
if err := os.MkdirAll(tempDownloadDir, 0755); err != nil {
return false, fmt.Errorf("create temp download dir: %w", err)
}
@@ -261,9 +343,17 @@ func (s *DownloadService) ExecuteDownload(d *models.Download) (bool, error) {
args = append(args, "-o", "item-%(autonumber)05d/%(title)s.%(ext)s")
args = append(args, d.URL)
if err := s.runYTDLP(d, args); err != nil {
s.finalizeError(d.ID, err)
return false, err
runErr := s.runYTDLP(ctx, d, args)
// Cancellation is checked before the run error, and both are checked before
// 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.ID)
}
if runErr != nil {
s.finalizeError(d.ID, runErr)
return false, runErr
}
mode := ""
@@ -274,32 +364,38 @@ func (s *DownloadService) ExecuteDownload(d *models.Download) (bool, error) {
// 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.
if err := s.refreshAndAddNew(d, preset, tempDownloadDir, ytdlpFlags); err != nil {
s.finalizeError(d.ID, err)
return false, err
}
postErr = s.refreshAndAddNew(ctx, d, preset, tempDownloadDir, ytdlpFlags)
} else {
imported, err := s.importDownloadedItems(d, tempDownloadDir, mode, ytdlpFlags)
if err != nil {
s.finalizeError(d.ID, err)
return false, err
}
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.
if sub == nil && imported == 0 {
err := fmt.Errorf("yt-dlp finished but no media files were downloaded")
s.finalizeError(d.ID, err)
return false, err
case sub == nil && imported == 0:
postErr = fmt.Errorf("yt-dlp finished but no media files were downloaded")
}
}
if sub != nil && sub.PruneRemoved {
s.pruneSubscription(d, sub)
if postErr == nil && sub != nil && sub.PruneRemoved {
s.pruneSubscription(ctx, d, sub)
}
// Check cancellation before postErr, not only when postErr is non-nil: a run
// stopped just after yt-dlp exited 0 leaves postErr nil, and without this the
// download falls through and is recorded as "completed".
if ctx.Err() != nil {
return false, s.finalizeCancelled(parent, d.ID)
}
if postErr != nil {
s.finalizeError(d.ID, postErr)
return false, postErr
}
if err := s.repo.MarkCompleted(d.ID, "completed"); err != nil {
@@ -311,15 +407,20 @@ func (s *DownloadService) ExecuteDownload(d *models.Download) (bool, error) {
// runYTDLP executes yt-dlp with args, streaming combined output into the live
// progress cache and periodically flushing it to the download's persisted log.
// It registers the process so Delete can kill it, and returns the exit error.
func (s *DownloadService) runYTDLP(d *models.Download, args []string) error {
// 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.Command(s.cfg.YTDLPPath, fullArgs...)
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 {
@@ -331,15 +432,6 @@ func (s *DownloadService) runYTDLP(d *models.Download, args []string) error {
return err
}
s.processMu.Lock()
s.processes[d.ID] = cmd.Process
s.processMu.Unlock()
defer func() {
s.processMu.Lock()
delete(s.processes, d.ID)
s.processMu.Unlock()
}()
ticker := time.NewTicker(10 * time.Second)
defer ticker.Stop()
done := make(chan struct{})
@@ -347,9 +439,7 @@ func (s *DownloadService) runYTDLP(d *models.Download, args []string) error {
for {
select {
case <-ticker.C:
if logs := s.cache.FlushLogs(d.ID); logs != "" {
s.repo.AppendLogs(d.ID, logs)
}
s.flushLogs(d.ID)
case <-done:
return
}
@@ -368,9 +458,7 @@ func (s *DownloadService) runYTDLP(d *models.Download, args []string) error {
}
close(done)
if logs := s.cache.FlushLogs(d.ID); logs != "" {
s.repo.AppendLogs(d.ID, logs)
}
s.flushLogs(d.ID)
return cmd.Wait()
}
@@ -384,7 +472,12 @@ func (s *DownloadService) appendCookies(args []string) ([]string, func()) {
if err != nil || strings.TrimSpace(cookies) == "" {
return args, cleanup
}
tmpFile, err := os.CreateTemp("", "cookies-*.txt")
// Written into the app's own temp dir rather than the system one, so the
// cookies file lands on the same volume the rest of the run uses.
if err := os.MkdirAll(s.cfg.TempDir, 0755); err != nil {
return args, cleanup
}
tmpFile, err := os.CreateTemp(s.cfg.TempDir, "cookies-*.txt")
if err != nil {
return args, cleanup
}
@@ -404,39 +497,73 @@ func (s *DownloadService) appendCookies(args []string) ([]string, func()) {
}
func (s *DownloadService) finalizeError(id int64, err error) {
s.flushLogs(id)
if markErr := s.repo.MarkError(id, err.Error()); markErr != nil {
log.Printf("download %d: failed to record error: %v", id, markErr)
}
}
// 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, id int64) error {
s.flushLogs(id)
if parent.Err() != nil {
return ErrCancelled
}
if err := s.repo.MarkCompleted(id, "cancelled"); err != nil {
log.Printf("download %d: failed to record cancellation: %v", id, err)
}
return ErrCancelled
}
// flushLogs persists whatever output has accumulated for a download.
func (s *DownloadService) flushLogs(id int64) {
logs := s.cache.FlushLogs(id)
if logs != "" {
s.repo.AppendLogs(id, logs)
if logs == "" {
return
}
if err := s.repo.AppendLogs(id, logs); err != nil {
log.Printf("download %d: failed to persist logs: %v", id, err)
}
s.repo.MarkError(id, err.Error())
}
// resolveBaseLibraryDir returns the absolute library directory a download writes
// into, applying the optional per-download OutputDir while rejecting any path
// that escapes the library root.
func (s *DownloadService) resolveBaseLibraryDir(d *models.Download) (string, error) {
baseLibraryDir := s.cfg.LibraryDir
if d.OutputDir.Valid && d.OutputDir.String != "" {
cleanDir := filepath.Clean(d.OutputDir.String)
fullPath := filepath.Join(baseLibraryDir, cleanDir)
resolvedPath, err := filepath.Abs(fullPath)
if err != nil {
return "", fmt.Errorf("invalid output directory: %w", err)
}
resolvedLibraryDir, _ := filepath.Abs(baseLibraryDir)
if !strings.HasPrefix(resolvedPath, resolvedLibraryDir+string(filepath.Separator)) && resolvedPath != resolvedLibraryDir {
return "", fmt.Errorf("invalid output directory: path traversal attempt detected")
}
baseLibraryDir = fullPath
if !d.OutputDir.Valid || d.OutputDir.String == "" {
return s.cfg.LibraryDir, nil
}
return baseLibraryDir, nil
// Reuse the library service's guard so both entry points enforce the boundary
// the same way — it resolves symlinks, which a plain prefix check does not.
dir, err := s.librarySvc.ResolveWithinLibrary(d.OutputDir.String)
if err != nil {
return "", fmt.Errorf("invalid output directory: %w", err)
}
return dir, nil
}
// importDownloadedItems moves each downloaded item from the temp dir into the
// library and returns the number of items successfully imported. Per-item
// failures are logged and skipped (a playlist with a few bad entries still
// imports the rest); a non-nil error means the import couldn't even start.
func (s *DownloadService) importDownloadedItems(d *models.Download, tempDownloadDir, mode, ytdlpFlags string) (int, error) {
//
// A cancel stops the import between items and returns ctx.Err() with the count
// imported so far. Importing a long playlist takes real time (a move plus an
// ffprobe per file), so a deleted download must not keep filling the library.
func (s *DownloadService) importDownloadedItems(ctx context.Context, d *models.Download, tempDownloadDir, mode, ytdlpFlags string) (int, error) {
entries, err := os.ReadDir(tempDownloadDir)
if err != nil {
return 0, err
@@ -464,7 +591,15 @@ func (s *DownloadService) importDownloadedItems(d *models.Download, tempDownload
imported := 0
for _, itemDir := range itemDirs {
if err := s.importItemDir(d.URL, itemDir, baseLibraryDir, mode, ytdlpFlags); err != nil {
if err := ctx.Err(); err != nil {
return imported, err
}
if err := s.importItemDir(ctx, d.URL, itemDir, baseLibraryDir, mode, ytdlpFlags); err != nil {
// A cancelled item isn't a bad item: stop instead of logging a warning
// for it and every one that follows.
if ctx.Err() != nil {
return imported, ctx.Err()
}
log.Printf("warning: failed to import item %s: %v", itemDir, err)
continue
}
@@ -476,7 +611,7 @@ func (s *DownloadService) importDownloadedItems(d *models.Download, tempDownload
return imported, nil
}
func (s *DownloadService) importItemDir(url, itemDir, baseLibraryDir, mode, ytdlpFlags string) error {
func (s *DownloadService) importItemDir(ctx context.Context, url, itemDir, baseLibraryDir, mode, ytdlpFlags string) error {
entries, err := os.ReadDir(itemDir)
if err != nil {
return err
@@ -494,7 +629,7 @@ func (s *DownloadService) importItemDir(url, itemDir, baseLibraryDir, mode, ytdl
path := filepath.Join(itemDir, name)
ext := strings.ToLower(filepath.Ext(name))
if name == "info.json" || strings.HasSuffix(name, ".info.json") {
if isInfoJSON(name) {
infoJSONPath = path
continue
}
@@ -513,8 +648,17 @@ func (s *DownloadService) importItemDir(url, itemDir, baseLibraryDir, mode, ytdl
return fmt.Errorf("no media files found in %s", itemDir)
}
name := s.deriveItemName(itemDir, infoJSONPath, mediaFiles)
videoID := readInfoID(infoJSONPath)
info := readInfoJSON(infoJSONPath)
name := s.deriveItemName(itemDir, info, mediaFiles)
videoID := info.ID
// Last point at which nothing has been written to the library yet: give up
// here on a cancel rather than part-way through, which would leave a folder
// with some of its files and no marker — or, in overwrite mode, delete the
// existing item and not replace it.
if err := ctx.Err(); err != nil {
return err
}
// Overwrite mode: replace the existing copy of this video in place rather than
// creating a duplicate folder. Removing the old dir lets uniqueDir reuse its
@@ -534,13 +678,13 @@ func (s *DownloadService) importItemDir(url, itemDir, baseLibraryDir, mode, ytdl
}
if infoJSONPath != "" {
if err := os.Rename(infoJSONPath, filepath.Join(targetDir, "info.json")); err != nil {
if err := moveFile(infoJSONPath, filepath.Join(targetDir, "info.json")); err != nil {
return err
}
}
for _, entry := range mediaFiles {
if err := os.Rename(filepath.Join(itemDir, entry.Name()), filepath.Join(targetDir, entry.Name())); err != nil {
if err := moveFile(filepath.Join(itemDir, entry.Name()), filepath.Join(targetDir, entry.Name())); err != nil {
return err
}
}
@@ -550,7 +694,7 @@ func (s *DownloadService) importItemDir(url, itemDir, baseLibraryDir, mode, ytdl
// serving pages. Files we can't probe simply get no duration.
fileDurations := make(map[string]int)
for _, entry := range mediaFiles {
if d, ok := probeDuration(s.cfg.FFprobePath, filepath.Join(targetDir, entry.Name())); ok {
if d, ok := probeDuration(ctx, s.cfg.FFprobePath, filepath.Join(targetDir, entry.Name())); ok {
fileDurations[entry.Name()] = d
}
}
@@ -561,7 +705,7 @@ func (s *DownloadService) importItemDir(url, itemDir, baseLibraryDir, mode, ytdl
return err
}
for _, sf := range subtitleFiles {
if err := os.Rename(sf, filepath.Join(subtitlesDir, filepath.Base(sf))); err != nil {
if err := moveFile(sf, filepath.Join(subtitlesDir, filepath.Base(sf))); err != nil {
return err
}
}
@@ -588,39 +732,47 @@ func (s *DownloadService) importItemDir(url, itemDir, baseLibraryDir, mode, ytdl
return nil
}
// readInfoID extracts the stable item identity (yt-dlp's video id) from an
// info.json. The id alone is sufficient to match items within a single
// subscription's owned directory, so the extractor is not used. Returns an
// empty string when the file is absent or unreadable.
func readInfoID(infoJSONPath string) string {
// infoJSON is the subset of yt-dlp's info.json VidArchive reads. ID is the
// stable item identity; within a single subscription's own directory it is
// enough to match items, so the extractor is not needed.
type infoJSON struct {
ID string `json:"id"`
Title string `json:"title"`
Description string `json:"description"`
WebpageURL string `json:"webpage_url"`
}
// readInfoJSON parses an info.json. A missing, unreadable or malformed file
// yields a zero-value struct: every caller treats absent fields as "unknown"
// and falls back, so there is nothing to distinguish.
func readInfoJSON(infoJSONPath string) infoJSON {
var info infoJSON
if infoJSONPath == "" {
return ""
return info
}
data, err := os.ReadFile(infoJSONPath)
if err != nil {
return ""
}
var info struct {
ID string `json:"id"`
return info
}
if err := json.Unmarshal(data, &info); err != nil {
return ""
log.Printf("ignoring malformed %s: %v", infoJSONPath, err)
return infoJSON{}
}
return info.ID
return info
}
// refreshAndAddNew handles a metadata-mode run. The main pass used
// --skip-download, so tempDownloadDir holds only info.json files. Existing
// library items have their markers refreshed in place; entries with no existing
// match are genuinely new and are downloaded as full items in a second pass.
func (s *DownloadService) refreshAndAddNew(d *models.Download, preset *models.Preset, tempDownloadDir, ytdlpFlags string) error {
func (s *DownloadService) refreshAndAddNew(ctx context.Context, d *models.Download, preset *models.Preset, tempDownloadDir, ytdlpFlags string) error {
// The library dir is not created here: a refresh that matches everything
// writes nothing, and importDownloadedItems creates it when a second pass
// actually has an item to add.
baseLibraryDir, err := s.resolveBaseLibraryDir(d)
if err != nil {
return err
}
if err := os.MkdirAll(baseLibraryDir, 0755); err != nil {
return err
}
entries, err := os.ReadDir(tempDownloadDir)
if err != nil {
@@ -629,6 +781,9 @@ func (s *DownloadService) refreshAndAddNew(d *models.Download, preset *models.Pr
var newURLs []string
for _, entry := range entries {
if err := ctx.Err(); err != nil {
return err
}
if !entry.IsDir() || !strings.HasPrefix(entry.Name(), "item-") {
continue
}
@@ -637,32 +792,32 @@ func (s *DownloadService) refreshAndAddNew(d *models.Download, preset *models.Pr
if infoJSONPath == "" {
continue
}
videoID := readInfoID(infoJSONPath)
if videoID == "" {
info := readInfoJSON(infoJSONPath)
if info.ID == "" {
continue
}
if existing, ok := s.librarySvc.FindByVideoID(baseLibraryDir, videoID); ok {
if err := s.applyMetadata(existing, infoJSONPath, videoID); err != nil {
if existing, ok := s.librarySvc.FindByVideoID(baseLibraryDir, info.ID); ok {
if err := s.applyMetadata(existing, info); err != nil {
log.Printf("warning: failed to refresh metadata for %s: %v", itemDir, err)
}
continue
}
if u := readWebpageURL(infoJSONPath); u != "" {
newURLs = append(newURLs, u)
if info.WebpageURL != "" {
newURLs = append(newURLs, info.WebpageURL)
}
}
if len(newURLs) == 0 {
return nil
}
return s.downloadFresh(d, preset, newURLs, ytdlpFlags)
return s.downloadFresh(ctx, d, preset, newURLs, ytdlpFlags)
}
// downloadFresh fetches the given item URLs as full downloads (media + info.json)
// and imports them into the download's library directory. Metadata mode uses this
// to add entries that don't exist in the library yet.
func (s *DownloadService) downloadFresh(d *models.Download, preset *models.Preset, urls []string, ytdlpFlags string) error {
tempDir := filepath.Join(s.cfg.TempDir, fmt.Sprintf("%d-new", d.ID))
func (s *DownloadService) downloadFresh(ctx context.Context, d *models.Download, preset *models.Preset, urls []string, ytdlpFlags string) error {
tempDir := s.tempNewDirFor(d.ID)
if err := os.MkdirAll(tempDir, 0755); err != nil {
return err
}
@@ -676,9 +831,13 @@ func (s *DownloadService) downloadFresh(d *models.Download, preset *models.Prese
args = append(args, "-o", "item-%(autonumber)05d/%(title)s.%(ext)s")
args = append(args, urls...)
runErr := s.runYTDLP(d, args)
runErr := s.runYTDLP(ctx, d, args)
// A cancelled second pass has nothing worth importing.
if ctx.Err() != nil {
return runErr
}
// Import whatever succeeded even if some entries errored.
if _, err := s.importDownloadedItems(d, tempDir, "", ytdlpFlags); err != nil {
if _, err := s.importDownloadedItems(ctx, d, tempDir, "", ytdlpFlags); err != nil {
log.Printf("warning: failed to import new metadata-mode items: %v", err)
}
return runErr
@@ -686,20 +845,7 @@ func (s *DownloadService) downloadFresh(d *models.Download, preset *models.Prese
// applyMetadata rewrites an existing item's marker (name/description/identity)
// from a fresh info.json without touching its media.
func (s *DownloadService) applyMetadata(existing, infoJSONPath, videoID string) error {
data, err := os.ReadFile(infoJSONPath)
if err != nil {
return err
}
var info struct {
Title string `json:"title"`
Description string `json:"description"`
WebpageURL string `json:"webpage_url"`
}
if err := json.Unmarshal(data, &info); err != nil {
log.Printf("applyMetadata: ignoring malformed %s: %v", infoJSONPath, err)
}
func (s *DownloadService) applyMetadata(existing string, info infoJSON) error {
meta, err := s.librarySvc.readOrCreateMetadata(existing)
if err != nil {
return fmt.Errorf("read metadata for %s: %w", existing, err)
@@ -713,7 +859,7 @@ func (s *DownloadService) applyMetadata(existing, infoJSONPath, videoID string)
if meta.SourceURL == "" && info.WebpageURL != "" {
meta.SourceURL = info.WebpageURL
}
meta.VideoID = videoID
meta.VideoID = info.ID
if err := s.librarySvc.writeMetadata(existing, meta); err != nil {
return err
@@ -724,19 +870,11 @@ func (s *DownloadService) applyMetadata(existing, infoJSONPath, videoID string)
return nil
}
// readWebpageURL returns the canonical entry URL from an info.json, or "".
func readWebpageURL(infoJSONPath string) string {
data, err := os.ReadFile(infoJSONPath)
if err != nil {
return ""
}
var info struct {
WebpageURL string `json:"webpage_url"`
}
if err := json.Unmarshal(data, &info); err != nil {
return ""
}
return info.WebpageURL
// isInfoJSON reports whether a file name is yt-dlp's metadata sidecar. yt-dlp
// writes "<title>.info.json" next to the media, but a bare "info.json" is what
// an already-imported item holds.
func isInfoJSON(name string) bool {
return name == "info.json" || strings.HasSuffix(name, ".info.json")
}
// findInfoJSON returns the path to an info.json directly inside itemDir, or "".
@@ -750,7 +888,7 @@ func findInfoJSON(itemDir string) string {
continue
}
name := entry.Name()
if name == "info.json" || strings.HasSuffix(name, ".info.json") {
if isInfoJSON(name) {
return filepath.Join(itemDir, name)
}
}
@@ -761,14 +899,14 @@ func findInfoJSON(itemDir string) string {
// directory that are no longer present upstream. It enumerates the current id
// set with a cheap flat-playlist listing; it never prunes when that enumeration
// fails or returns nothing, so a dead URL or network error can't wipe the dir.
func (s *DownloadService) pruneSubscription(d *models.Download, sub *models.Subscription) {
func (s *DownloadService) pruneSubscription(ctx context.Context, d *models.Download, sub *models.Subscription) {
baseLibraryDir, err := s.resolveBaseLibraryDir(d)
if err != nil {
log.Printf("subscription %d prune skipped: %v", sub.ID, err)
return
}
keep, err := s.enumeratePlaylistIDs(sub.URL)
keep, err := s.enumeratePlaylistIDs(ctx, sub.URL)
if err != nil {
log.Printf("subscription %d prune skipped: enumeration failed: %v", sub.ID, err)
return
@@ -792,14 +930,14 @@ func (s *DownloadService) pruneSubscription(d *models.Download, sub *models.Subs
// downloading, using yt-dlp --flat-playlist. Cookies are applied so private
// playlists enumerate correctly. Ids alone are sufficient to match items within
// a subscription's own directory (see FindByVideoID / PruneToIDSet).
func (s *DownloadService) enumeratePlaylistIDs(url string) (map[string]bool, error) {
func (s *DownloadService) enumeratePlaylistIDs(ctx context.Context, url string) (map[string]bool, error) {
args := []string{"--flat-playlist", "--no-warnings", "--print", "%(id)s"}
args, cleanup := s.appendCookies(args)
defer cleanup()
args = append(args, url)
out, err := exec.Command(s.cfg.YTDLPPath, args...).Output()
out, err := exec.CommandContext(ctx, s.cfg.YTDLPPath, args...).Output()
if err != nil {
return nil, err
}
@@ -816,29 +954,29 @@ func (s *DownloadService) enumeratePlaylistIDs(url string) (map[string]bool, err
return keep, nil
}
func (s *DownloadService) deriveItemName(itemDir, infoJSONPath string, mediaFiles []os.DirEntry) string {
if infoJSONPath != "" {
data, err := os.ReadFile(infoJSONPath)
if err == nil {
var info struct {
Title string `json:"title"`
}
if err := json.Unmarshal(data, &info); err == nil && info.Title != "" {
return sanitizeDirName(info.Title)
}
}
// deriveItemName names the imported item after its title, falling back to the
// largest media file's base name when there is no usable info.json.
func (s *DownloadService) deriveItemName(itemDir string, info infoJSON, mediaFiles []os.DirEntry) string {
if info.Title != "" {
return sanitizeDirName(info.Title)
}
sort.Slice(mediaFiles, func(i, j int) bool {
ii, _ := os.Stat(filepath.Join(itemDir, mediaFiles[i].Name()))
jj, _ := os.Stat(filepath.Join(itemDir, mediaFiles[j].Name()))
if ii == nil || jj == nil {
return false
// Stat once per file up front: doing it inside the comparator re-stats the
// same files O(n log n) times.
sizes := make(map[string]int64, len(mediaFiles))
for _, f := range mediaFiles {
if st, err := os.Stat(filepath.Join(itemDir, f.Name())); err == nil {
sizes[f.Name()] = st.Size()
}
return ii.Size() > jj.Size()
}
// Sort a copy: the caller iterates mediaFiles again to move and probe the
// files, and a naming helper must not reorder its caller's data.
byName := slices.Clone(mediaFiles)
slices.SortFunc(byName, func(a, b os.DirEntry) int {
return cmp.Compare(sizes[b.Name()], sizes[a.Name()])
})
base := strings.TrimSuffix(mediaFiles[0].Name(), filepath.Ext(mediaFiles[0].Name()))
base := strings.TrimSuffix(byName[0].Name(), filepath.Ext(byName[0].Name()))
return sanitizeDirName(base)
}
@@ -855,6 +993,47 @@ func (s *DownloadService) uniqueDir(base, name string) string {
}
}
// moveFile moves src to dst, falling back to copy-and-delete when the two are on
// different filesystems. The temp and library directories are independently
// configurable, so they can legitimately live on separate mounts — where a plain
// rename fails with EXDEV.
func moveFile(src, dst string) error {
if err := os.Rename(src, dst); err == nil {
return nil
} else if !errors.Is(err, syscall.EXDEV) {
return err
}
in, err := os.Open(src)
if err != nil {
return err
}
defer in.Close()
info, err := in.Stat()
if err != nil {
return err
}
out, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, info.Mode())
if err != nil {
return err
}
if _, err := io.Copy(out, in); err != nil {
out.Close()
os.Remove(dst)
return err
}
// Close explicitly: a deferred close would hide a flush error on the copy.
if err := out.Close(); err != nil {
os.Remove(dst)
return err
}
return os.Remove(src)
}
func sanitizeDirName(name string) string {
name = strings.TrimSpace(name)
replacer := strings.NewReplacer(
@@ -878,8 +1057,8 @@ func sanitizeDirName(name string) string {
// probeDuration returns the duration of a media file in whole seconds. The bool
// is false when ffprobe is unavailable or the file has no usable duration.
func probeDuration(ffprobePath, path string) (int, bool) {
out, err := exec.Command(ffprobePath, "-v", "error",
func probeDuration(ctx context.Context, ffprobePath, path string) (int, bool) {
out, err := exec.CommandContext(ctx, ffprobePath, "-v", "error",
"-show_entries", "format=duration",
"-of", "default=nw=1:nk=1", path).Output()
if err != nil {
@@ -1003,6 +1182,18 @@ var reservedFlags = map[string]string{
"--cookies": "cookies (set these in Settings instead)",
"--no-cookies": "cookies (set these in Settings instead)",
"--newline": "progress output formatting (VidArchive sets this to stream logs)",
// These hand yt-dlp an arbitrary command or binary to run. VidArchive passes
// custom flags through verbatim, so allowing them would turn the preset form
// into remote command execution.
"--exec": "running external commands (not permitted)",
"--exec-before-download": "running external commands (not permitted)",
"--postprocessor-args": "post-processor arguments (not permitted)",
"--ppa": "post-processor arguments (not permitted)",
"--downloader": "selecting an external downloader (not permitted)",
"--external-downloader": "selecting an external downloader (not permitted)",
"--downloader-args": "external downloader arguments (not permitted)",
"--external-downloader-args": "external downloader arguments (not permitted)",
}
// reservedSubscriptionFlags are additionally reserved for subscription runs,
@@ -1020,18 +1211,16 @@ var reservedSubscriptionFlags = map[string]string{
// controls, naming the offender. It matches both "--flag" and "--flag=value".
func checkReservedFlags(customFlags string, isSubscription bool) error {
for _, tok := range strings.Fields(customFlags) {
name := tok
if i := strings.IndexByte(name, '='); i >= 0 {
name = name[:i]
// Both "--flag value" and "--flag=value" name the same option.
name, _, _ := strings.Cut(tok, "=")
desc, ok := reservedFlags[name]
if !ok && isSubscription {
desc, ok = reservedSubscriptionFlags[name]
}
if desc, ok := reservedFlags[name]; ok {
if ok {
return fmt.Errorf("custom flag %q conflicts with VidArchive's handling of %s; remove it and try again", tok, desc)
}
if isSubscription {
if desc, ok := reservedSubscriptionFlags[name]; ok {
return fmt.Errorf("custom flag %q conflicts with VidArchive's handling of %s; remove it and try again", tok, desc)
}
}
}
return nil
}
Minternal/service/download_test.go
@@ -1,11 +1,16 @@
package service
import (
"context"
"database/sql"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"syscall"
"testing"
"time"
"vidarchive/internal/config"
"vidarchive/internal/models"
@@ -64,7 +69,7 @@ func TestDeriveItemName(t *testing.T) {
t.Fatal(err)
}
media := []os.DirEntry{dirEntry(t, itemDir, bigName)}
if got := svc.deriveItemName(itemDir, infoPath, media); got != "Cool- Video" {
if got := svc.deriveItemName(itemDir, readInfoJSON(infoPath), media); got != "Cool- Video" {
t.Errorf("deriveItemName with info.json = %q, want %q", got, "Cool- Video")
}
@@ -74,7 +79,7 @@ func TestDeriveItemName(t *testing.T) {
t.Fatal(err)
}
media = []os.DirEntry{dirEntry(t, itemDir, small), dirEntry(t, itemDir, bigName)}
if got := svc.deriveItemName(itemDir, "", media); got != "big" {
if got := svc.deriveItemName(itemDir, infoJSON{}, media); got != "big" {
t.Errorf("deriveItemName fallback = %q, want %q (largest file stem)", got, "big")
}
}
@@ -162,7 +167,7 @@ func TestImportItemDir(t *testing.T) {
t.Fatal(err)
}
if err := svc.importItemDir("https://example.com/v", src, libDir, "", "-f bestvideo+bestaudio"); err != nil {
if err := svc.importItemDir(context.Background(), "https://example.com/v", src, libDir, "", "-f bestvideo+bestaudio"); err != nil {
t.Fatalf("importItemDir: %v", err)
}
@@ -218,6 +223,9 @@ func TestCheckReservedFlags(t *testing.T) {
{"write-info-json sub", "--write-info-json", true, true},
{"no-write-info-json sub", "--no-write-info-json", true, true},
{"download-archive sub", "--download-archive a.txt", true, true},
// The "--flag=value" spelling names the same option as "--flag value",
// for the subscription-only table as well as the base one.
{"download-archive equals form sub", "--download-archive=a.txt", true, true},
// Base reserved flags still apply to subscriptions.
{"output sub", "-o x", true, true},
}
@@ -233,7 +241,10 @@ func TestCheckReservedFlags(t *testing.T) {
func TestImportDownloadedItemsRejectsOutputTraversal(t *testing.T) {
libDir := t.TempDir()
svc := &DownloadService{cfg: &config.Config{LibraryDir: libDir}}
svc := &DownloadService{
cfg: &config.Config{LibraryDir: libDir},
librarySvc: NewLibraryService(libDir, "ffmpeg", "ffprobe"),
}
// A temp download dir with one item subdir.
tempDir := t.TempDir()
@@ -246,7 +257,269 @@ func TestImportDownloadedItemsRejectsOutputTraversal(t *testing.T) {
URL: "u",
OutputDir: sql.NullString{String: "../escape", Valid: true},
}
if _, err := svc.importDownloadedItems(d, tempDir, "", ""); err == nil {
if _, err := svc.importDownloadedItems(context.Background(), d, tempDir, "", ""); err == nil {
t.Error("expected path-traversal output dir to be rejected")
}
}
// A download deleted while its items are being imported must stop, not keep
// filling the library with items whose queue row no longer exists.
func TestImportDownloadedItemsStopsOnCancel(t *testing.T) {
requireFFmpeg(t)
libDir := t.TempDir()
svc := &DownloadService{
cfg: &config.Config{LibraryDir: libDir, FFprobePath: "ffprobe"},
librarySvc: NewLibraryService(libDir, "ffmpeg", "ffprobe"),
}
tempDir := t.TempDir()
for _, name := range []string{"item-00001", "item-00002"} {
itemDir := filepath.Join(tempDir, name)
if err := os.MkdirAll(itemDir, 0755); err != nil {
t.Fatal(err)
}
makeTestVideo(t, filepath.Join(itemDir, "raw.mp4"))
}
ctx, cancel := context.WithCancel(context.Background())
cancel()
imported, err := svc.importDownloadedItems(ctx, &models.Download{URL: "u"}, tempDir, "", "")
if !errors.Is(err, context.Canceled) {
t.Errorf("error = %v, want context.Canceled", err)
}
if imported != 0 {
t.Errorf("imported = %d, want 0", imported)
}
entries, err := os.ReadDir(libDir)
if err != nil {
t.Fatal(err)
}
if len(entries) != 0 {
t.Errorf("library got %d entries, want 0", len(entries))
}
}
// A cancel part-way through a playlist must stop at an item boundary: the items
// already imported stay, and no half-written directory is left behind.
func TestImportDownloadedItemsStopsBetweenItems(t *testing.T) {
requireFFmpeg(t)
libDir := t.TempDir()
svc := &DownloadService{
cfg: &config.Config{LibraryDir: libDir, FFprobePath: "ffprobe"},
librarySvc: NewLibraryService(libDir, "ffmpeg", "ffprobe"),
}
tempDir := t.TempDir()
for i, name := range []string{"item-00001", "item-00002", "item-00003"} {
itemDir := filepath.Join(tempDir, name)
if err := os.MkdirAll(itemDir, 0755); err != nil {
t.Fatal(err)
}
makeTestVideo(t, filepath.Join(itemDir, "raw.mp4"))
info := fmt.Sprintf(`{"id":"v%d","title":"Clip %d"}`, i+1, i+1)
if err := os.WriteFile(filepath.Join(itemDir, "clip.info.json"), []byte(info), 0644); err != nil {
t.Fatal(err)
}
}
// Cancel as soon as the first item lands, so the second boundary check trips.
ctx, cancel := context.WithCancel(context.Background())
stop := make(chan struct{})
go func() {
defer close(stop)
for {
if _, err := os.Stat(filepath.Join(libDir, "Clip 1", itemMarkerName)); err == nil {
cancel()
return
}
time.Sleep(2 * time.Millisecond)
}
}()
imported, err := svc.importDownloadedItems(ctx, &models.Download{URL: "u"}, tempDir, "", "")
<-stop
cancel()
if !errors.Is(err, context.Canceled) {
t.Fatalf("error = %v, want context.Canceled", err)
}
if imported < 1 || imported > 2 {
t.Errorf("imported = %d, want the cancel to land after 1 or 2 items", imported)
}
// Whatever was imported must be complete: an item dir without a marker means
// the import stopped half-way through writing it.
entries, err := os.ReadDir(libDir)
if err != nil {
t.Fatal(err)
}
if len(entries) != imported {
t.Errorf("library has %d dirs but reported %d imported", len(entries), imported)
}
for _, entry := range entries {
marker := filepath.Join(libDir, entry.Name(), itemMarkerName)
if _, err := os.Stat(marker); err != nil {
t.Errorf("item %q left without a marker: %v", entry.Name(), err)
}
}
}
// Overwrite mode replaces the existing copy of a video in place instead of
// adding a second directory for it.
func TestImportItemDirOverwriteReplacesExisting(t *testing.T) {
requireFFmpeg(t)
libDir := t.TempDir()
svc := &DownloadService{
cfg: &config.Config{LibraryDir: libDir, FFprobePath: "ffprobe"},
librarySvc: NewLibraryService(libDir, "ffmpeg", "ffprobe"),
}
// First import establishes the item.
first := t.TempDir()
makeTestVideo(t, filepath.Join(first, "raw.mp4"))
if err := os.WriteFile(filepath.Join(first, "clip.info.json"), []byte(`{"id":"vid1","title":"Old Title"}`), 0644); err != nil {
t.Fatal(err)
}
if err := svc.importItemDir(context.Background(), "https://example.com/v", first, libDir, "overwrite", ""); err != nil {
t.Fatalf("first import: %v", err)
}
if err := os.WriteFile(filepath.Join(libDir, "Old Title", "stale.txt"), []byte("gone"), 0644); err != nil {
t.Fatal(err)
}
// Second import of the same video id, now retitled upstream.
second := t.TempDir()
makeTestVideo(t, filepath.Join(second, "raw.mp4"))
if err := os.WriteFile(filepath.Join(second, "clip.info.json"), []byte(`{"id":"vid1","title":"New Title"}`), 0644); err != nil {
t.Fatal(err)
}
if err := svc.importItemDir(context.Background(), "https://example.com/v", second, libDir, "overwrite", ""); err != nil {
t.Fatalf("second import: %v", err)
}
entries, err := os.ReadDir(libDir)
if err != nil {
t.Fatal(err)
}
if len(entries) != 1 || entries[0].Name() != "New Title" {
var names []string
for _, e := range entries {
names = append(names, e.Name())
}
t.Fatalf("library = %v, want exactly [New Title]", names)
}
if _, err := os.Stat(filepath.Join(libDir, "New Title", "stale.txt")); err == nil {
t.Error("the replaced item kept a file from the old copy")
}
}
// Without overwrite mode the same video is imported again beside the first, so
// the guard above is really the thing preventing duplicates.
func TestImportItemDirWithoutOverwriteKeepsBoth(t *testing.T) {
requireFFmpeg(t)
libDir := t.TempDir()
svc := &DownloadService{
cfg: &config.Config{LibraryDir: libDir, FFprobePath: "ffprobe"},
librarySvc: NewLibraryService(libDir, "ffmpeg", "ffprobe"),
}
for i := 0; i < 2; i++ {
src := t.TempDir()
makeTestVideo(t, filepath.Join(src, "raw.mp4"))
if err := os.WriteFile(filepath.Join(src, "clip.info.json"), []byte(`{"id":"vid1","title":"Same"}`), 0644); err != nil {
t.Fatal(err)
}
if err := svc.importItemDir(context.Background(), "https://example.com/v", src, libDir, "", ""); err != nil {
t.Fatalf("import %d: %v", i, err)
}
}
entries, err := os.ReadDir(libDir)
if err != nil {
t.Fatal(err)
}
if len(entries) != 2 {
t.Errorf("library has %d dirs, want 2 (Same and Same-1)", len(entries))
}
}
// moveFile falls back to copy+delete when the source and destination are on
// different filesystems, which is the normal case for a temp dir on tmpfs and a
// library on disk. /dev/shm is a second tmpfs mount, so a rename across it
// really does return EXDEV.
func TestMoveFileCrossDeviceFallback(t *testing.T) {
dstDir, err := os.MkdirTemp("/dev/shm", "vidarchive-move-")
if err != nil {
t.Skipf("no second filesystem available for a cross-device move: %v", err)
}
defer os.RemoveAll(dstDir)
srcDir := t.TempDir()
if sameDevice(t, srcDir, dstDir) {
t.Skip("temp dir and /dev/shm are on the same filesystem")
}
src := filepath.Join(srcDir, "clip.mp4")
content := []byte("not really a video, but the bytes must survive")
if err := os.WriteFile(src, content, 0640); err != nil {
t.Fatal(err)
}
// Confirm the fallback is actually what gets exercised.
dst := filepath.Join(dstDir, "clip.mp4")
if err := os.Rename(src, dst); !errors.Is(err, syscall.EXDEV) {
t.Skipf("rename across the two dirs did not report EXDEV: %v", err)
}
if err := moveFile(src, dst); err != nil {
t.Fatalf("moveFile: %v", err)
}
got, err := os.ReadFile(dst)
if err != nil {
t.Fatalf("read destination: %v", err)
}
if string(got) != string(content) {
t.Errorf("destination content = %q, want %q", got, content)
}
info, err := os.Stat(dst)
if err != nil {
t.Fatal(err)
}
if info.Mode().Perm() != 0640 {
t.Errorf("destination mode = %v, want 0640", info.Mode().Perm())
}
if _, err := os.Stat(src); !os.IsNotExist(err) {
t.Error("source still exists after a successful move")
}
}
// A move that cannot even read the source must not leave a destination behind.
func TestMoveFileMissingSourceLeavesNoDestination(t *testing.T) {
dir := t.TempDir()
dst := filepath.Join(dir, "dst.mp4")
if err := moveFile(filepath.Join(dir, "missing.mp4"), dst); err == nil {
t.Fatal("expected an error moving a file that does not exist")
}
if _, err := os.Stat(dst); !os.IsNotExist(err) {
t.Error("destination was created for a failed move")
}
}
func sameDevice(t *testing.T, a, b string) bool {
t.Helper()
var sa, sb syscall.Stat_t
if err := syscall.Stat(a, &sa); err != nil {
t.Fatal(err)
}
if err := syscall.Stat(b, &sb); err != nil {
t.Fatal(err)
}
return sa.Dev == sb.Dev
}
Ainternal/service/execute_download_test.go
@@ -0,0 +1,567 @@
package service
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
"time"
"vidarchive/internal/config"
"vidarchive/internal/models"
"vidarchive/internal/repository"
)
// execEnv is a DownloadService wired to a real database and temp directories,
// with yt-dlp and ffprobe replaced by scripts the test controls.
type execEnv struct {
svc *DownloadService
repo *repository.DownloadRepository
subRepo *repository.SubscriptionRepository
cfg *config.Config
// scratch holds the marker/signal files the fake tools read and write.
scratch string
}
func newExecEnv(t *testing.T) *execEnv {
t.Helper()
db := setupTestDB(t)
t.Cleanup(func() { db.Close() })
root := t.TempDir()
cfg := &config.Config{
LibraryDir: filepath.Join(root, "library"),
TempDir: filepath.Join(root, "temp"),
YTDLPPath: "/bin/false",
FFmpegPath: "ffmpeg",
FFprobePath: "ffprobe",
}
for _, dir := range []string{cfg.LibraryDir, cfg.TempDir} {
if err := os.MkdirAll(dir, 0755); err != nil {
t.Fatal(err)
}
}
downloadRepo := repository.NewDownloadRepository(db)
subRepo := repository.NewSubscriptionRepository(db)
svc := NewDownloadService(
downloadRepo,
NewLibraryService(cfg.LibraryDir, cfg.FFmpegPath, cfg.FFprobePath),
NewPresetService(repository.NewPresetRepository(db)),
NewSettingsService(repository.NewSettingsRepository(db)),
NewSubscriptionService(subRepo, cfg),
cfg,
)
return &execEnv{svc: svc, repo: downloadRepo, subRepo: subRepo, cfg: cfg, scratch: root}
}
// fakeYTDLP installs a stand-in for yt-dlp. body is shell run with $DEST set to
// the directory yt-dlp was told to write into (its -P argument), $SCRATCH set to
// the test's scratch dir, and $RUN set to the invocation count, so a body can
// behave differently on the metadata second pass.
func (e *execEnv) fakeYTDLP(t *testing.T, body string) {
t.Helper()
e.cfg.YTDLPPath = writeScript(t, filepath.Join(e.scratch, "yt-dlp"), `
DEST=""
prev=""
for a in "$@"; do
if [ "$prev" = "-P" ]; then DEST="$a"; fi
prev="$a"
done
export DEST
export SCRATCH="`+e.scratch+`"
RUN=$(( $(cat "$SCRATCH/runs" 2>/dev/null || echo 0) + 1 ))
echo "$RUN" > "$SCRATCH/runs"
export RUN
`+body)
}
// fakeFFprobe installs a stand-in for ffprobe. It reports a 3 second duration,
// which is what the real one would say about the test videos.
func (e *execEnv) fakeFFprobe(t *testing.T, body string) {
t.Helper()
e.cfg.FFprobePath = writeScript(t, filepath.Join(e.scratch, "ffprobe"), `
export SCRATCH="`+e.scratch+`"
`+body+`
echo 3.0`)
}
func writeScript(t *testing.T, path, body string) string {
t.Helper()
if err := os.WriteFile(path, []byte("#!/bin/sh\n"+body+"\n"), 0755); err != nil {
t.Fatal(err)
}
return path
}
// writeItem is shell that creates one importable item under $DEST. The media
// file is a copy of a real video, so the mimetype classification in
// importItemDir sees actual video content.
func (e *execEnv) writeItem(t *testing.T, dirName, title, videoID string) string {
t.Helper()
requireFFmpeg(t)
src := filepath.Join(e.scratch, "source-"+dirName+".mp4")
makeTestVideo(t, src)
return `
mkdir -p "$DEST/` + dirName + `"
cp "` + src + `" "$DEST/` + dirName + `/clip.mp4"
printf '{"id":"` + videoID + `","title":"` + title + `","webpage_url":"https://example.com/` + videoID + `"}' \
> "$DEST/` + dirName + `/clip.info.json"
`
}
// queue inserts a queued download, the state ExecuteDownload expects.
func (e *execEnv) queue(t *testing.T, d *models.Download) *models.Download {
t.Helper()
d.Status = "queued"
if d.URL == "" {
d.URL = "https://example.com/watch"
}
if err := e.repo.Create(d); err != nil {
t.Fatalf("create download: %v", err)
}
return d
}
func (e *execEnv) status(t *testing.T, id int64) *models.Download {
t.Helper()
got, err := e.repo.GetByID(id)
if err != nil {
t.Fatalf("reload download %d: %v", id, err)
}
return got
}
// waitForFile blocks until path exists. The fake tools touch a file to say they
// have started, which lets a test act at a known point instead of sleeping.
func waitForFile(t *testing.T, path string) {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for time.Now().Before(deadline) {
if _, err := os.Stat(path); err == nil {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("timed out waiting for %s", path)
}
func libraryEntries(t *testing.T, dir string) []string {
t.Helper()
entries, err := os.ReadDir(dir)
if err != nil {
t.Fatal(err)
}
var names []string
for _, entry := range entries {
names = append(names, entry.Name())
}
return names
}
func TestExecuteDownloadCompletes(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, e.writeItem(t, "item-00001", "My Clip", "abc123"))
d := e.queue(t, &models.Download{})
processed, err := e.svc.ExecuteDownload(context.Background(), d)
if err != nil {
t.Fatalf("ExecuteDownload: %v", err)
}
if !processed {
t.Error("processed = false, want true")
}
got := e.status(t, d.ID)
if got.Status != "completed" {
t.Errorf("status = %q, want completed", got.Status)
}
if !got.CompletedAt.Valid {
t.Error("completed_at not set")
}
if names := libraryEntries(t, e.cfg.LibraryDir); len(names) != 1 || names[0] != "My Clip" {
t.Errorf("library = %v, want [My Clip]", names)
}
if _, err := os.Stat(e.svc.tempDirFor(d.ID)); !os.IsNotExist(err) {
t.Error("temp download dir was not removed")
}
}
// A download already claimed by another worker must be left alone: Submit and
// the 2s queue checker can both enqueue the same row.
func TestExecuteDownloadSkipsAlreadyClaimed(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, `touch "$SCRATCH/ran"`)
d := e.queue(t, &models.Download{})
if _, err := e.repo.MarkStarted(d.ID); err != nil {
t.Fatal(err)
}
processed, err := e.svc.ExecuteDownload(context.Background(), d)
if err != nil {
t.Errorf("error = %v, want nil", err)
}
if processed {
t.Error("processed = true, want false for an already-claimed download")
}
if _, err := os.Stat(filepath.Join(e.scratch, "ran")); err == nil {
t.Error("yt-dlp ran for a download this worker did not claim")
}
}
func TestExecuteDownloadRecordsYTDLPFailure(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, `echo "ERROR: video unavailable" >&2; exit 3`)
d := e.queue(t, &models.Download{})
processed, err := e.svc.ExecuteDownload(context.Background(), d)
if err == nil {
t.Fatal("expected an error from a failing yt-dlp")
}
if processed {
t.Error("processed = true, want false")
}
got := e.status(t, d.ID)
if got.Status != "error" {
t.Errorf("status = %q, want error", got.Status)
}
if !got.ErrorMessage.Valid || got.ErrorMessage.String == "" {
t.Error("error_message not recorded")
}
if !strings.Contains(got.Logs.String, "video unavailable") {
t.Errorf("yt-dlp output not persisted to logs: %q", got.Logs.String)
}
}
// yt-dlp can exit 0 having downloaded nothing (every entry filtered out). For a
// plain download that is a failure, not a silent success.
func TestExecuteDownloadEmptyResultIsFailure(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, `mkdir -p "$DEST/item-00001"; exit 0`)
d := e.queue(t, &models.Download{})
if _, err := e.svc.ExecuteDownload(context.Background(), d); err == nil {
t.Fatal("expected an error when no media was downloaded")
}
got := e.status(t, d.ID)
if got.Status != "error" {
t.Errorf("status = %q, want error", got.Status)
}
if !strings.Contains(got.ErrorMessage.String, "no media files") {
t.Errorf("error_message = %q, want it to mention no media files", got.ErrorMessage.String)
}
}
// A reserved flag must fail before yt-dlp is ever started, and a preset's flags
// are checked as well as the download's own.
func TestExecuteDownloadRejectsReservedPresetFlag(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, `touch "$SCRATCH/ran"`)
preset := &models.Preset{Name: "Bad", CustomFlags: "-o /tmp/anywhere.mp4"}
if err := e.svc.presetSvc.Create(preset); err != nil {
t.Fatalf("create preset: %v", err)
}
d := e.queue(t, &models.Download{PresetID: sqlNullInt64(preset.ID)})
if _, err := e.svc.ExecuteDownload(context.Background(), d); err == nil {
t.Fatal("expected the reserved flag to be rejected")
}
if got := e.status(t, d.ID); got.Status != "error" {
t.Errorf("status = %q, want error", got.Status)
}
if _, err := os.Stat(filepath.Join(e.scratch, "ran")); err == nil {
t.Error("yt-dlp ran despite a reserved flag")
}
}
// A user-initiated cancel while yt-dlp runs is terminal: the row is recorded
// "cancelled" and nothing reaches the library.
func TestExecuteDownloadCancelDuringRun(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, `touch "$SCRATCH/started"; sleep 60`)
d := e.queue(t, &models.Download{})
done := make(chan error, 1)
go func() {
_, err := e.svc.ExecuteDownload(context.Background(), d)
done <- err
}()
waitForFile(t, filepath.Join(e.scratch, "started"))
e.svc.cancelDownload(d.ID)
select {
case err := <-done:
if err != ErrCancelled {
t.Errorf("error = %v, want ErrCancelled", err)
}
case <-time.After(20 * time.Second):
t.Fatal("ExecuteDownload did not return after cancel")
}
if got := e.status(t, d.ID); got.Status != "cancelled" {
t.Errorf("status = %q, want cancelled", got.Status)
}
if names := libraryEntries(t, e.cfg.LibraryDir); len(names) != 0 {
t.Errorf("library = %v, want empty", names)
}
}
// A shutdown is not a user cancel: the row must stay "downloading" so that
// ResetStalledDownloads re-queues it on the next start instead of losing it.
func TestExecuteDownloadShutdownLeavesRowResumable(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, `touch "$SCRATCH/started"; sleep 60`)
d := e.queue(t, &models.Download{})
// The parent context stands in for the worker pool's, which Stop cancels.
parent, stop := context.WithCancel(context.Background())
defer stop()
done := make(chan error, 1)
go func() {
_, err := e.svc.ExecuteDownload(parent, d)
done <- err
}()
waitForFile(t, filepath.Join(e.scratch, "started"))
stop()
select {
case err := <-done:
if err != ErrCancelled {
t.Errorf("error = %v, want ErrCancelled", err)
}
case <-time.After(20 * time.Second):
t.Fatal("ExecuteDownload did not return after shutdown")
}
if got := e.status(t, d.ID); got.Status != "downloading" {
t.Fatalf("status = %q, want downloading so the restart can resume it", got.Status)
}
if err := e.svc.ResetStalledDownloads(); err != nil {
t.Fatalf("ResetStalledDownloads: %v", err)
}
if got := e.status(t, d.ID); got.Status != "queued" {
t.Errorf("status after restart = %q, want queued", got.Status)
}
}
// A cancel that lands after yt-dlp exited 0 must not be recorded as a completed
// download. The cancel here arrives during the import, so the yt-dlp error is
// nil — the case that used to fall through to MarkCompleted.
func TestExecuteDownloadCancelAfterSuccessIsNotCompleted(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, e.writeItem(t, "item-00001", "My Clip", "abc123"))
// ffprobe runs during the import, after yt-dlp has already succeeded. Block
// there so the cancel lands at exactly that point.
e.fakeFFprobe(t, `
touch "$SCRATCH/probing"
while [ ! -f "$SCRATCH/proceed" ]; do sleep 0.05; done`)
d := e.queue(t, &models.Download{})
done := make(chan error, 1)
go func() {
_, err := e.svc.ExecuteDownload(context.Background(), d)
done <- err
}()
waitForFile(t, filepath.Join(e.scratch, "probing"))
e.svc.cancelDownload(d.ID)
if err := os.WriteFile(filepath.Join(e.scratch, "proceed"), nil, 0644); err != nil {
t.Fatal(err)
}
select {
case err := <-done:
if err != ErrCancelled {
t.Errorf("error = %v, want ErrCancelled", err)
}
case <-time.After(20 * time.Second):
t.Fatal("ExecuteDownload did not return after cancel")
}
if got := e.status(t, d.ID); got.Status != "cancelled" {
t.Errorf("status = %q, want cancelled (a cancelled run must not read as completed)", got.Status)
}
}
// The same rule for a metadata-mode subscription refresh, whose second pass
// returns the yt-dlp error rather than the import error — so a cancel there also
// leaves a nil error behind.
func TestExecuteDownloadMetadataCancelIsNotCompleted(t *testing.T) {
e := newExecEnv(t)
// Pass 1 (--skip-download) writes metadata for an entry the library does not
// have; pass 2 downloads it as a full item.
e.fakeYTDLP(t, `
if [ "$RUN" = "1" ]; then
mkdir -p "$DEST/item-00001"
printf '{"id":"new1","title":"Fresh","webpage_url":"https://example.com/new1"}' > "$DEST/item-00001/clip.info.json"
else
`+e.writeItem(t, "item-00001", "Fresh", "new1")+`
fi`)
e.fakeFFprobe(t, `
touch "$SCRATCH/probing"
while [ ! -f "$SCRATCH/proceed" ]; do sleep 0.05; done`)
sub := &models.Subscription{
Name: "Channel",
URL: "https://example.com/channel",
Enabled: true,
RefreshMode: "metadata",
ScheduleKind: "daily",
}
if err := e.subRepo.Create(sub); err != nil {
t.Fatalf("create subscription: %v", err)
}
d := e.queue(t, &models.Download{SubscriptionID: sqlNullInt64(sub.ID)})
done := make(chan error, 1)
go func() {
_, err := e.svc.ExecuteDownload(context.Background(), d)
done <- err
}()
waitForFile(t, filepath.Join(e.scratch, "probing"))
e.svc.cancelDownload(d.ID)
if err := os.WriteFile(filepath.Join(e.scratch, "proceed"), nil, 0644); err != nil {
t.Fatal(err)
}
select {
case err := <-done:
if err != ErrCancelled {
t.Errorf("error = %v, want ErrCancelled", err)
}
case <-time.After(20 * time.Second):
t.Fatal("ExecuteDownload did not return after cancel")
}
if got := e.status(t, d.ID); got.Status != "cancelled" {
t.Errorf("status = %q, want cancelled", got.Status)
}
}
// A restart must clear the temp dirs of downloads it re-queues: leaving them
// behind makes the re-run import into a fresh uniqueDir and the library ends up
// holding the same item twice.
func TestResetStalledDownloadsClearsTempDirs(t *testing.T) {
e := newExecEnv(t)
d := e.queue(t, &models.Download{})
if _, err := e.repo.MarkStarted(d.ID); err != nil {
t.Fatal(err)
}
dirs := e.svc.tempDirsFor(d.ID)
if len(dirs) != 2 {
t.Fatalf("tempDirsFor returned %d dirs, want the main and second-pass dirs", len(dirs))
}
for _, dir := range dirs {
if err := os.MkdirAll(dir, 0755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(dir, "partial.mp4.part"), []byte("x"), 0644); err != nil {
t.Fatal(err)
}
}
// A download that is not stalled must keep whatever it owns.
other := e.queue(t, &models.Download{})
keep := e.svc.tempDirFor(other.ID)
if err := os.MkdirAll(keep, 0755); err != nil {
t.Fatal(err)
}
if err := e.svc.ResetStalledDownloads(); err != nil {
t.Fatalf("ResetStalledDownloads: %v", err)
}
if got := e.status(t, d.ID); got.Status != "queued" {
t.Errorf("status = %q, want queued", got.Status)
}
for _, dir := range dirs {
if _, err := os.Stat(dir); !os.IsNotExist(err) {
t.Errorf("stale temp dir %s was not removed", dir)
}
}
if _, err := os.Stat(keep); err != nil {
t.Errorf("temp dir of a queued download was removed: %v", err)
}
}
// Deleting a download from the queue stops the work it started, so yt-dlp does
// not keep running for a row that no longer exists.
func TestDeleteCancelsRunningDownload(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, `touch "$SCRATCH/started"; sleep 60`)
d := e.queue(t, &models.Download{})
done := make(chan error, 1)
go func() {
_, err := e.svc.ExecuteDownload(context.Background(), d)
done <- err
}()
waitForFile(t, filepath.Join(e.scratch, "started"))
if err := e.svc.Delete(d.ID); err != nil {
t.Fatalf("Delete: %v", err)
}
select {
case err := <-done:
if err != ErrCancelled {
t.Errorf("error = %v, want ErrCancelled", err)
}
case <-time.After(20 * time.Second):
t.Fatal("deleting the download did not stop it")
}
if _, err := e.repo.GetByID(d.ID); err == nil {
t.Error("download row still exists after Delete")
}
if names := libraryEntries(t, e.cfg.LibraryDir); len(names) != 0 {
t.Errorf("library = %v, want empty", names)
}
}
// Clearing the queue must also stop what is running, for the same reason.
func TestDeleteAllCancelsRunningDownload(t *testing.T) {
e := newExecEnv(t)
e.fakeYTDLP(t, `touch "$SCRATCH/started"; sleep 60`)
d := e.queue(t, &models.Download{})
done := make(chan error, 1)
go func() {
_, err := e.svc.ExecuteDownload(context.Background(), d)
done <- err
}()
waitForFile(t, filepath.Join(e.scratch, "started"))
if err := e.svc.DeleteAll(); err != nil {
t.Fatalf("DeleteAll: %v", err)
}
select {
case err := <-done:
if err != ErrCancelled {
t.Errorf("error = %v, want ErrCancelled", err)
}
case <-time.After(20 * time.Second):
t.Fatal("clearing the queue did not stop the running download")
}
}
Minternal/service/library.go
@@ -137,6 +137,13 @@ func (s *LibraryService) scannedItem(itemDir, relPath string) (*models.LibraryIt
return item, nil
}
// ResolveWithinLibrary resolves a caller-supplied relative directory against the
// library root and rejects anything that escapes it. The directory need not
// exist yet, so it is safe to use when choosing a download's output location.
func (s *LibraryService) ResolveWithinLibrary(relPath string) (string, error) {
return s.resolveItemDir(filepath.Clean(relPath))
}
func (s *LibraryService) resolveItemDir(relPath string) (string, error) {
relPath = strings.Trim(relPath, string(filepath.Separator))
if relPath == "" {
@@ -848,10 +855,30 @@ func (s *LibraryService) FindByVideoID(baseDir, id string) (string, bool) {
if id == "" {
return "", false
}
var match string
// An unreadable base dir simply means no match here.
_ = s.eachItemDir(baseDir, "FindByVideoID", func(itemDir string, meta models.ItemMetadata) bool {
if meta.VideoID != id {
return true
}
match = itemDir
return false
})
return match, match != ""
}
// eachItemDir walks the marked library items directly under baseDir, reading
// each one's metadata, and calls fn until it returns false. Entries that aren't
// items, or whose metadata can't be read, are skipped — a single bad item must
// not abort a scan. logLabel names the caller in those skip messages.
func (s *LibraryService) eachItemDir(baseDir, logLabel string, fn func(itemDir string, meta models.ItemMetadata) bool) error {
entries, err := os.ReadDir(baseDir)
if err != nil {
return "", false
return err
}
for _, entry := range entries {
if !entry.IsDir() || entry.Name() == subtitlesDirName {
continue
@@ -862,14 +889,15 @@ func (s *LibraryService) FindByVideoID(baseDir, id string) (string, bool) {
}
meta, err := s.readOrCreateMetadata(itemDir)
if err != nil {
log.Printf("FindByVideoID: skipping %s: %v", itemDir, err)
log.Printf("%s: skipping %s: %v", logLabel, itemDir, err)
continue
}
if meta.VideoID == id {
return itemDir, true
if !fn(itemDir, meta) {
return nil
}
}
return "", false
return nil
}
// PruneToIDSet deletes items directly under baseDir whose identity key is not in
@@ -878,37 +906,23 @@ func (s *LibraryService) FindByVideoID(baseDir, id string) (string, bool) {
// never delete something we can't positively identify). Returns the number
// removed.
func (s *LibraryService) PruneToIDSet(baseDir string, keep map[string]bool) (int, error) {
entries, err := os.ReadDir(baseDir)
if err != nil {
return 0, err
}
removed := 0
for _, entry := range entries {
if !entry.IsDir() || entry.Name() == subtitlesDirName {
continue
}
itemDir := filepath.Join(baseDir, entry.Name())
if _, err := os.Stat(filepath.Join(itemDir, itemMarkerName)); err != nil {
continue
}
meta, err := s.readOrCreateMetadata(itemDir)
if err != nil {
log.Printf("PruneToIDSet: skipping %s: %v", itemDir, err)
continue
}
err := s.eachItemDir(baseDir, "PruneToIDSet", func(itemDir string, meta models.ItemMetadata) bool {
if meta.VideoID == "" || keep[meta.VideoID] {
continue
return true
}
if rel, err := filepath.Rel(s.libraryDir, itemDir); err == nil {
s.evictCachedScan(filepath.ToSlash(rel))
}
if err := os.RemoveAll(itemDir); err != nil {
log.Printf("warning: prune failed to remove %s: %v", itemDir, err)
continue
return true
}
removed++
}
return removed, nil
return true
})
return removed, err
}
func (s *LibraryService) SubtitleDir(relPath string) string {
@@ -919,8 +933,15 @@ func (s *LibraryService) SubtitleDir(relPath string) string {
return filepath.Join(itemDir, subtitlesDirName)
}
func (s *LibraryService) GetSubtitlePath(relPath, lang string) string {
return filepath.Join(s.SubtitleDir(relPath), lang+".vtt")
// GetSubtitlePath returns the .vtt path for a language. It errors when the item
// can't be resolved: joining onto an empty dir would yield a bare relative name
// that the caller would then serve relative to the process working directory.
func (s *LibraryService) GetSubtitlePath(relPath, lang string) (string, error) {
dir := s.SubtitleDir(relPath)
if dir == "" {
return "", fmt.Errorf("resolve subtitle dir for %q", relPath)
}
return filepath.Join(dir, lang+".vtt"), nil
}
func (s *LibraryService) GetSubtitles(ctx context.Context, relPath string) ([]models.SubtitleTrack, error) {
Minternal/service/subscription.go
@@ -2,6 +2,8 @@ package service
import (
"fmt"
"log"
"os"
"path/filepath"
"time"
@@ -104,6 +106,16 @@ func (s *SubscriptionService) MarkRun(id int64, lastRunAt, nextRunAt time.Time,
return s.repo.MarkRun(id, lastRunAt, nextRunAt, status)
}
// Delete removes the subscription and the download-archive file "skip" mode
// keeps for it. Leaving the archive behind would make a later subscription that
// reuses the id silently skip entries it never downloaded. The library
// directory is left alone: the archived media is the point of the tool.
func (s *SubscriptionService) Delete(id int64) error {
return s.repo.Delete(id)
if err := s.repo.Delete(id); err != nil {
return err
}
if err := os.Remove(s.ArchivePath(id)); err != nil && !os.IsNotExist(err) {
log.Printf("subscription %d: failed to remove archive file: %v", id, err)
}
return nil
}
Minternal/service/subscription_test.go
@@ -1,10 +1,14 @@
package service
import (
"os"
"path/filepath"
"testing"
"time"
"vidarchive/internal/config"
"vidarchive/internal/models"
"vidarchive/internal/repository"
)
func TestComputeNextRun(t *testing.T) {
@@ -64,3 +68,64 @@ func TestComputeNextRun(t *testing.T) {
c.check(t, next)
}
}
// Deleting a subscription removes its download-archive file. A later
// subscription that reuses the id would otherwise silently skip entries it
// never downloaded.
func TestSubscriptionDeleteRemovesArchive(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
cfg := &config.Config{DataDir: t.TempDir()}
svc := NewSubscriptionService(repository.NewSubscriptionRepository(db), cfg)
sub := &models.Subscription{
Name: "Channel",
URL: "https://example.com/channel",
Enabled: true,
RefreshMode: "skip",
ScheduleKind: "daily",
}
if err := svc.Create(sub); err != nil {
t.Fatalf("create: %v", err)
}
archive := svc.ArchivePath(sub.ID)
if err := os.MkdirAll(filepath.Dir(archive), 0755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(archive, []byte("youtube abc123\n"), 0644); err != nil {
t.Fatal(err)
}
if err := svc.Delete(sub.ID); err != nil {
t.Fatalf("delete: %v", err)
}
if _, err := os.Stat(archive); !os.IsNotExist(err) {
t.Errorf("archive file still present after delete: %v", err)
}
if _, err := svc.GetByID(sub.ID); err == nil {
t.Error("subscription row still exists after delete")
}
}
// A subscription with no archive file (any mode other than "skip") deletes
// cleanly rather than reporting the missing file as a failure.
func TestSubscriptionDeleteWithoutArchive(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
svc := NewSubscriptionService(repository.NewSubscriptionRepository(db), &config.Config{DataDir: t.TempDir()})
sub := &models.Subscription{
Name: "Channel",
URL: "https://example.com/channel",
RefreshMode: "overwrite",
ScheduleKind: "daily",
}
if err := svc.Create(sub); err != nil {
t.Fatalf("create: %v", err)
}
if err := svc.Delete(sub.ID); err != nil {
t.Errorf("delete without an archive file: %v", err)
}
}
Minternal/worker/pool.go
@@ -2,8 +2,9 @@ package worker
import (
"context"
"fmt"
"errors"
"log"
"sync"
"time"
"vidarchive/internal/models"
@@ -16,6 +17,7 @@ type Pool struct {
queue chan *models.Download
ctx context.Context
cancel context.CancelFunc
wg sync.WaitGroup
}
func New(downloadSvc *service.DownloadService, workers int) *Pool {
@@ -30,42 +32,56 @@ func New(downloadSvc *service.DownloadService, workers int) *Pool {
}
func (p *Pool) Start() {
// Reset any downloads that were in progress during a previous run
// Reset any downloads that were in progress during a previous run, and clear
// the temp dirs they left behind so a re-run doesn't import a duplicate.
if err := p.downloadSvc.ResetStalledDownloads(); err != nil {
log.Printf("Warning: failed to reset stalled downloads: %v", err)
}
for i := 0; i < p.workers; i++ {
p.wg.Add(1)
go p.worker(i)
}
// Queue checker - polls DB for queued downloads
p.wg.Add(1)
go p.queueChecker()
}
// Stop cancels running downloads and waits for the workers to return, so the
// process doesn't exit while a yt-dlp child is still writing to the temp dir.
func (p *Pool) Stop() {
p.cancel()
p.downloadSvc.CancelAll()
p.wg.Wait()
}
// Submit enqueues a download without blocking. When the buffer is full the row
// stays "queued" in the database and the queue checker picks it up on a later
// tick, so a burst of submissions can't stall the HTTP handler that made them.
func (p *Pool) Submit(d *models.Download) {
if p.ctx.Err() != nil {
return
}
select {
case p.queue <- d:
log.Printf("Download %d queued", d.ID)
case <-p.ctx.Done():
default:
log.Printf("Download %d left in database queue (worker buffer full)", d.ID)
}
}
func (p *Pool) worker(id int) {
defer p.wg.Done()
log.Printf("Worker %d started", id)
for {
select {
case d := <-p.queue:
if d == nil {
continue
}
log.Printf("Worker %d processing download %d", id, d.ID)
processed, err := p.downloadSvc.ExecuteDownload(d)
processed, err := p.downloadSvc.ExecuteDownload(p.ctx, d)
switch {
case errors.Is(err, service.ErrCancelled):
log.Printf("Worker %d download %d cancelled", id, d.ID)
case err != nil:
log.Printf("Worker %d download %d failed: %v", id, d.ID, err)
case processed:
@@ -81,6 +97,7 @@ func (p *Pool) worker(id int) {
}
func (p *Pool) queueChecker() {
defer p.wg.Done()
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
@@ -105,34 +122,10 @@ func (p *Pool) checkQueue() {
}
for _, d := range downloads {
// Try to submit - if queue is full, it'll block briefly
select {
case p.queue <- d:
default:
// Queue is full, skip for now
return
}
}
}
func (p *Pool) GetQueueStatus() (queued, active, completed, failed int, err error) {
all, err := p.downloadSvc.GetAll("all", "date")
if err != nil {
return 0, 0, 0, 0, fmt.Errorf("get all downloads: %w", err)
}
for _, d := range all {
switch d.Status {
case "queued":
queued++
case "downloading":
active++
case "completed":
completed++
case "error":
failed++
}
}
return queued, active, completed, failed, nil
}
Ainternal/worker/pool_test.go
@@ -0,0 +1,181 @@
package worker
import (
"os"
"path/filepath"
"testing"
"time"
"vidarchive/internal/config"
"vidarchive/internal/database"
"vidarchive/internal/models"
"vidarchive/internal/repository"
"vidarchive/internal/service"
)
// newTestPool builds a pool over a real service and database, with yt-dlp
// replaced by a script that reports it has started and then blocks. scratch is
// where that script writes its marker.
func newTestPool(t *testing.T) (pool *Pool, repo *repository.DownloadRepository, scratch string) {
t.Helper()
root := t.TempDir()
cfg := &config.Config{
DBPath: ":memory:",
LibraryDir: filepath.Join(root, "library"),
TempDir: filepath.Join(root, "temp"),
FFmpegPath: "ffmpeg",
FFprobePath: "ffprobe",
}
for _, dir := range []string{cfg.LibraryDir, cfg.TempDir} {
if err := os.MkdirAll(dir, 0755); err != nil {
t.Fatal(err)
}
}
// The script must not exit on its own: the point of these tests is that Stop
// interrupts a download rather than waiting one out.
cfg.YTDLPPath = filepath.Join(root, "yt-dlp")
script := "#!/bin/sh\ntouch " + filepath.Join(root, "started") + "\nsleep 300\n"
if err := os.WriteFile(cfg.YTDLPPath, []byte(script), 0755); err != nil {
t.Fatal(err)
}
db, err := database.New(cfg)
if err != nil {
t.Fatalf("init db: %v", err)
}
t.Cleanup(func() { db.Close() })
repo = repository.NewDownloadRepository(db)
downloadSvc := service.NewDownloadService(
repo,
service.NewLibraryService(cfg.LibraryDir, cfg.FFmpegPath, cfg.FFprobePath),
service.NewPresetService(repository.NewPresetRepository(db)),
service.NewSettingsService(repository.NewSettingsRepository(db)),
service.NewSubscriptionService(repository.NewSubscriptionRepository(db), cfg),
cfg,
)
return New(downloadSvc, 2), repo, root
}
func waitForFile(t *testing.T, path string) {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for time.Now().Before(deadline) {
if _, err := os.Stat(path); err == nil {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("timed out waiting for %s", path)
}
// Stop must return while a download is still running. If it waited for yt-dlp
// to finish on its own, a shutdown would hang for as long as the download takes.
func TestStopReturnsWhileDownloadRunning(t *testing.T) {
pool, repo, scratch := newTestPool(t)
d := &models.Download{URL: "https://example.com/v", Status: "queued"}
if err := repo.Create(d); err != nil {
t.Fatal(err)
}
pool.Start()
pool.Submit(d)
waitForFile(t, filepath.Join(scratch, "started"))
stopped := make(chan struct{})
go func() {
pool.Stop()
close(stopped)
}()
select {
case <-stopped:
case <-time.After(30 * time.Second):
t.Fatal("Stop did not return while a download was running")
}
// The row stays "downloading" after a shutdown so the next start resumes it.
got, err := repo.GetByID(d.ID)
if err != nil {
t.Fatal(err)
}
if got.Status != "downloading" {
t.Errorf("status = %q, want downloading", got.Status)
}
}
// Stop is called from the shutdown path and may race with an already-stopped
// pool; calling it twice must not panic or block.
func TestStopIsIdempotent(t *testing.T) {
pool, _, _ := newTestPool(t)
pool.Start()
done := make(chan struct{})
go func() {
pool.Stop()
pool.Stop()
close(done)
}()
select {
case <-done:
case <-time.After(30 * time.Second):
t.Fatal("second Stop blocked")
}
}
// A download submitted after shutdown must be dropped rather than sent to a
// channel nobody is reading.
func TestSubmitAfterStopIsNoOp(t *testing.T) {
pool, repo, _ := newTestPool(t)
pool.Start()
pool.Stop()
d := &models.Download{URL: "https://example.com/v", Status: "queued"}
if err := repo.Create(d); err != nil {
t.Fatal(err)
}
done := make(chan struct{})
go func() {
pool.Submit(d)
close(done)
}()
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("Submit blocked after Stop")
}
if len(pool.queue) != 0 {
t.Errorf("queue holds %d items, want 0 after Stop", len(pool.queue))
}
}
// Submit is called from an HTTP handler, so a full buffer must never block it:
// the row stays queued in the database and the queue checker picks it up later.
func TestSubmitDoesNotBlockWhenBufferFull(t *testing.T) {
pool, _, _ := newTestPool(t)
// Deliberately no Start: nothing drains the buffer.
done := make(chan struct{})
go func() {
for i := 0; i < cap(pool.queue)*2; i++ {
pool.Submit(&models.Download{ID: int64(i + 1), URL: "https://example.com/v"})
}
close(done)
}()
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("Submit blocked on a full buffer")
}
if len(pool.queue) != cap(pool.queue) {
t.Errorf("queue holds %d items, want the buffer to be full at %d", len(pool.queue), cap(pool.queue))
}
}
Mweb/static/style.css
@@ -626,6 +626,11 @@ button:disabled {
color: white;
}
.status-badge.cancelled {
background: var(--pico-muted-color);
color: var(--pico-background-color);
}
.error-message {
color: var(--pico-del-color);
font-size: 0.9rem;
Mweb/templates/queue.html
@@ -12,6 +12,7 @@
<option value="downloading" {{if eq .Data.Status "downloading"}}selected{{end}}>Active</option>
<option value="completed" {{if eq .Data.Status "completed"}}selected{{end}}>Completed</option>
<option value="error" {{if eq .Data.Status "error"}}selected{{end}}>Failed</option>
<option value="cancelled" {{if eq .Data.Status "cancelled"}}selected{{end}}>Cancelled</option>
</select>
<select name="sort">
<option value="date" {{if eq .Data.SortBy "date"}}selected{{end}}>Date</option>