Initial commit: playlink music library hardlink deduplicator

Go backend (chi, SQLite) with an embedded React/TypeScript UI that finds
songs present in both a host and a slave music library and replaces the
slave copy with a hardlink to the host copy.

- Path-first matching with format/title/album-folder fallbacks
- Auto-linking of byte-identical files, re-verified with cmp before linking
- Manual review with ffprobe quality comparison, downgrade guard and
  promote-to-host override
- Build mode to populate an empty slave library from the host
- Dry run by default, atomic stage+rename replacement, quarantine with
  retention, undo, path confinement and authenticated UI
- Dockerfile and compose file for Unraid

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
cdnutterandClaude Opus 5.5 committed 2026-09-22 15:38:25 -07:00
commit ac47e61b08
42 files changed
+8892

No files matched your search

+281
View File
@@ -0,0 +1,281 @@
package api
import (
"context"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"errors"
"net"
"net/http"
"net/url"
"strings"
"sync"
"time"
"golang.org/x/crypto/bcrypt"
"github.com/cdnutter/playlink/internal/store"
)
const (
cookieName = "playlink_session"
sessionTTL = 30 * 24 * time.Hour
csrfHeader = "X-Playlink-Request"
)
type ctxKey int
const userKey ctxKey = 1
func hashToken(t string) string {
h := sha256.Sum256([]byte(t))
return hex.EncodeToString(h[:])
}
func newToken() (string, error) {
var b [32]byte
if _, err := rand.Read(b[:]); err != nil {
return "", err
}
return base64.RawURLEncoding.EncodeToString(b[:]), nil
}
// limiter is a small per-IP fixed-window limiter for login attempts.
type limiter struct {
mu sync.Mutex
hits map[string][]time.Time
max int
window time.Duration
}
func newLimiter(max int, window time.Duration) *limiter {
return &limiter{hits: map[string][]time.Time{}, max: max, window: window}
}
func (l *limiter) allow(key string) bool {
l.mu.Lock()
defer l.mu.Unlock()
now := time.Now()
kept := l.hits[key][:0]
for _, t := range l.hits[key] {
if now.Sub(t) < l.window {
kept = append(kept, t)
}
}
if len(kept) >= l.max {
l.hits[key] = kept
return false
}
l.hits[key] = append(kept, now)
return true
}
func clientIP(r *http.Request) string {
host, _, err := net.SplitHostPort(r.RemoteAddr)
if err != nil {
return r.RemoteAddr
}
return host
}
func (s *Server) startSession(w http.ResponseWriter, r *http.Request, uid int64) error {
tok, err := newToken()
if err != nil {
return err
}
exp := time.Now().Add(sessionTTL)
if err := s.st.CreateSession(r.Context(), hashToken(tok), uid, exp.Unix()); err != nil {
return err
}
http.SetCookie(w, &http.Cookie{Name: cookieName, Value: tok, Path: "/", Expires: exp, HttpOnly: true,
Secure: s.cfg.SecureCookies, SameSite: http.SameSiteStrictMode})
return nil
}
func (s *Server) currentUser(r *http.Request) (store.User, string, error) {
c, err := r.Cookie(cookieName)
if err != nil || c.Value == "" {
return store.User{}, "", store.ErrNotFound
}
th := hashToken(c.Value)
uid, err := s.st.SessionUser(r.Context(), th)
if err != nil {
return store.User{}, "", err
}
u, err := s.st.UserByID(r.Context(), uid)
return u, th, err
}
// requireAuth rejects unauthenticated requests and, for state-changing
// methods, requests that are not same-origin API calls (CSRF defence in
// addition to the SameSite=Strict cookie).
func (s *Server) requireAuth(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
u, _, err := s.currentUser(r)
if err != nil {
writeErr(w, http.StatusUnauthorized, errors.New("not signed in"))
return
}
next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), userKey, u)))
})
}
func (s *Server) csrf(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet, http.MethodHead, http.MethodOptions:
default:
if r.Header.Get(csrfHeader) != "1" {
writeErr(w, http.StatusForbidden, errors.New("missing request header"))
return
}
if o := r.Header.Get("Origin"); o != "" {
u, err := url.Parse(o)
if err != nil || !strings.EqualFold(u.Host, r.Host) {
writeErr(w, http.StatusForbidden, errors.New("cross-origin request refused"))
return
}
}
}
next.ServeHTTP(w, r)
})
}
type credentials struct {
Username string `json:"username"`
Password string `json:"password"`
}
func validPassword(p string) error {
if len(p) < 10 {
return errors.New("password must be at least 10 characters")
}
if len(p) > 72 {
return errors.New("password must be at most 72 bytes")
}
return nil
}
func (s *Server) handleAuthState(w http.ResponseWriter, r *http.Request) {
n, err := s.st.UserCount(r.Context())
if err != nil {
writeErr(w, 500, err)
return
}
resp := map[string]any{"setup_required": n == 0, "authenticated": false}
if u, _, err := s.currentUser(r); err == nil {
resp["authenticated"] = true
resp["username"] = u.Username
}
writeJSON(w, 200, resp)
}
func (s *Server) handleSetup(w http.ResponseWriter, r *http.Request) {
var c credentials
if !readJSON(w, r, &c) {
return
}
c.Username = strings.TrimSpace(c.Username)
if c.Username == "" {
writeErr(w, 400, errors.New("username is required"))
return
}
if err := validPassword(c.Password); err != nil {
writeErr(w, 400, err)
return
}
hash, err := bcrypt.GenerateFromPassword([]byte(c.Password), 12)
if err != nil {
writeErr(w, 500, err)
return
}
if err := s.st.CreateFirstUser(r.Context(), c.Username, string(hash)); err != nil {
writeErr(w, 409, err)
return
}
u, err := s.st.UserByName(r.Context(), c.Username)
if err != nil {
writeErr(w, 500, err)
return
}
if err := s.startSession(w, r, u.ID); err != nil {
writeErr(w, 500, err)
return
}
s.log.Info("initial user created", "user", u.Username)
writeJSON(w, 200, map[string]any{"ok": true})
}
// dummyHash keeps login timing similar for unknown usernames.
var dummyHash, _ = bcrypt.GenerateFromPassword([]byte("playlink-timing-equalizer"), 12)
func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) {
if !s.loginLimit.allow(clientIP(r)) {
writeErr(w, http.StatusTooManyRequests, errors.New("too many attempts; try again in a few minutes"))
return
}
var c credentials
if !readJSON(w, r, &c) {
return
}
u, err := s.st.UserByName(r.Context(), strings.TrimSpace(c.Username))
hash := dummyHash
if err == nil {
hash = []byte(u.PasswordHash)
}
if bcrypt.CompareHashAndPassword(hash, []byte(c.Password)) != nil || err != nil {
s.log.Warn("failed login", "ip", clientIP(r))
writeErr(w, http.StatusUnauthorized, errors.New("invalid username or password"))
return
}
if err := s.startSession(w, r, u.ID); err != nil {
writeErr(w, 500, err)
return
}
writeJSON(w, 200, map[string]any{"ok": true})
}
func (s *Server) handleLogout(w http.ResponseWriter, r *http.Request) {
if c, err := r.Cookie(cookieName); err == nil {
_ = s.st.DeleteSession(r.Context(), hashToken(c.Value))
}
http.SetCookie(w, &http.Cookie{Name: cookieName, Value: "", Path: "/", MaxAge: -1, HttpOnly: true,
Secure: s.cfg.SecureCookies, SameSite: http.SameSiteStrictMode})
writeJSON(w, 200, map[string]any{"ok": true})
}
func (s *Server) handleChangePassword(w http.ResponseWriter, r *http.Request) {
var body struct {
Current string `json:"current"`
New string `json:"new"`
}
if !readJSON(w, r, &body) {
return
}
u, th, err := s.currentUser(r)
if err != nil {
writeErr(w, 401, err)
return
}
if bcrypt.CompareHashAndPassword([]byte(u.PasswordHash), []byte(body.Current)) != nil {
writeErr(w, 400, errors.New("current password is incorrect"))
return
}
if err := validPassword(body.New); err != nil {
writeErr(w, 400, err)
return
}
hash, err := bcrypt.GenerateFromPassword([]byte(body.New), 12)
if err != nil {
writeErr(w, 500, err)
return
}
if err := s.st.SetPassword(r.Context(), u.ID, string(hash)); err != nil {
writeErr(w, 500, err)
return
}
_ = s.st.DeleteUserSessions(r.Context(), u.ID, th) // sign out other devices
writeJSON(w, 200, map[string]any{"ok": true})
}
+561
View File
@@ -0,0 +1,561 @@
// Package api exposes the playlink HTTP API and serves the web UI.
package api
import (
"encoding/json"
"errors"
"io"
"io/fs"
"log/slog"
"net/http"
"os"
"path"
"path/filepath"
"sort"
"strconv"
"strings"
"time"
"github.com/go-chi/chi/v5"
"github.com/go-chi/chi/v5/middleware"
"github.com/cdnutter/playlink/internal/engine"
"github.com/cdnutter/playlink/internal/fsops"
"github.com/cdnutter/playlink/internal/store"
)
type Config struct {
SecureCookies bool
Version string
}
type Server struct {
cfg Config
st *store.Store
eng *engine.Engine
log *slog.Logger
web fs.FS
loginLimit *limiter
}
func New(cfg Config, st *store.Store, eng *engine.Engine, log *slog.Logger, web fs.FS) *Server {
return &Server{cfg: cfg, st: st, eng: eng, log: log, web: web, loginLimit: newLimiter(10, 5*time.Minute)}
}
func (s *Server) Handler() http.Handler {
r := chi.NewRouter()
r.Use(middleware.Recoverer)
r.Use(securityHeaders)
r.Get("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) })
r.Route("/api", func(r chi.Router) {
r.Use(s.csrf)
r.Use(func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
r.Body = http.MaxBytesReader(w, r.Body, 1<<20)
w.Header().Set("Cache-Control", "no-store")
next.ServeHTTP(w, r)
})
})
r.Get("/auth/state", s.handleAuthState)
r.Post("/auth/setup", s.handleSetup)
r.Post("/auth/login", s.handleLogin)
r.Post("/auth/logout", s.handleLogout)
r.Group(func(r chi.Router) {
r.Use(s.requireAuth)
r.Post("/auth/password", s.handleChangePassword)
r.Get("/info", s.handleInfo)
r.Get("/settings", s.handleGetSettings)
r.Put("/settings", s.handlePutSettings)
r.Get("/browse", s.handleBrowse)
r.Get("/pairs", s.handleListPairs)
r.Post("/pairs", s.handleSavePair)
r.Post("/pairs/validate", s.handleValidatePair)
r.Route("/pairs/{id}", func(r chi.Router) {
r.Get("/", s.handleGetPair)
r.Put("/", s.handleSavePair)
r.Delete("/", s.handleDeletePair)
r.Post("/scan", s.handleScan)
r.Get("/job", s.handleJob)
r.Post("/job/cancel", s.handleCancelJob)
r.Get("/matches", s.handleMatches)
r.Post("/link-identical", s.handleLinkIdentical)
r.Get("/tree", s.handleTree)
r.Post("/build", s.handleBuild)
r.Post("/remove", s.handleRemove)
})
r.Post("/matches/{id}/link", s.handleLink)
r.Post("/matches/{id}/promote", s.handlePromote)
r.Post("/matches/{id}/ignore", s.handleIgnore)
r.Get("/operations", s.handleOps)
r.Post("/operations/{id}/undo", s.handleUndo)
r.Post("/operations/{id}/purge", s.handlePurge)
})
})
r.NotFound(s.serveUI)
return r
}
func securityHeaders(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
h := w.Header()
h.Set("X-Content-Type-Options", "nosniff")
h.Set("X-Frame-Options", "DENY")
h.Set("Referrer-Policy", "same-origin")
h.Set("Content-Security-Policy", "default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; "+
"script-src 'self'; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
next.ServeHTTP(w, r)
})
}
// serveUI serves the embedded single-page app, falling back to index.html.
func (s *Server) serveUI(w http.ResponseWriter, r *http.Request) {
if strings.HasPrefix(r.URL.Path, "/api/") {
writeErr(w, 404, errors.New("not found"))
return
}
p := strings.TrimPrefix(path.Clean(r.URL.Path), "/")
if p == "" {
p = "index.html"
}
if _, err := fs.Stat(s.web, p); err != nil {
p = "index.html"
}
if strings.HasPrefix(p, "assets/") {
w.Header().Set("Cache-Control", "public, max-age=31536000, immutable")
} else {
w.Header().Set("Cache-Control", "no-cache")
}
http.ServeFileFS(w, r, s.web, p)
}
// ---- helpers ----
func writeJSON(w http.ResponseWriter, code int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
_ = json.NewEncoder(w).Encode(v)
}
func writeErr(w http.ResponseWriter, code int, err error) {
writeJSON(w, code, map[string]string{"error": err.Error()})
}
func readJSON(w http.ResponseWriter, r *http.Request, v any) bool {
dec := json.NewDecoder(r.Body)
dec.DisallowUnknownFields()
if err := dec.Decode(v); err != nil && !errors.Is(err, io.EOF) {
writeErr(w, 400, errors.New("invalid request body: "+err.Error()))
return false
}
return true
}
func idParam(r *http.Request) (int64, bool) {
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
return id, err == nil && id > 0
}
// opErr maps engine errors onto HTTP status codes.
func opErr(w http.ResponseWriter, err error) {
switch {
case errors.Is(err, store.ErrNotFound):
writeErr(w, 404, err)
case errors.Is(err, engine.ErrBusy), errors.Is(err, engine.ErrDowngrade), errors.Is(err, engine.ErrConfirm),
errors.Is(err, fsops.ErrChanged), errors.Is(err, fsops.ErrExists):
writeErr(w, 409, err)
case errors.Is(err, fsops.ErrOutsideRoot), errors.Is(err, fsops.ErrNotRegular), errors.Is(err, fsops.ErrCrossDevice):
writeErr(w, 400, err)
default:
writeErr(w, 500, err)
}
}
// ---- info & settings ----
func (s *Server) handleInfo(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, map[string]any{
"version": s.cfg.Version,
"allowed_roots": s.eng.AllowedRoots(),
"default_extensions": engine.DefaultExtensions,
"dry_run": s.st.DryRun(r.Context()),
})
}
func (s *Server) handleGetSettings(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, map[string]any{"dry_run": s.st.DryRun(r.Context())})
}
func (s *Server) handlePutSettings(w http.ResponseWriter, r *http.Request) {
var body struct {
DryRun *bool `json:"dry_run"`
}
if !readJSON(w, r, &body) {
return
}
if body.DryRun != nil {
v := "0"
if *body.DryRun {
v = "1"
}
if err := s.st.SetSetting(r.Context(), "dry_run", v); err != nil {
writeErr(w, 500, err)
return
}
s.log.Info("dry run changed", "dry_run", *body.DryRun)
}
s.handleGetSettings(w, r)
}
// handleBrowse lists subdirectories for the folder picker, confined to the
// allowed roots.
func (s *Server) handleBrowse(w http.ResponseWriter, r *http.Request) {
p := r.URL.Query().Get("path")
roots := s.eng.AllowedRoots()
if p == "" {
writeJSON(w, 200, map[string]any{"path": "", "parent": "", "dirs": roots})
return
}
p = filepath.Clean(p)
real, err := filepath.EvalSymlinks(p)
if err != nil {
writeErr(w, 400, err)
return
}
ok := false
for _, root := range roots {
if fsops.Within(root, real) {
ok = true
}
}
if !ok {
writeErr(w, 403, fsops.ErrOutsideRoot)
return
}
ents, err := os.ReadDir(real)
if err != nil {
writeErr(w, 400, err)
return
}
dirs := []string{}
for _, en := range ents {
if en.IsDir() && !strings.HasPrefix(en.Name(), ".") {
dirs = append(dirs, filepath.Join(real, en.Name()))
}
}
sort.Strings(dirs)
parent := filepath.Dir(real)
for _, root := range roots {
if real == root {
parent = ""
}
}
writeJSON(w, 200, map[string]any{"path": real, "parent": parent, "dirs": dirs})
}
// ---- pairs ----
type pairView struct {
store.Pair
Summary store.Summary `json:"summary"`
Job *engine.Job `json:"job"`
}
func (s *Server) view(r *http.Request, p store.Pair) pairView {
sum, _ := s.st.Summary(r.Context(), p.ID)
return pairView{Pair: p, Summary: sum, Job: s.eng.Job(p.ID)}
}
func (s *Server) handleListPairs(w http.ResponseWriter, r *http.Request) {
pairs, err := s.st.Pairs(r.Context())
if err != nil {
writeErr(w, 500, err)
return
}
out := make([]pairView, 0, len(pairs))
for _, p := range pairs {
out = append(out, s.view(r, p))
}
writeJSON(w, 200, out)
}
func (s *Server) handleGetPair(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
p, err := s.st.Pair(r.Context(), id)
if err != nil {
opErr(w, err)
return
}
writeJSON(w, 200, s.view(r, p))
}
type pairInput struct {
Name string `json:"name"`
HostRoot string `json:"host_root"`
SlaveRoot string `json:"slave_root"`
QuarantineDir string `json:"quarantine_dir"`
Extensions []string `json:"extensions"`
AutoLinkIdentical bool `json:"auto_link_identical"`
ScanIntervalMinutes int `json:"scan_interval_minutes"`
RetentionDays int `json:"retention_days"`
SettleMinutes int `json:"settle_minutes"`
}
func (in pairInput) apply(p *store.Pair) {
p.Name, p.HostRoot, p.SlaveRoot, p.QuarantineDir = in.Name, in.HostRoot, in.SlaveRoot, in.QuarantineDir
p.Extensions, p.AutoLinkIdentical = in.Extensions, in.AutoLinkIdentical
p.ScanIntervalMinutes, p.RetentionDays, p.SettleMinutes = in.ScanIntervalMinutes, in.RetentionDays, in.SettleMinutes
}
func (s *Server) handleValidatePair(w http.ResponseWriter, r *http.Request) {
var in pairInput
if !readJSON(w, r, &in) {
return
}
var p store.Pair
in.apply(&p)
warns, err := s.eng.NormalizePair(&p)
if err != nil {
writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error(), "warnings": []string{}})
return
}
if warns == nil {
warns = []string{}
}
writeJSON(w, 200, map[string]any{"ok": true, "warnings": warns, "pair": p})
}
func (s *Server) handleSavePair(w http.ResponseWriter, r *http.Request) {
var in pairInput
if !readJSON(w, r, &in) {
return
}
var p store.Pair
var old store.Pair
if id, ok := idParam(r); ok {
var err error
if old, err = s.st.Pair(r.Context(), id); err != nil {
opErr(w, err)
return
}
p = old
if j := s.eng.Job(id); j != nil && j.State == "running" {
writeErr(w, 409, engine.ErrBusy)
return
}
}
in.apply(&p)
warns, err := s.eng.NormalizePair(&p)
if err != nil {
writeErr(w, 400, err)
return
}
if err := s.st.SavePair(r.Context(), &p); err != nil {
writeErr(w, 500, err)
return
}
if old.ID != 0 && (old.HostRoot != p.HostRoot || old.SlaveRoot != p.SlaveRoot) {
_ = s.st.ClearMatches(r.Context(), p.ID) // results no longer apply
}
p, _ = s.st.Pair(r.Context(), p.ID)
if warns == nil {
warns = []string{}
}
writeJSON(w, 200, map[string]any{"pair": s.view(r, p), "warnings": warns})
}
func (s *Server) handleDeletePair(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
if j := s.eng.Job(id); j != nil && j.State == "running" {
writeErr(w, 409, engine.ErrBusy)
return
}
// Only configuration and history are removed; files are never touched.
if err := s.st.DeletePair(r.Context(), id); err != nil {
writeErr(w, 500, err)
return
}
writeJSON(w, 200, map[string]any{"ok": true})
}
func (s *Server) handleScan(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
if _, err := s.st.Pair(r.Context(), id); err != nil {
opErr(w, err)
return
}
j, err := s.eng.StartScan(id, "manual")
if err != nil {
opErr(w, err)
return
}
writeJSON(w, 202, j)
}
func (s *Server) handleJob(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
writeJSON(w, 200, s.eng.Job(id))
}
func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
writeJSON(w, 200, map[string]any{"cancelled": s.eng.CancelJob(id)})
}
func (s *Server) handleMatches(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
q := r.URL.Query()
limit, _ := strconv.Atoi(q.Get("limit"))
offset, _ := strconv.Atoi(q.Get("offset"))
ms, total, err := s.st.Matches(r.Context(), id, store.MatchFilter{Status: q.Get("status"), Query: q.Get("q"), Limit: limit, Offset: offset})
if err != nil {
writeErr(w, 500, err)
return
}
writeJSON(w, 200, map[string]any{"items": ms, "total": total})
}
func (s *Server) handleLinkIdentical(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
j, err := s.eng.StartLinkIdentical(id)
if err != nil {
opErr(w, err)
return
}
writeJSON(w, 202, j)
}
func (s *Server) handleTree(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
ents, err := s.eng.Tree(r.Context(), id, r.URL.Query().Get("path"))
if err != nil {
opErr(w, err)
return
}
writeJSON(w, 200, ents)
}
type pathsBody struct {
Paths []string `json:"paths"`
}
func (s *Server) handleBuild(w http.ResponseWriter, r *http.Request) {
s.pathsJob(w, r, s.eng.StartBuild)
}
func (s *Server) handleRemove(w http.ResponseWriter, r *http.Request) {
s.pathsJob(w, r, s.eng.StartRemove)
}
func (s *Server) pathsJob(w http.ResponseWriter, r *http.Request, start func(int64, []string) (*engine.Job, error)) {
id, _ := idParam(r)
var body pathsBody
if !readJSON(w, r, &body) {
return
}
if len(body.Paths) == 0 {
writeErr(w, 400, errors.New("no paths selected"))
return
}
for _, p := range body.Paths {
if _, err := fsops.CleanRel(p); err != nil {
writeErr(w, 400, err)
return
}
}
j, err := start(id, body.Paths)
if err != nil {
opErr(w, err)
return
}
writeJSON(w, 202, j)
}
// ---- match actions ----
func (s *Server) handleLink(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
var body struct {
AllowDowngrade bool `json:"allow_downgrade"`
}
if !readJSON(w, r, &body) {
return
}
op, err := s.eng.Link(r.Context(), id, engine.LinkOptions{AllowDowngrade: body.AllowDowngrade})
if err != nil {
opErr(w, err)
return
}
writeJSON(w, 200, op)
}
func (s *Server) handlePromote(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
var body struct {
Confirm bool `json:"confirm"`
}
if !readJSON(w, r, &body) {
return
}
op, err := s.eng.Promote(r.Context(), id, body.Confirm)
if err != nil {
opErr(w, err)
return
}
writeJSON(w, 200, op)
}
func (s *Server) handleIgnore(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
var body struct {
Ignored bool `json:"ignored"`
}
if !readJSON(w, r, &body) {
return
}
if err := s.st.SetIgnored(r.Context(), id, body.Ignored); err != nil {
writeErr(w, 500, err)
return
}
writeJSON(w, 200, map[string]any{"ok": true})
}
// ---- operations ----
func (s *Server) handleOps(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
pairID, _ := strconv.ParseInt(q.Get("pair"), 10, 64)
limit, _ := strconv.Atoi(q.Get("limit"))
offset, _ := strconv.Atoi(q.Get("offset"))
ops, total, err := s.st.Ops(r.Context(), store.OpFilter{PairID: pairID, Quarantine: q.Get("quarantine") == "1", Limit: limit, Offset: offset})
if err != nil {
writeErr(w, 500, err)
return
}
writeJSON(w, 200, map[string]any{"items": ops, "total": total})
}
func (s *Server) handleUndo(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
if err := s.eng.Undo(r.Context(), id); err != nil {
opErr(w, err)
return
}
writeJSON(w, 200, map[string]any{"ok": true})
}
func (s *Server) handlePurge(w http.ResponseWriter, r *http.Request) {
id, _ := idParam(r)
if err := s.eng.Purge(r.Context(), id); err != nil {
opErr(w, err)
return
}
writeJSON(w, 200, map[string]any{"ok": true})
}
+352
View File
@@ -0,0 +1,352 @@
// 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 host copy is lower quality than the slave copy; confirm the downgrade to continue")
ErrConfirm = errors.New("this action modifies the host 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
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
host 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) {
host, err := fsops.NewRoot(p.HostRoot)
if err != nil {
return nil, fmt.Errorf("host library: %w", err)
}
slave, err := fsops.NewRoot(p.SlaveRoot)
if err != nil {
return nil, fmt.Errorf("slave library: %w", err)
}
for _, r := range []string{host.Path, slave.Path} {
if !e.withinAllowed(r) {
return nil, fmt.Errorf("%s is outside the allowed roots %v", r, e.allowedRoots)
}
}
if fsops.Within(host.Path, slave.Path) || fsops.Within(slave.Path, host.Path) {
return nil, errors.New("host and slave libraries must not overlap or contain each other")
}
hi, err := fsops.Lstat(host.Path)
if err != nil {
return nil, err
}
si, err := fsops.Lstat(slave.Path)
if err != nil {
return nil, err
}
if hi.Dev != si.Dev {
return nil, fmt.Errorf("host and slave are on different filesystems (device %d vs %d): %w. "+
"In Docker, mount a single common parent folder instead of two separate volumes", hi.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(host.Path, q) {
return nil, errors.New("quarantine directory must not be inside the host 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)
}
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 != hi.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, host: host, slave: slave, quarantine: q, dev: hi.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) ([]string, error) {
p.Name = strings.TrimSpace(p.Name)
if p.Name == "" {
return nil, errors.New("name is required")
}
p.HostRoot = filepath.Clean(strings.TrimSpace(p.HostRoot))
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.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.HostRoot, p.SlaveRoot, p.QuarantineDir = l.host.Path, l.slave.Path, l.quarantine
return layoutWarnings(l), nil
}
// layoutWarnings does a cheap top-level comparison of the two libraries.
func layoutWarnings(l *libs) []string {
var warns []string
list := func(root string) (dirs map[string]bool, looseFiles int) {
dirs = map[string]bool{}
ents, _ := os.ReadDir(root)
for _, en := range ents {
if strings.HasPrefix(en.Name(), ".") {
continue
}
if en.IsDir() {
dirs[strings.ToLower(en.Name())] = true
} else if l.hasExt(en.Name()) {
looseFiles++
}
}
return
}
hd, hl := list(l.host.Path)
sd, sl := list(l.slave.Path)
if len(sd) == 0 && sl == 0 {
warns = append(warns, "The slave library is empty. Use Build to pick songs from the host library to hardlink into it.")
return warns
}
if hl > 0 || sl > 0 {
warns = append(warns, fmt.Sprintf("Found audio files directly in a library root (host: %d, slave: %d). Both libraries should use Artist/Album/Title.ext.", hl, sl))
}
shared := 0
for d := range sd {
if hd[d] {
shared++
}
}
if len(sd) > 0 && shared*2 < len(sd) {
warns = append(warns, fmt.Sprintf("Only %d of %d artist folders in the slave library also exist in the host library. "+
"The two libraries should share the same Artist/Album/Title.ext layout, or most files will not match.", shared, len(sd)))
}
return warns
}
// ---- 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
}
+481
View File
@@ -0,0 +1,481 @@
package engine
import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
"log/slog"
"os"
"path/filepath"
"testing"
"time"
"github.com/cdnutter/playlink/internal/fsops"
"github.com/cdnutter/playlink/internal/probe"
"github.com/cdnutter/playlink/internal/store"
)
type env struct {
t *testing.T
root string
host string
slave string
st *store.Store
e *Engine
pair store.Pair
}
func setup(t *testing.T, dryRun bool) *env {
t.Helper()
root, err := filepath.EvalSymlinks(t.TempDir())
if err != nil {
t.Fatal(err)
}
ev := &env{t: t, root: root, host: filepath.Join(root, "Albums"), slave: filepath.Join(root, "Singles")}
for _, d := range []string{ev.host, ev.slave, filepath.Join(root, "db")} {
if err := os.MkdirAll(d, 0o755); err != nil {
t.Fatal(err)
}
}
st, err := store.Open(filepath.Join(root, "db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { st.DB.Close() })
v := "0"
if dryRun {
v = "1"
}
if err := st.SetSetting(context.Background(), "dry_run", v); err != nil {
t.Fatal(err)
}
ev.st = st
ev.e = New(st, probe.Prober{}, []string{root}, slog.New(slog.NewTextHandler(io.Discard, nil)))
ev.pair = store.Pair{Name: "test", HostRoot: ev.host, SlaveRoot: ev.slave, RetentionDays: 30}
if _, err := ev.e.NormalizePair(&ev.pair); err != nil {
t.Fatal(err)
}
if err := st.SavePair(context.Background(), &ev.pair); err != nil {
t.Fatal(err)
}
return ev
}
func (ev *env) write(base, rel, content string) string {
ev.t.Helper()
p := filepath.Join(base, rel)
if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
ev.t.Fatal(err)
}
if err := os.WriteFile(p, []byte(content), 0o644); err != nil {
ev.t.Fatal(err)
}
return p
}
func (ev *env) read(p string) string {
ev.t.Helper()
b, err := os.ReadFile(p)
if err != nil {
ev.t.Fatal(err)
}
return string(b)
}
func (ev *env) scan() {
ev.t.Helper()
if _, err := ev.e.StartScan(ev.pair.ID, "test"); err != nil {
ev.t.Fatal(err)
}
ev.wait()
}
func (ev *env) wait() *Job {
ev.t.Helper()
deadline := time.Now().Add(10 * time.Second)
for time.Now().Before(deadline) {
if j := ev.e.Job(ev.pair.ID); j != nil && j.State != "running" {
if j.State != "done" {
ev.t.Fatalf("job %s: %s", j.State, j.Error)
}
return j
}
time.Sleep(10 * time.Millisecond)
}
ev.t.Fatal("job did not finish")
return nil
}
func (ev *env) match(rel string) store.Match {
ev.t.Helper()
m, err := ev.st.MatchBySlave(context.Background(), ev.pair.ID, rel)
if err != nil {
ev.t.Fatalf("match %s: %v", rel, err)
}
return m
}
func sameInode(t *testing.T, a, b string) bool {
t.Helper()
ia, err := fsops.Lstat(a)
if err != nil {
t.Fatal(err)
}
ib, err := fsops.Lstat(b)
if err != nil {
t.Fatal(err)
}
return ia.SameInode(ib)
}
var ctx = context.Background()
func TestIdenticalLinkAndUndo(t *testing.T) {
ev := setup(t, false)
rel := "bbno$/Bag or Die/edamame.flac"
h := ev.write(ev.host, rel, "same-bytes")
s := ev.write(ev.slave, rel, "same-bytes")
ev.scan()
m := ev.match(rel)
if m.Status != store.StatusIdentical || m.Kind != store.KindPath {
t.Fatalf("want identical/path, got %s/%s", m.Status, m.Kind)
}
op, err := ev.e.Link(ctx, m.ID, LinkOptions{})
if err != nil {
t.Fatal(err)
}
if op.QuarantinePath != "" {
t.Fatal("identical link must not quarantine")
}
if !sameInode(t, h, s) {
t.Fatal("slave is not a hardlink of host")
}
if ev.match(rel).Status != store.StatusLinked {
t.Fatal("match not marked linked")
}
if err := ev.e.Undo(ctx, op.ID); err != nil {
t.Fatal(err)
}
if sameInode(t, h, s) || ev.read(s) != "same-bytes" {
t.Fatal("undo should leave an independent identical copy")
}
}
func TestDifferentLinkQuarantinesAndUndoRestores(t *testing.T) {
ev := setup(t, false)
rel := "Artist/Album/song.flac"
h := ev.write(ev.host, rel, "host-version-longer")
s := ev.write(ev.slave, rel, "slave-version")
ev.scan()
m := ev.match(rel)
if m.Status != store.StatusDifferent {
t.Fatalf("want different, got %s", m.Status)
}
op, err := ev.e.Link(ctx, m.ID, LinkOptions{})
if err != nil {
t.Fatal(err)
}
if !sameInode(t, h, s) {
t.Fatal("not linked")
}
if op.QuarantinePath == "" || ev.read(op.QuarantinePath) != "slave-version" {
t.Fatal("original slave file not quarantined")
}
if err := ev.e.Undo(ctx, op.ID); err != nil {
t.Fatal(err)
}
if ev.read(s) != "slave-version" || ev.read(h) != "host-version-longer" {
t.Fatal("undo did not restore the original")
}
if _, err := os.Stat(op.QuarantinePath); !errors.Is(err, os.ErrNotExist) {
t.Fatal("quarantine entry should be consumed by the restore")
}
if err := ev.e.Undo(ctx, op.ID); err == nil {
t.Fatal("double undo must fail")
}
}
func TestSimilarExtensionLinkRenamesAndUndo(t *testing.T) {
ev := setup(t, false)
h := ev.write(ev.host, "A/B/01 - Song.flac", "flac-data")
s := ev.write(ev.slave, "A/B/01 - Song.mp3", "mp3")
ev.scan()
m := ev.match("A/B/01 - Song.mp3")
if m.Status != store.StatusSimilar || m.Kind != store.KindExtension || m.RelHost != "A/B/01 - Song.flac" {
t.Fatalf("unexpected match %+v", m)
}
op, err := ev.e.Link(ctx, m.ID, LinkOptions{})
if err != nil {
t.Fatal(err)
}
newSlave := filepath.Join(ev.slave, "A/B/01 - Song.flac")
if !sameInode(t, h, newSlave) {
t.Fatal("slave should now hold the host's .flac as a hardlink")
}
if _, err := os.Stat(s); !errors.Is(err, os.ErrNotExist) {
t.Fatal("old .mp3 should be moved to quarantine")
}
if err := ev.e.Undo(ctx, op.ID); err != nil {
t.Fatal(err)
}
if ev.read(s) != "mp3" {
t.Fatal(".mp3 not restored")
}
if _, err := os.Stat(newSlave); !errors.Is(err, os.ErrNotExist) {
t.Fatal("linked .flac should be removed on undo")
}
if ev.read(h) != "flac-data" {
t.Fatal("host must be untouched")
}
}
func TestFuzzyTitleMatch(t *testing.T) {
ev := setup(t, false)
ev.write(ev.host, "A/B/03. My Song!.flac", "x")
ev.write(ev.slave, "A/B/My Song.mp3", "y")
ev.write(ev.slave, "A/B/My Song (Live).mp3", "z")
ev.write(ev.host, "A/Other Album/Lonely.flac", "x")
ev.write(ev.slave, "A/Single/Lonely.flac", "x")
ev.scan()
if m := ev.match("A/B/My Song.mp3"); m.Kind != store.KindFuzzy {
t.Fatalf("want fuzzy, got %s", m.Kind)
}
if m := ev.match("A/B/My Song (Live).mp3"); m.Status != store.StatusUnmatched {
t.Fatalf("live version must not match studio version, got %s", m.Status)
}
if m := ev.match("A/Single/Lonely.flac"); m.Kind != store.KindElsewhere || m.Status != store.StatusSimilar {
t.Fatalf("want elsewhere/similar, got %s/%s", m.Kind, m.Status)
}
}
func TestDowngradeGuardAndPromote(t *testing.T) {
ev := setup(t, false)
rel := "A/B/song.flac"
h := ev.write(ev.host, rel, "lossy-host")
s := ev.write(ev.slave, rel, "lossless-slave!")
ev.scan()
m := ev.match(rel)
// Simulate the probe verdict that the slave is better.
v, _ := json.Marshal(probe.Verdict{Better: "slave"})
m.Verdict = v
if err := ev.st.UpsertMatch(ctx, &m); err != nil {
t.Fatal(err)
}
if _, err := ev.e.Link(ctx, m.ID, LinkOptions{}); !errors.Is(err, ErrDowngrade) {
t.Fatalf("want ErrDowngrade, got %v", err)
}
if _, err := ev.e.Promote(ctx, m.ID, false); !errors.Is(err, ErrConfirm) {
t.Fatalf("promote without confirm must fail, got %v", err)
}
op, err := ev.e.Promote(ctx, m.ID, true)
if err != nil {
t.Fatal(err)
}
if !sameInode(t, h, s) || ev.read(h) != "lossless-slave!" {
t.Fatal("host should now be the slave's file")
}
if ev.read(op.QuarantinePath) != "lossy-host" {
t.Fatal("host original not quarantined")
}
if err := ev.e.Undo(ctx, op.ID); err != nil {
t.Fatal(err)
}
if ev.read(h) != "lossy-host" || ev.read(s) != "lossless-slave!" || sameInode(t, h, s) {
t.Fatal("promote undo failed")
}
}
func TestChangedSinceScanIsRefused(t *testing.T) {
ev := setup(t, false)
rel := "A/B/c.flac"
ev.write(ev.host, rel, "aaa")
s := ev.write(ev.slave, rel, "bbb")
ev.scan()
m := ev.match(rel)
time.Sleep(10 * time.Millisecond)
if err := os.WriteFile(s, []byte("user edited this"), 0o644); err != nil {
t.Fatal(err)
}
if _, err := ev.e.Link(ctx, m.ID, LinkOptions{}); !errors.Is(err, fsops.ErrChanged) {
t.Fatalf("want ErrChanged, got %v", err)
}
if ev.read(s) != "user edited this" {
t.Fatal("file must be untouched")
}
}
func TestDryRunChangesNothing(t *testing.T) {
ev := setup(t, true)
rel := "A/B/d.flac"
h := ev.write(ev.host, rel, "same")
s := ev.write(ev.slave, rel, "same")
ev.scan()
op, err := ev.e.Link(ctx, ev.match(rel).ID, LinkOptions{})
if err != nil {
t.Fatal(err)
}
if !op.DryRun || op.Status != store.OpPlanned {
t.Fatalf("expected planned dry-run op, got %+v", op)
}
if sameInode(t, h, s) {
t.Fatal("dry run must not link")
}
}
func TestAutoLinkRespectsSettleTime(t *testing.T) {
ev := setup(t, false)
ev.pair.AutoLinkIdentical = true
ev.pair.SettleMinutes = 60
if err := ev.st.SavePair(ctx, &ev.pair); err != nil {
t.Fatal(err)
}
h := ev.write(ev.host, "A/B/e.flac", "same")
s := ev.write(ev.slave, "A/B/e.flac", "same")
ev.scan()
if sameInode(t, h, s) {
t.Fatal("freshly written files must not be auto-linked")
}
ev.pair.SettleMinutes = 0
if err := ev.st.SavePair(ctx, &ev.pair); err != nil {
t.Fatal(err)
}
ev.scan()
if !sameInode(t, h, s) {
t.Fatal("settled identical file should be auto-linked")
}
}
func TestBuildAndRemove(t *testing.T) {
ev := setup(t, false)
h1 := ev.write(ev.host, "bbno$/Album/one.flac", "1")
ev.write(ev.host, "bbno$/Album/two.flac", "2")
ev.write(ev.host, "bbno$/Album/cover.jpg", "img")
ev.write(ev.slave, "bbno$/Album/two.flac", "user's own different file")
if _, err := ev.e.StartBuild(ev.pair.ID, []string{"bbno$"}); err != nil {
t.Fatal(err)
}
j := ev.wait()
if j.Results["linked"] != 1 || j.Results["skipped"] != 1 {
t.Fatalf("unexpected results %+v", j.Results)
}
s1 := filepath.Join(ev.slave, "bbno$/Album/one.flac")
if !sameInode(t, h1, s1) {
t.Fatal("build did not hardlink")
}
if ev.read(filepath.Join(ev.slave, "bbno$/Album/two.flac")) != "user's own different file" {
t.Fatal("build must never overwrite an existing slave file")
}
if _, err := os.Stat(filepath.Join(ev.slave, "bbno$/Album/cover.jpg")); err == nil {
t.Fatal("non-audio files must not be linked")
}
if _, err := ev.e.StartRemove(ev.pair.ID, []string{"bbno$"}); err != nil {
t.Fatal(err)
}
j = ev.wait()
if j.Results["removed"] != 1 || j.Results["skipped"] != 1 {
t.Fatalf("unexpected remove results %+v", j.Results)
}
if _, err := os.Stat(s1); !errors.Is(err, os.ErrNotExist) {
t.Fatal("linked file should be removed")
}
if ev.read(h1) != "1" {
t.Fatal("host must be untouched")
}
if ev.read(filepath.Join(ev.slave, "bbno$/Album/two.flac")) != "user's own different file" {
t.Fatal("remove must never delete a non-linked file")
}
}
func TestSymlinkEscapeIgnored(t *testing.T) {
ev := setup(t, false)
outside := t.TempDir()
os.WriteFile(filepath.Join(outside, "secret.flac"), []byte("x"), 0o644)
os.MkdirAll(filepath.Join(ev.slave, "Evil"), 0o755)
if err := os.Symlink(outside, filepath.Join(ev.slave, "Evil", "Album")); err != nil {
t.Fatal(err)
}
ev.write(ev.host, "Evil/Album/secret.flac", "x")
ev.scan()
if _, err := ev.st.MatchBySlave(ctx, ev.pair.ID, "Evil/Album/secret.flac"); err == nil {
t.Fatal("files behind a symlinked directory must not be scanned")
}
root, _ := fsops.NewRoot(ev.slave)
if _, err := root.Resolve("Evil/Album/secret.flac"); !errors.Is(err, fsops.ErrOutsideRoot) {
t.Fatalf("resolve through an escaping symlink must fail, got %v", err)
}
if _, err := root.Resolve("../Albums/x.flac"); !errors.Is(err, fsops.ErrOutsideRoot) {
t.Fatalf("dot-dot must fail, got %v", err)
}
}
func TestPurgeAndOverlapValidation(t *testing.T) {
ev := setup(t, false)
rel := "A/B/f.flac"
ev.write(ev.host, rel, "one")
ev.write(ev.slave, rel, "two!")
ev.scan()
op, err := ev.e.Link(ctx, ev.match(rel).ID, LinkOptions{})
if err != nil {
t.Fatal(err)
}
if err := ev.e.Purge(ctx, op.ID); err != nil {
t.Fatal(err)
}
if _, err := os.Stat(op.QuarantinePath); !errors.Is(err, os.ErrNotExist) {
t.Fatal("purge did not delete quarantine")
}
if err := ev.e.Undo(ctx, op.ID); err == nil {
t.Fatal("undo after purge must fail")
}
bad := store.Pair{Name: "bad", HostRoot: ev.host, SlaveRoot: filepath.Join(ev.host, "A")}
if _, err := ev.e.NormalizePair(&bad); err == nil {
t.Fatal("nested libraries must be rejected")
}
outside := store.Pair{Name: "bad", HostRoot: ev.host, SlaveRoot: t.TempDir()}
if _, err := ev.e.NormalizePair(&outside); err == nil {
t.Fatal("paths outside allowed roots must be rejected")
}
}
func TestSameContent(t *testing.T) {
dir := t.TempDir()
big := bytes.Repeat([]byte("abcdefgh"), 300_000) // spans several chunks
a, b, c := filepath.Join(dir, "a"), filepath.Join(dir, "b"), filepath.Join(dir, "c")
os.WriteFile(a, big, 0o644)
os.WriteFile(b, big, 0o644)
alt := append([]byte(nil), big...)
alt[len(alt)-1] = 'X'
os.WriteFile(c, alt, 0o644)
if same, err := fsops.SameContent(a, b); err != nil || !same {
t.Fatalf("a,b should match: %v %v", same, err)
}
if same, _ := fsops.SameContent(a, c); same {
t.Fatal("a,c differ in the last byte")
}
}
func TestNormTitle(t *testing.T) {
cases := map[string]string{
"01 - Song.flac": "song",
"1-03 Song.mp3": "song",
"03. Song!.flac": "song",
"22.flac": "22",
"Song (Live).flac": "songlive",
"07 Song (Remix).flac": "songremix",
}
for in, want := range cases {
if got := normTitle(in); got != want {
t.Errorf("normTitle(%q) = %q, want %q", in, got, want)
}
}
}
+834
View File
@@ -0,0 +1,834 @@
package engine
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/fs"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"time"
"github.com/cdnutter/playlink/internal/fsops"
"github.com/cdnutter/playlink/internal/probe"
"github.com/cdnutter/playlink/internal/store"
)
type LinkOptions struct {
// AllowDowngrade confirms linking when the host copy scored lower.
AllowDowngrade bool
Trigger string
}
// snapshot holds the verified state of both files right before an operation.
type snapshot struct {
hostAbs, slaveAbs string
hi, si fsops.Info
}
// preflight re-validates a match against the filesystem: both files must be
// plain regular files inside their roots, on one device, and unchanged since
// the scan recorded them.
func preflight(l *libs, m store.Match) (snapshot, error) {
var s snapshot
var err error
if s.hostAbs, err = l.host.Resolve(m.RelHost); err != nil {
return s, err
}
if s.slaveAbs, err = l.slave.Resolve(m.RelSlave); err != nil {
return s, err
}
if s.hi, err = fsops.RegularFile(s.hostAbs); err != nil {
return s, fmt.Errorf("host file: %w", err)
}
if s.si, err = fsops.RegularFile(s.slaveAbs); err != nil {
return s, fmt.Errorf("slave file: %w", err)
}
if s.hi.Ino != m.HostIno || !s.hi.Unchanged(m.HostSize, m.HostMtime) {
return s, fmt.Errorf("host %w", fsops.ErrChanged)
}
if s.si.Ino != m.SlaveIno || !s.si.Unchanged(m.SlaveSize, m.SlaveMtime) {
return s, fmt.Errorf("slave %w", fsops.ErrChanged)
}
if s.hi.Dev != s.si.Dev || s.hi.Dev != l.dev {
return s, fsops.ErrCrossDevice
}
if s.hi.SameInode(s.si) {
return s, errors.New("already linked")
}
return s, nil
}
func quarantinePath(l *libs, opID int64, side, rel string) string {
return filepath.Join(l.quarantine, strconv.FormatInt(opID, 10), side, rel)
}
// quarantine adds an extra hardlink of p (whose identity is in) inside the
// quarantine. The original stays in place; nothing is lost if a later step
// fails. The caller removes or replaces p afterwards.
func quarantine(p string, in fsops.Info, q string) error {
if err := fsops.MkdirAllLike(filepath.Dir(q)); err != nil {
return err
}
return fsops.LinkNew(p, in, q)
}
// dropQuarantine removes a quarantine link that turned out not to be needed.
func dropQuarantine(l *libs, q string, keep fsops.Info) {
if err := fsops.RemoveLink(q, keep); err == nil {
pruneEmpty(filepath.Dir(q), l.quarantine)
}
}
// removeOriginal deletes an original file after it was safely superseded. The
// inode must still be the one we verified.
func removeOriginal(p string, expect fsops.Info) error {
now, err := fsops.RegularFile(p)
if err != nil {
return err
}
if !now.SameInode(expect) || !now.Unchanged(expect.Size, expect.MtimeNs) {
return fmt.Errorf("%s %w", p, fsops.ErrChanged)
}
if err := os.Remove(p); err != nil {
return err
}
return fsops.SyncDir(filepath.Dir(p))
}
// pruneEmpty removes empty directories from dir upward, stopping at stop.
func pruneEmpty(dir, stop string) {
for dir != stop && fsops.Within(stop, dir) {
if err := os.Remove(dir); err != nil {
return
}
dir = filepath.Dir(dir)
}
}
func (e *Engine) begin(ctx context.Context, o *store.Operation) error {
o.DryRun = e.st.DryRun(ctx)
if o.Trigger == "" {
o.Trigger = "manual"
}
o.Status = store.OpRunning
if o.DryRun {
o.Status = store.OpPlanned
}
return e.st.CreateOp(ctx, o)
}
func (e *Engine) finish(ctx context.Context, o *store.Operation, err error) error {
if err != nil {
o.Status, o.Error = store.OpFailed, err.Error()
e.log.Warn("operation failed", "op", o.ID, "action", o.Action, "slave", o.RelSlave, "err", err)
} else {
o.Status = store.OpDone
e.log.Info("operation done", "op", o.ID, "action", o.Action, "slave", o.RelSlave, "host", o.RelHost)
}
if ferr := e.st.FinishOp(ctx, o.ID, o.Status, o.Error); ferr != nil && err == nil {
return ferr
}
return err
}
func verdictOf(m store.Match) probe.Verdict {
var v probe.Verdict
_ = json.Unmarshal(m.Verdict, &v)
return v
}
// Link replaces the slave file with a hardlink of the host file.
//
// Identical files are re-compared byte for byte right before linking and no
// quarantine copy is kept (the data exists in the host). Non-identical files
// are quarantined first so the operation can be undone.
func (e *Engine) Link(ctx context.Context, matchID int64, opt LinkOptions) (store.Operation, error) {
e.mu.Lock()
defer e.mu.Unlock()
m, err := e.st.Match(ctx, matchID)
if err != nil {
return store.Operation{}, err
}
l, err := e.loadPair(ctx, m.PairID)
if err != nil {
return store.Operation{}, err
}
switch m.Status {
case store.StatusIdentical, store.StatusDifferent, store.StatusSimilar:
default:
return store.Operation{}, fmt.Errorf("cannot link a match with status %q", m.Status)
}
if m.Ignored {
return store.Operation{}, errors.New("match is ignored; un-ignore it first")
}
identical := m.Status == store.StatusIdentical
if !identical && verdictOf(m).Better == "slave" && !opt.AllowDowngrade {
return store.Operation{}, ErrDowngrade
}
s, err := preflight(l, m)
if err != nil {
return store.Operation{}, err
}
if identical {
same, err := fsops.SameContent(s.hostAbs, s.slaveAbs)
if err != nil {
return store.Operation{}, err
}
if !same {
e.refresh(ctx, l, m.RelSlave)
return store.Operation{}, errors.New("files are no longer identical; moved to manual review")
}
}
newRel := filepath.Join(filepath.Dir(m.RelSlave), filepath.Base(m.RelHost))
newAbs, err := l.slave.Resolve(newRel)
if err != nil {
return store.Operation{}, err
}
rename := newRel != m.RelSlave
if rename {
if _, err := os.Lstat(newAbs); err == nil {
return store.Operation{}, fmt.Errorf("slave already has %s: %w", newRel, fsops.ErrExists)
}
}
o := store.Operation{PairID: l.pair.ID, Action: store.ActLink, RelHost: m.RelHost, RelSlave: m.RelSlave,
RelNew: newRel, Trigger: opt.Trigger}
if identical {
o.Bytes = s.si.Size // space freed
} else {
o.QuarantineSide = "slave"
}
if err := e.begin(ctx, &o); err != nil {
return o, err
}
if o.DryRun {
return o, nil
}
err = func() error {
var q string
if !identical {
q = quarantinePath(l, o.ID, "slave", m.RelSlave)
if err := quarantine(s.slaveAbs, s.si, q); err != nil {
return fmt.Errorf("quarantine: %w", err)
}
o.QuarantinePath = q
if _, err := e.st.DB.ExecContext(ctx, `UPDATE operations SET quarantine_path = ? WHERE id = ?`, q, o.ID); err != nil {
dropQuarantine(l, q, s.si)
return err
}
}
if !rename {
if err := fsops.ReplaceWithLink(s.hostAbs, s.hi, s.slaveAbs); err != nil {
if q != "" {
if cur, e2 := fsops.Lstat(s.slaveAbs); e2 == nil && cur.SameInode(s.si) {
dropQuarantine(l, q, s.si) // original untouched
e.clearQuarantine(ctx, o.ID)
}
}
return err
}
return nil
}
if err := fsops.LinkNew(s.hostAbs, s.hi, newAbs); err != nil {
if q != "" {
dropQuarantine(l, q, s.si)
e.clearQuarantine(ctx, o.ID)
}
return err
}
return removeOriginal(s.slaveAbs, s.si)
}()
err = e.finish(ctx, &o, err)
e.refresh(ctx, l, m.RelSlave)
if rename {
e.refresh(ctx, l, newRel)
}
return o, err
}
func (e *Engine) clearQuarantine(ctx context.Context, opID int64) {
_, _ = e.st.DB.ExecContext(ctx, `UPDATE operations SET quarantine_path = '' WHERE id = ?`, opID)
}
// Promote is the override path: the slave copy is better, so it replaces the
// host file (whose original is quarantined) and both libraries share it.
func (e *Engine) Promote(ctx context.Context, matchID int64, confirmed bool) (store.Operation, error) {
if !confirmed {
return store.Operation{}, ErrConfirm
}
e.mu.Lock()
defer e.mu.Unlock()
m, err := e.st.Match(ctx, matchID)
if err != nil {
return store.Operation{}, err
}
l, err := e.loadPair(ctx, m.PairID)
if err != nil {
return store.Operation{}, err
}
if m.Status != store.StatusDifferent && m.Status != store.StatusSimilar {
return store.Operation{}, fmt.Errorf("cannot promote a match with status %q", m.Status)
}
s, err := preflight(l, m)
if err != nil {
return store.Operation{}, err
}
newHostRel := filepath.Join(filepath.Dir(m.RelHost), filepath.Base(m.RelSlave))
newHostAbs, err := l.host.Resolve(newHostRel)
if err != nil {
return store.Operation{}, err
}
rename := newHostRel != m.RelHost
if rename {
if _, err := os.Lstat(newHostAbs); err == nil {
return store.Operation{}, fmt.Errorf("host already has %s: %w", newHostRel, fsops.ErrExists)
}
}
o := store.Operation{PairID: l.pair.ID, Action: store.ActPromote, RelHost: m.RelHost, RelSlave: m.RelSlave,
RelNew: newHostRel, QuarantineSide: "host", Trigger: "manual"}
if err := e.begin(ctx, &o); err != nil {
return o, err
}
if o.DryRun {
return o, nil
}
err = func() error {
q := quarantinePath(l, o.ID, "host", m.RelHost)
if err := quarantine(s.hostAbs, s.hi, q); err != nil {
return fmt.Errorf("quarantine: %w", err)
}
o.QuarantinePath = q
if _, err := e.st.DB.ExecContext(ctx, `UPDATE operations SET quarantine_path = ? WHERE id = ?`, q, o.ID); err != nil {
dropQuarantine(l, q, s.hi)
return err
}
if !rename {
if err := fsops.ReplaceWithLink(s.slaveAbs, s.si, s.hostAbs); err != nil {
if cur, e2 := fsops.Lstat(s.hostAbs); e2 == nil && cur.SameInode(s.hi) {
dropQuarantine(l, q, s.hi)
e.clearQuarantine(ctx, o.ID)
}
return err
}
return nil
}
if err := fsops.LinkNew(s.slaveAbs, s.si, newHostAbs); err != nil {
dropQuarantine(l, q, s.hi)
e.clearQuarantine(ctx, o.ID)
return err
}
return removeOriginal(s.hostAbs, s.hi)
}()
err = e.finish(ctx, &o, err)
e.refresh(ctx, l, m.RelSlave)
return o, err
}
// ---- build (fresh slave library) and remove ----
// expand turns a list of relative files/folders into audio files under root.
func expand(ctx context.Context, l *libs, root fsops.Root, rels []string) ([]string, error) {
set := map[string]fsops.Info{}
for _, rel := range rels {
abs, err := root.Resolve(rel)
if err != nil {
return nil, err
}
fi, err := os.Lstat(abs)
if err != nil {
return nil, err
}
switch {
case fi.IsDir():
sub, err := fsops.NewRoot(abs)
if err != nil {
return nil, err
}
files := map[string]fsops.Info{}
if _, err := walk(ctx, l, sub, files); err != nil {
return nil, err
}
for r, in := range files {
set[filepath.Join(rel, r)] = in
}
case fi.Mode().IsRegular() && l.hasExt(abs):
set[filepath.Clean(rel)] = fsops.Info{}
}
}
return sortedKeys(set), nil
}
// StartBuild hardlinks the chosen host files/folders into the slave library,
// mirroring the host path. Existing slave files are never overwritten.
func (e *Engine) StartBuild(pairID int64, rels []string) (*Job, error) {
return e.startJob(pairID, "build", func(ctx context.Context, j *Job) error {
l, err := e.loadPair(ctx, pairID)
if err != nil {
return err
}
files, err := expand(ctx, l, l.host, rels)
if err != nil {
return err
}
j.phase("Linking into slave library", int64(len(files)))
for _, rel := range files {
if ctx.Err() != nil {
return ctx.Err()
}
e.buildOne(ctx, j, l, rel)
j.tick()
}
return nil
})
}
func (e *Engine) buildOne(ctx context.Context, j *Job, l *libs, rel string) {
e.mu.Lock()
defer e.mu.Unlock()
hostAbs, err := l.host.Resolve(rel)
if err != nil {
j.skip(rel + ": " + err.Error())
return
}
slaveAbs, err := l.slave.Resolve(rel)
if err != nil {
j.skip(rel + ": " + err.Error())
return
}
hi, err := fsops.RegularFile(hostAbs)
if err != nil {
j.skip(rel + ": " + err.Error())
return
}
if si, err := fsops.Lstat(slaveAbs); err == nil {
if si.SameInode(hi) {
j.count("already linked")
} else {
j.skip(rel + ": a different file already exists in the slave library (scan, then use Review)")
}
return
}
o := store.Operation{PairID: l.pair.ID, Action: store.ActBuild, RelHost: rel, RelSlave: rel, RelNew: rel, Trigger: "build"}
if err := e.begin(ctx, &o); err != nil {
j.skip(rel + ": " + err.Error())
return
}
if o.DryRun {
j.count("planned")
return
}
err = fsops.MkdirAllLike(filepath.Dir(slaveAbs))
if err == nil {
err = fsops.LinkNew(hostAbs, hi, slaveAbs)
}
if err = e.finish(ctx, &o, err); err != nil {
j.skip(rel + ": " + err.Error())
return
}
j.count("linked")
e.refresh(ctx, l, rel)
}
// StartRemove deletes slave files that are hardlinks of the host file at the
// same path. Anything that is not such a link is left alone.
func (e *Engine) StartRemove(pairID int64, rels []string) (*Job, error) {
return e.startJob(pairID, "remove", func(ctx context.Context, j *Job) error {
l, err := e.loadPair(ctx, pairID)
if err != nil {
return err
}
files, err := expand(ctx, l, l.slave, rels)
if err != nil {
return err
}
j.phase("Removing links from slave library", int64(len(files)))
for _, rel := range files {
if ctx.Err() != nil {
return ctx.Err()
}
e.removeOne(ctx, j, l, rel)
j.tick()
}
return nil
})
}
func (e *Engine) removeOne(ctx context.Context, j *Job, l *libs, rel string) {
e.mu.Lock()
defer e.mu.Unlock()
hostAbs, err1 := l.host.Resolve(rel)
slaveAbs, err2 := l.slave.Resolve(rel)
if err1 != nil || err2 != nil {
j.skip(rel + ": invalid path")
return
}
hi, err := fsops.RegularFile(hostAbs)
if err != nil {
j.skip(rel + ": no host file at the same path, so it is not removed")
return
}
si, err := fsops.RegularFile(slaveAbs)
if err != nil {
j.skip(rel + ": " + err.Error())
return
}
if !si.SameInode(hi) {
j.skip(rel + ": not a hardlink of the host file, so it is not removed")
return
}
o := store.Operation{PairID: l.pair.ID, Action: store.ActRemove, RelHost: rel, RelSlave: rel, Trigger: "build"}
if err := e.begin(ctx, &o); err != nil {
j.skip(rel + ": " + err.Error())
return
}
if o.DryRun {
j.count("planned")
return
}
err = fsops.RemoveLink(slaveAbs, hi)
if err = e.finish(ctx, &o, err); err != nil {
j.skip(rel + ": " + err.Error())
return
}
pruneEmpty(filepath.Dir(slaveAbs), l.slave.Path)
j.count("removed")
e.refresh(ctx, l, rel)
}
// ---- undo, restore and purge ----
// Undo reverts a completed operation. Every branch verifies that the current
// file is still exactly what the operation left behind before touching it.
func (e *Engine) Undo(ctx context.Context, opID int64) error {
e.mu.Lock()
defer e.mu.Unlock()
o, err := e.st.Op(ctx, opID)
if err != nil {
return err
}
if o.Status != store.OpDone || o.UndoneAt != nil {
return errors.New("only completed operations that were not undone can be reverted")
}
if o.PurgedAt != nil {
return errors.New("the quarantined original was already purged")
}
l, err := e.loadPair(ctx, o.PairID)
if err != nil {
return err
}
res := func(r fsops.Root, rel string) string {
p, e := r.Resolve(rel)
if e != nil && err == nil {
err = e
}
return p
}
hostAbs := res(l.host, o.RelHost)
slaveAbs := res(l.slave, o.RelSlave)
if err != nil {
return err
}
switch o.Action {
case store.ActBuild:
hi, err := fsops.RegularFile(hostAbs)
if err != nil {
return err
}
if err := fsops.RemoveLink(slaveAbs, hi); err != nil {
return err
}
pruneEmpty(filepath.Dir(slaveAbs), l.slave.Path)
case store.ActRemove:
hi, err := fsops.RegularFile(hostAbs)
if err != nil {
return err
}
if err := fsops.MkdirAllLike(filepath.Dir(slaveAbs)); err != nil {
return err
}
if err := fsops.LinkNew(hostAbs, hi, slaveAbs); err != nil {
return err
}
case store.ActLink:
newAbs := res(l.slave, o.RelNew)
if err != nil {
return err
}
hi, err := fsops.RegularFile(hostAbs)
if err != nil {
return err
}
cur, err := fsops.RegularFile(newAbs)
if err != nil {
return err
}
if !cur.SameInode(hi) {
return errors.New("the slave file is no longer linked to the host file; refusing to undo")
}
renamed := o.RelNew != o.RelSlave
if renamed {
if _, err := os.Lstat(slaveAbs); err == nil {
return fmt.Errorf("%s: %w", o.RelSlave, fsops.ErrExists)
}
}
if o.QuarantinePath == "" {
// Identical link: give the slave its own copy of the bytes again.
if err := fsops.CopyReplace(hostAbs, slaveAbs, hi.Mode); err != nil {
return err
}
} else {
qi, err := e.quarantined(l, o.QuarantinePath)
if err != nil {
return err
}
if renamed {
err = fsops.LinkNew(o.QuarantinePath, qi, slaveAbs)
} else {
err = fsops.ReplaceWithLink(o.QuarantinePath, qi, slaveAbs)
}
if err != nil {
return err
}
dropQuarantine(l, o.QuarantinePath, qi)
}
if renamed {
if err := fsops.RemoveLink(newAbs, hi); err != nil {
return err
}
}
e.refresh(ctx, l, o.RelNew)
case store.ActPromote:
newHostAbs := res(l.host, o.RelNew)
if err != nil {
return err
}
si, err := fsops.RegularFile(slaveAbs)
if err != nil {
return err
}
cur, err := fsops.RegularFile(newHostAbs)
if err != nil {
return err
}
if !cur.SameInode(si) {
return errors.New("the host file changed after promotion; refusing to undo")
}
qi, err := e.quarantined(l, o.QuarantinePath)
if err != nil {
return err
}
if o.RelNew == o.RelHost {
err = fsops.ReplaceWithLink(o.QuarantinePath, qi, hostAbs)
} else {
if _, e2 := os.Lstat(hostAbs); e2 == nil {
return fmt.Errorf("%s: %w", o.RelHost, fsops.ErrExists)
}
if err = fsops.LinkNew(o.QuarantinePath, qi, hostAbs); err == nil {
err = fsops.RemoveLink(newHostAbs, si)
}
}
if err != nil {
return err
}
dropQuarantine(l, o.QuarantinePath, qi)
default:
return fmt.Errorf("cannot undo %q", o.Action)
}
if err := e.st.MarkUndone(ctx, o.ID); err != nil {
return err
}
e.log.Info("operation undone", "op", o.ID, "action", o.Action)
e.refresh(ctx, l, o.RelSlave)
return nil
}
// quarantined validates a quarantine path recorded in the database.
func (e *Engine) quarantined(l *libs, q string) (fsops.Info, error) {
q = filepath.Clean(q)
if !fsops.Within(l.quarantine, q) || q == l.quarantine {
return fsops.Info{}, fmt.Errorf("quarantine path %s: %w", q, fsops.ErrOutsideRoot)
}
return fsops.RegularFile(q)
}
// Purge permanently deletes a quarantined original.
func (e *Engine) Purge(ctx context.Context, opID int64) error {
e.mu.Lock()
defer e.mu.Unlock()
o, err := e.st.Op(ctx, opID)
if err != nil {
return err
}
return e.purge(ctx, o)
}
func (e *Engine) purge(ctx context.Context, o store.Operation) error {
if o.QuarantinePath == "" || o.PurgedAt != nil || o.UndoneAt != nil {
return errors.New("nothing to purge for this operation")
}
l, err := e.loadPair(ctx, o.PairID)
if err != nil {
return err
}
opDir := filepath.Join(l.quarantine, strconv.FormatInt(o.ID, 10))
if !fsops.Within(l.quarantine, filepath.Clean(o.QuarantinePath)) || !fsops.Within(opDir, filepath.Clean(o.QuarantinePath)) {
return fsops.ErrOutsideRoot
}
if err := os.RemoveAll(opDir); err != nil {
return err
}
e.log.Info("quarantine purged", "op", o.ID, "path", o.QuarantinePath)
return e.st.MarkPurged(ctx, o.ID)
}
// Maintenance purges expired quarantine entries and prunes sessions.
func (e *Engine) Maintenance(ctx context.Context) {
e.st.PruneSessions(ctx)
pairs, err := e.st.Pairs(ctx)
if err != nil {
return
}
for _, p := range pairs {
if p.RetentionDays <= 0 {
continue // keep forever
}
cutoff := time.Now().Add(-time.Duration(p.RetentionDays) * 24 * time.Hour).Unix()
ops, err := e.st.ExpiredQuarantine(ctx, p.ID, cutoff)
if err != nil {
continue
}
for _, o := range ops {
e.mu.Lock()
if err := e.purge(ctx, o); err != nil {
e.log.Warn("retention purge failed", "op", o.ID, "err", err)
}
e.mu.Unlock()
}
}
}
// Recover runs at startup: operations left running by a crash are marked
// interrupted and their disposable staging files are removed. Originals are
// never at risk because every replacement is an atomic rename.
func (e *Engine) Recover(ctx context.Context) {
ops, err := e.st.RunningOps(ctx)
if err != nil {
return
}
for _, o := range ops {
if p, err := e.st.Pair(ctx, o.PairID); err == nil {
for _, root := range []string{p.SlaveRoot, p.HostRoot} {
for _, rel := range []string{o.RelSlave, o.RelNew, o.RelHost} {
if rel == "" {
continue
}
dir := filepath.Join(root, filepath.Dir(rel))
ents, _ := os.ReadDir(dir)
for _, en := range ents {
if strings.HasPrefix(en.Name(), fsops.TmpPrefix) && en.Type().IsRegular() {
_ = os.Remove(filepath.Join(dir, en.Name()))
}
}
}
}
}
_ = e.st.FinishOp(ctx, o.ID, store.OpInterrupted, "interrupted by shutdown; rescan to verify state")
e.log.Warn("marked interrupted operation", "op", o.ID)
}
}
// ---- browsing (build mode) ----
type TreeEntry struct {
Name string `json:"name"`
Rel string `json:"rel"`
Dir bool `json:"dir"`
Size int64 `json:"size,omitempty"`
State string `json:"state,omitempty"` // files: linked, conflict, missing
Linked int `json:"linked,omitempty"`
}
// Tree lists a host directory and shows how much of it is in the slave.
func (e *Engine) Tree(ctx context.Context, pairID int64, rel string) ([]TreeEntry, error) {
l, err := e.loadPair(ctx, pairID)
if err != nil {
return nil, err
}
dir := l.host.Path
if rel != "" {
if dir, err = l.host.Resolve(rel); err != nil {
return nil, err
}
}
ents, err := os.ReadDir(dir)
if err != nil {
return nil, err
}
linked, _ := e.st.LinkedHostPaths(ctx, pairID, rel)
perChild := map[string]int{}
for _, p := range linked {
sub := p
if rel != "" {
sub = strings.TrimPrefix(p, rel+string(filepath.Separator))
}
if i := strings.IndexByte(sub, filepath.Separator); i > 0 {
perChild[sub[:i]]++
}
}
out := []TreeEntry{}
for _, en := range ents {
name := en.Name()
if strings.HasPrefix(name, ".") || en.Type()&fs.ModeSymlink != 0 {
continue
}
child := name
if rel != "" {
child = filepath.Join(rel, name)
}
te := TreeEntry{Name: name, Rel: child, Dir: en.IsDir()}
if en.IsDir() {
te.Linked = perChild[name]
} else {
if !l.hasExt(name) {
continue
}
hi, err := fsops.Lstat(filepath.Join(dir, name))
if err != nil {
continue
}
te.Size = hi.Size
te.State = "missing"
if sp, err := l.slave.Resolve(child); err == nil {
if si, err := fsops.Lstat(sp); err == nil {
te.State = "conflict"
if si.SameInode(hi) {
te.State = "linked"
}
}
}
}
out = append(out, te)
}
sort.Slice(out, func(i, j int) bool {
if out[i].Dir != out[j].Dir {
return out[i].Dir
}
return strings.ToLower(out[i].Name) < strings.ToLower(out[j].Name)
})
return out, nil
}
+448
View File
@@ -0,0 +1,448 @@
package engine
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/fs"
"path/filepath"
"regexp"
"strings"
"time"
"unicode"
"github.com/cdnutter/playlink/internal/fsops"
"github.com/cdnutter/playlink/internal/probe"
"github.com/cdnutter/playlink/internal/store"
)
type walkStats struct {
files int
symlinks int
offDepth int // files not at Artist/Album/Title depth
}
// walk lists audio files beneath root. Hidden entries, playlink staging files,
// the quarantine and symlinks are never followed or returned.
func walk(ctx context.Context, l *libs, root fsops.Root, out map[string]fsops.Info) (walkStats, error) {
var ws walkStats
err := filepath.WalkDir(root.Path, func(p string, d fs.DirEntry, err error) error {
if err != nil {
if p == root.Path {
return err
}
return nil // unreadable subtree: skip rather than abort
}
if ctx.Err() != nil {
return ctx.Err()
}
name := d.Name()
if p != root.Path && strings.HasPrefix(name, ".") {
if d.IsDir() {
return filepath.SkipDir
}
return nil
}
if d.IsDir() {
if p == l.quarantine {
return filepath.SkipDir
}
return nil
}
if d.Type()&fs.ModeSymlink != 0 {
ws.symlinks++
return nil
}
if !d.Type().IsRegular() || !l.hasExt(name) {
return nil
}
in, err := fsops.Lstat(p)
if err != nil || !in.Mode.IsRegular() {
return nil
}
rel, err := root.Rel(p)
if err != nil {
return nil
}
out[rel] = in
ws.files++
if strings.Count(rel, string(filepath.Separator)) != 2 {
ws.offDepth++
}
return nil
})
return ws, err
}
var trackPrefix = regexp.MustCompile(`^\s*(?:\d{1,2}[-.]\s*)?\d{1,3}(?:\s*[-._)]\s*|\s+)`)
// normTitle makes a conservative comparison key from a file name: the track
// number prefix is stripped and punctuation/case ignored. Parenthesised text
// is kept, so "Song (Live)" never matches "Song".
func normTitle(name string) string {
stem := strings.TrimSuffix(name, filepath.Ext(name))
if s := trackPrefix.ReplaceAllString(stem, ""); strings.TrimSpace(s) != "" {
stem = s
}
var b strings.Builder
for _, r := range strings.ToLower(stem) {
if unicode.IsLetter(r) || unicode.IsDigit(r) {
b.WriteRune(r)
}
}
return b.String()
}
func firstComponent(rel string) string {
if i := strings.IndexByte(rel, filepath.Separator); i >= 0 {
return rel[:i]
}
return ""
}
// hostIndex supports the fallback lookups for slave files without an exact
// path counterpart.
type hostIndex struct {
files map[string]fsops.Info
lower map[string]string // lower(rel) -> rel
byDir map[string][]string // lower(dir) -> rels
byTitle map[string][]string // lower(artist)|normTitle -> rels
}
func buildIndex(files map[string]fsops.Info) *hostIndex {
ix := &hostIndex{files: files, lower: map[string]string{}, byDir: map[string][]string{}, byTitle: map[string][]string{}}
for _, rel := range sortedKeys(files) {
ix.lower[strings.ToLower(rel)] = rel
d := strings.ToLower(filepath.Dir(rel))
ix.byDir[d] = append(ix.byDir[d], rel)
k := strings.ToLower(firstComponent(rel)) + "|" + normTitle(filepath.Base(rel))
ix.byTitle[k] = append(ix.byTitle[k], rel)
}
return ix
}
// losslessRank orders extension candidates so a lossless host file is
// preferred when several share the same title.
func extRank(rel string) int {
switch strings.ToLower(filepath.Ext(rel)) {
case ".flac", ".wav", ".aiff", ".aif", ".alac", ".ape", ".wv", ".dsf", ".dff":
return 0
case ".m4a", ".opus", ".ogg":
return 1
}
return 2
}
// find locates the host counterpart of a slave file.
func (ix *hostIndex) find(relSlave string) (string, string) {
if _, ok := ix.files[relSlave]; ok {
return relSlave, store.KindPath
}
if r, ok := ix.lower[strings.ToLower(relSlave)]; ok {
return r, store.KindPathCase
}
dir := strings.ToLower(filepath.Dir(relSlave))
base := filepath.Base(relSlave)
stem := strings.ToLower(strings.TrimSuffix(base, filepath.Ext(base)))
var extHits, fuzzyHits []string
norm := normTitle(base)
for _, r := range ix.byDir[dir] {
b := filepath.Base(r)
if strings.ToLower(strings.TrimSuffix(b, filepath.Ext(b))) == stem {
extHits = append(extHits, r)
} else if norm != "" && normTitle(b) == norm {
fuzzyHits = append(fuzzyHits, r)
}
}
if len(extHits) > 0 {
best := extHits[0]
for _, r := range extHits[1:] {
if extRank(r) < extRank(best) {
best = r
}
}
return best, store.KindExtension
}
if len(fuzzyHits) == 1 {
return fuzzyHits[0], store.KindFuzzy
}
if artist := firstComponent(relSlave); artist != "" && norm != "" {
if hits := ix.byTitle[strings.ToLower(artist)+"|"+norm]; len(hits) == 1 {
return hits[0], store.KindElsewhere
}
}
return "", store.KindNone
}
// classify fills in status, probes and verdict for a match, reusing cached
// results from prev when neither file changed.
func (e *Engine) classify(ctx context.Context, l *libs, m *store.Match, hi, si fsops.Info, prev *store.Match) {
m.SlaveIno, m.SlaveSize, m.SlaveMtime = si.Ino, si.Size, si.MtimeNs
m.HostIno, m.HostSize, m.HostMtime = 0, 0, 0
m.HostProbe, m.SlaveProbe, m.Verdict, m.Note = nil, nil, nil, ""
if m.Kind == store.KindNone {
m.Status = store.StatusUnmatched
return
}
m.HostIno, m.HostSize, m.HostMtime = hi.Ino, hi.Size, hi.MtimeNs
unchanged := prev != nil && prev.RelHost == m.RelHost &&
prev.HostIno == hi.Ino && prev.HostSize == hi.Size && prev.HostMtime == hi.MtimeNs &&
prev.SlaveIno == si.Ino && prev.SlaveSize == si.Size && prev.SlaveMtime == si.MtimeNs
if prev != nil && prev.RelHost == m.RelHost {
m.Ignored = prev.Ignored
}
switch {
case hi.Dev != si.Dev:
m.Status, m.Note = store.StatusError, "files are on different filesystems"
return
case hi.SameInode(si):
m.Status = store.StatusLinked
return
}
pathKind := m.Kind == store.KindPath || m.Kind == store.KindPathCase
if pathKind && hi.Size == si.Size {
if unchanged && (prev.Status == store.StatusIdentical || prev.Status == store.StatusDifferent) {
m.Status = prev.Status
} else {
hostAbs := filepath.Join(l.host.Path, m.RelHost)
slaveAbs := filepath.Join(l.slave.Path, m.RelSlave)
same, err := fsops.SameContent(hostAbs, slaveAbs)
switch {
case err != nil:
m.Status, m.Note = store.StatusError, "compare failed: "+err.Error()
return
case same:
m.Status = store.StatusIdentical
default:
m.Status = store.StatusDifferent
}
}
if m.Status == store.StatusIdentical {
return
}
} else if pathKind {
m.Status = store.StatusDifferent
} else {
m.Status = store.StatusSimilar
}
// Manual review: gather codec details for both sides.
if unchanged && len(prev.HostProbe) > 0 && len(prev.SlaveProbe) > 0 {
m.HostProbe, m.SlaveProbe, m.Verdict = prev.HostProbe, prev.SlaveProbe, prev.Verdict
return
}
hp := e.prober.Probe(ctx, filepath.Join(l.host.Path, m.RelHost))
sp := e.prober.Probe(ctx, filepath.Join(l.slave.Path, m.RelSlave))
v := probe.Compare(hp, sp)
m.HostProbe, _ = json.Marshal(hp)
m.SlaveProbe, _ = json.Marshal(sp)
m.Verdict, _ = json.Marshal(v)
}
// StartScan launches a background scan of a pair.
func (e *Engine) StartScan(pairID int64, trigger string) (*Job, error) {
return e.startJob(pairID, "scan", func(ctx context.Context, j *Job) error {
return e.scan(ctx, j, pairID, trigger)
})
}
func (e *Engine) scan(ctx context.Context, j *Job, pairID int64, trigger string) error {
started := time.Now()
l, err := e.loadPair(ctx, pairID)
if err != nil {
_ = e.st.FinishScan(ctx, pairID, started.Unix(), err.Error(), nil)
return err
}
warns := layoutWarnings(l)
j.phase("Indexing host library", 0)
hostFiles := map[string]fsops.Info{}
hs, err := walk(ctx, l, l.host, hostFiles)
if err != nil {
_ = e.st.FinishScan(ctx, pairID, started.Unix(), err.Error(), warns)
return err
}
j.phase("Indexing slave library", 0)
slaveFiles := map[string]fsops.Info{}
ss, err := walk(ctx, l, l.slave, slaveFiles)
if err != nil {
_ = e.st.FinishScan(ctx, pairID, started.Unix(), err.Error(), warns)
return err
}
ix := buildIndex(hostFiles)
prev, err := e.st.AllMatches(ctx, pairID)
if err != nil {
return err
}
j.phase("Comparing files", int64(len(slaveFiles)))
keep := map[string]bool{}
kinds := map[string]int{}
for _, rel := range sortedKeys(slaveFiles) {
if ctx.Err() != nil {
return ctx.Err()
}
si := slaveFiles[rel]
m := store.Match{PairID: pairID, RelSlave: rel}
m.RelHost, m.Kind = ix.find(rel)
var pm *store.Match
if p, ok := prev[rel]; ok {
pm = &p
}
e.classify(ctx, l, &m, hostFiles[m.RelHost], si, pm)
if err := e.st.UpsertMatch(ctx, &m); err != nil {
return err
}
keep[rel] = true
kinds[m.Kind]++
j.count(m.Status)
j.tick()
}
if err := e.st.DeleteMatchesExcept(ctx, pairID, keep); err != nil {
return err
}
// Layout warnings based on what was actually found.
if ss.files > 0 {
if n := kinds[store.KindElsewhere] + kinds[store.KindNone]; n*4 > ss.files {
warns = append(warns, fmt.Sprintf("%d of %d slave files have no counterpart at the same path in the host library. "+
"Check that both libraries use the same Artist/Album/Title.ext layout.", n, ss.files))
}
if ss.offDepth > 0 {
warns = append(warns, fmt.Sprintf("%d slave files are not at Artist/Album/Title.ext depth.", ss.offDepth))
}
}
if hs.offDepth > 0 {
warns = append(warns, fmt.Sprintf("%d host files are not at Artist/Album/Title.ext depth.", hs.offDepth))
}
if hs.symlinks+ss.symlinks > 0 {
warns = append(warns, fmt.Sprintf("Ignored %d symlinks (playlink never follows or replaces symlinks).", hs.symlinks+ss.symlinks))
}
if e.prober.Bin == "" {
warns = append(warns, "ffprobe was not found, so codec and quality details are unavailable for manual review.")
}
if err := e.st.FinishScan(ctx, pairID, time.Now().Unix(), "", warns); err != nil {
return err
}
j.update(func(j *Job) {
j.Message = fmt.Sprintf("Scanned %d host and %d slave files in %s", hs.files, ss.files, time.Since(started).Round(time.Second))
})
if l.pair.AutoLinkIdentical {
return e.autoLink(ctx, j, l, trigger)
}
return nil
}
// autoLink links settled, byte-identical, same-path files after a scan.
func (e *Engine) autoLink(ctx context.Context, j *Job, l *libs, trigger string) error {
ids, err := e.st.IdenticalIDs(ctx, l.pair.ID)
if err != nil || len(ids) == 0 {
return err
}
if e.st.DryRun(ctx) {
j.update(func(j *Job) {
j.Message += fmt.Sprintf(". Dry run is on: %d identical files would be auto-linked.", len(ids))
})
return nil
}
return e.linkIDs(ctx, j, l.pair.ID, ids, "auto:"+trigger, true)
}
// linkIDs links a batch of identical matches.
func (e *Engine) linkIDs(ctx context.Context, j *Job, pairID int64, ids []int64, trigger string, onlySettled bool) error {
j.phase("Linking identical files", int64(len(ids)))
settle := time.Duration(0)
if onlySettled {
if p, err := e.st.Pair(ctx, pairID); err == nil {
settle = time.Duration(p.SettleMinutes) * time.Minute
}
}
cutoff := time.Now().Add(-settle).UnixNano()
for _, id := range ids {
if ctx.Err() != nil {
return ctx.Err()
}
m, err := e.st.Match(ctx, id)
if err != nil {
j.tick()
continue
}
if onlySettled && m.Kind != store.KindPath {
j.skip(m.RelSlave + ": path differs by case, link manually")
j.tick()
continue
}
if settle > 0 && (m.SlaveMtime > cutoff || m.HostMtime > cutoff) {
j.skip(m.RelSlave + ": modified recently, waiting for it to settle")
j.tick()
continue
}
op, err := e.Link(ctx, id, LinkOptions{Trigger: trigger})
switch {
case err != nil:
j.skip(m.RelSlave + ": " + err.Error())
case op.DryRun:
j.count("planned")
default:
j.count("linked")
}
j.tick()
}
return nil
}
// StartLinkIdentical links every identical match of a pair in the background.
func (e *Engine) StartLinkIdentical(pairID int64) (*Job, error) {
return e.startJob(pairID, "link-identical", func(ctx context.Context, j *Job) error {
ids, err := e.st.IdenticalIDs(ctx, pairID)
if err != nil {
return err
}
if len(ids) == 0 {
j.update(func(j *Job) { j.Message = "Nothing to link" })
return nil
}
return e.linkIDs(ctx, j, pairID, ids, "bulk", false)
})
}
// refresh re-evaluates a single slave file after an operation.
func (e *Engine) refresh(ctx context.Context, l *libs, relSlave string) {
si, err := fsops.RegularFile(filepath.Join(l.slave.Path, relSlave))
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
_ = e.st.DeleteMatch(ctx, l.pair.ID, relSlave)
}
return
}
m := store.Match{PairID: l.pair.ID, RelSlave: relSlave}
var pm *store.Match
if p, err := e.st.MatchBySlave(ctx, l.pair.ID, relSlave); err == nil {
pm = &p
}
var hi fsops.Info
if hp, err := l.host.Resolve(relSlave); err == nil {
if in, err := fsops.RegularFile(hp); err == nil {
hi = in
m.RelHost, m.Kind = relSlave, store.KindPath
}
}
if m.Kind == "" {
if pm != nil && pm.RelHost != "" {
if in, err := fsops.RegularFile(filepath.Join(l.host.Path, pm.RelHost)); err == nil {
hi = in
m.RelHost, m.Kind = pm.RelHost, pm.Kind
}
}
}
if m.Kind == "" {
m.Kind = store.KindNone
}
e.classify(ctx, l, &m, hi, si, pm)
_ = e.st.UpsertMatch(ctx, &m)
}
+405
View File
@@ -0,0 +1,405 @@
// Package fsops contains the low-level, safety-critical filesystem primitives
// used by playlink: confined path resolution, byte comparison, atomic
// hardlink replacement and directory fsyncs.
//
// Every mutating primitive in this package follows the same rule: a path that
// holds user data is never left missing. Replacements are staged as a
// temporary hardlink in the destination directory and then atomically renamed
// over the target.
package fsops
import (
"bytes"
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"io"
"io/fs"
"os"
"path/filepath"
"strings"
"syscall"
)
// TmpPrefix marks staging files created by playlink. Anything carrying this
// prefix is disposable: it is either a hardlink to a host file or a copy that
// has not been renamed into place yet.
const TmpPrefix = ".playlink-tmp-"
var (
ErrOutsideRoot = errors.New("path escapes library root")
ErrNotRegular = errors.New("not a regular file")
ErrCrossDevice = errors.New("files are on different filesystems; hardlinks are impossible")
ErrChanged = errors.New("file changed since it was scanned; rescan first")
ErrExists = errors.New("destination already exists")
ErrVerifyFailed = errors.New("post-operation verification failed")
)
// Info is the subset of stat(2) that playlink cares about.
type Info struct {
Dev uint64 `json:"dev"`
Ino uint64 `json:"ino"`
Nlink uint64 `json:"nlink"`
Size int64 `json:"size"`
MtimeNs int64 `json:"mtime_ns"`
Mode fs.FileMode `json:"-"`
}
// SameInode reports whether a and b are the same file on disk.
func (i Info) SameInode(o Info) bool { return i.Dev == o.Dev && i.Ino == o.Ino }
// Unchanged reports whether size and mtime still match a previous snapshot.
func (i Info) Unchanged(size, mtimeNs int64) bool { return i.Size == size && i.MtimeNs == mtimeNs }
// Lstat stats p without following a trailing symlink.
func Lstat(p string) (Info, error) {
fi, err := os.Lstat(p)
if err != nil {
return Info{}, err
}
return infoFrom(fi), nil
}
func infoFrom(fi fs.FileInfo) Info {
in := Info{Size: fi.Size(), MtimeNs: fi.ModTime().UnixNano(), Mode: fi.Mode()}
if st, ok := fi.Sys().(*syscall.Stat_t); ok {
in.Dev = uint64(st.Dev)
in.Ino = uint64(st.Ino)
in.Nlink = uint64(st.Nlink)
}
return in
}
// RegularFile lstat's p and fails unless it is a plain regular file (not a
// symlink, device, directory, ...).
func RegularFile(p string) (Info, error) {
in, err := Lstat(p)
if err != nil {
return in, err
}
if !in.Mode.IsRegular() {
return in, fmt.Errorf("%s: %w", p, ErrNotRegular)
}
return in, nil
}
// Root is an absolute, symlink-free directory that confines every path
// playlink touches for a library.
type Root struct {
Path string
}
// NewRoot validates dir and resolves symlinks so containment checks are exact.
func NewRoot(dir string) (Root, error) {
if !filepath.IsAbs(dir) {
return Root{}, fmt.Errorf("%q is not an absolute path", dir)
}
real, err := filepath.EvalSymlinks(filepath.Clean(dir))
if err != nil {
return Root{}, err
}
fi, err := os.Stat(real)
if err != nil {
return Root{}, err
}
if !fi.IsDir() {
return Root{}, fmt.Errorf("%q is not a directory", dir)
}
return Root{Path: real}, nil
}
// Within reports whether p equals base or lies beneath it. Both must be clean.
func Within(base, p string) bool {
if p == base {
return true
}
if base == "/" {
return strings.HasPrefix(p, "/")
}
return strings.HasPrefix(p, base+string(filepath.Separator))
}
// CleanRel validates a library-relative path coming from the database or an
// API request.
func CleanRel(rel string) (string, error) {
if rel == "" || filepath.IsAbs(rel) || strings.ContainsRune(rel, 0) {
return "", fmt.Errorf("invalid relative path %q", rel)
}
c := filepath.Clean(rel)
if c == "." || c == ".." || strings.HasPrefix(c, "../") {
return "", fmt.Errorf("%q: %w", rel, ErrOutsideRoot)
}
return c, nil
}
// Resolve maps a relative path to an absolute one inside the root. The
// deepest existing ancestor is resolved through symlinks and must still be
// inside the root, so a symlinked directory can never redirect a write
// outside the library. The final component itself is not followed.
func (r Root) Resolve(rel string) (string, error) {
c, err := CleanRel(rel)
if err != nil {
return "", err
}
full := filepath.Join(r.Path, c)
anc := filepath.Dir(full)
for {
real, err := filepath.EvalSymlinks(anc)
if err == nil {
if !Within(r.Path, real) {
return "", fmt.Errorf("%q: %w", rel, ErrOutsideRoot)
}
// Re-anchor the missing tail on the resolved ancestor.
tail, _ := filepath.Rel(anc, full)
return filepath.Join(real, tail), nil
}
if !errors.Is(err, fs.ErrNotExist) {
return "", err
}
if anc == r.Path || anc == "/" {
return "", err
}
anc = filepath.Dir(anc)
}
}
// Rel converts an absolute path inside the root back to a relative one.
func (r Root) Rel(abs string) (string, error) {
if !Within(r.Path, abs) || abs == r.Path {
return "", ErrOutsideRoot
}
return filepath.Rel(r.Path, abs)
}
// SameContent compares two files byte for byte (the equivalent of cmp(1)).
// It refuses to report equality if either file changed while being read.
func SameContent(a, b string) (bool, error) {
fa, err := os.Open(a)
if err != nil {
return false, err
}
defer fa.Close()
fb, err := os.Open(b)
if err != nil {
return false, err
}
defer fb.Close()
sa, err := fa.Stat()
if err != nil {
return false, err
}
sb, err := fb.Stat()
if err != nil {
return false, err
}
if !sa.Mode().IsRegular() || !sb.Mode().IsRegular() {
return false, ErrNotRegular
}
if sa.Size() != sb.Size() {
return false, nil
}
ia, ib := infoFrom(sa), infoFrom(sb)
if ia.SameInode(ib) {
return true, nil
}
const chunk = 1 << 20
bufA := make([]byte, chunk)
bufB := make([]byte, chunk)
for {
na, errA := io.ReadFull(fa, bufA)
nb, errB := io.ReadFull(fb, bufB)
if na != nb || !bytes.Equal(bufA[:na], bufB[:nb]) {
return false, nil
}
endA := errA == io.EOF || errA == io.ErrUnexpectedEOF
endB := errB == io.EOF || errB == io.ErrUnexpectedEOF
if errA != nil && !endA {
return false, errA
}
if errB != nil && !endB {
return false, errB
}
if endA || endB {
if endA != endB {
return false, nil
}
break
}
}
// Guard against a writer modifying either file mid-compare.
for _, pair := range []struct {
p string
prev Info
}{{a, ia}, {b, ib}} {
now, err := Lstat(pair.p)
if err != nil {
return false, err
}
if !now.SameInode(pair.prev) || !now.Unchanged(pair.prev.Size, pair.prev.MtimeNs) {
return false, ErrChanged
}
}
return true, nil
}
// SyncDir fsyncs a directory so renames and new links inside it are durable.
func SyncDir(dir string) error {
d, err := os.Open(dir)
if err != nil {
return err
}
defer d.Close()
if err := d.Sync(); err != nil && !errors.Is(err, syscall.EINVAL) {
return err
}
return nil
}
func tmpName(dir string) (string, error) {
var b [8]byte
if _, err := rand.Read(b[:]); err != nil {
return "", err
}
return filepath.Join(dir, TmpPrefix+hex.EncodeToString(b[:])), nil
}
// StageLink creates a temporary hardlink to src inside dstDir and verifies it
// points at the same inode. The caller must rename or remove the result.
func StageLink(src string, srcInfo Info, dstDir string) (string, error) {
tmp, err := tmpName(dstDir)
if err != nil {
return "", err
}
if err := os.Link(src, tmp); err != nil {
if errors.Is(err, syscall.EXDEV) {
return "", ErrCrossDevice
}
return "", err
}
got, err := Lstat(tmp)
if err != nil || !got.SameInode(srcInfo) {
_ = os.Remove(tmp)
if err == nil {
err = fmt.Errorf("%w: staged link does not share the source inode", ErrVerifyFailed)
}
return "", err
}
return tmp, nil
}
// LinkNew hardlinks src to dst, which must not exist. Verified by inode.
func LinkNew(src string, srcInfo Info, dst string) error {
if _, err := os.Lstat(dst); err == nil {
return fmt.Errorf("%s: %w", dst, ErrExists)
}
if err := os.Link(src, dst); err != nil {
if errors.Is(err, syscall.EXDEV) {
return ErrCrossDevice
}
return err
}
got, err := Lstat(dst)
if err != nil || !got.SameInode(srcInfo) {
// dst was created by us a moment ago and is only an extra link.
_ = os.Remove(dst)
if err == nil {
err = fmt.Errorf("%w: %s does not share the source inode", ErrVerifyFailed, dst)
}
return err
}
return SyncDir(filepath.Dir(dst))
}
// ReplaceWithLink atomically replaces dst with a hardlink to src. dst must be
// a regular file on the same device. The staged link is renamed over dst, so
// dst never disappears, even on a crash.
func ReplaceWithLink(src string, srcInfo Info, dst string) error {
tmp, err := StageLink(src, srcInfo, filepath.Dir(dst))
if err != nil {
return err
}
if err := os.Rename(tmp, dst); err != nil {
_ = os.Remove(tmp)
return err
}
if err := SyncDir(filepath.Dir(dst)); err != nil {
return err
}
got, err := Lstat(dst)
if err != nil {
return err
}
if !got.SameInode(srcInfo) {
return fmt.Errorf("%w: %s does not share the source inode after rename", ErrVerifyFailed, dst)
}
return nil
}
// RemoveLink deletes p only if it is a hardlink of keep (same inode and at
// least one other link). This is the only way playlink deletes a file outside
// the quarantine, and it never destroys the last copy of any data.
func RemoveLink(p string, keep Info) error {
in, err := RegularFile(p)
if err != nil {
return err
}
if !in.SameInode(keep) || in.Nlink < 2 {
return fmt.Errorf("refusing to remove %s: it is not a hardlink of the file being kept", p)
}
if err := os.Remove(p); err != nil {
return err
}
return SyncDir(filepath.Dir(p))
}
// CopyReplace replaces dst with an independent byte copy of src (used to undo
// a link between identical files). The copy is fsynced and verified before
// being renamed into place.
func CopyReplace(src, dst string, mode fs.FileMode) error {
tmp, err := tmpName(filepath.Dir(dst))
if err != nil {
return err
}
in, err := os.Open(src)
if err != nil {
return err
}
defer in.Close()
out, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_EXCL, mode.Perm())
if err != nil {
return err
}
cleanup := func(e error) error { out.Close(); _ = os.Remove(tmp); return e }
if _, err := io.Copy(out, in); err != nil {
return cleanup(err)
}
if err := out.Sync(); err != nil {
return cleanup(err)
}
if err := out.Close(); err != nil {
_ = os.Remove(tmp)
return err
}
same, err := SameContent(src, tmp)
if err != nil || !same {
_ = os.Remove(tmp)
if err == nil {
err = fmt.Errorf("%w: copy differs from source", ErrVerifyFailed)
}
return err
}
if err := os.Rename(tmp, dst); err != nil {
_ = os.Remove(tmp)
return err
}
return SyncDir(filepath.Dir(dst))
}
// MkdirAllLike creates dir and any missing parents with 0775 permissions.
func MkdirAllLike(dir string) error {
return os.MkdirAll(dir, 0o775)
}
+219
View File
@@ -0,0 +1,219 @@
// Package probe extracts audio stream details with ffprobe and ranks them.
package probe
import (
"context"
"encoding/json"
"fmt"
"math"
"os/exec"
"strconv"
"strings"
"time"
)
// Info describes the primary audio stream of a file.
type Info struct {
Codec string `json:"codec"`
CodecLong string `json:"codec_long,omitempty"`
Format string `json:"format,omitempty"`
Lossless bool `json:"lossless"`
Bitrate int64 `json:"bitrate"` // bits per second
SampleRate int `json:"sample_rate"`
BitDepth int `json:"bit_depth,omitempty"`
Channels int `json:"channels"`
Duration float64 `json:"duration"`
Title string `json:"title,omitempty"`
Artist string `json:"artist,omitempty"`
Album string `json:"album,omitempty"`
Error string `json:"error,omitempty"`
}
// Prober runs ffprobe. An empty Bin disables probing.
type Prober struct {
Bin string
Timeout time.Duration
}
type ffOut struct {
Streams []struct {
CodecName string `json:"codec_name"`
CodecLongName string `json:"codec_long_name"`
SampleRate string `json:"sample_rate"`
Channels int `json:"channels"`
BitsPerRawSample string `json:"bits_per_raw_sample"`
BitsPerSample int `json:"bits_per_sample"`
SampleFmt string `json:"sample_fmt"`
BitRate string `json:"bit_rate"`
Duration string `json:"duration"`
} `json:"streams"`
Format struct {
FormatName string `json:"format_name"`
Duration string `json:"duration"`
BitRate string `json:"bit_rate"`
Tags map[string]string `json:"tags"`
} `json:"format"`
}
// Probe inspects path. Failures are reported in Info.Error rather than as an
// error so a single unreadable file never aborts a scan.
func (p Prober) Probe(ctx context.Context, path string) Info {
if p.Bin == "" {
return Info{Error: "ffprobe not available"}
}
to := p.Timeout
if to == 0 {
to = 30 * time.Second
}
ctx, cancel := context.WithTimeout(ctx, to)
defer cancel()
out, err := exec.CommandContext(ctx, p.Bin, "-v", "error", "-print_format", "json",
"-show_format", "-show_streams", "-select_streams", "a:0", "--", path).Output()
if err != nil {
return Info{Error: fmt.Sprintf("ffprobe: %v", err)}
}
var ff ffOut
if err := json.Unmarshal(out, &ff); err != nil {
return Info{Error: fmt.Sprintf("ffprobe output: %v", err)}
}
if len(ff.Streams) == 0 {
return Info{Error: "no audio stream"}
}
s := ff.Streams[0]
in := Info{
Codec: s.CodecName,
CodecLong: s.CodecLongName,
Format: ff.Format.FormatName,
Channels: s.Channels,
Lossless: IsLossless(s.CodecName),
}
in.SampleRate, _ = strconv.Atoi(s.SampleRate)
in.BitDepth, _ = strconv.Atoi(s.BitsPerRawSample)
if in.BitDepth == 0 {
in.BitDepth = s.BitsPerSample
}
if in.BitDepth == 0 && in.Lossless {
in.BitDepth = depthFromSampleFmt(s.SampleFmt)
}
in.Bitrate, _ = strconv.ParseInt(s.BitRate, 10, 64)
if in.Bitrate == 0 {
in.Bitrate, _ = strconv.ParseInt(ff.Format.BitRate, 10, 64)
}
in.Duration, _ = strconv.ParseFloat(ff.Format.Duration, 64)
if in.Duration == 0 {
in.Duration, _ = strconv.ParseFloat(s.Duration, 64)
}
for k, v := range ff.Format.Tags {
switch strings.ToLower(k) {
case "title":
in.Title = v
case "artist":
in.Artist = v
case "album":
in.Album = v
}
}
return in
}
func depthFromSampleFmt(f string) int {
switch strings.TrimSuffix(f, "p") {
case "s16":
return 16
case "s32", "flt":
return 24 // FLAC/ALAC decode 24-bit into s32
case "u8":
return 8
}
return 0
}
// IsLossless reports whether an ffprobe codec name is a lossless codec.
func IsLossless(codec string) bool {
switch {
case codec == "flac", codec == "alac", codec == "ape", codec == "wavpack",
codec == "tta", codec == "mlp", codec == "truehd", codec == "wmalossless",
strings.HasPrefix(codec, "pcm_"), strings.HasPrefix(codec, "dsd_"):
return true
}
return false
}
// codecEfficiency scales a lossy bitrate into "MP3-equivalent" kbps so a 160k
// Opus file is not ranked below a 192k MP3.
func codecEfficiency(codec string) float64 {
switch codec {
case "opus":
return 1.6
case "aac", "aac_latm":
return 1.3
case "vorbis":
return 1.3
case "mp3", "mp3float":
return 1.0
case "wmav2", "wmav1":
return 0.9
}
return 1.0
}
// Score returns a comparable quality number. Lossless always beats lossy.
func Score(in Info) float64 {
if in.Error != "" || in.Codec == "" {
return -1
}
if in.Lossless {
depth := in.BitDepth
if depth == 0 {
depth = 16
}
return 1e6 + float64(depth)*1e3 + float64(in.SampleRate)/1e3
}
return float64(in.Bitrate) / 1000 * codecEfficiency(in.Codec)
}
// Verdict compares host and slave quality.
type Verdict struct {
// Better is "host", "slave", "equal" or "unknown".
Better string `json:"better"`
// DurationDelta is |host-slave| in seconds; large deltas suggest a
// different edit or version of the song rather than a re-encode.
DurationDelta float64 `json:"duration_delta"`
Notes []string `json:"notes"`
}
// Compare ranks host against slave. Within 5% is treated as equal.
func Compare(host, slave Info) Verdict {
v := Verdict{Better: "unknown"}
if host.Error != "" || slave.Error != "" {
v.Notes = append(v.Notes, "could not probe one of the files")
return v
}
if host.Duration > 0 && slave.Duration > 0 {
v.DurationDelta = math.Abs(host.Duration - slave.Duration)
if v.DurationDelta > 2 {
v.Notes = append(v.Notes, fmt.Sprintf("durations differ by %.1fs: possibly a different version or edit", v.DurationDelta))
}
}
hs, ss := Score(host), Score(slave)
switch {
case hs <= 0 || ss <= 0:
case math.Abs(hs-ss) <= 0.05*math.Max(hs, ss):
v.Better = "equal"
case hs > ss:
v.Better = "host"
default:
v.Better = "slave"
}
if host.Lossless != slave.Lossless {
if host.Lossless {
v.Notes = append(v.Notes, "host is lossless, slave is lossy")
} else {
v.Notes = append(v.Notes, "slave is lossless but host is lossy: linking would downgrade the slave")
}
}
if host.Channels != slave.Channels && host.Channels > 0 && slave.Channels > 0 {
v.Notes = append(v.Notes, fmt.Sprintf("channel count differs (%d vs %d)", host.Channels, slave.Channels))
}
return v
}
+57
View File
@@ -0,0 +1,57 @@
package probe
import (
"context"
"os/exec"
"path/filepath"
"testing"
)
func TestScoreOrdering(t *testing.T) {
flac24 := Info{Codec: "flac", Lossless: true, BitDepth: 24, SampleRate: 96000}
flac16 := Info{Codec: "flac", Lossless: true, BitDepth: 16, SampleRate: 44100}
opus160 := Info{Codec: "opus", Bitrate: 160_000}
mp3320 := Info{Codec: "mp3", Bitrate: 320_000}
mp3128 := Info{Codec: "mp3", Bitrate: 128_000}
order := []Info{flac24, flac16, mp3320, opus160, mp3128}
for i := 1; i < len(order); i++ {
if Score(order[i-1]) <= Score(order[i]) {
t.Errorf("%v should outrank %v", order[i-1], order[i])
}
}
if v := Compare(mp3128, flac16); v.Better != "slave" {
t.Errorf("lossless slave should win, got %s", v.Better)
}
if v := Compare(mp3320, Info{Codec: "mp3", Bitrate: 310_000}); v.Better != "equal" {
t.Errorf("within 5%% should be equal, got %s", v.Better)
}
}
func TestProbeRealFiles(t *testing.T) {
ffmpeg, err1 := exec.LookPath("ffmpeg")
ffprobe, err2 := exec.LookPath("ffprobe")
if err1 != nil || err2 != nil {
t.Skip("ffmpeg/ffprobe not installed")
}
dir := t.TempDir()
flac, mp3 := filepath.Join(dir, "a.flac"), filepath.Join(dir, "a.mp3")
for _, args := range [][]string{
{"-f", "lavfi", "-i", "sine=f=440:d=2", "-sample_fmt", "s16", flac},
{"-f", "lavfi", "-i", "sine=f=440:d=2", "-b:a", "128k", mp3},
} {
if out, err := exec.Command(ffmpeg, append([]string{"-v", "error", "-y"}, args...)...).CombinedOutput(); err != nil {
t.Fatalf("ffmpeg: %v %s", err, out)
}
}
p := Prober{Bin: ffprobe}
fi, mi := p.Probe(context.Background(), flac), p.Probe(context.Background(), mp3)
if fi.Error != "" || mi.Error != "" {
t.Fatalf("probe errors: %q %q", fi.Error, mi.Error)
}
if !fi.Lossless || fi.BitDepth != 16 || mi.Lossless || mi.Codec != "mp3" {
t.Fatalf("unexpected probe results %+v / %+v", fi, mi)
}
if v := Compare(mi, fi); v.Better != "slave" {
t.Fatalf("flac slave should beat mp3 host, got %+v", v)
}
}
+651
View File
@@ -0,0 +1,651 @@
package store
import (
"context"
"database/sql"
"encoding/json"
"errors"
"strings"
)
var ErrNotFound = errors.New("not found")
// ---- users & sessions ----
type User struct {
ID int64 `json:"id"`
Username string `json:"username"`
PasswordHash string `json:"-"`
}
func (s *Store) UserCount(ctx context.Context) (int, error) {
var n int
err := s.DB.QueryRowContext(ctx, `SELECT COUNT(*) FROM users`).Scan(&n)
return n, err
}
// CreateFirstUser inserts a user only if none exist (atomic setup).
func (s *Store) CreateFirstUser(ctx context.Context, username, hash string) error {
res, err := s.DB.ExecContext(ctx, `INSERT INTO users (username, password_hash, created_at)
SELECT ?, ?, ? WHERE NOT EXISTS (SELECT 1 FROM users)`, username, hash, now())
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return errors.New("setup already completed")
}
return nil
}
func (s *Store) UserByName(ctx context.Context, name string) (User, error) {
var u User
err := s.DB.QueryRowContext(ctx, `SELECT id, username, password_hash FROM users WHERE username = ?`, name).
Scan(&u.ID, &u.Username, &u.PasswordHash)
if errors.Is(err, sql.ErrNoRows) {
return u, ErrNotFound
}
return u, err
}
func (s *Store) UserByID(ctx context.Context, id int64) (User, error) {
var u User
err := s.DB.QueryRowContext(ctx, `SELECT id, username, password_hash FROM users WHERE id = ?`, id).
Scan(&u.ID, &u.Username, &u.PasswordHash)
if errors.Is(err, sql.ErrNoRows) {
return u, ErrNotFound
}
return u, err
}
func (s *Store) SetPassword(ctx context.Context, id int64, hash string) error {
_, err := s.DB.ExecContext(ctx, `UPDATE users SET password_hash = ? WHERE id = ?`, hash, id)
return err
}
func (s *Store) CreateSession(ctx context.Context, tokenHash string, userID, expires int64) error {
_, err := s.DB.ExecContext(ctx, `INSERT INTO sessions (token_hash, user_id, expires_at, created_at) VALUES (?, ?, ?, ?)`,
tokenHash, userID, expires, now())
return err
}
func (s *Store) SessionUser(ctx context.Context, tokenHash string) (int64, error) {
var uid int64
err := s.DB.QueryRowContext(ctx, `SELECT user_id FROM sessions WHERE token_hash = ? AND expires_at > ?`, tokenHash, now()).Scan(&uid)
if errors.Is(err, sql.ErrNoRows) {
return 0, ErrNotFound
}
return uid, err
}
func (s *Store) DeleteSession(ctx context.Context, tokenHash string) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM sessions WHERE token_hash = ?`, tokenHash)
return err
}
func (s *Store) DeleteUserSessions(ctx context.Context, userID int64, except string) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM sessions WHERE user_id = ? AND token_hash != ?`, userID, except)
return err
}
func (s *Store) PruneSessions(ctx context.Context) {
_, _ = s.DB.ExecContext(ctx, `DELETE FROM sessions WHERE expires_at <= ?`, now())
}
// ---- pairs ----
type Pair struct {
ID int64 `json:"id"`
Name string `json:"name"`
HostRoot string `json:"host_root"`
SlaveRoot string `json:"slave_root"`
QuarantineDir string `json:"quarantine_dir"`
Extensions []string `json:"extensions"`
AutoLinkIdentical bool `json:"auto_link_identical"`
ScanIntervalMinutes int `json:"scan_interval_minutes"`
RetentionDays int `json:"retention_days"`
SettleMinutes int `json:"settle_minutes"`
LastScanAt *int64 `json:"last_scan_at"`
LastScanError string `json:"last_scan_error"`
LastScanWarnings []string `json:"last_scan_warnings"`
CreatedAt int64 `json:"created_at"`
}
const pairCols = `id, name, host_root, slave_root, quarantine_dir, extensions, auto_link_identical,
scan_interval_minutes, retention_days, settle_minutes, last_scan_at, last_scan_error, last_scan_warnings, created_at`
type scanner interface{ Scan(...any) error }
func scanPair(r scanner) (Pair, error) {
var p Pair
var exts, warns string
var last sql.NullInt64
err := r.Scan(&p.ID, &p.Name, &p.HostRoot, &p.SlaveRoot, &p.QuarantineDir, &exts, &p.AutoLinkIdentical,
&p.ScanIntervalMinutes, &p.RetentionDays, &p.SettleMinutes, &last, &p.LastScanError, &warns, &p.CreatedAt)
if err != nil {
return p, err
}
if last.Valid {
p.LastScanAt = &last.Int64
}
p.Extensions = strings.Split(exts, ",")
_ = json.Unmarshal([]byte(warns), &p.LastScanWarnings)
if p.LastScanWarnings == nil {
p.LastScanWarnings = []string{}
}
return p, nil
}
func (s *Store) Pairs(ctx context.Context) ([]Pair, error) {
rows, err := s.DB.QueryContext(ctx, `SELECT `+pairCols+` FROM pairs ORDER BY id`)
if err != nil {
return nil, err
}
defer rows.Close()
out := []Pair{}
for rows.Next() {
p, err := scanPair(rows)
if err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
}
func (s *Store) Pair(ctx context.Context, id int64) (Pair, error) {
p, err := scanPair(s.DB.QueryRowContext(ctx, `SELECT `+pairCols+` FROM pairs WHERE id = ?`, id))
if errors.Is(err, sql.ErrNoRows) {
return p, ErrNotFound
}
return p, err
}
func (s *Store) SavePair(ctx context.Context, p *Pair) error {
exts := strings.Join(p.Extensions, ",")
if p.ID == 0 {
res, err := s.DB.ExecContext(ctx, `INSERT INTO pairs (name, host_root, slave_root, quarantine_dir, extensions,
auto_link_identical, scan_interval_minutes, retention_days, settle_minutes, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
p.Name, p.HostRoot, p.SlaveRoot, p.QuarantineDir, exts, p.AutoLinkIdentical,
p.ScanIntervalMinutes, p.RetentionDays, p.SettleMinutes, now())
if err != nil {
return err
}
p.ID, _ = res.LastInsertId()
return nil
}
_, err := s.DB.ExecContext(ctx, `UPDATE pairs SET name = ?, host_root = ?, slave_root = ?, quarantine_dir = ?,
extensions = ?, auto_link_identical = ?, scan_interval_minutes = ?, retention_days = ?, settle_minutes = ?
WHERE id = ?`,
p.Name, p.HostRoot, p.SlaveRoot, p.QuarantineDir, exts, p.AutoLinkIdentical,
p.ScanIntervalMinutes, p.RetentionDays, p.SettleMinutes, p.ID)
return err
}
func (s *Store) DeletePair(ctx context.Context, id int64) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM pairs WHERE id = ?`, id)
return err
}
func (s *Store) ClearMatches(ctx context.Context, pairID int64) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM matches WHERE pair_id = ?`, pairID)
return err
}
func (s *Store) FinishScan(ctx context.Context, pairID, at int64, scanErr string, warnings []string) error {
if warnings == nil {
warnings = []string{}
}
w, _ := json.Marshal(warnings)
_, err := s.DB.ExecContext(ctx, `UPDATE pairs SET last_scan_at = ?, last_scan_error = ?, last_scan_warnings = ? WHERE id = ?`,
at, scanErr, string(w), pairID)
return err
}
// ---- matches ----
// Match statuses.
const (
StatusLinked = "linked" // already the same inode
StatusIdentical = "identical" // byte-identical, safe to auto-link
StatusDifferent = "different" // same path, different content: manual review
StatusSimilar = "similar" // similar name (other extension / normalized title): manual review
StatusUnmatched = "unmatched" // exists only in the slave library
StatusError = "error" // cannot be linked (e.g. other filesystem)
)
// Match kinds (how the host candidate was found).
const (
KindPath = "path" // identical relative path
KindPathCase = "path_case" // relative path differs only by case
KindExtension = "extension" // same directory and title, different extension
KindFuzzy = "fuzzy" // same directory, normalized title matches
KindElsewhere = "elsewhere" // same artist, different album folder
KindNone = "none"
)
type Match struct {
ID int64 `json:"id"`
PairID int64 `json:"pair_id"`
RelSlave string `json:"rel_slave"`
RelHost string `json:"rel_host"`
Kind string `json:"kind"`
Status string `json:"status"`
HostIno uint64 `json:"-"`
HostSize int64 `json:"host_size"`
HostMtime int64 `json:"host_mtime"`
SlaveIno uint64 `json:"-"`
SlaveSize int64 `json:"slave_size"`
SlaveMtime int64 `json:"slave_mtime"`
HostProbe json.RawMessage `json:"host_probe,omitempty"`
SlaveProbe json.RawMessage `json:"slave_probe,omitempty"`
Verdict json.RawMessage `json:"verdict,omitempty"`
Note string `json:"note"`
Ignored bool `json:"ignored"`
UpdatedAt int64 `json:"updated_at"`
}
const matchCols = `id, pair_id, rel_slave, rel_host, kind, status, host_ino, host_size, host_mtime,
slave_ino, slave_size, slave_mtime, host_probe, slave_probe, verdict, note, ignored, updated_at`
func scanMatch(r scanner) (Match, error) {
var m Match
var hp, sp, v string
err := r.Scan(&m.ID, &m.PairID, &m.RelSlave, &m.RelHost, &m.Kind, &m.Status, &m.HostIno, &m.HostSize, &m.HostMtime,
&m.SlaveIno, &m.SlaveSize, &m.SlaveMtime, &hp, &sp, &v, &m.Note, &m.Ignored, &m.UpdatedAt)
if hp != "" {
m.HostProbe = json.RawMessage(hp)
}
if sp != "" {
m.SlaveProbe = json.RawMessage(sp)
}
if v != "" {
m.Verdict = json.RawMessage(v)
}
return m, err
}
func (s *Store) Match(ctx context.Context, id int64) (Match, error) {
m, err := scanMatch(s.DB.QueryRowContext(ctx, `SELECT `+matchCols+` FROM matches WHERE id = ?`, id))
if errors.Is(err, sql.ErrNoRows) {
return m, ErrNotFound
}
return m, err
}
func (s *Store) MatchBySlave(ctx context.Context, pairID int64, relSlave string) (Match, error) {
m, err := scanMatch(s.DB.QueryRowContext(ctx, `SELECT `+matchCols+` FROM matches WHERE pair_id = ? AND rel_slave = ?`, pairID, relSlave))
if errors.Is(err, sql.ErrNoRows) {
return m, ErrNotFound
}
return m, err
}
// AllMatches loads every match of a pair keyed by slave path (used by scans
// to reuse comparison and probe results for unchanged files).
func (s *Store) AllMatches(ctx context.Context, pairID int64) (map[string]Match, error) {
rows, err := s.DB.QueryContext(ctx, `SELECT `+matchCols+` FROM matches WHERE pair_id = ?`, pairID)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[string]Match{}
for rows.Next() {
m, err := scanMatch(rows)
if err != nil {
return nil, err
}
out[m.RelSlave] = m
}
return out, rows.Err()
}
type MatchFilter struct {
Status string // a status, "review" (different+similar), "ignored" or ""
Query string
Limit int
Offset int
}
func (s *Store) Matches(ctx context.Context, pairID int64, f MatchFilter) ([]Match, int, error) {
where := []string{"pair_id = ?"}
args := []any{pairID}
switch f.Status {
case "":
case "review":
where = append(where, "status IN ('different','similar') AND ignored = 0")
case "ignored":
where = append(where, "ignored = 1")
default:
where = append(where, "status = ? AND ignored = 0")
args = append(args, f.Status)
}
if f.Query != "" {
where = append(where, "(rel_slave LIKE ? ESCAPE '\\' OR rel_host LIKE ? ESCAPE '\\')")
q := "%" + strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(f.Query) + "%"
args = append(args, q, q)
}
w := strings.Join(where, " AND ")
var total int
if err := s.DB.QueryRowContext(ctx, `SELECT COUNT(*) FROM matches WHERE `+w, args...).Scan(&total); err != nil {
return nil, 0, err
}
if f.Limit <= 0 || f.Limit > 500 {
f.Limit = 100
}
rows, err := s.DB.QueryContext(ctx, `SELECT `+matchCols+` FROM matches WHERE `+w+
` ORDER BY rel_slave LIMIT ? OFFSET ?`, append(args, f.Limit, f.Offset)...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
out := []Match{}
for rows.Next() {
m, err := scanMatch(rows)
if err != nil {
return nil, 0, err
}
out = append(out, m)
}
return out, total, rows.Err()
}
// IdenticalIDs lists linkable identical matches (not ignored).
func (s *Store) IdenticalIDs(ctx context.Context, pairID int64) ([]int64, error) {
rows, err := s.DB.QueryContext(ctx, `SELECT id FROM matches WHERE pair_id = ? AND status = 'identical' AND ignored = 0 ORDER BY rel_slave`, pairID)
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 (s *Store) UpsertMatch(ctx context.Context, m *Match) error {
nullable := func(r json.RawMessage) string { return string(r) }
_, err := s.DB.ExecContext(ctx, `INSERT INTO matches (pair_id, rel_slave, rel_host, kind, status, host_ino, host_size,
host_mtime, slave_ino, slave_size, slave_mtime, host_probe, slave_probe, verdict, note, ignored, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(pair_id, rel_slave) DO UPDATE SET rel_host = excluded.rel_host, kind = excluded.kind,
status = excluded.status, host_ino = excluded.host_ino, host_size = excluded.host_size,
host_mtime = excluded.host_mtime, slave_ino = excluded.slave_ino, slave_size = excluded.slave_size,
slave_mtime = excluded.slave_mtime, host_probe = excluded.host_probe, slave_probe = excluded.slave_probe,
verdict = excluded.verdict, note = excluded.note, ignored = excluded.ignored, updated_at = excluded.updated_at`,
m.PairID, m.RelSlave, m.RelHost, m.Kind, m.Status, m.HostIno, m.HostSize, m.HostMtime, m.SlaveIno, m.SlaveSize,
m.SlaveMtime, nullable(m.HostProbe), nullable(m.SlaveProbe), nullable(m.Verdict), m.Note, m.Ignored, now())
return err
}
func (s *Store) DeleteMatch(ctx context.Context, pairID int64, relSlave string) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM matches WHERE pair_id = ? AND rel_slave = ?`, pairID, relSlave)
return err
}
// DeleteMatchesExcept removes rows for slave files that disappeared.
func (s *Store) DeleteMatchesExcept(ctx context.Context, pairID int64, keep map[string]bool) error {
existing, err := s.AllMatches(ctx, pairID)
if err != nil {
return err
}
for rel := range existing {
if !keep[rel] {
if err := s.DeleteMatch(ctx, pairID, rel); err != nil {
return err
}
}
}
return nil
}
func (s *Store) SetIgnored(ctx context.Context, id int64, ignored bool) error {
_, err := s.DB.ExecContext(ctx, `UPDATE matches SET ignored = ? WHERE id = ?`, ignored, id)
return err
}
// LinkedHostPaths returns host paths already linked into the slave under prefix.
func (s *Store) LinkedHostPaths(ctx context.Context, pairID int64, prefix string) ([]string, error) {
q := `SELECT rel_host FROM matches WHERE pair_id = ? AND status = 'linked'`
args := []any{pairID}
if prefix != "" {
q += ` AND rel_host LIKE ? ESCAPE '\'`
args = append(args, strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(prefix)+"/%")
}
rows, err := s.DB.QueryContext(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []string
for rows.Next() {
var r string
if err := rows.Scan(&r); err != nil {
return nil, err
}
out = append(out, r)
}
return out, rows.Err()
}
type Summary struct {
Counts map[string]int `json:"counts"`
Review int `json:"review"`
Ignored int `json:"ignored"`
ReclaimableSize int64 `json:"reclaimable_bytes"`
LinkedSize int64 `json:"linked_bytes"`
}
func (s *Store) Summary(ctx context.Context, pairID int64) (Summary, error) {
sum := Summary{Counts: map[string]int{}}
rows, err := s.DB.QueryContext(ctx, `SELECT status, ignored, COUNT(*), COALESCE(SUM(slave_size), 0) FROM matches WHERE pair_id = ? GROUP BY status, ignored`, pairID)
if err != nil {
return sum, err
}
defer rows.Close()
for rows.Next() {
var st string
var ign bool
var n int
var size int64
if err := rows.Scan(&st, &ign, &n, &size); err != nil {
return sum, err
}
if ign {
sum.Ignored += n
continue
}
sum.Counts[st] += n
switch st {
case StatusIdentical:
sum.ReclaimableSize += size
case StatusLinked:
sum.LinkedSize += size
case StatusDifferent, StatusSimilar:
sum.Review += n
}
}
return sum, rows.Err()
}
// ---- operations ----
type Operation struct {
ID int64 `json:"id"`
PairID int64 `json:"pair_id"`
Action string `json:"action"`
RelHost string `json:"rel_host"`
RelSlave string `json:"rel_slave"`
RelNew string `json:"rel_new"`
QuarantinePath string `json:"quarantine_path"`
QuarantineSide string `json:"quarantine_side"`
TmpPath string `json:"-"`
Bytes int64 `json:"bytes"`
DryRun bool `json:"dry_run"`
Trigger string `json:"trigger"`
Status string `json:"status"`
Error string `json:"error"`
CreatedAt int64 `json:"created_at"`
FinishedAt *int64 `json:"finished_at"`
UndoneAt *int64 `json:"undone_at"`
PurgedAt *int64 `json:"purged_at"`
}
// Operation actions.
const (
ActLink = "link" // slave replaced by hardlink of host
ActPromote = "promote" // host replaced by slave's file, then linked
ActBuild = "build" // host file linked into the slave (fresh library)
ActRemove = "remove" // slave hardlink removed (host copy kept)
)
// Operation statuses.
const (
OpRunning = "running"
OpDone = "done"
OpFailed = "failed"
OpPlanned = "planned" // dry run
OpInterrupted = "interrupted"
)
const opCols = `id, pair_id, action, rel_host, rel_slave, rel_new, quarantine_path, quarantine_side, tmp_path, bytes,
dry_run, trigger, status, error, created_at, finished_at, undone_at, purged_at`
func scanOp(r scanner) (Operation, error) {
var o Operation
var fin, und, pur sql.NullInt64
err := r.Scan(&o.ID, &o.PairID, &o.Action, &o.RelHost, &o.RelSlave, &o.RelNew, &o.QuarantinePath, &o.QuarantineSide,
&o.TmpPath, &o.Bytes, &o.DryRun, &o.Trigger, &o.Status, &o.Error, &o.CreatedAt, &fin, &und, &pur)
if fin.Valid {
o.FinishedAt = &fin.Int64
}
if und.Valid {
o.UndoneAt = &und.Int64
}
if pur.Valid {
o.PurgedAt = &pur.Int64
}
return o, err
}
func (s *Store) CreateOp(ctx context.Context, o *Operation) error {
o.CreatedAt = now()
res, err := s.DB.ExecContext(ctx, `INSERT INTO operations (pair_id, action, rel_host, rel_slave, rel_new, quarantine_path,
quarantine_side, tmp_path, bytes, dry_run, trigger, status, error, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
o.PairID, o.Action, o.RelHost, o.RelSlave, o.RelNew, o.QuarantinePath, o.QuarantineSide, o.TmpPath, o.Bytes,
o.DryRun, o.Trigger, o.Status, o.Error, o.CreatedAt)
if err != nil {
return err
}
o.ID, _ = res.LastInsertId()
return nil
}
func (s *Store) FinishOp(ctx context.Context, id int64, status, errMsg string) error {
_, err := s.DB.ExecContext(ctx, `UPDATE operations SET status = ?, error = ?, tmp_path = '', finished_at = ? WHERE id = ?`,
status, errMsg, now(), id)
return err
}
func (s *Store) MarkUndone(ctx context.Context, id int64) error {
_, err := s.DB.ExecContext(ctx, `UPDATE operations SET undone_at = ? WHERE id = ?`, now(), id)
return err
}
func (s *Store) MarkPurged(ctx context.Context, id int64) error {
_, err := s.DB.ExecContext(ctx, `UPDATE operations SET purged_at = ? WHERE id = ?`, now(), id)
return err
}
func (s *Store) Op(ctx context.Context, id int64) (Operation, error) {
o, err := scanOp(s.DB.QueryRowContext(ctx, `SELECT `+opCols+` FROM operations WHERE id = ?`, id))
if errors.Is(err, sql.ErrNoRows) {
return o, ErrNotFound
}
return o, err
}
type OpFilter struct {
PairID int64
Quarantine bool // only ops holding a live quarantined file
Limit int
Offset int
}
func (s *Store) Ops(ctx context.Context, f OpFilter) ([]Operation, int, error) {
where := []string{"1=1"}
var args []any
if f.PairID != 0 {
where = append(where, "pair_id = ?")
args = append(args, f.PairID)
}
if f.Quarantine {
where = append(where, "quarantine_path != '' AND purged_at IS NULL AND undone_at IS NULL AND status IN ('done','interrupted')")
}
w := strings.Join(where, " AND ")
var total int
if err := s.DB.QueryRowContext(ctx, `SELECT COUNT(*) FROM operations WHERE `+w, args...).Scan(&total); err != nil {
return nil, 0, err
}
if f.Limit <= 0 || f.Limit > 500 {
f.Limit = 100
}
rows, err := s.DB.QueryContext(ctx, `SELECT `+opCols+` FROM operations WHERE `+w+` ORDER BY id DESC LIMIT ? OFFSET ?`,
append(args, f.Limit, f.Offset)...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
out := []Operation{}
for rows.Next() {
o, err := scanOp(rows)
if err != nil {
return nil, 0, err
}
out = append(out, o)
}
return out, total, rows.Err()
}
// RunningOps returns operations left in the running state (after a crash).
func (s *Store) RunningOps(ctx context.Context) ([]Operation, error) {
rows, err := s.DB.QueryContext(ctx, `SELECT `+opCols+` FROM operations WHERE status = 'running'`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []Operation
for rows.Next() {
o, err := scanOp(rows)
if err != nil {
return nil, err
}
out = append(out, o)
}
return out, rows.Err()
}
// ExpiredQuarantine returns quarantine-holding ops older than cutoff.
func (s *Store) ExpiredQuarantine(ctx context.Context, pairID, cutoff int64) ([]Operation, error) {
rows, err := s.DB.QueryContext(ctx, `SELECT `+opCols+` FROM operations WHERE pair_id = ? AND quarantine_path != ''
AND purged_at IS NULL AND undone_at IS NULL AND status IN ('done','interrupted') AND created_at < ?`, pairID, cutoff)
if err != nil {
return nil, err
}
defer rows.Close()
var out []Operation
for rows.Next() {
o, err := scanOp(rows)
if err != nil {
return nil, err
}
out = append(out, o)
}
return out, rows.Err()
}
+162
View File
@@ -0,0 +1,162 @@
// Package store persists playlink state in SQLite.
package store
import (
"context"
"database/sql"
"fmt"
"path/filepath"
"time"
_ "modernc.org/sqlite"
)
type Store struct {
DB *sql.DB
}
var migrations = []string{
`CREATE TABLE users (
id INTEGER PRIMARY KEY,
username TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
created_at INTEGER NOT NULL
);
CREATE TABLE sessions (
token_hash TEXT PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
expires_at INTEGER NOT NULL,
created_at INTEGER NOT NULL
);
CREATE TABLE settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE pairs (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
host_root TEXT NOT NULL,
slave_root TEXT NOT NULL,
quarantine_dir TEXT NOT NULL,
extensions TEXT NOT NULL,
auto_link_identical INTEGER NOT NULL DEFAULT 0,
scan_interval_minutes INTEGER NOT NULL DEFAULT 0,
retention_days INTEGER NOT NULL DEFAULT 30,
settle_minutes INTEGER NOT NULL DEFAULT 10,
last_scan_at INTEGER,
last_scan_error TEXT NOT NULL DEFAULT '',
last_scan_warnings TEXT NOT NULL DEFAULT '[]',
created_at INTEGER NOT NULL
);
CREATE TABLE matches (
id INTEGER PRIMARY KEY,
pair_id INTEGER NOT NULL REFERENCES pairs(id) ON DELETE CASCADE,
rel_slave TEXT NOT NULL,
rel_host TEXT NOT NULL DEFAULT '',
kind TEXT NOT NULL,
status TEXT NOT NULL,
host_ino INTEGER NOT NULL DEFAULT 0,
host_size INTEGER NOT NULL DEFAULT 0,
host_mtime INTEGER NOT NULL DEFAULT 0,
slave_ino INTEGER NOT NULL DEFAULT 0,
slave_size INTEGER NOT NULL DEFAULT 0,
slave_mtime INTEGER NOT NULL DEFAULT 0,
host_probe TEXT NOT NULL DEFAULT '',
slave_probe TEXT NOT NULL DEFAULT '',
verdict TEXT NOT NULL DEFAULT '',
note TEXT NOT NULL DEFAULT '',
ignored INTEGER NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL,
UNIQUE(pair_id, rel_slave)
);
CREATE INDEX matches_pair_status ON matches(pair_id, status);
CREATE TABLE operations (
id INTEGER PRIMARY KEY,
pair_id INTEGER NOT NULL REFERENCES pairs(id) ON DELETE CASCADE,
action TEXT NOT NULL,
rel_host TEXT NOT NULL DEFAULT '',
rel_slave TEXT NOT NULL DEFAULT '',
rel_new TEXT NOT NULL DEFAULT '',
quarantine_path TEXT NOT NULL DEFAULT '',
quarantine_side TEXT NOT NULL DEFAULT '',
tmp_path TEXT NOT NULL DEFAULT '',
bytes INTEGER NOT NULL DEFAULT 0,
dry_run INTEGER NOT NULL DEFAULT 0,
trigger TEXT NOT NULL DEFAULT 'manual',
status TEXT NOT NULL,
error TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL,
finished_at INTEGER,
undone_at INTEGER,
purged_at INTEGER
);
CREATE INDEX operations_pair ON operations(pair_id, created_at DESC);`,
}
// Open opens (and migrates) the database in dataDir.
func Open(dataDir string) (*Store, error) {
dsn := "file:" + filepath.Join(dataDir, "playlink.db") +
"?_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_pragma=busy_timeout(10000)&_pragma=synchronous(NORMAL)"
db, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, err
}
db.SetMaxOpenConns(1) // SQLite: one writer; keeps things simple and safe.
s := &Store{DB: db}
if err := s.migrate(); err != nil {
db.Close()
return nil, err
}
return s, nil
}
func (s *Store) migrate() error {
if _, err := s.DB.Exec(`CREATE TABLE IF NOT EXISTS schema_version (v INTEGER NOT NULL)`); err != nil {
return err
}
var v int
if err := s.DB.QueryRow(`SELECT COALESCE(MAX(v), 0) FROM schema_version`).Scan(&v); err != nil {
return err
}
for i := v; i < len(migrations); i++ {
tx, err := s.DB.Begin()
if err != nil {
return err
}
if _, err := tx.Exec(migrations[i]); err != nil {
tx.Rollback()
return fmt.Errorf("migration %d: %w", i+1, err)
}
if _, err := tx.Exec(`INSERT INTO schema_version (v) VALUES (?)`, i+1); err != nil {
tx.Rollback()
return err
}
if err := tx.Commit(); err != nil {
return err
}
}
return nil
}
func now() int64 { return time.Now().Unix() }
// ---- settings ----
func (s *Store) Setting(ctx context.Context, key, def string) string {
var v string
if err := s.DB.QueryRowContext(ctx, `SELECT value FROM settings WHERE key = ?`, key).Scan(&v); err != nil {
return def
}
return v
}
func (s *Store) SetSetting(ctx context.Context, key, value string) error {
_, err := s.DB.ExecContext(ctx, `INSERT INTO settings (key, value) VALUES (?, ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value`, key, value)
return err
}
// DryRun is the global safety switch. It defaults to on for new installs.
func (s *Store) DryRun(ctx context.Context) bool {
return s.Setting(ctx, "dry_run", "1") == "1"
}