// Package datastore provides the tenant data-plane metadata store for dat9.
// Core metadata is split across file_nodes (dentry/path tree), inodes,
// contents, and semantic. The legacy files table is still supported for
// existing tenants during the split-table transition.
package datastore

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/pingcap/failpoint"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/mysqlutil"
	"go.uber.org/zap"
)

var (
	ErrNotFound                   = errors.New("not found")
	ErrUploadNotActive            = errors.New("upload is not in UPLOADING state")
	ErrUploadExpired              = errors.New("upload has expired")
	ErrPathConflict               = errors.New("path already exists")
	ErrDirectoryNotEmpty          = errors.New("directory not empty")
	ErrInvalidLinkTarget          = errors.New("invalid link target")
	ErrInvalidRootDentry          = errors.New("root path is implicit and cannot be mutated as a file node")
	ErrUploadConflict             = errors.New("active upload already exists for this path")
	ErrIdempotencyConflict        = errors.New("duplicate idempotency key")
	ErrJournalConflict            = errors.New("journal create conflict")
	ErrJournalClosed              = errors.New("journal is closed")
	ErrJournalValidation          = errors.New("journal validation failed")
	ErrJournalPayloadTooLarge     = errors.New("journal payload too large")
	ErrRevisionConflict           = errors.New("revision conflict")
	ErrContentLayoutConflict      = errors.New("content layout conflict")
	ErrFSLayerBaseInodeChanged    = errors.New("base inode changed")
	ErrFSLayerBasePathMissing     = errors.New("base path missing")
	ErrFSLayerBasePathExists      = errors.New("base path exists")
	ErrFSLayerBaseRevisionChanged = errors.New("base revision changed")
	ErrFileGCTaskLeaseMismatch    = errors.New("file gc task lease mismatch")
	ErrCommitOutcomeUnknown       = errors.New("transaction commit outcome unknown")
	// ErrDeleteIncomplete marks a bounded recursive delete that exhausted its
	// transaction/time budget before finishing. The operation is resumable:
	// callers may retry the same request to continue.
	ErrDeleteIncomplete = errors.New("recursive delete incomplete, retry to resume")
)

type StorageType string

const (
	StorageDB9 StorageType = "db9"
	StorageS3  StorageType = "s3"
)

// ContentLayout selects the physical object semantics for S3-backed content.
type ContentLayout string

const (
	ContentLayoutSingle    ContentLayout = "single"
	ContentLayoutAppendLog ContentLayout = "append_log"
)

const (
	fileTypeModeMask   uint32 = 0o170000
	permissionModeMask uint32 = 0o7777
)

type StorageEncryptionMode string

const (
	StorageEncryptionLegacy  StorageEncryptionMode = "legacy"
	StorageEncryptionNone    StorageEncryptionMode = "none"
	StorageEncryptionSSES3   StorageEncryptionMode = "sse-s3"
	StorageEncryptionSSEKMS  StorageEncryptionMode = "sse-kms"
	StorageEncryptionDSSEKMS StorageEncryptionMode = "dsse-kms"
)

type FileStatus string

const (
	StatusPending   FileStatus = "PENDING"
	StatusConfirmed FileStatus = "CONFIRMED"
	StatusDeleted   FileStatus = "DELETED"
)

type UploadStatus string

const (
	UploadInitiated UploadStatus = "INITIATED"
	UploadUploading UploadStatus = "UPLOADING"
	UploadCompleted UploadStatus = "COMPLETED"
	UploadAborted   UploadStatus = "ABORTED"
	UploadExpired   UploadStatus = "EXPIRED"
)

type UploadTargetKind string

const (
	UploadTargetFile    UploadTargetKind = "file"
	UploadTargetFSLayer UploadTargetKind = "fs_layer"
)

// FileNode represents a row in the file_nodes table (dentry).
type FileNode struct {
	NodeID      string
	Path        string
	ParentPath  string
	Name        string
	IsDirectory bool
	FileID      string // empty for directories
	InodeID     string // links to inodes.inode_id; for directories, this is the directory's inode
	CreatedAt   time.Time
}

// NodeFileID is the dentry's current file identity: confirmed file_id if
// present, otherwise the node file_id / inode_id.
func NodeFileID(nf *NodeWithFile) string {
	if nf == nil {
		return ""
	}
	if nf.File != nil && nf.File.FileID != "" {
		return nf.File.FileID
	}
	if nf.Node.FileID != "" {
		return nf.Node.FileID
	}
	return nf.Node.InodeID
}

func dentryIdentity(fileID, inodeID sql.NullString) string {
	if fileID.Valid && fileID.String != "" {
		return fileID.String
	}
	if inodeID.Valid {
		return inodeID.String
	}
	return ""
}

// File represents a row in the files table (inode).
type File struct {
	FileID                 string
	StorageType            StorageType
	ContentLayout          ContentLayout
	StorageRef             string
	StorageEncryptionMode  StorageEncryptionMode
	StorageEncryptionKeyID string
	ContentBlob            []byte
	ContentType            string
	SizeBytes              int64
	ChecksumSHA256         string
	Revision               int64
	Mode                   uint32
	// EmbeddingRevision tracks which file revision produced the stored embedding.
	EmbeddingRevision *int64
	Status            FileStatus
	SourceID          string
	ContentText       string
	Description       string
	// DescriptionEmbeddingRevision tracks which file revision produced the stored description_embedding.
	DescriptionEmbeddingRevision *int64
	CreatedAt                    time.Time
	ConfirmedAt                  *time.Time
	ExpiresAt                    *time.Time
}

// NodeWithFile joins file_nodes and files for stat/read operations.
type NodeWithFile struct {
	Node    FileNode
	File    *File // nil for directories
	Mode    uint32
	HasMode bool
}

// Upload represents a row in the uploads table.
type Upload struct {
	UploadID               string
	FileID                 string
	TargetKind             UploadTargetKind
	LayerID                string
	TargetPath             string
	BaseInodeID            string
	Mode                   uint32
	S3UploadID             string
	S3Key                  string
	StorageEncryptionMode  StorageEncryptionMode
	StorageEncryptionKeyID string
	TotalSize              int64
	PartSize               int64
	PartsTotal             int
	ExpectedRevision       *int64
	Status                 UploadStatus
	FingerprintSHA         string
	IdempotencyKey         string
	Description            string
	CreatedAt              time.Time
	UpdatedAt              time.Time
	ExpiresAt              time.Time
}

// Store is the metadata store backed by TiDB/MySQL (stand-in for db9).
type Store struct {
	db             *sql.DB
	useLegacyFiles bool // true when the legacy `files` table exists and needs dual-write
	scope          Scope
	// externalDB marks a Store wrapping a handle owned by someone else (a
	// shared-pool *sql.DB serving many tenants); Close must not close it.
	externalDB bool
}

func Open(dsn string) (*Store, error) {
	return OpenForTenant(context.Background(), dsn, "")
}

func OpenForTenant(ctx context.Context, dsn, tenantID string) (*Store, error) {
	return OpenForTenantWithOrg(ctx, dsn, tenantID, "")
}

func OpenForTenantWithOrg(ctx context.Context, dsn, tenantID, tidbCloudOrgID string) (*Store, error) {
	return OpenForTenantScoped(ctx, dsn, tenantID, tidbCloudOrgID, StandaloneScope(0))
}

// OpenForTenantScoped opens a Store whose SQL is shaped by scope: standalone
// (per-tenant DB, no fs_id columns) or shared (multi-tenant DB with fs_id
// row keys). The scope is fixed for the lifetime of the Store.
func OpenForTenantScoped(ctx context.Context, dsn, tenantID, tidbCloudOrgID string, scope Scope) (*Store, error) {
	lower := strings.ToLower(dsn)
	if strings.Contains(lower, "multistatements=true") || strings.Contains(lower, "multistatements=1") {
		return nil, fmt.Errorf("multiStatements is not allowed in production DSN")
	}
	db, err := mysqlutil.OpenInstrumentedForTenantWithOrg(ctx, dsn, mysqlutil.RoleUser, tenantID, tidbCloudOrgID)
	if err != nil {
		return nil, fmt.Errorf("open db: %w", err)
	}
	s := &Store{db: db, scope: scope}
	hasLegacy, err := s.detectLegacyFiles(ctx)
	if err != nil {
		_ = mysqlutil.CloseInstrumented(db)
		return nil, fmt.Errorf("detect legacy files table: %w", err)
	}
	s.useLegacyFiles = hasLegacy
	return s, nil
}

func (s *Store) Close() error {
	if s.externalDB {
		return nil
	}
	return mysqlutil.CloseInstrumented(s.db)
}
func (s *Store) DB() *sql.DB { return s.db }

// NewStoreWithDB wraps an externally owned *sql.DB (e.g. a shared-pool handle
// serving many tenants) with the given scope. Close is a no-op — the owner
// controls the handle's lifetime. No legacy-files detection runs: shared
// schema databases never carry the legacy files table.
func NewStoreWithDB(db *sql.DB, scope Scope) *Store {
	return &Store{db: db, scope: scope, externalDB: true}
}

// Scope returns the schema-shape scope fixed at construction.
func (s *Store) Scope() Scope { return s.scope }

// HasLegacyFiles reports whether the legacy `files` table exists in this
// tenant database. When false, all writes skip the `files` table entirely
// and only target the split tables (inodes / contents / semantic).
func (s *Store) HasLegacyFiles() bool { return s.useLegacyFiles }

func (s *Store) detectLegacyFiles(ctx context.Context) (bool, error) {
	var n int
	err := s.db.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM information_schema.tables
		 WHERE table_schema = DATABASE() AND table_name = 'files'`).Scan(&n)
	if err != nil {
		return false, err
	}
	return n > 0, nil
}

// InTx runs fn inside a database transaction. If fn returns an error, the
// transaction is rolled back; otherwise it is committed.
func (s *Store) InTx(ctx context.Context, fn func(tx *sql.Tx) error) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "in_tx", start, &err)

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	if err := fn(tx); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("%w: %w", ErrCommitOutcomeUnknown, err)
	}
	failpoint.Inject("inTxCommitOutcomeUnknown", func(val failpoint.Value) {
		err = fmt.Errorf("%w: %v", ErrCommitOutcomeUnknown, val)
	})
	return err
}

func (s *Store) columnExists(table, column string) bool {
	var count int
	err := s.db.QueryRow(
		`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = ? AND column_name = ?`,
		table, column).Scan(&count)
	return err == nil && count > 0
}

// --- file_nodes operations ---

func isRootFileNodePath(path string) bool {
	return path == "/"
}

func isRootFileNode(n FileNode) bool {
	return isRootFileNodePath(n.Path)
}

// fileNodePathHash is the lowercase 64-character SHA-256 hex hash used as the
// short indexed discriminator for file_nodes paths. Callers must keep the full
// path predicate alongside the hash predicate so a hash collision cannot alias
// two distinct paths.
func fileNodePathHash(path string) string {
	return StorageRefHash(path)
}

func (s *Store) InsertNode(ctx context.Context, n *FileNode) error {
	start := time.Now()
	var opErr error
	defer observeStoreOp(ctx, "insert_node", start, &opErr)

	if isRootFileNode(*n) {
		opErr = ErrInvalidRootDentry
		return ErrInvalidRootDentry
	}
	_, err := s.db.ExecContext(ctx, `INSERT INTO file_nodes (`+s.scope.InsCols(`node_id, path, path_hash, parent_path, parent_path_hash, name, is_directory, file_id, inode_id, created_at`)+`)
		VALUES (`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, ?, ?, ?, ?`)+`)`,
		s.scope.Args(n.NodeID, n.Path, fileNodePathHash(n.Path), n.ParentPath, fileNodePathHash(n.ParentPath),
			n.Name, n.IsDirectory, nullStr(n.FileID), nullStr(n.InodeID), n.CreatedAt.UTC())...)
	if isUniqueViolation(err) {
		opErr = ErrPathConflict
		return ErrPathConflict
	}
	opErr = err
	return err
}

func (s *Store) GetNode(ctx context.Context, path string) (*FileNode, error) {
	start := time.Now()
	var opErr error
	defer observeStoreOp(ctx, "get_node", start, &opErr)

	row := s.db.QueryRowContext(ctx, `SELECT node_id, path, parent_path, name, is_directory, file_id, inode_id, created_at
		FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`), s.scope.Args(fileNodePathHash(path), path)...)
	n, err := scanNode(row)
	opErr = err
	return n, err
}

// NodeExists reports whether any dentry occupies path without loading file
// metadata. It is used by create-if-absent paths that only need conflict
// precedence before doing heavier quota/storage work.
func (s *Store) NodeExists(ctx context.Context, path string) (exists bool, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "node_exists", start, &err)

	var one int
	err = s.db.QueryRowContext(ctx, `SELECT 1 FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` LIMIT 1`,
		s.scope.Args(fileNodePathHash(path), path)...).Scan(&one)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return true, nil
}

func (s *Store) ListNodes(ctx context.Context, parentPath string) (out []*FileNode, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "list_nodes", start, &err)

	rows, err := s.db.QueryContext(ctx, `SELECT node_id, path, parent_path, name, is_directory, file_id, inode_id, created_at
		FROM file_nodes WHERE `+s.scope.And(`parent_path_hash = ? AND parent_path = ?`)+` ORDER BY name`,
		s.scope.Args(fileNodePathHash(parentPath), parentPath)...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	nodes := make([]*FileNode, 0)
	for rows.Next() {
		n, err := scanNode(rows)
		if err != nil {
			return nil, err
		}
		if isRootFileNode(*n) {
			continue
		}
		nodes = append(nodes, n)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	out = nodes
	return out, nil
}

func (s *Store) DeleteNode(ctx context.Context, path string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "delete_node", start, &err)

	res, err := s.db.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`),
		s.scope.Args(fileNodePathHash(path), path)...)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

// DeleteEmptyDir atomically checks a directory is empty and deletes it. The
// post-DELETE re-check closes the TiDB no-gap-lock window where a concurrent
// create commits a child between the emptiness check and the delete.
func (s *Store) DeleteEmptyDir(ctx context.Context, path string) (err error) {
	return s.deleteEmptyDir(ctx, path, "")
}

// deleteEmptyDirFenced is DeleteEmptyDir constrained to one root generation:
// it first locks the directory row and verifies it still carries nodeID.
// ErrNotFound is returned when the row is gone OR a different node now
// occupies the path, so a batched recursive delete can never delete a
// replacement tree recreated at the same path (design doc §4.2 I2).
func (s *Store) deleteEmptyDirFenced(ctx context.Context, path, nodeID string) (err error) {
	return s.deleteEmptyDir(ctx, path, nodeID)
}

// DeleteEmptyDirIfInodeID deletes an empty directory at path only if it still
// names expectedInodeID. A missing path, non-directory occupant, or inode
// mismatch is a no-op so a concurrent replacement is left in place.
func (s *Store) DeleteEmptyDirIfInodeID(ctx context.Context, path, expectedInodeID string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "delete_empty_dir_if_inode_id", start, &err)
	if expectedInodeID == "" {
		return nil
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	var gotInode sql.NullString
	var isDir int
	err = tx.QueryRowContext(ctx, `SELECT inode_id, is_directory FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
		s.scope.Args(fileNodePathHash(path), path)...).Scan(&gotInode, &isDir)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	if isDir == 0 || !gotInode.Valid || gotInode.String != expectedInodeID {
		return tx.Commit()
	}

	hasChildren, err := s.dirHasChildrenTx(ctx, tx, path)
	if err != nil {
		return err
	}
	if hasChildren {
		return fmt.Errorf("%w: %s", ErrDirectoryNotEmpty, path)
	}

	res, err := tx.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ? AND is_directory = 1 AND inode_id = ?`),
		s.scope.Args(fileNodePathHash(path), path, expectedInodeID)...)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return tx.Commit()
	}
	stillHasChildren, err := s.dirHasChildrenTx(ctx, tx, path)
	if err != nil {
		return err
	}
	if stillHasChildren {
		return fmt.Errorf("%w: %s", ErrDirectoryNotEmpty, path)
	}
	return tx.Commit()
}

func (s *Store) deleteEmptyDir(ctx context.Context, path, nodeID string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "delete_empty_dir", start, &err)

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	if nodeID != "" {
		// Generation fence: lock the exact row by PRIMARY KEY and verify it is
		// still a directory at path carrying nodeID before deciding anything
		// (a PK point lock avoids secondary-index gap locks). A missing row or
		// a different node_id means the original generation is already gone —
		// report ErrNotFound so the caller stops instead of treating the path
		// as deletable.
		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(nodeID)...).Scan(&got, &gotPath, &isDir)
		if errors.Is(err, sql.ErrNoRows) || (err == nil && (got != nodeID || gotPath != path || isDir == 0)) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
	}
	hasChildren, err := s.dirHasChildrenTx(ctx, tx, path)
	if err != nil {
		return err
	}
	if hasChildren {
		return fmt.Errorf("%w: %s", ErrDirectoryNotEmpty, path)
	}
	deleteWhere := `path_hash = ? AND path = ? AND is_directory = 1`
	deleteArgs := []any{fileNodePathHash(path), path}
	if nodeID != "" {
		deleteWhere += ` AND node_id = ?`
		deleteArgs = append(deleteArgs, nodeID)
	}
	res, err := tx.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+s.scope.And(deleteWhere),
		s.scope.Args(deleteArgs...)...)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	stillHasChildren, err := s.dirHasChildrenTx(ctx, tx, path)
	if err != nil {
		return err
	}
	if stillHasChildren {
		return fmt.Errorf("%w: %s", ErrDirectoryNotEmpty, path)
	}
	err = tx.Commit()
	return err
}

func (s *Store) DeleteNodesByPrefix(ctx context.Context, prefix string) (n int64, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "delete_nodes_by_prefix", start, &err)

	where, args := pathPrefixPredicate("path", prefix)
	res, err := s.db.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+s.scope.And(where), s.scope.Args(args...)...)
	if err != nil {
		return 0, err
	}
	n, err = res.RowsAffected()
	return n, err
}

func (s *Store) UpdateNodePath(ctx context.Context, oldPath, newPath, newParentPath, newName string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "update_node_path", start, &err)

	if isRootFileNodePath(oldPath) || isRootFileNodePath(newPath) {
		return ErrInvalidRootDentry
	}
	res, err := s.db.ExecContext(ctx, `UPDATE file_nodes SET path = ?, path_hash = ?, parent_path = ?, parent_path_hash = ?, name = ?
		WHERE `+s.scope.And(`path_hash = ? AND path = ?`),
		append([]any{newPath, fileNodePathHash(newPath), newParentPath, fileNodePathHash(newParentPath), newName},
			s.scope.Args(fileNodePathHash(oldPath), oldPath)...)...)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

// RenameFileReplacingTarget atomically renames a file node. If newPath already
// names a file, that target dentry is replaced and its file is marked deleted
// when no other nodes reference it.
func (s *Store) RenameFileReplacingTarget(ctx context.Context, oldPath, newPath, newParentPath, newName string) (out *File, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "rename_file_replacing_target", start, &err)

	if isRootFileNodePath(oldPath) || isRootFileNodePath(newPath) {
		return nil, ErrInvalidRootDentry
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()

	type nodeRef struct {
		nodeID string
		fileID sql.NullString
		isDir  bool
	}
	var old nodeRef
	err = tx.QueryRow(`SELECT node_id, file_id, is_directory FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
		s.scope.Args(fileNodePathHash(oldPath), oldPath)...).Scan(&old.nodeID, &old.fileID, &old.isDir)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	if old.isDir {
		return nil, fmt.Errorf("source is a directory: %s", oldPath)
	}
	if oldPath == newPath {
		return nil, tx.Commit()
	}

	var dst nodeRef
	hasDst := false
	err = tx.QueryRow(`SELECT node_id, file_id, is_directory FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
		s.scope.Args(fileNodePathHash(newPath), newPath)...).Scan(&dst.nodeID, &dst.fileID, &dst.isDir)
	if err != nil {
		if !errors.Is(err, sql.ErrNoRows) {
			return nil, err
		}
	} else {
		hasDst = true
		if dst.isDir {
			return nil, ErrPathConflict
		}
		if dst.nodeID == old.nodeID {
			return nil, tx.Commit()
		}
		if _, err := tx.Exec(`DELETE FROM file_nodes WHERE `+s.scope.And(`node_id = ?`), s.scope.Args(dst.nodeID)...); err != nil {
			return nil, err
		}
	}

	res, err := tx.Exec(`UPDATE file_nodes SET path = ?, path_hash = ?, parent_path = ?, parent_path_hash = ?, name = ? WHERE `+s.scope.And(`node_id = ?`),
		append([]any{newPath, fileNodePathHash(newPath), newParentPath, fileNodePathHash(newParentPath), newName}, s.scope.Args(old.nodeID)...)...)
	if isUniqueViolation(err) {
		return nil, ErrPathConflict
	}
	if err != nil {
		return nil, err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return nil, ErrNotFound
	}

	if hasDst && dst.fileID.Valid && dst.fileID.String != "" {
		var count int64
		if err := tx.QueryRow(`SELECT COUNT(*) FROM file_nodes WHERE `+s.scope.And(`file_id = ?`)+` FOR UPDATE`, s.scope.Args(dst.fileID.String)...).Scan(&count); err != nil {
			return nil, err
		}
		if count == 0 {
			if s.useLegacyFiles {
				if _, err := tx.Exec(`UPDATE files SET status = 'DELETED' WHERE file_id = ?`, dst.fileID.String); err != nil {
					return nil, err
				}
			}
			if _, err := tx.Exec(`UPDATE inodes SET status = 'DELETED' WHERE `+s.scope.And(`inode_id = ?`), s.scope.Args(dst.fileID.String)...); err != nil {
				return nil, err
			}
			if _, err := tx.Exec(`DELETE FROM file_tags WHERE `+s.scope.And(`file_id = ?`), s.scope.Args(dst.fileID.String)...); err != nil {
				return nil, err
			}
			f, err := s.scanFileForGCTx(tx, dst.fileID.String)
			if err != nil {
				return nil, err
			}
			task, err := NewFileGCTaskFromFile(f, time.Now().UTC())
			if err != nil {
				return nil, err
			}
			if _, err := s.EnqueueFileGCTaskTx(tx, task); err != nil {
				return nil, err
			}
			out = f
		}
	}

	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return out, nil
}

// RenameFileNoReplace atomically renames a file node and fails if newPath
// already exists. It is used by fs-layer commit to avoid replacing a target
// that may have been created after the layer entry was recorded.
func (s *Store) RenameFileNoReplace(ctx context.Context, oldPath, newPath, newParentPath, newName string) error {
	return s.renameFileNoReplace(ctx, oldPath, newPath, newParentPath, newName, "")
}

// RenameFileNoReplaceIfFileID is RenameFileNoReplace constrained to one source
// inode: after locking the old path, a mismatch or missing expectedFileID is
// ErrPathConflict and the occupant is left unmoved.
func (s *Store) RenameFileNoReplaceIfFileID(ctx context.Context, oldPath, newPath, newParentPath, newName, expectedFileID string) error {
	return s.renameFileNoReplace(ctx, oldPath, newPath, newParentPath, newName, expectedFileID)
}

func (s *Store) renameFileNoReplace(ctx context.Context, oldPath, newPath, newParentPath, newName, expectedFileID string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "rename_file_no_replace", start, &err)

	if isRootFileNodePath(oldPath) || isRootFileNodePath(newPath) {
		return ErrInvalidRootDentry
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	type nodeRef struct {
		nodeID string
		isDir  bool
	}
	var old nodeRef
	var oldFileID, oldInodeID sql.NullString
	err = tx.QueryRowContext(ctx, `SELECT node_id, is_directory, file_id, inode_id FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
		s.scope.Args(fileNodePathHash(oldPath), oldPath)...).Scan(&old.nodeID, &old.isDir, &oldFileID, &oldInodeID)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return err
	}
	if old.isDir {
		return fmt.Errorf("source is a directory: %s", oldPath)
	}
	if expectedFileID != "" && dentryIdentity(oldFileID, oldInodeID) != expectedFileID {
		return ErrPathConflict
	}
	if oldPath == newPath {
		return tx.Commit()
	}

	var dst nodeRef
	err = tx.QueryRowContext(ctx, `SELECT node_id, is_directory FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
		s.scope.Args(fileNodePathHash(newPath), newPath)...).Scan(&dst.nodeID, &dst.isDir)
	if err != nil {
		if !errors.Is(err, sql.ErrNoRows) {
			return err
		}
	} else {
		if dst.nodeID == old.nodeID {
			return tx.Commit()
		}
		return ErrPathConflict
	}

	res, err := tx.ExecContext(ctx, `UPDATE file_nodes SET path = ?, path_hash = ?, parent_path = ?, parent_path_hash = ?, name = ?
		WHERE `+s.scope.And(`node_id = ?`),
		append([]any{newPath, fileNodePathHash(newPath), newParentPath, fileNodePathHash(newParentPath), newName}, s.scope.Args(old.nodeID)...)...)
	if err != nil {
		if strings.Contains(err.Error(), "Duplicate entry") {
			return ErrPathConflict
		}
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrNotFound
	}
	return tx.Commit()
}

func (s *Store) RenameDir(ctx context.Context, oldPrefix, newPrefix string) (count int64, err error) {
	return s.renameDir(ctx, oldPrefix, newPrefix, true, "")
}

// RenameDirNoReplace moves a directory tree and fails if newPrefix already
// exists, including an empty destination directory.
func (s *Store) RenameDirNoReplace(ctx context.Context, oldPrefix, newPrefix string) (count int64, err error) {
	return s.renameDir(ctx, oldPrefix, newPrefix, false, "")
}

// RenameDirNoReplaceIfInodeID is RenameDirNoReplace constrained to one source
// directory inode. After locking the source tree, a mismatch is ErrPathConflict
// and the occupant is left unmoved.
func (s *Store) RenameDirNoReplaceIfInodeID(ctx context.Context, oldPrefix, newPrefix, expectedInodeID string) (count int64, err error) {
	return s.renameDir(ctx, oldPrefix, newPrefix, false, expectedInodeID)
}

func (s *Store) renameDir(ctx context.Context, oldPrefix, newPrefix string, replaceEmptyDest bool, expectedInodeID string) (count int64, err error) {
	start := time.Now()
	op := "rename_dir"
	if !replaceEmptyDest {
		op = "rename_dir_no_replace"
	}
	defer observeStoreOp(ctx, op, start, &err)

	if oldPrefix == "/" || newPrefix == "/" {
		return 0, ErrInvalidRootDentry
	}
	if oldPrefix == newPrefix {
		return 0, nil
	}
	if strings.HasPrefix(newPrefix, oldPrefix) {
		return 0, ErrPathConflict
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer func() { _ = tx.Rollback() }()

	rows, err := tx.QueryContext(ctx, `SELECT node_id, path, parent_path, name, file_id, inode_id FROM file_nodes
		WHERE `+s.andAsGrouped("", `path = ? OR path LIKE ?`)+` ORDER BY path FOR UPDATE`,
		s.scope.Args(oldPrefix, oldPrefix+"%")...)
	if err != nil {
		return 0, err
	}
	type update struct {
		nodeID, newPath, newParent, newName string
	}
	var updates []update
	var srcIdentity string
	for rows.Next() {
		var nodeID, p, pp, name string
		var fileID, inodeID sql.NullString
		if err := rows.Scan(&nodeID, &p, &pp, &name, &fileID, &inodeID); err != nil {
			_ = rows.Close()
			return 0, err
		}
		newPath := newPrefix + strings.TrimPrefix(p, oldPrefix)
		newParent := newPrefix + strings.TrimPrefix(pp, oldPrefix)
		newName := name
		if p == oldPrefix {
			newParent = parentPath(newPrefix)
			newPath = newPrefix
			newName = baseName(newPrefix)
			srcIdentity = dentryIdentity(fileID, inodeID)
		}
		updates = append(updates, update{nodeID, newPath, newParent, newName})
	}
	_ = rows.Close()
	if len(updates) == 0 {
		return 0, ErrNotFound
	}
	if expectedInodeID != "" && srcIdentity != expectedInodeID {
		return 0, ErrPathConflict
	}

	type dstRef struct {
		nodeID string
		isDir  bool
	}
	var dst dstRef
	err = tx.QueryRowContext(ctx, `SELECT node_id, is_directory FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
		s.scope.Args(fileNodePathHash(newPrefix), newPrefix)...).Scan(&dst.nodeID, &dst.isDir)
	if err != nil {
		if !errors.Is(err, sql.ErrNoRows) {
			return 0, err
		}
	} else {
		if !replaceEmptyDest || !dst.isDir {
			return 0, ErrPathConflict
		}
		childRows, err := tx.QueryContext(ctx, `SELECT node_id FROM file_nodes WHERE `+s.scope.And(`parent_path_hash = ? AND parent_path = ?`)+` FOR UPDATE`,
			s.scope.Args(fileNodePathHash(newPrefix), newPrefix)...)
		if err != nil {
			return 0, err
		}
		hasChild := childRows.Next()
		if closeErr := childRows.Close(); closeErr != nil {
			return 0, closeErr
		}
		if hasChild {
			return 0, ErrPathConflict
		}
		if _, err := tx.ExecContext(ctx, `DELETE FROM file_nodes WHERE `+s.scope.And(`node_id = ?`), s.scope.Args(dst.nodeID)...); err != nil {
			return 0, err
		}
	}

	stmt, err := tx.PrepareContext(ctx, `UPDATE file_nodes SET path = ?, path_hash = ?, parent_path = ?, parent_path_hash = ?, name = ? WHERE `+s.scope.And(`node_id = ?`))
	if err != nil {
		return 0, err
	}
	defer func() { _ = stmt.Close() }()

	for _, u := range updates {
		if _, err := stmt.Exec(append([]any{u.newPath, fileNodePathHash(u.newPath), u.newParent, fileNodePathHash(u.newParent), u.newName}, s.scope.Args(u.nodeID)...)...); err != nil {
			if isUniqueViolation(err) {
				return 0, ErrPathConflict
			}
			return 0, err
		}
	}

	if err := tx.Commit(); err != nil {
		return 0, err
	}
	count = int64(len(updates))
	return count, nil
}

func (s *Store) RefCount(ctx context.Context, fileID string) (count int64, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "ref_count", start, &err)

	err = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM file_nodes WHERE `+s.scope.And(`file_id = ?`), s.scope.Args(fileID)...).Scan(&count)
	return count, err
}

// RefCounts returns path-reference counts for the supplied file IDs.
func (s *Store) RefCounts(ctx context.Context, fileIDs []string) (counts map[string]int64, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "ref_counts", start, &err)

	counts = make(map[string]int64)
	seen := make(map[string]struct{}, len(fileIDs))
	unique := make([]string, 0, len(fileIDs))
	for _, fileID := range fileIDs {
		if fileID == "" {
			continue
		}
		if _, ok := seen[fileID]; ok {
			continue
		}
		seen[fileID] = struct{}{}
		unique = append(unique, fileID)
	}
	const batchSize = 500
	for start := 0; start < len(unique); start += batchSize {
		end := start + batchSize
		if end > len(unique) {
			end = len(unique)
		}
		batch := unique[start:end]
		rows, err := s.db.QueryContext(ctx,
			`SELECT file_id, COUNT(*) FROM file_nodes WHERE `+s.scope.And(`file_id IN (`+questionPlaceholders(len(batch))+`)`)+` GROUP BY file_id`,
			s.scope.Args(stringsToAny(batch)...)...)
		if err != nil {
			return nil, err
		}
		for rows.Next() {
			var fileID string
			var count int64
			if err := rows.Scan(&fileID, &count); err != nil {
				_ = rows.Close()
				return nil, err
			}
			counts[fileID] = count
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			return nil, err
		}
		if err := rows.Close(); err != nil {
			return nil, err
		}
	}
	return counts, nil
}

// LinkFileNode atomically creates dstPath as another directory entry for the
// same confirmed file entity referenced by srcPath.
func (s *Store) LinkFileNode(ctx context.Context, srcPath, dstPath, dstParentPath, dstName, nodeID string, createdAt time.Time) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "link_file_node", start, &err)

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	if err := s.LinkFileNodeTx(ctx, tx, srcPath, dstPath, dstParentPath, dstName, nodeID, createdAt); err != nil {
		return err
	}
	return tx.Commit()
}

// LinkFileNodeTx creates a hardlink node inside an existing transaction.
func (s *Store) LinkFileNodeTx(ctx context.Context, db execer, srcPath, dstPath, dstParentPath, dstName, nodeID string, createdAt time.Time) error {
	if isRootFileNodePath(srcPath) || isRootFileNodePath(dstPath) {
		return ErrInvalidRootDentry
	}
	if dstParentPath != "/" {
		var parentIsDir bool
		err := db.QueryRowContext(ctx, `SELECT is_directory FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
			s.scope.Args(fileNodePathHash(dstParentPath), dstParentPath)...).Scan(&parentIsDir)
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return err
		}
		if !parentIsDir {
			return ErrNotFound
		}
	}

	var fileID, inodeID sql.NullString
	var isDir bool
	err := db.QueryRowContext(ctx, `SELECT file_id, inode_id, is_directory FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
		s.scope.Args(fileNodePathHash(srcPath), srcPath)...).Scan(&fileID, &inodeID, &isDir)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return err
	}
	if isDir {
		return ErrInvalidLinkTarget
	}
	if !fileID.Valid || fileID.String == "" {
		return ErrNotFound
	}
	if err := s.lockConfirmedFileForAttachTx(db, fileID.String); err != nil {
		return err
	}
	linkInodeID := fileID.String
	if inodeID.Valid && inodeID.String != "" {
		linkInodeID = inodeID.String
	}
	_, err = db.Exec(`INSERT INTO file_nodes (`+s.scope.InsCols(`node_id, path, path_hash, parent_path, parent_path_hash, name, is_directory, file_id, inode_id, created_at`)+`)
		VALUES (`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, 0, ?, ?, ?`)+`)`,
		s.scope.Args(nodeID, dstPath, fileNodePathHash(dstPath), dstParentPath, fileNodePathHash(dstParentPath), dstName, fileID.String, linkInodeID, createdAt.UTC())...)
	if isUniqueViolation(err) {
		return ErrPathConflict
	}
	if err != nil {
		return err
	}
	return nil
}

func (s *Store) EnsureParentDirs(ctx context.Context, path string, genID func() string) error {
	start := time.Now()
	var err error
	defer observeStoreOp(ctx, "ensure_parent_dirs", start, &err)

	// Run inside a transaction so the sequence of per-parent inserts is
	// atomic. Deadlocks can still happen when many goroutines contend for
	// the same unique-key lock on file_nodes.path; retry a bounded number
	// of times.
	const maxAttempts = 3
	for attempt := 0; attempt < maxAttempts; attempt++ {
		var tx *sql.Tx
		tx, err = s.db.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		_, err = s.ensureParentDirsWithExecer(ctx, tx, path, genID)
		if err != nil {
			_ = tx.Rollback()
			if isDeadlock(err) && attempt < maxAttempts-1 {
				continue
			}
			return err
		}
		err = tx.Commit()
		if err != nil {
			_ = tx.Rollback()
			if isDeadlock(err) && attempt < maxAttempts-1 {
				continue
			}
			return err
		}
		return nil
	}
	return err
}

// EnsureParentDirsCreated is EnsureParentDirs plus the inode IDs inserted for
// missing ancestors in the successful transaction.
func (s *Store) EnsureParentDirsCreated(ctx context.Context, path string, genID func() string) (created []string, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "ensure_parent_dirs", start, &err)

	const maxAttempts = 3
	for attempt := 0; attempt < maxAttempts; attempt++ {
		var tx *sql.Tx
		tx, err = s.db.BeginTx(ctx, nil)
		if err != nil {
			return nil, err
		}
		created, err = s.ensureParentDirsWithExecer(ctx, tx, path, genID)
		if err != nil {
			_ = tx.Rollback()
			if isDeadlock(err) && attempt < maxAttempts-1 {
				continue
			}
			return nil, err
		}
		err = tx.Commit()
		if err != nil {
			_ = tx.Rollback()
			if isDeadlock(err) && attempt < maxAttempts-1 {
				continue
			}
			return nil, err
		}
		return created, nil
	}
	return nil, err
}

// --- files operations ---

func (s *Store) InsertFile(ctx context.Context, f *File) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "insert_file", start, &err)

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	if s.useLegacyFiles {
		mode := fileStorageEncryptionModeForWrite(f.StorageEncryptionMode)
		_, err = tx.Exec(`INSERT INTO files
			(file_id, storage_type, content_layout, storage_ref, storage_ref_hash, storage_encryption_mode, storage_encryption_key_id,
			 content_blob, content_type, size_bytes, checksum_sha256,
			 revision, status, source_id, content_text, description, created_at, confirmed_at, expires_at)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			f.FileID, f.StorageType, contentLayoutForWrite(f.ContentLayout), f.StorageRef, StorageRefHash(f.StorageRef), mode,
			storageEncryptionKeyIDForWrite(mode, f.StorageEncryptionKeyID), nilBytes(f.ContentBlob), nullStr(f.ContentType),
			f.SizeBytes, nullStr(f.ChecksumSHA256), f.Revision, f.Status,
			nullStr(f.SourceID), nullStr(f.ContentText), nullStr(f.Description),
			f.CreatedAt.UTC(), nilTime(f.ConfirmedAt), nilTime(f.ExpiresAt))
		if err != nil {
			return err
		}
	}

	// Dual-write to split tables
	if err := s.insertSplitTablesTx(tx, f); err != nil {
		return err
	}

	return tx.Commit()
}

// insertSplitTablesTx inserts corresponding rows into inodes, contents, and semantic.
func (s *Store) insertSplitTablesTx(tx execer, f *File) error {
	now := time.Now().UTC()
	mode := f.Mode
	if mode == 0 {
		mode = 0o644
	}
	inode := &Inode{
		InodeID:     f.FileID,
		SizeBytes:   f.SizeBytes,
		Revision:    f.Revision,
		Mode:        mode,
		Status:      f.Status,
		CreatedAt:   f.CreatedAt,
		Mtime:       coalesceTime(f.ConfirmedAt, now),
		ConfirmedAt: f.ConfirmedAt,
		ExpiresAt:   f.ExpiresAt,
	}
	if err := s.InsertInodeTx(tx, inode); err != nil {
		return fmt.Errorf("insert inode: %w", err)
	}
	content := &Content{
		InodeID:                f.FileID,
		StorageType:            f.StorageType,
		ContentLayout:          f.ContentLayout,
		StorageRef:             f.StorageRef,
		StorageEncryptionMode:  f.StorageEncryptionMode,
		StorageEncryptionKeyID: f.StorageEncryptionKeyID,
		ContentBlob:            f.ContentBlob,
		ContentType:            f.ContentType,
		ChecksumSHA256:         f.ChecksumSHA256,
		SourceID:               f.SourceID,
	}
	if err := s.InsertContentTx(tx, content); err != nil {
		return fmt.Errorf("insert content: %w", err)
	}
	semantic := &Semantic{
		InodeID:                      f.FileID,
		EmbeddingRevision:            f.EmbeddingRevision,
		DescriptionEmbeddingRevision: f.DescriptionEmbeddingRevision,
		ContentText:                  f.ContentText,
		Description:                  f.Description,
	}
	if err := s.InsertSemanticTx(tx, semantic); err != nil {
		return fmt.Errorf("insert semantic: %w", err)
	}
	return nil
}

func (s *Store) GetFile(ctx context.Context, fileID string) (out *File, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "get_file", start, &err)

	inode, err := s.GetInode(ctx, fileID)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	content, contentErr := s.GetContent(ctx, fileID)
	if contentErr != nil && !errors.Is(contentErr, sql.ErrNoRows) {
		return nil, fmt.Errorf("get content: %w", contentErr)
	}
	semantic, semanticErr := s.GetSemantic(ctx, fileID)
	if semanticErr != nil && !errors.Is(semanticErr, sql.ErrNoRows) {
		return nil, fmt.Errorf("get semantic: %w", semanticErr)
	}
	out = assembleFile(inode, content, semantic)
	return out, nil
}

// HasConfirmedS3StorageRef reports whether a confirmed S3 file still points at
// the exact storage ref. Callers pass the hash so the query can use
// idx_contents_storage_ref_hash and then compare the full ref to avoid trusting
// the hash as the identity.
func (s *Store) HasConfirmedS3StorageRef(ctx context.Context, storageRefHash, storageRef string) (out bool, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "has_confirmed_s3_storage_ref", start, &err)
	if storageRefHash == "" || storageRef == "" {
		return false, nil
	}
	row := s.db.QueryRowContext(ctx, `SELECT 1
		FROM contents c JOIN inodes i ON i.inode_id = c.inode_id
		WHERE `+scopeWhereAnd(s.scope, `c.storage_type = ? AND i.status = ? AND c.storage_ref_hash = ? AND c.storage_ref = ?`, "c", "i")+`
		LIMIT 1`, scopeWhereArgs(s.scope, 2, StorageS3, StatusConfirmed, storageRefHash, storageRef)...)
	var one int
	if err := row.Scan(&one); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

// HasFSLayerStorageRef reports whether retained fs-layer history points at the
// exact storage ref. Layer refs are checked separately from the high-volume
// confirmed-content path so ordinary blob GC keeps using its existing index.
func (s *Store) HasFSLayerStorageRef(ctx context.Context, storageRefHash, storageRef string) (out bool, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "has_fs_layer_storage_ref", start, &err)
	if storageRefHash == "" || storageRef == "" {
		return false, nil
	}
	row := s.db.QueryRowContext(ctx, `SELECT 1
		FROM fs_layer_entries
		WHERE `+s.scope.And(`storage_ref_hash = ? AND storage_ref = ?`)+`
		LIMIT 1`, s.scope.Args(storageRefHash, storageRef)...)
	var one int
	if err := row.Scan(&one); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (s *Store) UpdateFileContent(ctx context.Context, fileID string, storageType StorageType, storageRef, contentType, checksum, contentText string, contentBlob []byte, size int64, description string) (newRevision int64, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "update_file_content", start, &err)

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer func() { _ = tx.Rollback() }()

	now := time.Now().UTC()
	var rev int64
	if s.useLegacyFiles {
		query := `UPDATE files SET storage_type = ?, storage_ref = ?, storage_ref_hash = ?,
			content_blob = ?, content_type = ?, size_bytes = ?, checksum_sha256 = ?, content_text = ?`
		args := []any{storageType, storageRef, StorageRefHash(storageRef), nilBytes(contentBlob), nullStr(contentType), size,
			nullStr(checksum), nullStr(contentText)}
		if description != "" {
			query += `, description = ?, description_embedding = NULL, description_embedding_revision = NULL`
			args = append(args, description)
		} else {
			query += `, description_embedding_revision = CASE
				WHEN description_embedding IS NOT NULL THEN revision + 1
				ELSE description_embedding_revision
				END`
		}
		query += `, revision = revision + 1, status = 'CONFIRMED', confirmed_at = ? WHERE file_id = ?`
		args = append(args, now, fileID)
		res, err := tx.Exec(query, args...)
		if err != nil {
			return 0, err
		}
		n, _ := res.RowsAffected()
		if n == 0 {
			return 0, ErrNotFound
		}
		if err := tx.QueryRow(`SELECT revision FROM files WHERE file_id = ?`, fileID).Scan(&rev); err != nil {
			return 0, fmt.Errorf("read revision after update: %w", err)
		}
	} else {
		var currentRev int64
		if err := tx.QueryRow(`SELECT revision FROM inodes WHERE `+s.scope.And(`inode_id = ? AND status = 'CONFIRMED'`)+` FOR UPDATE`, s.scope.Args(fileID)...).Scan(&currentRev); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return 0, ErrNotFound
			}
			return 0, fmt.Errorf("read current revision: %w", err)
		}
		rev = currentRev + 1
	}

	// Dual-write to split tables
	if err := s.UpdateInodeContentTx(tx, fileID, size, rev, StatusConfirmed, now); err != nil {
		return 0, fmt.Errorf("update inode: %w", err)
	}
	var encryptionMode StorageEncryptionMode
	if err := tx.QueryRow(`SELECT storage_encryption_mode FROM contents WHERE `+s.scope.And(`inode_id = ?`), s.scope.Args(fileID)...).Scan(&encryptionMode); err != nil {
		return 0, fmt.Errorf("read encryption mode: %w", err)
	}
	if err := s.UpdateContentTx(tx, fileID, storageType, storageRef, contentType, checksum, contentBlob, encryptionMode); err != nil {
		return 0, fmt.Errorf("update content: %w", err)
	}
	if err := s.UpdateSemanticTx(tx, fileID, contentText, description); err != nil {
		return 0, fmt.Errorf("update semantic: %w", err)
	}

	return rev, tx.Commit()
}

// UpdateFileSearchText updates files.content_text for search enrichment.
// When expectedRevision is positive, the update applies only to that revision.
// This prevents stale async tasks from overwriting a newer file revision.
func (s *Store) UpdateFileSearchText(ctx context.Context, fileID string, expectedRevision int64, contentText string) (updated bool, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "update_file_search_text", start, &err)

	res, err := s.updateFileSearchTextExec(s.db, fileID, expectedRevision, contentText)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	return n > 0, nil
}

// UpdateFileSearchTextTx updates files.content_text inside an existing
// transaction while preserving the revision gate used by async image tasks.
func (s *Store) UpdateFileSearchTextTx(db execer, fileID string, expectedRevision int64, contentText string) (bool, error) {
	res, err := s.updateFileSearchTextExec(db, fileID, expectedRevision, contentText)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	return n > 0, nil
}

// MatchFileRevisionForUpdateTx locks the authoritative file row and reports
// whether a confirmed file still has expectedRevision. It supports no-op
// writebacks whose UPDATE would otherwise report zero affected rows.
func (s *Store) MatchFileRevisionForUpdateTx(db execer, fileID string, expectedRevision int64) (bool, error) {
	var revision int64
	var err error
	if s.useLegacyFiles {
		err = db.QueryRow(`SELECT revision FROM files WHERE file_id = ? AND status = 'CONFIRMED' FOR UPDATE`, fileID).Scan(&revision)
	} else {
		err = db.QueryRow(`SELECT revision FROM inodes WHERE `+s.scope.And(`inode_id = ? AND status = 'CONFIRMED'`)+` FOR UPDATE`, s.scope.Args(fileID)...).Scan(&revision)
	}
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return expectedRevision <= 0 || revision == expectedRevision, nil
}

func (s *Store) updateFileSearchTextExec(db execer, fileID string, expectedRevision int64, contentText string) (sql.Result, error) {
	if s.useLegacyFiles {
		var res sql.Result
		var err error
		if expectedRevision > 0 {
			res, err = db.Exec(`UPDATE files SET content_text = ?
				WHERE file_id = ? AND status = 'CONFIRMED' AND revision = ?`,
				nullStr(contentText), fileID, expectedRevision)
		} else {
			res, err = db.Exec(`UPDATE files SET content_text = ?
				WHERE file_id = ? AND status = 'CONFIRMED'`,
				nullStr(contentText), fileID)
		}
		if err != nil {
			return nil, err
		}
		if expectedRevision > 0 {
			if _, err := db.Exec(`UPDATE semantic SET content_text = ?
				WHERE `+s.scope.And(`inode_id = ? AND EXISTS (SELECT 1 FROM inodes WHERE `+s.scope.And(`inode_id = ? AND status = 'CONFIRMED' AND revision = ?`)+`)`),
				append([]any{nullStr(contentText)}, append(s.scope.Args(fileID), s.scope.Args(fileID, expectedRevision)...)...)...); err != nil {
				return nil, fmt.Errorf("update semantic content_text: %w", err)
			}
			return res, nil
		}
		if _, err := db.Exec(`UPDATE semantic SET content_text = ?
			WHERE `+s.scope.And(`inode_id = ? AND EXISTS (SELECT 1 FROM inodes WHERE `+s.scope.And(`inode_id = ? AND status = 'CONFIRMED'`)+`)`),
			append([]any{nullStr(contentText)}, append(s.scope.Args(fileID), s.scope.Args(fileID)...)...)...); err != nil {
			return nil, fmt.Errorf("update semantic content_text: %w", err)
		}
		return res, nil
	}
	// New tenant without legacy files: write directly to semantic.
	if expectedRevision > 0 {
		return db.Exec(`UPDATE semantic SET content_text = ?
			WHERE `+s.scope.And(`inode_id = ? AND EXISTS (SELECT 1 FROM inodes WHERE `+s.scope.And(`inode_id = ? AND status = 'CONFIRMED' AND revision = ?`)+`)`),
			append([]any{nullStr(contentText)}, append(s.scope.Args(fileID), s.scope.Args(fileID, expectedRevision)...)...)...)
	}
	return db.Exec(`UPDATE semantic SET content_text = ?
		WHERE `+s.scope.And(`inode_id = ? AND EXISTS (SELECT 1 FROM inodes WHERE `+s.scope.And(`inode_id = ? AND status = 'CONFIRMED'`)+`)`),
		append([]any{nullStr(contentText)}, append(s.scope.Args(fileID), s.scope.Args(fileID)...)...)...)
}

// ReplaceFileTagsTx replaces all tags for fileID inside an existing transaction.
//
// Callers should only invoke this when they intend to replace the tag set for
// the current write revision. Passing an empty map clears all existing tags.
// Callers that intend to preserve the current tag set must skip this call.
func (s *Store) ReplaceFileTagsTx(db execer, fileID string, tags map[string]string) error {
	if _, err := db.Exec(`DELETE FROM file_tags WHERE `+s.scope.And(`file_id = ?`), s.scope.Args(fileID)...); err != nil {
		return err
	}
	if len(tags) == 0 {
		return nil
	}

	keys := make([]string, 0, len(tags))
	for k := range tags {
		keys = append(keys, k)
	}
	sort.Strings(keys)

	for _, k := range keys {
		if _, err := db.Exec(`INSERT INTO file_tags (`+s.scope.InsCols(`file_id, tag_key, tag_value`)+`) VALUES (`+s.scope.InsVals(`?, ?, ?`)+`)`,
			s.scope.Args(fileID, k, tags[k])...); err != nil {
			return err
		}
	}
	return nil
}

// ReplaceFileTagsByPrefixTx replaces only tags whose key starts with prefix
// inside an existing transaction.
//
// This is intended for system-owned tag namespaces. Passing an empty map clears
// that namespace while preserving all other file tags.
func (s *Store) ReplaceFileTagsByPrefixTx(db execer, fileID string, prefix string, tags map[string]string) error {
	prefix = strings.TrimSpace(prefix)
	if prefix == "" {
		return fmt.Errorf("tag prefix is required")
	}
	pattern := likeLiteralPrefixPattern(prefix)
	if _, err := db.Exec(`DELETE FROM file_tags WHERE `+s.scope.And(`file_id = ? AND tag_key LIKE ? ESCAPE '\\'`), s.scope.Args(fileID, pattern)...); err != nil {
		return err
	}
	if len(tags) == 0 {
		return nil
	}

	keys := make([]string, 0, len(tags))
	for k := range tags {
		if !strings.HasPrefix(k, prefix) {
			return fmt.Errorf("tag key %q does not match prefix %q", k, prefix)
		}
		keys = append(keys, k)
	}
	sort.Strings(keys)

	for _, k := range keys {
		if _, err := db.Exec(`INSERT INTO file_tags (`+s.scope.InsCols(`file_id, tag_key, tag_value`)+`) VALUES (`+s.scope.InsVals(`?, ?, ?`)+`)`,
			s.scope.Args(fileID, k, tags[k])...); err != nil {
			return err
		}
	}
	return nil
}

func likeLiteralPrefixPattern(prefix string) string {
	replacer := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`)
	return replacer.Replace(prefix) + "%"
}

// GetFileTags returns all tags for fileID.
func (s *Store) GetFileTags(ctx context.Context, fileID string) (out map[string]string, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "get_file_tags", start, &err)

	rows, err := s.db.QueryContext(ctx, `SELECT tag_key, tag_value FROM file_tags WHERE `+s.scope.And(`file_id = ?`)+` ORDER BY tag_key`, s.scope.Args(fileID)...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	tags := make(map[string]string)
	for rows.Next() {
		var k string
		var v string
		if err := rows.Scan(&k, &v); err != nil {
			return nil, err
		}
		tags[k] = v
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	out = tags
	return out, nil
}

// execer abstracts *sql.DB and *sql.Tx for shared query execution.
type execer interface {
	Exec(query string, args ...interface{}) (sql.Result, error)
	QueryRow(query string, args ...interface{}) *sql.Row
	Query(query string, args ...interface{}) (*sql.Rows, error)
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
}

// FileStorageMeta holds the lightweight storage metadata needed by upload
// overwrite logic, fetched with FOR UPDATE row locking.
type FileStorageMeta struct {
	StorageType   StorageType
	ContentLayout ContentLayout
	StorageRef    string
	Revision      int64
	SizeBytes     int64
	ContentType   string
}

// GetFileStorageMetaForUpdateTx returns confirmed file storage metadata with
// row-level locking inside an existing transaction.
func (s *Store) GetFileStorageMetaForUpdateTx(db execer, fileID string) (*FileStorageMeta, error) {
	var m FileStorageMeta
	var contentType sql.NullString
	err := db.QueryRow(`SELECT c.storage_type, c.content_layout, c.storage_ref, i.revision, i.size_bytes, COALESCE(c.content_type, '')
		FROM inodes i JOIN contents c ON i.inode_id = c.inode_id
		WHERE `+scopeWhereAnd(s.scope, `i.inode_id = ? AND i.status = 'CONFIRMED'`, "i", "c")+` FOR UPDATE`, scopeWhereArgs(s.scope, 2, fileID)...).Scan(
		&m.StorageType, &m.ContentLayout, &m.StorageRef, &m.Revision, &m.SizeBytes, &contentType)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	m.ContentType = contentType.String
	return &m, nil
}

func (s *Store) CompleteUploadTx(db execer, uploadID string) error {
	_, err := db.Exec(`UPDATE uploads SET status = 'COMPLETED',
		updated_at = ?
		WHERE `+s.scope.And(`upload_id = ? AND status = 'UPLOADING'`),
		append([]any{time.Now().UTC()}, s.scope.Args(uploadID)...)...)
	return err
}

// Chmod updates the permission bits (mode) of the file or directory at path.
// It returns ErrNotFound if the path does not exist or the node has no
// associated inode record (e.g. an old directory created before the
// split-table migration).
func (s *Store) Chmod(ctx context.Context, path string, mode uint32) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "chmod", start, &err)

	if isRootFileNodePath(path) {
		return ErrInvalidRootDentry
	}
	node, err := s.GetNode(ctx, path)
	if err != nil {
		return err
	}
	inodeID := node.InodeID
	if inodeID == "" && !node.IsDirectory {
		inodeID = node.FileID
	}
	if inodeID == "" {
		return ErrNotFound
	}

	var currentMode uint32
	if err := s.db.QueryRowContext(ctx, `SELECT mode FROM inodes WHERE `+s.scope.And(`inode_id = ? AND status = 'CONFIRMED'`), s.scope.Args(inodeID)...).Scan(&currentMode); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return err
	}
	mode = (mode & permissionModeMask) | (currentMode & fileTypeModeMask)
	res, err := s.db.ExecContext(ctx, `UPDATE inodes SET mode = ? WHERE `+s.scope.And(`inode_id = ? AND status = 'CONFIRMED'`),
		append([]any{mode}, s.scope.Args(inodeID)...)...)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *Store) EnsureParentDirsTx(db execer, path string, genID func() string) error {
	_, err := s.ensureParentDirsWithExecer(context.Background(), db, path, genID)
	return err
}

func (s *Store) EnsureParentDirsTxCreated(db execer, path string, genID func() string) ([]string, error) {
	return s.ensureParentDirsWithExecer(context.Background(), db, path, genID)
}

// deterministicNodeID returns a stable 64-char hex ID derived from the path.
// Using a deterministic ID means concurrent creators for the same missing
// parent path will collide on the same inode_id; INSERT IGNORE swallows the
// duplicate, so no orphan inode is ever created. This works on TiDB as well
// as MySQL because it does not rely on gap locking.
func deterministicNodeID(path string) string {
	h := sha256.Sum256([]byte(path))
	return hex.EncodeToString(h[:])
}

// ensureParentDirsWithExecer is the shared implementation used by both
// EnsureParentDirs and EnsureParentDirsTx. It uses a deterministic inode_id
// derived from the path so concurrent attempts to create the same missing
// parent cannot leak orphan inodes, even on TiDB which does not support gap
// locking for SELECT ... FOR UPDATE on non-existing rows.
func (s *Store) ensureParentDirsWithExecer(ctx context.Context, db execer, path string, genID func() string) ([]string, error) {
	var created []string
	var ancestors []string
	cur := path
	for {
		parent := parentPath(cur)
		if parent == cur || parent == "/" {
			break
		}
		ancestors = append(ancestors, parent)
		cur = parent
	}
	now := time.Now().UTC()
	for i := len(ancestors) - 1; i >= 0; i-- {
		dirPath := ancestors[i]
		pp := parentPath(dirPath)
		name := baseName(dirPath)

		// Check if the directory already exists and has an inode_id.
		var existingInodeID sql.NullString
		selectErr := db.QueryRowContext(ctx,
			`SELECT inode_id FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ? AND is_directory = 1`),
			s.scope.Args(fileNodePathHash(dirPath), dirPath)...).Scan(&existingInodeID)
		if selectErr == nil && existingInodeID.Valid {
			// Already exists and has an inode_id — nothing to do.
			continue
		}
		if selectErr != nil && !errors.Is(selectErr, sql.ErrNoRows) {
			return nil, fmt.Errorf("check parent dir %s: %w", dirPath, selectErr)
		}

		// Use a deterministic ID so concurrent creators for the same missing
		// parent path collide on the same inode_id. INSERT IGNORE swallows the
		// duplicate, so no orphan inode is ever created. This works on TiDB
		// as well as MySQL because it does not rely on gap locking.
		nodeID := deterministicNodeID(dirPath)
		_, err := db.ExecContext(ctx, `INSERT IGNORE INTO inodes
			(`+s.scope.InsCols(`inode_id, size_bytes, revision, mode, status, created_at, mtime`)+`)
			VALUES (`+s.scope.InsVals(`?, 0, 1, ?, 'CONFIRMED', ?, ?`)+`)`,
			s.scope.Args(nodeID, 0o755, now, now)...)
		if err != nil {
			return nil, fmt.Errorf("ensure parent inode %s: %w", dirPath, err)
		}

		if selectErr == nil && !existingInodeID.Valid {
			// Directory row exists from pre-migration (inode_id was NULL).
			// Backfill the inode_id.
			_, err = db.ExecContext(ctx,
				`UPDATE file_nodes SET inode_id = ? WHERE `+s.scope.And(`path_hash = ? AND path = ? AND is_directory = 1`),
				append([]any{nodeID}, s.scope.Args(fileNodePathHash(dirPath), dirPath)...)...)
			if err != nil {
				return nil, fmt.Errorf("backfill parent inode_id %s: %w", dirPath, err)
			}
			continue
		}

		// Directory does not exist — insert the dentry.
		res, err := db.ExecContext(ctx, `INSERT INTO file_nodes
			(`+s.scope.InsCols(`node_id, path, path_hash, parent_path, parent_path_hash, name, is_directory, inode_id, created_at`)+`)
			VALUES (`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, 1, ?, ?`)+`)
			ON DUPLICATE KEY UPDATE node_id = node_id`,
			s.scope.Args(nodeID, dirPath, fileNodePathHash(dirPath), pp, fileNodePathHash(pp), name, nodeID, now)...)
		if err != nil && !isUniqueViolation(err) {
			return nil, fmt.Errorf("ensure parent %s: %w", dirPath, err)
		}
		if err == nil {
			if n, _ := res.RowsAffected(); n == 1 {
				created = append(created, nodeID)
			}
		}
	}
	return created, nil
}

func (s *Store) InsertNodeTx(db execer, n *FileNode) error {
	if isRootFileNode(*n) {
		return ErrInvalidRootDentry
	}
	if !n.IsDirectory && n.FileID != "" {
		if err := s.lockConfirmedFileForAttachTx(db, n.FileID); err != nil {
			return err
		}
	}
	_, err := db.Exec(`INSERT INTO file_nodes (`+s.scope.InsCols(`node_id, path, path_hash, parent_path, parent_path_hash, name, is_directory, file_id, inode_id, created_at`)+`)
		VALUES (`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, ?, ?, ?, ?`)+`)`,
		s.scope.Args(n.NodeID, n.Path, fileNodePathHash(n.Path), n.ParentPath, fileNodePathHash(n.ParentPath),
			n.Name, n.IsDirectory, nullStr(n.FileID), nullStr(n.InodeID), n.CreatedAt.UTC())...)
	if isUniqueViolation(err) {
		return ErrPathConflict
	}
	return err
}

func (s *Store) lockConfirmedFileForAttachTx(db execer, fileID string) error {
	var status string
	if s.useLegacyFiles {
		if err := db.QueryRow(`SELECT status FROM files WHERE file_id = ? FOR UPDATE`, fileID).Scan(&status); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return err
		}
		if FileStatus(status) != StatusConfirmed {
			return ErrNotFound
		}
	}
	if err := db.QueryRow(`SELECT status FROM inodes WHERE `+s.scope.And(`inode_id = ?`)+` FOR UPDATE`, s.scope.Args(fileID)...).Scan(&status); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return err
	}
	if FileStatus(status) != StatusConfirmed {
		return ErrNotFound
	}
	return nil
}

func (s *Store) MarkFileDeleted(ctx context.Context, fileID string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "mark_file_deleted", start, &err)

	if s.useLegacyFiles {
		_, err = s.db.ExecContext(ctx, `UPDATE files SET status = 'DELETED' WHERE file_id = ?`, fileID)
		if err != nil {
			return err
		}
	}
	_, err = s.db.ExecContext(ctx, `UPDATE inodes SET status = 'DELETED' WHERE `+s.scope.And(`inode_id = ?`), s.scope.Args(fileID)...)
	return err
}

// ConfirmedStorageBytesTx returns the total bytes occupied by confirmed file
// entities in the current tenant database.
func (s *Store) ConfirmedStorageBytesTx(db execer) (int64, error) {
	var total sql.NullInt64
	if err := db.QueryRow(`SELECT COALESCE(SUM(size_bytes), 0) FROM inodes WHERE `+s.scope.And(`status = 'CONFIRMED'`), s.scope.Args()...).Scan(&total); err != nil {
		return 0, err
	}
	return total.Int64, nil
}

// ConfirmedMediaFileCountTx returns the number of confirmed files whose
// content_type starts with "image/" or "audio/". This count drives the
// per-tenant media LLM budget gate: files that exceed the quota are still
// stored but their LLM extraction tasks (img_extract_text, audio_extract_text)
// are not enqueued.
func (s *Store) ConfirmedMediaFileCountTx(db execer) (int64, error) {
	var count sql.NullInt64
	if err := db.QueryRow(`SELECT COUNT(*) FROM inodes i JOIN contents c ON i.inode_id = c.inode_id WHERE `+scopeWhereAnd(s.scope, `i.status = 'CONFIRMED' AND (c.content_type LIKE 'image/%' OR c.content_type LIKE 'audio/%')`, "i", "c"), scopeWhereArgs(s.scope, 2)...).Scan(&count); err != nil {
		return 0, err
	}
	return count.Int64, nil
}

// ActiveUploadReservedBytesTx returns the additional bytes reserved by active
// multipart uploads beyond what is already counted by the confirmed file set.
func (s *Store) ActiveUploadReservedBytesTx(db execer) (int64, error) {
	var total sql.NullInt64
	// TODO: This aggregation joins uploads -> file_nodes -> files on every quota
	// check. If upload concurrency grows, re-evaluate the access path and add a
	// more targeted uploads status/index strategy or a pre-aggregated quota state.
	//
	// In shared shape the fs_id predicate on the uploads driving alias is not
	// enough: target paths are not globally unique, so the file_nodes join must
	// also carry fs_id in its ON clause or one tenant's upload could join
	// another tenant's dentry at the same path and misread its size. The
	// inodes join needs the same guard: inode ids are unique only within a
	// tenant in shared shape, so an unscoped join could read another tenant's
	// confirmed size and understate this tenant's reservation.
	err := db.QueryRow(`SELECT COALESCE(SUM(
		CASE
			WHEN u.total_size > COALESCE(i.size_bytes, 0) THEN u.total_size - COALESCE(i.size_bytes, 0)
			ELSE 0
		END
	), 0)
		FROM uploads u
		LEFT JOIN file_nodes fn ON `+s.scope.AndAs("fn", `fn.path_hash = u.target_path_hash AND fn.path = u.target_path`)+`
		LEFT JOIN inodes i ON `+s.scope.AndOn(`i.inode_id = COALESCE(fn.inode_id, fn.file_id) AND i.status = 'CONFIRMED'`, "i")+`
		WHERE `+s.scope.AndAs("u", `u.target_kind = 'file' AND u.status IN ('INITIATED', 'UPLOADING') AND u.expires_at > ?`),
		append(append(s.scope.Args(), s.scope.Args()...), s.scope.Args(time.Now().UTC())...)...).Scan(&total)
	if err != nil {
		return 0, err
	}
	return total.Int64, nil
}

// ConfirmedFileSizeByPathTx returns the current confirmed file size at path.
// Missing paths and directories report zero bytes.
func (s *Store) ConfirmedFileSizeByPathTx(db execer, path string) (int64, error) {
	var size sql.NullInt64
	err := db.QueryRow(`SELECT i.size_bytes
		FROM file_nodes fn
		JOIN inodes i ON i.inode_id = COALESCE(fn.inode_id, fn.file_id)
		WHERE `+scopeWhereAnd(s.scope, `fn.path_hash = ? AND fn.path = ? AND fn.is_directory = 0 AND i.status = 'CONFIRMED'`, "fn", "i")+`
		LIMIT 1`, scopeWhereArgs(s.scope, 2, fileNodePathHash(path), path)...).Scan(&size)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	return size.Int64, nil
}

// ConfirmedFileSummary holds the minimal per-file info needed for quota reconciliation.
type ConfirmedFileSummary struct {
	FileID      string
	SizeBytes   int64
	ContentType string
}

// ConfirmedFileStorageTier is the minimal tenant-DB state needed to classify
// one live file in central billing metadata.
type ConfirmedFileStorageTier struct {
	FileID      string
	SizeBytes   int64
	IsMedia     bool
	StorageType StorageType
	LayerOwned  bool
}

// ListConfirmedFileStorageTiers keyset-paginates live core files.
func (s *Store) ListConfirmedFileStorageTiers(ctx context.Context, cursor string, limit int) ([]ConfirmedFileStorageTier, string, error) {
	if limit <= 0 {
		limit = 500
	}
	predicate := `i.status = 'CONFIRMED'`
	args := []any{limit}
	if cursor != "" {
		predicate += ` AND i.inode_id > ?`
		args = []any{cursor, limit}
	}
	rows, err := s.db.QueryContext(ctx,
		`SELECT i.inode_id, i.size_bytes, COALESCE(c.content_type, ''), c.storage_type,
				EXISTS (SELECT 1 FROM fs_layer_entries le WHERE `+s.scope.AndAs("le", `le.op = 'upsert' AND le.kind = 'file' AND le.storage_type = 's3' AND le.storage_ref_hash = c.storage_ref_hash AND le.storage_ref = c.storage_ref`)+`) AS layer_owned
			 FROM inodes i JOIN contents c ON i.inode_id = c.inode_id
			 WHERE `+scopeWhereAnd(s.scope, predicate, "i", "c")+`
			 ORDER BY i.inode_id ASC LIMIT ?`, scopeWhereArgs(s.scope, 3, args...)...)
	if err != nil {
		return nil, "", fmt.Errorf("query confirmed file storage tiers: %w", err)
	}
	defer func() { _ = rows.Close() }()
	var out []ConfirmedFileStorageTier
	for rows.Next() {
		var item ConfirmedFileStorageTier
		var contentType string
		if err := rows.Scan(&item.FileID, &item.SizeBytes, &contentType, &item.StorageType, &item.LayerOwned); err != nil {
			return nil, "", fmt.Errorf("scan confirmed file storage tier: %w", err)
		}
		item.IsMedia = strings.HasPrefix(contentType, "image/") || strings.HasPrefix(contentType, "audio/")
		out = append(out, item)
	}
	if err := rows.Err(); err != nil {
		return nil, "", err
	}
	if len(out) == limit {
		return out, out[len(out)-1].FileID, nil
	}
	return out, "", nil
}

// FSLayerStorageTierCursor identifies one immutable fs-layer entry payload.
type FSLayerStorageTierCursor struct {
	LayerID  string
	PathHash string
	EntrySeq int64
}

// FSLayerStorageTier is one physically persisted file payload version. Layer
// entries are append-only, so every matching row consumes storage even when a
// newer version at the same path is visible.
type FSLayerStorageTier struct {
	FSLayerStorageTierCursor
	SizeBytes   int64
	StorageType StorageType
}

// ListFSLayerStorageTiers keyset-paginates all physical fs-layer file-upsert
// payloads. Metadata-only operations do not own bytes and are excluded.
func (s *Store) ListFSLayerStorageTiers(ctx context.Context, cursor FSLayerStorageTierCursor, limit int) ([]FSLayerStorageTier, FSLayerStorageTierCursor, error) {
	if limit <= 0 {
		limit = 500
	}
	predicate := `op = 'upsert' AND kind = 'file'`
	args := []any{limit}
	if cursor != (FSLayerStorageTierCursor{}) {
		predicate += ` AND (layer_id > ? OR (layer_id = ? AND path_hash > ?) OR (layer_id = ? AND path_hash = ? AND entry_seq > ?))`
		args = []any{cursor.LayerID, cursor.LayerID, cursor.PathHash, cursor.LayerID, cursor.PathHash, cursor.EntrySeq, limit}
	}
	rows, err := s.db.QueryContext(ctx,
		`SELECT layer_id, path_hash, entry_seq, size_bytes,
			CASE WHEN storage_type = '' THEN 'db9' ELSE storage_type END
		 FROM fs_layer_entries
		 WHERE `+s.scope.And(predicate)+`
		 ORDER BY layer_id ASC, path_hash ASC, entry_seq ASC LIMIT ?`, s.scope.Args(args...)...)
	if err != nil {
		return nil, FSLayerStorageTierCursor{}, fmt.Errorf("query fs-layer storage tiers: %w", err)
	}
	defer func() { _ = rows.Close() }()
	var out []FSLayerStorageTier
	for rows.Next() {
		var item FSLayerStorageTier
		if err := rows.Scan(&item.LayerID, &item.PathHash, &item.EntrySeq, &item.SizeBytes, &item.StorageType); err != nil {
			return nil, FSLayerStorageTierCursor{}, fmt.Errorf("scan fs-layer storage tier: %w", err)
		}
		out = append(out, item)
	}
	if err := rows.Err(); err != nil {
		return nil, FSLayerStorageTierCursor{}, err
	}
	if len(out) == limit {
		return out, out[len(out)-1].FSLayerStorageTierCursor, nil
	}
	return out, FSLayerStorageTierCursor{}, nil
}

// ListConfirmedFileSummaries returns confirmed files in cursor-paginated batches,
// ordered by file_id. Pass cursor="" for the first page. Returns an empty cursor
// when there are no more rows.
func (s *Store) ListConfirmedFileSummaries(ctx context.Context, cursor string, limit int) ([]ConfirmedFileSummary, string, error) {
	if limit <= 0 {
		limit = 500
	}
	var rows *sql.Rows
	var err error
	if cursor == "" {
		rows, err = s.db.QueryContext(ctx,
			`SELECT i.inode_id, i.size_bytes, COALESCE(c.content_type, '')
			 FROM inodes i JOIN contents c ON i.inode_id = c.inode_id
			 WHERE `+scopeWhereAnd(s.scope, `i.status = 'CONFIRMED'`, "i", "c")+`
			 ORDER BY i.inode_id ASC LIMIT ?`, scopeWhereArgs(s.scope, 2, limit)...)
	} else {
		rows, err = s.db.QueryContext(ctx,
			`SELECT i.inode_id, i.size_bytes, COALESCE(c.content_type, '')
			 FROM inodes i JOIN contents c ON i.inode_id = c.inode_id
			 WHERE `+scopeWhereAnd(s.scope, `i.status = 'CONFIRMED' AND i.inode_id > ?`, "i", "c")+`
			 ORDER BY i.inode_id ASC LIMIT ?`, scopeWhereArgs(s.scope, 2, cursor, limit)...)
	}
	if err != nil {
		return nil, "", fmt.Errorf("query confirmed files: %w", err)
	}
	defer func() { _ = rows.Close() }()

	var out []ConfirmedFileSummary
	for rows.Next() {
		var f ConfirmedFileSummary
		if err := rows.Scan(&f.FileID, &f.SizeBytes, &f.ContentType); err != nil {
			return nil, "", fmt.Errorf("scan confirmed file: %w", err)
		}
		out = append(out, f)
	}
	if err := rows.Err(); err != nil {
		return nil, "", err
	}
	nextCursor := ""
	if len(out) == limit {
		nextCursor = out[len(out)-1].FileID
	}
	return out, nextCursor, nil
}

type ConfirmedS3Ref struct {
	StorageRef     string
	StorageRefHash string
	ContentLayout  ContentLayout
}

func (s *Store) ListConfirmedS3Refs(ctx context.Context, cursor string, limit int) ([]ConfirmedS3Ref, string, error) {
	if limit <= 0 {
		limit = 500
	}
	var rows *sql.Rows
	var err error
	if cursor == "" {
		rows, err = s.db.QueryContext(ctx,
			`SELECT c.storage_ref, c.storage_ref_hash, c.content_layout
			 FROM contents c JOIN inodes i ON i.inode_id = c.inode_id
			 WHERE `+scopeWhereAnd(s.scope, `i.status = 'CONFIRMED' AND c.storage_type = 's3' AND c.storage_ref <> ''`, "c", "i")+`
			 ORDER BY c.storage_ref ASC LIMIT ?`, scopeWhereArgs(s.scope, 2, limit)...)
	} else {
		rows, err = s.db.QueryContext(ctx,
			`SELECT c.storage_ref, c.storage_ref_hash, c.content_layout
			 FROM contents c JOIN inodes i ON i.inode_id = c.inode_id
			 WHERE `+scopeWhereAnd(s.scope, `i.status = 'CONFIRMED' AND c.storage_type = 's3' AND c.storage_ref <> '' AND c.storage_ref > ?`, "c", "i")+`
			 ORDER BY c.storage_ref ASC LIMIT ?`, scopeWhereArgs(s.scope, 2, cursor, limit)...)
	}
	if err != nil {
		return nil, "", fmt.Errorf("query confirmed s3 refs: %w", err)
	}
	defer func() { _ = rows.Close() }()

	out := make([]ConfirmedS3Ref, 0)
	for rows.Next() {
		var ref ConfirmedS3Ref
		if err := rows.Scan(&ref.StorageRef, &ref.StorageRefHash, &ref.ContentLayout); err != nil {
			return nil, "", fmt.Errorf("scan confirmed s3 ref: %w", err)
		}
		if ref.StorageRefHash == "" {
			ref.StorageRefHash = StorageRefHash(ref.StorageRef)
		}
		out = append(out, ref)
	}
	if err := rows.Err(); err != nil {
		return nil, "", err
	}
	nextCursor := ""
	if len(out) == limit {
		nextCursor = out[len(out)-1].StorageRef
	}
	return out, nextCursor, nil
}

func (s *Store) HasActiveUploads(ctx context.Context) (bool, error) {
	var one int
	err := s.db.QueryRowContext(ctx,
		`SELECT 1 FROM uploads
		 WHERE `+s.scope.And(`status IN ('INITIATED', 'UPLOADING') AND expires_at > ?`)+`
		 LIMIT 1`, s.scope.Args(time.Now().UTC())...).Scan(&one)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return true, nil
}

func (s *Store) SanitizeForkRuntimeState(ctx context.Context) error {
	return s.InTx(ctx, func(tx *sql.Tx) error {
		if _, err := tx.ExecContext(ctx, `DELETE fn FROM file_nodes fn JOIN inodes i ON i.inode_id = COALESCE(fn.inode_id, fn.file_id) WHERE `+scopeWhereAnd(s.scope, `i.status <> 'CONFIRMED'`, "fn", "i"), scopeWhereArgs(s.scope, 2)...); err != nil {
			return err
		}
		now := time.Now().UTC()
		if _, err := tx.ExecContext(ctx, `UPDATE contents c JOIN inodes i ON i.inode_id = c.inode_id SET c.storage_ref = '', c.storage_ref_hash = '', c.content_blob = NULL WHERE `+scopeWhereAnd(s.scope, `i.status <> 'CONFIRMED'`, "c", "i"), scopeWhereArgs(s.scope, 2)...); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `UPDATE inodes SET status = 'DELETED', expires_at = ? WHERE `+s.scope.And(`status <> 'CONFIRMED'`), append([]any{now}, s.scope.Args()...)...); err != nil {
			return err
		}
		if s.useLegacyFiles {
			if _, err := tx.ExecContext(ctx, `UPDATE files SET status = 'DELETED', storage_ref = '', storage_ref_hash = '', content_blob = NULL, expires_at = ? WHERE status <> 'CONFIRMED'`, now); err != nil {
				return err
			}
		}
		// Whole-table wipes. In shared shape every one of these tables carries
		// an fs_id column (vault tables included), so the DELETE must be
		// restricted to this tenant's rows — unscoped it would destroy other
		// tenants' runtime state. llm_usage is standalone-only (the central
		// meta DB ledger is authoritative in shared deployments) and is wiped
		// only there; a leftover standalone llm_usage table on a shared DB has
		// no fs_id column to restrict by, so it must not be touched.
		wipeTables := []string{"uploads", "file_gc_tasks", "semantic_tasks"}
		if !s.scope.Shared() {
			wipeTables = append(wipeTables, "llm_usage")
		}
		wipeTables = append(wipeTables,
			"vault_audit_log",
			"vault_grants",
			"vault_tokens",
			"vault_secret_fields",
			"vault_secrets",
			"vault_deks",
		)
		for _, tbl := range wipeTables {
			stmt := `DELETE FROM ` + tbl
			args := []any(nil)
			if s.scope.Shared() {
				stmt += ` WHERE fs_id = ?`
				args = s.scope.Args()
			}
			if _, err := tx.ExecContext(ctx, stmt, args...); err != nil {
				if isMissingTableError(err) {
					continue
				}
				return err
			}
		}
		return nil
	})
}

func isMissingTableError(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "error 1146") ||
		strings.Contains(msg, "table") && strings.Contains(msg, "doesn't exist")
}

// IsMissingTableError reports whether err is a MySQL 1146 ("table doesn't
// exist") failure. Callers use it to distinguish pre-migration tenants (the
// table was never created — a benign, expected state) from real connectivity
// or syntax failures.
func IsMissingTableError(err error) bool {
	return isMissingTableError(err)
}

// --- composite operations ---

func (s *Store) Stat(ctx context.Context, path string) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat", start, &err)

	node, err := s.GetNode(ctx, path)
	if err != nil {
		return nil, err
	}
	nf := &NodeWithFile{Node: *node}
	if !node.IsDirectory && node.FileID != "" {
		f, err := s.GetFile(ctx, node.FileID)
		if err != nil {
			return nil, err
		}
		nf.File = f
	}
	out = nf
	return out, nil
}

// StatTx is like Stat but reads through an existing transaction.
func (s *Store) StatTx(ctx context.Context, db execer, path string) (out *NodeWithFile, err error) {
	return s.statTx(ctx, db, path, false)
}

// StatForUpdateTx is StatTx with FOR UPDATE on the dentry (and joined inode).
func (s *Store) StatForUpdateTx(ctx context.Context, db execer, path string) (out *NodeWithFile, err error) {
	return s.statTx(ctx, db, path, true)
}

// LockDentryFileIDTx locks the file_nodes row for path inside an existing
// transaction and returns the dentry's current file identity
// (COALESCE(inode_id, file_id)). It returns ErrNotFound when the dentry does
// not exist. Callers use it to fence path→inode resolution races: a dentry
// deleted and recreated after resolution points at a different identity.
func (s *Store) LockDentryFileIDTx(db execer, path string) (string, error) {
	var id sql.NullString
	err := db.QueryRow(`SELECT COALESCE(inode_id, file_id) FROM file_nodes WHERE `+
		s.scope.And(`path_hash = ? AND path = ?`)+` FOR UPDATE`,
		s.scope.Args(fileNodePathHash(path), path)...).Scan(&id)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return "", ErrNotFound
		}
		return "", err
	}
	if !id.Valid || id.String == "" {
		return "", ErrNotFound
	}
	return id.String, nil
}

func (s *Store) statTx(ctx context.Context, db execer, path string, forUpdate bool) (out *NodeWithFile, err error) {
	query := `SELECT fn.node_id, fn.path, fn.parent_path, fn.name, fn.is_directory, fn.file_id, fn.created_at,
		i.inode_id, c.storage_type, c.content_layout, c.storage_ref, c.storage_encryption_mode, c.storage_encryption_key_id,
		c.content_blob, c.content_type, i.size_bytes,
		c.checksum_sha256, i.revision, i.mode, s.embedding_revision, i.status, c.source_id, s.content_text,
		s.description, s.description_embedding_revision, i.created_at, i.confirmed_at, i.expires_at
		FROM file_nodes fn
		LEFT JOIN inodes i ON ` + s.scope.AndOn(`COALESCE(fn.inode_id, fn.file_id) = i.inode_id AND i.status = 'CONFIRMED'`, "i") + `
		LEFT JOIN contents c ON ` + s.scope.AndOn(`i.inode_id = c.inode_id`, "c") + `
		LEFT JOIN semantic s ON ` + s.scope.AndOn(`i.inode_id = s.inode_id`, "s") + `
		WHERE ` + s.scope.AndAs("fn", `fn.path_hash = ? AND fn.path = ?`) + `
		LIMIT 1`
	if forUpdate {
		query += ` FOR UPDATE`
	}
	row := db.QueryRowContext(ctx, query, scopeWhereArgs(s.scope, 3, s.scope.Args(fileNodePathHash(path), path)...)...)
	return scanNodeWithFileWithBlob(row)
}

func (s *Store) StatPathFallback(ctx context.Context, primaryPath, fallbackPath string) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat_path_fallback", start, &err)

	out, err = statWithFallback(ctx, primaryPath, fallbackPath, func(ctx context.Context, path string) (*NodeWithFile, error) {
		return s.statTx(ctx, s.db, path, false)
	})
	return out, err
}

func statWithFallback(
	ctx context.Context,
	primaryPath, fallbackPath string,
	stat func(context.Context, string) (*NodeWithFile, error),
) (*NodeWithFile, error) {
	// These are deliberately separate autocommit point lookups so shared TiDB
	// can use (fs_id, path_hash). A fallback costs one extra round trip and the
	// pair is not a single snapshot; callers only use it for path-form resolution.
	out, err := stat(ctx, primaryPath)
	if err == nil || !errors.Is(err, ErrNotFound) || primaryPath == fallbackPath {
		return out, err
	}
	return stat(ctx, fallbackPath)
}

// StatMetadata fetches the fields needed by the enriched stat metadata API
// without loading file content or unrelated storage and semantic fields.
func (s *Store) StatMetadata(ctx context.Context, path string) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat_metadata", start, &err)

	out, err = s.statMetadata(ctx, path)
	if err == nil && out != nil && out.Node.IsDirectory {
		// Match Stat: explicit directory lookups expose dentry metadata only.
		out.File = nil
	}
	return out, err
}

// StatPathFallbackMetadata is like StatMetadata with primary/fallback path resolution;
// a directory found through fallback keeps its inode metadata attached as File.
func (s *Store) StatPathFallbackMetadata(ctx context.Context, primaryPath, fallbackPath string) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat_path_fallback_metadata", start, &err)

	out, err = statWithFallback(ctx, primaryPath, fallbackPath, s.statMetadata)
	return out, err
}

func (s *Store) statMetadata(ctx context.Context, path string) (*NodeWithFile, error) {
	row := s.db.QueryRowContext(ctx, `SELECT fn.node_id, fn.path, fn.parent_path, fn.name, fn.is_directory, fn.file_id, fn.created_at,
		i.inode_id, i.size_bytes, i.revision, i.created_at, i.confirmed_at, c.content_type, s.content_text, s.description
		FROM file_nodes fn
		LEFT JOIN inodes i ON `+s.scope.AndOn(`COALESCE(fn.inode_id, fn.file_id) = i.inode_id AND i.status = 'CONFIRMED'`, "i")+`
		LEFT JOIN contents c ON `+s.scope.AndOn(`i.inode_id = c.inode_id`, "c")+`
		LEFT JOIN semantic s ON `+s.scope.AndOn(`i.inode_id = s.inode_id`, "s")+`
		WHERE `+s.scope.AndAs("fn", `fn.path_hash = ? AND fn.path = ?`)+`
		LIMIT 1`, scopeWhereArgs(s.scope, 3, s.scope.Args(fileNodePathHash(path), path)...)...)
	return scanNodeWithFileMetadata(row)
}

// StatPathFallbackLite is like StatPathFallback but only fetches lightweight
// metadata fields needed for FUSE stat/HEAD operations.
func (s *Store) StatPathFallbackLite(ctx context.Context, primaryPath, fallbackPath string) (out *NodeWithFile, err error) {
	return s.statPathFallbackLite(ctx, primaryPath, fallbackPath, false)
}

// StatPathFallbackLiteWithChecksum includes persisted SHA-256 without loading blobs, text, or embeddings.
func (s *Store) StatPathFallbackLiteWithChecksum(ctx context.Context, primaryPath, fallbackPath string) (out *NodeWithFile, err error) {
	return s.statPathFallbackLite(ctx, primaryPath, fallbackPath, true)
}

func (s *Store) statPathFallbackLite(ctx context.Context, primaryPath, fallbackPath string, includeChecksum bool) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat_path_fallback_lite", start, &err)

	// Keep HEAD/FUSE on one round trip: this projection is already narrow and
	// the production slow-query evidence targeted the full enriched-stat join.
	checksumColumn := ""
	if includeChecksum {
		checksumColumn = ", c.checksum_sha256"
	}
	row := s.db.QueryRowContext(ctx, `SELECT fn.node_id, fn.path, fn.parent_path, fn.name, fn.is_directory, fn.file_id, fn.created_at,
			i.inode_id, i.size_bytes, i.revision, i.mode, i.status, i.created_at, i.confirmed_at, c.storage_type, c.content_layout`+checksumColumn+`
			FROM file_nodes fn
		LEFT JOIN inodes i ON `+s.scope.AndOn(`COALESCE(fn.inode_id, fn.file_id) = i.inode_id AND i.status = 'CONFIRMED'`, "i")+`
		LEFT JOIN contents c ON `+s.scope.AndOn(`i.inode_id = c.inode_id`, "c")+`
		WHERE `+s.andAsGrouped("fn", `(fn.path_hash = ? AND fn.path = ?) OR (fn.path_hash = ? AND fn.path = ?)`)+`
		ORDER BY CASE WHEN fn.path = ? THEN 0 ELSE 1 END
		LIMIT 1`, scopeWhereArgs(s.scope, 2, s.scope.Args(fileNodePathHash(primaryPath), primaryPath, fileNodePathHash(fallbackPath), fallbackPath, primaryPath)...)...)
	out, err = scanNodeWithFileLite(row, includeChecksum)
	return out, err
}

// StatLite is like Stat but only fetches lightweight metadata fields.
func (s *Store) StatLite(ctx context.Context, path string) (out *NodeWithFile, err error) {
	return s.statLite(ctx, path, false)
}

// StatLiteWithChecksum includes persisted SHA-256 without loading blobs, text, descriptions, or embeddings.
func (s *Store) StatLiteWithChecksum(ctx context.Context, path string) (out *NodeWithFile, err error) {
	return s.statLite(ctx, path, true)
}

func (s *Store) statLite(ctx context.Context, path string, includeChecksum bool) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat_lite", start, &err)

	checksumColumn := ""
	if includeChecksum {
		checksumColumn = ", c.checksum_sha256"
	}
	row := s.db.QueryRowContext(ctx, `SELECT fn.node_id, fn.path, fn.parent_path, fn.name, fn.is_directory, fn.file_id, fn.created_at,
			i.inode_id, i.size_bytes, i.revision, i.mode, i.status, i.created_at, i.confirmed_at, c.storage_type, c.content_layout`+checksumColumn+`
			FROM file_nodes fn
		LEFT JOIN inodes i ON `+s.scope.AndOn(`COALESCE(fn.inode_id, fn.file_id) = i.inode_id AND i.status = 'CONFIRMED'`, "i")+`
		LEFT JOIN contents c ON `+s.scope.AndOn(`i.inode_id = c.inode_id`, "c")+`
		WHERE `+s.scope.AndAs("fn", `fn.path_hash = ? AND fn.path = ?`)+`
		LIMIT 1`, scopeWhereArgs(s.scope, 2, s.scope.Args(fileNodePathHash(path), path)...)...)
	out, err = scanNodeWithFileLite(row, includeChecksum)
	return out, err
}

// StatForRead fetches the minimal fields needed to serve a GET request in a
// single pass: storage_type, storage_ref, content_blob (for db9), size_bytes,
// revision, and confirmed_at. This avoids the two-stat pattern of
// PresignGetObject-then-ReadCtx.
func (s *Store) StatForRead(ctx context.Context, path string) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat_for_read", start, &err)

	row := s.db.QueryRowContext(ctx, `SELECT fn.node_id, fn.path, fn.parent_path, fn.name, fn.is_directory, fn.file_id, fn.created_at,
		i.inode_id, c.storage_type, c.content_layout, c.storage_ref, c.content_blob, i.size_bytes, i.revision, i.mode, i.status, i.created_at, i.confirmed_at
		FROM file_nodes fn
		LEFT JOIN inodes i ON `+s.scope.AndOn(`COALESCE(fn.inode_id, fn.file_id) = i.inode_id AND i.status = 'CONFIRMED'`, "i")+`
		LEFT JOIN contents c ON `+s.scope.AndOn(`i.inode_id = c.inode_id`, "c")+`
		WHERE `+s.scope.AndAs("fn", `fn.path_hash = ? AND fn.path = ?`)+`
		LIMIT 1`, scopeWhereArgs(s.scope, 2, s.scope.Args(fileNodePathHash(path), path)...)...)
	out, err = scanNodeWithFileForRead(row)
	return out, err
}

// StatPathFallbackForRead is like StatForRead but with primary/fallback path resolution.
func (s *Store) StatPathFallbackForRead(ctx context.Context, primaryPath, fallbackPath string) (out *NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "stat_path_fallback_for_read", start, &err)

	row := s.db.QueryRowContext(ctx, `SELECT fn.node_id, fn.path, fn.parent_path, fn.name, fn.is_directory, fn.file_id, fn.created_at,
		i.inode_id, c.storage_type, c.content_layout, c.storage_ref, c.content_blob, i.size_bytes, i.revision, i.mode, i.status, i.created_at, i.confirmed_at
		FROM file_nodes fn
		LEFT JOIN inodes i ON `+s.scope.AndOn(`COALESCE(fn.inode_id, fn.file_id) = i.inode_id AND i.status = 'CONFIRMED'`, "i")+`
		LEFT JOIN contents c ON `+s.scope.AndOn(`i.inode_id = c.inode_id`, "c")+`
		WHERE `+s.andAsGrouped("fn", `(fn.path_hash = ? AND fn.path = ?) OR (fn.path_hash = ? AND fn.path = ?)`)+`
		ORDER BY CASE WHEN fn.path = ? THEN 0 ELSE 1 END
		LIMIT 1`, scopeWhereArgs(s.scope, 2, s.scope.Args(fileNodePathHash(primaryPath), primaryPath, fileNodePathHash(fallbackPath), fallbackPath, primaryPath)...)...)
	out, err = scanNodeWithFileForRead(row)
	return out, err
}

func (s *Store) ListDir(ctx context.Context, parentPath string) (out []*NodeWithFile, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "list_dir", start, &err)

	// TODO(#110): ReadDir only needs lightweight file metadata. Split this into a
	// metadata-only listing path so directory scans do not fetch or copy content_blob.
	q := `SELECT fn.node_id, fn.path, fn.parent_path, fn.name, fn.is_directory, fn.file_id, fn.created_at,
		i.inode_id, i.size_bytes, i.revision, i.mode, i.status, i.created_at, i.confirmed_at,
		s.embedding_revision
		FROM file_nodes fn
		LEFT JOIN inodes i ON ` + s.scope.AndOn(`COALESCE(fn.inode_id, fn.file_id) = i.inode_id AND i.status = 'CONFIRMED'`, "i") + `
		LEFT JOIN semantic s ON ` + s.scope.AndOn(`i.inode_id = s.inode_id`, "s") + `
		WHERE ` + s.scope.AndAs("fn", `fn.parent_path_hash = ? AND fn.parent_path = ?`) + `
		ORDER BY fn.name`
	rows, err := s.db.QueryContext(ctx, q, scopeWhereArgs(s.scope, 2, s.scope.Args(fileNodePathHash(parentPath), parentPath)...)...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	result := make([]*NodeWithFile, 0)
	for rows.Next() {
		var n FileNode
		var isDir int
		var nodeFileID sql.NullString
		var nodeCreatedAt time.Time
		var fFileID sql.NullString
		var fSizeBytes, fRevision sql.NullInt64
		var fMode sql.NullInt64
		var fStatus sql.NullString
		var fCreatedAt, fConfirmedAt sql.NullTime
		var fEmbeddingRevision sql.NullInt64
		if err := rows.Scan(&n.NodeID, &n.Path, &n.ParentPath, &n.Name, &isDir, &nodeFileID, &nodeCreatedAt,
			&fFileID, &fSizeBytes, &fRevision, &fMode, &fStatus, &fCreatedAt, &fConfirmedAt, &fEmbeddingRevision); err != nil {
			return nil, err
		}
		n.IsDirectory = isDir != 0
		n.FileID = nodeFileID.String
		n.CreatedAt = nodeCreatedAt.UTC()
		if isRootFileNode(n) {
			continue
		}
		nf := &NodeWithFile{Node: n}
		if fFileID.Valid {
			nf.File = &File{
				FileID:    fFileID.String,
				SizeBytes: fSizeBytes.Int64,
				Revision:  fRevision.Int64,
				Mode:      uint32(fMode.Int64),
				Status:    FileStatus(fStatus.String),
			}
			if fCreatedAt.Valid {
				nf.File.CreatedAt = fCreatedAt.Time.UTC()
			}
			if fConfirmedAt.Valid {
				t := fConfirmedAt.Time.UTC()
				nf.File.ConfirmedAt = &t
			}
			if fEmbeddingRevision.Valid {
				rev := fEmbeddingRevision.Int64
				nf.File.EmbeddingRevision = &rev
			}
		}
		if fMode.Valid {
			nf.Mode = uint32(fMode.Int64)
			nf.HasMode = true
		}
		result = append(result, nf)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	out = result
	return out, nil
}

func (s *Store) DeleteFileWithRefCheck(ctx context.Context, path string) (out *File, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "delete_file_with_ref_check", start, &err)

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()

	candidate, err := s.scanDeleteCandidateTx(ctx, tx, path)
	if err != nil {
		return nil, err
	}

	if candidate.isDir {
		return nil, ErrNotFound
	}

	if _, err := tx.Exec(`DELETE FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ?`),
		s.scope.Args(fileNodePathHash(path), path)...); err != nil {
		return nil, err
	}

	if candidate.fileID == "" {
		return nil, tx.Commit()
	}
	if err := s.lockFileIDsForDeleteTx(ctx, tx, []string{candidate.fileID}); err != nil {
		return nil, err
	}

	var count int64
	err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM file_nodes WHERE `+s.scope.And(`file_id = ?`)+` FOR UPDATE`, s.scope.Args(candidate.fileID)...).Scan(&count)
	if err != nil {
		return nil, err
	}

	if count > 0 {
		return nil, tx.Commit()
	}

	if candidate.file == nil {
		return nil, ErrNotFound
	}

	if err := s.markFilesDeletedTx(ctx, tx, []string{candidate.fileID}); err != nil {
		return nil, err
	}
	if err := s.deleteFileTagsByIDsTx(ctx, tx, []string{candidate.fileID}); err != nil {
		return nil, err
	}

	candidate.file.Status = StatusDeleted
	task, err := NewFileGCTaskFromFile(candidate.file, time.Now().UTC())
	if err != nil {
		return nil, err
	}
	if _, err := s.EnqueueFileGCTaskTx(tx, task); err != nil {
		return nil, err
	}

	if err := tx.Commit(); err != nil {
		return nil, err
	}
	out = candidate.file
	return out, nil
}

// DeleteFileIfFileID deletes the file dentry at path only if it still names
// expectedFileID. A mismatch or missing path is a no-op (the occupant is left).
func (s *Store) DeleteFileIfFileID(ctx context.Context, path, expectedFileID string) (out *File, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "delete_file_if_file_id", start, &err)
	if expectedFileID == "" {
		return nil, nil
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()

	candidate, err := s.scanDeleteCandidateTx(ctx, tx, path)
	if err != nil {
		if errors.Is(err, ErrNotFound) {
			return nil, nil
		}
		return nil, err
	}
	if candidate.isDir || candidate.fileID != expectedFileID {
		return nil, tx.Commit()
	}

	res, err := tx.Exec(`DELETE FROM file_nodes WHERE `+s.scope.And(`path_hash = ? AND path = ? AND file_id = ?`),
		s.scope.Args(fileNodePathHash(path), path, expectedFileID)...)
	if err != nil {
		return nil, err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return nil, tx.Commit()
	}

	if err := s.lockFileIDsForDeleteTx(ctx, tx, []string{candidate.fileID}); err != nil {
		return nil, err
	}

	var count int64
	err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM file_nodes WHERE `+s.scope.And(`file_id = ?`)+` FOR UPDATE`, s.scope.Args(candidate.fileID)...).Scan(&count)
	if err != nil {
		return nil, err
	}
	if count > 0 {
		return nil, tx.Commit()
	}
	if candidate.file == nil {
		return nil, tx.Commit()
	}
	if err := s.markFilesDeletedTx(ctx, tx, []string{candidate.fileID}); err != nil {
		return nil, err
	}
	if err := s.deleteFileTagsByIDsTx(ctx, tx, []string{candidate.fileID}); err != nil {
		return nil, err
	}
	candidate.file.Status = StatusDeleted
	task, err := NewFileGCTaskFromFile(candidate.file, time.Now().UTC())
	if err != nil {
		return nil, err
	}
	if _, err := s.EnqueueFileGCTaskTx(tx, task); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	out = candidate.file
	return out, nil
}

type deleteCandidate struct {
	fileID string
	isDir  bool
	file   *File
}

const deleteFileIDBatchSize = 1000

func (s *Store) scanDeleteCandidateTx(ctx context.Context, tx *sql.Tx, path string) (*deleteCandidate, error) {
	if s.useLegacyFiles {
		row := tx.QueryRowContext(ctx, `SELECT fn.file_id, fn.is_directory,
			f.file_id, f.storage_type, f.content_layout, f.storage_ref, f.storage_encryption_mode,
			f.storage_encryption_key_id, f.content_type, f.size_bytes, f.checksum_sha256,
			f.revision, f.embedding_revision, f.status, f.source_id, f.created_at, f.confirmed_at, f.expires_at
			FROM file_nodes fn
			LEFT JOIN files f ON f.file_id = fn.file_id
			WHERE fn.path_hash = ? AND fn.path = ?
			FOR UPDATE`, fileNodePathHash(path), path)
		return scanLegacyDeleteCandidate(row)
	}
	row := tx.QueryRowContext(ctx, `SELECT fn.file_id, fn.is_directory,
		c.storage_type, c.content_layout, c.storage_ref, i.size_bytes, COALESCE(c.content_type, ''),
		s.embedding_revision
		FROM file_nodes fn
		LEFT JOIN inodes i ON `+s.scope.AndOn(`i.inode_id = fn.file_id`, "i")+`
		LEFT JOIN contents c ON `+s.scope.AndOn(`c.inode_id = fn.file_id`, "c")+`
		LEFT JOIN semantic s ON `+s.scope.AndOn(`s.inode_id = fn.file_id`, "s")+`
		WHERE `+s.scope.AndAs("fn", `fn.path_hash = ? AND fn.path = ?`)+`
		FOR UPDATE`, scopeWhereArgs(s.scope, 3, s.scope.Args(fileNodePathHash(path), path)...)...)
	return scanSplitDeleteCandidate(row)
}

func scanLegacyDeleteCandidate(row *sql.Row) (*deleteCandidate, error) {
	var nodeFileID sql.NullString
	var isDir bool
	var fFileID, storageType, contentLayout, storageRef, encMode, encKeyID sql.NullString
	var contentType, checksum, status, sourceID sql.NullString
	var sizeBytes, revision, embeddingRevision sql.NullInt64
	var createdAt, confirmedAt, expiresAt sql.NullTime
	err := row.Scan(&nodeFileID, &isDir,
		&fFileID, &storageType, &contentLayout, &storageRef, &encMode, &encKeyID,
		&contentType, &sizeBytes, &checksum, &revision, &embeddingRevision,
		&status, &sourceID, &createdAt, &confirmedAt, &expiresAt)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	c := &deleteCandidate{fileID: nodeFileID.String, isDir: isDir}
	if fFileID.Valid {
		c.file = fileFromLegacyGCScan(fFileID.String, storageType, contentLayout, storageRef, encMode, encKeyID,
			contentType, sizeBytes, checksum, revision, embeddingRevision, status,
			sourceID, createdAt, confirmedAt, expiresAt)
	}
	return c, nil
}

func scanSplitDeleteCandidate(row *sql.Row) (*deleteCandidate, error) {
	var nodeFileID sql.NullString
	var isDir bool
	var storageType, contentLayout, storageRef, contentType sql.NullString
	var sizeBytes, embeddingRevision sql.NullInt64
	err := row.Scan(&nodeFileID, &isDir, &storageType, &contentLayout, &storageRef, &sizeBytes, &contentType, &embeddingRevision)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	c := &deleteCandidate{fileID: nodeFileID.String, isDir: isDir}
	if nodeFileID.Valid && storageType.Valid {
		f := &File{
			FileID:        nodeFileID.String,
			StorageType:   StorageType(storageType.String),
			ContentLayout: contentLayoutForWrite(ContentLayout(contentLayout.String)),
			StorageRef:    storageRef.String,
			ContentType:   contentType.String,
			SizeBytes:     sizeBytes.Int64,
			Status:        StatusDeleted,
		}
		if embeddingRevision.Valid {
			rev := embeddingRevision.Int64
			f.EmbeddingRevision = &rev
		}
		c.file = f
	}
	return c, nil
}

func (s *Store) scanOrphanedFilesByIDTx(ctx context.Context, tx *sql.Tx, fileIDs []string) ([]*File, error) {
	if len(fileIDs) == 0 {
		return nil, nil
	}
	var out []*File
	for start := 0; start < len(fileIDs); start += deleteFileIDBatchSize {
		end := start + deleteFileIDBatchSize
		if end > len(fileIDs) {
			end = len(fileIDs)
		}
		batch := fileIDs[start:end]
		args := stringsToAny(batch)
		placeholders := questionPlaceholders(len(batch))
		var query string
		if s.useLegacyFiles {
			query = `SELECT f.file_id, f.storage_type, f.content_layout, f.storage_ref, f.storage_encryption_mode,
				f.storage_encryption_key_id, f.content_type, f.size_bytes, f.checksum_sha256,
				f.revision, f.embedding_revision, f.status, f.source_id, f.created_at, f.confirmed_at, f.expires_at
				FROM files f
				WHERE f.file_id IN (` + placeholders + `)
				  AND NOT EXISTS (SELECT 1 FROM file_nodes fn WHERE fn.file_id = f.file_id)`
		} else {
			query = `SELECT i.inode_id, c.storage_type, c.content_layout, c.storage_ref, i.size_bytes,
				COALESCE(c.content_type, ''), s.embedding_revision
				FROM inodes i
				JOIN contents c ON c.inode_id = i.inode_id
				LEFT JOIN semantic s ON ` + s.scope.AndOn(`s.inode_id = i.inode_id`, "s") + `
				WHERE ` + s.scope.AndAs("i", s.scope.AndAs("c", `i.inode_id IN (`+placeholders+`)
				  AND NOT EXISTS (SELECT 1 FROM file_nodes fn WHERE `+s.scope.AndAs("fn", `fn.file_id = i.inode_id`)+`)`))
			args = append(scopeWhereArgs(s.scope, 2, s.scope.Args(args...)...), s.scope.Args()...)
		}
		rows, err := tx.QueryContext(ctx, query, args...)
		if err != nil {
			return nil, err
		}
		for rows.Next() {
			var f *File
			if s.useLegacyFiles {
				var fileID, storageType, contentLayout, storageRef, encMode, encKeyID sql.NullString
				var contentType, checksum, status, sourceID sql.NullString
				var sizeBytes, revision, embeddingRevision sql.NullInt64
				var createdAt, confirmedAt, expiresAt sql.NullTime
				if err := rows.Scan(&fileID, &storageType, &contentLayout, &storageRef, &encMode, &encKeyID,
					&contentType, &sizeBytes, &checksum, &revision, &embeddingRevision,
					&status, &sourceID, &createdAt, &confirmedAt, &expiresAt); err != nil {
					_ = rows.Close()
					return nil, err
				}
				f = fileFromLegacyGCScan(fileID.String, storageType, contentLayout, storageRef, encMode, encKeyID,
					contentType, sizeBytes, checksum, revision, embeddingRevision, status,
					sourceID, createdAt, confirmedAt, expiresAt)
				f.Status = StatusDeleted
			} else {
				var fileID, storageType, contentLayout, storageRef, contentType sql.NullString
				var sizeBytes, embeddingRevision sql.NullInt64
				if err := rows.Scan(&fileID, &storageType, &contentLayout, &storageRef, &sizeBytes, &contentType, &embeddingRevision); err != nil {
					_ = rows.Close()
					return nil, err
				}
				f = &File{
					FileID:        fileID.String,
					StorageType:   StorageType(storageType.String),
					ContentLayout: contentLayoutForWrite(ContentLayout(contentLayout.String)),
					StorageRef:    storageRef.String,
					ContentType:   contentType.String,
					SizeBytes:     sizeBytes.Int64,
					Status:        StatusDeleted,
				}
				if embeddingRevision.Valid {
					rev := embeddingRevision.Int64
					f.EmbeddingRevision = &rev
				}
			}
			out = append(out, f)
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			return nil, err
		}
		if err := rows.Close(); err != nil {
			return nil, err
		}
	}
	return out, nil
}

func (s *Store) lockFileIDsForDeleteTx(ctx context.Context, tx *sql.Tx, fileIDs []string) error {
	for start := 0; start < len(fileIDs); start += deleteFileIDBatchSize {
		end := start + deleteFileIDBatchSize
		if end > len(fileIDs) {
			end = len(fileIDs)
		}
		batch := fileIDs[start:end]
		if len(batch) == 0 {
			continue
		}
		args := stringsToAny(batch)
		placeholders := questionPlaceholders(len(batch))
		if s.useLegacyFiles {
			rows, err := tx.QueryContext(ctx, `SELECT file_id FROM files WHERE file_id IN (`+placeholders+`) FOR UPDATE`, args...)
			if err != nil {
				return err
			}
			if err := closeRowsAfterScan(rows); err != nil {
				return err
			}
		}
		rows, err := tx.QueryContext(ctx, `SELECT inode_id FROM inodes WHERE `+s.scope.And(`inode_id IN (`+placeholders+`)`)+` FOR UPDATE`, s.scope.Args(args...)...)
		if err != nil {
			return err
		}
		if err := closeRowsAfterScan(rows); err != nil {
			return err
		}
	}
	return nil
}

func closeRowsAfterScan(rows *sql.Rows) error {
	var discard string
	for rows.Next() {
		if err := rows.Scan(&discard); err != nil {
			_ = rows.Close()
			return err
		}
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return err
	}
	return rows.Close()
}

func (s *Store) markFilesDeletedTx(ctx context.Context, tx *sql.Tx, fileIDs []string) error {
	for start := 0; start < len(fileIDs); start += deleteFileIDBatchSize {
		end := start + deleteFileIDBatchSize
		if end > len(fileIDs) {
			end = len(fileIDs)
		}
		batch := fileIDs[start:end]
		if len(batch) == 0 {
			continue
		}
		args := stringsToAny(batch)
		placeholders := questionPlaceholders(len(batch))
		if s.useLegacyFiles {
			if _, err := tx.ExecContext(ctx, `UPDATE files SET status = 'DELETED' WHERE file_id IN (`+placeholders+`)`, args...); err != nil {
				return err
			}
		}
		if _, err := tx.ExecContext(ctx, `UPDATE inodes SET status = 'DELETED' WHERE `+s.scope.And(`inode_id IN (`+placeholders+`)`), s.scope.Args(args...)...); err != nil {
			return err
		}
	}
	return nil
}

func fileFromLegacyGCScan(fileID string, storageType, contentLayout, storageRef, encMode, encKeyID, contentType sql.NullString, sizeBytes sql.NullInt64, checksum sql.NullString, revision, embeddingRevision sql.NullInt64, status, sourceID sql.NullString, createdAt, confirmedAt, expiresAt sql.NullTime) *File {
	f := &File{
		FileID:                 fileID,
		StorageType:            StorageType(storageType.String),
		ContentLayout:          contentLayoutForWrite(ContentLayout(contentLayout.String)),
		StorageRef:             storageRef.String,
		StorageEncryptionMode:  StorageEncryptionMode(encMode.String),
		StorageEncryptionKeyID: encKeyID.String,
		ContentType:            contentType.String,
		SizeBytes:              sizeBytes.Int64,
		ChecksumSHA256:         checksum.String,
		Revision:               revision.Int64,
		Status:                 FileStatus(status.String),
		SourceID:               sourceID.String,
	}
	if embeddingRevision.Valid {
		rev := embeddingRevision.Int64
		f.EmbeddingRevision = &rev
	}
	if createdAt.Valid {
		f.CreatedAt = createdAt.Time.UTC()
	}
	if confirmedAt.Valid {
		t := confirmedAt.Time.UTC()
		f.ConfirmedAt = &t
	}
	if expiresAt.Valid {
		t := expiresAt.Time.UTC()
		f.ExpiresAt = &t
	}
	return f
}

func (s *Store) deleteFileTagsByIDsTx(ctx context.Context, tx *sql.Tx, fileIDs []string) error {
	for start := 0; start < len(fileIDs); start += deleteFileIDBatchSize {
		end := start + deleteFileIDBatchSize
		if end > len(fileIDs) {
			end = len(fileIDs)
		}
		batch := fileIDs[start:end]
		if len(batch) == 0 {
			continue
		}
		_, err := tx.ExecContext(ctx, `DELETE FROM file_tags WHERE `+s.scope.And(`file_id IN (`+questionPlaceholders(len(batch))+`)`),
			s.scope.Args(stringsToAny(batch)...)...)
		if err != nil {
			return err
		}
	}
	return nil
}

func stringsToAny(values []string) []any {
	args := make([]any, len(values))
	for i, value := range values {
		args[i] = value
	}
	return args
}

func (s *Store) dirHasChildrenTx(ctx context.Context, tx *sql.Tx, path string) (bool, error) {
	var one int
	// FORCE INDEX (idx_parent): the child probe must lock ONLY the parent's
	// child range. On a small table the optimizer can otherwise choose a full
	// scan, whose FOR UPDATE locks every row — including rows the caller's
	// root-generation fence holds — and serializes unrelated concurrent
	// operations (observed as spurious lock-wait timeouts in the recursive
	// delete lift race).
	err := tx.QueryRowContext(ctx, `SELECT 1 FROM file_nodes FORCE INDEX (idx_parent) WHERE `+s.scope.And(`parent_path_hash = ? AND parent_path = ?`)+` LIMIT 1 FOR UPDATE`,
		s.scope.Args(fileNodePathHash(path), path)...).Scan(&one)
	if err == nil {
		return true, nil
	}
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	return false, err
}

func pathPrefixPredicate(column, prefix string) (string, []any) {
	return "BINARY " + column + " LIKE BINARY ? ESCAPE '\\\\'", []any{likeLiteralPrefixPattern(prefix)}
}

// andAsGrouped prefixes pred with the scope's fs_id predicate for WHERE
// clauses whose top-level OR would otherwise escape the fs_id conjunct:
// shared shape wraps pred in parentheses before applying And/AndAs. alias may
// be empty for unqualified predicates. Standalone shape returns pred
// byte-identical.
func (s *Store) andAsGrouped(alias, pred string) string {
	if !s.scope.Shared() {
		return pred
	}
	if alias == "" {
		return s.scope.And("(" + pred + ")")
	}
	return s.scope.AndAs(alias, "("+pred+")")
}

// --- uploads operations ---

func (s *Store) InsertUpload(ctx context.Context, u *Upload) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "insert_upload", start, &err)

	err = s.InsertUploadTx(s.db, u)
	return err
}

func (s *Store) InsertUploadTx(db execer, u *Upload) error {
	mode := uploadStorageEncryptionModeForWrite(u.StorageEncryptionMode)
	_, err := db.Exec(`INSERT INTO uploads
		(`+s.scope.InsCols(`upload_id, file_id, target_kind, layer_id, target_path, target_path_hash, base_inode_id, mode,
		 s3_upload_id, s3_key, storage_encryption_mode,
		 storage_encryption_key_id, total_size, part_size,
		 parts_total, expected_revision, status, fingerprint_sha256, idempotency_key, description, created_at, updated_at, expires_at`)+`)
		VALUES (`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?`)+`)`,
		s.scope.Args(u.UploadID, u.FileID, uploadTargetKindForWrite(u.TargetKind), u.LayerID, u.TargetPath,
			uploadTargetPathHash(u), u.BaseInodeID, uploadModeForWrite(u.Mode), u.S3UploadID, u.S3Key,
			mode, storageEncryptionKeyIDForWrite(mode, u.StorageEncryptionKeyID),
			u.TotalSize, u.PartSize, u.PartsTotal, nullInt64Ptr(u.ExpectedRevision), u.Status,
			nullStr(u.FingerprintSHA), nullStr(u.IdempotencyKey), nullStr(u.Description),
			u.CreatedAt.UTC(), u.UpdatedAt.UTC(), u.ExpiresAt.UTC())...)
	if isUniqueViolation(err) {
		// Distinguish constraint: idx_idempotency (duplicate key) vs idx_uploads_active (concurrent path).
		// MySQL embeds the constraint name in the error message; no structured alternative exists.
		// If the constraint is renamed, update this string and corresponding tests.
		if strings.Contains(err.Error(), "idx_idempotency") {
			return ErrIdempotencyConflict
		}
		return ErrUploadConflict
	}
	return err
}

func (s *Store) GetUpload(ctx context.Context, uploadID string) (out *Upload, err error) {
	start := time.Now()
	var opErr error
	defer observeStoreOp(ctx, "get_upload", start, &opErr)

	queryStart := time.Now()
	rows, err := s.db.QueryContext(ctx, `SELECT upload_id, file_id, target_kind, layer_id, target_path, base_inode_id, mode,
		s3_upload_id, s3_key,
		storage_encryption_mode, storage_encryption_key_id,
		total_size, part_size, parts_total, expected_revision, status, fingerprint_sha256, idempotency_key,
		description, created_at, updated_at, expires_at
		FROM uploads WHERE `+s.scope.And(`upload_id = ?`), s.scope.Args(uploadID)...)
	queryContextDurationMs := datastorePhaseMs(queryStart)
	if err != nil {
		opErr = err
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	firstRowStart := time.Now()
	if !rows.Next() {
		firstRowDurationMs := datastorePhaseMs(firstRowStart)
		if err := rows.Err(); err != nil {
			opErr = err
			return nil, err
		}
		logger.InfoBenchTiming(ctx, "datastore_get_upload_timing",
			zap.String("upload_id", uploadID),
			zap.Float64("query_context_ms", queryContextDurationMs),
			zap.Float64("first_row_ms", firstRowDurationMs),
			zap.Float64("scan_ms", 0),
			zap.Float64("hydrate_ms", 0),
			zap.String("result", "not_found"),
			zap.Float64("total_ms", datastorePhaseMs(start)),
		)
		opErr = ErrNotFound
		return nil, ErrNotFound
	}
	firstRowDurationMs := datastorePhaseMs(firstRowStart)

	var u Upload
	var fingerprint, idempotencyKey sql.NullString
	var expectedRevision sql.NullInt64
	var createdAt, updatedAt, expiresAt time.Time
	scanStart := time.Now()
	var description sql.NullString
	err = rows.Scan(&u.UploadID, &u.FileID, &u.TargetKind, &u.LayerID, &u.TargetPath, &u.BaseInodeID, &u.Mode,
		&u.S3UploadID, &u.S3Key,
		&u.StorageEncryptionMode, &u.StorageEncryptionKeyID,
		&u.TotalSize, &u.PartSize, &u.PartsTotal, &expectedRevision, &u.Status,
		&fingerprint, &idempotencyKey, &description,
		&createdAt, &updatedAt, &expiresAt)
	if description.Valid {
		u.Description = description.String
	}
	scanDurationMs := datastorePhaseMs(scanStart)
	if err != nil {
		opErr = err
		return nil, err
	}
	if err := rows.Err(); err != nil {
		opErr = err
		return nil, err
	}

	hydrateStart := time.Now()
	u.FingerprintSHA = fingerprint.String
	u.IdempotencyKey = idempotencyKey.String
	if expectedRevision.Valid {
		rev := expectedRevision.Int64
		u.ExpectedRevision = &rev
	}
	u.CreatedAt = createdAt.UTC()
	u.UpdatedAt = updatedAt.UTC()
	u.ExpiresAt = expiresAt.UTC()
	hydrateDurationMs := datastorePhaseMs(hydrateStart)

	logger.InfoBenchTiming(ctx, "datastore_get_upload_timing",
		zap.String("upload_id", uploadID),
		zap.Int("parts_total", u.PartsTotal),
		zap.String("status", string(u.Status)),
		zap.Float64("query_context_ms", queryContextDurationMs),
		zap.Float64("first_row_ms", firstRowDurationMs),
		zap.Float64("scan_ms", scanDurationMs),
		zap.Float64("hydrate_ms", hydrateDurationMs),
		zap.String("result", "ok"),
		zap.Float64("total_ms", datastorePhaseMs(start)),
	)

	out = &u
	return out, nil
}

func (s *Store) GetUploadByPath(ctx context.Context, targetPath string) (out *Upload, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "get_upload_by_path", start, &err)

	row := s.db.QueryRowContext(ctx, `SELECT upload_id, file_id, target_kind, layer_id, target_path, base_inode_id, mode,
		s3_upload_id, s3_key,
		storage_encryption_mode, storage_encryption_key_id,
		total_size, part_size, parts_total, expected_revision, status, fingerprint_sha256, idempotency_key,
		description, created_at, updated_at, expires_at
		FROM uploads WHERE `+s.scope.And(`target_kind = ? AND target_path_hash = ? AND target_path = ?
		AND status IN ('INITIATED', 'UPLOADING') AND expires_at > ?`)+`
		ORDER BY created_at DESC LIMIT 1`, s.scope.Args(string(UploadTargetFile), fileNodePathHash(targetPath), targetPath, time.Now().UTC())...)
	out, err = scanUpload(row)
	return out, err
}

func (s *Store) GetFSLayerUploadByPath(ctx context.Context, layerID, targetPath string) (out *Upload, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "get_fs_layer_upload_by_path", start, &err)

	target := &Upload{TargetKind: UploadTargetFSLayer, LayerID: layerID, TargetPath: targetPath}
	row := s.db.QueryRowContext(ctx, `SELECT upload_id, file_id, target_kind, layer_id, target_path, base_inode_id, mode,
		s3_upload_id, s3_key,
		storage_encryption_mode, storage_encryption_key_id,
		total_size, part_size, parts_total, expected_revision, status, fingerprint_sha256, idempotency_key,
		description, created_at, updated_at, expires_at
		FROM uploads WHERE `+s.scope.And(`target_kind = ? AND layer_id = ? AND target_path_hash = ? AND target_path = ?
		AND status IN ('INITIATED', 'UPLOADING') AND expires_at > ?`)+`
		ORDER BY created_at DESC LIMIT 1`, s.scope.Args(string(UploadTargetFSLayer), layerID, uploadTargetPathHash(target), targetPath, time.Now().UTC())...)
	out, err = scanUpload(row)
	return out, err
}

func (s *Store) CompleteUpload(ctx context.Context, uploadID string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "complete_upload", start, &err)

	_, err = s.db.ExecContext(ctx, `UPDATE uploads SET status = 'COMPLETED',
		updated_at = ?
		WHERE `+s.scope.And(`upload_id = ? AND status = 'UPLOADING'`),
		append([]any{time.Now().UTC()}, s.scope.Args(uploadID)...)...)
	return err
}

func (s *Store) AbortUpload(ctx context.Context, uploadID string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "abort_upload", start, &err)

	_, err = s.db.ExecContext(ctx, `UPDATE uploads SET status = 'ABORTED',
		updated_at = ?
		WHERE `+s.scope.And(`upload_id = ? AND status = 'UPLOADING'`),
		append([]any{time.Now().UTC()}, s.scope.Args(uploadID)...)...)
	return err
}

// AbortUploadV2 sets an upload to ABORTED from either INITIATED or UPLOADING.
// The returned bool reports whether this call won the state transition.
// A false, nil result means the upload was already terminal or not found.
func (s *Store) AbortUploadV2(ctx context.Context, uploadID string) (aborted bool, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "abort_upload_v2", start, &err)

	res, err := s.db.ExecContext(ctx, `UPDATE uploads SET status = 'ABORTED',
		updated_at = ?
		WHERE `+s.scope.And(`upload_id = ? AND status IN ('INITIATED', 'UPLOADING')`),
		append([]any{time.Now().UTC()}, s.scope.Args(uploadID)...)...)
	if err != nil {
		return false, err
	}
	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return rowsAffected > 0, nil
}

// UpdateUploadStatus transitions an upload to a new status.
func (s *Store) UpdateUploadStatus(ctx context.Context, uploadID string, status UploadStatus) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "update_upload_status", start, &err)

	_, err = s.db.ExecContext(ctx, `UPDATE uploads SET status = ?,
		updated_at = ?
		WHERE `+s.scope.And(`upload_id = ?`),
		append([]any{string(status), time.Now().UTC()}, s.scope.Args(uploadID)...)...)
	return err
}

// TransitionUploadStatus atomically transitions an upload from expectedStatus to newStatus.
// Returns ErrUploadNotActive if the current status doesn't match expectedStatus.
func (s *Store) TransitionUploadStatus(ctx context.Context, uploadID string, expectedStatus, newStatus UploadStatus) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "transition_upload_status", start, &err)

	res, err := s.db.ExecContext(ctx, `UPDATE uploads SET status = ?,
		updated_at = ?
		WHERE `+s.scope.And(`upload_id = ? AND status = ?`),
		append([]any{string(newStatus), time.Now().UTC()}, s.scope.Args(uploadID, string(expectedStatus))...)...)
	if err != nil {
		if isUniqueViolation(err) {
			return ErrUploadConflict
		}
		return err
	}
	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if rowsAffected == 0 {
		return ErrUploadNotActive
	}
	return nil
}

func (s *Store) ListUploadsByPath(ctx context.Context, targetPath string, status UploadStatus) (out []*Upload, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "list_uploads_by_path", start, &err)

	rows, err := s.db.QueryContext(ctx, `SELECT upload_id, file_id, target_kind, layer_id, target_path, base_inode_id, mode,
		s3_upload_id, s3_key,
		storage_encryption_mode, storage_encryption_key_id,
		total_size, part_size, parts_total, expected_revision, status, fingerprint_sha256, idempotency_key,
		description, created_at, updated_at, expires_at
		FROM uploads WHERE `+s.scope.And(`target_kind = ? AND target_path_hash = ? AND target_path = ? AND status = ?`)+`
		ORDER BY created_at DESC`, s.scope.Args(string(UploadTargetFile), fileNodePathHash(targetPath), targetPath, status)...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	uploads := make([]*Upload, 0)
	for rows.Next() {
		u, err := scanUpload(rows)
		if err != nil {
			return nil, err
		}
		uploads = append(uploads, u)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	out = uploads
	return out, nil
}

// --- scan helpers ---

type scanner interface {
	Scan(dest ...interface{}) error
}

func scanNode(s scanner) (*FileNode, error) {
	var n FileNode
	var isDir int
	var fileID, inodeID sql.NullString
	var createdAt time.Time
	err := s.Scan(&n.NodeID, &n.Path, &n.ParentPath, &n.Name, &isDir, &fileID, &inodeID, &createdAt)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	n.IsDirectory = isDir != 0
	n.FileID = fileID.String
	n.InodeID = inodeID.String
	n.CreatedAt = createdAt.UTC()
	return &n, nil
}

func assembleFile(inode *Inode, content *Content, semantic *Semantic) *File {
	f := &File{
		FileID:      inode.InodeID,
		SizeBytes:   inode.SizeBytes,
		Revision:    inode.Revision,
		Mode:        inode.Mode,
		Status:      inode.Status,
		CreatedAt:   inode.CreatedAt,
		ConfirmedAt: inode.ConfirmedAt,
		ExpiresAt:   inode.ExpiresAt,
	}
	if content != nil {
		f.StorageType = content.StorageType
		f.ContentLayout = content.ContentLayout
		f.StorageRef = content.StorageRef
		f.StorageEncryptionMode = content.StorageEncryptionMode
		f.StorageEncryptionKeyID = content.StorageEncryptionKeyID
		f.ContentBlob = append([]byte(nil), content.ContentBlob...)
		f.ContentType = content.ContentType
		f.ChecksumSHA256 = content.ChecksumSHA256
		f.SourceID = content.SourceID
	}
	if semantic != nil {
		f.ContentText = semantic.ContentText
		f.Description = semantic.Description
		f.EmbeddingRevision = semantic.EmbeddingRevision
		f.DescriptionEmbeddingRevision = semantic.DescriptionEmbeddingRevision
	}
	return f
}

func scanFileForGC(s scanner) (*File, error) {
	var f File
	var contentType, checksum, sourceID sql.NullString
	var embeddingRevision sql.NullInt64
	var confirmedAt, expiresAt sql.NullTime
	var createdAt time.Time
	err := s.Scan(&f.FileID, &f.StorageType, &f.ContentLayout, &f.StorageRef, &f.StorageEncryptionMode,
		&f.StorageEncryptionKeyID, &contentType, &f.SizeBytes, &checksum,
		&f.Revision, &embeddingRevision, &f.Status, &sourceID, &createdAt, &confirmedAt, &expiresAt)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	f.ContentType = contentType.String
	f.ChecksumSHA256 = checksum.String
	f.SourceID = sourceID.String
	if embeddingRevision.Valid {
		rev := embeddingRevision.Int64
		f.EmbeddingRevision = &rev
	}
	f.CreatedAt = createdAt.UTC()
	if confirmedAt.Valid {
		t := confirmedAt.Time.UTC()
		f.ConfirmedAt = &t
	}
	if expiresAt.Valid {
		t := expiresAt.Time.UTC()
		f.ExpiresAt = &t
	}
	return &f, nil
}

// scanFileForGCTx reads the minimal file fields needed for GC task creation.
// For legacy tenants it queries the files table; for new tenants it queries
// the split tables (inodes/contents/semantic).
func (s *Store) scanFileForGCTx(db execer, fileID string) (*File, error) {
	if s.useLegacyFiles {
		row := db.QueryRow(`SELECT file_id, storage_type, content_layout, storage_ref, storage_encryption_mode,
			storage_encryption_key_id, content_type, size_bytes, checksum_sha256,
			revision, embedding_revision, status, source_id, created_at, confirmed_at, expires_at
			FROM files WHERE file_id = ?`, fileID)
		return scanFileForGC(row)
	}
	var f File
	f.FileID = fileID
	f.Status = StatusDeleted
	var contentType sql.NullString
	var embRev sql.NullInt64
	err := db.QueryRow(`SELECT c.storage_type, c.content_layout, c.storage_ref, i.size_bytes, COALESCE(c.content_type, ''),
		s.embedding_revision
		FROM inodes i
		JOIN contents c ON i.inode_id = c.inode_id
		LEFT JOIN semantic s ON `+s.scope.AndOn(`i.inode_id = s.inode_id`, "s")+`
		WHERE `+scopeWhereAnd(s.scope, `i.inode_id = ?`, "i", "c"), scopeWhereArgs(s.scope, 3, fileID)...).Scan(
		&f.StorageType, &f.ContentLayout, &f.StorageRef, &f.SizeBytes, &contentType, &embRev)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	f.ContentType = contentType.String
	if embRev.Valid {
		f.EmbeddingRevision = &embRev.Int64
	}
	return &f, nil
}

func scanNodeWithFileWithBlob(s scanner) (*NodeWithFile, error) {
	var n FileNode
	var isDir int
	var nodeFileID sql.NullString
	var nodeCreatedAt time.Time

	var fFileID, fStorageType, fContentLayout, fStorageRef, fStorageEncryptionMode, fStorageEncryptionKeyID sql.NullString
	var fContentBlob []byte
	var fContentType, fChecksum, fSourceID, fContentText, fDescription sql.NullString
	var fSizeBytes, fRevision, fMode, fEmbeddingRevision, fDescriptionEmbeddingRevision sql.NullInt64
	var fStatus sql.NullString
	var fCreatedAt, fConfirmedAt, fExpiresAt sql.NullTime

	err := s.Scan(&n.NodeID, &n.Path, &n.ParentPath, &n.Name, &isDir, &nodeFileID, &nodeCreatedAt,
		&fFileID, &fStorageType, &fContentLayout, &fStorageRef, &fStorageEncryptionMode, &fStorageEncryptionKeyID,
		&fContentBlob, &fContentType, &fSizeBytes,
		&fChecksum, &fRevision, &fMode, &fEmbeddingRevision, &fStatus, &fSourceID, &fContentText,
		&fDescription, &fDescriptionEmbeddingRevision, &fCreatedAt, &fConfirmedAt, &fExpiresAt)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}

	n.IsDirectory = isDir != 0
	n.FileID = nodeFileID.String
	n.CreatedAt = nodeCreatedAt.UTC()

	nf := &NodeWithFile{Node: n}
	if fFileID.Valid {
		nf.File = &File{
			FileID:                 fFileID.String,
			StorageType:            StorageType(fStorageType.String),
			ContentLayout:          contentLayoutForWrite(ContentLayout(fContentLayout.String)),
			StorageRef:             fStorageRef.String,
			StorageEncryptionMode:  StorageEncryptionMode(fStorageEncryptionMode.String),
			StorageEncryptionKeyID: fStorageEncryptionKeyID.String,
			ContentBlob:            append([]byte(nil), fContentBlob...),
			ContentType:            fContentType.String,
			SizeBytes:              fSizeBytes.Int64,
			ChecksumSHA256:         fChecksum.String,
			Revision:               fRevision.Int64,
			Mode:                   uint32(fMode.Int64),
			Status:                 FileStatus(fStatus.String),
			SourceID:               fSourceID.String,
			ContentText:            fContentText.String,
			Description:            fDescription.String,
		}
		if fEmbeddingRevision.Valid {
			rev := fEmbeddingRevision.Int64
			nf.File.EmbeddingRevision = &rev
		}
		if fDescriptionEmbeddingRevision.Valid {
			rev := fDescriptionEmbeddingRevision.Int64
			nf.File.DescriptionEmbeddingRevision = &rev
		}
		if fCreatedAt.Valid {
			nf.File.CreatedAt = fCreatedAt.Time.UTC()
		}
		if fConfirmedAt.Valid {
			t := fConfirmedAt.Time.UTC()
			nf.File.ConfirmedAt = &t
		}
		if fExpiresAt.Valid {
			t := fExpiresAt.Time.UTC()
			nf.File.ExpiresAt = &t
		}
	}
	if fMode.Valid {
		nf.Mode = uint32(fMode.Int64)
		nf.HasMode = true
	}
	return nf, nil
}

func scanNodeWithFileMetadata(s scanner) (*NodeWithFile, error) {
	// Scan order mirrors statMetadata's dentry, inode, content, and semantic projection.
	var n FileNode
	var isDir int
	var nodeFileID sql.NullString
	var nodeCreatedAt time.Time

	var fFileID, fContentType, fContentText, fDescription sql.NullString
	var fSizeBytes, fRevision sql.NullInt64
	var fCreatedAt, fConfirmedAt sql.NullTime

	err := s.Scan(&n.NodeID, &n.Path, &n.ParentPath, &n.Name, &isDir, &nodeFileID, &nodeCreatedAt,
		&fFileID, &fSizeBytes, &fRevision, &fCreatedAt, &fConfirmedAt, &fContentType, &fContentText, &fDescription)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}

	n.IsDirectory = isDir != 0
	n.FileID = nodeFileID.String
	n.CreatedAt = nodeCreatedAt.UTC()

	nf := &NodeWithFile{Node: n}
	if fFileID.Valid {
		nf.File = &File{
			FileID:      fFileID.String,
			ContentType: fContentType.String,
			SizeBytes:   fSizeBytes.Int64,
			Revision:    fRevision.Int64,
			ContentText: fContentText.String,
			Description: fDescription.String,
		}
		if fCreatedAt.Valid {
			nf.File.CreatedAt = fCreatedAt.Time.UTC()
		}
		if fConfirmedAt.Valid {
			t := fConfirmedAt.Time.UTC()
			nf.File.ConfirmedAt = &t
		}
	}
	return nf, nil
}

// scanNodeWithFileForRead scans the result set for ReadPlan: node fields +
// file_id, storage_type, storage_ref, content_blob, size_bytes, revision,
// status, created_at, confirmed_at. Includes blob for db9 inline reads but
// excludes content_text, description, embedding columns.
func scanNodeWithFileForRead(s scanner) (*NodeWithFile, error) {
	var n FileNode
	var isDir int
	var nodeFileID sql.NullString
	var nodeCreatedAt time.Time

	var fFileID, fStorageType, fContentLayout, fStorageRef sql.NullString
	var fContentBlob []byte
	var fSizeBytes, fRevision sql.NullInt64
	var fMode sql.NullInt64
	var fStatus sql.NullString
	var fCreatedAt, fConfirmedAt sql.NullTime

	err := s.Scan(&n.NodeID, &n.Path, &n.ParentPath, &n.Name, &isDir, &nodeFileID, &nodeCreatedAt,
		&fFileID, &fStorageType, &fContentLayout, &fStorageRef, &fContentBlob, &fSizeBytes, &fRevision, &fMode, &fStatus, &fCreatedAt, &fConfirmedAt)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}

	n.IsDirectory = isDir != 0
	n.FileID = nodeFileID.String
	n.CreatedAt = nodeCreatedAt.UTC()

	nf := &NodeWithFile{Node: n}
	if fFileID.Valid {
		nf.File = &File{
			FileID:        fFileID.String,
			StorageType:   StorageType(fStorageType.String),
			ContentLayout: contentLayoutForWrite(ContentLayout(fContentLayout.String)),
			StorageRef:    fStorageRef.String,
			ContentBlob:   append([]byte(nil), fContentBlob...),
			SizeBytes:     fSizeBytes.Int64,
			Revision:      fRevision.Int64,
			Status:        FileStatus(fStatus.String),
		}
		if fCreatedAt.Valid {
			nf.File.CreatedAt = fCreatedAt.Time.UTC()
		}
		if fConfirmedAt.Valid {
			t := fConfirmedAt.Time.UTC()
			nf.File.ConfirmedAt = &t
		}
	}
	if fMode.Valid {
		nf.Mode = uint32(fMode.Int64)
		nf.HasMode = true
	}
	return nf, nil
}

// scanNodeWithFileLite scans lite metadata with an optional trailing checksum.
func scanNodeWithFileLite(s scanner, includeChecksum bool) (*NodeWithFile, error) {
	var n FileNode
	var isDir int
	var nodeFileID sql.NullString
	var nodeCreatedAt time.Time

	var fFileID sql.NullString
	var fSizeBytes, fRevision sql.NullInt64
	var fMode sql.NullInt64
	var fStatus sql.NullString
	var fCreatedAt, fConfirmedAt sql.NullTime
	var fStorageType, fContentLayout sql.NullString
	var fChecksum sql.NullString

	dest := []any{&n.NodeID, &n.Path, &n.ParentPath, &n.Name, &isDir, &nodeFileID, &nodeCreatedAt,
		&fFileID, &fSizeBytes, &fRevision, &fMode, &fStatus, &fCreatedAt, &fConfirmedAt, &fStorageType, &fContentLayout}
	if includeChecksum {
		dest = append(dest, &fChecksum)
	}
	err := s.Scan(dest...)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}

	n.IsDirectory = isDir != 0
	n.FileID = nodeFileID.String
	n.CreatedAt = nodeCreatedAt.UTC()

	nf := &NodeWithFile{Node: n}
	if fFileID.Valid {
		nf.File = &File{
			FileID:         fFileID.String,
			ContentLayout:  contentLayoutForWrite(ContentLayout(fContentLayout.String)),
			SizeBytes:      fSizeBytes.Int64,
			ChecksumSHA256: fChecksum.String,
			Revision:       fRevision.Int64,
			Mode:           uint32(fMode.Int64),
			Status:         FileStatus(fStatus.String),
			StorageType:    StorageType(fStorageType.String),
		}
		if fCreatedAt.Valid {
			nf.File.CreatedAt = fCreatedAt.Time.UTC()
		}
		if fConfirmedAt.Valid {
			t := fConfirmedAt.Time.UTC()
			nf.File.ConfirmedAt = &t
		}
	}
	if fMode.Valid {
		nf.Mode = uint32(fMode.Int64)
		nf.HasMode = true
	}
	return nf, nil
}

func nilBytes(b []byte) any {
	if len(b) == 0 {
		return nil
	}
	return b
}

func fileStorageEncryptionModeForWrite(mode StorageEncryptionMode) StorageEncryptionMode {
	if mode == "" {
		return StorageEncryptionLegacy
	}
	return mode
}

func contentLayoutForWrite(layout ContentLayout) ContentLayout {
	if layout == "" {
		return ContentLayoutSingle
	}
	return layout
}

func uploadStorageEncryptionModeForWrite(mode StorageEncryptionMode) StorageEncryptionMode {
	if mode == "" {
		return StorageEncryptionNone
	}
	return mode
}

func uploadTargetKindForWrite(kind UploadTargetKind) UploadTargetKind {
	if kind == "" {
		return UploadTargetFile
	}
	return kind
}

func uploadModeForWrite(mode uint32) uint32 {
	if mode == 0 {
		return 0o644
	}
	return mode & 0o777
}

func uploadTargetPathHash(upload *Upload) string {
	if upload != nil && uploadTargetKindForWrite(upload.TargetKind) == UploadTargetFSLayer {
		return StorageRefHash("fs_layer\x00" + upload.LayerID + "\x00" + upload.TargetPath)
	}
	if upload == nil {
		return fileNodePathHash("")
	}
	return fileNodePathHash(upload.TargetPath)
}

func storageEncryptionKeyIDForWrite(mode StorageEncryptionMode, keyID string) string {
	switch mode {
	case StorageEncryptionSSEKMS, StorageEncryptionDSSEKMS:
		return keyID
	default:
		return ""
	}
}

func scanUpload(s scanner) (*Upload, error) {
	var u Upload
	var fingerprint, idempotencyKey, description sql.NullString
	var expectedRevision sql.NullInt64
	var createdAt, updatedAt, expiresAt time.Time
	err := s.Scan(&u.UploadID, &u.FileID, &u.TargetKind, &u.LayerID, &u.TargetPath, &u.BaseInodeID, &u.Mode,
		&u.S3UploadID, &u.S3Key,
		&u.StorageEncryptionMode, &u.StorageEncryptionKeyID,
		&u.TotalSize, &u.PartSize, &u.PartsTotal, &expectedRevision, &u.Status,
		&fingerprint, &idempotencyKey, &description,
		&createdAt, &updatedAt, &expiresAt)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	u.FingerprintSHA = fingerprint.String
	u.IdempotencyKey = idempotencyKey.String
	u.Description = description.String
	if expectedRevision.Valid {
		rev := expectedRevision.Int64
		u.ExpectedRevision = &rev
	}
	u.CreatedAt = createdAt.UTC()
	u.UpdatedAt = updatedAt.UTC()
	u.ExpiresAt = expiresAt.UTC()
	return &u, nil
}

// --- string helpers ---

func nullStr(s string) interface{} {
	if s == "" {
		return nil
	}
	return s
}

func nullInt64Ptr(v *int64) interface{} {
	if v == nil {
		return nil
	}
	return *v
}

func nilTime(t *time.Time) interface{} {
	if t == nil {
		return nil
	}
	return t.UTC()
}

func coalesceTime(t *time.Time, fallback time.Time) time.Time {
	if t != nil {
		return *t
	}
	return fallback
}

func parentPath(p string) string {
	if p == "/" {
		return "/"
	}
	p = strings.TrimSuffix(p, "/")
	idx := strings.LastIndex(p, "/")
	if idx <= 0 {
		return "/"
	}
	return p[:idx+1]
}

func baseName(p string) string {
	p = strings.TrimSuffix(p, "/")
	idx := strings.LastIndex(p, "/")
	if idx < 0 {
		return p
	}
	return p[idx+1:]
}

func isUniqueViolation(err error) bool {
	if err == nil {
		return false
	}
	msg := err.Error()
	return strings.Contains(msg, "Duplicate entry") || strings.Contains(msg, "UNIQUE constraint failed")
}

func isDeadlock(err error) bool {
	if err == nil {
		return false
	}
	msg := err.Error()
	return strings.Contains(msg, "Deadlock found") || strings.Contains(msg, "deadlock")
}

func datastorePhaseMs(start time.Time) float64 {
	return float64(time.Since(start).Microseconds()) / 1000.0
}

func observeStoreOp(ctx context.Context, op string, start time.Time, errp *error) {
	elapsed := time.Since(start)
	result := "ok"
	if errp != nil && *errp != nil {
		result = storeOpResultForError(*errp)
		if shouldLogStoreOpFailure(result) {
			logger.Error(ctx, "datastore_op_failed", zap.String("operation", op), zap.String("result", result), zap.Error(*errp))
		} else {
			logger.Warn(ctx, "datastore_op_failed", zap.String("operation", op), zap.String("result", result), zap.String("detail", (*errp).Error()))
		}
	}
	logger.InfoBenchTiming(ctx, "datastore_op_timing",
		zap.String("operation", op),
		zap.String("result", result),
		zap.Float64("duration_ms", float64(elapsed.Microseconds())/1000.0))
	metrics.RecordOperation("datastore", op, result, elapsed)
}

func storeOpResultForError(err error) string {
	switch {
	case err == nil:
		return "ok"
	case errors.Is(err, ErrNotFound):
		return "not_found"
	case errors.Is(err, ErrPathConflict),
		errors.Is(err, ErrDirectoryNotEmpty),
		errors.Is(err, ErrRevisionConflict),
		errors.Is(err, ErrUploadConflict),
		errors.Is(err, ErrUploadNotActive),
		errors.Is(err, ErrIdempotencyConflict),
		errors.Is(err, ErrJournalConflict):
		return "conflict"
	case errors.Is(err, context.Canceled):
		return "canceled"
	case errors.Is(err, context.DeadlineExceeded):
		return "deadline_exceeded"
	case errors.Is(err, sql.ErrConnDone):
		return "conn_closed"
	default:
		if strings.Contains(err.Error(), "database is closed") {
			return "conn_closed"
		}
		return "error"
	}
}

func shouldLogStoreOpFailure(result string) bool {
	switch result {
	case "not_found", "conflict", "canceled", "deadline_exceeded", "conn_closed":
		return false
	default:
		return true
	}
}