playlink/internal/engine/engine.go

// Package engine implements scanning, matching and the hardlink operations.
package engine

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

	"github.com/cdnutter/playlink/internal/fsops"
	"github.com/cdnutter/playlink/internal/probe"
	"github.com/cdnutter/playlink/internal/store"
)

// DefaultExtensions are the audio file types considered by default.
var DefaultExtensions = []string{"flac", "mp3", "m4a", "aac", "alac", "ogg", "oga", "opus", "wav", "aif", "aiff", "wma", "ape", "wv", "dsf", "dff", "mka"}

// QuarantineDirName is the default quarantine folder created in the slave root.
const QuarantineDirName = ".playlink-trash"

var (
	ErrBusy      = errors.New("another job is already running for this library pair")
	ErrDowngrade = errors.New("the master copy is lower quality than the slave copy; confirm the downgrade to continue")
	ErrConfirm   = errors.New("this action modifies the master library and must be confirmed")
)

type Engine struct {
	st           *store.Store
	prober       probe.Prober
	allowedRoots []string
	log          *slog.Logger

	// mu serializes every filesystem mutation across all pairs. Throughput
	// is irrelevant next to the guarantee that two operations never race on
	// the same files.
	mu sync.Mutex

	beforeAction func(ctx context.Context, reason string) error

	jobsMu sync.Mutex
	jobs   map[int64]*Job
	nextID int64
}

func New(st *store.Store, prober probe.Prober, allowedRoots []string, log *slog.Logger) *Engine {
	return &Engine{st: st, prober: prober, allowedRoots: allowedRoots, log: log, jobs: map[int64]*Job{}}
}

func (e *Engine) AllowedRoots() []string { return e.allowedRoots }

// ---- jobs ----

type Job struct {
	ID         int64          `json:"id"`
	PairID     int64          `json:"pair_id"`
	Kind       string         `json:"kind"`
	State      string         `json:"state"` // running, done, failed, cancelled
	Phase      string         `json:"phase"`
	Done       int64          `json:"done"`
	Total      int64          `json:"total"`
	Message    string         `json:"message"`
	Error      string         `json:"error"`
	Results    map[string]int `json:"results"`
	Skipped    []string       `json:"skipped"`
	StartedAt  int64          `json:"started_at"`
	FinishedAt int64          `json:"finished_at,omitempty"`

	cancel context.CancelFunc
	e      *Engine
}

func (j *Job) update(f func(j *Job)) {
	j.e.jobsMu.Lock()
	f(j)
	j.e.jobsMu.Unlock()
}

func (j *Job) phase(p string, total int64) {
	j.update(func(j *Job) { j.Phase, j.Total, j.Done = p, total, 0 })
}

func (j *Job) tick() { j.update(func(j *Job) { j.Done++ }) }

func (j *Job) count(key string) { j.update(func(j *Job) { j.Results[key]++ }) }

func (j *Job) skip(msg string) {
	j.update(func(j *Job) {
		j.Results["skipped"]++
		if len(j.Skipped) < 200 {
			j.Skipped = append(j.Skipped, msg)
		}
	})
}

// Job returns a snapshot of the latest job for a pair.
func (e *Engine) Job(pairID int64) *Job {
	e.jobsMu.Lock()
	defer e.jobsMu.Unlock()
	j, ok := e.jobs[pairID]
	if !ok {
		return nil
	}
	cp := *j
	cp.Results = map[string]int{}
	for k, v := range j.Results {
		cp.Results[k] = v
	}
	cp.Skipped = append([]string{}, j.Skipped...) // never nil: the UI expects an array
	return &cp
}

func (e *Engine) CancelJob(pairID int64) bool {
	e.jobsMu.Lock()
	defer e.jobsMu.Unlock()
	if j, ok := e.jobs[pairID]; ok && j.State == "running" {
		j.cancel()
		return true
	}
	return false
}

func (e *Engine) startJob(pairID int64, kind string, run func(ctx context.Context, j *Job) error) (*Job, error) {
	e.jobsMu.Lock()
	if j, ok := e.jobs[pairID]; ok && j.State == "running" {
		e.jobsMu.Unlock()
		return nil, ErrBusy
	}
	e.nextID++
	ctx, cancel := context.WithCancel(context.Background())
	j := &Job{ID: e.nextID, PairID: pairID, Kind: kind, State: "running", Results: map[string]int{},
		Skipped: []string{}, StartedAt: time.Now().Unix(), cancel: cancel, e: e}
	e.jobs[pairID] = j
	e.jobsMu.Unlock()

	go func() {
		defer cancel()
		err := run(ctx, j)
		j.update(func(j *Job) {
			j.FinishedAt = time.Now().Unix()
			switch {
			case errors.Is(err, context.Canceled):
				j.State, j.Error = "cancelled", "cancelled"
			case err != nil:
				j.State, j.Error = "failed", err.Error()
			default:
				j.State = "done"
			}
		})
		if err != nil {
			e.log.Warn("job finished with error", "pair", pairID, "kind", kind, "err", err)
		} else {
			e.log.Info("job finished", "pair", pairID, "kind", kind)
		}
	}()
	return e.Job(pairID), nil
}

// ---- pair validation ----

type libs struct {
	pair       store.Pair
	master     fsops.Root
	slave      fsops.Root
	quarantine string
	dev        uint64
	exts       map[string]bool
}

func (l *libs) hasExt(name string) bool {
	return l.exts[strings.TrimPrefix(strings.ToLower(filepath.Ext(name)), ".")]
}

func (e *Engine) withinAllowed(p string) bool {
	for _, r := range e.allowedRoots {
		if fsops.Within(r, p) {
			return true
		}
	}
	return false
}

// open resolves a pair's directories and re-checks every invariant. It is
// called before every scan and operation so configuration drift (a remount,
// a moved folder) is caught before anything is touched.
func (e *Engine) open(p store.Pair) (*libs, error) {
	master, err := fsops.NewRoot(p.MasterRoot)
	if err != nil {
		return nil, fmt.Errorf("master library: %w", err)
	}
	slave, err := fsops.NewRoot(p.SlaveRoot)
	if err != nil {
		return nil, fmt.Errorf("slave library: %w", err)
	}
	for _, r := range []string{master.Path, slave.Path} {
		if !e.withinAllowed(r) {
			return nil, fmt.Errorf("%s is outside the allowed roots %v", r, e.allowedRoots)
		}
	}
	if fsops.Within(master.Path, slave.Path) || fsops.Within(slave.Path, master.Path) {
		return nil, errors.New("master and slave libraries must not overlap or contain each other")
	}
	mi, err := fsops.Lstat(master.Path)
	if err != nil {
		return nil, err
	}
	si, err := fsops.Lstat(slave.Path)
	if err != nil {
		return nil, err
	}
	if mi.Dev != si.Dev {
		return nil, fmt.Errorf("master and slave are on different filesystems (device %d vs %d): %w. "+
			"In Docker, mount a single common parent folder instead of two separate volumes", mi.Dev, si.Dev, fsops.ErrCrossDevice)
	}

	q := filepath.Clean(p.QuarantineDir)
	if !filepath.IsAbs(q) {
		return nil, errors.New("quarantine directory must be absolute")
	}
	if fsops.Within(master.Path, q) {
		return nil, errors.New("quarantine directory must not be inside the master library")
	}
	if q == slave.Path || !e.withinAllowed(q) {
		return nil, errors.New("quarantine directory must be inside the allowed roots and not the slave root itself")
	}
	if err := os.MkdirAll(q, 0o775); err != nil {
		return nil, fmt.Errorf("quarantine directory: %w", err)
	}
	// Keep the quarantine hidden from Plex when enabled, including quarantine
	// folders created after the setting was turned on. Never fatal.
	if e.plexIgnoreEnabled() {
		if err := syncPlexIgnore(q, true); err != nil && !errors.Is(err, ErrForeignIgnore) {
			e.log.Warn(".plexignore not written", "err", plexIgnoreError(q, err))
		}
	}
	if rq, err := filepath.EvalSymlinks(q); err != nil || (rq != q && !e.withinAllowed(rq)) {
		return nil, errors.New("quarantine directory resolves outside the allowed roots")
	}
	qi, err := fsops.Lstat(q)
	if err != nil {
		return nil, err
	}
	if qi.Dev != mi.Dev {
		return nil, errors.New("quarantine directory must be on the same filesystem as the libraries")
	}

	exts := map[string]bool{}
	for _, x := range p.Extensions {
		x = strings.TrimPrefix(strings.ToLower(strings.TrimSpace(x)), ".")
		if x != "" {
			exts[x] = true
		}
	}
	if len(exts) == 0 {
		return nil, errors.New("no file extensions configured")
	}
	return &libs{pair: p, master: master, slave: slave, quarantine: q, dev: mi.Dev, exts: exts}, nil
}

// NormalizePair fills defaults and validates a pair. It returns non-fatal
// warnings (such as layout mismatches) for the UI.
func (e *Engine) NormalizePair(p *store.Pair) ([]store.Warning, error) {
	p.Name = strings.TrimSpace(p.Name)
	if p.Name == "" {
		return nil, errors.New("name is required")
	}
	p.MasterRoot = filepath.Clean(strings.TrimSpace(p.MasterRoot))
	p.SlaveRoot = filepath.Clean(strings.TrimSpace(p.SlaveRoot))
	if strings.TrimSpace(p.QuarantineDir) == "" {
		p.QuarantineDir = filepath.Join(p.SlaveRoot, QuarantineDirName)
	}
	p.QuarantineDir = filepath.Clean(strings.TrimSpace(p.QuarantineDir))
	if len(p.Extensions) == 0 || (len(p.Extensions) == 1 && strings.TrimSpace(p.Extensions[0]) == "") {
		p.Extensions = DefaultExtensions
	}
	var clean []string
	for _, x := range p.Extensions {
		x = strings.TrimPrefix(strings.ToLower(strings.TrimSpace(x)), ".")
		if x != "" && !strings.ContainsAny(x, ",/") {
			clean = append(clean, x)
		}
	}
	p.Extensions = clean
	if p.RetentionDays < 0 || p.ScanIntervalMinutes < 0 || p.SettleMinutes < 0 {
		return nil, errors.New("intervals must not be negative")
	}
	if p.ApproxThreshold == 0 {
		p.ApproxThreshold = 85
	}
	if p.ApproxThreshold < 60 || p.ApproxThreshold > 100 {
		return nil, errors.New("approximate match threshold must be between 60 and 100 percent")
	}
	if p.ScanIntervalMinutes > 0 && p.ScanIntervalMinutes < 5 {
		return nil, errors.New("scan interval must be 0 (manual) or at least 5 minutes")
	}
	l, err := e.open(*p)
	if err != nil {
		return nil, err
	}
	// Store resolved paths so later containment checks compare like with like.
	p.MasterRoot, p.SlaveRoot, p.QuarantineDir = l.master.Path, l.slave.Path, l.quarantine
	return layoutWarnings(l), nil
}

// layoutWarnings is the quick check run when a pair is saved and at the
// start of each scan. Content that exists in only one library is normal and
// never warned about; only folders that are the same but named differently are.
func layoutWarnings(l *libs) []store.Warning {
	empty := true
	if ents, _ := os.ReadDir(l.slave.Path); len(ents) > 0 {
		for _, en := range ents {
			if !strings.HasPrefix(en.Name(), ".") && (en.IsDir() || l.hasExt(en.Name())) {
				empty = false
				break
			}
		}
	}
	if empty {
		return []store.Warning{{Title: "The slave library is empty",
			Detail: "Use Build to pick songs from the master library to hardlink into it."}}
	}
	if w := renamedWarning(renamedFolders(l)); w != nil {
		for _, r := range planRenames(l) {
			if r.Ready {
				w.Fixable++
			}
		}
		return []store.Warning{*w}
	}
	return nil
}

// ---- helpers ----

func (e *Engine) loadPair(ctx context.Context, id int64) (*libs, error) {
	p, err := e.st.Pair(ctx, id)
	if err != nil {
		return nil, err
	}
	return e.open(p)
}

func sortedKeys(m map[string]fsops.Info) []string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}

// SetBeforeAction installs the hook that backs up the database before an
// action in live mode.
func (e *Engine) SetBeforeAction(f func(ctx context.Context, reason string) error) {
	e.beforeAction = f
}

// BeforeAction runs the pre-action backup, once per user action (not per
// file). It does nothing in dry run, where no files change. An error means
// the action must not run.
func (e *Engine) BeforeAction(ctx context.Context, reason string) error {
	if e.beforeAction == nil || e.st.DryRun(ctx) {
		return nil
	}
	return e.beforeAction(ctx, reason)
}