package workflow

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

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

const fastflowSchemaCheckTimeout = 5 * time.Second

func validateFastflowHasCmdSchema(cfg *gomysql.Config) error {
	sqlDB, err := sql.Open("mysql", cfg.FormatDSN())
	if err != nil {
		return fmt.Errorf("open workflow database for schema validation failed: %w", err)
	}
	defer sqlDB.Close()

	ctx, cancel := context.WithTimeout(context.Background(), fastflowSchemaCheckTimeout)
	defer cancel()
	if err := sqlDB.PingContext(ctx); err != nil {
		return fmt.Errorf("ping workflow database for schema validation failed: %w", err)
	}

	return validateFastflowHasCmdSchemaWithDB(ctx, sqlDB)
}

func validateFastflowHasCmdSchemaWithDB(ctx context.Context, db *sql.DB) error {
	var extra, generationExpression string
	err := db.QueryRowContext(ctx, `
		SELECT COALESCE(extra, ''), COALESCE(generation_expression, '')
		FROM information_schema.columns
		WHERE table_schema = DATABASE()
			AND table_name = 'dag_instances'
			AND column_name = 'has_cmd'
	`).Scan(&extra, &generationExpression)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return fmt.Errorf("fastflow schema invalid: dag_instances.has_cmd is missing after migration")
		}
		return fmt.Errorf("check dag_instances.has_cmd schema failed: %w", err)
	}

	normalizedExtra := strings.ToLower(extra)
	normalizedExpression := strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(generationExpression, "`", ""), " ", ""))
	if !strings.Contains(normalizedExtra, "generated") ||
		!strings.Contains(normalizedExpression, "cmd") ||
		!strings.Contains(normalizedExpression, "notnull") {
		return fmt.Errorf(
			"fastflow schema invalid: dag_instances.has_cmd must be a generated column from cmd, got extra=%q generation_expression=%q",
			extra,
			generationExpression,
		)
	}

	rows, err := db.QueryContext(ctx, `
		SELECT column_name
		FROM information_schema.statistics
		WHERE table_schema = DATABASE()
			AND table_name = 'dag_instances'
			AND index_name = 'idx_worker_has_cmd'
		ORDER BY seq_in_index
	`)
	if err != nil {
		return fmt.Errorf("check idx_worker_has_cmd schema failed: %w", err)
	}
	defer rows.Close()

	var indexColumns []string
	for rows.Next() {
		var column string
		if err := rows.Scan(&column); err != nil {
			return fmt.Errorf("scan idx_worker_has_cmd schema failed: %w", err)
		}
		indexColumns = append(indexColumns, column)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("scan idx_worker_has_cmd schema failed: %w", err)
	}
	if len(indexColumns) != 2 || indexColumns[0] != "worker" || indexColumns[1] != "has_cmd" {
		return fmt.Errorf("fastflow schema invalid: idx_worker_has_cmd must be (worker, has_cmd), got %v", indexColumns)
	}

	return nil
}