// 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/notify" "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 notify func(ctx context.Context, ev notify.Event) emptyMu sync.Mutex empty map[int64]*NoMusicScan // latest scan for folders without music, per pair dupeMu sync.Mutex dupes map[int64]*DupeScan // latest duplicate search, per pair 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 } // SetNotify installs the hook that tells the user about things that // happened without them (see the notify package). func (e *Engine) SetNotify(f func(ctx context.Context, ev notify.Event)) { e.notify = f } func (e *Engine) emit(ctx context.Context, l *libs, ev notify.Event) { if e.notify == nil { return } ev.PairID, ev.Pair = l.pair.ID, l.pair.Name e.notify(ctx, ev) } func plural(n int, one, many string) string { if n == 1 { return "1 " + one } return fmt.Sprintf("%d %s", n, many) } func formatBytes(n int64) string { const unit = 1024 if n < unit { return fmt.Sprintf("%d B", n) } div, exp := int64(unit), 0 for m := n / unit; m >= unit; m /= unit { div *= unit exp++ } return fmt.Sprintf("%.1f %cB", float64(n)/float64(div), "KMGTPE"[exp]) } // 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) }