package executor

import (
	"context"

	"go.uber.org/zap"

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

func (e *dmlExecutor) executeDMLStatement(ctx context.Context, exec sqlExecutor, user, password string, stmt parser.Statement, params []interface{}, plan *dmlPlan) (*QueryResult, error) {
	// Retry the whole logical write on deadlock (when not inside a caller-owned
	// transaction); a deadlock rolls the statement back, so re-running is safe.
	return e.executeWriteWithRetry(ctx, exec, func() (*QueryResult, error) {
		return e.executeDMLStatementOnce(ctx, exec, user, password, stmt, params, plan)
	})
}

func (e *dmlExecutor) executeDMLStatementOnce(ctx context.Context, exec sqlExecutor, user, password string, stmt parser.Statement, params []interface{}, plan *dmlPlan) (*QueryResult, error) {
	switch s := stmt.(type) {
	case *parser.InsertStatement:
		if s.JSON {
			return e.executeJSONInsert(ctx, exec, user, password, s, params)
		}
	}

	if stmt != nil {
		var columnTypes map[string]string
		var err error
		switch s := stmt.(type) {
		case *parser.InsertStatement:
			columnTypes, err = e.lookupColumnTypesIfAvailable(ctx, user, password, s.Table)
		case *parser.UpdateStatement:
			columnTypes, err = e.lookupColumnTypesIfAvailable(ctx, user, password, s.Table)
		case *parser.DeleteStatement:
			columnTypes, err = e.lookupColumnTypesIfAvailable(ctx, user, password, s.From)
		}
		if err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
		updated, updatedParams, changed, err := applyCastConversionsForDML(stmt, params, columnTypes)
		if err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
		stmt = updated
		params = updatedParams
		if changed {
			plan = nil
		}

		updated, updatedParams, timeChanged, err := applyTimeConversionsForDML(stmt, params, columnTypes)
		if err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
		stmt = updated
		params = updatedParams
		if timeChanged {
			plan = nil
		}
	}

	if updateStmt, ok := stmt.(*parser.UpdateStatement); ok {
		if result, handled, err := e.executeCollectionMutationUpdate(ctx, exec, user, password, updateStmt, params); err != nil || handled {
			return result, err
		}
	}
	if deleteStmt, ok := stmt.(*parser.DeleteStatement); ok {
		if assignments := deleteAssignments(deleteStmt); len(assignments) > 0 {
			updateStmt := &parser.UpdateStatement{
				Token:       deleteStmt.Token,
				Table:       deleteStmt.From,
				Assigments:  assignments,
				Where:       deleteStmt.Where,
				IfExists:    deleteStmt.IfExists,
				IfCondition: deleteStmt.IfCondition,
				Timestamp:   deleteStmt.Timestamp,
			}
			if result, handled, err := e.executeCollectionMutationUpdate(ctx, exec, user, password, updateStmt, params); err != nil || handled {
				return result, err
			}
		}
	}

	if len(params) > 0 && hasUnset(params) {
		switch s := stmt.(type) {
		case *parser.InsertStatement:
			return e.executeInsertWithUnset(ctx, exec, user, password, s, params, plan)
		case *parser.UpdateStatement:
			return e.executeUpdateWithUnset(ctx, exec, user, password, s, params, plan)
		case *parser.DeleteStatement:
			return e.executeDeleteWithUnset(ctx, exec, user, password, s, params, plan)
		}
	}

	if plan == nil {
		var err error
		plan, err = e.buildDMLPlan(ctx, user, password, stmt)
		if err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
	}
	if plan.sql != "" {
		e.logger.Debug("Executing DML query",
			zap.String("user", user),
			zap.Stringer("cql", stmt),
			zap.String("sql", plan.sql))
	}

	return e.executeDMLPlan(ctx, exec, plan, params)
}