package uploads import ( "bytes" "context" "encoding/hex" "errors" "fmt" "io/fs" "os" "os/exec" "path" "path/filepath" "strconv" "strings" "time" "github.com/google/uuid" "pipu/api/internal/library" "pipu/api/internal/library/meta" "pipu/api/internal/storageagent" ) // Фоновые задачи очереди: обработка принятых файлов, импорт одобренных, истечение недокачанных. // По одному файлу за раз — диск и NFS не перегружаются, а порядок понятен админу. const ( pollEvery = 30 * time.Second cleanupEvery = 10 * time.Minute // FingerprintTolerance — разница длительностей, при которой отпечатки ещё сравниваются. FingerprintTolerance = 5000 // MetadataTolerance — «та же длительность» для совпадения метаданных (docs/uploads.md#4). MetadataTolerance = 2000 ) // Run запускает фоновые задачи до отмены ctx. func (s *Service) Run(ctx context.Context) { s.reconcile(ctx) go s.loop(ctx, s.wakeProcess, pollEvery, func(ctx context.Context) bool { return s.processNext(ctx) }) go s.loop(ctx, s.wakeImport, pollEvery, func(ctx context.Context) bool { return s.importNext(ctx) }) go s.loop(ctx, nil, cleanupEvery, func(ctx context.Context) bool { s.expire(ctx) if err := s.store.ExpireMetadataCache(ctx); err != nil { s.log.Warn("uploads: expire metadata cache", "err", err) } if err := s.store.ReconcileQuota(ctx); err != nil { s.log.Error("uploads: reconcile quota", "err", err) } return false }) if err := s.store.ReconcileQuota(ctx); err != nil { s.log.Error("uploads: reconcile quota", "err", err) } wake(s.wakeProcess) wake(s.wakeImport) if s.catalogue != nil { go s.loop(ctx, nil, pollEvery, func(ctx context.Context) bool { return s.lookupNextRelease(ctx) }) } } // loop вызывает step, пока тот находит работу, затем ждёт сигнала или таймера. func (s *Service) loop(ctx context.Context, signal chan struct{}, every time.Duration, step func(ctx context.Context) bool) { t := time.NewTicker(every) defer t.Stop() for { for ctx.Err() == nil && step(ctx) { } select { case <-ctx.Done(): return case <-t.C: case <-signal: } } } func (s *Service) stagingAbs(rel string) string { return filepath.Join(s.cfg.StagingDir, filepath.FromSlash(rel)) } // --- Обработка --- // processNext обрабатывает один файл в статусе processing; false — работы нет. func (s *Service) processNext(ctx context.Context) bool { f, err := s.store.NextWithStatus(ctx, StatusProcessing) if errors.Is(err, ErrNotFound) { return false } if err != nil { s.log.Error("uploads: next to process", "err", err) return false } if err := s.process(ctx, f); err != nil { s.log.Error("uploads: process", "upload_id", f.ID, "err", err) // Сбой БД / диска: файл остаётся в processing, повтор — со следующим тиком. return false } return true } // outcome — итог обработки для записи в БД. type outcome struct { status Status reason string probe library.Probe sha []byte print *library.Fingerprint tags map[string]string meta *Meta dupTrack *uuid.UUID dupLevel string possible bool quality bool place placement } func (s *Service) process(ctx context.Context, f File) error { abs := s.stagingAbs(f.StagingPath) o := s.analyze(ctx, f, abs) if err := s.store.SaveOutcome(ctx, f, o); err != nil { return err } if o.status != StatusPendingReview { s.removeStaging(ctx, f) } s.log.Info("upload processed", "upload_id", f.ID, "status", o.status, "reason", o.reason) return nil } // analyze — шаги process_upload_file (docs/uploads.md#6): ffprobe, SHA-256, Chromaprint → дубли; метаданные; путь. func (s *Service) analyze(ctx context.Context, f File, abs string) outcome { info, err := os.Stat(abs) if err != nil || info.Size() != f.Size { return outcome{status: StatusInvalid, reason: CodeSizeMismatch} } probe, err := s.prober.Probe(ctx, abs) if err != nil || probe.DurationMs < MinDurationMs || probe.DurationMs > MaxDurationMs { return outcome{status: StatusInvalid, reason: CodeInvalidAudio} } o := outcome{probe: probe, tags: tagsMap(probe.Tags)} for key, value := range probe.RawTags { if strings.HasPrefix(key, "musicbrainz") { o.tags[key] = value } } if o.sha, err = library.FileSHA256(abs); err != nil { return outcome{status: StatusInvalid, reason: CodeInvalidAudio, probe: probe} } if len(f.SHA256) > 0 && !bytes.Equal(f.SHA256, o.sha) { o.status, o.reason = StatusInvalid, CodeHashMismatch return o } // 1. Точная копия. if track, dup, err := s.store.DuplicateBySHA(ctx, o.sha, f.ID); err == nil && dup { o.status, o.reason, o.dupTrack, o.dupLevel = StatusDuplicate, CodeDuplicate, track, DupSHA256 return o } // 3. Акустический отпечаток (fpcalc может отсутствовать — тогда уровень пропускается). if fp, err := s.fpcalc.Fingerprint(ctx, abs); err == nil { o.print = &fp cands, err := s.store.FingerprintCandidates(ctx, probe.DurationMs, FingerprintTolerance, f.ID) if err == nil { if c, ok := bestMatch(fp.Points, cands); ok { if s.keepAsUpgrade(ctx, &o, c.TrackID, probe) { // лучшее качество — дальше как обычный файл на рассмотрение } else { o.status, o.reason, o.dupTrack, o.dupLevel = StatusDuplicate, CodeDuplicate, c.TrackID, DupFingerprint return o } } } } else if !errors.Is(err, library.ErrNoFPCalc) { s.log.Warn("uploads: fpcalc", "upload_id", f.ID, "err", err) } // Метаданные: теги → путь из перетащенной папки по шаблонам целевого хранилища → имя файла. src, _ := s.pickSource(ctx, f.TargetSourceID) rel := f.OriginalName if f.ClientRelPath != nil { rel = *f.ClientRelPath } var fromPath meta.Fields if src.ID != uuid.Nil { if patterns := s.sourcePatterns(ctx, src.ID); len(patterns) > 0 { fromPath = meta.FromPath(meta.MatchPath(patterns, rel).Fields) } } m := normalizeMeta(FromMerged(meta.Merge(probe.Tags, fromPath, true, rel))) if m.Album == "" { m.Album = library.UnknownAlbum } o.meta = &m // 2. Совпадение метаданных — не отказ, а пометка для админа. if track, err := s.store.MetadataMatch(ctx, meta.Key(m.Artist), meta.Key(m.Title), probe.DurationMs, MetadataTolerance); err == nil && track != nil { o.possible, o.dupTrack, o.dupLevel = true, track, DupMetadata } codec := probe.Codec ff := f ff.Codec = &codec o.place = s.plan(ctx, ff, m, f.TargetSourceID) o.status, o.reason = StatusPendingReview, o.place.Reason return o } // keepAsUpgrade — совпадение по отпечатку, но файл лучше библиотечного и режим замены включён. // true — не считать дублем, оставить на рассмотрение. func (s *Service) keepAsUpgrade(ctx context.Context, o *outcome, track *uuid.UUID, probe library.Probe) bool { if track == nil { return false } st, err := s.store.Settings(ctx) if err != nil || !st.AllowQualityUpgrade { return false } codec, format, bitrate, err := s.store.BestAudio(ctx, *track) if err != nil || !BetterQuality(probe.Codec, probe.Format, probe.Bitrate, codec, format, bitrate) { return false } o.quality = true o.dupTrack = track o.dupLevel = DupFingerprint return true } // bestMatch — отпечаток, похожий не меньше порога library.SameRecording. func bestMatch(points []uint32, cands []PrintCandidate) (PrintCandidate, bool) { best, score := PrintCandidate{}, 0.0 for _, c := range cands { if sim := library.Similarity(points, c.Points); sim > score { best, score = c, sim } } return best, score >= library.SameRecording } func tagsMap(t meta.Fields) map[string]string { out := map[string]string{} for k, v := range map[string]string{ "artist": t.Artist, "albumartist": t.AlbumArtist, "album": t.Album, "title": t.Title, "genre": t.Genre, "year": t.Year, "track": t.Track, "disc": t.Disc, } { if v != "" { out[k] = v } } return out } func (s *Service) sourcePatterns(ctx context.Context, id uuid.UUID) []meta.Pattern { rows, err := s.lib.Patterns(ctx, id) if err != nil { return nil } var exprs []string for _, r := range rows { if r.IsEnabled { exprs = append(exprs, r.Regex) } } patterns, _ := meta.CompilePatterns(exprs) return patterns } // --- Импорт --- // importNext импортирует один одобренный файл; false — работы нет. func (s *Service) importNext(ctx context.Context) bool { f, err := s.store.NextWithStatus(ctx, StatusApproved) if errors.Is(err, ErrNotFound) { return false } if err != nil { s.log.Error("uploads: next to import", "err", err) return false } if err := s.store.ChangeStatus(ctx, f.ID, []Status{StatusApproved}, StatusImporting); err != nil { // Файл успели отклонить — берём следующий; сбой БД — ждём тика. return errors.Is(err, ErrWrongStatus) } if err := s.importFile(ctx, f); err != nil { s.log.Warn("upload import failed", "upload_id", f.ID, "err", err) msg := err.Error() if ferr := s.store.FailImport(ctx, f.ID, CodeImportFailed+": "+msg); ferr != nil { s.log.Error("uploads: mark import failed", "upload_id", f.ID, "err", ferr) } } return true } // importFile — import_upload_file (docs/uploads.md#9): путь по финальным метаданным → (теги) → запись агентом → // точечный скан → imported. Повторяемо: если файл с тем же хэшем уже лежит по пути — запись пропускается. func (s *Service) importFile(ctx context.Context, f File) error { if f.Meta == nil { return errors.New("no metadata") } pl := s.plan(ctx, f, *f.Meta, f.TargetSourceID) if pl.Reason != "" { return errors.New(pl.Reason) } rel := *pl.RelPath stagingRel, sum := f.StagingPath, f.SHA256 if f.MetadataEdited || (f.CoverPath != nil && *f.CoverPath != "") { tagged, tsum, err := s.writeTags(ctx, f, *f.Meta, path.Ext(rel)) if err != nil { return fmt.Errorf("write tags: %w", err) } if tagged != "" { stagingRel, sum = tagged, tsum } } target := filepath.Join(pl.root, filepath.FromSlash(rel)) if have, err := library.FileSHA256(target); err != nil || !bytes.Equal(have, sum) { req := storageagent.WriteRequest{SourceID: pl.src.ID.String(), RelPath: rel, StagingPath: stagingRel, SHA256: hex.EncodeToString(sum)} if pl.src.Type == library.SourceLocal { lp := "" if pl.src.LocalPath != nil { lp = *pl.src.LocalPath } req.LocalPath = &lp } res, err := s.agent.WriteFile(ctx, req) if err != nil { return fmt.Errorf("storage-agent: %w", err) } rel = res.RelPath } fileID, err := s.scanner.ScanFile(ctx, pl.src, pl.root, rel) if err != nil { return fmt.Errorf("scan: %w", err) } if f.ArtistImagePath != nil && s.artistImages != nil { image, err := os.ReadFile(s.stagingAbs(*f.ArtistImagePath)) if err != nil { return fmt.Errorf("read artist image: %w", err) } if err := s.artistImages.ApplyArtistImage(ctx, fileID, image); err != nil { return fmt.Errorf("artist image: %w", err) } } if err := s.store.CompleteImport(ctx, f.ID, pl.src.ID, rel, fileID); err != nil { return err } s.removeStaging(ctx, f) s.store.Audit(ctx, f.ReviewedBy, "upload.import", f.ID, map[string]any{"source_id": pl.src.ID, "rel_path": rel}) s.log.Info("upload imported", "upload_id", f.ID, "source_id", pl.src.ID, "rel_path", rel) return nil } // tagFormats — контейнеры, в которые ffmpeg пишет теги без перекодирования. var tagFormats = map[string]bool{".flac": true, ".mp3": true, ".m4a": true, ".ogg": true, ".opus": true} // writeTags записывает правленые метаданные в копию файла рядом в staging (`ffmpeg -map 0 -c copy -metadata …`). // "" — формат без поддержки тегов, импортируется исходный файл. func (s *Service) writeTags(ctx context.Context, f File, m Meta, ext string) (string, []byte, error) { if !tagFormats[strings.ToLower(ext)] { return "", nil, nil } rel := f.StagingPath + ".tagged" + strings.ToLower(ext) out := s.stagingAbs(rel) args := []string{"-v", "error", "-y", "-i", s.stagingAbs(f.StagingPath)} if f.CoverPath != nil && *f.CoverPath != "" { args = append(args, "-i", s.stagingAbs(*f.CoverPath), "-map", "0:a", "-map", "1", "-c", "copy", "-disposition:v:0", "attached_pic", "-map_metadata", "0") } else { args = append(args, "-map", "0", "-c", "copy", "-map_metadata", "0") } set := func(k, v string) { args = append(args, "-metadata", k+"="+v) } set("title", m.Title) set("artist", m.Artist) set("album_artist", m.AlbumArtist) set("album", m.Album) set("genre", m.Genre) for _, key := range []string{"musicbrainz_albumid", "musicbrainz_releasegroupid", "musicbrainz_trackid"} { if value := f.Tags[key]; value != "" { set(key, value) } } num := func(p *int) string { if p == nil { return "" } return strconv.Itoa(*p) } set("date", num(m.Year)) set("track", num(m.Track)) set("disc", num(m.Disc)) args = append(args, "--", out) bin := s.cfg.FFmpeg if bin == "" { bin = "ffmpeg" } ctx, cancel := context.WithTimeout(ctx, 5*time.Minute) defer cancel() if msg, err := exec.CommandContext(ctx, bin, args...).CombinedOutput(); err != nil { _ = os.Remove(out) return "", nil, fmt.Errorf("ffmpeg: %s", strings.TrimSpace(string(msg))) } sum, err := library.FileSHA256(out) if err != nil { return "", nil, err } return rel, sum, nil } // --- Очистка --- // expire переводит недокачанные за 24 ч загрузки в expired и удаляет их файлы. func (s *Service) expire(ctx context.Context) { ids, err := s.store.Stale(ctx, s.now().Add(-ExpireAfter)) if err != nil { s.log.Error("uploads: stale", "err", err) return } for _, id := range ids { f, err := s.store.File(ctx, id) if err != nil { continue } if err := s.store.ChangeStatus(ctx, id, []Status{StatusUploading}, StatusExpired); err == nil { s.removeStaging(ctx, f) } } } // reconcile при старте: прерванный импорт — снова в approved; осиротевшие файлы staging удаляются. func (s *Service) reconcile(ctx context.Context) { if err := s.store.ResetImporting(ctx); err != nil { s.log.Error("uploads: reset importing", "err", err) } inUse, err := s.store.StagingInUse(ctx) if err != nil { s.log.Error("uploads: staging in use", "err", err) return } removed := 0 _ = filepath.WalkDir(s.cfg.StagingDir, func(p string, d fs.DirEntry, err error) error { if err != nil || d.IsDir() { return nil } rel, err := filepath.Rel(s.cfg.StagingDir, p) if err != nil { return nil } rel = filepath.ToSlash(rel) if base, _, ok := strings.Cut(rel, ".tagged"); ok { rel = base } if !inUse[rel] { if os.Remove(p) == nil { removed++ } } return nil }) if entries, err := os.ReadDir(s.cfg.StagingDir); err == nil { for _, e := range entries { if e.IsDir() { _ = os.Remove(filepath.Join(s.cfg.StagingDir, e.Name())) // только пустые } } } if removed > 0 { s.log.Info("uploads: removed orphan staging files", "count", removed) } }