A per-pair "Approximate matching" setting adds a second pass for slave files the strict rules could not place. It searches the same album with disc folders (Disc 02, CD2, ...) ignored, then the top-level folder, and scores titles by edit distance and word containment against a configurable threshold (default 85%), with a small track-number bonus. Guards against false matches: version words (live, remix, reprise, instrumental, ...) must agree, near-tied candidates are left unmatched, and approximate matches always go to manual review, never auto-link. Each match records a human-readable reason. Review now flags matches where the master copy is better as upgrades, with an Upgrades filter and count and an "Upgrade to master copy" action. The redundant "master is lossless" warning note is dropped. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
670 lines
22 KiB
Go
670 lines
22 KiB
Go
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"`
|
|
MasterRoot string `json:"master_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"`
|
|
ApproxMatching bool `json:"approx_matching"`
|
|
ApproxThreshold int `json:"approx_threshold"` // percent
|
|
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, master_root, slave_root, quarantine_dir, extensions, auto_link_identical,
|
|
scan_interval_minutes, retention_days, settle_minutes, approx_matching, approx_threshold, 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.MasterRoot, &p.SlaveRoot, &p.QuarantineDir, &exts, &p.AutoLinkIdentical,
|
|
&p.ScanIntervalMinutes, &p.RetentionDays, &p.SettleMinutes, &p.ApproxMatching, &p.ApproxThreshold, &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, master_root, slave_root, quarantine_dir, extensions,
|
|
auto_link_identical, scan_interval_minutes, retention_days, settle_minutes, approx_matching, approx_threshold, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
p.Name, p.MasterRoot, p.SlaveRoot, p.QuarantineDir, exts, p.AutoLinkIdentical,
|
|
p.ScanIntervalMinutes, p.RetentionDays, p.SettleMinutes, p.ApproxMatching, p.ApproxThreshold, now())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
p.ID, _ = res.LastInsertId()
|
|
return nil
|
|
}
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE pairs SET name = ?, master_root = ?, slave_root = ?, quarantine_dir = ?,
|
|
extensions = ?, auto_link_identical = ?, scan_interval_minutes = ?, retention_days = ?, settle_minutes = ?,
|
|
approx_matching = ?, approx_threshold = ? WHERE id = ?`,
|
|
p.Name, p.MasterRoot, p.SlaveRoot, p.QuarantineDir, exts, p.AutoLinkIdentical,
|
|
p.ScanIntervalMinutes, p.RetentionDays, p.SettleMinutes, p.ApproxMatching, p.ApproxThreshold, 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 master 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 top-level folder and title, different subfolder
|
|
KindApprox = "approximate" // opt-in: similar title, disc folders ignored
|
|
KindNone = "none"
|
|
)
|
|
|
|
type Match struct {
|
|
ID int64 `json:"id"`
|
|
PairID int64 `json:"pair_id"`
|
|
RelSlave string `json:"rel_slave"`
|
|
RelMaster string `json:"rel_master"`
|
|
Kind string `json:"kind"`
|
|
Status string `json:"status"`
|
|
MasterIno uint64 `json:"-"`
|
|
MasterSize int64 `json:"master_size"`
|
|
MasterMtime int64 `json:"master_mtime"`
|
|
SlaveIno uint64 `json:"-"`
|
|
SlaveSize int64 `json:"slave_size"`
|
|
SlaveMtime int64 `json:"slave_mtime"`
|
|
MasterProbe json.RawMessage `json:"master_probe,omitempty"`
|
|
SlaveProbe json.RawMessage `json:"slave_probe,omitempty"`
|
|
Verdict json.RawMessage `json:"verdict,omitempty"`
|
|
Note string `json:"note"`
|
|
Reason string `json:"reason"` // why an approximate match was chosen
|
|
Ignored bool `json:"ignored"`
|
|
UpdatedAt int64 `json:"updated_at"`
|
|
}
|
|
|
|
const matchCols = `id, pair_id, rel_slave, rel_master, kind, status, master_ino, master_size, master_mtime,
|
|
slave_ino, slave_size, slave_mtime, master_probe, slave_probe, verdict, note, reason, ignored, updated_at`
|
|
|
|
func scanMatch(r scanner) (Match, error) {
|
|
var m Match
|
|
var mp, sp, v string
|
|
err := r.Scan(&m.ID, &m.PairID, &m.RelSlave, &m.RelMaster, &m.Kind, &m.Status, &m.MasterIno, &m.MasterSize, &m.MasterMtime,
|
|
&m.SlaveIno, &m.SlaveSize, &m.SlaveMtime, &mp, &sp, &v, &m.Note, &m.Reason, &m.Ignored, &m.UpdatedAt)
|
|
if mp != "" {
|
|
m.MasterProbe = json.RawMessage(mp)
|
|
}
|
|
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")
|
|
case "upgrade":
|
|
where = append(where, "status IN ('different','similar') AND ignored = 0 AND "+upgradeCond)
|
|
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_master 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_master, kind, status, master_ino, master_size,
|
|
master_mtime, slave_ino, slave_size, slave_mtime, master_probe, slave_probe, verdict, note, reason, ignored, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(pair_id, rel_slave) DO UPDATE SET rel_master = excluded.rel_master, kind = excluded.kind,
|
|
status = excluded.status, master_ino = excluded.master_ino, master_size = excluded.master_size,
|
|
master_mtime = excluded.master_mtime, slave_ino = excluded.slave_ino, slave_size = excluded.slave_size,
|
|
slave_mtime = excluded.slave_mtime, master_probe = excluded.master_probe, slave_probe = excluded.slave_probe,
|
|
verdict = excluded.verdict, note = excluded.note, reason = excluded.reason, ignored = excluded.ignored, updated_at = excluded.updated_at`,
|
|
m.PairID, m.RelSlave, m.RelMaster, m.Kind, m.Status, m.MasterIno, m.MasterSize, m.MasterMtime, m.SlaveIno, m.SlaveSize,
|
|
m.SlaveMtime, nullable(m.MasterProbe), nullable(m.SlaveProbe), nullable(m.Verdict), m.Note, m.Reason, 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
|
|
}
|
|
|
|
// LinkedMasterPaths returns master paths already linked into the slave under prefix.
|
|
func (s *Store) LinkedMasterPaths(ctx context.Context, pairID int64, prefix string) ([]string, error) {
|
|
q := `SELECT rel_master FROM matches WHERE pair_id = ? AND status = 'linked'`
|
|
args := []any{pairID}
|
|
if prefix != "" {
|
|
q += ` AND rel_master 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()
|
|
}
|
|
|
|
// upgradeCond selects review matches where the master copy scored better, so
|
|
// linking upgrades the slave.
|
|
const upgradeCond = `verdict LIKE '%"better":"master"%'`
|
|
|
|
type Summary struct {
|
|
Counts map[string]int `json:"counts"`
|
|
Review int `json:"review"`
|
|
Upgrades int `json:"upgrades"`
|
|
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
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return sum, err
|
|
}
|
|
err = s.DB.QueryRowContext(ctx, `SELECT COUNT(*) FROM matches WHERE pair_id = ? AND ignored = 0
|
|
AND status IN ('different','similar') AND `+upgradeCond, pairID).Scan(&sum.Upgrades)
|
|
return sum, err
|
|
}
|
|
|
|
// ---- operations ----
|
|
|
|
type Operation struct {
|
|
ID int64 `json:"id"`
|
|
PairID int64 `json:"pair_id"`
|
|
Action string `json:"action"`
|
|
RelMaster string `json:"rel_master"`
|
|
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 master
|
|
ActPromote = "promote" // master replaced by slave's file, then linked
|
|
ActBuild = "build" // master file linked into the slave (fresh library)
|
|
ActRemove = "remove" // slave hardlink removed (master copy kept)
|
|
)
|
|
|
|
// Operation statuses.
|
|
const (
|
|
OpRunning = "running"
|
|
OpDone = "done"
|
|
OpFailed = "failed"
|
|
OpPlanned = "planned" // dry run
|
|
OpInterrupted = "interrupted"
|
|
)
|
|
|
|
const opCols = `id, pair_id, action, rel_master, 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.RelMaster, &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_master, rel_slave, rel_new, quarantine_path,
|
|
quarantine_side, tmp_path, bytes, dry_run, trigger, status, error, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
o.PairID, o.Action, o.RelMaster, 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()
|
|
}
|