package executor

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

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/cqlerrors"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
)

func (e *ddlExecutor) executeCreateTrigger(ctx context.Context, user, password string, stmt *parser.CreateTriggerStatement) (*QueryResult, error) {
	if stmt == nil || stmt.Table == nil {
		return nil, cqlerrors.Invalid("trigger table is required")
	}
	if err := e.validateTriggerName(stmt.Name); err != nil {
		return nil, err
	}
	if err := e.validateTableName(stmt.Table); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(stmt.Table.Schema); err != nil {
		return nil, err
	}
	if !config.GetConfig().Features.MetadataOnlyTriggers {
		return nil, cqlerrors.Invalid("CREATE TRIGGER is not supported")
	}
	if err := e.ensureSchemaExists(ctx, user, password, stmt.Table.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureTableExists(ctx, user, password, stmt.Table.Schema, stmt.Table.Name); err != nil {
		return nil, err
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}
	existing, _, err := e.schemaManager.CQLTrigger(ctx, user, password, stmt.Table.Schema, stmt.Table.Name, stmt.Name)
	if err != nil {
		return nil, err
	}
	if existing != nil {
		return nil, cqlerrors.AlreadyExists(fmt.Sprintf("trigger %q already exists", stmt.Name))
	}

	result, err := e.executeDDLWithMeta(ctx, user, password, "", func(ctx context.Context, conn *sql.Conn) error {
		return e.writeTriggerMeta(ctx, conn, stmt)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	out, err := e.withSchemaChange(ctx, user, password, result, "CREATED", "TABLE", stmt.Table.Schema, stmt.Table.Name)
	if err != nil {
		return nil, err
	}
	out.Warnings = append(out.Warnings, "trigger stored in metadata only; triggers are not executed")
	return out, nil
}

func (e *ddlExecutor) executeDropTrigger(ctx context.Context, user, password string, stmt *parser.DropTriggerStatement) (*QueryResult, error) {
	if stmt == nil || stmt.Table == nil {
		return nil, cqlerrors.Invalid("trigger table is required")
	}
	if err := e.validateTriggerName(stmt.Name); err != nil {
		return nil, err
	}
	if err := e.validateTableName(stmt.Table); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(stmt.Table.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureSchemaExists(ctx, user, password, stmt.Table.Schema); err != nil {
		return nil, err
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}
	existing, _, err := e.schemaManager.CQLTrigger(ctx, user, password, stmt.Table.Schema, stmt.Table.Name, stmt.Name)
	if err != nil {
		return nil, err
	}
	if existing == nil {
		return nil, cqlerrors.Invalid(fmt.Sprintf("trigger %q does not exist", stmt.Name))
	}

	result, err := e.executeDDLWithMeta(ctx, user, password, "", func(ctx context.Context, conn *sql.Conn) error {
		return e.deleteTriggerMeta(ctx, conn, stmt.Table.Schema, stmt.Table.Name, stmt.Name)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	return e.withSchemaChange(ctx, user, password, result, "DROPPED", "TABLE", stmt.Table.Schema, stmt.Table.Name)
}