package mysqlbackend

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	mysql "github.com/go-sql-driver/mysql"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
)

const cqlMetaSchema = "cql_meta"

var cqlMetaDDL = []string{
	"CREATE SCHEMA IF NOT EXISTS cql_meta",
	`CREATE TABLE IF NOT EXISTS cql_meta.keyspaces (
		keyspace_name VARCHAR(255) NOT NULL,
		durable_writes BOOLEAN NOT NULL,
		replication_json JSON NOT NULL,
		PRIMARY KEY (keyspace_name)
	) ENGINE=InnoDB`,
	`CREATE TABLE IF NOT EXISTS cql_meta.tables (
		keyspace_name VARCHAR(255) NOT NULL,
		table_name VARCHAR(255) NOT NULL,
		options_json JSON NOT NULL,
		comment TEXT,
		PRIMARY KEY (keyspace_name, table_name)
	) ENGINE=InnoDB`,
	`CREATE TABLE IF NOT EXISTS cql_meta.columns (
			keyspace_name VARCHAR(255) NOT NULL,
			table_name VARCHAR(255) NOT NULL,
			column_name VARCHAR(255) NOT NULL,
		cql_type TEXT NOT NULL,
		default_expr TEXT NULL,
		on_update_expr TEXT NULL,
			kind VARCHAR(32) NOT NULL,
			position INT NOT NULL,
			clustering_order VARCHAR(4) NOT NULL DEFAULT 'ASC',
			PRIMARY KEY (keyspace_name, table_name, column_name)
		) ENGINE=InnoDB`,
	`CREATE TABLE IF NOT EXISTS cql_meta.types (
		keyspace_name VARCHAR(255) NOT NULL,
		type_name VARCHAR(255) NOT NULL,
		field_names_json JSON NOT NULL,
		field_types_json JSON NOT NULL,
		PRIMARY KEY (keyspace_name, type_name)
	) ENGINE=InnoDB`,
	`CREATE TABLE IF NOT EXISTS cql_meta.indexes (
		keyspace_name VARCHAR(255) NOT NULL,
		table_name VARCHAR(255) NOT NULL,
		index_name VARCHAR(255) NOT NULL,
		kind VARCHAR(32) NOT NULL,
		options_json JSON NOT NULL,
		PRIMARY KEY (keyspace_name, table_name, index_name)
	) ENGINE=InnoDB`,
	`CREATE TABLE IF NOT EXISTS cql_meta.views (
		keyspace_name VARCHAR(255) NOT NULL,
		view_name VARCHAR(255) NOT NULL,
		base_table VARCHAR(255) NOT NULL,
		where_clause TEXT NOT NULL,
		options_json JSON NOT NULL,
		PRIMARY KEY (keyspace_name, view_name)
	) ENGINE=InnoDB`,
	`CREATE TABLE IF NOT EXISTS cql_meta.triggers (
		keyspace_name VARCHAR(255) NOT NULL,
		table_name VARCHAR(255) NOT NULL,
		trigger_name VARCHAR(255) NOT NULL,
		options_json JSON NOT NULL,
		PRIMARY KEY (keyspace_name, table_name, trigger_name)
	) ENGINE=InnoDB`,
	`CREATE TABLE IF NOT EXISTS cql_meta.schema_version (
		id TINYINT NOT NULL DEFAULT 1,
		version CHAR(36) NOT NULL,
		updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
		PRIMARY KEY (id)
	) ENGINE=InnoDB`,
}

// EnsureCQLMeta creates the cql_meta schema and tables if they are missing.
func (m *SchemaManager) EnsureCQLMeta(ctx context.Context, user, password string) error {
	if m == nil || m.poolManager == nil {
		return fmt.Errorf("schema manager not initialized")
	}
	key := userKey{user: user, password: password}

	m.metaMu.Lock()
	if _, ok := m.metaEnsured[key]; ok {
		m.metaMu.Unlock()
		return nil
	}
	m.metaMu.Unlock()

	conn, release, err := getConn(ctx, m.poolManager, user, password)
	if err != nil {
		return err
	}
	if release != nil {
		defer release()
	}

	for _, stmt := range cqlMetaDDL {
		if _, err := execContextWithMetrics(ctx, conn, metrics.MySQLOpExecWrite, stmt); err != nil {
			return fmt.Errorf("failed to ensure cql_meta schema: %w", err)
		}
	}
	if err := ensureCQLMetaOptionalColumn(ctx, conn, "cql_meta.columns", "default_expr", "TEXT NULL"); err != nil {
		return fmt.Errorf("failed to ensure cql_meta.columns.default_expr: %w", err)
	}
	if err := ensureCQLMetaOptionalColumn(ctx, conn, "cql_meta.columns", "on_update_expr", "TEXT NULL"); err != nil {
		return fmt.Errorf("failed to ensure cql_meta.columns.on_update_expr: %w", err)
	}
	if err := ensureCQLMetaIndexDropped(ctx, conn, cqlMetaSchema, "columns", "uq_cql_meta_columns_position"); err != nil {
		return fmt.Errorf("failed to remove obsolete cql_meta.columns position index: %w", err)
	}

	m.metaMu.Lock()
	m.metaEnsured[key] = struct{}{}
	m.metaMu.Unlock()
	return nil
}

func ensureCQLMetaOptionalColumn(ctx context.Context, conn *sql.Conn, table, column, definition string) error {
	stmt := fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", table, column, definition)
	if _, err := execContextWithMetrics(ctx, conn, metrics.MySQLOpExecWrite, stmt); err != nil {
		if isDuplicateColumnError(err) {
			return nil
		}
		return err
	}
	return nil
}

func ensureCQLMetaIndexDropped(ctx context.Context, conn *sql.Conn, schema, table, index string) error {
	var exists int
	err := conn.QueryRowContext(ctx, `
		SELECT 1
		FROM information_schema.statistics
		WHERE table_schema = ? AND table_name = ? AND index_name = ?
		LIMIT 1
	`, schema, table, index).Scan(&exists)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		return err
	}

	stmt := fmt.Sprintf("ALTER TABLE %s.%s DROP INDEX %s", schema, table, index)
	_, err = execContextWithMetrics(ctx, conn, metrics.MySQLOpExecWrite, stmt)
	return err
}

func isDuplicateColumnError(err error) bool {
	var myErr *mysql.MySQLError
	if errors.As(err, &myErr) {
		return myErr.Number == 1060
	}
	return false
}

// CQLKeyspaces returns keyspace metadata and whether cql_meta is available.
func (m *SchemaManager) CQLKeyspaces(ctx context.Context, user, password string) ([]CQLKeyspaceInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Keyspaces(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	return entry.list, entry.available, nil
}

// CQLTables returns table metadata and whether cql_meta is available.
func (m *SchemaManager) CQLTables(ctx context.Context, user, password string) ([]CQLTableInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Tables(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	return entry.list, entry.available, nil
}

// CQLTablesForKeyspace returns table metadata for a keyspace.
func (m *SchemaManager) CQLTablesForKeyspace(ctx context.Context, user, password, keyspace string) ([]CQLTableInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Tables(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	if tables, ok := entry.bySchema[keyspace]; ok {
		return tables, entry.available, nil
	}
	return entry.bySchemaLower[normalizeLookup(keyspace)], entry.available, nil
}

// CQLColumnsForTable returns column metadata for a table.
func (m *SchemaManager) CQLColumnsForTable(ctx context.Context, user, password, keyspace, table string) ([]CQLColumnInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Columns(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	if cols, ok := entry.byTable[tableKey{schema: keyspace, table: table}]; ok {
		return cols, entry.available, nil
	}
	return entry.byTableLower[normalizeTableKey(keyspace, table)], entry.available, nil
}

// CQLColumns returns column metadata and whether cql_meta is available.
func (m *SchemaManager) CQLColumns(ctx context.Context, user, password string) ([]CQLColumnInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Columns(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	return entry.list, entry.available, nil
}

// CQLType returns metadata for a specific user-defined type.
func (m *SchemaManager) CQLType(ctx context.Context, user, password, keyspace, typeName string) (*CQLTypeInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Types(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	typ, ok := entry.byName[typeKey{schema: keyspace, name: typeName}]
	if !ok {
		typ, ok = entry.byNameLower[normalizeTypeKey(keyspace, typeName)]
		if !ok {
			return nil, entry.available, nil
		}
	}
	out := typ
	return &out, entry.available, nil
}

// CQLTypes returns type metadata and whether cql_meta is available.
func (m *SchemaManager) CQLTypes(ctx context.Context, user, password string) ([]CQLTypeInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Types(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	return entry.list, entry.available, nil
}

// CQLIndexes returns index metadata and whether cql_meta is available.
func (m *SchemaManager) CQLIndexes(ctx context.Context, user, password string) ([]CQLIndexInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Indexes(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	return entry.list, entry.available, nil
}

// CQLViews returns view metadata and whether cql_meta is available.
func (m *SchemaManager) CQLViews(ctx context.Context, user, password string) ([]CQLViewInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Views(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	return entry.list, entry.available, nil
}

// CQLView returns metadata for a specific view.
func (m *SchemaManager) CQLView(ctx context.Context, user, password, keyspace, viewName string) (*CQLViewInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Views(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	view, ok := entry.byName[viewKey{schema: keyspace, name: viewName}]
	if !ok {
		view, ok = entry.byNameLower[normalizeViewKey(keyspace, viewName)]
		if !ok {
			return nil, entry.available, nil
		}
	}
	out := view
	return &out, entry.available, nil
}

// CQLTriggers returns trigger metadata and whether cql_meta is available.
func (m *SchemaManager) CQLTriggers(ctx context.Context, user, password string) ([]CQLTriggerInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Triggers(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	return entry.list, entry.available, nil
}

// CQLTrigger returns metadata for a specific trigger.
func (m *SchemaManager) CQLTrigger(ctx context.Context, user, password, keyspace, table, triggerName string) (*CQLTriggerInfo, bool, error) {
	if m == nil || m.metaCache == nil {
		return nil, false, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.metaCache.Triggers(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, false, err
	}
	trigger, ok := entry.byName[triggerKey{schema: keyspace, table: table, name: triggerName}]
	if !ok {
		trigger, ok = entry.byNameLower[normalizeTriggerKey(keyspace, table, triggerName)]
		if !ok {
			return nil, entry.available, nil
		}
	}
	out := trigger
	return &out, entry.available, nil
}