package controlplane

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"testing"

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

type selectedLivePreflightSchemaProvisioner struct {
	runtimeports.SchemaProvisioner
	verify func(*database.TiDBConnection) error
}

func (p selectedLivePreflightSchemaProvisioner) VerifySchemaReadiness(
	db *database.TiDBConnection,
) error {
	return p.verify(db)
}

func TestPreflightApplyAllSourceRevisionAdvanceSelectedTargetMintsExactOrderedReceipt(t *testing.T) {
	queries := []string{}
	matcher := sqlmock.QueryMatcherFunc(func(expectedSQL, actualSQL string) error {
		if err := sqlmock.QueryMatcherRegexp.Match(expectedSQL, actualSQL); err != nil {
			return err
		}
		queries = append(queries, actualSQL)
		return nil
	})
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(matcher))
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	prepared, selected, dependencyRow := testApplyAllSourceRevisionAdvanceSelectedLivePreflightPrepared(t, db)
	runtime := prepared.admission.identity.proof.executionRuntime

	originalSchemaProvisioner := schemaProvisioner
	schemaChecked := false
	schemaProvisioner = selectedLivePreflightSchemaProvisioner{
		SchemaProvisioner: originalSchemaProvisioner,
		verify: func(candidate *database.TiDBConnection) error {
			if candidate != runtime.runnerDatabase {
				return fmt.Errorf("readiness used a different TiDB connection")
			}
			if len(queries) != 0 {
				return fmt.Errorf("readiness ran after another live query")
			}
			schemaChecked = true
			return nil
		},
	}
	t.Cleanup(func() { schemaProvisioner = originalSchemaProvisioner })

	expectSelectedLivePreflightDependencyRow(mock, dependencyRow, "1.0.0")
	expectInstalledPackStateLookup(mock, dependencyRow.Name, installedPackStatusActive, 0)
	mock.ExpectQuery(`(?s)SELECT Id.*FROM ProcessDefinition.*WHERE COALESCE\(OwnerPack, ''\) = \?`).
		WithArgs(selected.installedCompatibility.Name()).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	expectApplyAllSourceRevisionAdvanceAttributionInstalledPackListRows(mock, dependencyRow)
	expectInstalledPackStateLookup(mock, dependencyRow.Name, installedPackStatusActive, 0)
	allowedObjects, err := AllowedL0ObjectAPINames(context.Background(), db)
	if err != nil {
		t.Fatalf("allowed L0 objects: %v", err)
	}
	allowedPacks := map[string]struct{}{
		strings.ToLower(selected.installedCompatibility.Name()): {},
		strings.ToLower(dependencyRow.Name):                     {},
	}
	for _, runtimePack := range packs.RuntimeRequiredPacks() {
		allowedPacks[strings.ToLower(runtimePack.Name())] = struct{}{}
	}
	expectSuccessfulApplyAllSourceRevisionAdvanceAttributionSurfaceChecks(
		mock, sortedKeys(allowedObjects), sortedKeys(allowedPacks),
	)
	expectInstalledPackRuntimeDeclarationColumns(mock, true)
	expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, true)
	expectApplyAllSourceRevisionAdvanceHostDigestColumn(mock, true)
	expectApplyAllSourceRevisionAdvancePreconditionRow(mock, selected.current.row.Name, selected.current.row)

	receipt, err := preflightApplyAllSourceRevisionAdvanceSelectedTarget(
		context.Background(), prepared, nil,
	)
	if err != nil {
		t.Fatalf("selected live preflight: %v", err)
	}
	if !schemaChecked {
		t.Fatal("selected live preflight skipped schema readiness")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
	if len(queries) == 0 || !strings.Contains(queries[len(queries)-1], "SELECT Id, Name, COALESCE(PackInstallId, '')") {
		t.Fatalf("final live query was not the whole selected installed-row reread: %v", queries)
	}
	state, err := requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(receipt)
	if err != nil {
		t.Fatalf("require selected live-preflight receipt: %v", err)
	}
	if state.prepared != prepared || state.target != selected.installedCompatibility ||
		state.database != runtime.database || state.runnerDatabase != runtime.runnerDatabase ||
		state.applyDeps.DB != runtime.database || state.applyDeps.DBConn != runtime.runnerDatabase ||
		!state.registrations.SameSnapshot(runtime.componentKindRegistrationSet) ||
		state.applyOptions != prepared.applyOptions || state.applyDeps.Options != prepared.applyOptions {
		t.Fatalf("receipt state lost exact prepared authority: %#v", state)
	}
}

func testApplyAllSourceRevisionAdvanceSelectedLivePreflightPrepared(
	t *testing.T,
	db *sql.DB,
) (*applyAllSourceRevisionAdvanceAttemptIdentity, applyAllSourceRevisionAdvanceExecutionProofEntry, installedPackRow) {
	t.Helper()
	makeEntry := func(
		advance normalizedApplyAllSourceRevisionAdvance,
		currentTarget *packs.SourceBackedPack,
		target *packs.SourceBackedPack,
	) applyAllSourceRevisionAdvanceExecutionProofEntry {
		current := testApplyAllSourceRevisionAdvanceFrozenCurrentTargetProof(
			t, advance, currentTarget, installedPackStatusActive, advance.Kind,
			testCurrentTargetV1Hash, testCurrentTargetV1Hash,
		)
		atom, err := packs.FreezeSourceBackedExecutionAtom(target)
		if err != nil {
			t.Fatalf("freeze %s execution atom: %v", advance.Name, err)
		}
		return applyAllSourceRevisionAdvanceExecutionProofEntry{
			name: advance.Name, current: current.snapshot.entries[0],
			currentTarget: currentTarget, installedCompatibility: target, atom: atom,
		}
	}

	stableAdvance := testApplyAllSourceRevisionAdvance("stable", 1)
	stableTarget := testApplyAllSourceRevisionTarget(
		t, stableAdvance.Name, "1.0.0", stableAdvance.Kind, stableAdvance.URI, stableAdvance.ToRevision,
	)
	stable := makeEntry(stableAdvance, stableTarget, stableTarget)
	if stable.current.pending != nil {
		t.Fatalf("stable fixture pending = %#v, want nil", stable.current.pending)
	}

	dependencyAdvance := testApplyAllSourceRevisionAdvance("dependency", 2)
	dependencyCurrent := testApplyAllSourceRevisionTarget(
		t, dependencyAdvance.Name, "1.0.0", dependencyAdvance.Kind, dependencyAdvance.URI, dependencyAdvance.FromRevision,
	)
	dependencyTarget := testApplyAllSourceRevisionTarget(
		t, dependencyAdvance.Name, "2.0.0", dependencyAdvance.Kind, dependencyAdvance.URI, dependencyAdvance.ToRevision,
	)
	dependency := makeEntry(dependencyAdvance, dependencyCurrent, dependencyTarget)

	selectedAdvance := testApplyAllSourceRevisionAdvance("selected", 3)
	selectedCurrent := testApplyAllSourceRevisionTarget(
		t, selectedAdvance.Name, "1.0.0", selectedAdvance.Kind, selectedAdvance.URI, selectedAdvance.FromRevision,
	)
	selectedTarget := testApplyAllSourceRevisionTarget(
		t, selectedAdvance.Name, "1.0.0", selectedAdvance.Kind, selectedAdvance.URI, selectedAdvance.ToRevision,
		packs.PackDependency{Name: dependencyAdvance.Name, VersionConstraint: ">=2.0.0"},
	)
	selected := makeEntry(selectedAdvance, selectedCurrent, selectedTarget)
	entries := []applyAllSourceRevisionAdvanceExecutionProofEntry{stable, dependency, selected}
	admission := testApplyAllSourceRevisionAdvanceExecutionAdmissionFromEntries(t, db, entries)
	runtime := admission.identity.proof.executionRuntime
	registrations := runnerTestRegistrationSet(t)
	runtime.componentKindRegistrationSet = registrations
	runtime.provenance.componentKindRegistrationSet = registrations
	if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err != nil {
		t.Fatalf("validate registered live-preflight admission: %v", err)
	}
	applyOptions := packs.ApplyOptions{
		SkipPolymorphicDiscriminatorBackfill: true,
		RepairLegacyUniqueDuplicates:         true,
	}
	prepared, err := prepareApplyAllSourceRevisionAdvanceAttempt(admission, 1, applyOptions)
	if err != nil {
		t.Fatalf("prepare selected live-preflight attempt: %v", err)
	}
	if prepared.pendingOrdinal != 1 || prepared.selectedProofIndex != 2 {
		t.Fatalf("prepared selection = ordinal %d index %d, want ordinal 1 index 2", prepared.pendingOrdinal, prepared.selectedProofIndex)
	}
	return prepared, selected, dependency.current.row
}

func expectSelectedLivePreflightDependencyRow(
	mock sqlmock.Sqlmock,
	row installedPackRow,
	liveVersion string,
) {
	mock.ExpectQuery(`(?s)SELECT Name, Category, Status.*HostRequirementsDigest.*FROM _System_InstalledPack.*COALESCE\(IsDeleted, 0\) = 0 AND Name = \?.*LIMIT 1`).
		WithArgs(row.Name).
		WillReturnRows(sqlmock.NewRows([]string{
			"Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility",
			"DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest",
		}).AddRow(
			row.Name, row.Category, row.Status, liveVersion,
			testApplyAllSourceRevisionAdvanceNullValue(row.ManifestFormatVersion),
			testApplyAllSourceRevisionAdvanceNullValue(row.PlatformCompatibility),
			testApplyAllSourceRevisionAdvanceNullValue(row.DesiredPayloadHash),
			testApplyAllSourceRevisionAdvanceNullValue(row.AppliedPayloadHash),
			testApplyAllSourceRevisionAdvanceNullValue(row.PayloadHashAlgorithm),
			testApplyAllSourceRevisionAdvanceNullValue(row.SourceKind),
			testApplyAllSourceRevisionAdvanceNullValue(row.SourceURI),
			testApplyAllSourceRevisionAdvanceNullValue(row.SourceRevision),
			testApplyAllSourceRevisionAdvanceNullValue(row.HostRequirementsDigest),
		))
}

func TestPreflightApplyAllSourceRevisionAdvanceSelectedTargetRejectsFinalRowDrift(t *testing.T) {
	tests := []struct {
		name   string
		change func(*installedPackRow) bool
	}{
		{name: "missing", change: func(*installedPackRow) bool { return false }},
		{name: "row identity", change: func(row *installedPackRow) bool { row.ID += "-changed"; return true }},
		{name: "incarnation", change: func(row *installedPackRow) bool { row.PackInstallID += "-changed"; return true }},
		{name: "status", change: func(row *installedPackRow) bool { row.Status = installedPackStatusFailed; return true }},
		{name: "deletion", change: func(row *installedPackRow) bool { row.IsDeleted = 1; return true }},
		{name: "source revision", change: func(row *installedPackRow) bool { row.SourceRevision.String += "-changed"; return true }},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			receipt, err, _ := testRunApplyAllSourceRevisionAdvanceSelectedLivePreflight(
				t, test.change,
			)
			if err == nil {
				t.Fatal("selected live preflight accepted final installed-row drift")
			}
			if receipt.identity != nil {
				t.Fatalf("failed live preflight returned receipt %#v", receipt)
			}
		})
	}
}

func TestSelectedLivePreflightReceiptRejectsCoordinatedAuthorityMutation(t *testing.T) {
	if _, err := requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(
		applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt{},
	); err == nil {
		t.Fatal("zero selected live-preflight receipt was accepted")
	}
	tests := []struct {
		name   string
		mutate func(*testing.T, applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt)
	}{
		{name: "apply options", mutate: func(t *testing.T, receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt) {
			prepared := receipt.identity.prepared
			prepared.applyOptions.RepairLegacyUniqueDuplicates = false
			prepared.provenance.applyOptions = prepared.applyOptions
			assertPreparedSelfConsistentButReceiptRejected(t, receipt)
		}},
		{name: "pending selection", mutate: func(t *testing.T, receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt) {
			prepared := receipt.identity.prepared
			prepared.pendingOrdinal = 0
			prepared.selectedProofIndex = 1
			prepared.provenance.pendingOrdinal = 0
			prepared.provenance.selectedProofIndex = 1
			assertPreparedSelfConsistentButReceiptRejected(t, receipt)
		}},
		{name: "registration snapshot", mutate: func(t *testing.T, receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt) {
			prepared := receipt.identity.prepared
			runtime := prepared.admission.identity.proof.executionRuntime
			replacement, err := packs.NewComponentKindRegistrationSet(
				runtime.componentKindRegistrationSet.Registrations(),
			)
			if err != nil {
				t.Fatalf("rebuild registration set: %v", err)
			}
			runtime.componentKindRegistrationSet = replacement
			runtime.provenance.componentKindRegistrationSet = replacement
			prepared.provenance.executionRuntimeWitness.componentKindRegistrationSet = replacement
			prepared.provenance.registrations = replacement
			assertPreparedSelfConsistentButReceiptRejected(t, receipt)
		}},
		{name: "one-shot claim identity", mutate: func(t *testing.T, receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt) {
			replacement := &applyAllSourceRevisionAdvanceSelectedLivePreflightClaim{}
			receipt.identity.claim = replacement
			receipt.identity.provenance.claim = replacement
			if _, err := requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(receipt); err == nil {
				t.Fatal("receipt accepted a coordinated one-shot claim substitution")
			}
		}},
		{name: "final row witness", mutate: func(t *testing.T, receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt) {
			receipt.identity.finalRow.Status = installedPackStatusFailed
			receipt.identity.provenance.finalRow = receipt.identity.finalRow
			if _, err := requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(receipt); err == nil {
				t.Fatal("receipt accepted a coordinated final-row witness change")
			}
		}},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			receipt, err, _ := testRunApplyAllSourceRevisionAdvanceSelectedLivePreflight(
				t, func(*installedPackRow) bool { return true },
			)
			if err != nil {
				t.Fatalf("mint selected live-preflight receipt: %v", err)
			}
			test.mutate(t, receipt)
		})
	}
}

func TestSelectedLivePreflightOwnsCompleteReadOrderWithoutExecution(t *testing.T) {
	preflight := parseControlplaneFuncDecl(
		t,
		"apply_all_source_revision_advance_selected_live_preflight.go",
		"preflightApplyAllSourceRevisionAdvanceSelectedTarget",
	)
	lastPosition := preflight.Pos()
	for _, name := range []string{
		"ensureInstallPackSchemaReadinessForConnection",
		"validateLivePackDependenciesWithPreflightOverrides",
		"validateSourceBackedPackInstallPreflights",
		"installedPackHasRuntimeDeclarationColumns",
		"installedPackHasRuntimeStartupShallowDeclarationColumn",
		"installedPackHasRuntimeFormulaFunctionBundlesColumn",
		"installedPackHasHostRequirementsDigestColumn",
		"getInstalledPackRowByNameForSourceRevisionAdvance",
	} {
		call := findFuncCallByName(preflight, name)
		if call == nil {
			t.Fatalf("selected live preflight is missing %s", name)
		}
		if call.Pos() <= lastPosition {
			t.Fatalf("selected live-preflight call %s is out of order", name)
		}
		lastPosition = call.Pos()
	}
	for _, forbidden := range []string{
		"beginApplyAllSourceRevisionAdvanceAttempt",
		"captureApplyAllSourceRevisionAdvanceSnapshots",
		"registerApplyAllSourceRevisionAdvanceRuntime",
		"ApplySourceBackedExecutionAtom",
	} {
		if findFuncCallByName(preflight, forbidden) != nil {
			t.Fatalf("production-inert selected live preflight calls %s", forbidden)
		}
	}
}

func assertPreparedSelfConsistentButReceiptRejected(
	t *testing.T,
	receipt applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt,
) {
	t.Helper()
	if err := validateApplyAllSourceRevisionAdvanceAttemptIdentity(receipt.identity.prepared, false); err != nil {
		t.Fatalf("coordinated mutation invalidated prepared identity: %v", err)
	}
	if _, err := requireApplyAllSourceRevisionAdvanceSelectedLivePreflightReceipt(receipt); err == nil {
		t.Fatal("receipt accepted coordinated prepared-authority mutation")
	}
}

func testRunApplyAllSourceRevisionAdvanceSelectedLivePreflight(
	t *testing.T,
	finalRow func(*installedPackRow) bool,
) (
	applyAllSourceRevisionAdvanceSelectedLivePreflightReceipt,
	error,
	*applyAllSourceRevisionAdvanceAttemptIdentity,
) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer db.Close()
	prepared, selected, dependencyRow := testApplyAllSourceRevisionAdvanceSelectedLivePreflightPrepared(t, db)
	runtime := prepared.admission.identity.proof.executionRuntime
	originalSchemaProvisioner := schemaProvisioner
	schemaProvisioner = selectedLivePreflightSchemaProvisioner{
		SchemaProvisioner: originalSchemaProvisioner,
		verify: func(candidate *database.TiDBConnection) error {
			if candidate != runtime.runnerDatabase {
				return fmt.Errorf("readiness used a different TiDB connection")
			}
			return nil
		},
	}
	defer func() { schemaProvisioner = originalSchemaProvisioner }()

	expectSelectedLivePreflightBeforeLifecycle(t, mock, db, selected, dependencyRow)
	expectInstalledPackRuntimeDeclarationColumns(mock, true)
	expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, true)
	expectApplyAllSourceRevisionAdvanceHostDigestColumn(mock, true)
	live := selected.current.row
	if finalRow(&live) {
		expectApplyAllSourceRevisionAdvancePreconditionRow(mock, selected.current.row.Name, live)
	} else {
		mock.ExpectQuery(applyAllSourceRevisionAdvanceRowQueryPattern()).
			WithArgs(selected.current.row.Name).
			WillReturnRows(testApplyAllSourceRevisionAdvancePreconditionRows())
	}
	receipt, preflightErr := preflightApplyAllSourceRevisionAdvanceSelectedTarget(
		context.Background(), prepared, nil,
	)
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
	return receipt, preflightErr, prepared
}

func expectSelectedLivePreflightBeforeLifecycle(
	t *testing.T,
	mock sqlmock.Sqlmock,
	db *sql.DB,
	selected applyAllSourceRevisionAdvanceExecutionProofEntry,
	dependencyRow installedPackRow,
) {
	t.Helper()
	expectSelectedLivePreflightDependencyRow(mock, dependencyRow, "1.0.0")
	expectInstalledPackStateLookup(mock, dependencyRow.Name, installedPackStatusActive, 0)
	mock.ExpectQuery(`(?s)SELECT Id.*FROM ProcessDefinition.*WHERE COALESCE\(OwnerPack, ''\) = \?`).
		WithArgs(selected.installedCompatibility.Name()).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	expectApplyAllSourceRevisionAdvanceAttributionInstalledPackListRows(mock, dependencyRow)
	expectInstalledPackStateLookup(mock, dependencyRow.Name, installedPackStatusActive, 0)
	allowedObjects, err := AllowedL0ObjectAPINames(context.Background(), db)
	if err != nil {
		t.Fatalf("allowed L0 objects: %v", err)
	}
	allowedPacks := map[string]struct{}{
		strings.ToLower(selected.installedCompatibility.Name()): {},
		strings.ToLower(dependencyRow.Name):                     {},
	}
	for _, runtimePack := range packs.RuntimeRequiredPacks() {
		allowedPacks[strings.ToLower(runtimePack.Name())] = struct{}{}
	}
	expectSuccessfulApplyAllSourceRevisionAdvanceAttributionSurfaceChecks(
		mock, sortedKeys(allowedObjects), sortedKeys(allowedPacks),
	)
}