package catalogpublicationv2store

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

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

func TestCreateHaltedShadowCreatesOnlyTheFixedStrictV2Shape(t *testing.T) {
	db, mock, closeDB := newSchemaMock(t)
	defer closeDB()
	mock.ExpectExec(regexp.QuoteMeta(shadowTableDDL)).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectExactSchema(mock, shadowTableName)

	table, err := CreateHaltedShadow(context.Background(), db)
	if err != nil {
		t.Fatal(err)
	}
	if table == nil || table.db != db || table.identity != shadowTableIdentity {
		t.Fatalf("shadow handle = %#v", table)
	}
	assertSchemaExpectations(t, mock)
}

func TestVerifyLiveAcceptsOnlyTheFixedCanonicalIdentity(t *testing.T) {
	db, mock, closeDB := newSchemaMock(t)
	defer closeDB()
	expectExactSchema(mock, liveTableName)

	table, err := VerifyLive(context.Background(), db)
	if err != nil {
		t.Fatal(err)
	}
	if table == nil || table.db != db || table.identity != liveTableIdentity {
		t.Fatalf("live handle = %#v", table)
	}
	assertSchemaExpectations(t, mock)
}

func TestVerifyLiveRejectsEveryNonV2ColumnShape(t *testing.T) {
	tests := []struct {
		name    string
		columns []columnShape
	}{
		{
			name: "legacy BindingVersion column",
			columns: func() []columnShape {
				columns := append([]columnShape(nil), strictV2Columns...)
				columns = append(columns[:1], append([]columnShape{{
					name:       "BindingVersion",
					columnType: "varchar(64)",
					nullable:   "YES",
					charSet:    "ascii",
					collation:  "ascii_bin",
				}}, columns[1:]...)...)
				return columns
			}(),
		},
		{
			name: "nullable owner receipt digest",
			columns: func() []columnShape {
				columns := append([]columnShape(nil), strictV2Columns...)
				columns[7].nullable = "YES"
				return columns
			}(),
		},
		{
			name: "extra legacy column",
			columns: append(
				append([]columnShape(nil), strictV2Columns...),
				columnShape{name: "LegacyReceiptGate", columnType: "varchar(16)", nullable: "NO", charSet: "ascii", collation: "ascii_bin"},
			),
		},
	}

	for _, test := range tests {
		test := test
		t.Run(test.name, func(t *testing.T) {
			db, mock, closeDB := newSchemaMock(t)
			defer closeDB()
			expectColumns(mock, liveTableName, test.columns)

			table, err := VerifyLive(context.Background(), db)
			if !errors.Is(err, ErrSchema) || table != nil {
				t.Fatalf("VerifyLive = (%#v, %v), want (nil, ErrSchema)", table, err)
			}
			assertSchemaExpectations(t, mock)
		})
	}
}

func TestVerifyLiveRejectsWrongPrimaryOrAdditionalIndexes(t *testing.T) {
	tests := []struct {
		name    string
		indexes []indexColumnShape
	}{
		{
			name: "wrong primary",
			indexes: []indexColumnShape{
				{indexName: "PRIMARY", nonUnique: 0, sequence: 1, column: "OrgId"},
				{indexName: "idx_catalog_publication_org_lineage", nonUnique: 1, sequence: 1, column: "OrgId"},
				{indexName: "idx_catalog_publication_org_lineage", nonUnique: 1, sequence: 2, column: "CatalogRevision"},
				{indexName: "idx_catalog_publication_org_lineage", nonUnique: 1, sequence: 3, column: "SecurityEpoch"},
			},
		},
		{
			name: "additional index",
			indexes: append(
				append([]indexColumnShape(nil), strictV2Indexes...),
				indexColumnShape{indexName: "legacy_version", nonUnique: 1, sequence: 1, column: "OrgId"},
			),
		},
	}
	for _, test := range tests {
		test := test
		t.Run(test.name, func(t *testing.T) {
			db, mock, closeDB := newSchemaMock(t)
			defer closeDB()
			expectColumns(mock, liveTableName, strictV2Columns)
			expectIndexes(mock, liveTableName, test.indexes)

			table, err := VerifyLive(context.Background(), db)
			if !errors.Is(err, ErrSchema) || table != nil {
				t.Fatalf("VerifyLive = (%#v, %v), want (nil, ErrSchema)", table, err)
			}
			assertSchemaExpectations(t, mock)
		})
	}
}

func TestStrictV2DDLHasNoVersionSwitchDefaultOrNullableReceipt(t *testing.T) {
	for _, forbidden := range []string{
		"BindingVersion",
		"DEFAULT",
		"ReceiptEnforcementGate",
		"CatalogPublicationV1",
	} {
		if strings.Contains(shadowTableDDL, forbidden) {
			t.Fatalf("strict v2 DDL contains forbidden %q", forbidden)
		}
	}
	for _, required := range []string{
		"CREATE TABLE IF NOT EXISTS " + shadowTableName,
		"LayoutBundleOwnerVerificationReceiptSetDigest VARCHAR(71) CHARACTER SET ascii COLLATE ascii_bin NOT NULL",
		"PRIMARY KEY (PublicationDigest)",
	} {
		if !strings.Contains(shadowTableDDL, required) {
			t.Fatalf("strict v2 DDL is missing %q", required)
		}
	}
	if strings.Count(shadowTableDDL, " NOT NULL") != len(strictV2Columns) {
		t.Fatalf("strict v2 DDL NOT NULL count = %d, want %d", strings.Count(shadowTableDDL, " NOT NULL"), len(strictV2Columns))
	}
}

func TestNilDatabaseAndUnverifiedHandlesFailClosed(t *testing.T) {
	for name, verify := range map[string]func(context.Context, *sql.DB) (*VerifiedTable, error){
		"create shadow": CreateHaltedShadow,
		"verify shadow": VerifyShadow,
		"verify live":   VerifyLive,
	} {
		if table, err := verify(context.Background(), nil); err == nil || table != nil {
			t.Fatalf("%s nil database = (%#v, %v)", name, table, err)
		}
	}
	if _, _, err := requireVerifiedTable(nil); !errors.Is(err, ErrSchema) {
		t.Fatalf("nil handle error = %v, want ErrSchema", err)
	}
	if _, _, err := requireVerifiedTable(&VerifiedTable{}); !errors.Is(err, ErrSchema) {
		t.Fatalf("zero handle error = %v, want ErrSchema", err)
	}
}

func newSchemaMock(t *testing.T) (*sql.DB, sqlmock.Sqlmock, func()) {
	t.Helper()
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	if err != nil {
		t.Fatal(err)
	}
	return db, mock, func() {
		_ = db.Close()
	}
}

func expectExactSchema(mock sqlmock.Sqlmock, tableName string) {
	expectColumns(mock, tableName, strictV2Columns)
	expectIndexes(mock, tableName, strictV2Indexes)
}

func expectColumns(mock sqlmock.Sqlmock, tableName string, columns []columnShape) {
	rows := sqlmock.NewRows([]string{
		"COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "EXTRA",
		"CHARACTER_SET_NAME", "COLLATION_NAME",
	})
	for _, column := range columns {
		var charSet, collation any
		if column.charSet != "" {
			charSet = column.charSet
		}
		if column.collation != "" {
			collation = column.collation
		}
		rows.AddRow(column.name, column.columnType, column.nullable, nil, "", charSet, collation)
	}
	mock.ExpectQuery(regexp.QuoteMeta(columnsQuery)).
		WithArgs(tableName).
		WillReturnRows(rows)
}

func expectIndexes(mock sqlmock.Sqlmock, tableName string, indexes []indexColumnShape) {
	rows := sqlmock.NewRows([]string{
		"INDEX_NAME", "NON_UNIQUE", "SEQ_IN_INDEX", "COLUMN_NAME", "SUB_PART",
	})
	for _, index := range indexes {
		rows.AddRow(index.indexName, index.nonUnique, index.sequence, index.column, nil)
	}
	mock.ExpectQuery(regexp.QuoteMeta(indexesQuery)).
		WithArgs(tableName).
		WillReturnRows(rows)
}

func assertSchemaExpectations(t *testing.T, mock sqlmock.Sqlmock) {
	t.Helper()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}