//go:build integration

package controlplane

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

	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/pingcap-inc/nexuscrm-controlplane/bootstrapimpl"
)

const connectorIngressControlTable = "_System_ConnectorIngressControl"

func TestConnectorIngressFenceEpochSchemaUpgradePreservesExistingControlRow(t *testing.T) {
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	db := conn.DB()

	const rowID = "ingress-control-upgrade-proof"
	if _, err := db.ExecContext(ctx, `
		INSERT INTO `+connectorIngressControlTable+`
			(Id, Namespace, PackInstallId, Channel, Mode, RetryAfterSeconds)
		VALUES (?, ?, ?, ?, ?, ?)`,
		rowID,
		"connector-upgrade-proof",
		"",
		"webhook",
		"active",
		120,
	); err != nil {
		t.Fatalf("insert pre-upgrade ingress control row: %v", err)
	}

	if _, err := db.ExecContext(ctx, "ALTER TABLE `"+connectorIngressControlTable+"` DROP COLUMN `FenceEpoch`"); err != nil {
		t.Fatalf("prepare supported prior ingress-control schema: %v", err)
	}
	assertConnectorIngressFenceEpochMissing(t, db)

	// InitializeSchema invokes this authoritative reconciliation phase before
	// metadata and index reconciliation. Isolating the affected table here
	// proves the supported warm upgrade without making the evidence depend on
	// unrelated DDL in the rest of the managed catalog.
	if err := bootstrapimpl.ReconcileTableColumns(db, []schema.TableDefinition{connectorIngressControlDefinition(t)}); err != nil {
		t.Fatalf("upgrade ingress-control schema from supported prior state: %v", err)
	}
	assertConnectorIngressFenceEpochReady(t, db)

	var epoch int64
	var mode string
	if err := db.QueryRowContext(ctx,
		"SELECT FenceEpoch, Mode FROM `"+connectorIngressControlTable+"` WHERE Id = ?",
		rowID,
	).Scan(&epoch, &mode); err != nil {
		t.Fatalf("read upgraded ingress control row: %v", err)
	}
	if epoch != 0 || mode != "active" {
		t.Fatalf("upgraded ingress control row = epoch %d mode %q, want epoch 0 active", epoch, mode)
	}
}

// TestConnectorIngressFenceWarmUpgradeCanonicalizesNullableScopeAndEpoch
// proves the supported no-reset upgrade path for the two nullable legacy
// values which would otherwise make the namespace fence ambiguous.  The
// canonical catalog row is (Namespace, PackInstallId=”, Channel='*') with
// FenceEpoch=0; ReconcileTableColumns must establish that representation
// before a new connector runtime is allowed to use the row as a fence.
func TestConnectorIngressFenceWarmUpgradeCanonicalizesNullableScopeAndEpoch(t *testing.T) {
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	db := conn.DB()

	if _, err := db.ExecContext(ctx, "ALTER TABLE `"+connectorIngressControlTable+"` MODIFY COLUMN `PackInstallId` VARCHAR(255) NULL DEFAULT NULL"); err != nil {
		t.Fatalf("prepare nullable legacy PackInstallId: %v", err)
	}
	if _, err := db.ExecContext(ctx, "ALTER TABLE `"+connectorIngressControlTable+"` MODIFY COLUMN `FenceEpoch` BIGINT NULL DEFAULT NULL"); err != nil {
		t.Fatalf("prepare nullable legacy FenceEpoch: %v", err)
	}

	const rowID = "ingress-control-nullable-warm-upgrade"
	if _, err := db.ExecContext(ctx, `
		INSERT INTO `+connectorIngressControlTable+`
			(Id, Namespace, PackInstallId, Channel, Mode, FenceEpoch, RetryAfterSeconds)
		VALUES (?, ?, NULL, ?, ?, NULL, ?)`,
		rowID,
		"connector-nullable-warm-upgrade",
		"*",
		"paused",
		120,
	); err != nil {
		t.Fatalf("insert nullable legacy namespace fence: %v", err)
	}

	if err := bootstrapimpl.ReconcileTableColumns(db, []schema.TableDefinition{connectorIngressControlDefinition(t)}); err != nil {
		t.Fatalf("reconcile nullable ingress-control warm upgrade: %v", err)
	}
	assertConnectorIngressControlScopeReady(t, db)
	assertConnectorIngressFenceEpochReady(t, db)

	var packInstallID string
	var epoch int64
	var channel, mode string
	if err := db.QueryRowContext(ctx,
		"SELECT PackInstallId, FenceEpoch, Channel, Mode FROM `"+connectorIngressControlTable+"` WHERE Id = ?",
		rowID,
	).Scan(&packInstallID, &epoch, &channel, &mode); err != nil {
		t.Fatalf("read normalized namespace fence: %v", err)
	}
	if packInstallID != "" || epoch != 0 || channel != "*" || mode != "paused" {
		t.Fatalf("normalized namespace fence = packInstallId %q epoch %d channel %q mode %q, want empty/0/*/paused", packInstallID, epoch, channel, mode)
	}
}

// TestConnectorIngressFenceWarmUpgradeRejectsAmbiguousGlobalScope proves that
// a deployment never chooses an arbitrary row when a legacy NULL global fence
// and the canonical empty-string global fence coexist.  TiDB considers them
// distinct under the physical unique key, but the runtime treats both as the
// same logical scope.  Reconciliation must therefore fail closed before it
// can make the column non-null, leaving the ambiguous legacy row untouched.
func TestConnectorIngressFenceWarmUpgradeRejectsAmbiguousGlobalScope(t *testing.T) {
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	db := conn.DB()

	if _, err := db.ExecContext(ctx, "ALTER TABLE `"+connectorIngressControlTable+"` MODIFY COLUMN `PackInstallId` VARCHAR(255) NULL DEFAULT NULL"); err != nil {
		t.Fatalf("prepare nullable legacy PackInstallId: %v", err)
	}

	const namespace = "connector-ambiguous-global-scope"
	if _, err := db.ExecContext(ctx, `
		INSERT INTO `+connectorIngressControlTable+`
			(Id, Namespace, PackInstallId, Channel, Mode, FenceEpoch, RetryAfterSeconds)
		VALUES (?, ?, NULL, ?, ?, ?, ?), (?, ?, ?, ?, ?, ?, ?)`,
		"ingress-control-legacy-null-global",
		namespace,
		"*",
		"paused",
		7,
		120,
		"ingress-control-canonical-global",
		namespace,
		"",
		"*",
		"active",
		11,
		120,
	); err != nil {
		t.Fatalf("insert ambiguous global namespace fences: %v", err)
	}

	err := bootstrapimpl.ReconcileTableColumns(db, []schema.TableDefinition{connectorIngressControlDefinition(t)})
	if err == nil {
		t.Fatal("reconcile ambiguous global scope = nil, want fail-closed duplicate-key error")
	}
	if !strings.Contains(err.Error(), "_System_ConnectorIngressControl.PackInstallId") {
		t.Fatalf("reconcile ambiguous global scope error = %q, want PackInstallId context", err)
	}

	var legacyPackInstallID sql.NullString
	if err := db.QueryRowContext(ctx,
		"SELECT PackInstallId FROM `"+connectorIngressControlTable+"` WHERE Id = ?",
		"ingress-control-legacy-null-global",
	).Scan(&legacyPackInstallID); err != nil {
		t.Fatalf("read rejected legacy global fence: %v", err)
	}
	if legacyPackInstallID.Valid {
		t.Fatalf("ambiguous legacy global fence was mutated to PackInstallId %q, want NULL retained for operator resolution", legacyPackInstallID.String)
	}
}

func connectorIngressControlDefinition(t *testing.T) schema.TableDefinition {
	t.Helper()

	for _, definition := range bootstrapimpl.GetManagedTableDefinitions() {
		if definition.TableName == connectorIngressControlTable {
			return definition
		}
	}
	t.Fatalf("managed definition for %s is missing", connectorIngressControlTable)
	return schema.TableDefinition{}
}

func assertConnectorIngressFenceEpochMissing(t *testing.T, db *sql.DB) {
	t.Helper()

	var count int
	if err := db.QueryRow(
		`SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS
		 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
		connectorIngressControlTable,
		"FenceEpoch",
	).Scan(&count); err != nil {
		t.Fatalf("query missing ingress fence epoch: %v", err)
	}
	if count != 0 {
		t.Fatalf("%s.FenceEpoch count = %d, want 0 in supported prior schema", connectorIngressControlTable, count)
	}
}

func assertConnectorIngressFenceEpochReady(t *testing.T, db *sql.DB) {
	t.Helper()

	var columnType, nullable, defaultValue string
	if err := db.QueryRow(
		`SELECT COLUMN_TYPE, IS_NULLABLE, COALESCE(COLUMN_DEFAULT, '')
		 FROM INFORMATION_SCHEMA.COLUMNS
		 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
		connectorIngressControlTable,
		"FenceEpoch",
	).Scan(&columnType, &nullable, &defaultValue); err != nil {
		t.Fatalf("query upgraded ingress fence epoch: %v", err)
	}
	if !strings.HasPrefix(strings.ToLower(columnType), "bigint") || nullable != "NO" || defaultValue != "0" {
		t.Fatalf("%s.FenceEpoch = type %q nullable %q default %q, want non-null BIGINT default 0", connectorIngressControlTable, columnType, nullable, defaultValue)
	}
}

func assertConnectorIngressControlScopeReady(t *testing.T, db *sql.DB) {
	t.Helper()

	var columnType, nullable string
	if err := db.QueryRow(
		`SELECT COLUMN_TYPE, IS_NULLABLE
		 FROM INFORMATION_SCHEMA.COLUMNS
		 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
		connectorIngressControlTable,
		"PackInstallId",
	).Scan(&columnType, &nullable); err != nil {
		t.Fatalf("query normalized ingress-control scope: %v", err)
	}
	if !strings.HasPrefix(strings.ToLower(columnType), "varchar") || nullable != "NO" {
		t.Fatalf("%s.PackInstallId = type %q nullable %q, want non-null VARCHAR with empty global scope", connectorIngressControlTable, columnType, nullable)
	}
}