package datastore

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"go.uber.org/zap"

	"github.com/pingcap/failpoint"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/metrics"
)

// Batched recursive directory delete, per
// docs/design/recursive-delete-batched-design.md. This is the default and
// only recursive-delete behavior: small trees are deleted in one atomic
// transaction (fast path), large trees in bounded per-batch transactions
// (post-order sweep); success means the root dentry is gone.
//
// Invariants (see the design doc §4.2):
//
//	I1  a directory dentry is deleted only in a transaction that verified it
//	    childless, with a post-DELETE re-check in the same transaction;
//	I2  the root dentry is deleted last — while it exists it pins the path
//	    against recreation (idx_path uniqueness), and the operation carries
//	    the captured root node_id as a generation fence: every sweep
//	    transaction and the final root delete re-validate that exact
//	    identity, and a same-path replacement tree is never traversed or
//	    deleted;
//	I3  orphan/GC decisions derive only from rows deleted in the same
//	    transaction;
//	I4  a directory that was non-empty at first visit is re-evaluated for
//	    unlinking after it drains (post-order lift);
//	I5  a drain batch that deleted zero rows breaks out — batch fullness is
//	    never a progress signal.

// errDirDeleteRaced marks a batch/lift whose post-DELETE re-check found new
// children: the transaction must roll back and the directory be revisited.
var errDirDeleteRaced = errors.New("directory delete raced with concurrent create")

// errRootReplaced marks a transaction whose root-generation fence failed: the
// original root dentry is gone or a different node now occupies the path. The
// recursive delete must stop immediately — it must never traverse or delete a
// replacement tree recreated at the same path (design doc §2.5/§4.2 I2).
var errRootReplaced = errors.New("recursive delete root replaced by a new generation")

// DeleteDirRecursiveSummary reports the outcome of a recursive delete.
type DeleteDirRecursiveSummary struct {
	NodesDeleted    int64
	OrphansEnqueued int64
	TxCount         int64
}

// DeleteDirRecursive deletes dirPath and its whole subtree.
func (s *Store) DeleteDirRecursive(ctx context.Context, dirPath string) (out *DeleteDirRecursiveSummary, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "delete_dir_recursive", start, &err)

	return s.deleteDirRecursiveBatched(ctx, dirPath)
}

const (
	// deleteDirEstimatePageSize is the name-paged enumeration page used by the
	// subtree estimate walk.
	deleteDirEstimatePageSize = 1000
)

// Tunables are package-level variables so tests can shrink them.
var (
	// deleteDirEstimateMaxDirs bounds the counting walk used to derive the
	// transaction budget. Deliberately small: the budget only needs to be
	// sufficient, not exact — maxDuration is the hard backstop. A variable so
	// tests can prove the walk stops at the cap even on ultra-wide parents.
	deleteDirEstimateMaxDirs = int64(10_000)
	// deleteDirSweepBatchSize bounds the rows touched by one sweep
	// transaction.
	deleteDirSweepBatchSize = deleteFileIDBatchSize
	// deleteDirMaxDuration bounds one recursive delete call; the sweep is
	// resumable, so a bounded caller may simply retry.
	deleteDirMaxDuration = 5 * time.Minute
	// deleteDirRootAttempts bounds root-dentry retries against racing
	// concurrent writers before ErrDirectoryNotEmpty is returned.
	deleteDirRootAttempts = 5
	// deleteDirBatchRetries bounds deadlock/lock-wait retries per batch
	// transaction.
	deleteDirBatchRetries = 3
	// deleteDirMaxBatchesCap caps the derived transaction budget.
	deleteDirMaxBatchesCap = int64(1_000_000)
)

// DeleteDirMaxDuration returns the current per-call wall-clock budget for
// recursive directory deletes. Exported so server-side durability lead times
// can be kept strictly larger than the delete budget (a crash-recovery marker
// must never become due while a delete could still be in flight).
func DeleteDirMaxDuration() time.Duration { return deleteDirMaxDuration }

func isLockWaitTimeout(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(err.Error(), "Lock wait timeout exceeded")
}

func isRetryableTxnError(err error) bool {
	return isDeadlock(err) || isLockWaitTimeout(err)
}

// deleteChildRow is one file_nodes row selected during a drain batch.
type deleteChildRow struct {
	nodeID string
	path   string
	fileID string
	isDir  bool
}

// dirDeleteQueue is a LIFO stack of directory paths with set dedup, giving a
// naturally post-order-ish traversal (deep directories are processed before
// their ancestors).
type dirDeleteQueue struct {
	stack []string
	in    map[string]bool
}

func newDirDeleteQueue() *dirDeleteQueue { return &dirDeleteQueue{in: make(map[string]bool)} }

func (q *dirDeleteQueue) push(p string) {
	if p == "" || q.in[p] {
		return
	}
	q.in[p] = true
	q.stack = append(q.stack, p)
}

func (q *dirDeleteQueue) pop() (string, bool) {
	if len(q.stack) == 0 {
		return "", false
	}
	p := q.stack[len(q.stack)-1]
	q.stack = q.stack[:len(q.stack)-1]
	delete(q.in, p)
	return p, true
}

func (q *dirDeleteQueue) empty() bool { return len(q.stack) == 0 }

// AbortUploadsByTargetPrefix best-effort aborts active uploads whose target
// path lies under dirPath, so their finalize cannot materialize nodes inside
// a tree being deleted. uploads has no target_path prefix index, so this is a
// bounded scan of the tenant's active uploads. Note this only marks the
// metadata rows ABORTED (which blocks finalize); the S3 multipart sessions
// themselves are left to the usual lifecycle/abort sweeper, as the store
// layer has no S3 client.
func (s *Store) AbortUploadsByTargetPrefix(ctx context.Context, dirPath string) (n int64, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "abort_uploads_by_target_prefix", start, &err)

	where, args := pathPrefixPredicate("target_path", dirPath)
	res, err := s.db.ExecContext(ctx, `UPDATE uploads SET status = 'ABORTED', updated_at = ?
		WHERE `+s.scope.And(`target_kind = 'file' AND status IN ('INITIATED', 'UPLOADING') AND `+where),
		append([]any{time.Now().UTC()}, s.scope.Args(args...)...)...)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// deleteDirRecursiveBatched implements the flag-on path: fast path for small
// trees, post-order batched sweep otherwise.
func (s *Store) deleteDirRecursiveBatched(ctx context.Context, dirPath string) (*DeleteDirRecursiveSummary, error) {
	if dirPath == "/" {
		return nil, ErrInvalidRootDentry
	}

	// Step 0: best-effort abort of in-flight uploads into the subtree.
	if n, err := s.AbortUploadsByTargetPrefix(ctx, dirPath); err != nil {
		logger.Warn(ctx, "delete_dir_recursive_abort_uploads_failed", zap.String("path", dirPath), zap.Error(err))
	} else if n > 0 {
		logger.Info(ctx, "delete_dir_recursive_abort_uploads", zap.String("path", dirPath), zap.Int64("aborted", n))
	}

	// The root dentry must exist, and its node_id is the generation fence for
	// the whole operation (I2): every sweep transaction and the final root
	// delete re-validate this exact identity, so a same-path replacement tree
	// is never traversed or deleted.
	rootNodeID, err := s.lookupDirNodeID(ctx, dirPath)
	if err != nil {
		return nil, err
	}

	// Step 1: fast path — the whole subtree in one transaction.
	summary, done, err := s.deleteDirFastPath(ctx, dirPath, rootNodeID)
	if err != nil {
		if errors.Is(err, errRootReplaced) {
			// The original root generation disappeared before this
			// transaction started: the goal state (original root gone) was
			// reached by a concurrent actor, and the replacement is not ours
			// to delete.
			logger.Info(ctx, "delete_dir_recursive_root_replaced",
				zap.String("path", dirPath), zap.String("mode", "fast_path"))
			return &DeleteDirRecursiveSummary{}, nil
		}
		return nil, err
	}
	if done {
		logger.Info(ctx, "delete_dir_recursive_done",
			zap.String("path", dirPath), zap.String("mode", "fast_path"),
			zap.Int64("nodes_deleted", summary.NodesDeleted),
			zap.Int64("orphans_enqueued", summary.OrphansEnqueued),
			zap.Int64("tx_count", summary.TxCount))
		return summary, nil
	}
	summary = &DeleteDirRecursiveSummary{}

	// The wall-clock budget is established BEFORE the estimate walk so the
	// estimate itself is bounded by the operation deadline, not just by the
	// directory cap. Budget derived from the (walkable) subtree estimate:
	// deep chains cost ~2 transactions per directory, wide dirs ~1 per batch
	// of files.
	deadline := time.Now().Add(deleteDirMaxDuration)
	estCtx, cancelEst := context.WithDeadline(ctx, deadline)
	files, dirs := s.estimateSubtreeSize(estCtx, dirPath)
	cancelEst()
	maxBatches := int64(32) + 8*(files/int64(deleteDirSweepBatchSize)+dirs+1)
	if maxBatches > deleteDirMaxBatchesCap {
		maxBatches = deleteDirMaxBatchesCap
	}

	sw := &dirSweeper{
		s:          s,
		root:       dirPath,
		rootNodeID: rootNodeID,
		queue:      newDirDeleteQueue(),
		maxBatches: maxBatches,
		deadline:   deadline,
		summary:    summary,
	}

	// Steps 2+3: sweep, then delete the root last; a non-empty root means
	// concurrent writers raced us — re-seed and go again, bounded.
	for attempt := 0; ; attempt++ {
		sw.queue.push(dirPath)
		if err := sw.sweep(ctx); err != nil {
			if errors.Is(err, errRootReplaced) {
				// The fence failed inside a sweep transaction: stop now, never
				// re-seed against the replacement tree (the original root is
				// gone — the delete's goal state is reached).
				logger.Info(ctx, "delete_dir_recursive_root_replaced",
					zap.String("path", dirPath), zap.String("mode", "sweep"))
				return sw.summary, nil
			}
			return sw.summary, err
		}
		failpoint.InjectCall("deleteDirBeforeRootDelete", dirPath, attempt)
		err := s.deleteEmptyDirFenced(ctx, dirPath, rootNodeID)
		if err == nil || errors.Is(err, ErrNotFound) {
			// nil: the fenced root row was deleted.
			// ErrNotFound: the original root generation is gone (concurrent
			// delete, or a replacement now occupies the path) — the goal
			// state is reached either way and the replacement is untouched.
			logger.Info(ctx, "delete_dir_recursive_done",
				zap.String("path", dirPath), zap.String("mode", "sweep"),
				zap.Int64("nodes_deleted", sw.summary.NodesDeleted),
				zap.Int64("orphans_enqueued", sw.summary.OrphansEnqueued),
				zap.Int64("tx_count", sw.summary.TxCount))
			return sw.summary, nil
		}
		if !errors.Is(err, ErrDirectoryNotEmpty) {
			return sw.summary, err
		}
		// ErrDirectoryNotEmpty with the fence intact means OUR root still has
		// children (concurrent creates): re-seed the sweep. If the root row no
		// longer carries our node_id, deleteEmptyDirFenced returns ErrNotFound
		// above and we never reach this branch.
		if attempt+1 >= deleteDirRootAttempts {
			return sw.summary, fmt.Errorf("%w: %s", ErrDirectoryNotEmpty, dirPath)
		}
	}
}

// lookupDirNodeID returns the root directory's node_id, or ErrNotFound.
func (s *Store) lookupDirNodeID(ctx context.Context, dirPath string) (string, error) {
	var nodeID string
	err := s.db.QueryRowContext(ctx, `SELECT node_id FROM file_nodes WHERE `+
		s.scope.And(`path_hash = ? AND path = ? AND is_directory = 1`),
		s.scope.Args(fileNodePathHash(dirPath), dirPath)...).Scan(&nodeID)
	if errors.Is(err, sql.ErrNoRows) {
		return "", ErrNotFound
	}
	if err != nil {
		return "", err
	}
	return nodeID, nil
}

// estimateSubtreeSize counts files and directories under dirPath with an
// indexed per-directory COUNT walk, bounding the walk at
// deleteDirEstimateMaxDirs directories. The cap bounds BOTH the directory
// count returned and the child-directory paths enqueued, so an ultra-wide
// parent cannot force unbounded paging queries or unbounded queue memory
// (the count query is a single aggregate; the per-child name-page walk is
// what must stop at the cap). The estimate only feeds the transaction
// budget, so it errs on the generous side: any query failure or deadline
// expiry returns the walk cap (budget then saturates at
// deleteDirMaxBatchesCap and maxDuration remains the real backstop).
func (s *Store) estimateSubtreeSize(ctx context.Context, dirPath string) (files, dirs int64) {
	generous := func() (int64, int64) { return files, deleteDirEstimateMaxDirs }
	queue := []string{dirPath}
	for len(queue) > 0 && dirs < deleteDirEstimateMaxDirs {
		if err := ctx.Err(); err != nil {
			return generous()
		}
		parent := queue[len(queue)-1]
		queue = queue[:len(queue)-1]
		var total, dirCount int64
		err := s.db.QueryRowContext(ctx, `SELECT COUNT(*), COALESCE(SUM(is_directory), 0) FROM file_nodes
			WHERE `+s.scope.And(`parent_path_hash = ? AND parent_path = ?`),
			s.scope.Args(fileNodePathHash(parent), parent)...).Scan(&total, &dirCount)
		if err != nil {
			return generous()
		}
		files += total - dirCount
		dirs += dirCount
		// Once the directory budget is exhausted, stop entirely: no further
		// child enumeration and no more queries (a wide parent whose COUNT
		// alone crossed the cap must not page through its children).
		if dirs >= deleteDirEstimateMaxDirs {
			dirs = deleteDirEstimateMaxDirs
			break
		}
		// Enqueue child directories by name page, bounded by the remaining
		// directory budget so a single ultra-wide parent cannot queue
		// unbounded paths or issue unlimited pages.
		remaining := deleteDirEstimateMaxDirs - dirs
		offset := ""
		for remaining > 0 {
			if err := ctx.Err(); err != nil {
				return generous()
			}
			limit := remaining
			if limit > deleteDirEstimatePageSize {
				limit = deleteDirEstimatePageSize
			}
			rows, err := s.db.QueryContext(ctx, `SELECT path FROM file_nodes
				WHERE `+s.scope.And(`parent_path_hash = ? AND parent_path = ? AND is_directory = 1 AND name > ?`)+`
				ORDER BY name LIMIT ?`,
				s.scope.Args(fileNodePathHash(parent), parent, offset, limit)...)
			if err != nil {
				return generous()
			}
			n := 0
			for rows.Next() {
				var p string
				if err := rows.Scan(&p); err != nil {
					_ = rows.Close()
					return generous()
				}
				queue = append(queue, p)
				offset = baseName(p)
				n++
				remaining--
			}
			_ = rows.Close()
			if n < int(limit) {
				break
			}
		}
	}
	return files, dirs
}

// --- fast path ---

// deleteDirFastPath attempts the whole subtree in a single transaction when
// it enumerates to at most deleteDirSweepBatchSize nodes. rootNodeID is the
// generation fence captured by the caller: the transaction first verifies the
// root row still carries that exact node_id, so a same-path replacement tree
// is never traversed. done=false means the caller should continue with the
// batched sweep.
func (s *Store) deleteDirFastPath(ctx context.Context, dirPath, rootNodeID string) (summary *DeleteDirRecursiveSummary, done bool, err error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, false, err
	}
	defer func() { _ = tx.Rollback() }()

	// Root-generation fence: lock the exact root row and verify identity
	// before touching anything under the path (I2).
	ok, err := s.rootMatchesTx(ctx, tx, dirPath, rootNodeID)
	if err != nil {
		return nil, false, err
	}
	if !ok {
		return nil, false, errRootReplaced
	}

	// Bounded BFS enumeration, children locked FOR UPDATE as we go.
	var nodes []deleteChildRow
	queue := []string{dirPath}
	for len(queue) > 0 {
		parent := queue[len(queue)-1]
		queue = queue[:len(queue)-1]
		remaining := deleteDirSweepBatchSize - len(nodes) + 1
		rows, err := s.selectChildrenForUpdateTx(ctx, tx, parent, remaining)
		if err != nil {
			return nil, false, err
		}
		for _, r := range rows {
			if r.isDir {
				queue = append(queue, r.path)
			}
		}
		nodes = append(nodes, rows...)
		if len(nodes) > deleteDirSweepBatchSize {
			return nil, false, nil // overflow → sweep
		}
	}

	// Children before parents: directory paths end in '/', so a deeper path
	// always sorts after its ancestors.
	sort.Slice(nodes, func(i, j int) bool { return nodes[i].path > nodes[j].path })

	var fileIDs []string
	for start := 0; start < len(nodes); start += deleteFileIDBatchSize {
		end := start + deleteFileIDBatchSize
		if end > len(nodes) {
			end = len(nodes)
		}
		batch := nodes[start:end]
		ids := make([]any, 0, len(batch))
		for _, n := range batch {
			ids = append(ids, n.nodeID)
			if !n.isDir && n.fileID != "" {
				fileIDs = append(fileIDs, n.fileID)
			}
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+
			s.scope.And(`node_id IN (`+questionPlaceholders(len(batch))+`)`),
			s.scope.Args(ids...)...); err != nil {
			return nil, false, err
		}
	}

	orphans, err := s.processOrphansTx(ctx, tx, fileIDs)
	if err != nil {
		return nil, false, err
	}

	// Root last (I2), with a post-DELETE re-check (I1). The DELETE carries the
	// node_id so it can never remove a replacement generation.
	if _, err := tx.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+
		s.scope.And(`path_hash = ? AND path = ? AND node_id = ? AND is_directory = 1`),
		s.scope.Args(fileNodePathHash(dirPath), dirPath, rootNodeID)...); err != nil {
		return nil, false, err
	}
	hasChildren, err := s.dirHasChildrenTx(ctx, tx, dirPath)
	if err != nil {
		return nil, false, err
	}
	if hasChildren {
		return nil, false, nil // raced with a concurrent create → sweep
	}

	if err := tx.Commit(); err != nil {
		return nil, false, err
	}
	return &DeleteDirRecursiveSummary{
		NodesDeleted:    int64(len(nodes)) + 1,
		OrphansEnqueued: orphans,
		TxCount:         1,
	}, true, nil
}

// --- batched sweep ---

type dirSweeper struct {
	s          *Store
	root       string
	rootNodeID string
	queue      *dirDeleteQueue
	maxBatches int64
	deadline   time.Time
	summary    *DeleteDirRecursiveSummary
}

// rootMatchesTx locks the root dentry in the current transaction and reports
// whether it still carries the captured generation. It is the fence for I2:
// every sweep transaction re-validates the exact root identity, so a
// concurrent delete+recreate of the path aborts the sweep instead of letting
// it traverse (and delete) the replacement tree. Lock order is always root
// row first, then the batch/lift rows — matching EnsureParentDirsTx's
// ancestor-first (root-first) backfill order, so a racing create blocks
// briefly on the root lock rather than deadlocking.
func (s *Store) rootMatchesTx(ctx context.Context, tx *sql.Tx, dirPath, rootNodeID string) (bool, error) {
	// Lock the root row by PRIMARY KEY (node_id), never through the secondary
	// path_hash index: a point lock on the PK touches only the clustered
	// record, while a locking read on idx_path can take next-key/gap locks
	// that inadvertently serialize unrelated range scans (e.g. a concurrent
	// rmdir's child probe on idx_parent). The locked row is then verified to
	// still be a directory at dirPath carrying the captured node_id.
	var got, gotPath string
	var isDir int
	err := tx.QueryRowContext(ctx, `SELECT node_id, path, is_directory FROM file_nodes WHERE `+
		s.scope.And(`node_id = ?`)+` FOR UPDATE`,
		s.scope.Args(rootNodeID)...).Scan(&got, &gotPath, &isDir)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return got == rootNodeID && gotPath == dirPath && isDir != 0, nil
}

func (sw *dirSweeper) checkBudget(ctx context.Context) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	if sw.summary.TxCount >= sw.maxBatches || time.Now().After(sw.deadline) {
		return fmt.Errorf("%w: %s (tx=%d)", ErrDeleteIncomplete, sw.root, sw.summary.TxCount)
	}
	return nil
}

// sweep drains the work queue in bounded per-batch transactions.
func (sw *dirSweeper) sweep(ctx context.Context) error {
	for !sw.queue.empty() {
		parent, _ := sw.queue.pop()

		// Drain this directory in batches.
		raced := false
		for {
			if err := sw.checkBudget(ctx); err != nil {
				return err
			}
			failpoint.InjectCall("drainBatchBeforeSelect", parent)
			out, err := sw.runDrainBatch(ctx, parent)
			if errors.Is(err, errDirDeleteRaced) {
				// Post-DELETE re-check failed: revisit the raced subdir and
				// this parent promptly (I1); skip the lift this round.
				sw.queue.push(parent)
				sw.queue.push(out.racedDir)
				raced = true
				break
			}
			if err != nil {
				return err
			}
			sw.summary.NodesDeleted += out.deleted
			sw.summary.OrphansEnqueued += out.orphans
			for _, sub := range out.subdirs {
				sw.queue.push(sub)
			}
			if out.deleted == 0 {
				break // I5: nothing deletable at this level — descend first
			}
			if out.rows < int64(deleteDirSweepBatchSize) {
				break // level drained
			}
		}

		// Post-order lift (I4): delete the drained directory's own dentry.
		// The root is owned by the caller's DeleteEmptyDir step.
		if raced || parent == sw.root {
			continue
		}
		if err := sw.checkBudget(ctx); err != nil {
			return err
		}
		lifted, err := sw.runLift(ctx, parent)
		if err != nil {
			return err
		}
		if lifted {
			sw.summary.NodesDeleted++
			sw.queue.push(parentPath(parent))
		}
	}
	return nil
}

type drainOutcome struct {
	rows     int64
	deleted  int64
	orphans  int64
	fileIDs  []string
	subdirs  []string
	racedDir string
}

// runDrainBatch executes one drain batch transaction with bounded retries on
// deadlock/lock-wait.
func (sw *dirSweeper) runDrainBatch(ctx context.Context, parent string) (*drainOutcome, error) {
	var out *drainOutcome
	err := sw.withBatchRetry(ctx, "delete_dir_recursive_batch", func(tx *sql.Tx) error {
		ok, err := sw.s.rootMatchesTx(ctx, tx, sw.root, sw.rootNodeID)
		if err != nil {
			return err
		}
		if !ok {
			return errRootReplaced
		}
		out, err = sw.s.drainChildrenBatchTx(ctx, tx, parent)
		return err
	})
	if err != nil {
		if out == nil {
			out = &drainOutcome{}
		}
		return out, err
	}
	sw.summary.TxCount++
	return out, nil
}

// runLift executes one post-order lift transaction. lifted=true means the
// dentry was deleted and the parent should be revisited. A non-empty or
// already-gone directory is a committed no-op (idempotent).
func (sw *dirSweeper) runLift(ctx context.Context, dirPath string) (lifted bool, err error) {
	err = sw.withBatchRetry(ctx, "delete_dir_recursive_lift", func(tx *sql.Tx) error {
		lifted = false
		ok, err := sw.s.rootMatchesTx(ctx, tx, sw.root, sw.rootNodeID)
		if err != nil {
			return err
		}
		if !ok {
			return errRootReplaced
		}
		failpoint.InjectCall("liftBeforePathLookup", dirPath)
		var nodeID string
		err = tx.QueryRowContext(ctx, `SELECT node_id FROM file_nodes WHERE `+
			sw.s.scope.And(`path_hash = ? AND path = ? AND is_directory = 1`)+` FOR UPDATE`,
			sw.s.scope.Args(fileNodePathHash(dirPath), dirPath)...).Scan(&nodeID)
		if errors.Is(err, sql.ErrNoRows) {
			return nil // already gone (concurrent rmdir): idempotent success
		}
		if err != nil {
			return err
		}
		has, err := sw.s.dirHasChildrenTx(ctx, tx, dirPath)
		if err != nil {
			return err
		}
		if has {
			return nil // legit children queued below, or racing creates
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+
			sw.s.scope.And(`node_id = ?`), sw.s.scope.Args(nodeID)...); err != nil {
			return err
		}
		still, err := sw.s.dirHasChildrenTx(ctx, tx, dirPath)
		if err != nil {
			return err
		}
		if still {
			return errDirDeleteRaced // rollback; step-3 re-seed rediscovers it
		}
		lifted = true
		return nil
	})
	if errors.Is(err, errDirDeleteRaced) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	sw.summary.TxCount++
	return lifted, nil
}

// withBatchRetry runs fn in a transaction, retrying deadlock/lock-wait
// failures with backoff. errDirDeleteRaced and other errors propagate without
// retry.
func (sw *dirSweeper) withBatchRetry(ctx context.Context, op string, fn func(tx *sql.Tx) error) (err error) {
	start := time.Now()
	defer func() {
		result := "ok"
		if err != nil {
			result = "error"
			if errors.Is(err, errDirDeleteRaced) {
				result = "raced"
			} else if errors.Is(err, errRootReplaced) {
				result = "root_replaced"
			}
		}
		metrics.RecordOperation("datastore", op, result, time.Since(start))
	}()

	for attempt := 0; ; attempt++ {
		tx, err := sw.s.db.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		fnErr := fn(tx)
		if fnErr == nil {
			fnErr = deleteDirBatchFailpointError(op)
		}
		if fnErr != nil {
			_ = tx.Rollback()
			if isRetryableTxnError(fnErr) && attempt < deleteDirBatchRetries {
				metrics.RecordOperation("datastore", op+"_retry", "retryable", 0)
				if !sleepWithContext(ctx, time.Duration(50*(1<<attempt))*time.Millisecond) {
					return ctx.Err()
				}
				continue
			}
			return fnErr
		}
		if err := tx.Commit(); err != nil {
			if isRetryableTxnError(err) && attempt < deleteDirBatchRetries {
				metrics.RecordOperation("datastore", op+"_retry", "retryable", 0)
				if !sleepWithContext(ctx, time.Duration(50*(1<<attempt))*time.Millisecond) {
					return ctx.Err()
				}
				continue
			}
			return err
		}
		return nil
	}
}

func sleepWithContext(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}

// errDeleteDirBatchInjected is the synthetic deadlock returned by
// deleteDirBatchFailpointError when its failpoint is armed; the message keeps
// isRetryableTxnError true so the batch retry path is exercised.
var errDeleteDirBatchInjected = errors.New("injected failpoint: Deadlock found when trying to get lock; try restarting transaction")

// deleteDirBatchFailpointError lets failpoint tests inject one synthetic
// retryable error per batch operation (op). It is a no-op unless the
// failpoint is armed: the failpoint.Inject marker compiles to an empty call
// until failpoint-ctl rewrites it, so normal runs see exactly the
// non-instrumented behavior.
func deleteDirBatchFailpointError(op string) error {
	var injected error
	failpoint.Inject("deleteDirBatchError", func(val failpoint.Value) {
		if name, ok := val.(string); ok && name == op {
			injected = errDeleteDirBatchInjected
		}
	})
	return injected
}

// selectChildrenForUpdateTx selects up to limit children of parent, locking
// them FOR UPDATE. The (parent_path_hash, name) index makes this a bounded
// range scan.
func (s *Store) selectChildrenForUpdateTx(ctx context.Context, tx *sql.Tx, parent string, limit int) ([]deleteChildRow, error) {
	rows, err := tx.QueryContext(ctx, `SELECT node_id, path, COALESCE(file_id, ''), is_directory FROM file_nodes
		WHERE `+s.scope.And(`parent_path_hash = ? AND parent_path = ?`)+`
		ORDER BY name LIMIT ? FOR UPDATE`,
		s.scope.Args(fileNodePathHash(parent), parent, limit)...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	var out []deleteChildRow
	for rows.Next() {
		var r deleteChildRow
		var isDir int
		if err := rows.Scan(&r.nodeID, &r.path, &r.fileID, &isDir); err != nil {
			return nil, err
		}
		r.isDir = isDir != 0
		out = append(out, r)
	}
	return out, rows.Err()
}

// drainChildrenBatchTx processes one batch of parent's children: files are
// deleted and fed to the orphan pipeline (I3), empty directories are deleted
// with a post-DELETE re-check (I1), non-empty directories are queued for
// descent.
func (s *Store) drainChildrenBatchTx(ctx context.Context, tx *sql.Tx, parent string) (*drainOutcome, error) {
	rows, err := s.selectChildrenForUpdateTx(ctx, tx, parent, deleteDirSweepBatchSize)
	if err != nil {
		return nil, err
	}
	failpoint.InjectCall("drainChildrenBatchAfterSelect", parent)
	out := &drainOutcome{rows: int64(len(rows))}
	for _, r := range rows {
		if !r.isDir {
			if _, err := tx.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+
				s.scope.And(`node_id = ?`), s.scope.Args(r.nodeID)...); err != nil {
				return nil, err
			}
			out.deleted++
			if r.fileID != "" {
				out.fileIDs = append(out.fileIDs, r.fileID)
			}
			continue
		}
		has, err := s.dirHasChildrenTx(ctx, tx, r.path)
		if err != nil {
			return nil, err
		}
		if has {
			out.subdirs = append(out.subdirs, r.path)
			continue
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+
			s.scope.And(`node_id = ?`), s.scope.Args(r.nodeID)...); err != nil {
			return nil, err
		}
		out.deleted++
		failpoint.InjectCall("drainEmptyDirBeforeRecheck", r.path)
		still, err := s.dirHasChildrenTx(ctx, tx, r.path)
		if err != nil {
			return nil, err
		}
		if still {
			out.racedDir = r.path
			return out, errDirDeleteRaced
		}
	}
	orphans, err := s.processOrphansTx(ctx, tx, out.fileIDs)
	if err != nil {
		return nil, err
	}
	out.orphans = orphans
	return out, nil
}

// processOrphansTx runs the orphan pipeline for file IDs whose dentries were
// deleted in this transaction: lock the inodes, keep only IDs with no
// remaining file_nodes reference, mark them DELETED, drop their tags, and
// enqueue batched GC tasks.
func (s *Store) processOrphansTx(ctx context.Context, tx *sql.Tx, fileIDs []string) (int64, error) {
	if len(fileIDs) == 0 {
		return 0, nil
	}
	sort.Strings(fileIDs)
	// Destructive in-place dedup: this reorders and truncates the caller's
	// backing array (out.fileIDs). Safe today because callers never read the
	// slice after processOrphansTx; keep that contract explicit.
	deduped := fileIDs[:0]
	for i, id := range fileIDs {
		if i == 0 || id != fileIDs[i-1] {
			deduped = append(deduped, id)
		}
	}
	if err := s.lockFileIDsForDeleteTx(ctx, tx, deduped); err != nil {
		return 0, err
	}
	orphaned, err := s.scanOrphanedFilesByIDTx(ctx, tx, deduped)
	if err != nil {
		return 0, err
	}
	if len(orphaned) == 0 {
		return 0, nil
	}
	orphanIDs := make([]string, 0, len(orphaned))
	for _, f := range orphaned {
		orphanIDs = append(orphanIDs, f.FileID)
	}
	if err := s.markFilesDeletedTx(ctx, tx, orphanIDs); err != nil {
		return 0, err
	}
	if err := s.deleteFileTagsByIDsTx(ctx, tx, orphanIDs); err != nil {
		return 0, err
	}
	now := time.Now().UTC()
	tasks := make([]*FileGCTask, 0, len(orphaned))
	for _, f := range orphaned {
		task, err := NewFileGCTaskFromFile(f, now)
		if err != nil {
			return 0, err
		}
		tasks = append(tasks, task)
	}
	return s.enqueueFileGCTasksTx(tx, tasks)
}