MusicServer/internal/backend/sync.go

package backend

import (
	"context"
	"crypto/md5"
	"crypto/sha1"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"music-server/internal/db/repository"
	"music-server/internal/logging"

	"github.com/MShekow/directory-checksum/directory_checksum"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/spf13/afero"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

const (
	soundtrackWorkers = 10
	songWorkers       = 4
	brokenWorkers     = 50
)

// foldersToSkip are directories in the music path that are never synced.
var foldersToSkip = map[string]struct{}{".sync": {}, "characters": {}, "dist": {}, "old": {}}

type SyncResponse struct {
	SoundtracksAdded          []string          `json:"soundtracks_added"`
	SoundtracksReAdded        []string          `json:"soundtracks_re_added"`
	SoundtracksChangedTitle   map[string]string `json:"soundtracks_changed_title"`
	SoundtracksChangedContent []string          `json:"soundtracks_changed_content"`
	SoundtracksRemoved        []string          `json:"soundtracks_removed"`
	CatchedErrors             []string          `json:"catched_errors"`
	TotalTime                 string            `json:"total_time"`
}

type ProgressResponse struct {
	Progress  string `json:"progress"`
	TimeSpent string `json:"time_spent"`
}

type SoundtrackStatus int

const (
	NotChanged SoundtrackStatus = iota
	TitleChanged
	SoundtrackChanged
	NewSoundtrack
)

var statusName = map[SoundtrackStatus]string{
	NotChanged:        "Not changed",
	TitleChanged:      "Title changed",
	SoundtrackChanged: "Soundtrack changed",
	NewSoundtrack:     "New soundtrack",
}

func (ss SoundtrackStatus) String() string {
	return statusName[ss]
}

// SyncService synchronises the music folder with the database.
type SyncService struct {
	q          repository.Querier
	musicPath  string
	onComplete func(context.Context)

	running atomic.Bool
	wg      sync.WaitGroup

	foldersSynced atomic.Int64
	foldersTotal  atomic.Int64

	mu      sync.Mutex
	started time.Time
	result  SyncResponse
}

// NewSyncService creates a SyncService. onComplete is called after every sync, before it is reported as finished.
func NewSyncService(q repository.Querier, musicPath string, onComplete func(context.Context)) *SyncService {
	if !strings.HasSuffix(musicPath, "/") {
		musicPath += "/"
	}
	return &SyncService{q: q, musicPath: musicPath, onComplete: onComplete}
}

// IsSyncing reports whether a sync is running.
func (s *SyncService) IsSyncing() bool {
	return s.running.Load()
}

// Start starts a sync in the background. It returns false if a sync is already running.
func (s *SyncService) Start(full bool) bool {
	if !s.running.CompareAndSwap(false, true) {
		return false
	}
	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		defer s.running.Store(false)
		s.run(context.Background(), full)
	}()
	return true
}

// Wait blocks until the running sync, if any, has finished.
func (s *SyncService) Wait() {
	s.wg.Wait()
}

// Progress returns how far the running sync has come.
func (s *SyncService) Progress() ProgressResponse {
	progress := 0
	if total := s.foldersTotal.Load(); total > 0 {
		progress = int(s.foldersSynced.Load() * 100 / total)
	}
	s.mu.Lock()
	timeSpent := time.Since(s.started)
	s.mu.Unlock()
	return ProgressResponse{
		Progress:  fmt.Sprintf("%d", progress),
		TimeSpent: formatDuration(timeSpent),
	}
}

// Result returns the result of the last finished sync. Lists are empty, not nil, before the first sync.
func (s *SyncService) Result() SyncResponse {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.result.withEmptyLists()
}

func (r SyncResponse) withEmptyLists() SyncResponse {
	for _, list := range []*[]string{&r.SoundtracksAdded, &r.SoundtracksReAdded, &r.SoundtracksChangedContent, &r.SoundtracksRemoved, &r.CatchedErrors} {
		if *list == nil {
			*list = []string{}
		}
	}
	if r.SoundtracksChangedTitle == nil {
		r.SoundtracksChangedTitle = map[string]string{}
	}
	return r
}

// ResetDB deletes all songs and soundtracks.
func (s *SyncService) ResetDB(ctx context.Context) error {
	if err := s.q.ClearSongs(ctx); err != nil {
		return err
	}
	return s.q.ClearSoundtracks(ctx)
}

func formatDuration(d time.Duration) string {
	return time.Time{}.Add(d).Format("15:04:05")
}

// syncRun holds the state of a single sync.
type syncRun struct {
	q         repository.Querier
	musicPath string
	full      bool
	progress  func()

	existing []repository.Soundtrack

	mu             sync.Mutex
	foundIDs       []int32
	added          []string
	reAdded        []string
	changedTitle   map[string]string
	changedContent []string
	errs           []string
}

func (s *SyncService) run(ctx context.Context, full bool) {
	start := time.Now()
	s.mu.Lock()
	s.started = start
	s.mu.Unlock()
	s.foldersSynced.Store(0)
	s.foldersTotal.Store(0)

	logger := logging.GetLogger()
	logger.Info("Starting sync", zap.String("music_path", s.musicPath), zap.Bool("full", full))

	run := &syncRun{q: s.q, musicPath: s.musicPath, full: full, progress: func() { s.foldersSynced.Add(1) }}
	result := run.execute(ctx, func(total int) { s.foldersTotal.Store(int64(total)) })
	result.TotalTime = formatDuration(time.Since(start))

	if s.onComplete != nil {
		s.onComplete(ctx)
	}

	s.mu.Lock()
	s.result = result
	s.mu.Unlock()

	logger.Info("Sync completed",
		zap.Duration("total_time", time.Since(start)),
		zap.Int("added", len(result.SoundtracksAdded)),
		zap.Int("re_added", len(result.SoundtracksReAdded)),
		zap.Int("changed_title", len(result.SoundtracksChangedTitle)),
		zap.Int("changed_content", len(result.SoundtracksChangedContent)),
		zap.Int("removed", len(result.SoundtracksRemoved)),
		zap.Int("errors", len(result.CatchedErrors)))
}

func (r *syncRun) execute(ctx context.Context, setTotal func(int)) SyncResponse {
	var err error
	r.existing, err = r.q.GetAllSoundtracksIncludingDeleted(ctx)
	if err != nil {
		r.addError("GetAllSoundtracksIncludingDeleted", err, "")
		return r.response(nil)
	}

	directories, err := os.ReadDir(r.musicPath)
	if err != nil {
		r.addError("ReadDir", err, r.musicPath)
		return r.response(nil)
	}
	setTotal(len(directories))

	g := new(errgroup.Group)
	g.SetLimit(soundtrackWorkers)
	for _, dir := range directories {
		g.Go(func() error {
			defer r.progress()
			if _, skip := foldersToSkip[dir.Name()]; dir.IsDir() && !skip {
				r.syncSoundtrack(ctx, dir.Name())
			}
			return nil
		})
	}
	_ = g.Wait()

	if err := r.q.MarkSoundtracksDeletedExcept(ctx, r.foundIDs); err != nil {
		r.addError("MarkSoundtracksDeletedExcept", err, "")
	}
	if r.full {
		r.removeBrokenSongs(ctx)
	}
	return r.response(removedSoundtracks(r.existing, r.foundIDs))
}

func (r *syncRun) response(removed []string) SyncResponse {
	r.mu.Lock()
	defer r.mu.Unlock()
	return SyncResponse{
		SoundtracksAdded:          sorted(r.added),
		SoundtracksReAdded:        sorted(r.reAdded),
		SoundtracksChangedTitle:   r.changedTitle,
		SoundtracksChangedContent: sorted(r.changedContent),
		SoundtracksRemoved:        removed,
		CatchedErrors:             r.errs,
	}
}

func sorted(s []string) []string {
	slices.Sort(s)
	return s
}

// removedSoundtracks returns the names of soundtracks that were active before the sync but not found by it.
func removedSoundtracks(existing []repository.Soundtrack, foundIDs []int32) []string {
	var removed []string
	for _, s := range existing {
		if s.Deleted == nil && !slices.Contains(foundIDs, s.ID) {
			removed = append(removed, s.SoundtrackName)
		}
	}
	return removed
}

func (r *syncRun) addError(op string, err error, detail string) {
	logging.GetLogger().Error("Sync error", zap.String("operation", op), zap.String("detail", detail), zap.Error(err))
	msg := fmt.Sprintf("Func: %s\nError message: %s", op, err)
	if detail != "" {
		msg += "\nDebug message: " + detail
	}
	r.mu.Lock()
	r.errs = append(r.errs, msg)
	r.mu.Unlock()
}

func (r *syncRun) syncSoundtrack(ctx context.Context, name string) {
	logger := logging.GetLogger()
	dir := r.musicPath + name + "/"

	fp, err := fingerprint(dir)
	if err != nil {
		r.addError("fingerprint", err, dir)
		return
	}

	byName := findByName(r.existing, name)
	if !r.full && byName != nil && byName.Path == dir && byName.Fingerprint != "" && byName.Fingerprint == fp {
		r.found(byName.ID)
		if byName.Deleted != nil {
			r.reAddSoundtrack(ctx, *byName, name)
			r.syncSongs(ctx, *byName, dir)
		}
		logger.Debug("Soundtrack not changed", zap.String("soundtrack", name))
		return
	}

	hash, err := dirHash(dir)
	if err != nil {
		r.addError("dirHash", err, dir)
		return
	}
	status, match := classify(r.existing, name, hash)
	// A soundtrack is also treated as changed when the music path has moved.
	changed := status != NotChanged || r.full || match.Path != dir

	if changed {
		if removeIDFiles(dir) {
			if fp, err = fingerprint(dir); err != nil {
				r.addError("fingerprint", err, dir)
				return
			}
			if hash, err = dirHash(dir); err != nil {
				r.addError("dirHash", err, dir)
				return
			}
		}
	}

	logger.Debug("Syncing soundtrack", zap.String("soundtrack", name), zap.String("status", status.String()))

	var soundtrack repository.Soundtrack
	switch status {
	case NewSoundtrack:
		soundtrack, err = r.insertSoundtrack(ctx, name, dir, hash)
		if err != nil {
			r.addError("InsertSoundtrack", err, dir)
			return
		}
		r.mu.Lock()
		r.added = append(r.added, name)
		r.mu.Unlock()
	case SoundtrackChanged:
		soundtrack = *match
		r.mu.Lock()
		r.changedContent = append(r.changedContent, name)
		r.mu.Unlock()
	case TitleChanged:
		soundtrack = *match
		if err := r.q.UpdateSoundtrackName(ctx, repository.UpdateSoundtrackNameParams{Name: name, Path: dir, ID: soundtrack.ID}); err != nil {
			r.addError("UpdateSoundtrackName", err, dir)
		}
		r.mu.Lock()
		if r.changedTitle == nil {
			r.changedTitle = make(map[string]string)
		}
		r.changedTitle[soundtrack.SoundtrackName] = name
		r.mu.Unlock()
	case NotChanged:
		soundtrack = *match
	}
	r.found(soundtrack.ID)

	if status != NewSoundtrack && status != TitleChanged && soundtrack.Path != dir {
		if err := r.q.UpdateSoundtrackName(ctx, repository.UpdateSoundtrackNameParams{Name: name, Path: dir, ID: soundtrack.ID}); err != nil {
			r.addError("UpdateSoundtrackName", err, dir)
		}
	}
	if status != NewSoundtrack && soundtrack.Hash != hash {
		if err := r.q.UpdateSoundtrackHash(ctx, repository.UpdateSoundtrackHashParams{Hash: hash, ID: soundtrack.ID}); err != nil {
			r.addError("UpdateSoundtrackHash", err, dir)
		}
	}

	if soundtrack.Deleted != nil {
		r.reAddSoundtrack(ctx, soundtrack, name)
	}
	if changed || soundtrack.Deleted != nil {
		r.syncSongs(ctx, soundtrack, dir)
	}

	if fp != soundtrack.Fingerprint {
		if err := r.q.SetSoundtrackFingerprint(ctx, repository.SetSoundtrackFingerprintParams{Fingerprint: fp, ID: soundtrack.ID}); err != nil {
			r.addError("SetSoundtrackFingerprint", err, dir)
		}
	}
}

func (r *syncRun) found(id int32) {
	r.mu.Lock()
	r.foundIDs = append(r.foundIDs, id)
	r.mu.Unlock()
}

// reAddSoundtrack brings back a soundtrack that was deleted in an earlier sync.
func (r *syncRun) reAddSoundtrack(ctx context.Context, soundtrack repository.Soundtrack, name string) {
	if err := r.q.RemoveSoundtrackDeletionDate(ctx, soundtrack.ID); err != nil {
		r.addError("RemoveSoundtrackDeletionDate", err, name)
	}
	r.mu.Lock()
	r.reAdded = append(r.reAdded, name)
	r.mu.Unlock()
}

func (r *syncRun) insertSoundtrack(ctx context.Context, name, dir, hash string) (repository.Soundtrack, error) {
	params := repository.InsertSoundtrackParams{SoundtrackName: name, Path: dir, Hash: hash, Uuid: uuid.New()}
	row, err := r.q.InsertSoundtrack(ctx, params)
	if isUniqueViolation(err) {
		logging.GetLogger().Warn("ID collision detected, resetting soundtrack sequence", zap.String("soundtrack", name))
		if _, err := r.q.ResetSoundtrackIdSeq(ctx); err != nil {
			return repository.Soundtrack{}, err
		}
		params.Uuid = uuid.New()
		row, err = r.q.InsertSoundtrack(ctx, params)
	}
	if err != nil {
		return repository.Soundtrack{}, err
	}
	return repository.Soundtrack{ID: row.ID, Uuid: row.Uuid, SoundtrackName: name, Path: dir, Hash: hash}, nil
}

func isUniqueViolation(err error) bool {
	var pgErr *pgconn.PgError
	return errors.As(err, &pgErr) && pgErr.Code == "23505"
}

// syncSongs makes the songs in the database match the song files in the soundtrack folder.
func (r *syncRun) syncSongs(ctx context.Context, soundtrack repository.Soundtrack, dir string) {
	existing, err := r.q.FindSongsFromSoundtrack(ctx, soundtrack.ID)
	if err != nil {
		r.addError("FindSongsFromSoundtrack", err, dir)
		return
	}
	files, err := r.songFiles(dir)
	if err != nil {
		r.addError("songFiles", err, dir)
		return
	}

	ops := reconcileSongs(existing, files)
	// Delete first so paths of removed songs are free for renamed ones.
	if len(ops.delete) > 0 {
		if err := r.q.RemoveBrokenSongs(ctx, ops.delete); err != nil {
			r.addError("RemoveBrokenSongs", err, dir)
		}
	}
	for _, u := range ops.update {
		err := r.q.UpdateSongByUUID(ctx, repository.UpdateSongByUUIDParams{
			SongName: u.file.name,
			FileName: &u.file.fileName,
			Path:     u.file.path,
			Hash:     u.file.hash,
			Uuid:     u.uuid,
		})
		if err != nil {
			r.addError("UpdateSongByUUID", err, u.file.path)
		}
	}
	for _, song := range ops.insert {
		err := r.q.AddSong(ctx, repository.AddSongParams{
			SoundtrackID:   soundtrack.ID,
			SongName:       song.name,
			Path:           song.path,
			FileName:       &song.fileName,
			Hash:           song.hash,
			Uuid:           uuid.New(),
			SoundtrackUuid: soundtrack.Uuid,
		})
		if err != nil {
			r.addError("AddSong", err, song.path)
		}
	}
}

// songFile is a song found on disk.
type songFile struct {
	name     string
	fileName string
	path     string
	hash     string
}

func (r *syncRun) songFiles(dir string) ([]songFile, error) {
	entries, err := os.ReadDir(dir)
	if err != nil {
		return nil, err
	}

	var files []songFile
	for _, entry := range entries {
		if isSong(entry) {
			files = append(files, songFile{name: songName(entry.Name()), fileName: entry.Name(), path: dir + entry.Name()})
		}
	}

	g := new(errgroup.Group)
	g.SetLimit(songWorkers)
	for i := range files {
		g.Go(func() error {
			hash, err := fileHash(files[i].path)
			if err != nil {
				return err
			}
			files[i].hash = hash
			return nil
		})
	}
	return files, g.Wait()
}

type songUpdate struct {
	uuid uuid.UUID
	file songFile
}

type songOps struct {
	insert []songFile
	update []songUpdate
	delete []uuid.UUID
}

// reconcileSongs works out which database changes make existing match the files on disk.
// Songs are matched by path first, then by hash so renamed files keep their play counts.
func reconcileSongs(existing []repository.Song, files []songFile) songOps {
	var ops songOps
	used := make(map[uuid.UUID]bool)
	byPath := make(map[string]repository.Song)
	for _, song := range existing {
		byPath[song.Path] = song
	}

	var unmatched []songFile
	for _, file := range files {
		song, ok := byPath[file.path]
		if !ok {
			unmatched = append(unmatched, file)
			continue
		}
		used[song.Uuid] = true
		if song.Hash != file.hash || song.SongName != file.name || song.FileName == nil || *song.FileName != file.fileName {
			ops.update = append(ops.update, songUpdate{uuid: song.Uuid, file: file})
		}
	}

	for _, file := range unmatched {
		i := slices.IndexFunc(existing, func(song repository.Song) bool {
			return !used[song.Uuid] && song.Hash != "" && song.Hash == file.hash
		})
		if i < 0 {
			ops.insert = append(ops.insert, file)
			continue
		}
		used[existing[i].Uuid] = true
		ops.update = append(ops.update, songUpdate{uuid: existing[i].Uuid, file: file})
	}

	for _, song := range existing {
		if !used[song.Uuid] {
			ops.delete = append(ops.delete, song.Uuid)
		}
	}
	return ops
}

// classify decides how a soundtrack folder relates to the soundtracks already in the database.
func classify(existing []repository.Soundtrack, name, hash string) (SoundtrackStatus, *repository.Soundtrack) {
	if s := findByName(existing, name); s != nil {
		if s.Hash == hash {
			return NotChanged, s
		}
		return SoundtrackChanged, s
	}
	for i := range existing {
		if existing[i].Hash == hash {
			return TitleChanged, &existing[i]
		}
	}
	return NewSoundtrack, nil
}

func findByName(existing []repository.Soundtrack, name string) *repository.Soundtrack {
	for i := range existing {
		if existing[i].SoundtrackName == name {
			return &existing[i]
		}
	}
	return nil
}

// removeBrokenSongs deletes songs whose file no longer exists.
func (r *syncRun) removeBrokenSongs(ctx context.Context) {
	songs, err := r.q.FetchAllSongs(ctx)
	if err != nil {
		r.addError("FetchAllSongs", err, "")
		return
	}

	var mu sync.Mutex
	var broken []uuid.UUID
	g := new(errgroup.Group)
	g.SetLimit(brokenWorkers)
	for _, song := range songs {
		g.Go(func() error {
			if _, err := os.Stat(song.Path); err != nil {
				logging.GetLogger().Warn("Broken song found", zap.String("path", song.Path))
				mu.Lock()
				broken = append(broken, song.Uuid)
				mu.Unlock()
			}
			return nil
		})
	}
	_ = g.Wait()

	if len(broken) > 0 {
		if err := r.q.RemoveBrokenSongs(ctx, broken); err != nil {
			r.addError("RemoveBrokenSongs", err, "")
		}
	}
}

// fingerprint summarises the names, sizes and modification times of all files in dir
// without reading their contents.
func fingerprint(dir string) (string, error) {
	h := sha1.New()
	err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}
		if d.IsDir() {
			return nil
		}
		info, err := d.Info()
		if err != nil {
			return err
		}
		rel, _ := filepath.Rel(dir, path)
		fmt.Fprintf(h, "%s\x00%d\x00%d\n", rel, info.Size(), info.ModTime().UnixNano())
		return nil
	})
	if err != nil {
		return "", err
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

func dirHash(dir string) (string, error) {
	directory, err := directory_checksum.ScanDirectory(dir, afero.NewOsFs())
	if err != nil {
		return "", err
	}
	return directory.ComputeDirectoryChecksums()
}

func fileHash(path string) (string, error) {
	f, err := os.Open(path)
	if err != nil {
		return "", err
	}
	defer f.Close()
	h := md5.New()
	if _, err := io.Copy(h, f); err != nil {
		return "", err
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

func isSong(entry fs.DirEntry) bool {
	return !entry.IsDir() && strings.EqualFold(filepath.Ext(entry.Name()), ".mp3")
}

func songName(fileName string) string {
	return strings.TrimSuffix(fileName, filepath.Ext(fileName))
}

// removeIDFiles removes legacy .id files from a soundtrack folder (UUID migration, see issue #29).
// It reports whether any file was removed.
func removeIDFiles(dir string) bool {
	entries, err := os.ReadDir(dir)
	if err != nil {
		logging.GetLogger().Error("Failed to read soundtrack directory for .id file cleanup",
			zap.String("path", dir), zap.Error(err))
		return false
	}

	anyRemoved := false
	for _, entry := range entries {
		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".id") {
			continue
		}
		path := filepath.Join(dir, entry.Name())
		if err := os.Remove(path); err != nil {
			logging.GetLogger().Error("Failed to remove .id file", zap.String("path", path), zap.Error(err))
			continue
		}
		logging.GetLogger().Info("Removed .id file", zap.String("path", path))
		anyRemoved = true
	}
	return anyRemoved
}