package datastore

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

	"github.com/pingcap/failpoint"

	"github.com/mem9-ai/drive9/pkg/pathutil"
)

type FSLayerState string
type FSLayerDurabilityMode string
type FSLayerEntryOp string
type FSLayerEntryKind string

var ErrFSLayerRefAmbiguous = errors.New("fs layer ref is ambiguous")
var ErrFSLayerStateConflict = errors.New("fs layer state conflict")
var ErrFSLayerInvalidArgument = errors.New("invalid fs layer argument")
var ErrFSLayerDepthExceeded = errors.New("fs layer depth exceeded")
var ErrFSLayerHasChildren = errors.New("fs layer has pinned descendants")

const (
	FSLayerStateActive     FSLayerState = "active"
	FSLayerStateSealed     FSLayerState = "sealed"
	FSLayerStateCommitting FSLayerState = "committing"
	FSLayerStateCommitted  FSLayerState = "committed"
	FSLayerStateAbandoned  FSLayerState = "abandoned"
	FSLayerStateConflicted FSLayerState = "conflicted"

	FSLayerDurabilityRestoreSafe  FSLayerDurabilityMode = "restore-safe"
	FSLayerDurabilityWriteThrough FSLayerDurabilityMode = "write-through"
	FSLayerDurabilityLocalFast    FSLayerDurabilityMode = "local-fast"

	FSLayerEntryOpUpsert   FSLayerEntryOp = "upsert"
	FSLayerEntryOpWhiteout FSLayerEntryOp = "whiteout"
	FSLayerEntryOpMkdir    FSLayerEntryOp = "mkdir"
	FSLayerEntryOpSymlink  FSLayerEntryOp = "symlink"
	FSLayerEntryOpChmod    FSLayerEntryOp = "chmod"
	FSLayerEntryOpRename   FSLayerEntryOp = "rename"

	FSLayerEntryKindFile    FSLayerEntryKind = "file"
	FSLayerEntryKindDir     FSLayerEntryKind = "dir"
	FSLayerEntryKindSymlink FSLayerEntryKind = "symlink"

	FSLayerIDMaxBytes           = 64
	FSLayerCheckpointIDMaxBytes = 64
	fsLayerEventIDMaxBytes      = 64

	FSLayerOriginCreate = "create"
	FSLayerOriginFork   = "fork"

	// FSLayerDefaultMaxDepth is the default soft cap (D15). Hard top is FSLayerHardMaxDepth.
	FSLayerDefaultMaxDepth = 8
	FSLayerHardMaxDepth    = 16

	// fsLayerSelectCols is the column list for scanning fs_layers rows.
	fsLayerSelectCols = `layer_id, base_root_path, name, state, durability_mode, actor_id, durable_seq,
	parent_layer_id, origin_seq, origin_checkpoint_id, root_layer_id, depth, origin,
	commit_applied, created_at, updated_at, sealed_at`

	// fsLayerEventOpRollback is the synthetic event op emitted into
	// fs_layer_events when a layer is rolled back. It lets SSE/layer-event
	// consumers detect the rollback and refresh their view without a remount.
	// Unlike entry ops it is never written to fs_layer_entries and bypasses
	// validateFSLayerEntryOp. Both equal "rollback".
	fsLayerEventOpRollback = "rollback"
)

type FSLayer struct {
	LayerID            string
	BaseRootPath       string
	Name               string
	Tags               map[string]string
	State              FSLayerState
	DurabilityMode     FSLayerDurabilityMode
	ActorID            string
	DurableSeq         int64
	ParentLayerID      string
	OriginSeq          int64
	OriginCheckpointID string
	RootLayerID        string
	Depth              int
	Origin             string
	CommitApplied      bool
	CreatedAt          time.Time
	UpdatedAt          time.Time
	SealedAt           *time.Time
}

type FSLayerEntry struct {
	LayerID                string
	Path                   string
	PathHash               string
	ParentPath             string
	ParentPathHash         string
	Name                   string
	Op                     FSLayerEntryOp
	Kind                   FSLayerEntryKind
	BaseInodeID            string
	BaseRevision           int64
	StorageType            string
	StorageRef             string
	StorageRefHash         string
	StorageEncryptionMode  StorageEncryptionMode
	StorageEncryptionKeyID string
	ContentBlob            []byte
	ContentType            string
	ContentText            string
	ChecksumSHA256         string
	SizeBytes              int64
	Mode                   uint32
	EntrySeq               int64
	CreatedAt              time.Time
	UpdatedAt              time.Time
}

type FSLayerCheckpoint struct {
	CheckpointID string
	LayerID      string
	DurableSeq   int64
	Label        string
	CreatedAt    time.Time
}

type FSLayerEvent struct {
	EventID   string
	LayerID   string
	Seq       int64
	ActorID   string
	Op        string
	Path      string
	CreatedAt time.Time
}

func (s *Store) CreateFSLayer(ctx context.Context, layer *FSLayer) error {
	if layer == nil {
		return fmt.Errorf("fs layer is required")
	}
	layer.LayerID = strings.TrimSpace(layer.LayerID)
	if err := validateFSLayerID(layer.LayerID); err != nil {
		return err
	}
	root, err := pathutil.CanonicalizeDir(layer.BaseRootPath)
	if err != nil {
		return fmt.Errorf("invalid fs layer base root: %w", err)
	}
	layer.BaseRootPath = root
	layer.Name = strings.TrimSpace(layer.Name)
	layer.ActorID = strings.TrimSpace(layer.ActorID)
	if layer.State == "" {
		layer.State = FSLayerStateActive
	}
	if err := validateFSLayerState(layer.State); err != nil {
		return err
	}
	if layer.DurabilityMode == "" {
		layer.DurabilityMode = FSLayerDurabilityRestoreSafe
	}
	if err := validateFSLayerDurability(layer.DurabilityMode); err != nil {
		return err
	}
	// Root layers (create): parent empty, self root, depth 0.
	if strings.TrimSpace(layer.ParentLayerID) == "" {
		layer.ParentLayerID = ""
		layer.OriginSeq = 0
		layer.OriginCheckpointID = ""
		if layer.RootLayerID == "" {
			layer.RootLayerID = layer.LayerID
		}
		layer.Depth = 0
		if layer.Origin == "" {
			layer.Origin = FSLayerOriginCreate
		}
	}
	tags, err := normalizeFSLayerTags(layer.Tags)
	if err != nil {
		return err
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin create fs layer transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	_, err = tx.ExecContext(ctx, `
INSERT INTO fs_layers (
	`+s.scope.InsCols(`layer_id, base_root_path, name, state, durability_mode, actor_id, durable_seq,
	parent_layer_id, origin_seq, origin_checkpoint_id, root_layer_id, depth, origin,
	created_at, updated_at, sealed_at`)+`
) VALUES (`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, UTC_TIMESTAMP(3), UTC_TIMESTAMP(3), ?`)+`)`,
		s.scope.Args(layer.LayerID, layer.BaseRootPath, layer.Name, string(layer.State), string(layer.DurabilityMode), layer.ActorID,
			layer.DurableSeq, layer.ParentLayerID, layer.OriginSeq, layer.OriginCheckpointID, layer.RootLayerID, layer.Depth, layer.Origin,
			nilTime(layer.SealedAt))...)
	if err != nil {
		return fmt.Errorf("create fs layer %s: %w", layer.LayerID, err)
	}
	if err := s.replaceFSLayerTagsTx(ctx, tx, layer.LayerID, tags); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit create fs layer %s: %w", layer.LayerID, err)
	}
	layer.Tags = tags
	return nil
}

func (s *Store) GetFSLayer(ctx context.Context, layerID string) (*FSLayer, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	row := s.db.QueryRowContext(ctx, `
SELECT `+fsLayerSelectCols+`
	FROM fs_layers
	WHERE `+s.scope.And(`layer_id = ?`), s.scope.Args(layerID)...)
	layer, err := scanFSLayer(row)
	if err != nil {
		return nil, err
	}
	if err := s.loadFSLayerTags(ctx, layer); err != nil {
		return nil, err
	}
	return layer, nil
}

func (s *Store) ListFSLayers(ctx context.Context) ([]FSLayer, error) {
	// A full layer list has no natural WHERE clause; in shared shape still
	// restrict the scan to this tenant's fs_id.
	query := `
SELECT ` + fsLayerSelectCols + `
FROM fs_layers`
	var args []any
	if s.scope.Shared() {
		query += `
WHERE fs_id = ?`
		args = append(args, s.scope.FsID())
	}
	query += `
ORDER BY updated_at DESC, layer_id`
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list fs layers: %w", err)
	}
	defer func() { _ = rows.Close() }()
	var out []FSLayer
	for rows.Next() {
		layer, err := scanFSLayer(rows)
		if err != nil {
			return nil, fmt.Errorf("scan fs layer: %w", err)
		}
		if err := s.loadFSLayerTags(ctx, layer); err != nil {
			return nil, err
		}
		out = append(out, *layer)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list fs layers: %w", err)
	}
	return out, nil
}

// ListFSLayersByRoot returns layers in the same fork tree, including tags.
// Roots created before root_layer_id was filled store an empty root and are
// matched by layer_id.
func (s *Store) ListFSLayersByRoot(ctx context.Context, rootID string) ([]FSLayer, error) {
	return s.listFSLayersByRoot(ctx, rootID, true)
}

// ListFSLayersByRootMeta is ListFSLayersByRoot without per-row tag loads.
// Hop-lineage callers only need state, parent, and origin_seq.
func (s *Store) ListFSLayersByRootMeta(ctx context.Context, rootID string) ([]FSLayer, error) {
	return s.listFSLayersByRoot(ctx, rootID, false)
}

func (s *Store) listFSLayersByRoot(ctx context.Context, rootID string, withTags bool) ([]FSLayer, error) {
	rootID = strings.TrimSpace(rootID)
	if rootID == "" {
		return nil, fmt.Errorf("fs layer root id is required")
	}
	query := `
SELECT ` + fsLayerSelectCols + `
FROM fs_layers
WHERE ` + s.scope.And(`(root_layer_id = ? OR (root_layer_id = '' AND layer_id = ?))`) + `
ORDER BY updated_at DESC, layer_id`
	rows, err := s.db.QueryContext(ctx, query, s.scope.Args(rootID, rootID)...)
	if err != nil {
		return nil, fmt.Errorf("list fs layers by root %s: %w", rootID, err)
	}
	defer func() { _ = rows.Close() }()
	return s.scanFSLayers(ctx, rows, "list fs layers by root", withTags)
}

func (s *Store) ResolveFSLayerRef(ctx context.Context, ref string) (*FSLayer, error) {
	ref = strings.TrimSpace(ref)
	if ref == "" {
		return nil, fmt.Errorf("fs layer ref is required")
	}
	layer, err := s.GetFSLayer(ctx, ref)
	if err == nil {
		return layer, nil
	}
	if !errors.Is(err, ErrNotFound) {
		return nil, err
	}
	if !strings.HasPrefix(ref, "tag:") {
		layers, err := s.listFSLayersByName(ctx, ref)
		if err != nil {
			return nil, err
		}
		if len(layers) == 1 {
			return &layers[0], nil
		}
		if len(layers) > 1 {
			return nil, fmt.Errorf("%w: name %q matched %d layers", ErrFSLayerRefAmbiguous, ref, len(layers))
		}
	}
	key, value, hasValue, ok := parseFSLayerTagRef(ref)
	if !ok {
		return nil, ErrNotFound
	}
	layers, err := s.listFSLayersByTag(ctx, key, value, hasValue)
	if err != nil {
		return nil, err
	}
	if len(layers) == 0 {
		return nil, ErrNotFound
	}
	if len(layers) > 1 {
		return nil, fmt.Errorf("%w: tag %q matched %d layers", ErrFSLayerRefAmbiguous, ref, len(layers))
	}
	return &layers[0], nil
}

func (s *Store) SetFSLayerTags(ctx context.Context, layerID string, tags map[string]string) error {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return fmt.Errorf("fs layer id is required")
	}
	normalized, err := normalizeFSLayerTags(tags)
	if err != nil {
		return err
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin set fs layer tags transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	var existing string
	if err := tx.QueryRowContext(ctx, `SELECT layer_id FROM fs_layers WHERE `+s.scope.And(`layer_id = ?`), s.scope.Args(layerID)...).Scan(&existing); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return fmt.Errorf("read fs layer %s: %w", layerID, err)
	}
	if err := s.replaceFSLayerTagsTx(ctx, tx, layerID, normalized); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit set fs layer tags %s: %w", layerID, err)
	}
	return nil
}

func (s *Store) GetFSLayerTags(ctx context.Context, layerID string) (map[string]string, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	rows, err := s.db.QueryContext(ctx, `
SELECT tag_key, tag_value
FROM fs_layer_tags
WHERE `+s.scope.And(`layer_id = ?`)+`
ORDER BY tag_key`, s.scope.Args(layerID)...)
	if err != nil {
		return nil, fmt.Errorf("list fs layer tags: %w", err)
	}
	defer func() { _ = rows.Close() }()
	out := make(map[string]string)
	for rows.Next() {
		var key, value string
		if err := rows.Scan(&key, &value); err != nil {
			return nil, fmt.Errorf("scan fs layer tag: %w", err)
		}
		out[key] = value
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list fs layer tags: %w", err)
	}
	return out, nil
}

func (s *Store) SetFSLayerState(ctx context.Context, layerID string, state FSLayerState) error {
	if strings.TrimSpace(layerID) == "" {
		return fmt.Errorf("fs layer id is required")
	}
	if err := validateFSLayerState(state); err != nil {
		return err
	}
	query := `UPDATE fs_layers SET state = ?, updated_at = UTC_TIMESTAMP(3)`
	setArgs := []any{string(state)}
	if state == FSLayerStateSealed || state == FSLayerStateCommitting || state == FSLayerStateCommitted || state == FSLayerStateAbandoned || state == FSLayerStateConflicted {
		query += `, sealed_at = COALESCE(sealed_at, UTC_TIMESTAMP(3))`
	}
	query += ` WHERE ` + s.scope.And(`layer_id = ?`)
	args := append(setArgs, s.scope.Args(layerID)...)
	res, err := s.db.ExecContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("set fs layer state %s: %w", layerID, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		if _, err := s.GetFSLayer(ctx, layerID); err != nil {
			return err
		}
	}
	return nil
}

func (s *Store) SetFSLayerStateIf(ctx context.Context, layerID string, from []FSLayerState, to FSLayerState) error {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return fmt.Errorf("fs layer id is required")
	}
	if len(from) == 0 {
		return fmt.Errorf("fs layer source states are required")
	}
	if err := validateFSLayerState(to); err != nil {
		return err
	}
	setArgs := []any{string(to)}
	query := `UPDATE fs_layers SET state = ?, updated_at = UTC_TIMESTAMP(3)`
	if to == FSLayerStateSealed || to == FSLayerStateCommitting || to == FSLayerStateCommitted || to == FSLayerStateAbandoned || to == FSLayerStateConflicted {
		query += `, sealed_at = COALESCE(sealed_at, UTC_TIMESTAMP(3))`
	}
	query += ` WHERE ` + s.scope.And(`layer_id = ? AND state IN (`)
	whereArgs := []any{layerID}
	for i, state := range from {
		if err := validateFSLayerState(state); err != nil {
			return err
		}
		if i > 0 {
			query += `, `
		}
		query += `?`
		whereArgs = append(whereArgs, string(state))
	}
	query += `)`
	args := append(setArgs, s.scope.Args(whereArgs...)...)
	res, err := s.db.ExecContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("set fs layer state %s: %w", layerID, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n > 0 {
		return nil
	}
	if _, err := s.GetFSLayer(ctx, layerID); err != nil {
		return err
	}
	return ErrFSLayerStateConflict
}

// FinalizeFSLayerCommit flips committing → committed and sets commit_applied.
// Logical DELETE may later abandon the row; the column stays so hop landing
// can still see the flatten.
func (s *Store) FinalizeFSLayerCommit(ctx context.Context, layerID string) error {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return fmt.Errorf("fs layer id is required")
	}
	res, err := s.db.ExecContext(ctx, `
UPDATE fs_layers
SET state = ?, commit_applied = 1, sealed_at = COALESCE(sealed_at, UTC_TIMESTAMP(3)), updated_at = UTC_TIMESTAMP(3)
WHERE `+s.scope.And(`layer_id = ? AND state = ?`),
		append([]any{string(FSLayerStateCommitted)},
			s.scope.Args(layerID, string(FSLayerStateCommitting))...)...)
	if err != nil {
		return fmt.Errorf("finalize fs layer commit %s: %w", layerID, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n > 0 {
		return nil
	}
	if _, err := s.GetFSLayer(ctx, layerID); err != nil {
		return err
	}
	return ErrFSLayerStateConflict
}

func (s *Store) BeginFSLayerCommit(ctx context.Context, layerID string) error {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return fmt.Errorf("fs layer id is required")
	}
	res, err := s.db.ExecContext(ctx, `
UPDATE fs_layers
SET state = ?, sealed_at = COALESCE(sealed_at, UTC_TIMESTAMP(3)), updated_at = UTC_TIMESTAMP(3)
WHERE `+s.scope.And(`layer_id = ? AND state IN (?, ?)`),
		append([]any{string(FSLayerStateCommitting)},
			s.scope.Args(layerID, string(FSLayerStateActive), string(FSLayerStateSealed))...)...)
	if err != nil {
		return fmt.Errorf("begin fs layer commit %s: %w", layerID, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n > 0 {
		return nil
	}
	if _, err := s.GetFSLayer(ctx, layerID); err != nil {
		return err
	}
	return ErrFSLayerStateConflict
}

func (s *Store) RollbackFSLayer(ctx context.Context, layerID string) error {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return fmt.Errorf("fs layer id is required")
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin rollback fs layer transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	// Flip the layer state to abandoned. The UPDATE is conditional on the
	// current state being one of the rollback-eligible states, mirroring
	// SetFSLayerStateIf but inside this transaction so the synthetic event
	// is committed atomically with the state transition.
	res, err := tx.ExecContext(ctx, `
UPDATE fs_layers
SET state = ?, updated_at = UTC_TIMESTAMP(3), sealed_at = COALESCE(sealed_at, UTC_TIMESTAMP(3))
WHERE `+s.scope.And(`layer_id = ? AND state IN (?, ?, ?)`),
		append([]any{string(FSLayerStateAbandoned)},
			s.scope.Args(layerID, string(FSLayerStateActive), string(FSLayerStateSealed), string(FSLayerStateConflicted))...)...)
	if err != nil {
		return fmt.Errorf("rollback fs layer %s: %w", layerID, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("rollback fs layer %s rows affected: %w", layerID, err)
	}
	if n == 0 {
		// No row was updated: either the layer does not exist, or it was
		// already in a terminal state. Distinguish the two and, for the
		// already-abandoned case, return nil idempotently WITHOUT
		// re-emitting the rollback event.
		var state string
		if err := tx.QueryRowContext(ctx, `SELECT state FROM fs_layers WHERE `+s.scope.And(`layer_id = ?`)+` FOR UPDATE`, s.scope.Args(layerID)...).Scan(&state); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrNotFound
			}
			return fmt.Errorf("read fs layer %s: %w", layerID, err)
		}
		if FSLayerState(state) == FSLayerStateAbandoned {
			return nil
		}
		return ErrFSLayerStateConflict
	}

	// Allocate a seq strictly greater than any existing event/entry seq so
	// the FUSE layer-event watcher (seq > lastSeq) observes the rollback.
	// Rollback writes no fs_layer_entries row, so the shared allocator
	// consults both tables under the already-held layer row lock.
	seq, err := s.nextFSLayerSeq(ctx, tx, layerID)
	if err != nil {
		return err
	}
	eventID := compactFSLayerEventID(fmt.Sprintf("%s:rollback:%d", layerID, seq))
	if _, err := tx.ExecContext(ctx, `
INSERT INTO fs_layer_events (`+s.scope.InsCols(`event_id, layer_id, seq, actor_id, op, path, created_at`)+`)
VALUES (`+s.scope.InsVals(`?, ?, ?, '', ?, '/', UTC_TIMESTAMP(3)`)+`)`,
		s.scope.Args(eventID, layerID, seq, fsLayerEventOpRollback)...); err != nil {
		return fmt.Errorf("insert fs layer rollback event %s: %w", layerID, err)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit rollback fs layer %s: %w", layerID, err)
	}
	return nil
}

func (s *Store) ListFSLayerEntryLog(ctx context.Context, layerID string) ([]FSLayerEntry, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	return s.listFSLayerEntryLog(ctx, layerID, nil)
}

func (s *Store) ListFSLayerEntryLogAtSeq(ctx context.Context, layerID string, maxSeq int64) ([]FSLayerEntry, error) {
	if maxSeq < 0 {
		return nil, fmt.Errorf("fs layer max seq must be non-negative")
	}
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	return s.listFSLayerEntryLog(ctx, layerID, &maxSeq)
}

func (s *Store) listFSLayerEntryLog(ctx context.Context, layerID string, maxSeq *int64) ([]FSLayerEntry, error) {
	where := "WHERE " + s.scope.And("layer_id = ?")
	args := s.scope.Args(layerID)
	if maxSeq != nil {
		where += " AND entry_seq <= ?"
		args = append(args, *maxSeq)
	}
	rows, err := s.db.QueryContext(ctx, `
SELECT layer_id, path, path_hash, parent_path, parent_path_hash, name, op, kind, base_inode_id, base_revision,
	storage_type, storage_ref, storage_ref_hash, storage_encryption_mode, storage_encryption_key_id,
	content_blob, content_type, content_text, checksum_sha256, size_bytes, mode, entry_seq, created_at, updated_at
FROM fs_layer_entries
`+where+`
ORDER BY entry_seq ASC, path ASC`, args...)
	if err != nil {
		return nil, fmt.Errorf("list fs layer entry log: %w", err)
	}
	defer func() { _ = rows.Close() }()
	var out []FSLayerEntry
	for rows.Next() {
		entry, err := scanFSLayerEntry(rows)
		if err != nil {
			return nil, fmt.Errorf("scan fs layer entry log: %w", err)
		}
		out = append(out, *entry)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list fs layer entry log: %w", err)
	}
	return out, nil
}

// nextFSLayerSeq returns the next shared seq for this layer's entries and
// events. The caller must already hold the fs_layers row lock so no
// concurrent writer can insert between the two current reads.
//
// A plain MAX() after FOR UPDATE is still a snapshot read in TiDB (and
// MySQL REPEATABLE READ) pessimistic transactions: lock wait does not
// advance start_ts, so two writers that began together both observe N
// and both insert seq=N+1. That is legal for fs_layer_entries (PK is
// layer_id, path_hash, entry_seq) when the paths differ, but
// uk_fs_layer_events_seq(layer_id, seq) then returns Error 1062.
// ORDER BY ... LIMIT 1 FOR UPDATE is a current read of the latest
// committed row. Events and entries share one namespace, so we take
// GREATEST of both heads (rollback writes an event with no entry).
func (s *Store) nextFSLayerSeq(ctx context.Context, tx *sql.Tx, layerID string) (int64, error) {
	entrySeq, err := scanLockedInt64(tx.QueryRowContext(ctx, `
SELECT entry_seq FROM fs_layer_entries
WHERE `+s.scope.And(`layer_id = ?`)+`
ORDER BY entry_seq DESC
LIMIT 1
FOR UPDATE`, s.scope.Args(layerID)...))
	if err != nil {
		return 0, fmt.Errorf("read fs layer max entry seq: %w", err)
	}
	eventSeq, err := scanLockedInt64(tx.QueryRowContext(ctx, `
SELECT seq FROM fs_layer_events
WHERE `+s.scope.And(`layer_id = ?`)+`
ORDER BY seq DESC
LIMIT 1
FOR UPDATE`, s.scope.Args(layerID)...))
	if err != nil {
		return 0, fmt.Errorf("read fs layer max event seq: %w", err)
	}
	maxSeq := entrySeq
	if eventSeq > maxSeq {
		maxSeq = eventSeq
	}
	return maxSeq + 1, nil
}

func scanLockedInt64(row *sql.Row) (int64, error) {
	var v sql.NullInt64
	if err := row.Scan(&v); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return 0, nil
		}
		return 0, err
	}
	if !v.Valid {
		return 0, nil
	}
	return v.Int64, nil
}

func (s *Store) UpsertFSLayerEntry(ctx context.Context, entry *FSLayerEntry) error {
	if entry == nil {
		return fmt.Errorf("fs layer entry is required")
	}
	if err := normalizeFSLayerEntry(entry); err != nil {
		return err
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin upsert fs layer entry transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	if err := s.upsertFSLayerEntryTx(ctx, tx, entry); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit upsert fs layer entry %s:%s: %w: %w", entry.LayerID, entry.Path, ErrCommitOutcomeUnknown, err)
	}
	return nil
}

func (s *Store) upsertFSLayerEntryTx(ctx context.Context, tx *sql.Tx, entry *FSLayerEntry) error {
	var (
		baseRoot string
		state    string
	)
	if err := tx.QueryRowContext(ctx, `SELECT base_root_path, state FROM fs_layers WHERE `+s.scope.And(`layer_id = ?`)+` FOR UPDATE`, s.scope.Args(entry.LayerID)...).Scan(&baseRoot, &state); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return fmt.Errorf("read fs layer %s: %w", entry.LayerID, err)
	}
	if FSLayerState(state) != FSLayerStateActive {
		return ErrFSLayerStateConflict
	}
	if err := validateFSLayerEntryWithinBaseRoot(entry, baseRoot); err != nil {
		return err
	}
	// Synchronous base-revision CAS for file content upserts. Without this
	// gate, an upsert based on a stale base revision is persisted silently
	// (e.g. a FUSE lazy flush that fetched retained parts after a sibling
	// commit): the spliced bytes enter the layer, the writer's dirty state
	// is cleared, and the mismatch only surfaces at layer-commit preflight —
	// escalating the whole layer to Conflicted. Reject here, while the
	// writer can still retry. base_revision <= 0 means "no base claim"
	// (new or layer-local file) and skips the check, matching the commit
	// preflight's semantics.
	if entry.Op == FSLayerEntryOpUpsert && entry.Kind == FSLayerEntryKindFile && entry.BaseRevision > 0 {
		bn, statErr := s.StatTx(ctx, tx, entry.Path)
		switch {
		case errors.Is(statErr, ErrNotFound) && entry.BaseInodeID != "":
			// Dest-space copy-up: dest is not on main yet. The claim is the
			// remapped live-main identity; commit preflight checks it after
			// the ancestor hop lands.
		case errors.Is(statErr, ErrNotFound):
			return fmt.Errorf("%w: %w for %s (base_revision=%d)", ErrRevisionConflict, ErrFSLayerBasePathMissing, entry.Path, entry.BaseRevision)
		case statErr != nil:
			return fmt.Errorf("stat base file for layer upsert %s: %w", entry.Path, statErr)
		default:
			var currentRev int64
			if bn.File != nil {
				currentRev = bn.File.Revision
			}
			if bn.File == nil || currentRev != entry.BaseRevision {
				return fmt.Errorf("%w: %w for %s (current=%d, want=%d)", ErrRevisionConflict, ErrFSLayerBaseRevisionChanged, entry.Path, currentRev, entry.BaseRevision)
			}
			if entry.BaseInodeID != "" && bn.File.FileID != entry.BaseInodeID {
				return fmt.Errorf("%w: %w for %s", ErrRevisionConflict, ErrFSLayerBaseInodeChanged, entry.Path)
			}
		}
	}
	if entry.EntrySeq <= 0 {
		seq, err := s.nextFSLayerSeq(ctx, tx, entry.LayerID)
		if err != nil {
			return err
		}
		entry.EntrySeq = seq
	}
	_, err := tx.ExecContext(ctx, `
INSERT INTO fs_layer_entries (
	`+s.scope.InsCols(`layer_id, path, path_hash, parent_path, parent_path_hash, name, op, kind, base_inode_id, base_revision,
	storage_type, storage_ref, storage_ref_hash, storage_encryption_mode, storage_encryption_key_id,
	content_blob, content_type, content_text, checksum_sha256, size_bytes, mode, entry_seq, created_at, updated_at`)+`
) VALUES (`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, UTC_TIMESTAMP(3), UTC_TIMESTAMP(3)`)+`)`,
		s.scope.Args(entry.LayerID, entry.Path, entry.PathHash, entry.ParentPath, entry.ParentPathHash, entry.Name,
			string(entry.Op), string(entry.Kind), entry.BaseInodeID, entry.BaseRevision, entry.StorageType,
			entry.StorageRef, entry.StorageRefHash, fileStorageEncryptionModeForWrite(entry.StorageEncryptionMode),
			storageEncryptionKeyIDForWrite(entry.StorageEncryptionMode, entry.StorageEncryptionKeyID),
			nilBytes(entry.ContentBlob), nullStr(entry.ContentType),
			nullStr(entry.ContentText), entry.ChecksumSHA256, entry.SizeBytes, entry.Mode, entry.EntrySeq)...)
	if err != nil {
		return fmt.Errorf("upsert fs layer entry %s:%s: %w", entry.LayerID, entry.Path, err)
	}
	eventID := compactFSLayerEventID(fmt.Sprintf("%s:%d", entry.LayerID, entry.EntrySeq))
	if _, err := tx.ExecContext(ctx, `
INSERT INTO fs_layer_events (`+s.scope.InsCols(`event_id, layer_id, seq, actor_id, op, path, created_at`)+`)
VALUES (`+s.scope.InsVals(`?, ?, ?, '', ?, ?, UTC_TIMESTAMP(3)`)+`)`,
		s.scope.Args(eventID, entry.LayerID, entry.EntrySeq, string(entry.Op), entry.Path)...); err != nil {
		return fmt.Errorf("insert fs layer event %s:%s: %w", entry.LayerID, entry.Path, err)
	}
	// Held under the fs_layers row lock so a racing fork's parent
	// FOR UPDATE waits here. Used to prove fork origin_seq is a
	// current read of the write that commits during that wait.
	failpoint.InjectCall("fsLayerUpsertBeforeCommit", entry.LayerID)
	return nil
}

func (s *Store) PublishFSLayerUpload(ctx context.Context, uploadID string, entry *FSLayerEntry) error {
	_, err := s.PublishFSLayerUploadResult(ctx, uploadID, entry)
	return err
}

// PublishFSLayerUploadResult publishes an upload and reports whether this call
// performed the UPLOADING to COMPLETED transition. Retries that find the
// existing published entry return false.
func (s *Store) PublishFSLayerUploadResult(ctx context.Context, uploadID string, entry *FSLayerEntry) (bool, error) {
	if entry == nil {
		return false, fmt.Errorf("fs layer entry is required")
	}
	if err := normalizeFSLayerEntry(entry); err != nil {
		return false, err
	}
	published := false
	err := s.InTx(ctx, func(tx *sql.Tx) error {
		var (
			targetKind       UploadTargetKind
			layerID          string
			targetPath       string
			baseInodeID      string
			mode             uint32
			s3Key            string
			totalSize        int64
			expectedRevision sql.NullInt64
			status           UploadStatus
		)
		err := tx.QueryRowContext(ctx, `SELECT target_kind, layer_id, target_path, base_inode_id, mode,
			s3_key, total_size, expected_revision, status
			FROM uploads WHERE `+s.scope.And(`upload_id = ?`)+` FOR UPDATE`, s.scope.Args(uploadID)...).Scan(
			&targetKind, &layerID, &targetPath, &baseInodeID, &mode,
			&s3Key, &totalSize, &expectedRevision, &status,
		)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return fmt.Errorf("read fs layer upload %s: %w", uploadID, err)
		}
		baseRevision := int64(0)
		if expectedRevision.Valid {
			baseRevision = expectedRevision.Int64
		}
		if targetKind != UploadTargetFSLayer || layerID != entry.LayerID || targetPath != entry.Path ||
			baseInodeID != entry.BaseInodeID || baseRevision != entry.BaseRevision || mode != entry.Mode ||
			s3Key != entry.StorageRef || totalSize != entry.SizeBytes || entry.StorageType != string(StorageS3) {
			return fmt.Errorf("%w: fs layer upload metadata does not match completed object", ErrFSLayerInvalidArgument)
		}
		if status == UploadCompleted {
			var entrySeq int64
			err := tx.QueryRowContext(ctx, `SELECT entry_seq FROM fs_layer_entries
				WHERE `+s.scope.And(`layer_id = ? AND path_hash = ? AND path = ? AND storage_ref_hash = ? AND storage_ref = ?`)+`
				ORDER BY entry_seq DESC LIMIT 1`, s.scope.Args(entry.LayerID, entry.PathHash, entry.Path, entry.StorageRefHash, entry.StorageRef)...).Scan(&entrySeq)
			if errors.Is(err, sql.ErrNoRows) {
				return ErrUploadNotActive
			}
			if err != nil {
				return fmt.Errorf("read published fs layer upload %s: %w", uploadID, err)
			}
			entry.EntrySeq = entrySeq
			return nil
		}
		if status != UploadUploading {
			return ErrUploadNotActive
		}
		if err := s.upsertFSLayerEntryTx(ctx, tx, entry); err != nil {
			return err
		}
		res, err := tx.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)...)...)
		if err != nil {
			return fmt.Errorf("complete fs layer upload %s: %w", uploadID, err)
		}
		rowsAffected, err := res.RowsAffected()
		if err != nil {
			return fmt.Errorf("complete fs layer upload %s rows affected: %w", uploadID, err)
		}
		if rowsAffected != 1 {
			return ErrUploadNotActive
		}
		published = true
		return nil
	})
	if err != nil {
		return false, fmt.Errorf("publish fs layer upload %s: %w", uploadID, err)
	}
	return published, nil
}

// FSLayerHasOverlayDrafts reports whether the layer log (optionally capped)
// contains rename, whiteout, or chmod entries. Parentless GETs without those
// ops can use a path-hash point lookup instead of replaying the full log.
func (s *Store) FSLayerHasOverlayDrafts(ctx context.Context, layerID string, tipMax *int64) (bool, error) {
	return s.fsLayerHasOverlayOps(ctx, layerID, tipMax, FSLayerEntryOpRename, FSLayerEntryOpWhiteout, FSLayerEntryOpChmod)
}

// FSLayerChainHasRemapDrafts reports whether any visible chain log contains
// rename or whiteout (ops that can hide or remap a dest-space path). Chmod
// is ignored: it cannot remap. Ancestor frames are capped at origin_seq.
func (s *Store) FSLayerChainHasRemapDrafts(ctx context.Context, layerID string, tipMax *int64) (bool, error) {
	chain, err := s.ListFSLayerChain(ctx, layerID)
	if err != nil {
		return false, err
	}
	chain = applyTipLimit(chain, tipMax)
	for i := range chain {
		var capSeq *int64
		if chain[i].LimitSeq >= 0 {
			seq := chain[i].LimitSeq
			capSeq = &seq
		}
		has, err := s.fsLayerHasOverlayOps(ctx, chain[i].Layer.LayerID, capSeq, FSLayerEntryOpRename, FSLayerEntryOpWhiteout)
		if err != nil {
			return false, err
		}
		if has {
			return true, nil
		}
	}
	return false, nil
}

func (s *Store) fsLayerHasOverlayOps(ctx context.Context, layerID string, tipMax *int64, ops ...FSLayerEntryOp) (bool, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return false, fmt.Errorf("fs layer id is required")
	}
	if len(ops) == 0 {
		return false, nil
	}
	placeholders := make([]string, len(ops))
	args := make([]any, 0, 2+len(ops))
	args = append(args, s.scope.Args(layerID)...)
	for i, op := range ops {
		placeholders[i] = "?"
		args = append(args, string(op))
	}
	query := `
SELECT 1 FROM fs_layer_entries
WHERE ` + s.scope.And(`layer_id = ? AND op IN (`+strings.Join(placeholders, ", ")+`)`)
	if tipMax != nil {
		query += `
AND entry_seq <= ?`
		args = append(args, *tipMax)
	}
	query += `
LIMIT 1`
	var one int
	err := s.db.QueryRowContext(ctx, query, args...).Scan(&one)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("check fs layer overlay drafts %s: %w", layerID, err)
	}
	return true, nil
}

func (s *Store) GetFSLayerEntry(ctx context.Context, layerID, path string) (*FSLayerEntry, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	canonical, err := canonicalFSLayerLookupPath(path)
	if err != nil {
		return nil, err
	}
	row := s.db.QueryRowContext(ctx, `
SELECT layer_id, path, path_hash, parent_path, parent_path_hash, name, op, kind, base_inode_id, base_revision,
	storage_type, storage_ref, storage_ref_hash, storage_encryption_mode, storage_encryption_key_id,
	content_blob, content_type, content_text, checksum_sha256, size_bytes, mode, entry_seq, created_at, updated_at
FROM fs_layer_entries
WHERE `+s.scope.And(`layer_id = ? AND path_hash = ? AND path = ?`)+`
ORDER BY entry_seq DESC
LIMIT 1`, s.scope.Args(layerID, fsLayerPathHash(canonical), canonical)...)
	return scanFSLayerEntry(row)
}

func (s *Store) GetFSLayerEntryAtSeq(ctx context.Context, layerID, path string, maxSeq int64) (*FSLayerEntry, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	if maxSeq < 0 {
		return nil, fmt.Errorf("fs layer max seq must be non-negative")
	}
	canonical, err := canonicalFSLayerLookupPath(path)
	if err != nil {
		return nil, err
	}
	row := s.db.QueryRowContext(ctx, `
SELECT layer_id, path, path_hash, parent_path, parent_path_hash, name, op, kind, base_inode_id, base_revision,
	storage_type, storage_ref, storage_ref_hash, storage_encryption_mode, storage_encryption_key_id,
	content_blob, content_type, content_text, checksum_sha256, size_bytes, mode, entry_seq, created_at, updated_at
FROM fs_layer_entries
WHERE `+s.scope.And(`layer_id = ? AND path_hash = ? AND path = ? AND entry_seq <= ?`)+`
ORDER BY entry_seq DESC
LIMIT 1`, s.scope.Args(layerID, fsLayerPathHash(canonical), canonical, maxSeq)...)
	return scanFSLayerEntry(row)
}

func (s *Store) ListFSLayerEntries(ctx context.Context, layerID string) ([]FSLayerEntry, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	return s.listFSLayerEntriesLatest(ctx, layerID, nil)
}

func (s *Store) ListFSLayerEntriesAtSeq(ctx context.Context, layerID string, maxSeq int64) ([]FSLayerEntry, error) {
	if maxSeq < 0 {
		return nil, fmt.Errorf("fs layer max seq must be non-negative")
	}
	return s.listFSLayerEntriesLatest(ctx, layerID, &maxSeq)
}

func (s *Store) listFSLayerEntriesLatest(ctx context.Context, layerID string, maxSeq *int64) ([]FSLayerEntry, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	where := "WHERE " + s.scope.And("layer_id = ?")
	args := s.scope.Args(layerID)
	if maxSeq != nil {
		where += " AND entry_seq <= ?"
		args = append(args, *maxSeq)
	}
	// Both the derived latest table and the outer fs_layer_entries alias get
	// an fs_id predicate in shared shape: the same (layer_id, path) pair can
	// exist under another fs_id, and an unfiltered MAX(entry_seq) could then
	// hide this tenant's newest entry.
	query := `
SELECT e.layer_id, e.path, e.path_hash, e.parent_path, e.parent_path_hash, e.name, e.op, e.kind, e.base_inode_id, e.base_revision,
	e.storage_type, e.storage_ref, e.storage_ref_hash, e.storage_encryption_mode, e.storage_encryption_key_id,
	e.content_blob, e.content_type, e.content_text, e.checksum_sha256, e.size_bytes, e.mode, e.entry_seq, e.created_at, e.updated_at
FROM fs_layer_entries e
JOIN (
		SELECT layer_id, path_hash, path, MAX(entry_seq) AS entry_seq
		FROM fs_layer_entries
		` + where + `
		GROUP BY layer_id, path_hash, path
	) latest ON latest.layer_id = e.layer_id AND latest.path_hash = e.path_hash AND latest.path = e.path AND latest.entry_seq = e.entry_seq
WHERE ` + s.scope.AndAs("e", "e.layer_id = ?") + `
ORDER BY e.entry_seq, e.path`
	args = append(args, s.scope.Args(layerID)...)
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list fs layer entries: %w", err)
	}
	defer func() { _ = rows.Close() }()
	var out []FSLayerEntry
	for rows.Next() {
		entry, err := scanFSLayerEntry(rows)
		if err != nil {
			return nil, fmt.Errorf("scan fs layer entry: %w", err)
		}
		out = append(out, *entry)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list fs layer entries: %w", err)
	}
	return out, nil
}

func (s *Store) CreateFSLayerCheckpoint(ctx context.Context, checkpoint *FSLayerCheckpoint) error {
	if checkpoint == nil {
		return fmt.Errorf("fs layer checkpoint is required")
	}
	checkpointID, err := normalizeFSLayerCheckpointID(checkpoint.CheckpointID)
	if err != nil {
		return err
	}
	checkpoint.CheckpointID = checkpointID
	checkpoint.LayerID = strings.TrimSpace(checkpoint.LayerID)
	if checkpoint.LayerID == "" {
		return fmt.Errorf("%w: layer_id is required", ErrFSLayerInvalidArgument)
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin fs layer checkpoint transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	var existingLayerID string
	if err := tx.QueryRowContext(ctx, `SELECT layer_id FROM fs_layers WHERE `+s.scope.And(`layer_id = ?`)+` FOR UPDATE`, s.scope.Args(checkpoint.LayerID)...).Scan(&existingLayerID); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return fmt.Errorf("read fs layer %s: %w", checkpoint.LayerID, err)
	}
	seq, err := scanLockedInt64(tx.QueryRowContext(ctx, `
SELECT entry_seq FROM fs_layer_entries
WHERE `+s.scope.And(`layer_id = ?`)+`
ORDER BY entry_seq DESC
LIMIT 1
FOR UPDATE`, s.scope.Args(checkpoint.LayerID)...))
	if err != nil {
		return fmt.Errorf("read fs layer max seq: %w", err)
	}
	checkpoint.DurableSeq = seq
	_, err = tx.ExecContext(ctx, `
INSERT INTO fs_layer_checkpoints (`+s.scope.InsCols(`checkpoint_id, layer_id, durable_seq, label, created_at`)+`)
VALUES (`+s.scope.InsVals(`?, ?, ?, ?, UTC_TIMESTAMP(3)`)+`)`,
		s.scope.Args(checkpoint.CheckpointID, checkpoint.LayerID, checkpoint.DurableSeq, checkpoint.Label)...)
	if err != nil {
		return fmt.Errorf("create fs layer checkpoint %s: %w", checkpoint.CheckpointID, err)
	}
	if _, err := tx.ExecContext(ctx, `
UPDATE fs_layers SET durable_seq = GREATEST(durable_seq, ?), updated_at = UTC_TIMESTAMP(3)
WHERE `+s.scope.And(`layer_id = ?`),
		append([]any{checkpoint.DurableSeq}, s.scope.Args(checkpoint.LayerID)...)...); err != nil {
		return fmt.Errorf("update fs layer durable seq %s: %w", checkpoint.LayerID, err)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit fs layer checkpoint %s: %w", checkpoint.CheckpointID, err)
	}
	return nil
}

func (s *Store) GetFSLayerCheckpoint(ctx context.Context, checkpointID string) (*FSLayerCheckpoint, error) {
	checkpointID, err := normalizeFSLayerCheckpointID(checkpointID)
	if err != nil {
		return nil, err
	}
	row := s.db.QueryRowContext(ctx, `
SELECT checkpoint_id, layer_id, durable_seq, label, created_at
FROM fs_layer_checkpoints
WHERE `+s.scope.And(`checkpoint_id = ?`), s.scope.Args(checkpointID)...)
	return scanFSLayerCheckpoint(row)
}

func normalizeFSLayerCheckpointID(checkpointID string) (string, error) {
	checkpointID = strings.TrimSpace(checkpointID)
	if checkpointID == "" {
		return "", fmt.Errorf("%w: checkpoint_id is required", ErrFSLayerInvalidArgument)
	}
	if len(checkpointID) > FSLayerCheckpointIDMaxBytes {
		return "", fmt.Errorf("%w: checkpoint_id exceeds %d bytes", ErrFSLayerInvalidArgument, FSLayerCheckpointIDMaxBytes)
	}
	return checkpointID, nil
}

func (s *Store) ListFSLayerEvents(ctx context.Context, layerID string, since int64, limit int) ([]FSLayerEvent, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	if limit <= 0 || limit > 1000 {
		limit = 1000
	}
	rows, err := s.db.QueryContext(ctx, `
SELECT event_id, layer_id, seq, actor_id, op, path, created_at
FROM fs_layer_events
WHERE `+s.scope.And(`layer_id = ? AND seq > ?`)+`
ORDER BY seq ASC
LIMIT ?`, s.scope.Args(layerID, since, limit)...)
	if err != nil {
		return nil, fmt.Errorf("list fs layer events %s: %w", layerID, err)
	}
	defer func() { _ = rows.Close() }()
	var out []FSLayerEvent
	for rows.Next() {
		var ev FSLayerEvent
		if err := rows.Scan(&ev.EventID, &ev.LayerID, &ev.Seq, &ev.ActorID, &ev.Op, &ev.Path, &ev.CreatedAt); err != nil {
			return nil, err
		}
		ev.CreatedAt = ev.CreatedAt.UTC()
		out = append(out, ev)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if out == nil {
		out = []FSLayerEvent{}
	}
	return out, nil
}

func scanFSLayer(row interface{ Scan(dest ...any) error }) (*FSLayer, error) {
	var layer FSLayer
	var state, durability string
	var sealedAt sql.NullTime
	var commitApplied int64
	if err := row.Scan(
		&layer.LayerID, &layer.BaseRootPath, &layer.Name, &state, &durability, &layer.ActorID, &layer.DurableSeq,
		&layer.ParentLayerID, &layer.OriginSeq, &layer.OriginCheckpointID, &layer.RootLayerID, &layer.Depth, &layer.Origin,
		&commitApplied, &layer.CreatedAt, &layer.UpdatedAt, &sealedAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	layer.State = FSLayerState(state)
	layer.DurabilityMode = FSLayerDurabilityMode(durability)
	layer.CommitApplied = commitApplied != 0
	if layer.RootLayerID == "" {
		layer.RootLayerID = layer.LayerID
	}
	if layer.Origin == "" {
		layer.Origin = FSLayerOriginCreate
	}
	layer.CreatedAt = layer.CreatedAt.UTC()
	layer.UpdatedAt = layer.UpdatedAt.UTC()
	if sealedAt.Valid {
		t := sealedAt.Time.UTC()
		layer.SealedAt = &t
	}
	return &layer, nil
}

func (s *Store) listFSLayersByName(ctx context.Context, name string) ([]FSLayer, error) {
	rows, err := s.db.QueryContext(ctx, `
SELECT `+fsLayerSelectCols+`
FROM fs_layers
WHERE `+s.scope.And(`name = ?`)+`
ORDER BY updated_at DESC, layer_id`, s.scope.Args(name)...)
	if err != nil {
		return nil, fmt.Errorf("list fs layers by name: %w", err)
	}
	defer func() { _ = rows.Close() }()
	return s.scanFSLayersWithTags(ctx, rows, "list fs layers by name")
}

// fsLayerJoinArgs prepends the two fs_id bind arguments used when a query
// filters both fs_layer join aliases in shared shape; standalone shape
// leaves args unchanged.
func (s *Store) fsLayerJoinArgs(args ...any) []any {
	if s.scope.Shared() {
		return append([]any{s.scope.FsID(), s.scope.FsID()}, args...)
	}
	return args
}

func (s *Store) listFSLayersByTag(ctx context.Context, key, value string, hasValue bool) ([]FSLayer, error) {
	// The same layer_id can exist under two fs_ids in shared shape, so both
	// join aliases get an fs_id predicate: one-sided filtering could join a
	// tag row to another tenant's layer row (or duplicate the result rows).
	query := `
SELECT l.layer_id, l.base_root_path, l.name, l.state, l.durability_mode, l.actor_id, l.durable_seq,
	l.parent_layer_id, l.origin_seq, l.origin_checkpoint_id, l.root_layer_id, l.depth, l.origin,
	l.commit_applied, l.created_at, l.updated_at, l.sealed_at
FROM fs_layers l
JOIN fs_layer_tags t ON t.layer_id = l.layer_id
WHERE ` + s.scope.AndAs("l", s.scope.AndAs("t", "t.tag_key = ?"))
	args := s.fsLayerJoinArgs(key)
	if hasValue {
		query += ` AND t.tag_value = ?`
		args = append(args, value)
	}
	query += `
ORDER BY l.updated_at DESC, l.layer_id`
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list fs layers by tag: %w", err)
	}
	defer func() { _ = rows.Close() }()
	return s.scanFSLayersWithTags(ctx, rows, "list fs layers by tag")
}

func (s *Store) scanFSLayersWithTags(ctx context.Context, rows *sql.Rows, label string) ([]FSLayer, error) {
	return s.scanFSLayers(ctx, rows, label, true)
}

func (s *Store) scanFSLayers(ctx context.Context, rows *sql.Rows, label string, withTags bool) ([]FSLayer, error) {
	var out []FSLayer
	for rows.Next() {
		layer, err := scanFSLayer(rows)
		if err != nil {
			return nil, fmt.Errorf("scan fs layer: %w", err)
		}
		if withTags {
			if err := s.loadFSLayerTags(ctx, layer); err != nil {
				return nil, err
			}
		}
		out = append(out, *layer)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("%s: %w", label, err)
	}
	return out, nil
}

func (s *Store) loadFSLayerTags(ctx context.Context, layer *FSLayer) error {
	if layer == nil {
		return nil
	}
	tags, err := s.GetFSLayerTags(ctx, layer.LayerID)
	if err != nil {
		return err
	}
	layer.Tags = tags
	return nil
}

func (s *Store) replaceFSLayerTagsTx(ctx context.Context, tx *sql.Tx, layerID string, tags map[string]string) error {
	if _, err := tx.ExecContext(ctx, `DELETE FROM fs_layer_tags WHERE `+s.scope.And(`layer_id = ?`), s.scope.Args(layerID)...); err != nil {
		return fmt.Errorf("delete fs layer tags %s: %w", layerID, err)
	}
	for key, value := range tags {
		if _, err := tx.ExecContext(ctx, `
INSERT INTO fs_layer_tags (`+s.scope.InsCols(`layer_id, tag_key, tag_value, created_at`)+`)
VALUES (`+s.scope.InsVals(`?, ?, ?, UTC_TIMESTAMP(3)`)+`)`, s.scope.Args(layerID, key, value)...); err != nil {
			return fmt.Errorf("insert fs layer tag %s:%s: %w", layerID, key, err)
		}
	}
	return nil
}

func scanFSLayerEntry(row interface{ Scan(dest ...any) error }) (*FSLayerEntry, error) {
	var entry FSLayerEntry
	var op, kind string
	var contentBlob []byte
	var contentType, contentText, storageEncryptionMode, storageEncryptionKeyID sql.NullString
	var mode int64
	if err := row.Scan(
		&entry.LayerID, &entry.Path, &entry.PathHash, &entry.ParentPath, &entry.ParentPathHash,
		&entry.Name, &op, &kind, &entry.BaseInodeID, &entry.BaseRevision, &entry.StorageType,
		&entry.StorageRef, &entry.StorageRefHash, &storageEncryptionMode, &storageEncryptionKeyID,
		&contentBlob, &contentType, &contentText,
		&entry.ChecksumSHA256, &entry.SizeBytes, &mode, &entry.EntrySeq, &entry.CreatedAt, &entry.UpdatedAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	entry.Op = FSLayerEntryOp(op)
	entry.Kind = FSLayerEntryKind(kind)
	entry.ContentBlob = contentBlob
	entry.ContentType = contentType.String
	entry.ContentText = contentText.String
	entry.StorageEncryptionMode = StorageEncryptionMode(storageEncryptionMode.String)
	entry.StorageEncryptionKeyID = storageEncryptionKeyID.String
	entry.Mode = uint32(mode)
	entry.CreatedAt = entry.CreatedAt.UTC()
	entry.UpdatedAt = entry.UpdatedAt.UTC()
	return &entry, nil
}

func scanFSLayerCheckpoint(row interface{ Scan(dest ...any) error }) (*FSLayerCheckpoint, error) {
	var checkpoint FSLayerCheckpoint
	if err := row.Scan(
		&checkpoint.CheckpointID, &checkpoint.LayerID, &checkpoint.DurableSeq, &checkpoint.Label, &checkpoint.CreatedAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	checkpoint.CreatedAt = checkpoint.CreatedAt.UTC()
	return &checkpoint, nil
}

func normalizeFSLayerEntry(entry *FSLayerEntry) error {
	entry.LayerID = strings.TrimSpace(entry.LayerID)
	if entry.LayerID == "" {
		return fmt.Errorf("fs layer id is required")
	}
	if entry.Op == "" {
		entry.Op = FSLayerEntryOpUpsert
	}
	if err := validateFSLayerEntryOp(entry.Op); err != nil {
		return err
	}
	if entry.Kind == "" {
		switch entry.Op {
		case FSLayerEntryOpMkdir:
			entry.Kind = FSLayerEntryKindDir
		case FSLayerEntryOpSymlink:
			entry.Kind = FSLayerEntryKindSymlink
		default:
			entry.Kind = FSLayerEntryKindFile
		}
	}
	if err := validateFSLayerEntryKind(entry.Kind); err != nil {
		return err
	}
	var (
		p   string
		err error
	)
	if entry.Kind == FSLayerEntryKindDir || entry.Op == FSLayerEntryOpMkdir {
		p, err = pathutil.CanonicalizeDir(entry.Path)
	} else {
		p, err = pathutil.Canonicalize(entry.Path)
	}
	if err != nil {
		return fmt.Errorf("invalid fs layer entry path: %w", err)
	}
	entry.Path = p
	entry.PathHash = fsLayerPathHash(p)
	entry.ParentPath = pathutil.ParentPath(p)
	entry.ParentPathHash = fsLayerPathHash(entry.ParentPath)
	entry.Name = pathutil.BaseName(p)
	if entry.Mode == 0 {
		entry.Mode = 0o644
		if entry.Kind == FSLayerEntryKindDir {
			entry.Mode = 0o755
		}
	}
	if entry.StorageRefHash == "" && entry.StorageRef != "" {
		entry.StorageRefHash = StorageRefHash(entry.StorageRef)
	}
	if entry.StorageRef != "" {
		prefix := "layers/" + entry.LayerID + "/"
		objectID := strings.TrimPrefix(entry.StorageRef, prefix)
		if objectID == entry.StorageRef || objectID == "" || strings.Contains(objectID, "/") {
			return fmt.Errorf("fs layer S3 storage ref must belong to layer %q", entry.LayerID)
		}
	}
	return nil
}

func validateFSLayerEntryWithinBaseRoot(entry *FSLayerEntry, baseRoot string) error {
	root, err := pathutil.CanonicalizeDir(baseRoot)
	if err != nil {
		return fmt.Errorf("invalid fs layer base root: %w", err)
	}
	if !fsLayerPathWithinBaseRoot(entry.Path, root) {
		return fmt.Errorf("fs layer entry path %q is outside base root %q", entry.Path, root)
	}
	if entry.Op != FSLayerEntryOpRename {
		return nil
	}
	target := strings.TrimSpace(entry.ContentText)
	if target == "" && len(entry.ContentBlob) > 0 {
		target = strings.TrimSpace(string(entry.ContentBlob))
	}
	if target == "" {
		return nil
	}
	targetPath, err := canonicalFSLayerLookupPath(target)
	if err != nil {
		return fmt.Errorf("invalid fs layer rename target: %w", err)
	}
	if !fsLayerPathWithinBaseRoot(targetPath, root) {
		return fmt.Errorf("fs layer rename target %q is outside base root %q", targetPath, root)
	}
	return nil
}

func fsLayerPathWithinBaseRoot(p, baseRoot string) bool {
	if baseRoot == "/" {
		return strings.HasPrefix(p, "/")
	}
	baseRoot = strings.TrimRight(baseRoot, "/")
	if p == baseRoot {
		return true
	}
	return strings.HasPrefix(p, baseRoot+"/")
}

func validateFSLayerID(layerID string) error {
	if layerID == "" {
		return fmt.Errorf("%w: layer_id is required", ErrFSLayerInvalidArgument)
	}
	if len(layerID) > FSLayerIDMaxBytes {
		return fmt.Errorf("%w: layer_id exceeds %d bytes", ErrFSLayerInvalidArgument, FSLayerIDMaxBytes)
	}
	if strings.Contains(layerID, ":") {
		return fmt.Errorf("%w: layer_id must not contain \":\"", ErrFSLayerInvalidArgument)
	}
	return nil
}

func validateFSLayerState(state FSLayerState) error {
	switch state {
	case FSLayerStateActive, FSLayerStateSealed, FSLayerStateCommitting, FSLayerStateCommitted, FSLayerStateAbandoned, FSLayerStateConflicted:
		return nil
	default:
		return fmt.Errorf("%w: invalid fs layer state %q", ErrFSLayerInvalidArgument, state)
	}
}

func compactFSLayerEventID(eventID string) string {
	if len(eventID) <= fsLayerEventIDMaxBytes {
		return eventID
	}
	// Keep readable derived IDs when possible and hash only oversized IDs to fit the bounded primary key.
	return fmt.Sprintf("%x", sha256.Sum256([]byte(eventID)))
}

func validateFSLayerDurability(mode FSLayerDurabilityMode) error {
	switch mode {
	case FSLayerDurabilityRestoreSafe, FSLayerDurabilityWriteThrough, FSLayerDurabilityLocalFast:
		return nil
	default:
		return fmt.Errorf("%w: invalid fs layer durability mode %q", ErrFSLayerInvalidArgument, mode)
	}
}

func validateFSLayerEntryOp(op FSLayerEntryOp) error {
	switch op {
	case FSLayerEntryOpUpsert, FSLayerEntryOpWhiteout, FSLayerEntryOpMkdir, FSLayerEntryOpSymlink, FSLayerEntryOpChmod, FSLayerEntryOpRename:
		return nil
	default:
		return fmt.Errorf("invalid fs layer entry op %q", op)
	}
}

func validateFSLayerEntryKind(kind FSLayerEntryKind) error {
	switch kind {
	case FSLayerEntryKindFile, FSLayerEntryKindDir, FSLayerEntryKindSymlink:
		return nil
	default:
		return fmt.Errorf("invalid fs layer entry kind %q", kind)
	}
}

func normalizeFSLayerTags(tags map[string]string) (map[string]string, error) {
	if len(tags) == 0 {
		return nil, nil
	}
	out := make(map[string]string, len(tags))
	for rawKey, rawValue := range tags {
		key := strings.TrimSpace(rawKey)
		value := strings.TrimSpace(rawValue)
		if key == "" {
			return nil, fmt.Errorf("%w: fs layer tag key is required", ErrFSLayerInvalidArgument)
		}
		if len(key) > 255 {
			return nil, fmt.Errorf("%w: fs layer tag key %q exceeds 255 bytes", ErrFSLayerInvalidArgument, key)
		}
		if len(value) > 255 {
			return nil, fmt.Errorf("%w: fs layer tag %q value exceeds 255 bytes", ErrFSLayerInvalidArgument, key)
		}
		if strings.ContainsRune(key, '\x00') || strings.ContainsRune(value, '\x00') {
			return nil, fmt.Errorf("%w: fs layer tag contains NUL", ErrFSLayerInvalidArgument)
		}
		if strings.Contains(key, "=") {
			return nil, fmt.Errorf("%w: fs layer tag key %q must not contain '='", ErrFSLayerInvalidArgument, key)
		}
		out[key] = value
	}
	return out, nil
}

func parseFSLayerTagRef(ref string) (key string, value string, hasValue bool, ok bool) {
	ref = strings.TrimSpace(ref)
	hasTagPrefix := strings.HasPrefix(ref, "tag:")
	if strings.HasPrefix(ref, "tag:") {
		ref = strings.TrimSpace(strings.TrimPrefix(ref, "tag:"))
	}
	if ref == "" {
		return "", "", false, false
	}
	if k, v, found := strings.Cut(ref, "="); found {
		k = strings.TrimSpace(k)
		v = strings.TrimSpace(v)
		if k == "" {
			return "", "", false, false
		}
		return k, v, true, true
	}
	if strings.Contains(ref, " ") {
		return "", "", false, false
	}
	return ref, "", false, hasTagPrefix
}

func canonicalFSLayerLookupPath(p string) (string, error) {
	if strings.HasSuffix(p, "/") {
		canonical, err := pathutil.CanonicalizeDir(p)
		if err != nil {
			return "", fmt.Errorf("invalid fs layer entry path: %w", err)
		}
		return canonical, nil
	}
	canonical, err := pathutil.Canonicalize(p)
	if err != nil {
		return "", fmt.Errorf("invalid fs layer entry path: %w", err)
	}
	return canonical, nil
}

func fsLayerPathHash(path string) string {
	return StorageRefHash(path)
}