package serving

import (
	"database/sql"
	"errors"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
)

const servingSchemaFixtureDatabase = "nexuscrm_serving_test"

type servingSchemaFixture struct {
	tables []servingTableSchemaFixture
}

type servingTableSchemaFixture struct {
	name        string
	columns     []servingColumnFixture
	indexes     []servingIndexFixture
	foreignKeys []servingForeignKeyFixture
}

type servingColumnFixture struct {
	expectedServingColumn
	extra string
}

type servingIndexFixture struct {
	name      string
	unique    bool
	columns   []string
	prefixAt  int
	sequence0 int
}

type servingForeignKeyFixture struct {
	constraint       string
	column           string
	referencedSchema string
	referencedTable  string
	referencedColumn string
	currentSchema    string
	updateRule       string
	deleteRule       string
}

func TestVerifyServingSchemaAcceptsExactCatalogShape(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)

	if err := verifyServingSchema(t.Context(), db); err != nil {
		t.Fatalf("verifyServingSchema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyServingSchemaAcceptsTiDBIntegerDisplayWidthAliases(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	fixture := exactServingSchemaFixture()
	for _, table := range fixture.tables {
		for index := range table.columns {
			switch table.columns[index].columnType {
			case "int(11)":
				table.columns[index].columnType = "int"
			case "bigint(20)":
				table.columns[index].columnType = "bigint"
			}
		}
	}
	expectServingSchema(t, mock, fixture, len(servingTableSchemas)*3)

	if err := verifyServingSchema(t.Context(), db); err != nil {
		t.Fatalf("verifyServingSchema with TiDB integer aliases: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyServingSchemaRejectsColumnDrift(t *testing.T) {
	tests := []struct {
		name         string
		throughTable string
		mutate       func(*servingSchemaFixture)
		want         string
	}{
		{
			name: "missing column",
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochTable)
				table.columns = table.columns[:len(table.columns)-1]
			},
			want: "missing column LastModifiedDate",
		},
		{
			name: "extra column",
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochTable)
				table.columns = append(table.columns, servingColumnFixture{
					expectedServingColumn: expectedServingColumn{
						name: "LegacyCatalogView", columnType: "mediumblob",
					},
				})
			},
			want: "unexpected column LegacyCatalogView",
		},
		{
			name: "widened type",
			mutate: func(fixture *servingSchemaFixture) {
				fixture.table(servingEpochTable).columns[0].columnType = "varchar(512)"
			},
			want: "want type=varchar(255)",
		},
		{
			name: "unsafe collation",
			mutate: func(fixture *servingSchemaFixture) {
				fixture.table(servingEpochTable).columns[0].collation = "utf8mb4_general_ci"
			},
			want: "want type=varchar(255) nullable=NO collation=utf8mb4_bin",
		},
		{
			name: "weakened nullability",
			mutate: func(fixture *servingSchemaFixture) {
				fixture.table(servingEpochTable).columns[1].nullable = true
			},
			want: "want type=char(71) nullable=NO",
		},
		{
			name: "missing timestamp default",
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochTable)
				table.columns[len(table.columns)-2].hasDefault = false
			},
			want: "CreatedDate is type=datetime(6)",
		},
		{
			name: "wrong timestamp default",
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochTable)
				table.columns[len(table.columns)-1].columnDefault = "CURRENT_TIMESTAMP"
			},
			want: "default=CURRENT_TIMESTAMP, want type=datetime(6) nullable=NO collation=NULL default=CURRENT_TIMESTAMP(6)",
		},
		{
			name: "wrong state default",
			mutate: func(fixture *servingSchemaFixture) {
				fixture.table(servingEpochTable).columns[4].columnDefault = "verified"
			},
			want: "State is type=varchar(32)",
		},
		{
			name: "unexpected nullable empty default",
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochTable)
				table.columns[3].hasDefault = true
				table.columns[3].columnDefault = ""
			},
			want: "ActiveScopeDigest is type=char(64)",
		},
		{
			name: "on update timestamp",
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochTable)
				table.columns[len(table.columns)-1].extra = "DEFAULT_GENERATED on update CURRENT_TIMESTAMP(6)"
			},
			want: "extra=\"DEFAULT_GENERATED on update CURRENT_TIMESTAMP(6)\"",
		},
		{
			name:         "binding owner pack coordinate must inherit parent collation",
			throughTable: servingEpochBindingOwnerTable,
			mutate: func(fixture *servingSchemaFixture) {
				fixture.table(servingEpochBindingOwnerTable).columns[1].collation = "utf8mb4_bin"
			},
			want: "PackGenerationId is type=varchar(255) nullable=NO collation=utf8mb4_bin",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock := newServingSchemaMock(t)
			fixture := exactServingSchemaFixture()
			test.mutate(&fixture)
			throughTable := test.throughTable
			if throughTable == "" {
				throughTable = servingEpochTable
			}
			for _, table := range fixture.tables {
				expectServingColumns(mock, table.name, table.columns)
				if table.name == throughTable {
					break
				}
			}

			err := verifyServingSchema(t.Context(), db)
			requireSchemaUnavailable(t, err, test.want)
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestVerifyServingSchemaAcceptsBenignTimestampDefaultGeneratedExtra(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	fixture := exactServingSchemaFixture()
	for index := range fixture.tables {
		columns := fixture.tables[index].columns
		columns[len(columns)-2].extra = "DEFAULT_GENERATED"
		columns[len(columns)-1].extra = "default_generated"
	}
	expectServingSchema(t, mock, fixture, len(servingTableSchemas)*3)

	if err := verifyServingSchema(t.Context(), db); err != nil {
		t.Fatalf("verifyServingSchema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyServingSchemaRejectsUnsafeRequiredIndexes(t *testing.T) {
	tests := []struct {
		name         string
		throughTable string
		mutate       func(*servingSchemaFixture)
		want         string
	}{
		{
			name: "missing uniqueness",
			mutate: func(fixture *servingSchemaFixture) {
				fixture.table(servingEpochTable).indexes[1].unique = false
			},
			want: "ux_connector_serving_epoch_digest",
		},
		{
			name: "prefix index",
			mutate: func(fixture *servingSchemaFixture) {
				fixture.table(servingEpochTable).indexes[1].prefixAt = 1
			},
			want: "unsafe sequence, column, uniqueness, or prefix",
		},
		{
			name: "wrong order",
			mutate: func(fixture *servingSchemaFixture) {
				indexes := fixture.table(servingEpochTable).indexes
				indexes[3].columns[0], indexes[3].columns[1] =
					indexes[3].columns[1], indexes[3].columns[0]
			},
			want: "idx_connector_serving_epoch_scope_state",
		},
		{
			name: "unexpected unique index",
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochTable)
				table.indexes = append(table.indexes, servingIndexFixture{
					name:    "ux_unreviewed_org",
					unique:  true,
					columns: []string{"OrgId"},
				})
			},
			want: "unexpected unique index ux_unreviewed_org",
		},
		{
			name: "unexpected nonunique index",
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochTable)
				table.indexes = append(table.indexes, servingIndexFixture{
					name:    "idx_unreviewed_org",
					columns: []string{"OrgId"},
				})
			},
			want: "unexpected index idx_unreviewed_org",
		},
		{
			name:         "missing credential proof composite primary key",
			throughTable: servingEpochCredentialGenerationTable,
			mutate: func(fixture *servingSchemaFixture) {
				table := fixture.table(servingEpochCredentialGenerationTable)
				table.indexes = table.indexes[1:]
			},
			want: "index PRIMARY",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock := newServingSchemaMock(t)
			fixture := exactServingSchemaFixture()
			test.mutate(&fixture)
			expectServingAllColumns(mock, fixture)
			throughTable := test.throughTable
			if throughTable == "" {
				throughTable = servingEpochTable
			}
			for _, table := range fixture.tables {
				expectServingIndexes(mock, table.name, table.indexes)
				if table.name == throughTable {
					break
				}
			}

			err := verifyServingSchema(t.Context(), db)
			requireSchemaUnavailable(t, err, test.want)
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestVerifyServingSchemaRejectsWrongForeignKeySet(t *testing.T) {
	tests := []struct {
		name   string
		table  string
		mutate func(*servingTableSchemaFixture)
		want   string
	}{
		{
			name:  "cascading delete",
			table: servingEpochBindingTable,
			mutate: func(table *servingTableSchemaFixture) {
				table.foreignKeys[0].deleteRule = "CASCADE"
			},
			want: "delete=CASCADE",
		},
		{
			name:  "cross-database binding owner parent",
			table: servingEpochBindingOwnerTable,
			mutate: func(table *servingTableSchemaFixture) {
				table.foreignKeys[1].referencedSchema = "foreign_serving_database"
			},
			want: "foreign_serving_database._System_PackGenerationConnectorBindingDefinitionSet.PackGenerationId",
		},
		{
			name:  "missing reproducibility foreign key",
			table: servingEpochReproducibilityTable,
			mutate: func(table *servingTableSchemaFixture) {
				table.foreignKeys = nil
			},
			want: "missing foreign key on column ServingEpochId",
		},
		{
			name:  "extra credential proof foreign key",
			table: servingEpochCredentialGenerationTable,
			mutate: func(table *servingTableSchemaFixture) {
				table.foreignKeys = append(table.foreignKeys, servingForeignKeyFixture{
					constraint:       "fk_unexpected",
					column:           "ReceiptProtocolVersion",
					referencedSchema: servingSchemaFixtureDatabase,
					referencedTable:  "_System_ConnectorServingEpoch",
					referencedColumn: "Id",
					currentSchema:    servingSchemaFixtureDatabase,
					updateRule:       "RESTRICT",
					deleteRule:       "RESTRICT",
				})
			},
			want: "foreign key fk_unexpected",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock := newServingSchemaMock(t)
			fixture := exactServingSchemaFixture()
			test.mutate(fixture.table(test.table))
			expectServingAllColumns(mock, fixture)
			expectServingAllIndexes(mock, fixture)
			for _, table := range fixture.tables {
				expectServingForeignKeys(mock, table.name, table.foreignKeys)
				if table.name == test.table {
					break
				}
			}

			err := verifyServingSchema(t.Context(), db)
			requireSchemaUnavailable(t, err, test.want)
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestVerifyServingSchemaWrapsInspectionFailures(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	mock.ExpectQuery(regexp.QuoteMeta(inspectServingColumnsSQL)).
		WithArgs(servingEpochTable).
		WillReturnError(errors.New("information schema unavailable"))

	err := verifyServingSchema(t.Context(), db)
	requireSchemaUnavailable(t, err, "information schema unavailable")
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyServingSchemaRejectsMissingDependencies(t *testing.T) {
	for _, test := range []struct {
		name string
		ctx  bool
		db   bool
	}{
		{name: "nil context", db: true},
		{name: "nil database", ctx: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			var db *sql.DB
			if test.db {
				opened, _ := newServingSchemaMock(t)
				db = opened
			}
			var err error
			if test.ctx {
				err = verifyServingSchema(t.Context(), db)
			} else {
				err = verifyServingSchema(nil, db)
			}
			requireSchemaUnavailable(t, err, "database or context is unavailable")
		})
	}
}

func newServingSchemaMock(t *testing.T) (*sql.DB, sqlmock.Sqlmock) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	return db, mock
}

func exactServingSchemaFixture() servingSchemaFixture {
	fixture := servingSchemaFixture{
		tables: make([]servingTableSchemaFixture, len(servingTableSchemas)),
	}
	for tableIndex, table := range servingTableSchemas {
		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 (fixture *servingSchemaFixture) table(name string) *servingTableSchemaFixture {
	for index := range fixture.tables {
		if fixture.tables[index].name == name {
			return &fixture.tables[index]
		}
	}
	panic("serving schema fixture has no table " + name)
}

func servingColumnFixtures(columns []expectedServingColumn) []servingColumnFixture {
	result := make([]servingColumnFixture, len(columns))
	for i, column := range columns {
		result[i] = servingColumnFixture{expectedServingColumn: column}
	}
	return result
}

func servingIndexFixtures(indexes []expectedServingIndex) []servingIndexFixture {
	result := make([]servingIndexFixture, len(indexes))
	for i, index := range indexes {
		result[i] = servingIndexFixture{
			name:    index.name,
			unique:  index.unique,
			columns: append([]string(nil), index.columns...),
		}
	}
	return result
}

func expectServingSchema(
	t *testing.T,
	mock sqlmock.Sqlmock,
	fixture servingSchemaFixture,
	_ int,
) {
	t.Helper()
	expectServingAllColumns(mock, fixture)
	expectServingAllIndexes(mock, fixture)
	for _, table := range fixture.tables {
		expectServingForeignKeys(mock, table.name, table.foreignKeys)
	}
}

func expectServingAllColumns(mock sqlmock.Sqlmock, fixture servingSchemaFixture) {
	for _, table := range fixture.tables {
		expectServingColumns(mock, table.name, table.columns)
	}
}

func expectServingAllIndexes(mock sqlmock.Sqlmock, fixture servingSchemaFixture) {
	for _, table := range fixture.tables {
		expectServingIndexes(mock, table.name, table.indexes)
	}
}

func expectServingColumns(mock sqlmock.Sqlmock, table string, columns []servingColumnFixture) {
	rows := sqlmock.NewRows([]string{
		"COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLLATION_NAME", "COLUMN_DEFAULT", "EXTRA",
	})
	for _, column := range columns {
		nullable := "NO"
		if column.nullable {
			nullable = "YES"
		}
		var collation any
		if column.collation != "" {
			collation = column.collation
		}
		var columnDefault any
		if column.hasDefault {
			columnDefault = column.columnDefault
		}
		rows.AddRow(column.name, column.columnType, nullable, collation, columnDefault, column.extra)
	}
	mock.ExpectQuery(regexp.QuoteMeta(inspectServingColumnsSQL)).
		WithArgs(table).
		WillReturnRows(rows)
}

func expectServingIndexes(mock sqlmock.Sqlmock, table string, indexes []servingIndexFixture) {
	rows := sqlmock.NewRows([]string{"INDEX_NAME", "NON_UNIQUE", "SEQ_IN_INDEX", "COLUMN_NAME", "SUB_PART"})
	for _, index := range indexes {
		nonUnique := 1
		if index.unique {
			nonUnique = 0
		}
		for columnIndex, column := range index.columns {
			var subPart any
			if index.prefixAt == columnIndex+1 {
				subPart = 16
			}
			rows.AddRow(index.name, nonUnique, index.sequence0+columnIndex+1, column, subPart)
		}
	}
	mock.ExpectQuery(regexp.QuoteMeta(inspectServingIndexesSQL)).
		WithArgs(table).
		WillReturnRows(rows)
}

func expectServingForeignKeys(
	mock sqlmock.Sqlmock,
	table string,
	foreignKeys []servingForeignKeyFixture,
) {
	rows := sqlmock.NewRows([]string{
		"CONSTRAINT_NAME", "COLUMN_NAME", "REFERENCED_TABLE_SCHEMA",
		"REFERENCED_TABLE_NAME", "REFERENCED_COLUMN_NAME", "CURRENT_SCHEMA",
		"UPDATE_RULE", "DELETE_RULE",
	})
	for _, foreignKey := range foreignKeys {
		rows.AddRow(
			foreignKey.constraint,
			foreignKey.column,
			foreignKey.referencedSchema,
			foreignKey.referencedTable,
			foreignKey.referencedColumn,
			foreignKey.currentSchema,
			foreignKey.updateRule,
			foreignKey.deleteRule,
		)
	}
	mock.ExpectQuery(regexp.QuoteMeta(inspectServingForeignKeysSQL)).
		WithArgs(table).
		WillReturnRows(rows)
}

func requireSchemaUnavailable(t *testing.T, err error, detail string) {
	t.Helper()
	if !errors.Is(err, ErrSchemaUnavailable) {
		t.Fatalf("error = %v, want ErrSchemaUnavailable", err)
	}
	if !strings.Contains(err.Error(), detail) {
		t.Fatalf("error = %v, want detail %q", err, detail)
	}
}