package serving

import (
	"errors"
	"slices"
	"strings"
	"testing"
)

func TestVerifyIngressReceiptAuthoritySchemaAcceptsExactCatalogShape(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	expectServingSchema(t, mock, exactIngressReceiptSchemaFixture(), len(ingressReceiptAuthoritySchemas)*3)
	if err := verifyIngressReceiptAuthoritySchema(t.Context(), db); err != nil {
		t.Fatalf("verify exact ingress receipt authority schema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestIngressReceiptTransportIdentityIndexExcludesServingScope(t *testing.T) {
	for _, index := range v2ReceiptAdmissionKeySchema.indexes {
		if index.name != "ux_connector_v2_receipt_key_transport" {
			continue
		}
		want := []string{"ConnectorAccountId", "IngressChannel", "TransportIdempotencyKey"}
		if !index.unique || len(index.columns) != len(want) {
			t.Fatalf("transport identity index = %#v", index)
		}
		for position := range want {
			if index.columns[position] != want[position] {
				t.Fatalf("transport identity index columns = %v, want %v", index.columns, want)
			}
		}
		return
	}
	t.Fatal("transport identity index is absent")
}

func TestIngressReceiptCompletenessIndexLeadsWithConnectorAccount(t *testing.T) {
	for _, index := range v2ReceiptSchema.indexes {
		if index.name != "idx_connector_v2_receipt_account" {
			continue
		}
		if index.unique || !slices.Equal(index.columns, []string{"ConnectorAccountId", "Id"}) {
			t.Fatalf("receipt account index = unique:%t columns:%v", index.unique, index.columns)
		}
		return
	}
	t.Fatal("receipt schema is missing the account-leading completeness index")
}

func TestIngressReceiptProtocolVersionUsesClosedWidth(t *testing.T) {
	for _, column := range v2ReceiptSchema.columns {
		if column.name != "ReceiptProtocolVersion" {
			continue
		}
		if column.columnType != "varchar(32)" || column.nullable || column.collation != "utf8mb4_bin" {
			t.Fatalf("receipt protocol column = %#v", column)
		}
		return
	}
	t.Fatal("receipt schema is missing ReceiptProtocolVersion")
}

func TestIngressReceiptComponentSchemaRequiresCompleteOrderedChildren(t *testing.T) {
	if !slices.Equal(
		v2ReceiptComponentSchema.indexes[1].columns,
		[]string{"ReceiptId", "ComponentOrdinal"},
	) || !v2ReceiptComponentSchema.indexes[1].unique {
		t.Fatalf("component ordinal index = %#v", v2ReceiptComponentSchema.indexes[1])
	}
	if len(v2ReceiptComponentSchema.foreignKeys) != 1 ||
		v2ReceiptComponentSchema.foreignKeys[0].column != "ReceiptId" ||
		v2ReceiptComponentSchema.foreignKeys[0].referencedTable != v2ReceiptTable ||
		v2ReceiptComponentSchema.foreignKeys[0].referencedColumn != "Id" {
		t.Fatalf("component parent foreign key = %#v", v2ReceiptComponentSchema.foreignKeys)
	}
}

func TestVerifyIngressReceiptAuthoritySchemaRejectsReceiptGraphDrift(t *testing.T) {
	t.Run("missing receipt ciphertext column", func(t *testing.T) {
		db, mock := newServingSchemaMock(t)
		fixture := exactIngressReceiptSchemaFixture()
		table := fixture.table(v2ReceiptTable)
		table.columns = append(table.columns[:11], table.columns[12:]...)
		for _, current := range fixture.tables {
			expectServingColumns(mock, current.name, current.columns)
			if current.name == v2ReceiptTable {
				break
			}
		}
		err := verifyIngressReceiptAuthoritySchema(t.Context(), db)
		requireIngressReceiptSchemaUnavailable(t, err)
	})

	t.Run("wide receipt protocol column", func(t *testing.T) {
		db, mock := newServingSchemaMock(t)
		fixture := exactIngressReceiptSchemaFixture()
		table := fixture.table(v2ReceiptTable)
		table.columns[1].columnType = "varchar(255)"
		for _, current := range fixture.tables {
			expectServingColumns(mock, current.name, current.columns)
			if current.name == v2ReceiptTable {
				break
			}
		}
		err := verifyIngressReceiptAuthoritySchema(t.Context(), db)
		requireIngressReceiptSchemaUnavailable(t, err)
	})

	t.Run("prefix transport uniqueness", func(t *testing.T) {
		db, mock := newServingSchemaMock(t)
		fixture := exactIngressReceiptSchemaFixture()
		fixture.table(v2ReceiptAdmissionKeyTable).indexes[1].prefixAt = 3
		expectServingAllColumns(mock, fixture)
		for _, current := range fixture.tables {
			expectServingIndexes(mock, current.name, current.indexes)
			if current.name == v2ReceiptAdmissionKeyTable {
				break
			}
		}
		err := verifyIngressReceiptAuthoritySchema(t.Context(), db)
		requireIngressReceiptSchemaUnavailable(t, err)
		if !strings.Contains(err.Error(), "ux_connector_v2_receipt_key_transport") {
			t.Fatalf("prefix drift error = %v, want exact transport index", err)
		}
	})

	t.Run("missing component payload digest", func(t *testing.T) {
		db, mock := newServingSchemaMock(t)
		fixture := exactIngressReceiptSchemaFixture()
		table := fixture.table(v2ReceiptComponentTable)
		table.columns = append(table.columns[:6], table.columns[7:]...)
		for _, current := range fixture.tables {
			expectServingColumns(mock, current.name, current.columns)
			if current.name == v2ReceiptComponentTable {
				break
			}
		}
		err := verifyIngressReceiptAuthoritySchema(t.Context(), db)
		requireIngressReceiptSchemaUnavailable(t, err)
	})

	t.Run("prefix component ordinal uniqueness", func(t *testing.T) {
		db, mock := newServingSchemaMock(t)
		fixture := exactIngressReceiptSchemaFixture()
		fixture.table(v2ReceiptComponentTable).indexes[1].prefixAt = 1
		expectServingAllColumns(mock, fixture)
		for _, current := range fixture.tables {
			expectServingIndexes(mock, current.name, current.indexes)
			if current.name == v2ReceiptComponentTable {
				break
			}
		}
		err := verifyIngressReceiptAuthoritySchema(t.Context(), db)
		requireIngressReceiptSchemaUnavailable(t, err)
		if !strings.Contains(err.Error(), "ux_connector_v2_receipt_component_ordinal") {
			t.Fatalf("prefix drift error = %v, want exact component ordinal index", err)
		}
	})

	t.Run("receipt FK cascade", func(t *testing.T) {
		db, mock := newServingSchemaMock(t)
		fixture := exactIngressReceiptSchemaFixture()
		fixture.table(v2ReceiptTable).foreignKeys[0].deleteRule = "CASCADE"
		expectServingAllColumns(mock, fixture)
		expectServingAllIndexes(mock, fixture)
		for _, current := range fixture.tables {
			expectServingForeignKeys(mock, current.name, current.foreignKeys)
			if current.name == v2ReceiptTable {
				break
			}
		}
		err := verifyIngressReceiptAuthoritySchema(t.Context(), db)
		requireIngressReceiptSchemaUnavailable(t, err)
	})

	t.Run("component FK cascade", func(t *testing.T) {
		db, mock := newServingSchemaMock(t)
		fixture := exactIngressReceiptSchemaFixture()
		fixture.table(v2ReceiptComponentTable).foreignKeys[0].deleteRule = "CASCADE"
		expectServingAllColumns(mock, fixture)
		expectServingAllIndexes(mock, fixture)
		for _, current := range fixture.tables {
			expectServingForeignKeys(mock, current.name, current.foreignKeys)
			if current.name == v2ReceiptComponentTable {
				break
			}
		}
		err := verifyIngressReceiptAuthoritySchema(t.Context(), db)
		requireIngressReceiptSchemaUnavailable(t, err)
	})
}

func exactIngressReceiptSchemaFixture() servingSchemaFixture {
	fixture := servingSchemaFixture{tables: make([]servingTableSchemaFixture, len(ingressReceiptAuthoritySchemas))}
	for tableIndex, table := range ingressReceiptAuthoritySchemas {
		foreignKeys := make([]servingForeignKeyFixture, len(table.foreignKeys))
		for foreignKeyIndex, foreignKey := range table.foreignKeys {
			foreignKeys[foreignKeyIndex] = servingForeignKeyFixture{
				constraint:       "fk_" + table.name + "_" + foreignKey.column,
				column:           foreignKey.column,
				referencedSchema: servingSchemaFixtureDatabase,
				referencedTable:  foreignKey.referencedTable,
				referencedColumn: foreignKey.referencedColumn,
				currentSchema:    servingSchemaFixtureDatabase,
				updateRule:       "RESTRICT",
				deleteRule:       "RESTRICT",
			}
		}
		fixture.tables[tableIndex] = servingTableSchemaFixture{
			name:        table.name,
			columns:     servingColumnFixtures(table.columns),
			indexes:     servingIndexFixtures(table.indexes),
			foreignKeys: foreignKeys,
		}
	}
	return fixture
}

func requireIngressReceiptSchemaUnavailable(t *testing.T, err error) {
	t.Helper()
	if !errors.Is(err, ErrSchemaUnavailable) || !errors.Is(err, ErrIngressReceiptSchemaUnavailable) {
		t.Fatalf("schema drift error = %v, want generic and ingress receipt schema sentinels", err)
	}
}