package controlplane

import (
	"context"
	"database/sql"
	"errors"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

func TestAdmitApplyAllSourceRevisionAdvanceExecutionRereadsWholeProofAndDetachesPending(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	proof := testApplyAllSourceRevisionAdvanceExecutionProofForAdmission(t, db, 2)
	for _, entry := range proof.entries {
		expectApplyAllSourceRevisionAdvancePreconditionRow(mock, entry.current.advance.Name, entry.current.row)
	}

	admission, err := admitApplyAllSourceRevisionAdvanceExecution(context.Background(), db, proof)
	if err != nil {
		t.Fatalf("admit execution proof: %v", err)
	}
	if admission.identity == nil || !admission.identity.minted {
		t.Fatalf("admission identity = %#v, want minted proof carrier", admission.identity)
	}
	retained := &admission.identity.proof
	if retained.activeRuntimeIdentity != proof.activeRuntimeIdentity ||
		retained.executionRuntime != proof.executionRuntime ||
		retained.provenance.activeRuntimeIdentity != proof.provenance.activeRuntimeIdentity ||
		retained.provenance.executionRuntime != proof.provenance.executionRuntime ||
		retained.provenance.activeRuntimeDatabase != db ||
		retained.provenance.activeRuntimeSnapshot != proof.provenance.activeRuntimeSnapshot {
		t.Fatalf("retained proof = %#v, want exact runtime lineage", retained)
	}
	if retained.entries == nil || len(retained.entries) != 2 ||
		retained.provenance.entries == nil || len(retained.provenance.entries) != 2 {
		t.Fatalf("retained entries/provenance = %#v/%#v, want complete detached tuple", retained.entries, retained.provenance.entries)
	}
	outerProvenance := &admission.identity.provenance
	if outerProvenance.executionRuntime != proof.executionRuntime ||
		outerProvenance.activeRuntimeIdentity != proof.activeRuntimeIdentity ||
		outerProvenance.activeRuntimeDatabase != db ||
		outerProvenance.activeRuntimeSnapshot != proof.provenance.activeRuntimeSnapshot ||
		outerProvenance.entries == nil || len(outerProvenance.entries) != 2 {
		t.Fatalf("outer admission provenance = %#v, want exact detached proof witness", outerProvenance)
	}
	for index, entry := range retained.entries {
		if !equalApplyAllSourceRevisionAdvanceExecutionProofEntry(entry, proof.entries[index]) {
			t.Fatalf("admitted entry %d = %#v, want complete proof entry %#v", index, entry, proof.entries[index])
		}
		if entry.currentTarget != proof.entries[index].currentTarget ||
			entry.installedCompatibility != proof.entries[index].installedCompatibility ||
			entry.atom != proof.entries[index].atom {
			t.Fatalf("admitted entry %d did not retain exact target and atom constructions", index)
		}
	}
	if retained.entries[1].current.pending == nil ||
		retained.entries[1].current.pending == proof.entries[1].current.pending ||
		retained.entries[1].current.pending == retained.provenance.entries[1].expected.current.pending ||
		retained.provenance.entries[1].expected.current.pending == proof.provenance.entries[1].expected.current.pending ||
		outerProvenance.entries[1].expected.current.pending == retained.entries[1].current.pending ||
		outerProvenance.entries[1].expected.current.pending == retained.provenance.entries[1].expected.current.pending ||
		outerProvenance.entries[1].expected.current.pending == proof.provenance.entries[1].expected.current.pending {
		t.Fatal("admitted pending classification is missing or aliases proof state")
	}
	if &retained.entries[0] == &proof.entries[0] ||
		&retained.provenance.entries[0] == &proof.provenance.entries[0] ||
		retained.provenance.entries[0].currentTargetIdentity != proof.provenance.entries[0].currentTargetIdentity ||
		retained.provenance.entries[0].atomIdentity != proof.provenance.entries[0].atomIdentity {
		t.Fatal("admission did not detach mutable slices while retaining proof-minted identity witnesses")
	}
	wantKind := retained.entries[1].current.pending.Kind
	proof.entries[1].current.pending.Kind = ApplyAllSourceRevisionAdvancePendingRetry
	proof.provenance.entries[1].expected.current.pending.Kind = ApplyAllSourceRevisionAdvancePendingCanonicalize
	proof.entries = proof.entries[:1]
	proof.provenance.entries = proof.provenance.entries[:1]
	proof.executionRuntime = nil
	proof.provenance.executionRuntime = nil
	if retained.entries[1].current.pending.Kind != wantKind || len(retained.entries) != 2 ||
		len(retained.provenance.entries) != 2 ||
		validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission) != nil {
		t.Fatal("admission changed after source proof slice or pending mutation")
	}
	retained.executionRuntime.applyDeps.Options.SkipPolymorphicDiscriminatorBackfill = true
	if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err == nil {
		t.Fatal("admission accepted mutation behind the retained source-proof runtime token")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestAdmitApplyAllSourceRevisionAdvanceExecutionAcceptsMintedEmptyWithoutDatabase(t *testing.T) {
	proof := testApplyAllSourceRevisionAdvanceExecutionProofForAdmission(t, nil, 0)
	admission, err := admitApplyAllSourceRevisionAdvanceExecution(context.Background(), nil, proof)
	if err != nil {
		t.Fatalf("admit empty proof: %v", err)
	}
	if admission.identity == nil || !admission.identity.minted ||
		admission.identity.proof.activeRuntimeIdentity != nil ||
		admission.identity.proof.executionRuntime != nil ||
		admission.identity.proof.entries == nil || len(admission.identity.proof.entries) != 0 ||
		!admission.identity.proof.provenance.minted ||
		admission.identity.proof.provenance.entries == nil ||
		len(admission.identity.proof.provenance.entries) != 0 ||
		!admission.identity.provenance.minted || admission.identity.provenance.entries == nil ||
		len(admission.identity.provenance.entries) != 0 {
		t.Fatalf("empty admission = %#v, want minted empty capability", admission)
	}
	if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err != nil {
		t.Fatalf("validate empty admission: %v", err)
	}

	admission, err = admitApplyAllSourceRevisionAdvanceExecution(
		context.Background(), nil, applyAllSourceRevisionAdvanceExecutionProof{},
	)
	assertApplyAllSourceRevisionComposeError(t, err, -1, "")
	assertZeroApplyAllSourceRevisionAdvanceExecutionAdmission(t, admission)
}

func TestAdmitApplyAllSourceRevisionAdvanceExecutionValidatesProofBeforeDatabaseRead(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	proof := testApplyAllSourceRevisionAdvanceExecutionProofForAdmission(t, db, 1)
	proof.entries[0].current.advance.ToRevision += "-mutated"
	proof.entries[0].current.pending.Authorization.ToRevision = proof.entries[0].current.advance.ToRevision
	proof.provenance.entries[0].expected = cloneApplyAllSourceRevisionAdvanceExecutionProofEntry(proof.entries[0])

	admission, err := admitApplyAllSourceRevisionAdvanceExecution(context.Background(), db, proof)
	assertApplyAllSourceRevisionComposeError(t, err, proof.entries[0].current.advance.index, proof.entries[0].current.advance.Name)
	assertZeroApplyAllSourceRevisionAdvanceExecutionAdmission(t, admission)
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL was attempted before proof validation: %v", err)
	}
}

func TestAdmitApplyAllSourceRevisionAdvanceExecutionRequiresExactDatabaseBeforeRead(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	otherDB, otherMock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("other sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = otherDB.Close() })
	proof := testApplyAllSourceRevisionAdvanceExecutionProofForAdmission(t, db, 1)

	for name, candidate := range map[string]*sql.DB{"nil": nil, "different": otherDB} {
		t.Run(name, func(t *testing.T) {
			admission, err := admitApplyAllSourceRevisionAdvanceExecution(context.Background(), candidate, proof)
			assertApplyAllSourceRevisionComposeError(t, err, -1, "")
			assertZeroApplyAllSourceRevisionAdvanceExecutionAdmission(t, admission)
		})
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("proof database was read before identity validation: %v", err)
	}
	if err := otherMock.ExpectationsWereMet(); err != nil {
		t.Fatalf("different database was read before identity validation: %v", err)
	}
}

func TestApplyAllSourceRevisionAdvanceExecutionAdmissionRejectsMutationBeforeConsumerEffects(t *testing.T) {
	tests := []struct {
		name   string
		mutate func(*testing.T, *applyAllSourceRevisionAdvanceExecutionAdmission, *sql.DB)
	}{
		{name: "coordinated runtime substitution", mutate: func(t *testing.T, admission *applyAllSourceRevisionAdvanceExecutionAdmission, db *sql.DB) {
			replacement, err := freezeApplyAllSourceRevisionAdvanceExecutionRuntime(
				testApplyAllSourceRevisionAdvanceExecutionProofRunner(db),
			)
			if err != nil {
				t.Fatalf("freeze replacement runtime: %v", err)
			}
			admission.identity.proof.executionRuntime = replacement
			admission.identity.proof.provenance.executionRuntime = replacement
		}},
		{name: "coordinated active runtime snapshot substitution", mutate: func(_ *testing.T, admission *applyAllSourceRevisionAdvanceExecutionAdmission, db *sql.DB) {
			replacement := &applyAllSourceRevisionAdvanceActiveRuntimeIdentity{
				db: db, snapshot: &packs.ActiveInstalledRuntimeSnapshot{},
			}
			admission.identity.proof.activeRuntimeIdentity = replacement
			admission.identity.proof.provenance.activeRuntimeIdentity = replacement
			admission.identity.proof.provenance.activeRuntimeDatabase = db
			admission.identity.proof.provenance.activeRuntimeSnapshot = replacement.snapshot
		}},
		{name: "coordinated tuple shrink", mutate: func(_ *testing.T, admission *applyAllSourceRevisionAdvanceExecutionAdmission, _ *sql.DB) {
			admission.identity.proof.entries = admission.identity.proof.entries[:1]
			admission.identity.proof.provenance.entries = admission.identity.proof.provenance.entries[:1]
			admission.identity.proof.provenance.entryCount = 1
		}},
		{name: "coordinated tuple reorder", mutate: func(_ *testing.T, admission *applyAllSourceRevisionAdvanceExecutionAdmission, _ *sql.DB) {
			entries := admission.identity.proof.entries
			entries[0], entries[1] = entries[1], entries[0]
			provenance := admission.identity.proof.provenance.entries
			provenance[0], provenance[1] = provenance[1], provenance[0]
		}},
		{name: "equivalent current target replacement", mutate: func(t *testing.T, admission *applyAllSourceRevisionAdvanceExecutionAdmission, _ *sql.DB) {
			proof := &admission.identity.proof
			entry := &proof.entries[0]
			advance := entry.current.advance
			replacement := testApplyAllSourceRevisionTarget(
				t, advance.Name, "1.0.0", advance.Kind, advance.URI, entry.current.row.SourceRevision.String,
			)
			entry.currentTarget = replacement
			proof.provenance.entries[0].expected.currentTarget = replacement
			identity, err := packs.FreezeSourceBackedPackIdentity(replacement)
			if err != nil {
				t.Fatalf("freeze replacement current target identity: %v", err)
			}
			proof.provenance.entries[0].currentTargetIdentity = identity
		}},
		{name: "equivalent installed target and atom replacement", mutate: func(t *testing.T, admission *applyAllSourceRevisionAdvanceExecutionAdmission, _ *sql.DB) {
			proof := &admission.identity.proof
			entry := &proof.entries[0]
			advance := entry.current.advance
			replacement := testApplyAllSourceRevisionTarget(
				t, advance.Name, "1.0.0", advance.Kind, advance.URI, advance.ToRevision,
			)
			atom, err := packs.FreezeSourceBackedExecutionAtom(replacement)
			if err != nil {
				t.Fatalf("freeze replacement atom: %v", err)
			}
			entry.installedCompatibility = replacement
			entry.atom = atom
			proof.provenance.entries[0].expected.installedCompatibility = replacement
			proof.provenance.entries[0].expected.atom = atom
			identity, err := packs.FreezeSourceBackedExecutionAtomIdentity(atom)
			if err != nil {
				t.Fatalf("freeze replacement atom identity: %v", err)
			}
			proof.provenance.entries[0].atomIdentity = identity
		}},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			proof := testApplyAllSourceRevisionAdvanceExecutionProofForAdmission(t, db, 2)
			for _, entry := range proof.entries {
				expectApplyAllSourceRevisionAdvancePreconditionRow(mock, entry.current.advance.Name, entry.current.row)
			}
			admission, err := admitApplyAllSourceRevisionAdvanceExecution(context.Background(), db, proof)
			if err != nil {
				t.Fatalf("admit execution proof: %v", err)
			}

			test.mutate(t, &admission, db)
			gotDB, gotEntry, err := applyAllSourceRevisionAdvanceAdmittedPendingEntry(admission, 0)
			var planErr *ApplyAllSourceRevisionAdvancePlanError
			if !errors.As(err, &planErr) || planErr.Stage != ApplyAllSourceRevisionAdvancePlanCompose {
				t.Fatalf("consume mutated admission error = %v, want compose-stage integrity error", err)
			}
			if gotDB != nil || gotEntry.name != "" || gotEntry.currentTarget != nil ||
				gotEntry.installedCompatibility != nil || gotEntry.atom != nil {
				t.Fatalf("consume mutated admission = %p/%#v, want zero result", gotDB, gotEntry)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("consumer attempted SQL after admission mutation: %v", err)
			}
		})
	}
}

func TestAdmitApplyAllSourceRevisionAdvanceExecutionRejectsEveryProjectedFieldChange(t *testing.T) {
	mutators := []struct {
		name string
		edit func(*installedPackRow)
	}{
		{name: "id", edit: func(row *installedPackRow) { row.ID = "changed" }},
		{name: "name", edit: func(row *installedPackRow) { row.Name = "changed" }},
		{name: "pack install ID", edit: func(row *installedPackRow) { row.PackInstallID = "changed" }},
		{name: "category", edit: func(row *installedPackRow) { row.Category = "changed" }},
		{name: "status", edit: func(row *installedPackRow) { row.Status = installedPackStatusFailed }},
		{name: "is deleted", edit: func(row *installedPackRow) { row.IsDeleted = 1 }},
		{name: "pack version", edit: func(row *installedPackRow) { row.PackVersion = sql.NullString{} }},
		{name: "manifest format", edit: func(row *installedPackRow) { row.ManifestFormatVersion = sql.NullString{} }},
		{name: "platform compatibility", edit: func(row *installedPackRow) { row.PlatformCompatibility = sql.NullString{} }},
		{name: "desired hash", edit: func(row *installedPackRow) {
			row.DesiredPayloadHash = sql.NullString{Valid: true, String: "changed"}
		}},
		{name: "applied hash", edit: func(row *installedPackRow) { row.AppliedPayloadHash = sql.NullString{} }},
		{name: "hash algorithm", edit: func(row *installedPackRow) { row.PayloadHashAlgorithm = sql.NullString{} }},
		{name: "source kind", edit: func(row *installedPackRow) { row.SourceKind = sql.NullString{} }},
		{name: "source URI", edit: func(row *installedPackRow) { row.SourceURI = sql.NullString{} }},
		{name: "source revision", edit: func(row *installedPackRow) { row.SourceRevision = sql.NullString{} }},
		{name: "host requirements", edit: func(row *installedPackRow) { row.HostRequirementsDigest = sql.NullString{} }},
	}
	for _, test := range mutators {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			proof := testApplyAllSourceRevisionAdvanceExecutionProofForAdmission(t, db, 1)
			entry := proof.entries[0]
			changed := entry.current.row
			test.edit(&changed)
			expectApplyAllSourceRevisionAdvancePreconditionRow(mock, entry.current.advance.Name, changed)

			admission, err := admitApplyAllSourceRevisionAdvanceExecution(context.Background(), db, proof)
			assertApplyAllSourceRevisionAdvancePreconditionError(
				t, err, ApplyAllSourceRevisionAdvancePlanState, entry.current.advance.index, entry.current.advance.Name,
			)
			assertZeroApplyAllSourceRevisionAdvanceExecutionAdmission(t, admission)
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestAdmitApplyAllSourceRevisionAdvanceExecutionReturnsZeroOnReadOrStateFailure(t *testing.T) {
	tests := []struct {
		name      string
		wantStage ApplyAllSourceRevisionAdvancePlanErrorStage
		setup     func(sqlmock.Sqlmock, applyAllSourceRevisionAdvanceExecutionProof)
		wantEntry int
	}{
		{name: "read", wantStage: ApplyAllSourceRevisionAdvancePlanRead, setup: func(mock sqlmock.Sqlmock, proof applyAllSourceRevisionAdvanceExecutionProof) {
			entry := proof.entries[0]
			mock.ExpectQuery(applyAllSourceRevisionAdvanceRowQueryPattern()).WithArgs(entry.current.advance.Name).WillReturnError(errors.New("read failed"))
		}},
		{name: "missing", wantStage: ApplyAllSourceRevisionAdvancePlanState, setup: func(mock sqlmock.Sqlmock, proof applyAllSourceRevisionAdvanceExecutionProof) {
			entry := proof.entries[0]
			mock.ExpectQuery(applyAllSourceRevisionAdvanceRowQueryPattern()).WithArgs(entry.current.advance.Name).WillReturnRows(testApplyAllSourceRevisionAdvancePreconditionRows())
		}},
		{name: "second entry changed", wantStage: ApplyAllSourceRevisionAdvancePlanState, wantEntry: 1, setup: func(mock sqlmock.Sqlmock, proof applyAllSourceRevisionAdvanceExecutionProof) {
			expectApplyAllSourceRevisionAdvancePreconditionRow(mock, proof.entries[0].current.advance.Name, proof.entries[0].current.row)
			changed := proof.entries[1].current.row
			changed.SourceRevision.String = "changed"
			expectApplyAllSourceRevisionAdvancePreconditionRow(mock, proof.entries[1].current.advance.Name, changed)
		}},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			proof := testApplyAllSourceRevisionAdvanceExecutionProofForAdmission(t, db, 2)
			test.setup(mock, proof)
			entry := proof.entries[test.wantEntry].current.advance

			admission, err := admitApplyAllSourceRevisionAdvanceExecution(context.Background(), db, proof)
			assertApplyAllSourceRevisionAdvancePreconditionError(t, err, test.wantStage, entry.index, entry.Name)
			assertZeroApplyAllSourceRevisionAdvanceExecutionAdmission(t, admission)
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func testApplyAllSourceRevisionAdvanceExecutionProofForAdmission(
	t *testing.T,
	db *sql.DB,
	count int,
) applyAllSourceRevisionAdvanceExecutionProof {
	t.Helper()
	return testApplyAllSourceRevisionAdvanceExecutionProofForDatabase(t, db, count)
}

func testApplyAllSourceRevisionAdvanceExecutionAdmissionFromEntries(
	t *testing.T,
	db *sql.DB,
	entries []applyAllSourceRevisionAdvanceExecutionProofEntry,
) applyAllSourceRevisionAdvanceExecutionAdmission {
	t.Helper()
	proof := testApplyAllSourceRevisionAdvanceExecutionProofForDatabase(t, db, len(entries))
	proof.entries = make([]applyAllSourceRevisionAdvanceExecutionProofEntry, len(entries))
	proof.provenance.entries = make(
		[]applyAllSourceRevisionAdvanceExecutionProofProvenanceEntry, len(entries),
	)
	proof.provenance.entryCount = len(entries)
	for index, entry := range entries {
		if entry.replayBase == nil {
			entry.replayBase = entry.currentTarget
		}
		proof.entries[index] = cloneApplyAllSourceRevisionAdvanceExecutionProofEntry(entry)
		currentTargetIdentity, err := packs.FreezeSourceBackedPackIdentity(entry.currentTarget)
		if err != nil {
			t.Fatalf("freeze admission fixture current target: %v", err)
		}
		replayBaseIdentity, err := packs.FreezeSourceBackedPackIdentity(entry.replayBase)
		if err != nil {
			t.Fatalf("freeze admission fixture replay base: %v", err)
		}
		atomIdentity, err := packs.FreezeSourceBackedExecutionAtomIdentity(entry.atom)
		if err != nil {
			t.Fatalf("freeze admission fixture atom: %v", err)
		}
		proof.provenance.entries[index] = applyAllSourceRevisionAdvanceExecutionProofProvenanceEntry{
			expected:              cloneApplyAllSourceRevisionAdvanceExecutionProofEntry(entry),
			currentTargetIdentity: currentTargetIdentity,
			replayBaseIdentity:    replayBaseIdentity,
			atomIdentity:          atomIdentity,
		}
	}
	if err := validateApplyAllSourceRevisionAdvanceExecutionProof(proof); err != nil {
		t.Fatalf("validate admission fixture proof: %v", err)
	}
	admission := applyAllSourceRevisionAdvanceExecutionAdmission{
		identity: &applyAllSourceRevisionAdvanceExecutionAdmissionIdentity{
			proof:      cloneApplyAllSourceRevisionAdvanceExecutionProof(proof),
			provenance: cloneApplyAllSourceRevisionAdvanceExecutionProofProvenance(proof.provenance),
			minted:     true,
		},
	}
	if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err != nil {
		t.Fatalf("validate admission fixture: %v", err)
	}
	return admission
}

func assertZeroApplyAllSourceRevisionAdvanceExecutionAdmission(
	t *testing.T,
	admission applyAllSourceRevisionAdvanceExecutionAdmission,
) {
	t.Helper()
	if admission.identity != nil {
		t.Fatalf("admission = %#v, want all-zero admission", admission)
	}
}