package packs

import (
	"context"
	"database/sql"
	"os"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
)

func testFormulaStorageTransitionDefinition(formula string) []domainschema.TableDefinition {
	return []domainschema.TableDefinition{{
		TableName: "Lead",
		Columns: []domainschema.ColumnDefinition{{
			Name:               "EmailDomain__c",
			LogicalType:        "formula",
			Formula:            formula,
			FormulaMode:        "active",
			FormulaStorageMode: "write_materialized",
			ReturnType:         "string",
		}},
	}}
}

func expectLiveViewOnlyFormula(t *testing.T, mock sqlmock.Sqlmock, formula string) {
	t.Helper()
	mock.ExpectQuery("SELECT f.Type").
		WithArgs("Lead", "EmailDomain__c").
		WillReturnRows(sqlmock.NewRows([]string{"Type", "FormulaMode", "FormulaStorageMode", "Formula", "ReturnType", "ProjectionConfig"}).
			AddRow("formula", "active", "", formula, "string", ""))
}

func TestPlanObjectSpecFormulaStorageTransitionsDetectsExactViewOnlyEdge(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	formula := `SUBSTITUTE(Email, LEFT(Email, FIND("@", Email)), NULL)`
	expectLiveViewOnlyFormula(t, mock, formula)

	got, err := planObjectSpecFormulaStorageTransitions(context.Background(), db, testFormulaStorageTransitionDefinition(formula))
	if err != nil {
		t.Fatalf("plan transitions: %v", err)
	}
	if len(got) != 1 || got[0].objectAPIName != "Lead" || got[0].fieldAPIName != "EmailDomain__c" || got[0].viewName != "v_lead" {
		t.Fatalf("transitions = %#v", got)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestPlanObjectSpecFormulaStorageTransitionsRejectsSemanticChange(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectQuery("SELECT f.Type").
		WithArgs("Lead", "EmailDomain__c").
		WillReturnRows(sqlmock.NewRows([]string{"Type", "FormulaMode", "FormulaStorageMode", "Formula", "ReturnType", "ProjectionConfig"}).
			AddRow("formula", "active", "", "LOWER(Email)", "string", ""))

	_, err = planObjectSpecFormulaStorageTransitions(context.Background(), db, testFormulaStorageTransitionDefinition("UPPER(Email)"))
	if err == nil || !regexp.MustCompile("split semantic and storage changes").MatchString(err.Error()) {
		t.Fatalf("error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestPlanObjectSpecFormulaStorageTransitionsRejectsProjectionSemanticChange(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectQuery("SELECT f.Type").
		WithArgs("Lead", "EmailDomain__c").
		WillReturnRows(sqlmock.NewRows([]string{"Type", "FormulaMode", "FormulaStorageMode", "Formula", "ReturnType", "ProjectionConfig"}).
			AddRow("formula", "active", "", "", "string", `{"kind":"lookup","targetObject":"Account"}`))

	_, err = planObjectSpecFormulaStorageTransitions(
		context.Background(), db, testFormulaStorageTransitionDefinition("UPPER(Email)"),
	)
	if err == nil || !strings.Contains(err.Error(), "split semantic and storage changes") {
		t.Fatalf("error = %v, want projection semantic-change rejection", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestApplyObjectSpecFormulaStorageTransitionsCopiesAndVerifies(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	boundary := "SELECT `Id` FROM `Lead` WHERE `Id` > ? ORDER BY `Id` LIMIT ?, 1"
	tail := "SELECT `Id` FROM `Lead` WHERE `Id` > ? ORDER BY `Id` DESC LIMIT 1"
	mock.ExpectQuery(regexp.QuoteMeta(boundary)).
		WithArgs("", objectSpecFormulaStorageTransitionBatchRows-1).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(tail)).
		WithArgs("").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("lead-2"))
	update := "UPDATE `Lead` AS target JOIN `v_lead` AS computed ON target.`Id` = computed.`Id` SET target.`EmailDomain__c` = computed.`EmailDomain__c` WHERE target.`Id` > ? AND target.`Id` <= ? AND computed.`Id` > ? AND computed.`Id` <= ? AND (NOT (target.`EmailDomain__c` <=> computed.`EmailDomain__c`))"
	// Keep the SQL assertion exact: every mutating statement must be bounded by
	// one physical primary-key range inside the stopped-Runtime candidate.
	mock.ExpectExec(regexp.QuoteMeta(update)).
		WithArgs("", "lead-2", "", "lead-2").
		WillReturnResult(sqlmock.NewResult(0, 2))
	verify := "SELECT COUNT(*) FROM `Lead` AS target JOIN `v_lead` AS computed ON target.`Id` = computed.`Id` WHERE target.`Id` > ? AND target.`Id` <= ? AND computed.`Id` > ? AND computed.`Id` <= ? AND (NOT (target.`EmailDomain__c` <=> computed.`EmailDomain__c`))"
	mock.ExpectQuery(regexp.QuoteMeta(verify)).
		WithArgs("", "lead-2", "", "lead-2").
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(0))
	mock.ExpectQuery(regexp.QuoteMeta(boundary)).
		WithArgs("lead-2", objectSpecFormulaStorageTransitionBatchRows-1).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(regexp.QuoteMeta(tail)).
		WithArgs("lead-2").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))

	err = applyObjectSpecFormulaStorageTransitions(context.Background(), db, []objectSpecFormulaStorageTransition{{
		objectAPIName: "Lead",
		fieldAPIName:  "EmailDomain__c",
		viewName:      "v_lead",
	}})
	if err != nil {
		t.Fatalf("apply transitions: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestRebuildObjectSpecFormulaStorageTransitionSourcesRejectsDeferredRebuild(t *testing.T) {
	err := rebuildObjectSpecFormulaStorageTransitionSources(
		context.Background(),
		ApplyDeps{Options: ApplyOptions{DeferFormulaViewRebuild: true}},
		[]objectSpecFormulaStorageTransition{{objectAPIName: "Lead", fieldAPIName: "EmailDomain__c", viewName: "v_lead"}},
	)
	if err == nil || !strings.Contains(err.Error(), "cannot defer formula-view rebuild") {
		t.Fatalf("deferred transition error = %v, want fail-closed conflict", err)
	}
}

func TestFormulaStorageTransitionReusesCompleteDependencyOrderedViewRebuild(t *testing.T) {
	raw, err := os.ReadFile("object_spec_formula_storage_transition.go")
	if err != nil {
		t.Fatal(err)
	}
	body := string(raw)
	start := strings.Index(body, "func rebuildObjectSpecFormulaStorageTransitionSources(")
	end := strings.Index(body, "func sameFormulaStorageTransitionSemantics(")
	if start < 0 || end <= start {
		t.Fatalf("formula transition rebuild function bounds = (%d, %d)", start, end)
	}
	rebuild := body[start:end]
	if !strings.Contains(rebuild, "deps.FormulaView.RebuildAllViews(ctx)") {
		t.Fatal("formula transition must rebuild the complete active dependency graph")
	}
	if strings.Contains(rebuild, "deps.FormulaView.RebuildView(") || strings.Contains(rebuild, "deps.FormulaView.OrderForBuild(") {
		t.Fatal("formula transition must not maintain a partial dependency-ordering path")
	}
}

func TestFormulaStorageTransitionTiDBCopiesSwitchesAndUsesDeclaredIndex(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openFlowRuntimeMaintenanceTestDB(t)
	defer cleanup()

	formula := `SUBSTITUTE(Email, LEFT(Email, FIND("@", Email)), NULL)`
	for _, statement := range []string{
		`DROP VIEW IF EXISTS v_lead`,
		"DROP TABLE IF EXISTS `_System_Field`, `_System_Object`, `Lead`",
		`CREATE TABLE _System_Object (
			Id VARCHAR(64) PRIMARY KEY,
			ApiName VARCHAR(255) NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT FALSE
		)`,
		`CREATE TABLE _System_Field (
			Id VARCHAR(64) PRIMARY KEY,
			ObjectId VARCHAR(64) NOT NULL,
			ApiName VARCHAR(255) NOT NULL,
			Type VARCHAR(64) NOT NULL,
			FormulaMode VARCHAR(64) NULL,
			FormulaStorageMode VARCHAR(64) NULL,
			Formula TEXT NULL,
			ReturnType VARCHAR(64) NULL,
			ProjectionConfig JSON NULL,
			IsDeleted BOOL NOT NULL DEFAULT FALSE
		)`,
		"CREATE TABLE `Lead` (\n" + `
			Id VARCHAR(64) PRIMARY KEY,
			Email VARCHAR(255) NULL,
			EmailDomain__c VARCHAR(255) NULL,
			IsDeleted BOOL NOT NULL DEFAULT FALSE,
			IsConverted BOOL NOT NULL DEFAULT FALSE
		)`,
		`CREATE SQL SECURITY INVOKER VIEW v_lead AS
		 SELECT Id, Email,
		        REPLACE(Email, LEFT(Email, LOCATE('@', Email)), '') AS EmailDomain__c,
		        IsDeleted, IsConverted
		   FROM ` + "`Lead`",
	} {
		if _, err := db.ExecContext(ctx, statement); err != nil {
			t.Fatalf("prepare TiDB transition fixture: %v\nSQL:\n%s", err, statement)
		}
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_Object (Id, ApiName) VALUES ('obj-lead', 'Lead')`); err != nil {
		t.Fatalf("seed TiDB object authority: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_Field
			(Id, ObjectId, ApiName, Type, FormulaMode, FormulaStorageMode, Formula, ReturnType)
		VALUES ('field-domain', 'obj-lead', 'EmailDomain__c', 'formula', 'active', '', ?, 'string')`, formula); err != nil {
		t.Fatalf("seed TiDB formula authority: %v", err)
	}
	if _, err := db.ExecContext(ctx, "INSERT INTO `Lead`\n"+`
		(Id, Email, EmailDomain__c, IsDeleted, IsConverted) VALUES
		('1', 'one@example.com', NULL, FALSE, FALSE),
		('2', 'two@example.com', 'stale.invalid', FALSE, FALSE),
		('3', 'three@other.test', 'other.test', FALSE, FALSE),
		('4', 'four@example.com', NULL, TRUE, FALSE),
		('5', 'five@example.com', NULL, FALSE, TRUE)`); err != nil {
		t.Fatalf("seed TiDB Lead rows: %v", err)
	}

	defs := testFormulaStorageTransitionDefinition(formula)
	transitions, err := planObjectSpecFormulaStorageTransitions(ctx, db, defs)
	if err != nil {
		t.Fatalf("plan TiDB transition: %v", err)
	}
	if len(transitions) != 1 {
		t.Fatalf("TiDB transitions = %#v, want one", transitions)
	}
	if err := applyObjectSpecFormulaStorageTransitions(ctx, db, transitions); err != nil {
		t.Fatalf("apply TiDB transition: %v", err)
	}
	// A stopped-candidate retry before metadata publication must be harmless.
	if err := applyObjectSpecFormulaStorageTransitions(ctx, db, transitions); err != nil {
		t.Fatalf("retry TiDB transition: %v", err)
	}

	var mismatches int
	if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM `Lead` AS target\n"+`
		JOIN v_lead AS computed ON target.Id = computed.Id
		WHERE NOT (target.EmailDomain__c <=> computed.EmailDomain__c)`).Scan(&mismatches); err != nil {
		t.Fatalf("verify TiDB copied formula values: %v", err)
	}
	if mismatches != 0 {
		t.Fatalf("TiDB copied formula mismatches = %d, want 0", mismatches)
	}

	defs[0].Indices = []domainschema.IndexDefinition{{
		Name:    "idx_lead_email_domain_deleted_converted",
		Columns: []string{"EmailDomain__c", "IsDeleted", "IsConverted"},
	}}
	if err := reconcilePackTableIndices(db, defs); err != nil {
		t.Fatalf("reconcile TiDB formula index: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_Field
		SET FormulaStorageMode = 'write_materialized'
		WHERE Id = 'field-domain'`); err != nil {
		t.Fatalf("publish TiDB materialized metadata: %v", err)
	}
	if _, err := db.ExecContext(ctx, "CREATE OR REPLACE SQL SECURITY INVOKER VIEW v_lead AS SELECT Id, Email, EmailDomain__c, IsDeleted, IsConverted FROM `Lead`"); err != nil {
		t.Fatalf("switch TiDB formula read path: %v", err)
	}

	transitions, err = planObjectSpecFormulaStorageTransitions(ctx, db, defs)
	if err != nil {
		t.Fatalf("re-plan completed TiDB transition: %v", err)
	}
	if len(transitions) != 0 {
		t.Fatalf("completed TiDB transitions = %#v, want none", transitions)
	}

	rows, err := db.QueryContext(ctx, `EXPLAIN FORMAT='brief'
		SELECT Id, EmailDomain__c FROM v_lead
		WHERE EmailDomain__c = 'example.com' AND IsDeleted = FALSE AND IsConverted = FALSE`)
	if err != nil {
		t.Fatalf("explain TiDB indexed formula lookup: %v", err)
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatalf("read TiDB explain columns: %v", err)
	}
	var plan strings.Builder
	for rows.Next() {
		values := make([]sql.NullString, len(columns))
		dest := make([]any, len(columns))
		for i := range values {
			dest[i] = &values[i]
		}
		if err := rows.Scan(dest...); err != nil {
			t.Fatalf("scan TiDB explain row: %v", err)
		}
		for _, value := range values {
			if value.Valid {
				plan.WriteString(value.String)
				plan.WriteByte(' ')
			}
		}
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate TiDB explain rows: %v", err)
	}
	if !strings.Contains(strings.ToLower(plan.String()), "idx_lead_email_domain_deleted_converted") {
		t.Fatalf("TiDB lookup plan does not use declared formula index:\n%s", plan.String())
	}
}

func TestApplyObjectSpecsOrdersFormulaStorageTransitionBeforeIndexAndMetadata(t *testing.T) {
	raw, err := os.ReadFile("platform_core.go")
	if err != nil {
		t.Fatalf("read platform_core.go: %v", err)
	}
	body := string(raw)
	plan := strings.Index(body, "planObjectSpecFormulaStorageTransitions")
	guard := strings.Index(body, "offlinecandidateauthority.FormulaStorageTransitionAllowed(")
	rebuild := strings.Index(body, "rebuildObjectSpecFormulaStorageTransitionSources")
	create := strings.Index(body, "deps.Schema.BatchCreatePhysicalTables")
	reconcile := strings.Index(body, "schemaProvisioner.ReconcileTableColumns")
	backfill := strings.Index(body, "applyObjectSpecFormulaStorageTransitions")
	index := strings.Index(body, "reconcilePackTableIndices")
	publish := strings.Index(body, "applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt")
	if plan < 0 || guard < 0 || rebuild < 0 || create < 0 || reconcile < 0 || backfill < 0 || index < 0 || publish < 0 {
		t.Fatalf("formula transition ordering anchors missing: plan=%d guard=%d rebuild=%d create=%d reconcile=%d backfill=%d index=%d publish=%d", plan, guard, rebuild, create, reconcile, backfill, index, publish)
	}
	if !(plan < guard && guard < rebuild && rebuild < create && create < reconcile && reconcile < backfill && backfill < index && index < publish) {
		t.Fatalf("formula transition must fail closed and rebuild its authority before DDL, then reconcile column, backfill, index, and publish metadata: plan=%d guard=%d rebuild=%d create=%d reconcile=%d backfill=%d index=%d publish=%d", plan, guard, rebuild, create, reconcile, backfill, index, publish)
	}
}