package main

import (
	"cmp"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"syscall"
)

// The duplicate ladder (see hashSignature and README "Duplicate
// detection"). A same-size candidate below headTailMin is hashed in
// full and compared directly; a larger one is separated first by the
// hashes of its end windows, then by a content hash that is exact below
// wholeFileMax and deliberately sampled at or above it. The hash phase
// reads only the end windows of a larger file; the content phase reads
// it for its content hash only once its size, head, and tail match
// another file's.

// headTailMin is the size threshold for the end-window gate. A file
// smaller than this is hashed in full directly, with no separate head
// and tail step: its head, tail, and content all carry the whole-file
// hash. A file this size or larger is separated first by its end
// windows.
const headTailMin = 10 * 1024 * 1024

// headTailWindow is the number of bytes hashed from each end of a file
// at or above headTailMin (the head and tail rungs). Because
// headTailMin is far larger than two windows, the head and tail windows
// never overlap.
const headTailWindow = 64 * 1024

// wholeFileMax is the size boundary between the two content rungs: a
// file strictly smaller than this is content-hashed in full; a file
// this size or larger is content-hashed by sampling.
const wholeFileMax = 50 * 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

// 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 int64
	dev   uint64
	ino   uint64
}

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

// runScan implements the scan subcommand: four sequential phases —
// walk (which stats each file as it is discovered), hash, update,
// content — that synchronize the persistent database with the
// filesystem state under the PATH operands. Only files whose size at
// least one other file shares are ever hashed: a size-unique file
// cannot be a duplicate. A file of headTailMin or more gets its content
// hash only when its size, head, and tail match another file's. Flag
// parsing and the at-least-one-operand check are done by cobra. Errors
// are returned rather than exiting, so that the deferred close — which
// checkpoints the SQLite WAL — always runs. Cancelling ctx unwinds the
// worker pools and aborts the scan with the context's error.
func runScan(ctx context.Context, roots []string, workers int,
	oneFS bool,
) error {
	if workers < 1 {
		workers = 1
	}

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

	dbPath := databasePath()

	db, err := openScanDatabase(ctx, dbPath)
	if err != nil {
		return err
	}

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

	st, err := syncScan(ctx, db, roots, workers, oneFS)
	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
}

// 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. 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) []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 {
		if !underAnyRoot(r, kept) {
			kept = append(kept, r)
		}
	}

	return kept
}

// scanStats summarizes one scan's database synchronization for the
// final stderr summary.
type scanStats struct {
	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
	sizes     []int64   // size census: every walked file, plus records outside the roots
	toHash    []fileRec // files whose size is shared: must be read
	sentinels []fileRec // new/changed size-unique files: recorded without hashes
	batch     []scanRec
	st        scanStats
}

// syncScan synchronizes the database with the filesystem under roots
// in four sequential phases: walk (enumerate and stat every file,
// building a complete size census), hash (read only the new or
// changed — or previously unhashed — files whose size at least one
// other file shares, committing results in batches as they arrive),
// update (record the size-unique files without reading them, and
// delete the records the scan no longer verifies), and content (fill
// in the content hash of every record of headTailMin or more whose
// size, head, and tail match another record's). Records outside the
// roots are never touched, except that the content phase fills in
// their content hash.
func syncScan(ctx context.Context, db *sql.DB, roots []string,
	workers int, oneFS bool,
) (scanStats, error) {
	roots = pruneRoots(roots)

	s := &scanState{db: db}

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

	changed, unhashed := s.walkPhase(startWalk(ctx, roots, oneFS, workers))

	// A cancelled walk stops early, so its size census covers only part
	// of the roots, and every file it never reached looks vanished to
	// the update phase. Defence in depth rather than the only barrier:
	// that phase would today fail on its first BeginTx with the same
	// cancelled context before deleting anything. But it is the barrier
	// that survives a later decision to let an interrupted scan commit
	// what it has, and it turns a confusing failure deep in the update
	// phase into a clean abort at the phase boundary.
	err = ctx.Err()
	if err != nil {
		return s.st, err
	}

	s.partition(changed, unhashed)

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

	err = s.updatePhase(ctx)
	if err != nil {
		return s.st, err
	}

	return s.st, s.contentPhase(ctx, workers)
}

// loadIndex indexes the database records under the scan roots for
// change detection and collects the sizes of every record outside
// them: out-of-scope records join the size census so a scanned file
// can be recognized as a possible duplicate of a tree scanned
// separately into the same database.
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.
	prog := newProgress("load", -1)
	defer prog.finish()

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

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

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

				return
			}

			s.sizes = append(s.sizes, size)
		})
}

// walkPhase drains the walk, appending every walked file's size to
// the census and resolving what it can immediately: an unchanged file
// whose record already has hashes needs nothing from the hash phase
// (the content phase may still fill in its content hash). It returns
// the new-or-changed files and the unchanged files whose records lack
// hashes; both remain candidates until the census decides whether
// their sizes are shared.
func (s *scanState) walkPhase(
	events <-chan walkEvent,
) ([]fileRec, []fileRec) {
	prog := newProgress("walk", -1)

	var changed, unhashed []fileRec

	for ev := range events {
		if ev.fail {
			s.st.skipped++

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

			continue
		}

		s.sizes = append(s.sizes, ev.rec.size)

		prog.increment()

		old, ok := s.existing[ev.rec.path]

		switch {
		case !ok || old.size != ev.rec.size || old.mtime < ev.rec.mtime:
			changed = append(changed, ev.rec)
		case old.hashed:
			delete(s.existing, ev.rec.path)

			s.st.unchanged++
		default:
			unhashed = append(unhashed, ev.rec)
		}
	}

	prog.finish()

	return changed, unhashed
}

// partition decides each candidate's disposition now that the size
// census is complete. A file whose size no other file shares cannot
// be a duplicate and is never read: a new or changed one is recorded
// without hashes, an unchanged unhashed one keeps its record. Every
// file with a shared size queues for the hash phase.
func (s *scanState) partition(changed, unhashed []fileRec) {
	slices.Sort(s.sizes)

	for _, rec := range changed {
		if s.sizeShared(rec.size) {
			s.toHash = append(s.toHash, rec)

			continue
		}

		s.sentinels = append(s.sentinels, rec)
		s.resolve(rec.path)
	}

	for _, rec := range unhashed {
		if s.sizeShared(rec.size) {
			s.toHash = append(s.toHash, rec)

			continue
		}

		delete(s.existing, rec.path)

		s.st.unchanged++
	}

	s.sizes = nil
}

// sizeShared reports whether at least two census entries have this
// size. Every candidate's own size is in the census exactly once, so
// a second entry means another file (or a record outside the scan
// roots) could share its content.
func (s *scanState) sizeShared(size int64) bool {
	i, found := slices.BinarySearch(s.sizes, size)

	return found && i+1 < len(s.sizes) && s.sizes[i+1] == size
}

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

// hashPhase hashes every queued file with hashSignature — the head and
// tail of a file of headTailMin or more, the whole file below that —
// 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
// skipped; stale records for its paths, if any, are deleted by the
// update phase.
func (s *scanState) hashPhase(ctx context.Context, workers int) error {
	runs := hashRuns(s.toHash)
	s.toHash = nil

	return s.readRuns(ctx, workers, "hash", runs, hashSignature, s.recordRun)
}

// readRuns reads runs with the worker pool, one read per inode run, in
// the order given, under a progress display named label. The workers
// compute each run's hashes with hash, and each result goes to record;
// a run that fails to read is warned about and counted as skipped
// instead. The total counts actual reads, so the bar shows a real ETA.
//
// 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) readRuns(ctx context.Context, workers int,
	label string, runs [][]fileRec,
	hash func(path string, size int64) (string, string, string, error),
	record func(ctx context.Context, r hashResult) error,
) error {
	pool := startHashPool(ctx, runs, workers, hash)
	defer pool.stop()

	prog := newProgress(label, int64(len(runs)))
	defer prog.finish()

	for range runs {
		var r hashResult

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

		prog.increment()

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

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

			continue
		}

		err := record(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.
func (s *scanState) commitFullBatch(ctx context.Context) error {
	if len(s.batch) < updateBatchSize {
		return nil
	}

	err := applyBatch(ctx, s.db, s.batch, nil, nil)
	s.batch = s.batch[:0]

	return err
}

// updatePhase writes the scan's tail under one progress display: the
// final partial batch of hashed records, a hash-less record for every
// size-unique new or changed file, 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(s.sentinels) + len(deletes)
	prog := newProgress("update", int64(total))

	defer prog.finish()

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

	s.batch = nil

	// Sentinel records are converted in batch-sized chunks rather than
	// materialized all at once; a first scan can have millions.
	for chunk := range slices.Chunk(s.sentinels, updateBatchSize) {
		recs := make([]scanRec, 0, len(chunk))
		for _, rec := range chunk {
			recs = append(recs, scanRec{
				size: rec.size, mtime: rec.mtime, path: rec.path,
			})
		}

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

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

// contentPhase fills in the content hash of every record of headTailMin
// or more that lacks one and whose size, head, and tail equal another
// record's, anywhere in the database: records from this scan and
// records stored by earlier scans, inside or outside the roots. Only
// such a file can still be a duplicate, so no other file of headTailMin
// or more is read beyond its end windows. The files are read with the
// hash phase's worker pool and their records written back in batches. A
// failed read is warned about and counted as skipped; the record keeps
// its empty content, so it is never grouped, and a later scan tries
// again.
func (s *scanState) contentPhase(ctx context.Context, workers int) error {
	toRead, recs, err := contentCandidates(ctx, s.db)
	if err != nil {
		return err
	}

	err = s.readRuns(ctx, workers, "content", hashRuns(toRead),
		hashContentOnly, func(ctx context.Context, r hashResult) error {
			// Every path in the run keeps its record's head and tail
			// and gains the one content hash read for the run.
			for _, f := range r.run {
				rec := recs[f.path]
				rec.content = r.content
				s.batch = append(s.batch, rec)
			}

			return s.commitFullBatch(ctx)
		})
	if err != nil {
		return err
	}

	return applyChanges(ctx, s.db, s.batch, nil, nil)
}

// contentCandidates returns the files the content phase reads, and the
// records of the files that passed the check, by path. Every record
// contentCandidatesSQL returns has its file checked with lstat: a file
// that is gone, is no longer a regular file, or has changed by the
// walk's rule keeps its record as it is and is not a duplicate. The
// files of a group that pass are read only if the group still has at
// least minGroupSize members, counting its records that already have a
// content hash, so a group whose other members are all stale costs no
// reads.
func contentCandidates(ctx context.Context,
	db *sql.DB,
) ([]fileRec, map[string]scanRec, error) {
	var (
		toRead []fileRec
		passed []fileRec // the current group's files that passed the check
		first  scanRec   // the current group's first record
		hashed int       // the current group's records with a content hash
	)

	recs := make(map[string]scanRec)

	// endGroup queues the current group's files that passed the check,
	// if the group still has at least minGroupSize members.
	endGroup := func() {
		if len(passed)+hashed >= minGroupSize {
			toRead = append(toRead, passed...)
		}

		passed = nil
	}

	err := loadContentCandidates(ctx, db, func(r scanRec, groupHashed int) {
		if r.size != first.size || r.head != first.head || r.tail != first.tail {
			endGroup()

			first, hashed = r, groupHashed
		}

		f, ok := unchangedFile(r)
		if ok {
			passed = append(passed, f)
			recs[r.path] = r
		}
	})
	if err != nil {
		return nil, nil, err
	}

	endGroup()

	return toRead, recs, nil
}

// unchangedFile lstats the file r names and returns it for reading if
// it is still the regular file r records: the same size, and an mtime
// no newer than recorded (the walk's change rule). Otherwise it
// reports false.
func unchangedFile(r scanRec) (fileRec, bool) {
	fi, err := os.Lstat(r.path)
	if err != nil || !fi.Mode().IsRegular() || fi.Size() != r.size ||
		fi.ModTime().Unix() > r.mtime {
		return fileRec{}, false
	}

	dev, ino := inodeOfInfo(fi)

	return fileRec{
		path: r.path, size: r.size, mtime: r.mtime, dev: dev, ino: ino,
	}, true
}

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

// 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, or a warning when fail is set.
type walkEvent struct {
	rec  fileRec
	warn string
	fail bool
}

// 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. 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,
	workers int,
) <-chan walkEvent {
	jobs, subdirs, events := startWalkWorkers(ctx, workers, oneFS)

	go func() {
		initial := make([]dirJob, 0, len(roots))
		for _, root := range roots {
			initial = append(initial, seedRoot(ctx, root, 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():
	}
}

// seedRoot turns one PATH operand into the walk's starting state: a
// regular-file operand is statted and emitted directly, a directory
// operand becomes an initial job, and a symlink or other non-regular
// operand yields nothing (symlinks are never followed, including as
// operands).
func seedRoot(ctx context.Context, root string,
	events chan<- walkEvent,
) []dirJob {
	fi, err := os.Lstat(root)
	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("walk %s: %v", root, err),
			fail: true,
		})

		return nil
	}

	switch {
	case fi.IsDir():
		if filepath.Base(root) == ".zfs" {
			return nil
		}

		dev, ok := deviceOfInfo(fi)

		return []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}
	case fi.Mode().IsRegular():
		dev, ino := inodeOfInfo(fi)

		sendEvent(ctx, events, walkEvent{rec: fileRec{
			path:  root,
			size:  fi.Size(),
			mtime: fi.ModTime().Unix(),
			dev:   dev,
			ino:   ino,
		}})

		return nil
	default:
		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,
) (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, 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.
func walkOneDir(ctx context.Context, job dirJob, oneFS bool,
	events chan<- walkEvent,
) []dirJob {
	entries, err := os.ReadDir(job.path)
	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 {
		p := filepath.Join(job.path, e.Name())

		if e.IsDir() {
			if sub, ok := subdirJob(ctx, p, e, job, oneFS, events); 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, 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 (or stops being a
// regular file) between the directory read and the lstat is warned
// about and skipped.
func emitFile(ctx context.Context, p string, e fs.DirEntry,
	events chan<- walkEvent,
) {
	info, err := e.Info()
	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("stat %s: %v", p, err),
			fail: true,
		})

		return
	}

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

	dev, ino := inodeOfInfo(info)

	sendEvent(ctx, events, walkEvent{rec: fileRec{
		path:  p,
		size:  info.Size(),
		mtime: info.ModTime().Unix(),
		dev:   dev,
		ino:   ino,
	}})
}

// subdirJob applies the descent rules to directory p: never enter
// .zfs (ZFS snapshot pseudo-dirs would list every file once per
// snapshot), and with -x never enter a directory on a different
// filesystem than its operand.
func subdirJob(ctx context.Context, p string, e fs.DirEntry,
	parent dirJob, oneFS bool, events chan<- walkEvent,
) (dirJob, bool) {
	if e.Name() == ".zfs" {
		return dirJob{}, false
	}

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

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

		return dirJob{}, false
	}

	if dev, ok := deviceOfInfo(info); ok && dev != parent.rootDev {
		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
// phase that started the pool: head, tail, and content from
// hashSignature, content alone from hashContentOnly.
type hashResult struct {
	run     []fileRec
	head    string
	tail    string
	content string
	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) (string, string, string, 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) (string, string, string, error),
) {
	for run := range jobs {
		if ctx.Err() != nil {
			continue
		}

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

// emptyHash is the lowercase-hex SHA-256 of the empty input: the head,
// tail, and content hash of every zero-length file.
const emptyHash = "e3b0c44298fc1c149afbf4c8996fb924" +
	"27ae41e4649b934ca495991b7852b855"

// hashSignature computes the hashes the hash phase records for a file
// whose size is shared; with the file size they form its duplicate
// signature. A file below headTailMin is hashed in full and its
// whole-file SHA-256 is returned as head, tail, and content alike —
// that range takes no separate end-window step. For a file at or above
// headTailMin only the head and tail are computed, the SHA-256 of its
// first and last headTailWindow bytes, and content is returned empty:
// the content phase computes it with hashContentOnly once the file's
// size, head, and tail match another file's. Two files are duplicates
// only when all four agree; any mismatch means not a duplicate. size
// is the value recorded when the file was statted; a zero-length file
// has constant hashes and is never opened.
func hashSignature(path string, size int64) (string, string, string, error) {
	if size == 0 {
		return emptyHash, emptyHash, emptyHash, nil
	}

	//nolint:gosec // hashing operator-supplied paths is the tool's purpose
	f, err := os.Open(path)
	if err != nil {
		return "", "", "", err
	}

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

	// Below the threshold the whole file is hashed directly, with no
	// end-window step: head and tail both carry the whole-file hash.
	if size < int64(headTailMin) {
		content, err := hashWhole(f, size)
		if err != nil {
			return "", "", "", err
		}

		return content, content, content, nil
	}

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

	return head, tail, "", nil
}

// hashContentOnly returns the content hash of the file at path, which
// is at least headTailMin bytes: the content phase's read. head and
// tail are returned empty, because the content phase keeps the ones its
// records already hold.
func hashContentOnly(path string, size int64) (string, string, string, error) {
	//nolint:gosec // hashing operator-supplied paths is the tool's purpose
	f, err := os.Open(path)
	if err != nil {
		return "", "", "", err
	}

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

	content, err := hashContent(f, size)

	return "", "", content, err
}

// hashEnds returns the SHA-256 of the first and last headTailWindow
// bytes of f. It is called only for files at least headTailMin, which
// is far larger than two windows, so the windows never overlap and both
// reads are always full.
func hashEnds(f *os.File, size int64) (string, string, error) {
	buf := make([]byte, headTailWindow)

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

	h := sha256.Sum256(buf)
	head := hex.EncodeToString(h[:])

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

	t := sha256.Sum256(buf)

	return head, hex.EncodeToString(t[:]), nil
}

// hashContent returns the content-rung 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) (string, 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) (string, error) {
	h := sha256.New()

	n, err := io.Copy(h, io.NewSectionReader(f, 0, size))
	if err != nil {
		return "", err
	}

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

	return hex.EncodeToString(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. This is
// the probabilistic large-file rung: 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, files of different sizes
// never reach this comparison, so the sample boundaries always align.
func hashSamples(f *os.File, size int64) (string, 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 "", err
		}

		h.Write(buf[:n])
	}

	return hex.EncodeToString(h.Sum(nil)), nil
}