package executor

import (
	"context"
	"fmt"
	"sort"
	"strings"

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

func (e *ddlExecutor) executeGrant(ctx context.Context, user, password string, stmt *parser.GrantStatement) (*QueryResult, error) {
	if stmt == nil {
		return nil, cqlerrors.Invalid("GRANT statement is required")
	}
	if strings.ToUpper(stmt.Resource) != "KEYSPACE" {
		return nil, cqlerrors.Invalid("only GRANT ... ON KEYSPACE is supported")
	}
	if err := e.validateKeyspaceName(stmt.Keyspace); err != nil {
		return nil, err
	}
	if keyspace.IsBuiltIn(stmt.Keyspace) || keyspace.IsInternalSchema(stmt.Keyspace) {
		return nil, cqlerrors.Invalid(fmt.Sprintf("keyspace %q is reserved", stmt.Keyspace))
	}
	if err := validateTiDBUserName(stmt.Role); err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}

	conn, release, err := e.getConn(ctx, user, password)
	if err != nil {
		return nil, cqlerrors.Invalid(fmt.Sprintf("failed to get connection: %s", err.Error()))
	}
	if release != nil {
		defer release()
	}

	exists, err := e.tidbUserExists(ctx, conn, stmt.Role)
	if err != nil {
		return nil, err
	}
	if !exists {
		return nil, cqlerrors.Invalid(fmt.Sprintf("user %q does not exist", stmt.Role))
	}

	if e.schemaManager != nil {
		exists, err := e.schemaManager.SchemaExists(ctx, user, password, stmt.Keyspace)
		if err != nil {
			return nil, err
		}
		if !exists {
			return nil, cqlerrors.Invalid(fmt.Sprintf("keyspace %q does not exist", stmt.Keyspace))
		}
	}

	privileges, err := grantPrivileges(stmt.Permissions)
	if err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}
	sql := fmt.Sprintf("GRANT %s ON %s.* TO %s",
		strings.Join(privileges, ", "),
		quoteMySQLIdentifier(stmt.Keyspace),
		quoteTiDBAccount(stmt.Role, defaultTiDBUserHost))
	result, err := e.executeDDL(ctx, conn, sql)
	if err != nil {
		return nil, err
	}
	if e.poolManager != nil {
		e.poolManager.InvalidateUser(stmt.Role)
	}
	return result, nil
}

func grantPrivileges(permissions []string) ([]string, error) {
	if len(permissions) == 0 {
		return nil, fmt.Errorf("GRANT requires at least one permission")
	}
	privileges := make(map[string]struct{})
	for _, permission := range permissions {
		switch strings.ToUpper(permission) {
		case "ALL":
			for _, privilege := range []string{"ALTER", "CREATE", "DELETE", "DROP", "INDEX", "INSERT", "SELECT", "UPDATE"} {
				privileges[privilege] = struct{}{}
			}
		case "SELECT":
			privileges["SELECT"] = struct{}{}
		case "MODIFY":
			privileges["DELETE"] = struct{}{}
			privileges["INSERT"] = struct{}{}
			privileges["UPDATE"] = struct{}{}
		case "CREATE":
			privileges["CREATE"] = struct{}{}
		case "ALTER":
			privileges["ALTER"] = struct{}{}
			privileges["INDEX"] = struct{}{}
		case "DROP":
			privileges["DROP"] = struct{}{}
		case "AUTHORIZE":
			return nil, fmt.Errorf("GRANT AUTHORIZE is not supported in TiDB-native user mode")
		default:
			return nil, fmt.Errorf("unsupported GRANT permission %q", permission)
		}
	}

	out := make([]string, 0, len(privileges))
	for privilege := range privileges {
		out = append(out, privilege)
	}
	sort.Strings(out)
	return out, nil
}