ddc/internal/jobs/manager.go

package jobs

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"ddc/internal/config"
	"ddc/internal/diskspace"
	"ddc/internal/fetch"
	"ddc/internal/files"
	"ddc/internal/guard"
	"ddc/internal/history"
	"ddc/internal/libraries"
	"ddc/internal/preferences"
)

var (
	ErrJobNotFound    = errors.New("download not found")
	ErrJobBusy        = errors.New("download is busy, try again when it is ready")
	ErrInvalidRequest = errors.New("invalid request")
)

type Manager struct {
	settings    config.Settings
	preferences *preferences.Store
	history     *history.Store
	policy      *guard.Policy
	downloader  *fetch.Downloader
	rootContext context.Context
	cancelAll   context.CancelFunc
	slots       slotLimiter
	mutex       sync.RWMutex
	jobs        map[string]*Job
}

func NewManager(settings config.Settings, store *preferences.Store, historyStore *history.Store) *Manager {
	policy := guard.NewPolicy(store.Get().AllowedPorts)
	rootContext, cancelAll := context.WithCancel(context.Background())
	return &Manager{
		settings:    settings,
		preferences: store,
		history:     historyStore,
		policy:      policy,
		downloader:  fetch.NewDownloader(policy),
		rootContext: rootContext,
		cancelAll:   cancelAll,
		jobs:        make(map[string]*Job),
	}
}

func (manager *Manager) Destinations() []libraries.Destination {
	return libraries.Discover(manager.settings.LibraryRoots)
}

func (manager *Manager) Preferences() preferences.Preferences {
	return manager.preferences.Get()
}

func (manager *Manager) UpdatePreferences(next preferences.Preferences) error {
	downloadPaths, err := manager.normalizeDownloadPaths(next.DownloadPaths)
	if err != nil {
		return err
	}
	next.DownloadPaths = downloadPaths
	if err := manager.preferences.Update(next); err != nil {
		return err
	}
	manager.policy.SetAllowedPorts(next.AllowedPorts)
	slog.Debug("preferences saved",
		"parallel_downloads", next.MaxConcurrentDownloads,
		"connections_per_download", next.ConnectionsPerDownload,
		"min_free_gb", next.MinFreeGB,
		"download_paths", len(next.DownloadPaths),
	)
	return nil
}

func (manager *Manager) concurrencyLimit() int {
	return manager.preferences.Get().MaxConcurrentDownloads
}

func (manager *Manager) reservedBytes() int64 {
	return manager.preferences.Get().MinFreeBytes()
}

func (manager *Manager) AvailableBytes() (int64, error) {
	return diskspace.Available(manager.settings.DownloadDir)
}

type CreateRequest struct {
	URL         string `json:"url"`
	FileName    string `json:"fileName"`
	AutoExtract bool   `json:"autoExtract"`
	Destination string `json:"destination"`
	Subfolder   string `json:"subfolder"`
}

func (manager *Manager) Create(request CreateRequest) (Snapshot, error) {
	target, err := manager.policy.ValidateURL(request.URL)
	if err != nil {
		return Snapshot{}, invalid(err)
	}
	requestedName := fetch.SanitizeFileName(request.FileName)
	if strings.TrimSpace(request.FileName) != "" && requestedName == "" {
		return Snapshot{}, invalid(errors.New("file name is not allowed"))
	}
	moveTarget, err := manager.resolveTarget(request.Destination, request.Subfolder)
	if err != nil {
		return Snapshot{}, err
	}
	if err := diskspace.EnsureFree(manager.settings.DownloadDir, 0, manager.reservedBytes()); err != nil {
		return Snapshot{}, err
	}

	id, err := newJobID()
	if err != nil {
		return Snapshot{}, err
	}
	directory := filepath.Join(manager.settings.DownloadDir, id)
	if err := os.Mkdir(directory, 0o755); err != nil {
		return Snapshot{}, err
	}

	job := manager.newJob(id, directory, time.Now())
	job.sourceURL = target.String()
	job.displayURL = target.Redacted()
	job.requestedName = requestedName
	job.autoExtract = request.AutoExtract
	job.target = moveTarget
	job.status = StatusQueued
	job.fileName = initialFileName(requestedName, target.Path, target.Host)
	job.workers.Add(1)

	manager.store(job)
	job.saveRecord()
	job.logger().Info("download added",
		"url", job.displayURL,
		"file_name", requestedName,
		"target", job.snapshot().MoveTarget,
	)
	manager.launchDownload(job)
	return job.snapshot(), nil
}

func (manager *Manager) List() []Snapshot {
	manager.mutex.RLock()
	snapshots := make([]Snapshot, 0, len(manager.jobs))
	for _, job := range manager.jobs {
		snapshot := job.snapshot()
		if snapshot.Status == StatusQueued {
			snapshot.QueuePosition = manager.slots.position(job.id)
		}
		snapshots = append(snapshots, snapshot)
	}
	manager.mutex.RUnlock()

	sort.Slice(snapshots, func(first, second int) bool {
		return snapshots[first].CreatedAt.After(snapshots[second].CreatedAt)
	})
	return snapshots
}

func (manager *Manager) Files(id string) ([]files.Node, error) {
	job, err := manager.lookup(id)
	if err != nil {
		return nil, err
	}
	if job.currentStatus() != StatusReady {
		return nil, ErrJobBusy
	}
	return files.BuildTree(job.directory)
}

func (manager *Manager) Delete(id string) error {
	job, err := manager.lookup(id)
	if err != nil {
		return err
	}
	if !job.markDeleted() {
		return ErrJobBusy
	}

	job.cancel()
	job.workers.Wait()

	manager.mutex.Lock()
	delete(manager.jobs, id)
	manager.mutex.Unlock()

	if err := os.RemoveAll(job.directory); err != nil {
		job.logger().Error("could not remove download files", "directory", job.directory, "error", err)
		return err
	}
	job.logger().Info("download removed")
	return nil
}

func (manager *Manager) Shutdown(ctx context.Context) {
	manager.cancelAll()
	manager.mutex.RLock()
	allJobs := make([]*Job, 0, len(manager.jobs))
	for _, job := range manager.jobs {
		allJobs = append(allJobs, job)
	}
	manager.mutex.RUnlock()

	finished := make(chan struct{})
	go func() {
		for _, job := range allJobs {
			job.workers.Wait()
		}
		close(finished)
	}()
	select {
	case <-finished:
	case <-ctx.Done():
		slog.Warn("stopped waiting for downloads to save progress", "error", ctx.Err())
	}
}

func (manager *Manager) newJob(id, directory string, createdAt time.Time) *Job {
	jobContext, cancel := context.WithCancel(manager.rootContext)
	return &Job{
		id:         id,
		directory:  directory,
		createdAt:  createdAt,
		jobContext: jobContext,
		cancel:     cancel,
	}
}

func (manager *Manager) store(job *Job) {
	manager.mutex.Lock()
	defer manager.mutex.Unlock()
	manager.jobs[job.id] = job
}

func (manager *Manager) lookup(id string) (*Job, error) {
	manager.mutex.RLock()
	defer manager.mutex.RUnlock()
	job, found := manager.jobs[id]
	if !found {
		return nil, ErrJobNotFound
	}
	return job, nil
}

func newJobID() (string, error) {
	buffer := make([]byte, 16)
	if _, err := rand.Read(buffer); err != nil {
		return "", err
	}
	return hex.EncodeToString(buffer), nil
}

func initialFileName(requestedName, urlPath, host string) string {
	if requestedName != "" {
		return requestedName
	}
	if name := fetch.SanitizeFileName(path.Base(urlPath)); name != "" {
		return name
	}
	return host
}

func invalid(err error) error {
	return fmt.Errorf("%w: %w", ErrInvalidRequest, err)
}