package catalogassignmentidentity

import (
	"fmt"
	"regexp"
	"strings"
	"testing"

	catalogassignmentidentityv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogassignmentidentityv1"
	catalogpublicationv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogpublicationv1"
	"github.com/pingcap/tidb/pkg/parser"
	_ "github.com/pingcap/tidb/pkg/parser/test_driver"
)

func TestSchemaParsesWithTiDB(t *testing.T) {
	sqlParser := parser.New()
	for _, statement := range SchemaStatementsForTest() {
		if _, _, err := sqlParser.Parse(statement, "", ""); err != nil {
			t.Fatalf("assignment-identity schema does not parse: %v\n%s", err, statement)
		}
	}
}

func TestSchemaOwnsExactlyOneEnvelopeAndFourIdentityDimensions(t *testing.T) {
	statements := SchemaStatementsForTest()
	if len(statements) != 5 {
		t.Fatalf("schema has %d statements, want exactly five immutable artifact tables", len(statements))
	}

	tables := []string{
		"_System_CatalogAssignmentIdentityView",
		"_System_CatalogAssignmentIdentityViewObject",
		"_System_CatalogAssignmentIdentityViewRecordType",
		"_System_CatalogAssignmentIdentityViewProductApp",
		"_System_CatalogAssignmentIdentityViewNavigationApp",
	}
	joined := strings.Join(statements, "\n")
	for index, table := range tables {
		create := "CREATE TABLE IF NOT EXISTS " + table + " "
		if count := strings.Count(joined, create); count != 1 {
			t.Fatalf("schema creates %s %d times, want exactly once", table, count)
		}
		if !strings.HasPrefix(
			normalizeAssignmentIdentitySQL(statements[index]),
			"CREATE TABLE IF NOT EXISTS "+table+" (",
		) {
			t.Fatalf("schema statement %d does not create %s", index, table)
		}
	}
	if count := strings.Count(
		joined,
		"CREATE TABLE IF NOT EXISTS _System_CatalogAssignmentIdentity",
	); count != len(tables) {
		t.Fatalf("schema creates %d assignment-identity tables, want %d", count, len(tables))
	}
}

func TestEnvelopeSchemaPreservesTheCompleteCanonicalView(t *testing.T) {
	envelope := normalizeAssignmentIdentitySQL(SchemaStatementsForTest()[0])
	requireAssignmentIdentitySQLFragments(t, envelope, []string{
		fmt.Sprintf("OrgId VARBINARY(%d) NOT NULL", catalogpublicationv1.MaxOrgIDBytes),
		"ViewDigest VARBINARY(71) NOT NULL",
		"CatalogRevision DECIMAL(20,0) NOT NULL",
		"CatalogSnapshotDigest VARBINARY(71) NOT NULL",
		"SchemaGeneration DECIMAL(20,0) NOT NULL",
		"EngineGeneration DECIMAL(20,0) NOT NULL",
		"CanonicalizationVersion VARBINARY(64) NOT NULL",
		"KeyGrammarVersion VARBINARY(64) NOT NULL",
		"ObjectKeyNormalizationVersion VARBINARY(64) NOT NULL",
		"CanonicalBytes LONGBLOB NOT NULL",
		"CanonicalByteLength DECIMAL(20,0) NOT NULL",
		"ObjectCount DECIMAL(20,0) NOT NULL",
		"RecordTypeCount DECIMAL(20,0) NOT NULL",
		"ProductAppCount DECIMAL(20,0) NOT NULL",
		"NavigationAppCount DECIMAL(20,0) NOT NULL",
		"CreatedDate DATETIME(6) NOT NULL",
		"PRIMARY KEY (OrgId, ViewDigest)",
		"OCTET_LENGTH(ViewDigest) = 71",
		"OCTET_LENGTH(CatalogSnapshotDigest) = 71",
		"CatalogRevision > 0",
		"SchemaGeneration > 0",
		"EngineGeneration > 0",
		"CanonicalizationVersion = '" + catalogassignmentidentityv1.CanonicalizationVersion + "'",
		"KeyGrammarVersion = '" + catalogassignmentidentityv1.KeyGrammarVersion + "'",
		"ObjectKeyNormalizationVersion = '" + catalogassignmentidentityv1.ObjectKeyNormalizationVersion + "'",
		fmt.Sprintf("ObjectCount <= %d", catalogassignmentidentityv1.MaxEntriesPerDimension),
		"ObjectCount >= 0",
		fmt.Sprintf("RecordTypeCount <= %d", catalogassignmentidentityv1.MaxEntriesPerDimension),
		"RecordTypeCount >= 0",
		fmt.Sprintf("ProductAppCount <= %d", catalogassignmentidentityv1.MaxEntriesPerDimension),
		"ProductAppCount >= 0",
		fmt.Sprintf("NavigationAppCount <= %d", catalogassignmentidentityv1.MaxEntriesPerDimension),
		"NavigationAppCount >= 0",
		fmt.Sprintf("CanonicalByteLength <= %d", catalogassignmentidentityv1.MaxCanonicalViewBytes),
		"CanonicalByteLength = OCTET_LENGTH(CanonicalBytes)",
	})
}

func TestDimensionSchemasPreserveCanonicalKeysTargetsStatesAndOrder(t *testing.T) {
	statements := SchemaStatementsForTest()
	tests := []struct {
		name      string
		statement string
		required  []string
	}{
		{
			name:      "objects",
			statement: statements[1],
			required: []string{
				fmt.Sprintf(
					"NormalizedAPIName VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxObjectAPINameBytes,
				),
				fmt.Sprintf(
					"ResolvedObjectId VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxTargetLocalIDBytes,
				),
				"UNIQUE KEY ux_catalog_assignment_identity_object_key (OrgId, ViewDigest, NormalizedAPIName)",
				"UNIQUE KEY ux_catalog_assignment_identity_object_target (OrgId, ViewDigest, ResolvedObjectId)",
			},
		},
		{
			name:      "record types",
			statement: statements[2],
			required: []string{
				fmt.Sprintf(
					"ObjectNormalizedAPIName VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxObjectAPINameBytes,
				),
				fmt.Sprintf(
					"NamespaceId VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxKeyPartBytes,
				),
				fmt.Sprintf(
					"LocalKey VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxKeyPartBytes,
				),
				fmt.Sprintf(
					"ResolvedRecordTypeId VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxTargetLocalIDBytes,
				),
				"UNIQUE KEY ux_catalog_assignment_identity_record_type_key (OrgId, ViewDigest, ObjectNormalizedAPIName, NamespaceId, LocalKey)",
				"UNIQUE KEY ux_catalog_assignment_identity_record_type_target (OrgId, ViewDigest, ResolvedRecordTypeId)",
			},
		},
		{
			name:      "Product Apps",
			statement: statements[3],
			required: []string{
				fmt.Sprintf(
					"AppId VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxKeyPartBytes,
				),
				"ResolvedTargetKind TINYINT UNSIGNED NOT NULL",
				"ResolvedAppInstallationId VARBINARY(36) NULL",
				"UNIQUE KEY ux_catalog_assignment_identity_product_app_key (OrgId, ViewDigest, AppId)",
				"UNIQUE KEY ux_catalog_assignment_identity_product_app_target (OrgId, ViewDigest, ResolvedAppInstallationId)",
				"ResolvedTargetKind = 0",
				"ResolvedTargetKind = 1",
				"ResolvedAppInstallationId IS NULL",
				"ResolvedAppInstallationId IS NOT NULL",
			},
		},
		{
			name:      "navigation Apps",
			statement: statements[4],
			required: []string{
				fmt.Sprintf(
					"NamespaceId VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxKeyPartBytes,
				),
				fmt.Sprintf(
					"LocalKey VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxKeyPartBytes,
				),
				fmt.Sprintf(
					"ResolvedAppConfigId VARBINARY(%d) NOT NULL",
					catalogassignmentidentityv1.MaxTargetLocalIDBytes,
				),
				"UNIQUE KEY ux_catalog_assignment_identity_navigation_app_key (OrgId, ViewDigest, NamespaceId, LocalKey)",
				"UNIQUE KEY ux_catalog_assignment_identity_navigation_app_target (OrgId, ViewDigest, ResolvedAppConfigId)",
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			statement := normalizeAssignmentIdentitySQL(test.statement)
			requireAssignmentIdentitySQLFragments(t, statement, append([]string{
				fmt.Sprintf("OrgId VARBINARY(%d) NOT NULL", catalogpublicationv1.MaxOrgIDBytes),
				"ViewDigest VARBINARY(71) NOT NULL",
				"EntryOrdinal DECIMAL(20,0) NOT NULL",
				"State VARBINARY(16) NOT NULL",
				"PRIMARY KEY (OrgId, ViewDigest, EntryOrdinal)",
				"OCTET_LENGTH(ViewDigest) = 71",
				"EntryOrdinal >= 0",
				fmt.Sprintf("EntryOrdinal < %d", catalogassignmentidentityv1.MaxEntriesPerDimension),
				"State IN ('ACTIVE','TOMBSTONE')",
			}, test.required...))
			for _, forbidden := range []string{"CreatedDate", "LastModifiedDate", "FOREIGN KEY"} {
				if strings.Contains(statement, forbidden) {
					t.Fatalf("%s dimension contains forbidden mutable/coupling field %q", test.name, forbidden)
				}
			}
		})
	}
}

func TestSchemaCannotSelectOrMutateAnArtifact(t *testing.T) {
	joined := normalizeAssignmentIdentitySQL(strings.Join(SchemaStatementsForTest(), "\n"))
	for _, forbidden := range []string{
		" ALTER TABLE ",
		" DELETE FROM ",
		" DROP TABLE ",
		" INSERT INTO ",
		" ON UPDATE ",
		" REPLACE INTO ",
		" UPDATE ",
		"Current",
		"Latest",
		"Pointer",
		"Selected",
		"Selection",
	} {
		if strings.Contains(" "+joined+" ", forbidden) {
			t.Fatalf("immutable artifact schema contains forbidden selector or mutation vocabulary %q", forbidden)
		}
	}

	sourceBindingUnique := regexp.MustCompile(
		`(?i)UNIQUE\s+(?:KEY|INDEX)[^(]*\([^)]*(?:CatalogRevision|CatalogSnapshotDigest|SchemaGeneration|EngineGeneration)`,
	)
	if sourceBindingUnique.MatchString(joined) {
		t.Fatal("source Catalog binding must not be a uniqueness selector for an unselected content artifact")
	}
}

func TestSchemaStatementsForTestReturnsDefensiveCopy(t *testing.T) {
	first := SchemaStatementsForTest()
	first[0] = "changed"
	if SchemaStatementsForTest()[0] == "changed" {
		t.Fatal("SchemaStatementsForTest returned shared storage")
	}
}

func normalizeAssignmentIdentitySQL(statement string) string {
	return strings.Join(strings.Fields(statement), " ")
}

func requireAssignmentIdentitySQLFragments(t *testing.T, statement string, required []string) {
	t.Helper()
	for _, fragment := range required {
		if !strings.Contains(statement, fragment) {
			t.Fatalf("assignment-identity schema is missing %q\n%s", fragment, statement)
		}
	}
}