pipu-music/services/api/internal/uploads/worker.go

package uploads

import (
	"bytes"
	"context"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"

	"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); return false })
	wake(s.wakeProcess)
	wake(s.wakeImport)
}

// 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
	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.saveOutcome(ctx, f, o); err != nil {
		return err
	}
	if o.status != StatusPendingReview {
		s.removeStaging(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)}
	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 {
				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
}

// 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
}

func (s *Service) saveOutcome(ctx context.Context, f File, o outcome) error {
	return s.store.Transition(ctx, f.ID, []Status{StatusProcessing}, o.status, func(tx pgx.Tx) error {
		var (
			tags, metaJSON []byte
			points         []int32
		)
		if o.tags != nil {
			tags, _ = json.Marshal(o.tags)
		}
		if o.meta != nil {
			metaJSON, _ = json.Marshal(o.meta)
		}
		if o.print != nil {
			points = make([]int32, len(o.print.Points))
			for i, p := range o.print.Points {
				points[i] = int32(p)
			}
		}
		p := o.probe
		_, err := tx.Exec(ctx, `
			UPDATE upload_files SET sha256 = coalesce($2, sha256), format = NULLIF($3, ''), codec = NULLIF($4, ''),
				bitrate = NULLIF($5, 0), sample_rate = NULLIF($6, 0), bit_depth = NULLIF($7, 0), duration_ms = NULLIF($8, 0),
				tags = $9, metadata = $10, fingerprint = $11, duplicate_of_track_id = $12, duplicate_level = NULLIF($13, ''),
				possible_duplicate = $14, target_source_id = $15, target_rel_path = $16, status_reason = NULLIF($17, ''),
				matched_artist_id = $18, matched_album_id = $19
			WHERE id = $1`,
			f.ID, o.sha, p.Format, p.Codec, p.Bitrate, p.SampleRate, p.BitDepth, p.DurationMs, tags, metaJSON, points,
			o.dupTrack, o.dupLevel, o.possible, o.place.SourceID, o.place.RelPath, o.reason, o.place.ArtistID, o.place.AlbumID)
		return err
	})
}

// --- Импорт ---

// 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.Transition(ctx, f.ID, []Status{StatusApproved}, StatusImporting, nil); 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.Transition(ctx, f.ID, []Status{StatusImporting}, StatusImportFailed, func(tx pgx.Tx) error {
			_, err := tx.Exec(ctx, `UPDATE upload_files SET status_reason = $2 WHERE id = $1`, f.ID, CodeImportFailed+": "+msg)
			return err
		}); 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 {
		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 err := s.store.Transition(ctx, f.ID, []Status{StatusImporting}, StatusImported, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, `UPDATE upload_files SET target_source_id = $2, target_rel_path = $3 WHERE id = $1`,
			f.ID, pl.src.ID, rel); err != nil {
			return err
		}
		return MarkImported(ctx, tx, f.ID, fileID)
	}); err != nil {
		return err
	}
	s.removeStaging(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), "-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)
	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.Transition(ctx, id, []Status{StatusUploading}, StatusExpired, nil); err == nil {
			s.removeStaging(f)
		}
	}
}

// reconcile при старте: прерванный импорт — снова в approved; осиротевшие файлы staging удаляются.
func (s *Service) reconcile(ctx context.Context) {
	if _, err := s.store.pool.Exec(ctx, `UPDATE upload_files SET status = 'approved', updated_at = now() WHERE status = 'importing'`); 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)
	}
}