package catalogpublicationv2store

import (
	"context"
	"database/sql"
	"fmt"
	"reflect"
	"sort"
)

const shadowTableDDL = `CREATE TABLE IF NOT EXISTS _System_CatalogPublicationV2Shadow (
PublicationDigest VARCHAR(71) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
OrgId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
CatalogRevision DECIMAL(20,0) NOT NULL,
CatalogSnapshotDigest VARCHAR(71) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
SchemaGeneration DECIMAL(20,0) NOT NULL,
EngineGeneration DECIMAL(20,0) NOT NULL,
LayoutValidationViewDigest VARCHAR(71) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
LayoutBundleOwnerVerificationReceiptSetDigest VARCHAR(71) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
LayoutActivationSetDigest VARCHAR(71) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
SecurityEpoch DECIMAL(20,0) NOT NULL,
LayoutRoutingSecurityViewDigest VARCHAR(71) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
CreatedDate DATETIME(6) NOT NULL,
PRIMARY KEY (PublicationDigest),
INDEX idx_catalog_publication_org_lineage (OrgId, CatalogRevision, SecurityEpoch))`

type columnShape struct {
	name       string
	columnType string
	nullable   string
	charSet    string
	collation  string
}

var strictV2Columns = []columnShape{
	{name: "PublicationDigest", columnType: "varchar(71)", nullable: "NO", charSet: "ascii", collation: "ascii_bin"},
	{name: "OrgId", columnType: "varchar(255)", nullable: "NO", charSet: "utf8mb4", collation: "utf8mb4_bin"},
	{name: "CatalogRevision", columnType: "decimal(20,0)", nullable: "NO"},
	{name: "CatalogSnapshotDigest", columnType: "varchar(71)", nullable: "NO", charSet: "ascii", collation: "ascii_bin"},
	{name: "SchemaGeneration", columnType: "decimal(20,0)", nullable: "NO"},
	{name: "EngineGeneration", columnType: "decimal(20,0)", nullable: "NO"},
	{name: "LayoutValidationViewDigest", columnType: "varchar(71)", nullable: "NO", charSet: "ascii", collation: "ascii_bin"},
	{name: "LayoutBundleOwnerVerificationReceiptSetDigest", columnType: "varchar(71)", nullable: "NO", charSet: "ascii", collation: "ascii_bin"},
	{name: "LayoutActivationSetDigest", columnType: "varchar(71)", nullable: "NO", charSet: "ascii", collation: "ascii_bin"},
	{name: "SecurityEpoch", columnType: "decimal(20,0)", nullable: "NO"},
	{name: "LayoutRoutingSecurityViewDigest", columnType: "varchar(71)", nullable: "NO", charSet: "ascii", collation: "ascii_bin"},
	{name: "CreatedDate", columnType: "datetime(6)", nullable: "NO"},
}

type indexColumnShape struct {
	indexName string
	nonUnique int64
	sequence  int64
	column    string
}

var strictV2Indexes = []indexColumnShape{
	{indexName: "PRIMARY", nonUnique: 0, sequence: 1, column: "PublicationDigest"},
	{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"},
}

const columnsQuery = `SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE,
COLUMN_DEFAULT, EXTRA, CHARACTER_SET_NAME, COLLATION_NAME
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
ORDER BY ORDINAL_POSITION`

const indexesQuery = `SELECT INDEX_NAME, NON_UNIQUE, SEQ_IN_INDEX, COLUMN_NAME, SUB_PART
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
ORDER BY INDEX_NAME, SEQ_IN_INDEX`

// CreateHaltedShadow creates only the fixed strict-v2 shadow table and returns
// a handle after exact structural readback. It is an explicit offline operator
// primitive, not a startup migration.
func CreateHaltedShadow(ctx context.Context, db *sql.DB) (*VerifiedTable, error) {
	if db == nil {
		return nil, errorsDatabaseRequired()
	}
	if _, err := db.ExecContext(ctx, shadowTableDDL); err != nil {
		return nil, err
	}
	return verifyTable(ctx, db, shadowTableIdentity)
}

// VerifyShadow returns a capability only when the fixed halted-migration
// shadow has the complete strict-v2 shape.
func VerifyShadow(ctx context.Context, db *sql.DB) (*VerifiedTable, error) {
	return verifyTable(ctx, db, shadowTableIdentity)
}

// VerifyLive returns a capability only when the future canonical table is
// already the exact strict-v2 table. The current mixed/v1 table fails this
// check; this function never alters it.
func VerifyLive(ctx context.Context, db *sql.DB) (*VerifiedTable, error) {
	return verifyTable(ctx, db, liveTableIdentity)
}

func verifyTable(
	ctx context.Context,
	db *sql.DB,
	identity *tableIdentity,
) (*VerifiedTable, error) {
	if db == nil {
		return nil, errorsDatabaseRequired()
	}
	if identity != liveTableIdentity && identity != shadowTableIdentity {
		return nil, ErrSchema
	}
	columns, err := readColumnShapes(ctx, db, identity.name)
	if err != nil {
		return nil, err
	}
	if !reflect.DeepEqual(columns, strictV2Columns) {
		return nil, fmt.Errorf(
			"%w: table %s columns = %#v, want %#v",
			ErrSchema,
			identity.name,
			columns,
			strictV2Columns,
		)
	}
	indexes, err := readIndexShapes(ctx, db, identity.name)
	if err != nil {
		return nil, err
	}
	if !reflect.DeepEqual(indexes, strictV2Indexes) {
		return nil, fmt.Errorf(
			"%w: table %s indexes = %#v, want %#v",
			ErrSchema,
			identity.name,
			indexes,
			strictV2Indexes,
		)
	}
	return &VerifiedTable{db: db, identity: identity}, nil
}

func readColumnShapes(ctx context.Context, db *sql.DB, tableName string) ([]columnShape, error) {
	rows, err := db.QueryContext(ctx, columnsQuery, tableName)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var result []columnShape
	for rows.Next() {
		var shape columnShape
		var defaultValue, charSet, collation sql.NullString
		var extra string
		if err := rows.Scan(
			&shape.name,
			&shape.columnType,
			&shape.nullable,
			&defaultValue,
			&extra,
			&charSet,
			&collation,
		); err != nil {
			return nil, err
		}
		if defaultValue.Valid || extra != "" {
			return nil, fmt.Errorf(
				"%w: table %s column %s has default %q or extra %q",
				ErrSchema,
				tableName,
				shape.name,
				defaultValue.String,
				extra,
			)
		}
		if charSet.Valid {
			shape.charSet = charSet.String
		}
		if collation.Valid {
			shape.collation = collation.String
		}
		result = append(result, shape)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return result, nil
}

func readIndexShapes(ctx context.Context, db *sql.DB, tableName string) ([]indexColumnShape, error) {
	rows, err := db.QueryContext(ctx, indexesQuery, tableName)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var result []indexColumnShape
	for rows.Next() {
		var shape indexColumnShape
		var subPart sql.NullInt64
		if err := rows.Scan(
			&shape.indexName,
			&shape.nonUnique,
			&shape.sequence,
			&shape.column,
			&subPart,
		); err != nil {
			return nil, err
		}
		if subPart.Valid {
			return nil, fmt.Errorf(
				"%w: table %s index %s uses a prefix length",
				ErrSchema,
				tableName,
				shape.indexName,
			)
		}
		result = append(result, shape)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	sort.Slice(result, func(i, j int) bool {
		if result[i].indexName != result[j].indexName {
			return result[i].indexName < result[j].indexName
		}
		return result[i].sequence < result[j].sequence
	})
	return result, nil
}

func errorsDatabaseRequired() error {
	return fmt.Errorf("catalog publication v2 store: database is required")
}