navidrome/scanner/phase_1_folders.go

package scanner

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"maps"
	"path"
	"slices"
	"sync"
	"sync/atomic"
	"time"

	"github.com/Masterminds/squirrel"
	ppl "github.com/google/go-pipeline/pkg/pipeline"
	"github.com/navidrome/navidrome/conf"
	"github.com/navidrome/navidrome/consts"
	"github.com/navidrome/navidrome/core/storage"
	"github.com/navidrome/navidrome/log"
	"github.com/navidrome/navidrome/model"
	"github.com/navidrome/navidrome/model/metadata"
	"github.com/navidrome/navidrome/utils"
	"github.com/navidrome/navidrome/utils/pl"
	"github.com/navidrome/navidrome/utils/slice"
)

func createPhaseFolders(ctx context.Context, state *scanState, ds model.DataStore) *phaseFolders {
	var jobs []*scanJob

	// Create scan jobs for all libraries
	for _, lib := range state.libraries {
		// Get target folders for this library if selective scan
		var targetFolders []string
		if state.isSelectiveScan() {
			targetFolders = state.targets[lib.ID]
		}

		job, err := newScanJob(ctx, ds, lib, state.fullScan, targetFolders)
		if err != nil {
			log.Error(ctx, "Scanner: Error creating scan context", "lib", lib.Name, err)
			state.sendError(err)
			continue
		}
		jobs = append(jobs, job)
	}

	walkCtx, stopWalk := context.WithCancelCause(ctx)
	return &phaseFolders{jobs: jobs, ctx: ctx, walkCtx: walkCtx, stopWalk: stopWalk, ds: ds, state: state,
		imageChanges: &imageChangeCollector{ds: ds}}
}

type scanJob struct {
	lib           model.Library
	fs            storage.MusicFS
	lastUpdates   map[string]model.FolderUpdateInfo // Holds last update info for all (DB) folders in this library
	targetFolders []string                          // Specific folders to scan (including all descendants)
	lock          sync.Mutex
	numFolders    atomic.Int64
}

func newScanJob(ctx context.Context, ds model.DataStore, lib model.Library, fullScan bool, targetFolders []string) (*scanJob, error) {
	// Get folder updates, optionally filtered to specific target folders
	lastUpdates, err := ds.Folder().GetFolderUpdateInfo(ctx, lib, targetFolders...)
	if err != nil {
		return nil, fmt.Errorf("getting last updates: %w", err)
	}

	fileStore, err := storage.For(lib.Path)
	if err != nil {
		log.Error(ctx, "Error getting storage for library", "library", lib.Name, "path", lib.Path, err)
		return nil, fmt.Errorf("getting storage for library: %w", err)
	}
	fsys, err := fileStore.FS()
	if err != nil {
		log.Error(ctx, "Error getting fs for library", "library", lib.Name, "path", lib.Path, err)
		return nil, fmt.Errorf("getting fs for library: %w", err)
	}

	// Ensure FullScanInProgress reflects the current scan request.
	// This is important when resuming an interrupted quick scan as a full scan:
	// the DB may have FullScanInProgress=false, but we need it true for isOutdated() to work correctly.
	lib.FullScanInProgress = lib.FullScanInProgress || fullScan

	return &scanJob{
		lib:           lib,
		fs:            fsys,
		lastUpdates:   lastUpdates,
		targetFolders: targetFolders,
	}, nil
}

// popLastUpdate retrieves and removes the last update info for the given folder ID
// This is used to track which folders have been found during the walk_dir_tree
func (j *scanJob) popLastUpdate(folderID string) model.FolderUpdateInfo {
	j.lock.Lock()
	defer j.lock.Unlock()

	lastUpdate := j.lastUpdates[folderID]
	delete(j.lastUpdates, folderID)
	return lastUpdate
}

// createFolderEntry creates a new folderEntry for the given path, using the last update info from the job
// to populate the previous update time and hash. It also removes the folder from the job's lastUpdates map.
// This is used to track which folders have been found during the walk_dir_tree.
func (j *scanJob) createFolderEntry(path string) *folderEntry {
	id := model.FolderID(j.lib, path)
	info := j.popLastUpdate(id)
	return newFolderEntry(j, id, path, info)
}

// phaseFolders represents the first phase of the scanning process, which is responsible
// for scanning all libraries and importing new or updated files. This phase involves
// traversing the directory tree of each library, identifying new or modified media files,
// and updating the database with the relevant information.
//
// The phaseFolders struct holds the context, data store, and jobs required for the scanning
// process. Each job represents a library being scanned, and contains information about the
// library, file system, and the last updates of the folders.
//
// The phaseFolders struct implements the phase interface, providing methods to produce
// folder entries, process folders, persist changes to the database, and log the results.
type phaseFolders struct {
	jobs             []*scanJob
	ds               model.DataStore
	ctx              context.Context //nolint:containedctx // phase runs under a single scan ctx
	walkCtx          context.Context //nolint:containedctx // cancelled when a folder fails to persist, so the walk stops early
	stopWalk         context.CancelCauseFunc
	state            *scanState
	prevAlbumPIDConf string
	imageChanges     *imageChangeCollector
}

func (p *phaseFolders) description() string {
	return "Scan all libraries and import new/updated files"
}

func (p *phaseFolders) producer() ppl.Producer[*folderEntry] {
	return ppl.NewProducer(func(put func(entry *folderEntry)) error {
		var err error
		p.prevAlbumPIDConf, err = p.ds.Property().DefaultGet(p.ctx, consts.PIDAlbumKey, "")
		if err != nil {
			return fmt.Errorf("getting album PID conf: %w", err)
		}

		// TODO Parallelize multiple job when we have multiple libraries
		var total int64
		var totalChanged int64
		for _, job := range p.jobs {
			if utils.IsCtxDone(p.walkCtx) {
				break
			}

			outputChan, err := walkDirTree(p.walkCtx, job, job.targetFolders...)
			if err != nil {
				log.Warn(p.ctx, "Scanner: Error scanning library", "lib", job.lib.Name, err)
			}
			for folder := range pl.ReadOrDone(p.walkCtx, outputChan) {
				job.numFolders.Add(1)
				p.state.sendProgress(&ProgressInfo{
					LibID:     job.lib.ID,
					FileCount: uint32(len(folder.audioFiles)),
					Path:      folder.path,
					Phase:     "1",
				})

				// Log folder info
				log.Trace(p.ctx, "Scanner: Checking folder state", " folder", folder.path, "_updTime", folder.updTime,
					"_modTime", folder.modTime, "_lastScanStartedAt", folder.job.lib.LastScanStartedAt,
					"numAudioFiles", len(folder.audioFiles), "numImageFiles", len(folder.imageFiles),
					"numPlaylists", len(folder.playlistFiles), "numSubfolders", folder.numSubFolders)

				// Check if folder is outdated
				if folder.isOutdated() {
					if !p.state.fullScan {
						// Ancestor folders need a row even with no files of their own: artwork
						// resolution climbs them, and an image added later needs a state to diff.
						if folder.isEmpty() && folder.isNew() {
							log.Trace(p.ctx, "Scanner: Skipping new empty folder", "folder", folder.path, "lib", job.lib.Name)
							continue
						}
						log.Debug(p.ctx, "Scanner: Detected changes in folder", "folder", folder.path, "lastUpdate", folder.modTime, "lib", job.lib.Name)
					}
					totalChanged++
					folder.elapsed.Stop()
					put(folder)
				} else {
					log.Trace(p.ctx, "Scanner: Skipping up-to-date folder", "folder", folder.path, "lastUpdate", folder.modTime, "lib", job.lib.Name)
				}
			}
			total += job.numFolders.Load()
		}
		log.Debug(p.ctx, "Scanner: Finished loading all folders", "numFolders", total, "numChanged", totalChanged)
		return nil
	}, ppl.Name("traverse filesystem"))
}

func (p *phaseFolders) measure(entry *folderEntry) func() time.Duration {
	entry.elapsed.Start()
	return func() time.Duration { return entry.elapsed.Stop() }
}

func (p *phaseFolders) stages() []ppl.Stage[*folderEntry] {
	return []ppl.Stage[*folderEntry]{
		ppl.NewStage(p.processFolder, ppl.Name("process folder"), ppl.Concurrency(conf.Server.DevScannerThreads)),
		// persistChanges is not reentrant, so it always has to run with concurrency=1
		ppl.NewStage(p.persistChanges, ppl.Name("persist changes"), ppl.Concurrency(1)),
		ppl.NewStage(p.logFolder, ppl.Name("log results")),
	}
}

func (p *phaseFolders) processFolder(entry *folderEntry) (*folderEntry, error) {
	defer p.measure(entry)()
	if err := context.Cause(p.walkCtx); err != nil {
		return entry, err
	}

	// Load children mediafiles from DB
	cursor, err := p.ds.MediaFile().GetCursor(p.ctx, model.QueryOptions{
		Filters: squirrel.And{squirrel.Eq{"folder_id": entry.id}},
	})
	if err != nil {
		log.Error(p.ctx, "Scanner: Error loading mediafiles from DB", "folder", entry.path, err)
		return entry, err
	}
	dbTracks := make(map[string]*model.MediaFile)
	for mf, err := range cursor {
		if err != nil {
			log.Error(p.ctx, "Scanner: Error loading mediafiles from DB", "folder", entry.path, err)
			return entry, err
		}
		dbTracks[mf.Path] = &mf
	}

	// Get list of files to import, based on modtime (or all if fullScan),
	// leave in dbTracks only tracks that are missing (not found in the FS)
	filesToImport := make(map[string]*model.MediaFile, len(entry.audioFiles))
	for afPath, af := range entry.audioFiles {
		fullPath := path.Join(entry.path, afPath)
		dbTrack, foundInDB := dbTracks[fullPath]
		if !foundInDB || p.state.fullScan {
			filesToImport[fullPath] = dbTrack
		} else {
			info, err := af.Info()
			if err != nil {
				log.Warn(p.ctx, "Scanner: Error getting file info", "folder", entry.path, "file", af.Name(), err)
				p.state.sendWarning(fmt.Sprintf("Error getting file info for %s/%s: %v", entry.path, af.Name(), err))
				return entry, nil
			}
			if info.ModTime().After(dbTrack.UpdatedAt) || dbTrack.Missing {
				filesToImport[fullPath] = dbTrack
			}
		}
		delete(dbTracks, fullPath)
	}

	// Remaining dbTracks are tracks that were not found in the FS, so they should be marked as missing
	entry.missingTracks = slices.Collect(maps.Values(dbTracks))

	// Load metadata from files that need to be imported
	if len(filesToImport) > 0 {
		err = p.loadTagsFromFiles(entry, filesToImport)
		if err != nil {
			log.Warn(p.ctx, "Scanner: Error loading tags from files. Skipping", "folder", entry.path, err)
			p.state.sendWarning(fmt.Sprintf("Error loading tags from files in %s: %v", entry.path, err))
			return entry, nil
		}

		p.createAlbumsFromMediaFiles(entry)
		p.createArtistsFromMediaFiles(entry)
	}

	return entry, nil
}

const filesBatchSize = 200

// loadTagsFromFiles reads metadata from the files in the given list and populates
// the entry's tracks and tags with the results.
func (p *phaseFolders) loadTagsFromFiles(entry *folderEntry, toImport map[string]*model.MediaFile) error {
	tracks := make([]model.MediaFile, 0, len(toImport))
	uniqueTags := make(map[string]model.Tag, len(toImport))
	for chunk := range slice.CollectChunks(maps.Keys(toImport), filesBatchSize) {
		allInfo, err := entry.job.fs.ReadTags(chunk...)
		if err != nil {
			log.Warn(p.ctx, "Scanner: Error extracting metadata from files. Skipping", "folder", entry.path, err)
			return err
		}
		for filePath, info := range allInfo {
			md := metadata.New(filePath, info)
			track := md.ToMediaFile(entry.job.lib.ID, entry.id)
			tracks = append(tracks, track)
			for _, t := range track.Tags.FlattenAll() {
				uniqueTags[t.ID] = t
			}

			// Keep track of any album ID changes, to reassign annotations later
			prevAlbumID := ""
			if prev := toImport[filePath]; prev != nil {
				prevAlbumID = prev.AlbumID
			} else {
				prevAlbumID = md.AlbumID(track, p.prevAlbumPIDConf)
			}
			_, ok := entry.albumIDMap[track.AlbumID]
			if prevAlbumID != track.AlbumID && !ok {
				entry.albumIDMap[track.AlbumID] = prevAlbumID
			}
		}
	}
	entry.tracks = tracks
	entry.tags = slices.Collect(maps.Values(uniqueTags))
	return nil
}

// createAlbumsFromMediaFiles groups the entry's tracks by album ID and creates albums
func (p *phaseFolders) createAlbumsFromMediaFiles(entry *folderEntry) {
	grouped := slice.Group(entry.tracks, func(mf model.MediaFile) string { return mf.AlbumID })
	albums := make(model.Albums, 0, len(grouped))
	for _, group := range grouped {
		songs := model.MediaFiles(group)
		album := songs.ToAlbum()
		albums = append(albums, album)
	}
	entry.albums = albums
}

// createArtistsFromMediaFiles creates artists from the entry's tracks
func (p *phaseFolders) createArtistsFromMediaFiles(entry *folderEntry) {
	participants := make(model.Participants, len(entry.tracks)*3) // preallocate ~3 artists per track
	for _, track := range entry.tracks {
		participants.Merge(track.Participants)
	}
	entry.artists = participants.AllArtists()
}

func (p *phaseFolders) persistChanges(entry *folderEntry) (*folderEntry, error) {
	defer p.measure(entry)()
	p.state.changesDetected.Store(true)

	ctx := log.NewContext(p.ctx, "folder", entry.path)
	err := p.ds.WithTxRetry(ctx, func(ctx context.Context, tx model.DataStore) error {
		return p.persistFolder(ctx, tx, entry)
	}, "scanner: persist changes")
	if err != nil {
		log.Error(ctx, "Scanner: Error persisting changes to DB", err)
		p.stopWalk(err)
		return entry, err
	}

	// A new folder's albums/artists were enqueued with it; only pre-existing folders need the diff.
	if !entry.isNew() {
		if changed, artistImage := entry.imagesChanged(); changed {
			p.imageChanges.record(entry.job.lib, imageChangedFolder{
				id: entry.id, path: entry.path, artistImage: artistImage,
			})
		}
	}
	return entry, nil
}

// persistFolder writes the folder in tx. WithTxRetry may rerun it after a rollback.
func (p *phaseFolders) persistFolder(ctx context.Context, tx model.DataStore, entry *folderEntry) error {
	// Collect artwork queue items for changed albums/artists, enqueued in the same transaction
	var queueItems []model.ArtworkQueueItem
	// persistAlbum consumes the map, so a rerun needs the original
	albumIDMap := maps.Clone(entry.albumIDMap)

	// Instantiate all repositories just once per folder
	folderRepo := tx.Folder()
	tagRepo := tx.Tag()
	artistRepo := tx.Artist()
	libraryRepo := tx.Library()
	albumRepo := tx.Album()
	mfRepo := tx.MediaFile()

	// Save folder to DB
	folder := entry.toFolder()
	err := folderRepo.Put(ctx, folder)
	if err != nil {
		return fmt.Errorf("persisting folder: %w", err)
	}

	// Save all tags to DB
	err = tagRepo.Add(ctx, entry.job.lib.ID, entry.tags...)
	if err != nil {
		return fmt.Errorf("persisting tags: %w", err)
	}

	// Save all new/modified artists to DB. Their information will be incomplete, but they will be refreshed later
	for i := range entry.artists {
		err = artistRepo.Put(ctx, &entry.artists[i], "name",
			"mbz_artist_id", "sort_artist_name", "order_artist_name", "full_text", "search_normalized", "updated_at")
		if err != nil {
			return fmt.Errorf("persisting artist %q: %w", entry.artists[i].Name, err)
		}
		err = libraryRepo.AddArtist(ctx, entry.job.lib.ID, entry.artists[i].ID)
		if err != nil {
			return fmt.Errorf("adding artist %q to library: %w", entry.artists[i].Name, err)
		}
		if entry.artists[i].Name != consts.UnknownArtist && entry.artists[i].Name != consts.VariousArtists {
			queueItems = append(queueItems, scanArtworkItem(model.KindArtistArtwork, entry.artists[i].ID))
		}
	}

	// Save all new/modified albums to DB. Their information will be incomplete, but they will be refreshed later
	for i := range entry.albums {
		err = p.persistAlbum(albumRepo, &entry.albums[i], albumIDMap)
		if err != nil {
			return err
		}
		if entry.albums[i].Name != consts.UnknownAlbum {
			queueItems = append(queueItems, scanArtworkItem(model.KindAlbumArtwork, entry.albums[i].ID))
		}
	}

	// Save all tracks to DB
	for i := range entry.tracks {
		err = mfRepo.Put(ctx, &entry.tracks[i])
		if err != nil {
			return fmt.Errorf("persisting track %q: %w", entry.tracks[i].Path, err)
		}
	}

	// A re-imported track returns to unresolved so new embedded art is picked up lazily.
	if len(entry.tracks) > 0 {
		trackIDs := slice.Map(entry.tracks, func(t model.MediaFile) string { return t.ID })
		if err := tx.Artwork().DeleteForItems(ctx, model.KindMediaFileArtwork, trackIDs); err != nil {
			log.Warn(ctx, "Scanner: could not invalidate media_file artwork", err)
		}
	}

	// Mark all missing tracks as not available
	if len(entry.missingTracks) > 0 {
		err = mfRepo.MarkMissing(ctx, true, entry.missingTracks...)
		if err != nil {
			return fmt.Errorf("marking missing tracks: %w", err)
		}

		// Touch all albums that have missing tracks, so they get refreshed in later phases
		groupedMissingTracks := slice.ToMap(entry.missingTracks, func(mf *model.MediaFile) (string, struct{}) {
			return mf.AlbumID, struct{}{}
		})
		albumsToUpdate := slices.Collect(maps.Keys(groupedMissingTracks))
		err = albumRepo.Touch(ctx, albumsToUpdate...)
		if err != nil {
			return fmt.Errorf("touching albums %v: %w", albumsToUpdate, err)
		}
	}

	// Enqueue artwork resolution for changed albums/artists. Never fails the scan.
	// A full scan re-imports every track, so a re-import is no evidence the art changed.
	if len(queueItems) > 0 {
		queue := tx.ArtworkQueue()
		enqueue := queue.Enqueue
		if p.state.fullScan {
			enqueue = queue.EnqueueIfMissing
		}
		if err := enqueue(ctx, queueItems...); err != nil {
			log.Warn(ctx, "Scanner: could not enqueue artwork resolution", err)
		}
	}
	return nil
}

// persistAlbum persists the given album to the database, and reassigns annotations from the previous album ID
func (p *phaseFolders) persistAlbum(repo model.AlbumRepository, a *model.Album, idMap map[string]string) error {
	prevID := idMap[a.ID]
	log.Trace(p.ctx, "Persisting album", "album", a.Name, "albumArtist", a.AlbumArtist, "id", a.ID, "prevID", cmp.Or(prevID, "nil"))
	if err := repo.Put(p.ctx, a); err != nil {
		return fmt.Errorf("persisting album %s: %w", a.ID, err)
	}
	if prevID == "" {
		return nil
	}

	// Reassign annotation from previous album to new album
	log.Trace(p.ctx, "Reassigning album annotations", "from", prevID, "to", a.ID, "album", a.Name)
	if err := repo.ReassignAnnotation(p.ctx, prevID, a.ID); err != nil {
		log.Warn(p.ctx, "Scanner: Could not reassign annotations", "from", prevID, "to", a.ID, "album", a.Name, err)
		p.state.sendWarning(fmt.Sprintf("Could not reassign annotations from %s to %s ('%s'): %v", prevID, a.ID, a.Name, err))
	}

	// Keep created_at field from previous instance of the album
	if err := repo.CopyAttributes(p.ctx, prevID, a.ID, "created_at"); err != nil {
		// Silently ignore when the previous album is not found
		if !errors.Is(err, model.ErrNotFound) {
			log.Warn(p.ctx, "Scanner: Could not copy fields", "from", prevID, "to", a.ID, "album", a.Name, err)
			p.state.sendWarning(fmt.Sprintf("Could not copy fields from %s to %s ('%s'): %v", prevID, a.ID, a.Name, err))
		}
	}
	// Don't keep track of this mapping anymore
	delete(idMap, a.ID)
	return nil
}

func (p *phaseFolders) logFolder(entry *folderEntry) (*folderEntry, error) {
	logCall := log.Info
	if entry.isEmpty() {
		logCall = log.Trace
	}
	logCall(p.ctx, "Scanner: Completed processing folder",
		"audioCount", len(entry.audioFiles), "imageCount", len(entry.imageFiles), "plsCount", len(entry.playlistFiles),
		"elapsed", entry.elapsed.Elapsed(), "tracksMissing", len(entry.missingTracks),
		"tracksImported", len(entry.tracks), "library", entry.job.lib.Name, consts.Zwsp+"folder", entry.path)
	return entry, nil
}

func (p *phaseFolders) finalize(err error) error {
	p.stopWalk(nil)
	defer p.imageChanges.enqueue(p.ctx)
	// A failed phase may not have walked every folder, and unvisited ones must not be marked missing
	if err != nil {
		return err
	}
	return p.ds.WithTxRetry(p.ctx, func(ctx context.Context, tx model.DataStore) error {
		for _, job := range p.jobs {
			// Mark all folders that were not updated as missing
			if len(job.lastUpdates) == 0 {
				continue
			}
			folderIDs := slices.Collect(maps.Keys(job.lastUpdates))
			if err := tx.Folder().MarkMissing(ctx, true, folderIDs...); err != nil {
				return fmt.Errorf("marking missing folders in %s: %w", job.lib.Name, err)
			}
			if err := tx.MediaFile().MarkMissingByFolder(ctx, true, folderIDs...); err != nil {
				return fmt.Errorf("marking tracks in missing folders in %s: %w", job.lib.Name, err)
			}
			// Touch all albums that have missing folders, so they get refreshed in later phases
			if _, err := tx.Album().TouchByMissingFolder(ctx); err != nil {
				return fmt.Errorf("touching albums with missing folders in %s: %w", job.lib.Name, err)
			}
		}
		return nil
	}, "scanner: finalize phaseFolders")
}

var _ phase[*folderEntry] = (*phaseFolders)(nil)