package importer import ( "context" "encoding/json" "errors" "fmt" "log/slog" "os" "path/filepath" "slices" "strconv" "sync" "time" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/acquire" "git.fabledsword.com/bvandeusen/minstrel/internal/catalogue" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/downloader" "git.fabledsword.com/bvandeusen/minstrel/internal/indexer" ) // Import states, as grabs.import_state holds them. const ( StateDownloading = "downloading" StateWaiting = "waiting" StateImporting = "importing" StateImported = "imported" StateRefused = "refused" StateReview = "review" StateHeld = "held" StateFailed = "failed" ) const ( // interval: a finished download is noticed within a minute. One // Torrents call per download client, so the client barely notices. interval = time.Minute // passTimeout bounds a pass: fingerprinting a long album against // AcoustID's pace takes a minute or two, so half an hour is a hang. passTimeout = 30 * time.Minute // staleMinutes: an import still "importing" after this was cut short // (a restart mid-copy) and is tried again. staleMinutes = 30 ) // Scanner reads new files into the library. type Scanner interface { ScanFiles(ctx context.Context, paths []string) ([]pgtype.UUID, error) } // Catalogue lists a release group's editions. type Catalogue interface { Releases(ctx context.Context, groupMBID string) (catalogue.Editions, error) } // Importer watches every live grab's download and imports it once finished. type Importer struct { pool *pgxpool.Pool Scanner Scanner Catalogue Catalogue Printer Printer // Roots are the library's folders (the scan paths), read on every use // because they change in Settings (#6129). New music is filed under the // one the settings name, or the first. Roots func() []string // DataDir is where cached art lives, for tidying an artist an upgrade // leaves with nothing (#5524). Empty skips it. DataDir string // ClientFor builds the download client a grab went to. Defaults to // acquire.DownloadClientFor. ClientFor func(ctx context.Context, code string) (downloader.Driver, error) logger *slog.Logger wake chan struct{} mu sync.Mutex // one pass at a time } // New returns an Importer. printer may be nil: the check then rests on // lengths and tags. func New(pool *pgxpool.Pool, scanner Scanner, cat Catalogue, printer Printer, roots func() []string, logger *slog.Logger) *Importer { if logger == nil { logger = slog.Default() } im := &Importer{pool: pool, Scanner: scanner, Catalogue: cat, Printer: printer, Roots: roots, logger: logger, wake: make(chan struct{}, 1)} im.ClientFor = func(ctx context.Context, code string) (downloader.Driver, error) { return acquire.DownloadClientFor(ctx, pool, code) } return im } // Wake asks for a pass now: after a grab, or a person's decision. func (im *Importer) Wake() { select { case im.wake <- struct{}{}: default: } } // Run passes every minute, and when woken, until ctx ends. func (im *Importer) Run(ctx context.Context) { t := time.NewTicker(interval) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: case <-im.wake: } rep := im.Pass(ctx) if rep.Imported+rep.Refused+rep.Review+rep.Held+rep.Failed > 0 || rep.Error != "" { im.logger.Info("import: pass", "imported", rep.Imported, "refused", rep.Refused, "review", rep.Review, "held", rep.Held, "failed", rep.Failed, "downloading", rep.Downloading, "error", rep.Error) } } } // Report is one pass. type Report struct { Downloading, Imported, Refused, Review, Held, Failed, Waiting int Error string } // Pass looks at every live grab once. func (im *Importer) Pass(ctx context.Context) Report { im.mu.Lock() defer im.mu.Unlock() ctx, cancel := context.WithTimeout(ctx, passTimeout) defer cancel() var rep Report q := dbq.New(im.pool) if _, err := q.ReclaimStaleImports(ctx, staleMinutes); err != nil { rep.Error = err.Error() return rep } grabs, err := q.ListLiveGrabs(ctx) if err != nil { rep.Error = err.Error() return rep } byClient := map[string][]dbq.ListLiveGrabsRow{} var order []string for _, g := range grabs { if _, ok := byClient[g.Client]; !ok { order = append(order, g.Client) } byClient[g.Client] = append(byClient[g.Client], g) } for _, code := range order { im.passClient(ctx, q, code, byClient[code], &rep) } return rep } func (im *Importer) passClient(ctx context.Context, q *dbq.Queries, code string, gs []dbq.ListLiveGrabsRow, rep *Report) { wait := func(g dbq.ListLiveGrabsRow, why string) { rep.Waiting++ _ = q.SetGrabProgress(ctx, dbq.SetGrabProgressParams{ ID: g.ID, Progress: g.Progress, ImportState: StateWaiting, ImportReason: why}) } d, err := im.ClientFor(ctx, code) if err != nil { for _, g := range gs { wait(g, fmt.Sprintf("Can't reach its download client: %v.", err)) } return } hashes := make([]string, 0, len(gs)) for _, g := range gs { hashes = append(hashes, g.InfoHash) } ts, err := d.Torrents(ctx, hashes...) if err != nil { for _, g := range gs { wait(g, fmt.Sprintf("Can't reach its download client %s: %s", code, indexer.FirstLine(err.Error()))) } return } byHash := map[string]downloader.Torrent{} for _, t := range ts { byHash[indexer.NormalizeHash(t.Hash)] = t } for _, g := range gs { if ctx.Err() != nil { return } switch g.ImportState { case StateReview, StateFailed, StateImporting: continue } t, ok := byHash[indexer.NormalizeHash(g.InfoHash)] if !ok { continue // gone from the client: the lifecycle settles it (#5523) } if !g.WantID.Valid { wait(g, "Its album is no longer wanted.") continue } if !finished(ctx, d, t) { rep.Downloading++ _ = q.SetGrabProgress(ctx, dbq.SetGrabProgressParams{ ID: g.ID, Progress: float32(min(t.Progress, 0.999)), ImportState: StateDownloading}) continue } claimed, err := q.ClaimGrabImport(ctx, g.ID) if err != nil { continue // claimed by a pass racing this one, or ended } switch out := im.importOne(ctx, q, claimed, d, t); out { case StateImported: rep.Imported++ case StateRefused: rep.Refused++ case StateReview: rep.Review++ case StateHeld: rep.Held++ case StateFailed: rep.Failed++ case StateWaiting: rep.Waiting++ } } } // upStates are qBittorrent's states for a torrent that has every byte and // is seeding, paused or queued to seed. Exchange's states arrive in the same // vocabulary. var upStates = map[string]bool{ "uploading": true, "stalledUP": true, "pausedUP": true, "stoppedUP": true, "queuedUP": true, "forcedUP": true, "checkingUP": true, } // finished: whole, in a seeding state, and no wanted file short of bytes. func finished(ctx context.Context, d downloader.Driver, t downloader.Torrent) bool { if !t.Done() || !upStates[t.State] { return false } if fp, ok := d.(downloader.FileProgress); ok { if states, err := fp.FileStates(ctx, t.Hash); err == nil { for _, f := range states { if f.Unfinished() { return false } } } } return true } // Detail is what grabs.import_detail keeps: the verdict and what it rested // on, the edition chosen, and what happened to the files. type Detail struct { Verdict Verdict `json:"verdict"` Edition string `json:"edition,omitempty"` EditionMBID string `json:"edition_mbid,omitempty"` Files []string `json:"files,omitempty"` Placed int `json:"placed,omitempty"` Skipped int `json:"skipped,omitempty"` Untagged int `json:"untagged,omitempty"` // RehomedFrom: the album it was grabbed for, when it turned out to be // another wanted album. RehomedFrom string `json:"rehomed_from,omitempty"` } // importOne checks and files one finished download and records the outcome. func (im *Importer) importOne(ctx context.Context, q *dbq.Queries, g dbq.Grab, d downloader.Driver, t downloader.Torrent) string { record := func(state, reason, end string, det Detail, files int, dest string) string { body, _ := json.Marshal(det) // The outcome is written even when the pass's context is ending: // a claimed grab left "importing" waits half an hour to be retried. wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 15*time.Second) defer cancel() if _, err := q.RecordImport(wctx, dbq.RecordImportParams{ ID: g.ID, ImportState: state, ImportReason: reason, ImportDetail: body, ImportedFiles: int32(files), Destination: dest, EndReason: end, }); err != nil { im.logger.Warn("import: record outcome", "grab", g.Name, "state", state, "err", err) } im.logger.Info("import: "+state, "grab", g.Name, "reason", reason) return state } w, err := im.want(ctx, q, g.WantID) if err != nil { return record(StateWaiting, "Its album could not be read: "+err.Error(), "", Detail{}, 0, "") } editions, err := im.editions(ctx, w.GroupMBID) if err != nil { return record(StateWaiting, "MusicBrainz could not be reached to check it against the album; trying again shortly.", "", Detail{}, 0, "") } src := t.ContentPath if src == "" { src = filepath.Join(g.SavePath, t.Name) } files, printed, err := inspect(ctx, src, im.Printer) if err != nil { return record(StateFailed, fmt.Sprintf("Couldn't read the download at %s: %v", src, err), "", Detail{}, 0, "") } v := Verify(w, editions, files, printed, g.ImportForced) det := Detail{Verdict: v} for _, f := range files { det.Files = append(det.Files, f.Rel) } if v.Edition >= 0 { det.Edition, det.EditionMBID = describe(editions[v.Edition]), editions[v.Edition].MBID } switch v.Outcome { case Review: return record(StateReview, v.Reason, "", det, 0, "") case Refuse: if v.Code == CodeWrongAlbum { if out, ok := im.rehome(ctx, q, g, d, w, files, printed, record); ok { return out } } im.refuse(ctx, q, g, d, w, v.Reason, pgtype.UUID{}) return record(StateRefused, v.Reason, StateRefused, det, 0, "") } return im.place(ctx, q, g, d, w, editions, files, v, det, record) } type recordFunc func(state, reason, end string, det Detail, files int, dest string) string // place files an accepted download and records it imported, or held when // the library already had every track. func (im *Importer) place(ctx context.Context, q *dbq.Queries, g dbq.Grab, d downloader.Driver, w Want, editions []Edition, files []File, v Verdict, det Detail, record recordFunc) string { root, err := im.root(ctx, q) if err != nil { return record(StateFailed, "Not imported: "+err.Error()+".", "", det, 0, "") } var e *Edition if v.Edition >= 0 { e = &editions[v.Edition] } plan := Plan(root, w, e, files, v.Pairs) if w.Upgrade != "" { return im.placeUpgrade(ctx, q, g, d, w, e, files, plan, v, det, record) } heldRec, heldPos, err := im.held(ctx, q, w, plan) if err != nil { return record(StateWaiting, "Couldn't read what the library holds: "+err.Error(), "", det, 0, "") } var placed []string for _, pl := range plan { if (pl.Recording != "" && heldRec[pl.Recording]) || heldPos[[2]int{pl.Disc, pl.Track}] { det.Skipped++ continue } tags := Tags{Title: pl.Title, Artist: w.Artist, Album: w.Title, AlbumArtist: w.Artist, Year: w.Year, Disc: pl.Disc, Discs: pl.Discs, Track: pl.Track, Tracks: pl.Tracks, GroupMBID: w.GroupMBID, RecordingMBID: pl.Recording, ArtistMBID: w.ArtistMBID} if e != nil { tags.ReleaseMBID = e.MBID } tagged, err := Place(ctx, files[pl.File].Path, pl.Dest, tags) if errors.Is(err, ErrOccupied) { det.Skipped++ continue } if err != nil { for _, p := range placed { _ = os.Remove(p) } return record(StateFailed, fmt.Sprintf("Copying into the library failed, and nothing was left behind: %v", err), "", det, 0, "") } placed = append(placed, pl.Dest) if !tagged { det.Untagged++ } } det.Placed = len(placed) if len(placed) == 0 { im.removeFromClient(ctx, d, g) return record(StateHeld, fmt.Sprintf("The library already holds every track of %s, so nothing was imported.", w.Title), StateHeld, det, 0, "") } if _, err := im.Scanner.ScanFiles(ctx, placed); err != nil { // The watcher and the next scan find the files anyway. im.logger.Warn("import: scan the new files", "grab", g.Name, "err", err) } if err := q.MarkWantSearched(ctx, g.WantID); err != nil { im.logger.Warn("import: mark the want searched", "grab", g.Name, "err", err) } dest := filepath.Dir(placed[0]) if e != nil && e.Media() > 1 { dest = filepath.Dir(dest) } reason := fmt.Sprintf("Imported %s into %s. %s", tracks(len(placed)), dest, v.Reason) if det.Skipped > 0 { reason += fmt.Sprintf(" %s already in the library left as they were.", tracks(det.Skipped)) } if det.Untagged > 0 { reason += fmt.Sprintf(" %s copied with their own tags: ffmpeg could not rewrite them.", tracks(det.Untagged)) } return record(StateImported, reason, StateImported, det, len(placed), dest) } // rehome imports a download that turned out to be another wanted album, // named by its tags, for that album, when its own grab is not running. func (im *Importer) rehome(ctx context.Context, q *dbq.Queries, g dbq.Grab, d downloader.Driver, from Want, files []File, printed bool, record recordFunc) (string, bool) { group := majority(files, func(f File) string { return f.GroupMBID }) if group == "" { return "", false } others, err := q.OpenWantsForGroups(ctx, []string{group}) if err != nil || len(others) == 0 { return "", false } other := others[0] w, err := im.want(ctx, q, other.ID) if err != nil { return "", false } editions, err := im.editions(ctx, w.GroupMBID) if err != nil { return "", false } v := Verify(w, editions, files, printed, false) if v.Outcome != Accept { return "", false } moved, err := q.RehomeGrab(ctx, dbq.RehomeGrabParams{ID: g.ID, WantID: other.ID}) if err != nil { return "", false } if err := q.SearchWantAgain(ctx, g.WantID); err != nil { im.logger.Warn("import: search the first album again", "want", from.Title, "err", err) } det := Detail{Verdict: v, RehomedFrom: from.Title} for _, f := range files { det.Files = append(det.Files, f.Rel) } if v.Edition >= 0 { det.Edition, det.EditionMBID = describe(editions[v.Edition]), editions[v.Edition].MBID } v.Reason = fmt.Sprintf("Grabbed for %s, it is %s, which is wanted too. %s", from.Title, w.Title, v.Reason) det.Verdict = v return im.place(ctx, q, moved, d, w, editions, files, v, det, record), true } // refuse blocks the release, asks for the album to be searched again and // removes the download with its files: it is not the album. func (im *Importer) refuse(ctx context.Context, q *dbq.Queries, g dbq.Grab, d downloader.Driver, w Want, why string, by pgtype.UUID) { if err := acquire.NewMemory(im.pool).Block(ctx, dbq.BlockReleaseParams{ InfoHash: g.InfoHash, Name: g.Name, Reason: acquire.BlockImport, Detail: why, WantID: g.WantID, BlockedBy: by, }); err != nil { im.logger.Warn("import: block the release", "grab", g.Name, "err", err) } if err := q.SearchWantAgain(ctx, g.WantID); err != nil { im.logger.Warn("import: search again", "want", w.Title, "err", err) } im.removeFromClient(ctx, d, g) } func (im *Importer) removeFromClient(ctx context.Context, d downloader.Driver, g dbq.Grab) { if d == nil || g.InfoHash == "" { return } if err := d.Delete(ctx, true, g.InfoHash); err != nil { im.logger.Warn("import: remove the download", "grab", g.Name, "err", err) } } // Refuse is a person's "refuse" on a download the check held for them, or // one that failed: it is refused as the check would have. func (im *Importer) Refuse(ctx context.Context, grabID, by pgtype.UUID) (dbq.Grab, error) { q := dbq.New(im.pool) g, err := q.GetGrab(ctx, grabID) if err != nil { return dbq.Grab{}, err } if g.EndedAt.Valid || (g.ImportState != StateReview && g.ImportState != StateFailed) { return dbq.Grab{}, ErrNotDecidable } var w Want if g.WantID.Valid { w, _ = im.want(ctx, q, g.WantID) } d, err := im.ClientFor(ctx, g.Client) if err != nil { d = nil } why := "A person refused it: it is not the album." im.refuse(ctx, q, g, d, w, why, by) out, err := q.RecordImport(ctx, dbq.RecordImportParams{ ID: g.ID, ImportState: StateRefused, ImportReason: why, ImportDetail: g.ImportDetail, EndReason: StateRefused, }) im.Wake() return out, err } // ErrNotDecidable: the grab is not waiting on a person. var ErrNotDecidable = errors.New("importer: that download is not waiting on a decision") func (im *Importer) want(ctx context.Context, q *dbq.Queries, id pgtype.UUID) (Want, error) { row, err := q.GetWantForSearch(ctx, id) if err != nil { return Want{}, err } w := Want{ GroupMBID: row.ReleaseGroupMbid, ArtistMBID: row.ArtistMbid, Artist: row.ArtistName, Title: row.Title, Year: yearOf(row.FirstReleaseDate), PrimaryType: row.PrimaryType, SecondaryTypes: row.SecondaryTypes, } if row.FulfilledAt.Valid && row.Upgrade != "" && row.AlbumID.Valid { w.Upgrade, w.AlbumID = row.Upgrade, row.AlbumID } return w, nil } func (im *Importer) editions(ctx context.Context, group string) ([]Edition, error) { if im.Catalogue == nil { return nil, errors.New("no catalogue") } eds, err := im.Catalogue.Releases(ctx, group) if err != nil { return nil, err } out := make([]Edition, 0, len(eds.Editions)) for _, e := range eds.Editions { ed := Edition{MBID: e.Release.Mbid, Title: e.Release.Title, Status: e.Release.Status, Date: e.Release.ReleaseDate, Country: e.Release.Country} for _, t := range e.Tracks { ed.Tracks = append(ed.Tracks, Track{Medium: int(t.MediumPosition), Format: t.MediumFormat, Position: int(t.Position), Title: t.Title, RecordingMBID: t.RecordingMbid, LengthMs: int(t.LengthMs)}) } slices.SortStableFunc(ed.Tracks, func(a, b Track) int { if a.Medium != b.Medium { return a.Medium - b.Medium } return a.Position - b.Position }) out = append(out, ed) } return out, nil } // root is the library folder new music is filed into. func (im *Importer) root(ctx context.Context, q *dbq.Queries) (string, error) { roots := im.Roots() if len(roots) == 0 { return "", errors.New("there is no library folder to file new music into; add one in Settings → Library") } s, err := q.GetAcquisitionSettings(ctx) if err == nil && s.ImportRoot != "" && slices.Contains(roots, s.ImportRoot) { return s.ImportRoot, nil } return roots[0], nil } // held reads which of the planned tracks the library already has: by // recording, and by disc and track of an album of the same group. func (im *Importer) held(ctx context.Context, q *dbq.Queries, w Want, plan []Placement) (map[string]bool, map[[2]int]bool, error) { var recs []string for _, pl := range plan { if pl.Recording != "" { recs = append(recs, pl.Recording) } } rec := map[string]bool{} if len(recs) > 0 { got, err := q.HeldRecordings(ctx, recs) if err != nil { return nil, nil, err } for _, r := range got { rec[r] = true } } pos := map[[2]int]bool{} got, err := q.HeldPositions(ctx, &w.GroupMBID) if err != nil { return nil, nil, err } for _, p := range got { pos[[2]int{int(p.Disc), int(p.Track)}] = true } return rec, pos, nil } func yearOf(date string) int { if len(date) < 4 { return 0 } y, err := strconv.Atoi(date[:4]) if err != nil { return 0 } return y } func tracks(n int) string { if n == 1 { return "1 track" } return strconv.Itoa(n) + " tracks" }