package controlplane

import (
	"database/sql"
	"fmt"
	"log"
	"strings"

	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// applyAllSourceRevisionAdvanceExecutionProof is the only value that binds one
// frozen active-runtime lineage and every frozen installed row to its exact
// current-runtime target, installed-compatibility target, and packs-minted
// execution atom. It is inert: it performs no reads or writes and exposes no
// execution method.
type applyAllSourceRevisionAdvanceExecutionProof struct {
	activeRuntimeIdentity *applyAllSourceRevisionAdvanceActiveRuntimeIdentity
	executionRuntime      *applyAllSourceRevisionAdvanceExecutionRuntime
	entries               []applyAllSourceRevisionAdvanceExecutionProofEntry
	provenance            applyAllSourceRevisionAdvanceExecutionProofProvenance
}

type applyAllSourceRevisionAdvanceExecutionProofEntry struct {
	name                   string
	current                applyAllSourceRevisionAdvanceCurrentSnapshotEntry
	currentTarget          *packs.SourceBackedPack
	replayBase             *packs.SourceBackedPack
	installedCompatibility *packs.SourceBackedPack
	atom                   *packs.SourceBackedExecutionAtom
}

// applyAllSourceRevisionAdvanceExecutionProofProvenance is an inert,
// independently detached witness for the complete proof assembled by the
// binder. A later validator compares the operational proof back to this value.
type applyAllSourceRevisionAdvanceExecutionProofProvenance struct {
	activeRuntimeIdentity *applyAllSourceRevisionAdvanceActiveRuntimeIdentity
	executionRuntime      *applyAllSourceRevisionAdvanceExecutionRuntime
	activeRuntimeDatabase *sql.DB
	activeRuntimeSnapshot *packs.ActiveInstalledRuntimeSnapshot
	entries               []applyAllSourceRevisionAdvanceExecutionProofProvenanceEntry
	entryCount            int
	minted                bool
}

type applyAllSourceRevisionAdvanceExecutionProofProvenanceEntry struct {
	expected              applyAllSourceRevisionAdvanceExecutionProofEntry
	currentTargetIdentity *packs.SourceBackedPackIdentity
	replayBaseIdentity    *packs.SourceBackedPackIdentity
	atomIdentity          *packs.SourceBackedExecutionAtomIdentity
}

func bindApplyAllSourceRevisionAdvanceExecutionProof(
	r *Runner,
	precondition applyAllSourceRevisionAdvancePreconditionProof,
	source applyAllSourceRevisionAdvanceSourceTuple,
	compatibility applyAllSourceRevisionAdvanceInstalledCompatibilityTuple,
) (applyAllSourceRevisionAdvanceExecutionProof, error) {
	if err := validateApplyAllSourceRevisionAdvanceTargetEquivalence(source, compatibility); err != nil {
		return applyAllSourceRevisionAdvanceExecutionProof{}, err
	}
	if err := validateApplyAllSourceRevisionAdvanceCurrentAtTargetCompatibility(precondition, compatibility); err != nil {
		return applyAllSourceRevisionAdvanceExecutionProof{}, err
	}
	if err := validateApplyAllSourceRevisionAdvanceReplayTuple(precondition); err != nil {
		return applyAllSourceRevisionAdvanceExecutionProof{}, err
	}
	var activeRuntimeIdentity *applyAllSourceRevisionAdvanceActiveRuntimeIdentity
	var executionRuntime *applyAllSourceRevisionAdvanceExecutionRuntime
	if len(compatibility.plan) == 0 {
		if precondition.activeRuntime.minted || precondition.activeRuntime.identity != nil ||
			compatibility.activeRuntimeIdentity != nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				-1, "", fmt.Errorf("empty roles retain an active-runtime identity"),
			)
		}
	} else {
		var err error
		activeRuntimeIdentity, err = precondition.activeRuntime.identityToken()
		if err != nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				-1, "", err,
			)
		}
		if compatibility.activeRuntimeIdentity != activeRuntimeIdentity {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				-1, "", fmt.Errorf("installed compatibility belongs to a different active-runtime observation"),
			)
		}
		executionRuntime, err = freezeApplyAllSourceRevisionAdvanceExecutionRuntime(r)
		if err != nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				-1, "", fmt.Errorf("freeze execution runtime: %w", err),
			)
		}
		if executionRuntime.database != activeRuntimeIdentity.db {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				-1, "", fmt.Errorf("execution runtime and active-runtime observation use different databases"),
			)
		}
	}
	// Durable equivalence has served its only purpose. Drop the fresh-source
	// authority before constructing the capability so the proof can retain only
	// the frozen current and installed-compatibility roles.
	source = applyAllSourceRevisionAdvanceSourceTuple{}
	proof := applyAllSourceRevisionAdvanceExecutionProof{
		activeRuntimeIdentity: activeRuntimeIdentity,
		executionRuntime:      executionRuntime,
		entries:               make([]applyAllSourceRevisionAdvanceExecutionProofEntry, 0, len(compatibility.plan)),
		provenance: applyAllSourceRevisionAdvanceExecutionProofProvenance{
			activeRuntimeIdentity: activeRuntimeIdentity,
			executionRuntime:      executionRuntime,
			entries:               make([]applyAllSourceRevisionAdvanceExecutionProofProvenanceEntry, 0, len(compatibility.plan)),
			entryCount:            len(compatibility.plan),
			minted:                true,
		},
	}
	if activeRuntimeIdentity != nil {
		proof.provenance.activeRuntimeDatabase = activeRuntimeIdentity.db
		proof.provenance.activeRuntimeSnapshot = activeRuntimeIdentity.snapshot
	}
	for _, advance := range compatibility.plan {
		name := strings.ToLower(strings.TrimSpace(advance.Name))
		current, ok := precondition.snapshot.entry(name)
		if !ok || current.advance != advance {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				advance.index, advance.Name,
				fmt.Errorf("frozen current entry does not match the installed compatibility plan"),
			)
		}
		currentTarget := precondition.current.targets[name]
		if currentTarget == nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				advance.index, advance.Name, fmt.Errorf("frozen current target is missing"),
			)
		}
		replayBase := precondition.replay.targets[name]
		if replayBase == nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				advance.index, advance.Name, fmt.Errorf("frozen replay base is missing"),
			)
		}
		target, ok := compatibility.targets[name]
		if !ok || target == nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				advance.index, advance.Name, fmt.Errorf("installed compatibility target is missing"),
			)
		}
		currentTargetIdentity, err := packs.FreezeSourceBackedPackIdentity(currentTarget)
		if err != nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				advance.index, advance.Name,
				fmt.Errorf("freeze frozen current target identity: %w", err),
			)
		}
		replayBaseIdentity, err := packs.FreezeSourceBackedPackIdentity(replayBase)
		if err != nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				advance.index, advance.Name,
				fmt.Errorf("freeze frozen replay base identity: %w", err),
			)
		}
		// Begin always mints a fresh PackInstallId for a source-revision
		// advance. Components whose live rows are owned by that incarnation
		// therefore require an idempotent replay even when their declaration
		// payload is unchanged. Component equality is not ownership equality.
		atom, _, err := packs.FreezeSourceBackedReplayExecutionAtom(replayBase, target)
		if err == nil {
			reason := "fresh-pack-install-incarnation"
			if current.pending != nil && current.pending.Kind == ApplyAllSourceRevisionAdvancePendingRetry {
				reason = "pending-retry"
			} else if advance.completeReplay {
				reason = "offline-candidate-prior-attempt-may-have-partial-writes"
			}
			log.Printf(
				"controlplane: source revision advance pack=%s plan=complete reason=%s",
				target.Name(), reason,
			)
		}
		if err != nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				advance.index, advance.Name,
				fmt.Errorf("freeze installed compatibility execution atom: %w", err),
			)
		}
		atomIdentity, err := packs.FreezeSourceBackedExecutionAtomIdentity(atom)
		if err != nil {
			return applyAllSourceRevisionAdvanceExecutionProof{}, applyAllSourceRevisionAdvanceTargetEquivalenceError(
				advance.index, advance.Name,
				fmt.Errorf("freeze installed compatibility execution atom identity: %w", err),
			)
		}
		entry := applyAllSourceRevisionAdvanceExecutionProofEntry{
			name:                   name,
			current:                current,
			currentTarget:          currentTarget,
			replayBase:             replayBase,
			installedCompatibility: target,
			atom:                   atom,
		}
		proof.entries = append(proof.entries, entry)
		proof.provenance.entries = append(
			proof.provenance.entries,
			applyAllSourceRevisionAdvanceExecutionProofProvenanceEntry{
				expected:              cloneApplyAllSourceRevisionAdvanceExecutionProofEntry(entry),
				currentTargetIdentity: currentTargetIdentity,
				replayBaseIdentity:    replayBaseIdentity,
				atomIdentity:          atomIdentity,
			},
		)
	}
	if err := validateApplyAllSourceRevisionAdvanceExecutionProof(proof); err != nil {
		return applyAllSourceRevisionAdvanceExecutionProof{}, err
	}
	return proof, nil
}

func cloneApplyAllSourceRevisionAdvanceExecutionProofEntry(
	entry applyAllSourceRevisionAdvanceExecutionProofEntry,
) applyAllSourceRevisionAdvanceExecutionProofEntry {
	out := entry
	if entry.current.pending != nil {
		pending := *entry.current.pending
		out.current.pending = &pending
	}
	return out
}

func cloneApplyAllSourceRevisionAdvanceExecutionProof(
	proof applyAllSourceRevisionAdvanceExecutionProof,
) applyAllSourceRevisionAdvanceExecutionProof {
	out := proof
	out.entries = make([]applyAllSourceRevisionAdvanceExecutionProofEntry, len(proof.entries))
	for index, entry := range proof.entries {
		out.entries[index] = cloneApplyAllSourceRevisionAdvanceExecutionProofEntry(entry)
	}
	out.provenance = cloneApplyAllSourceRevisionAdvanceExecutionProofProvenance(proof.provenance)
	return out
}

func cloneApplyAllSourceRevisionAdvanceExecutionProofProvenance(
	provenance applyAllSourceRevisionAdvanceExecutionProofProvenance,
) applyAllSourceRevisionAdvanceExecutionProofProvenance {
	out := provenance
	out.entries = make(
		[]applyAllSourceRevisionAdvanceExecutionProofProvenanceEntry,
		len(provenance.entries),
	)
	for index, entry := range provenance.entries {
		out.entries[index] = entry
		out.entries[index].expected = cloneApplyAllSourceRevisionAdvanceExecutionProofEntry(entry.expected)
	}
	return out
}

// validateApplyAllSourceRevisionAdvanceExecutionProof checks the binder-minted
// envelope and its detached runtime, cardinality, entry, and target-construction
// provenance. It performs no database or source reads, resolution, execution,
// or writes.
func validateApplyAllSourceRevisionAdvanceExecutionProof(
	proof applyAllSourceRevisionAdvanceExecutionProof,
) error {
	provenance := proof.provenance
	if !provenance.minted || proof.entries == nil || provenance.entries == nil {
		return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
			-1, "", fmt.Errorf("proof was not minted by its binder"),
		)
	}
	if provenance.entryCount < 0 || len(proof.entries) != provenance.entryCount ||
		len(provenance.entries) != provenance.entryCount {
		return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
			-1, "", fmt.Errorf("proof entry cardinality does not match its provenance"),
		)
	}
	if provenance.entryCount == 0 {
		if proof.activeRuntimeIdentity != nil || provenance.activeRuntimeIdentity != nil ||
			proof.executionRuntime != nil || provenance.executionRuntime != nil ||
			provenance.activeRuntimeDatabase != nil || provenance.activeRuntimeSnapshot != nil {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				-1, "", fmt.Errorf("empty proof retains active-runtime provenance"),
			)
		}
		return nil
	}
	if proof.activeRuntimeIdentity == nil ||
		proof.activeRuntimeIdentity != provenance.activeRuntimeIdentity ||
		proof.executionRuntime == nil || proof.executionRuntime != provenance.executionRuntime ||
		provenance.activeRuntimeDatabase == nil || provenance.activeRuntimeSnapshot == nil ||
		proof.activeRuntimeIdentity.db != provenance.activeRuntimeDatabase ||
		proof.activeRuntimeIdentity.snapshot != provenance.activeRuntimeSnapshot {
		return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
			-1, "", fmt.Errorf("active-runtime identity does not match its provenance"),
		)
	}
	if err := proof.executionRuntime.validate(); err != nil ||
		proof.executionRuntime.database != provenance.activeRuntimeDatabase {
		if err == nil {
			err = fmt.Errorf("execution runtime database does not match active-runtime provenance")
		}
		return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
			-1, "", fmt.Errorf("validate execution runtime: %w", err),
		)
	}
	installedPlan := make([]normalizedApplyAllSourceRevisionAdvance, 0, provenance.entryCount)
	installedTargets := make(map[string]*packs.SourceBackedPack, provenance.entryCount)
	for index, entry := range proof.entries {
		witness := provenance.entries[index]
		advance := witness.expected.current.advance
		if !equalApplyAllSourceRevisionAdvanceExecutionProofEntry(entry, witness.expected) {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				advance.index, advance.Name, fmt.Errorf("entry does not match its detached provenance"),
			)
		}
		if entry.current.pending != nil && entry.current.pending == witness.expected.current.pending {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				advance.index, advance.Name, fmt.Errorf("entry pending classification aliases its provenance"),
			)
		}
		name := strings.ToLower(strings.TrimSpace(advance.Name))
		if entry.name != name {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				advance.index, advance.Name, fmt.Errorf("entry name is not the canonical advance name"),
			)
		}
		if strings.TrimSpace(entry.current.row.ID) == "" || strings.TrimSpace(entry.current.row.PackInstallID) == "" {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				advance.index, advance.Name, fmt.Errorf("installed row and pack-install incarnation identity are required"),
			)
		}
		if err := packs.ValidateSourceBackedPackIdentityTarget(
			witness.currentTargetIdentity, entry.currentTarget,
		); err != nil {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				advance.index, advance.Name, fmt.Errorf("validate current target identity: %w", err),
			)
		}
		if err := packs.ValidateSourceBackedPackIdentityTarget(
			witness.replayBaseIdentity, entry.replayBase,
		); err != nil {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				advance.index, advance.Name, fmt.Errorf("validate replay base identity: %w", err),
			)
		}
		if err := packs.ValidateSourceBackedExecutionAtomIdentityTarget(
			witness.atomIdentity, entry.atom, entry.installedCompatibility,
		); err != nil {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				advance.index, advance.Name, fmt.Errorf("validate installed execution atom identity: %w", err),
			)
		}
		if err := validateApplyAllSourceRevisionAdvanceFrozenCurrentTargetEntry(
			entry.current, entry.currentTarget,
		); err != nil {
			return applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
				advance.index, advance.Name, fmt.Errorf("validate frozen current target: %w", err),
			)
		}
		installedPlan = append(installedPlan, advance)
		installedTargets[name] = entry.installedCompatibility
	}
	if err := validateApplyAllSourceRevisionAdvanceTargetTuple(
		"execution proof installed compatibility",
		applyAllSourceRevisionAdvanceTuple{plan: installedPlan, targets: installedTargets},
	); err != nil {
		return err
	}
	return nil
}

func equalApplyAllSourceRevisionAdvanceExecutionProofEntry(
	left applyAllSourceRevisionAdvanceExecutionProofEntry,
	right applyAllSourceRevisionAdvanceExecutionProofEntry,
) bool {
	return left.name == right.name &&
		left.current.advance == right.current.advance &&
		left.current.row == right.current.row &&
		equalApplyAllSourceRevisionAdvancePending(left.current.pending, right.current.pending) &&
		left.currentTarget == right.currentTarget &&
		left.replayBase == right.replayBase &&
		left.installedCompatibility == right.installedCompatibility &&
		left.atom == right.atom
}

func applyAllSourceRevisionAdvanceExecutionProofIntegrityError(
	index int,
	name string,
	err error,
) error {
	return applyAllSourceRevisionAdvancePlanError(
		ApplyAllSourceRevisionAdvancePlanCompose, index, name,
		fmt.Errorf("validate execution proof integrity: %w", err),
	)
}