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" "music-server/internal/metrics" "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, besides hidden // ones like .sync and .uploads. var foldersToSkip = map[string]struct{}{"characters": {}, "dist": {}, "old": {}} // skipFolder reports whether a directory in the music path is not a soundtrack. func skipFolder(name string) bool { _, skip := foldersToSkip[name] return skip || strings.HasPrefix(name, ".") } 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 } // SyncFolder runs prepare and then syncs the soundtrack folder name, while no other sync // can run. It returns ErrSyncing when a sync is running. prepare can put the folder in // place; when it fails, nothing is synced. onComplete is not called, as only one // soundtrack changes. func (s *SyncService) SyncFolder(ctx context.Context, name string, prepare func() error) (repository.Soundtrack, error) { if !s.running.CompareAndSwap(false, true) { return repository.Soundtrack{}, ErrSyncing } defer s.running.Store(false) existing, err := s.q.GetAllSoundtracksIncludingDeleted(ctx) if err != nil { return repository.Soundtrack{}, err } if err := prepare(); err != nil { return repository.Soundtrack{}, err } run := &syncRun{q: s.q, musicPath: s.musicPath, existing: existing, progress: func() {}} run.syncSoundtrack(ctx, name) if len(run.errs) > 0 { return repository.Soundtrack{}, fmt.Errorf("syncing %q: %s", name, strings.Join(run.errs, "; ")) } if len(run.foundIDs) != 1 { return repository.Soundtrack{}, fmt.Errorf("syncing %q found %d soundtracks", name, len(run.foundIDs)) } if err := s.UpdateLibraryMetrics(ctx); err != nil { logging.GetLogger().Error("Failed to count the library", zap.Error(err)) } return s.q.GetSoundtrackById(ctx, run.foundIDs[0]) } // 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)) s.recordMetrics(ctx, full, time.Since(start), len(result.CatchedErrors) == 0) 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 init() { // Start both sync types at 0, so the series exist before the first sync. metrics.SyncDuration.WithLabelValues(syncType(false)) metrics.SyncDuration.WithLabelValues(syncType(true)) } // syncType is the type label of the sync metrics. func syncType(full bool) string { if full { return "full" } return "changes" } // recordMetrics updates the sync and library metrics after a sync. func (s *SyncService) recordMetrics(ctx context.Context, full bool, took time.Duration, success bool) { metrics.SyncDuration.WithLabelValues(syncType(full)).Observe(took.Seconds()) if success { metrics.SyncLastSuccess.SetToCurrentTime() } if err := s.UpdateLibraryMetrics(ctx); err != nil { logging.GetLogger().Error("Failed to count the library", zap.Error(err)) } } // UpdateLibraryMetrics sets the library size metrics from the database. func (s *SyncService) UpdateLibraryMetrics(ctx context.Context) error { count, err := s.q.CountLibrary(ctx) if err != nil { return err } metrics.LibrarySoundtracks.Set(float64(count.Soundtracks)) metrics.LibrarySongs.Set(float64(count.Songs)) return nil } 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 dir.IsDir() && !skipFolder(dir.Name()) { 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() metrics.BrokenSongs.Set(float64(len(broken))) 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 }