package datastore

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

	"github.com/pingcap/failpoint"
)

// FSLayerForkOptions controls ForkFSLayer.
type FSLayerForkOptions struct {
	// ChildLayerID is required (caller allocates id).
	ChildLayerID string
	// ParentLayerID is the layer to fork from (required).
	ParentLayerID string
	// Name optional display name for the child.
	Name string
	// ActorID optional.
	ActorID string
	// CheckpointID optional; when set, pin to that checkpoint's durable_seq (D2).
	CheckpointID string
	// MaxDepth soft cap; 0 means FSLayerDefaultMaxDepth.
	MaxDepth int
}

// FSLayerChainFrame is one hop on the parent chain from root toward tip.
type FSLayerChainFrame struct {
	Layer FSLayer
	// LimitSeq is the max entry_seq visible on this layer for the requesting child.
	// For the tip layer itself, LimitSeq is -1 meaning tip (no max).
	LimitSeq int64
}

// ForkFSLayer creates a child layer pinned to the parent's tip (or checkpoint).
// D1: tip pin = MAX(entry_seq) under parent FOR UPDATE.
// D2: checkpoint pin when CheckpointID set.
// D13: parent must be active or sealed.
// D14/D15: inherit base_root/durability; depth check.
func (s *Store) ForkFSLayer(ctx context.Context, opts FSLayerForkOptions) (*FSLayer, error) {
	childID := strings.TrimSpace(opts.ChildLayerID)
	parentID := strings.TrimSpace(opts.ParentLayerID)
	if err := validateFSLayerID(childID); err != nil {
		return nil, err
	}
	if parentID == "" {
		return nil, fmt.Errorf("%w: parent layer_id is required", ErrFSLayerInvalidArgument)
	}
	maxDepth := opts.MaxDepth
	if maxDepth <= 0 {
		maxDepth = FSLayerDefaultMaxDepth
	}
	if maxDepth > FSLayerHardMaxDepth {
		maxDepth = FSLayerHardMaxDepth
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("begin fork fs layer: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	var (
		baseRoot, state, durability, parentRoot string
		parentDepth                             int
	)
	failpoint.InjectCall("fsLayerForkAfterBegin", parentID)
	err = tx.QueryRowContext(ctx, `
SELECT base_root_path, state, durability_mode, root_layer_id, depth
FROM fs_layers WHERE `+s.scope.And(`layer_id = ?`)+` FOR UPDATE`,
		s.scope.Args(parentID)...).Scan(
		&baseRoot, &state, &durability, &parentRoot, &parentDepth,
	)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("read parent fs layer %s: %w", parentID, err)
	}
	st := FSLayerState(state)
	if st != FSLayerStateActive && st != FSLayerStateSealed {
		return nil, fmt.Errorf("%w: parent layer %s state %s cannot be forked", ErrFSLayerStateConflict, parentID, state)
	}
	childDepth := parentDepth + 1
	if childDepth > maxDepth {
		return nil, fmt.Errorf("%w: depth %d exceeds max %d", ErrFSLayerDepthExceeded, childDepth, maxDepth)
	}
	if parentRoot == "" {
		parentRoot = parentID
	}

	var originSeq int64
	var originCheckpointID string
	checkpointID := strings.TrimSpace(opts.CheckpointID)
	if checkpointID != "" {
		normalized, nerr := normalizeFSLayerCheckpointID(checkpointID)
		if nerr != nil {
			return nil, nerr
		}
		checkpointID = normalized
		var cpLayer string
		var cpSeq int64
		if err := tx.QueryRowContext(ctx, `
SELECT layer_id, durable_seq FROM fs_layer_checkpoints
WHERE `+s.scope.And(`checkpoint_id = ?`), s.scope.Args(checkpointID)...).Scan(&cpLayer, &cpSeq); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return nil, fmt.Errorf("%w: checkpoint %s", ErrNotFound, checkpointID)
			}
			return nil, fmt.Errorf("read checkpoint %s: %w", checkpointID, err)
		}
		if cpLayer != parentID {
			return nil, fmt.Errorf("%w: checkpoint %s does not belong to parent layer %s", ErrFSLayerStateConflict, checkpointID, parentID)
		}
		originSeq = cpSeq
		originCheckpointID = checkpointID
	} else {
		// D1 tip pin: latest committed entry_seq, not durable_seq.
		// Plain MAX() after the parent FOR UPDATE is still a snapshot
		// read in TiDB pessimistic txns (#72): a parent write that
		// committed while this fork waited on the layer lock would be
		// invisible. ORDER BY … LIMIT 1 FOR UPDATE is a current read.
		seq, serr := 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(parentID)...))
		if serr != nil {
			return nil, fmt.Errorf("read parent max entry_seq: %w", serr)
		}
		originSeq = seq
	}

	name := strings.TrimSpace(opts.Name)
	actor := strings.TrimSpace(opts.ActorID)
	child := FSLayer{
		LayerID:            childID,
		BaseRootPath:       baseRoot,
		Name:               name,
		State:              FSLayerStateActive,
		DurabilityMode:     FSLayerDurabilityMode(durability),
		ActorID:            actor,
		ParentLayerID:      parentID,
		OriginSeq:          originSeq,
		OriginCheckpointID: originCheckpointID,
		RootLayerID:        parentRoot,
		Depth:              childDepth,
		Origin:             FSLayerOriginFork,
	}
	_, 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(`?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?, UTC_TIMESTAMP(3), UTC_TIMESTAMP(3), NULL`)+`)`,
		s.scope.Args(child.LayerID, child.BaseRootPath, child.Name, string(child.State), string(child.DurabilityMode), child.ActorID,
			child.ParentLayerID, child.OriginSeq, child.OriginCheckpointID, child.RootLayerID, child.Depth, child.Origin)...)
	if err != nil {
		if isUniqueViolation(err) {
			return nil, fmt.Errorf("%w: layer %s already exists", ErrFSLayerStateConflict, childID)
		}
		return nil, fmt.Errorf("insert forked fs layer %s: %w", childID, err)
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("commit fork fs layer %s: %w", childID, err)
	}
	return s.GetFSLayer(ctx, childID)
}

// ListFSLayerChain returns frames from root to tip (inclusive).
// Ancestor LimitSeq is the origin_seq of the next hop toward tip.
// Tip LimitSeq is -1 (unbounded).
func (s *Store) ListFSLayerChain(ctx context.Context, layerID string) ([]FSLayerChainFrame, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return nil, fmt.Errorf("fs layer id is required")
	}
	var rev []FSLayer
	seen := map[string]struct{}{}
	for curID := layerID; curID != ""; {
		if _, ok := seen[curID]; ok {
			return nil, fmt.Errorf("fs layer chain cycle at %s", curID)
		}
		seen[curID] = struct{}{}
		layer, err := s.GetFSLayer(ctx, curID)
		if err != nil {
			return nil, err
		}
		rev = append(rev, *layer)
		if layer.ParentLayerID == "" {
			break
		}
		curID = layer.ParentLayerID
		if len(rev) > FSLayerHardMaxDepth+2 {
			return nil, fmt.Errorf("fs layer chain too deep")
		}
	}
	frames := make([]FSLayerChainFrame, 0, len(rev))
	for i := len(rev) - 1; i >= 0; i-- {
		frames = append(frames, FSLayerChainFrame{Layer: rev[i], LimitSeq: -1})
	}
	for i := 0; i < len(frames)-1; i++ {
		frames[i].LimitSeq = frames[i+1].Layer.OriginSeq
	}
	return frames, nil
}

func applyTipLimit(chain []FSLayerChainFrame, tipMaxSeq *int64) []FSLayerChainFrame {
	if tipMaxSeq == nil || len(chain) == 0 {
		return chain
	}
	chain[len(chain)-1].LimitSeq = *tipMaxSeq
	return chain
}

// ResolveFSLayerPath walks the chain for a path (D8).
// Returns (entry, true, nil) on hit (including whiteout), (nil, false, nil) on miss → main.
func (s *Store) ResolveFSLayerPath(ctx context.Context, layerID, path string) (*FSLayerEntry, bool, error) {
	return s.ResolveFSLayerPathAtSeq(ctx, layerID, path, nil)
}

// ResolveFSLayerPathAtSeq is ResolveFSLayerPath with an optional tip max_seq.
func (s *Store) ResolveFSLayerPathAtSeq(ctx context.Context, layerID, path string, tipMaxSeq *int64) (*FSLayerEntry, bool, error) {
	canonical, err := canonicalFSLayerLookupPath(path)
	if err != nil {
		return nil, false, err
	}
	tree, err := s.MaterializeFSLayerOverlay(ctx, layerID, tipMaxSeq)
	if err != nil {
		return nil, false, err
	}
	n, ok := OverlayLookup(tree, canonical)
	if !ok {
		hidden, _ := OverlayHideOrRemap(tree, canonical)
		if hidden {
			e := FSLayerEntry{Path: canonical, Op: FSLayerEntryOpWhiteout}
			return &e, true, nil
		}
		return nil, false, nil
	}
	e := OverlayNodeEntry(n)
	return &e, true, nil
}

// ListFSLayerChainMergedEntries returns latest-per-path effective overlay entries
// across the chain (tip wins). Used for diff/status listing.
// Whiteouts are included so callers can hide base paths.
func (s *Store) ListFSLayerChainMergedEntries(ctx context.Context, layerID string) ([]FSLayerEntry, error) {
	return s.listFSLayerChainMergedEntries(ctx, layerID, nil)
}

func (s *Store) ListFSLayerChainMergedEntriesAtSeq(ctx context.Context, layerID string, tipMaxSeq int64) ([]FSLayerEntry, error) {
	return s.listFSLayerChainMergedEntries(ctx, layerID, &tipMaxSeq)
}

func (s *Store) listFSLayerChainMergedEntries(ctx context.Context, layerID string, tipMaxSeq *int64) ([]FSLayerEntry, error) {
	tree, err := s.MaterializeFSLayerOverlay(ctx, layerID, tipMaxSeq)
	if err != nil {
		return nil, err
	}
	out := make([]FSLayerEntry, 0, len(tree))
	for _, n := range tree {
		if n.ConsumedByRename {
			continue
		}
		out = append(out, OverlayNodeEntry(n))
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Path == out[j].Path {
			return out[i].EntrySeq < out[j].EntrySeq
		}
		return out[i].Path < out[j].Path
	})
	return out, nil
}

// ListFSLayerChainEffectiveLog returns ordered ops (root→tip, origin limits). D19.
func (s *Store) ListFSLayerChainEffectiveLog(ctx context.Context, layerID string) ([]FSLayerEntry, error) {
	return s.listFSLayerChainEffectiveLog(ctx, layerID, nil)
}

func (s *Store) ListFSLayerChainEffectiveLogAtSeq(ctx context.Context, layerID string, tipMaxSeq int64) ([]FSLayerEntry, error) {
	return s.listFSLayerChainEffectiveLog(ctx, layerID, &tipMaxSeq)
}

// FSLayerStillPins reports whether any descendant still needs this layer (D17).
// True if any descendant is active/sealed/committing/conflicted, or is
// abandoned/committed but itself still_pins (e.g. abandoned middle + active grandchild).
func (s *Store) FSLayerStillPins(ctx context.Context, layerID string) (bool, error) {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return false, fmt.Errorf("fs layer id is required")
	}
	return s.fsLayerStillPinsRec(ctx, s.db, layerID, map[string]struct{}{})
}

type fsLayerQueryer interface {
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}

func (s *Store) fsLayerStillPinsRec(ctx context.Context, q fsLayerQueryer, layerID string, seen map[string]struct{}) (bool, error) {
	if _, ok := seen[layerID]; ok {
		return false, nil
	}
	seen[layerID] = struct{}{}
	rows, err := q.QueryContext(ctx, `
SELECT layer_id, state FROM fs_layers WHERE `+s.scope.And(`parent_layer_id = ?`),
		s.scope.Args(layerID)...)
	if err != nil {
		return false, fmt.Errorf("list child layers of %s: %w", layerID, err)
	}
	type child struct {
		id    string
		state FSLayerState
	}
	var children []child
	for rows.Next() {
		var id, state string
		if err := rows.Scan(&id, &state); err != nil {
			_ = rows.Close()
			return false, err
		}
		children = append(children, child{id: id, state: FSLayerState(state)})
	}
	err = rows.Err()
	_ = rows.Close()
	if err != nil {
		return false, err
	}
	for _, c := range children {
		switch c.state {
		case FSLayerStateActive, FSLayerStateSealed, FSLayerStateCommitting, FSLayerStateConflicted:
			return true, nil
		}
		// abandoned/committed: pin if their subtree still needs them
		sub, err := s.fsLayerStillPinsRec(ctx, q, c.id, seen)
		if err != nil {
			return false, err
		}
		if sub {
			return true, nil
		}
	}
	return false, nil
}

func (s *Store) listFSLayerDirectChildIDs(ctx context.Context, q fsLayerQueryer, layerID string) ([]string, error) {
	rows, err := q.QueryContext(ctx, `
SELECT layer_id FROM fs_layers WHERE `+s.scope.And(`parent_layer_id = ?`)+`
ORDER BY layer_id`, s.scope.Args(layerID)...)
	if err != nil {
		return nil, fmt.Errorf("list child layer ids of %s: %w", layerID, err)
	}
	defer func() { _ = rows.Close() }()
	var ids []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return ids, nil
}

// ListFSLayerDirectChildren returns layers whose parent_layer_id = layerID.
func (s *Store) ListFSLayerDirectChildren(ctx context.Context, layerID string) ([]FSLayer, error) {
	layerID = strings.TrimSpace(layerID)
	rows, err := s.db.QueryContext(ctx, `
SELECT `+fsLayerSelectCols+`
FROM fs_layers WHERE `+s.scope.And(`parent_layer_id = ?`)+`
ORDER BY layer_id`, s.scope.Args(layerID)...)
	if err != nil {
		return nil, fmt.Errorf("list child layers: %w", err)
	}
	defer func() { _ = rows.Close() }()
	return s.scanFSLayersWithTags(ctx, rows, "list child layers")
}

// DeleteFSLayerOptions controls DeleteFSLayer (logical abandon).
type DeleteFSLayerOptions struct {
	// Cascade abandons the whole descendant tree first (depth-first).
	Cascade bool
}

// DeleteFSLayer logically abandons a layer (D17). Physical row/entry delete is
// not performed while still_pins. Without cascade, still_pins → ErrFSLayerHasChildren.
//
// Non-cascade: the pins check and state flip share one row lock. A fork
// that committed while DELETE waited on that lock can still be invisible
// to the snapshot pins SELECT (same TiDB start_ts gap as D1). That is
// only a 409-vs-200 window: abandon stays logical and later still_pins
// walks still see the child. Cascade is one short transaction per layer,
// not a tree-wide lock: between iterations the parent stays active and
// unlocked, so a racing fork may 201 and then be abandoned by the next
// cascade pass. The loop re-checks pins under the lock each time; after
// the final flip, further forks fail with ErrFSLayerStateConflict.
func (s *Store) DeleteFSLayer(ctx context.Context, layerID string, opts DeleteFSLayerOptions) error {
	layerID = strings.TrimSpace(layerID)
	if layerID == "" {
		return fmt.Errorf("%w: fs layer id is required", ErrFSLayerInvalidArgument)
	}
	for {
		children, err := s.abandonFSLayerLocked(ctx, layerID, opts.Cascade)
		if err != nil {
			return err
		}
		if len(children) == 0 {
			return nil
		}
		for i := range children {
			if err := s.DeleteFSLayer(ctx, children[i], DeleteFSLayerOptions{Cascade: true}); err != nil {
				return err
			}
		}
	}
}

func (s *Store) abandonFSLayerLocked(ctx context.Context, layerID string, cascade bool) ([]string, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("begin delete fs layer %s: %w", layerID, err)
	}
	defer func() { _ = tx.Rollback() }()

	var state string
	err = tx.QueryRowContext(ctx, `
SELECT state FROM fs_layers WHERE `+s.scope.And(`layer_id = ?`)+` FOR UPDATE`,
		s.scope.Args(layerID)...).Scan(&state)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("lock fs layer %s: %w", layerID, err)
	}
	st := FSLayerState(state)
	if st == FSLayerStateCommitting {
		return nil, fmt.Errorf("%w: layer %s is committing", ErrFSLayerStateConflict, layerID)
	}

	pins, err := s.fsLayerStillPinsRec(ctx, tx, layerID, map[string]struct{}{})
	if err != nil {
		return nil, err
	}
	if pins {
		if !cascade {
			ids, listErr := s.listFSLayerDirectChildIDs(ctx, tx, layerID)
			if listErr != nil {
				return nil, listErr
			}
			return nil, fmt.Errorf("%w: layer %s has descendants %v; use cascade", ErrFSLayerHasChildren, layerID, ids)
		}
		ids, err := s.listFSLayerDirectChildIDs(ctx, tx, layerID)
		if err != nil {
			return nil, err
		}
		if err := tx.Commit(); err != nil {
			return nil, fmt.Errorf("commit delete fs layer %s: %w", layerID, err)
		}
		return ids, nil
	}
	if st == FSLayerStateAbandoned {
		if err := tx.Commit(); err != nil {
			return nil, fmt.Errorf("commit delete fs layer %s: %w", layerID, err)
		}
		return nil, nil
	}

	// Pins are false while this row is locked, so fork cannot insert a child.
	failpoint.InjectCall("fsLayerDeleteAfterPinsCheck", layerID)

	setSQL := `state = ?, updated_at = UTC_TIMESTAMP(3), sealed_at = COALESCE(sealed_at, UTC_TIMESTAMP(3))`
	setArgs := []any{string(FSLayerStateAbandoned)}
	if st == FSLayerStateCommitted {
		// Keep the apply mark when abandoning a committed row so hop
		// landing survives the state flip.
		setSQL = `state = ?, commit_applied = 1, updated_at = UTC_TIMESTAMP(3), sealed_at = COALESCE(sealed_at, UTC_TIMESTAMP(3))`
	}
	res, err := tx.ExecContext(ctx, `
UPDATE fs_layers
SET `+setSQL+`
WHERE `+s.scope.And(`layer_id = ? AND state IN (?, ?, ?, ?)`),
		append(setArgs,
			s.scope.Args(layerID,
				string(FSLayerStateActive), string(FSLayerStateSealed),
				string(FSLayerStateConflicted), string(FSLayerStateCommitted))...)...)
	if err != nil {
		return nil, fmt.Errorf("abandon fs layer %s: %w", layerID, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return nil, err
	}
	if n == 0 {
		return nil, fmt.Errorf("%w: layer %s", ErrFSLayerStateConflict, layerID)
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("commit delete fs layer %s: %w", layerID, err)
	}
	return nil, nil
}

// HasParent returns true if the layer is a non-root (forked) layer.
func (l *FSLayer) HasParent() bool {
	return l != nil && strings.TrimSpace(l.ParentLayerID) != ""
}