package main

import (
	"cmp"
	"context"
	"crypto/sha256"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"os/signal"
	"path/filepath"
	"runtime"
	"slices"
	"strings"
	"sync"
	"syscall"
	"time"
)

// The hash phase gives every file it reads three hashes (see
// hashSignature and README "Duplicate detection"): head and tail, of
// its first and last headTailWindow bytes, and content. The content
// hash covers the whole file below wholeFileMax, and only sampled
// windows at or above it, so such a file is never read in full.

// headTailWindow is the number of bytes hashed from each end of a file
// for its head and tail. In a file no larger than this, each window is
// the whole file.
const headTailWindow = 64 * 1024

// wholeFileMax is the size boundary between the two content hashes: a
// file strictly smaller than this is content-hashed in full; a file
// this size or larger is content-hashed by sampling.
const wholeFileMax = 100 * 1024 * 1024

// sampleStride is the spacing between content samples for large files:
// one window is read at each gigabyte-aligned offset (0, 1 GiB, ...).
const sampleStride = 1024 * 1024 * 1024

// sampleWindow is the number of bytes read at each large-file sample
// offset, truncated at end of file.
const sampleWindow = 1024 * 1024

// workQueueDepth bounds the job and result channels feeding the walk
// and hash worker pools.
const workQueueDepth = 1024

// errInterrupted reports a scan stopped by SIGINT or SIGTERM. runScan
// has already printed its line, so run prints nothing more.
var errInterrupted = errors.New("scan interrupted")

// fileRec carries one statted file between the scan phases. dev and
// ino identify the underlying inode so hard-linked paths can share
// one read; both are zero when the platform exposes no inode.
type fileRec struct {
	path  string
	size  int64
	mtime time.Time
	dev   uint64
	ino   uint64
}

// fileMeta is the in-memory index entry for one existing database
// record: just enough for change detection. Hashes stay on disk; at
// tens of millions of records they would dominate the scan's memory.
type fileMeta struct {
	size  int64
	mtime time.Time
}

// scanUnit is what every scan phase counts, as its progress display
// names it.
const scanUnit = "files"

// scanPhase names one phase of a scan with its place among all of them,
// as "[2/4] walk", for the phase's progress display. The list here is
// the one place the phases are counted.
func scanPhase(name string) string {
	return phaseLabel([]string{"load", "walk", "hash", "update"}, name)
}

// runScan implements the scan subcommand: four sequential phases —
// load (which reads the existing records), walk (which stats each file
// as it is discovered), hash, update — that synchronize the persistent
// database with the filesystem state under the PATH operands. Flag
// parsing and the at-least-one-operand check are done by cobra. Before
// anything else, the scan names the database on stderr, saying whether
// it is new. The scan holds the lock on the database for its whole run,
// so a second scan fails before it walks the filesystem or opens the
// database. Errors are returned rather than exiting, so that the
// deferred close — which takes the database out of WAL mode — always
// runs, and the lock is released after it. When ctx is cancelled, as by
// the SIGINT or SIGTERM that interruptContext catches, the scan keeps
// what it has hashed (see syncScan), prints how many files its walk
// reached, and returns errInterrupted. workers must be at least 1; the
// scan command rejects anything less. Before the lock is taken, the
// scan lists the filesystems that are not persistent, on which the walk
// records nothing; unless noDefaultExcludes is set, the walk does not
// enter the default excludes either. Once the database is open, the
// scan notes its files, which the walk passes over.
func runScan(ctx context.Context, roots []string, workers int,
	oneFS, noDefaultExcludes bool,
) error {
	dbPath, source, err := databasePath()
	if err != nil {
		return err
	}

	// A database file missing now is created by lockScanDatabase below.
	what := "database"

	_, err = os.Stat(dbPath)
	if errors.Is(err, fs.ErrNotExist) {
		what = "new database"
	}

	fmt.Fprintf(os.Stderr, "scan: %s %s (%s)\n", what, escapePath(dbPath),
		source)

	roots, err = resolveRoots(roots)
	if err != nil {
		return err
	}

	nonPersistent, err := nonPersistentFilesystems()
	if err != nil {
		return fmt.Errorf("list the mounted filesystems: %w", err)
	}

	rules := walkRules{nonPersistent: nonPersistent}
	if !noDefaultExcludes {
		rules.excludes = defaultExcludes(runtime.GOOS)
	}

	lock, err := lockScanDatabase(dbPath)
	if err != nil {
		return err
	}

	defer func() { _ = lock.Close() }()

	db, err := openScanDatabase(ctx, dbPath)
	if err != nil && ctx.Err() != nil {
		// Interrupted while opening; SQLite may report that with an
		// error of its own rather than the context's.
		return interrupted(0)
	}

	if err != nil {
		return err
	}

	defer closeScanDatabase(ctx, db, dbPath)

	rules.databaseFiles, err = databaseFiles(dbPath)
	if err != nil {
		return err
	}

	st, err := syncScan(ctx, db, roots, workers, oneFS, rules)
	if errors.Is(err, context.Canceled) {
		return interrupted(st.walked)
	}

	if err != nil {
		return fmt.Errorf("update database %s: %w", dbPath, err)
	}

	fmt.Fprintf(os.Stderr,
		"scan: %d files seen (%d added, %d updated, %d removed, "+
			"%d unchanged), %d skipped\n",
		st.added+st.updated+st.unchanged, st.added, st.updated,
		st.removed, st.unchanged, st.skipped)

	return nil
}

// interruptContext returns a copy of ctx that the first SIGINT or
// SIGTERM cancels; the scan command runs the scan under it. stop
// releases the signals.
func interruptContext(ctx context.Context) (context.Context, func()) {
	// A SIGINT ignored from the start, as by a script's background job,
	// stays ignored.
	signals := []os.Signal{syscall.SIGTERM}
	if !signal.Ignored(syscall.SIGINT) {
		signals = append(signals, syscall.SIGINT)
	}

	ctx, stop := signal.NotifyContext(ctx, signals...)

	// Stopping restores the default handling, so a second signal ends
	// the process at once.
	context.AfterFunc(ctx, stop)

	return ctx, stop
}

// interrupted prints the line for a scan stopped by a signal after its
// walk reached walked files, and returns errInterrupted.
func interrupted(walked int) error {
	fmt.Fprintf(os.Stderr, "scan: interrupted after %d files\n", walked)

	return errInterrupted
}

// resolveRoots converts each PATH operand to an absolute, lexically
// cleaned path (symlinks are not resolved) and verifies that it
// exists. Database records are keyed by absolute path, so scan results
// must not depend on the working directory.
func resolveRoots(roots []string) ([]string, error) {
	abs := make([]string, 0, len(roots))

	for _, root := range roots {
		a, err := filepath.Abs(root)
		if err != nil {
			return nil, fmt.Errorf("resolve %s: %w", root, err)
		}

		// A nonexistent operand is a fatal error before any scanning.
		_, err = os.Lstat(a)
		if err != nil {
			return nil, err
		}

		abs = append(abs, a)
	}

	return abs, nil
}

// pruneRoots drops operands already covered by another operand:
// duplicates and any operand lying under another one. An operand under
// another one is kept, though, when the walk of the deepest kept operand
// above it does not reach it (see walkReaches). Every remaining file is
// then reachable through exactly one root, so walked paths are unique
// without keeping a scan-wide set of every path seen.
func pruneRoots(roots []string, oneFS bool, rules walkRules,
	device func(path string) (uint64, bool),
) []string {
	// Shorter-first ordering guarantees an ancestor is kept before any
	// operand under it is considered.
	sorted := slices.Clone(roots)
	slices.SortFunc(sorted, func(a, b string) int {
		if len(a) != len(b) {
			return len(a) - len(b)
		}

		return strings.Compare(a, b)
	})

	kept := make([]string, 0, len(sorted))

	for _, r := range sorted {
		above := deepestRoot(r, kept)
		if above == "" {
			kept = append(kept, r)

			continue
		}

		// A duplicate is dropped, and so is an operand the walk of above
		// reaches.
		if above != r && !walkReaches(above, r, oneFS, rules, device) {
			kept = append(kept, r)
		}
	}

	return kept
}

// walkReaches reports whether the walk of operand above reaches r, an
// operand under it. That walk does not enter a default exclude in
// rules, nor a directory that device puts on a filesystem in rules that
// is not persistent, nor, with -x, one that device puts on a different
// filesystem than above, when device knows above's. It reaches every
// regular file in a directory it enters, so for a regular-file operand
// the directories checked start at its directory; they end below above.
func walkReaches(above, r string, oneFS bool, rules walkRules,
	device func(path string) (uint64, bool),
) bool {
	aboveDev, aboveDevOK := device(above)

	dir := r

	fi, err := os.Lstat(r)
	if err == nil && fi.Mode().IsRegular() {
		dir = filepath.Dir(r)
	}

	for ; dir != above; dir = filepath.Dir(dir) {
		if isDefaultExclude(dir, rules.excludes) {
			return false
		}

		dev, ok := device(dir)
		if !ok {
			continue
		}

		if _, found := rules.nonPersistent[dev]; found {
			return false
		}

		if oneFS && aboveDevOK && dev != aboveDev {
			return false
		}
	}

	return true
}

// deviceOfPath returns the filesystem device ID of the file at path,
// when lstat succeeds and the platform exposes one.
func deviceOfPath(path string) (uint64, bool) {
	fi, err := os.Lstat(path)
	if err != nil {
		return 0, false
	}

	return deviceOfInfo(fi)
}

// scanStats summarizes one scan's database synchronization for the
// final stderr summary, or for the line an interrupted scan prints.
type scanStats struct {
	walked    int // files the walk reached
	added     int
	updated   int
	removed   int
	unchanged int
	skipped   int
}

// scanState carries one scan's evolving state across its phases.
// Entries consumed from existing mark files verified this run;
// whatever remains after the walk and hash phases is deleted by the
// update phase.
type scanState struct {
	db       *sql.DB
	existing map[string]fileMeta
	toHash   []fileRec // new and changed files: must be read
	batch    []scanRec
	st       scanStats
}

// syncScan synchronizes the database with the filesystem under roots;
// see runPhases. When ctx is cancelled, as by an interrupt, it commits
// the hashed records still waiting in the batch, starts no other write
// or deletion, and returns the cancellation.
func syncScan(ctx context.Context, db *sql.DB, roots []string,
	workers int, oneFS bool, rules walkRules,
) (scanStats, error) {
	s := &scanState{db: db}

	err := s.runPhases(ctx, roots, workers, oneFS, rules)
	if err == nil || ctx.Err() == nil {
		return s.st, err
	}

	// The one write made after the cancellation, so it cannot use ctx.
	err = applyChanges(context.WithoutCancel(ctx), db, s.batch, nil, nil)
	if err != nil {
		return s.st, err
	}

	return s.st, ctx.Err()
}

// runPhases synchronizes the database with the filesystem under roots
// in four sequential phases: load (read the existing records), walk
// (enumerate and stat every file), hash (read only the new or changed
// files, committing results in batches as they arrive), and update
// (delete the records the scan no longer verifies). Records outside the
// roots are never touched. Operands the walk cannot start from, or that
// are not on a persistent filesystem, are dropped first, so the records
// beneath them are treated as if those operands had not been given.
// Below an operand, the walk follows rules.
func (s *scanState) runPhases(ctx context.Context, roots []string,
	workers int, oneFS bool, rules walkRules,
) error {
	// Types and filesystems are checked before pruning so that an
	// operand under a dropped one is still scanned, not dropped as lying
	// under it, and so that an operand inside a .zfs directory gets its
	// warning even when it lies under another operand.
	roots = pruneRoots(s.walkableRoots(roots, rules), oneFS, rules,
		deviceOfPath)

	err := s.loadIndex(ctx, roots)
	if err != nil {
		return err
	}

	s.walkPhase(startWalk(ctx, roots, oneFS, rules, workers), roots)

	// A cancelled walk stops early, so every file it never reached would
	// look vanished to the update phase. Stop before anything is written
	// or deleted.
	err = ctx.Err()
	if err != nil {
		return err
	}

	err = s.hashPhase(ctx, workers)
	if err != nil {
		return err
	}

	return s.updatePhase(ctx)
}

// walkableRoots returns the operands the walk can start from: regular
// files, and directories not named .zfs, none of them inside a
// directory named .zfs. Every other operand is warned about, counted as
// skipped, and dropped. A dropped operand is no longer a root, so the
// records stored beneath it are treated as if it had not been given:
// those under no other root count as outside the roots and are not
// deleted as unverified. An operand on a filesystem in rules that is not
// persistent is dropped the same way, with a line saying so that is not
// a warning, and is not counted as skipped. An operand that fails lstat
// here is kept, and the walk warns about it unless it no longer exists.
func (s *scanState) walkableRoots(roots []string, rules walkRules) []string {
	kept := make([]string, 0, len(roots))

	for _, root := range roots {
		fi, err := os.Lstat(root)
		if err == nil {
			warn := operandWarning(root, fi)
			if warn != "" {
				s.st.skipped++

				fmt.Fprintln(os.Stderr, escapePath(warn))

				continue
			}

			line := rules.nonPersistentLine(root, fi)
			if line != "" {
				fmt.Fprintln(os.Stderr, escapePath(line))

				continue
			}
		}

		kept = append(kept, root)
	}

	return kept
}

// loadIndex indexes the database records under the scan roots for
// change detection. It reads every record, those outside the roots
// included, so its display counts against the number of records.
func (s *scanState) loadIndex(ctx context.Context, roots []string) error {
	// Indexing tens of millions of records takes real time; without a
	// display the scan looks hung before the walk begins. Counting them
	// first gives the display its total.
	prog := newProgress(progressParams{
		label: scanPhase("load"), unit: scanUnit, total: -1,
		waitingFor: countingRecords, stalledOn: "db",
	})
	defer prog.finish()

	total, err := countRecords(ctx, s.db)
	if err != nil {
		return err
	}

	prog.setTotal(total)

	s.existing = make(map[string]fileMeta)

	return loadFileMeta(ctx, s.db,
		func(path string, size int64, mtime time.Time) {
			prog.increment()

			if underAnyRoot(path, roots) {
				s.existing[path] = fileMeta{size: size, mtime: mtime}
			}
		})
}

// walkPhase drains the walk and queues every new or changed file for
// the hash phase. An unchanged file needs nothing more, and its record
// counts as verified. A record at or below a path the walk left out,
// because -x stopped at it, it is a default exclude, or it is not on a
// persistent filesystem, counts as outside the roots: it is not
// deleted. That does not hold below a root at or inside that path,
// since that root's own walk verifies the records below it; nor is a
// line printed for such a root.
func (s *scanState) walkPhase(events <-chan walkEvent, roots []string) {
	// The records the last scan left under the roots estimate how many
	// files the walk will find. A first scan has none, and no estimate.
	estimate := int64(len(s.existing))
	if estimate == 0 {
		estimate = -1
	}

	prog := newProgress(progressParams{
		label: scanPhase("walk"), unit: scanUnit, total: estimate,
		estimate: true, stalledOn: "disk",
	})

	var leftOut []string

	for ev := range events {
		if ev.otherFSDir != "" {
			leftOut = append(leftOut, ev.otherFSDir)

			continue
		}

		// The line saying so is printed as a warning is, but what the
		// walk leaves out is not counted as skipped. A default exclude
		// that is itself a root is walked as one, so it gets no line.
		if ev.leftOut != "" {
			leftOut = append(leftOut, ev.leftOut)

			if !slices.Contains(roots, ev.leftOut) {
				prog.warnf("%s", ev.warn)
			}

			continue
		}

		if ev.fail {
			s.st.skipped++

			prog.warnf("%s", ev.warn)

			continue
		}

		s.st.walked++

		prog.increment()

		old, ok := s.existing[ev.rec.path]
		if !ok || old.size != ev.rec.size || mtimeAfter(ev.rec.mtime, old.mtime) {
			s.toHash = append(s.toHash, ev.rec)

			continue
		}

		delete(s.existing, ev.rec.path)

		s.st.unchanged++
	}

	prog.finish()

	for path := range s.existing {
		// Kept only when a path the walk left out lies below the deepest
		// root above the record, or is the record's own. A root at or
		// below that path was walked on its own and did not find the
		// file. Both contain the record, so the longer path is the
		// deeper.
		left := deepestRoot(path, leftOut)
		if len(left) > len(deepestRoot(path, roots)) {
			delete(s.existing, path)
		}
	}
}

// resolve counts one written record as added or updated and marks its
// path verified.
func (s *scanState) resolve(path string) {
	if _, ok := s.existing[path]; ok {
		s.st.updated++

		delete(s.existing, path)

		return
	}

	s.st.added++
}

// hashRuns sorts the queued files into inode order and groups paths
// sharing an inode into runs. Inode-ordered reads minimize seeking on
// spinning disks, and each run of hard-linked paths is read once,
// with every path sharing the result. Files without an inode identity
// are never merged.
func hashRuns(toHash []fileRec) [][]fileRec {
	slices.SortFunc(toHash, func(a, b fileRec) int {
		if c := cmp.Compare(a.dev, b.dev); c != 0 {
			return c
		}

		if c := cmp.Compare(a.ino, b.ino); c != 0 {
			return c
		}

		return strings.Compare(a.path, b.path)
	})

	var runs [][]fileRec

	start := 0

	for i := 1; i <= len(toHash); i++ {
		if i == len(toHash) || !sameInode(toHash[i-1], toHash[i]) {
			runs = append(runs, toHash[start:i])
			start = i
		}
	}

	return runs
}

// sameInode reports whether two records name the same underlying
// inode. Records without an inode identity never match.
func sameInode(a, b fileRec) bool {
	return (a.dev != 0 || a.ino != 0) && a.dev == b.dev && a.ino == b.ino
}

// isDatabaseFile reports whether f, a file the walk reached, is one of
// the scan's own database files in rules, which the walk passes over
// without a line and without counting it.
func isDatabaseFile(f fileRec, rules walkRules) bool {
	return slices.ContainsFunc(rules.databaseFiles, func(d fileRec) bool {
		return sameInode(d, f)
	})
}

// hashPhase hashes every queued file with hashSignature, one read per
// inode run, committing completed records to the database in batches
// as results arrive, so a long scan persists its progress as it goes
// (an interrupted scan resumes cheaply: the next run skips everything
// already recorded). A run that fails to hash is warned about and
// counted as skipped; stale records for its paths, if any, are deleted
// by the update phase. A run whose file no longer exists gets no record
// and no warning, and is not counted as skipped. The total counts actual
// reads, so the display's time remaining is meaningful.
//
// Returning early — a failed database write, or a cancelled scan — must
// not strand the pool: the feeder would park forever on a full jobs
// channel and every worker on a full results channel. The deferred stop
// is what prevents that.
func (s *scanState) hashPhase(ctx context.Context, workers int) error {
	runs := hashRuns(s.toHash)
	s.toHash = nil

	pool := startHashPool(ctx, runs, workers, hashSignature)
	defer pool.stop()

	// Between results the phase may wait on a file being read or on a
	// batch of records being written.
	prog := newProgress(progressParams{
		label: scanPhase("hash"), unit: scanUnit, total: int64(len(runs)),
		stalledOn: "disk/db",
	})
	defer prog.finish()

	for range runs {
		var r hashResult

		select {
		case r = <-pool.results:
		case <-ctx.Done():
			return ctx.Err()
		}

		prog.increment()

		// A file removed since the walk is no error: it gets no record.
		if errors.Is(r.err, fs.ErrNotExist) {
			continue
		}

		if r.err != nil {
			s.st.skipped += len(r.run)

			prog.warnf("hash %s: %v", r.run[0].path, r.err)

			continue
		}

		err := s.recordRun(ctx, r)
		if err != nil {
			return err
		}
	}

	return nil
}

// recordRun folds one hash result into the running batch: every path
// in the run (one file, or several hard links to it) gets a record
// with the shared hashes.
func (s *scanState) recordRun(ctx context.Context, r hashResult) error {
	for _, rec := range r.run {
		s.resolve(rec.path)

		s.batch = append(s.batch, scanRec{
			size:    rec.size,
			mtime:   rec.mtime,
			head:    r.head,
			tail:    r.tail,
			content: r.content,
			path:    rec.path,
		})
	}

	return s.commitFullBatch(ctx)
}

// commitFullBatch commits the running batch once it holds
// updateBatchSize records. A batch that fails to commit is kept: the
// commit fails when the scan is interrupted, and syncScan then commits
// the batch itself.
func (s *scanState) commitFullBatch(ctx context.Context) error {
	if len(s.batch) < updateBatchSize {
		return nil
	}

	err := applyBatch(ctx, s.db, s.batch, nil, nil)
	if err != nil {
		return err
	}

	s.batch = s.batch[:0]

	return nil
}

// updatePhase writes the scan's tail under one progress display: the
// final partial batch of hashed records, and deletions for every record
// the scan did not verify (vanished files, plus paths that failed to
// stat or hash).
func (s *scanState) updatePhase(ctx context.Context) error {
	deletes := make([]string, 0, len(s.existing))
	for path := range s.existing {
		deletes = append(deletes, path)
	}

	// Sorted deletes keep the update phase deterministic.
	slices.Sort(deletes)

	s.st.removed = len(deletes)

	total := len(s.batch) + len(deletes)
	prog := newProgress(progressParams{
		label: scanPhase("update"), unit: scanUnit, total: int64(total),
		stalledOn: "db",
	})

	defer prog.finish()

	err := applyChanges(ctx, s.db, s.batch, nil, prog)
	if err != nil {
		return err
	}

	s.batch = nil

	return applyChanges(ctx, s.db, nil, deletes, prog)
}

// mtimeAfter reports whether mtime a is later than mtime b.
// Not a.After(b): time.Time wraps an mtime past year 292 billion; Unix() undoes it.
func mtimeAfter(a, b time.Time) bool {
	if a.Unix() != b.Unix() {
		return a.Unix() > b.Unix()
	}

	return a.Nanosecond() > b.Nanosecond()
}

// underAnyRoot reports whether path is any of the roots or lies under
// one of them.
func underAnyRoot(path string, roots []string) bool {
	for _, root := range roots {
		if underRoot(path, root) {
			return true
		}
	}

	return false
}

// underRoot reports whether path is root itself or lies under it. Both
// must be absolute and lexically clean.
func underRoot(path, root string) bool {
	if path == root {
		return true
	}

	prefix := root
	if !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}

	return strings.HasPrefix(path, prefix)
}

// deepestRoot returns the longest of the roots that path is or lies
// under, or "" when it lies under none.
func deepestRoot(path string, roots []string) string {
	deepest := ""

	for _, root := range roots {
		if underRoot(path, root) && len(root) > len(deepest) {
			deepest = root
		}
	}

	return deepest
}

// dirJob is one directory awaiting traversal by the walk workers. It
// carries its operand's filesystem device so -x can stop at
// filesystem boundaries.
type dirJob struct {
	path      string
	rootDev   uint64
	rootDevOK bool
}

// walkEvent is one walk result delivered to the walk phase: a regular
// file's statted record, a warning when fail is set, or a path the walk
// left out: when otherFSDir is set, a directory -x did not enter
// because it is on another filesystem than its operand, and when
// leftOut is set, a default exclude or a directory or file on a
// filesystem that is not persistent, with warn holding the line that
// says so.
type walkEvent struct {
	rec        fileRec
	warn       string
	fail       bool
	otherFSDir string
	leftOut    string
}

// startWalk seeds every root into the shared walk worker pool and
// returns the event stream: one record per regular file, one warning
// event per per-path error, and one event per path -x or rules kept the
// walk from entering or recording. A path that no longer exists when
// the walk reaches it is no error, and gets no event. The channel is
// closed when the walk completes, and also when ctx is cancelled —
// every goroutine in the pool abandons its blocking send in that case,
// so the consumer sees a truncated but properly terminated stream
// instead of a stalled one.
func startWalk(ctx context.Context, roots []string, oneFS bool,
	rules walkRules, workers int,
) <-chan walkEvent {
	jobs, subdirs, events := startWalkWorkers(ctx, workers, oneFS, rules)

	go func() {
		initial := make([]dirJob, 0, len(roots))
		for _, root := range roots {
			initial = append(initial, seedRoot(ctx, root, rules, events)...)
		}

		dispatchDirs(ctx, initial, jobs, subdirs)
	}()

	return events
}

// sendEvent delivers one walk event, abandoning the send when the scan
// is cancelled. Every walk goroutine reaches the consumer through this
// one channel, so this is where a cancelled walk unwinds rather than
// parking on a buffer nobody is draining.
func sendEvent(ctx context.Context, events chan<- walkEvent,
	ev walkEvent,
) {
	select {
	case events <- ev:
	case <-ctx.Done():
	}
}

// operandWarning returns the one-line warning for an operand the walk
// does not start from, naming the path and what it is, or "" for one it
// does: a regular file, or a directory not named .zfs, that is not
// inside a directory named .zfs, neither as given nor with its symlinks
// resolved. Symlinks are never followed, including as operands.
func operandWarning(root string, fi fs.FileInfo) string {
	// The walk never enters a .zfs directory, so it starts from nothing
	// inside one either, not even through a symlink. An operand whose
	// symlinks cannot be resolved is checked as given.
	resolved, err := filepath.EvalSymlinks(root)
	if err != nil {
		resolved = root
	}

	if slices.Contains(strings.Split(filepath.Dir(root), "/"), ".zfs") ||
		slices.Contains(strings.Split(filepath.Dir(resolved), "/"), ".zfs") {
		return fmt.Sprintf("walk %s: skipping operand inside a .zfs directory",
			root)
	}

	var kind string

	switch mode := fi.Mode(); {
	case mode.IsRegular():
		return ""
	case mode.IsDir():
		if filepath.Base(root) != ".zfs" {
			return ""
		}

		kind = ".zfs directory"
	case mode&fs.ModeSymlink != 0:
		kind = "symlink"
	case mode&fs.ModeSocket != 0:
		kind = "socket"
	case mode&fs.ModeNamedPipe != 0:
		kind = "FIFO"
	case mode&fs.ModeDevice != 0:
		kind = "device node"
	default:
		kind = "non-regular file"
	}

	return fmt.Sprintf("walk %s: skipping %s operand", root, kind)
}

// seedRoot turns one PATH operand into the walk's starting state: a
// regular-file operand is statted and emitted directly, unless it is
// one of the scan's own database files in rules, and a directory
// operand becomes an initial job. walkableRoots has already dropped
// every other operand. One that has changed into something else since
// is warned about and skipped here, and one removed since is skipped
// without a warning. Either is still a root, so the records stored
// beneath it, other than those under another root, are deleted as
// unverified.
func seedRoot(ctx context.Context, root string, rules walkRules,
	events chan<- walkEvent,
) []dirJob {
	fi, err := os.Lstat(root)
	if errors.Is(err, fs.ErrNotExist) {
		// Removed since the scan started: no error, and no files.
		return nil
	}

	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("walk %s: %v", root, err),
			fail: true,
		})

		return nil
	}

	warn := operandWarning(root, fi)
	if warn != "" {
		sendEvent(ctx, events, walkEvent{warn: warn, fail: true})

		return nil
	}

	if fi.IsDir() {
		dev, ok := deviceOfInfo(fi)

		return []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}
	}

	dev, ino := inodeOfInfo(fi)
	rec := fileRec{
		path:  root,
		size:  fi.Size(),
		mtime: fi.ModTime(),
		dev:   dev,
		ino:   ino,
	}

	if !isDatabaseFile(rec, rules) {
		sendEvent(ctx, events, walkEvent{rec: rec})
	}

	return nil
}

// startWalkWorkers starts the walk worker pool. Each worker processes
// one directory at a time, emitting an event per regular file and
// handing discovered subdirectories back to the dispatcher; events is
// closed once every worker has finished. A cancelled scan makes the
// workers drop the directories still queued rather than stop reading
// jobs, so the range always runs out and the pool always tears down.
func startWalkWorkers(ctx context.Context, workers int,
	oneFS bool, rules walkRules,
) (chan dirJob, chan []dirJob, chan walkEvent) {
	jobs := make(chan dirJob, workQueueDepth)
	subdirs := make(chan []dirJob, workers)
	events := make(chan walkEvent, workQueueDepth)

	var wg sync.WaitGroup

	for range workers {
		wg.Go(func() {
			for job := range jobs {
				if ctx.Err() != nil {
					continue
				}

				select {
				case subdirs <- walkOneDir(ctx, job, oneFS, rules, events):
				case <-ctx.Done():
				}
			}
		})
	}

	go func() {
		wg.Wait()
		close(events)
	}()

	return jobs, subdirs, events
}

// dispatchDirs feeds directory jobs to the walk workers, queueing
// newly discovered subdirectories (newest first, which keeps the
// frontier small) until every directory has been processed, then
// closes jobs. jobs is closed on every path out, cancellation
// included: the workers range over it, and a dispatcher that returned
// without closing would strand all of them.
func dispatchDirs(ctx context.Context, initial []dirJob,
	jobs chan<- dirJob, subdirs <-chan []dirJob,
) {
	go func() {
		defer close(jobs)

		queue := slices.Clone(initial)
		pending := len(queue)

		for pending > 0 {
			var (
				out  chan<- dirJob
				next dirJob
			)

			if len(queue) > 0 {
				out = jobs
				next = queue[len(queue)-1]
			}

			select {
			case out <- next:
				queue = queue[:len(queue)-1]
			case subs := <-subdirs:
				pending += len(subs) - 1
				queue = append(queue, subs...)
			case <-ctx.Done():
				return
			}
		}
	}()
}

// walkOneDir reads one directory, emitting an event per regular-file
// entry and a warning event per unreadable one, and returns the
// subdirectories to descend into. A directory or entry removed since
// the walk found it gets no event.
func walkOneDir(ctx context.Context, job dirJob, oneFS bool,
	rules walkRules, events chan<- walkEvent,
) []dirJob {
	entries, err := os.ReadDir(job.path)
	if errors.Is(err, fs.ErrNotExist) {
		// Removed since its parent was read: no error, and no files.
		return nil
	}

	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("walk %s: %v", job.path, err),
			fail: true,
		})

		return nil
	}

	var subs []dirJob

	for _, e := range entries {
		// A cancelled scan wants nothing more from this directory: stop
		// rather than lstat the rest of a large one.
		if ctx.Err() != nil {
			return nil
		}

		p := filepath.Join(job.path, e.Name())

		if e.IsDir() {
			sub, ok := subdirJob(ctx, p, e, job, oneFS, rules, events)
			if ok {
				subs = append(subs, sub)
			}

			continue
		}
		// Regular files only: skip symlinks, sockets, FIFOs, and
		// device nodes.
		if !e.Type().IsRegular() {
			continue
		}

		emitFile(ctx, p, e, rules, events)
	}

	return subs
}

// emitFile stats one regular directory entry and emits its record.
// The lstat happens here in the walk worker, while the directory's
// metadata is still hot; a path that fails to stat between the
// directory read and the lstat is warned about and skipped, unless it
// no longer exists, and one that has stopped being a regular file is
// skipped. A file on a filesystem in rules that is not persistent, as
// through a bind mount of one file, is not recorded: it gets an event
// naming it instead. One of the scan's own database files in rules gets
// no event at all.
func emitFile(ctx context.Context, p string, e fs.DirEntry,
	rules walkRules, events chan<- walkEvent,
) {
	info, err := e.Info()
	if errors.Is(err, fs.ErrNotExist) {
		// Removed since its directory was read: no error.
		return
	}

	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("stat %s: %v", p, err),
			fail: true,
		})

		return
	}

	if !info.Mode().IsRegular() {
		return
	}

	line := rules.nonPersistentLine(p, info)
	if line != "" {
		sendEvent(ctx, events, walkEvent{leftOut: p, warn: line})

		return
	}

	dev, ino := inodeOfInfo(info)
	rec := fileRec{
		path:  p,
		size:  info.Size(),
		mtime: info.ModTime(),
		dev:   dev,
		ino:   ino,
	}

	if !isDatabaseFile(rec, rules) {
		sendEvent(ctx, events, walkEvent{rec: rec})
	}
}

// subdirJob applies the descent rules to directory p: never enter
// .zfs (ZFS snapshot pseudo-dirs would list every file once per
// snapshot), never enter a directory on a filesystem in rules that is
// not persistent, nor a default exclude in rules, and with -x never
// enter a directory on a different filesystem than its operand. For the
// last three it sends an event that names the directory, so the scan
// keeps the records below it, except those below an operand at or
// inside it.
func subdirJob(ctx context.Context, p string, e fs.DirEntry,
	parent dirJob, oneFS bool, rules walkRules,
	events chan<- walkEvent,
) (dirJob, bool) {
	if e.Name() == ".zfs" {
		return dirJob{}, false
	}

	info, err := e.Info()
	if errors.Is(err, fs.ErrNotExist) {
		// Removed since its parent was read: no error, and no files.
		return dirJob{}, false
	}

	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("walk %s: %v", p, err),
			fail: true,
		})

		return dirJob{}, false
	}

	line := rules.nonPersistentLine(p, info)
	if line != "" {
		sendEvent(ctx, events, walkEvent{leftOut: p, warn: line})

		return dirJob{}, false
	}

	if isDefaultExclude(p, rules.excludes) {
		sendEvent(ctx, events, walkEvent{
			leftOut: p,
			warn:    fmt.Sprintf("scan: skipped %s (default exclude)", p),
		})

		return dirJob{}, false
	}

	job := dirJob{path: p, rootDev: parent.rootDev, rootDevOK: parent.rootDevOK}
	if !oneFS || !parent.rootDevOK {
		return job, true
	}

	if dev, ok := deviceOfInfo(info); ok && dev != parent.rootDev {
		sendEvent(ctx, events, walkEvent{otherFSDir: p})

		return dirJob{}, false
	}

	return job, true
}

// deviceOfInfo extracts the filesystem device ID from a FileInfo, when
// the platform exposes one.
func deviceOfInfo(fi fs.FileInfo) (uint64, bool) {
	st, ok := fi.Sys().(*syscall.Stat_t)
	if !ok {
		return 0, false
	}

	return statDev(st), true
}

// inodeOfInfo extracts the (device, inode) pair identifying a file's
// underlying inode; (0, 0) when the platform exposes none (such files
// are never merged as hard links).
func inodeOfInfo(fi fs.FileInfo) (uint64, uint64) {
	st, ok := fi.Sys().(*syscall.Stat_t)
	if !ok {
		return 0, 0
	}

	return statDev(st), st.Ino
}

// hashResult carries the hashes computed for one inode run (or the
// error that prevented computing them) from the pool's workers to the
// hash phase. Its run leaves out the paths found removed.
type hashResult struct {
	run     []fileRec
	head    [sha256.Size]byte
	tail    [sha256.Size]byte
	content [sha256.Size]byte
	err     error
}

// hashPool owns every goroutine of the hash worker pool: the feeder
// that queues the inode runs and the workers that read them. Both block
// on channel sends, so both are cancellable — the pool's context is
// derived from the scan's, and stop cancels it and waits the goroutines
// out. The consumer must call stop on every path out of the phase, not
// just the happy one.
type hashPool struct {
	results <-chan hashResult
	cancel  context.CancelFunc
	done    <-chan struct{}
}

// startHashPool starts the feeder and the workers over runs. Workers
// hash each run's first path with hash (all paths in a run are hard
// links to the same inode) and write one result per run.
func startHashPool(ctx context.Context, runs [][]fileRec, workers int,
	hash func(path string, size int64) (
		[sha256.Size]byte, [sha256.Size]byte, [sha256.Size]byte, error),
) *hashPool {
	ctx, cancel := context.WithCancel(ctx)

	jobs := make(chan []fileRec, workQueueDepth)
	results := make(chan hashResult, workQueueDepth)

	var wg sync.WaitGroup

	wg.Go(func() { feedHashJobs(ctx, runs, jobs) })

	for range workers {
		wg.Go(func() { hashWorker(ctx, jobs, results, hash) })
	}

	done := make(chan struct{})

	go func() {
		wg.Wait()
		close(done)
	}()

	return &hashPool{results: results, cancel: cancel, done: done}
}

// stop cancels the pool and blocks until every one of its goroutines
// has exited, draining results while it waits: a worker already parked
// on a send observes the cancellation only once a receiver frees it.
// Calling stop more than once is safe.
func (p *hashPool) stop() {
	p.cancel()

	for {
		select {
		case <-p.results:
		case <-p.done:
			return
		}
	}
}

// feedHashJobs queues every run for the workers, closing jobs on the
// way out — including when the scan is cancelled mid-queue, so that the
// workers' range over jobs always terminates.
func feedHashJobs(ctx context.Context, runs [][]fileRec,
	jobs chan<- []fileRec,
) {
	defer close(jobs)

	for _, run := range runs {
		select {
		case jobs <- run:
		case <-ctx.Done():
			return
		}
	}
}

// hashWorker hashes one inode run at a time with hash until jobs is
// closed or the scan is cancelled. A cancelled worker drops the runs
// still queued instead of stopping its reads of jobs: the range must
// run out for the pool to tear down, and reading a file nobody wants
// the hash of only delays that.
func hashWorker(ctx context.Context, jobs <-chan []fileRec,
	results chan<- hashResult,
	hash func(path string, size int64) (
		[sha256.Size]byte, [sha256.Size]byte, [sha256.Size]byte, error),
) {
	for run := range jobs {
		if ctx.Err() != nil {
			continue
		}

		head, tail, content, err := hash(run[0].path, run[0].size)

		// A path removed since the walk is dropped from its run. The
		// next path is a hard link to the same file, and is read instead.
		for errors.Is(err, fs.ErrNotExist) && len(run) > 1 {
			run = run[1:]
			head, tail, content, err = hash(run[0].path, run[0].size)
		}

		select {
		case results <- hashResult{
			run: run, head: head, tail: tail, content: content, err: err,
		}:
		case <-ctx.Done():
			return
		}
	}
}

// errNotRegular reports a path that is no longer a regular file when it
// is opened to be read.
var errNotRegular = errors.New("not a regular file")

// openRegularFile opens path for reading and returns it only if it is
// still a regular file, since the path may have been replaced after it
// was statted. O_NONBLOCK keeps the open of a FIFO or a device node
// from waiting forever, and fstat then rejects it. Reads of a regular
// file ignore O_NONBLOCK.
func openRegularFile(path string) (*os.File, error) {
	//nolint:gosec // hashing operator-supplied paths is the tool's purpose
	f, err := os.OpenFile(path, os.O_RDONLY|syscall.O_NONBLOCK, 0)
	if err != nil {
		return nil, err
	}

	fi, err := f.Stat()
	if err != nil {
		_ = f.Close()

		return nil, err
	}

	if !fi.Mode().IsRegular() {
		_ = f.Close()

		return nil, errNotRegular
	}

	return f, nil
}

// hashSignature computes the three hashes the hash phase records for a
// file: head and tail (see hashEnds) and content (see hashContent).
// With the file size they form its duplicate signature, and two files
// are duplicates only when all four agree. size is the value recorded
// when the file was statted; a zero-length file has constant hashes and
// is never opened. On an error every hash is left zero.
func hashSignature(path string, size int64) (
	[sha256.Size]byte, [sha256.Size]byte, [sha256.Size]byte, error,
) {
	var zero [sha256.Size]byte

	if size == 0 {
		empty := sha256.Sum256(nil)

		return empty, empty, empty, nil
	}

	f, err := openRegularFile(path)
	if err != nil {
		return zero, zero, zero, err
	}

	defer func() { _ = f.Close() }()

	head, tail, err := hashEnds(f, size)
	if err != nil {
		return zero, zero, zero, err
	}

	content, err := hashContent(f, size)
	if err != nil {
		return zero, zero, zero, err
	}

	return head, tail, content, nil
}

// hashEnds returns the SHA-256 of the first and last headTailWindow
// bytes of f. In a file no larger than headTailWindow, each window is
// the whole file. Reading fewer bytes than asked means the file shrank
// between the stat and the hash, and is an error.
func hashEnds(f *os.File, size int64) (
	[sha256.Size]byte, [sha256.Size]byte, error,
) {
	var zero [sha256.Size]byte

	buf := make([]byte, min(int64(headTailWindow), size))

	_, err := f.ReadAt(buf, 0)
	if err != nil {
		return zero, zero, err
	}

	head := sha256.Sum256(buf)

	_, err = f.ReadAt(buf, size-int64(len(buf)))
	if err != nil {
		return zero, zero, err
	}

	return head, sha256.Sum256(buf), nil
}

// hashContent returns the content hash of f: the SHA-256 of the whole
// file when it is smaller than wholeFileMax, or of sampled windows when
// it is that size or larger.
func hashContent(f *os.File, size int64) ([sha256.Size]byte, error) {
	if size >= int64(wholeFileMax) {
		return hashSamples(f, size)
	}

	return hashWhole(f, size)
}

// hashWhole returns the SHA-256 of the entire file. A SectionReader is
// used so the read is independent of the offset left by any end-window
// reads. Reading fewer than size bytes means the file shrank between
// the stat and the hash; that is an error rather than a hash of content
// that no longer matches the recorded size.
func hashWhole(f *os.File, size int64) ([sha256.Size]byte, error) {
	h := sha256.New()

	n, err := io.Copy(h, io.NewSectionReader(f, 0, size))
	if err != nil {
		return [sha256.Size]byte{}, err
	}

	if n != size {
		return [sha256.Size]byte{}, fmt.Errorf("read %d of %d bytes: %w",
			n, size, io.ErrUnexpectedEOF)
	}

	return [sha256.Size]byte(h.Sum(nil)), nil
}

// hashSamples feeds sampleWindow bytes at each gigabyte-aligned offset
// (0, sampleStride, 2*sampleStride, ... while inside the file), in
// order, into one hash, each window truncated at end of file. Two files
// of equal size agreeing on every sample are reported as duplicates
// without every byte being read. Because size is part of the signature,
// only files of one size are compared, so their sample boundaries
// always align.
func hashSamples(f *os.File, size int64) ([sha256.Size]byte, error) {
	h := sha256.New()
	buf := make([]byte, sampleWindow)

	for off := int64(0); off < size; off += int64(sampleStride) {
		n := min(int64(sampleWindow), size-off)

		_, err := f.ReadAt(buf[:n], off)
		if err != nil {
			return [sha256.Size]byte{}, err
		}

		h.Write(buf[:n])
	}

	return [sha256.Size]byte(h.Sum(nil)), nil
}