package runtimedeploytask

import (
	"context"
	_ "embed"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/codex-k8s/kodex/libs/go/errs"
	domainrepo "github.com/codex-k8s/kodex/services/internal/control-plane/internal/domain/repository/runtimedeploytask"
	entitytypes "github.com/codex-k8s/kodex/services/internal/control-plane/internal/domain/types/entity"
	querytypes "github.com/codex-k8s/kodex/services/internal/control-plane/internal/domain/types/query"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/jackc/pgx/v5/pgxpool"
)

var (
	//go:embed sql/get_by_run_id.sql
	queryGetByRunID string
	//go:embed sql/find_active_by_namespace.sql
	queryFindActiveByNamespace string
	//go:embed sql/select_by_run_id_for_update.sql
	querySelectByRunIDForUpdate string
	//go:embed sql/insert_pending.sql
	queryInsertPending string
	//go:embed sql/reset_desired_to_pending.sql
	queryResetDesiredToPending string
	//go:embed sql/cancel_superseded_deploy_only.sql
	queryCancelSupersededDeployOnly string
	//go:embed sql/claim_next.sql
	queryClaimNext string
	//go:embed sql/mark_succeeded.sql
	queryMarkSucceeded string
	//go:embed sql/mark_failed.sql
	queryMarkFailed string
	//go:embed sql/renew_lease.sql
	queryRenewLease string
	//go:embed sql/requeue_running.sql
	queryRequeueRunning string
	//go:embed sql/request_action.sql
	queryRequestAction string
	//go:embed sql/count_recent.sql
	queryCountRecent string
	//go:embed sql/list_recent.sql
	queryListRecent string
	//go:embed sql/append_log.sql
	queryAppendLog string
	//go:embed sql/cleanup_task_logs_updated_before.sql
	queryCleanupTaskLogsUpdatedBefore string
)

// Repository persists runtime_deploy_tasks state in PostgreSQL.
type Repository struct {
	db *pgxpool.Pool
}

// NewRepository constructs PostgreSQL runtime_deploy_tasks repository.
func NewRepository(db *pgxpool.Pool) *Repository {
	return &Repository{db: db}
}

// UpsertDesired creates or updates one run-bound desired deployment state.
func (r *Repository) UpsertDesired(ctx context.Context, params domainrepo.UpsertDesiredParams) (domainrepo.Task, error) {
	normalized, err := normalizeUpsertParams(params)
	if err != nil {
		return domainrepo.Task{}, err
	}

	tx, err := r.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return domainrepo.Task{}, fmt.Errorf("begin runtime deploy upsert transaction: %w", err)
	}
	defer func() {
		_ = tx.Rollback(ctx)
	}()

	existing, found, err := getByRunIDForUpdate(ctx, tx, normalized.RunID)
	if err != nil {
		return domainrepo.Task{}, err
	}
	if !found {
		inserted, insertErr := insertPending(ctx, tx, normalized)
		if insertErr != nil {
			return domainrepo.Task{}, insertErr
		}
		if shouldCancelSupersededDeployOnly(inserted, normalized) {
			if _, cancelErr := cancelSupersededDeployOnlyTasks(ctx, tx, inserted.RunID, normalized); cancelErr != nil {
				return domainrepo.Task{}, cancelErr
			}
		}
		if err := tx.Commit(ctx); err != nil {
			return domainrepo.Task{}, fmt.Errorf("commit runtime deploy upsert transaction: %w", err)
		}
		return inserted, nil
	}

	if !shouldResetDesired(existing, normalized) {
		if err := tx.Commit(ctx); err != nil {
			return domainrepo.Task{}, fmt.Errorf("commit runtime deploy upsert transaction: %w", err)
		}
		return existing, nil
	}

	updated, err := resetDesiredToPending(ctx, tx, normalized)
	if err != nil {
		return domainrepo.Task{}, err
	}
	if shouldCancelSupersededDeployOnly(updated, normalized) {
		if _, cancelErr := cancelSupersededDeployOnlyTasks(ctx, tx, updated.RunID, normalized); cancelErr != nil {
			return domainrepo.Task{}, cancelErr
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return domainrepo.Task{}, fmt.Errorf("commit runtime deploy upsert transaction: %w", err)
	}
	return updated, nil
}

// GetByRunID returns one runtime deploy task by run id.
func (r *Repository) GetByRunID(ctx context.Context, runID string) (domainrepo.Task, bool, error) {
	return r.getOptionalTask(ctx, queryGetByRunID, strings.TrimSpace(runID), "query runtime deploy task by run_id=%s")
}

// FindActiveByNamespace returns one pending/running task for namespace when present.
func (r *Repository) FindActiveByNamespace(ctx context.Context, namespace string) (domainrepo.Task, bool, error) {
	return r.getOptionalTask(ctx, queryFindActiveByNamespace, strings.TrimSpace(namespace), "query active runtime deploy task by namespace=%s")
}

func (r *Repository) getOptionalTask(ctx context.Context, query string, lookupValue string, errorFormat string) (domainrepo.Task, bool, error) {
	if lookupValue == "" {
		return domainrepo.Task{}, false, nil
	}
	row := r.db.QueryRow(ctx, query, lookupValue)
	task, err := scanTask(row)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return domainrepo.Task{}, false, nil
		}
		return domainrepo.Task{}, false, fmt.Errorf(errorFormat+": %w", lookupValue, err)
	}
	return task, true, nil
}

// ClaimNext acquires one pending/expired-running task lease.
func (r *Repository) ClaimNext(ctx context.Context, params domainrepo.ClaimParams) (domainrepo.Task, bool, error) {
	leaseOwner := strings.TrimSpace(params.LeaseOwner)
	leaseTTL := strings.TrimSpace(params.LeaseTTL)
	staleRunningTimeout := strings.TrimSpace(params.StaleRunningTimeout)
	if leaseOwner == "" {
		return domainrepo.Task{}, false, fmt.Errorf("claim runtime deploy task: lease_owner is required")
	}
	if leaseTTL == "" {
		return domainrepo.Task{}, false, fmt.Errorf("claim runtime deploy task: lease_ttl is required")
	}
	if staleRunningTimeout == "" {
		staleRunningTimeout = "2 minutes"
	}

	row := r.db.QueryRow(ctx, queryClaimNext, leaseOwner, leaseTTL, staleRunningTimeout)
	task, err := scanTask(row)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return domainrepo.Task{}, false, nil
		}
		return domainrepo.Task{}, false, fmt.Errorf("claim runtime deploy task: %w", err)
	}
	return task, true, nil
}

// MarkSucceeded sets successful terminal state for one leased task.
func (r *Repository) MarkSucceeded(ctx context.Context, params domainrepo.MarkSucceededParams) (bool, error) {
	runID := strings.TrimSpace(params.RunID)
	leaseOwner := strings.TrimSpace(params.LeaseOwner)
	if runID == "" {
		return false, fmt.Errorf("mark runtime deploy task succeeded: run_id is required")
	}
	if leaseOwner == "" {
		return false, fmt.Errorf("mark runtime deploy task succeeded: lease_owner is required")
	}

	var returnedRunID string
	err := r.db.QueryRow(ctx, queryMarkSucceeded, runID, leaseOwner, strings.TrimSpace(params.ResultNamespace), strings.TrimSpace(params.ResultTargetEnv)).Scan(&returnedRunID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return false, nil
		}
		return false, fmt.Errorf("mark runtime deploy task %s succeeded: %w", runID, err)
	}
	return strings.TrimSpace(returnedRunID) != "", nil
}

// MarkFailed sets failed terminal state for one leased task.
func (r *Repository) MarkFailed(ctx context.Context, params domainrepo.MarkFailedParams) (bool, error) {
	runID := strings.TrimSpace(params.RunID)
	leaseOwner := strings.TrimSpace(params.LeaseOwner)
	if runID == "" {
		return false, fmt.Errorf("mark runtime deploy task failed: run_id is required")
	}
	if leaseOwner == "" {
		return false, fmt.Errorf("mark runtime deploy task failed: lease_owner is required")
	}

	var returnedRunID string
	err := r.db.QueryRow(ctx, queryMarkFailed, runID, leaseOwner, strings.TrimSpace(params.LastError)).Scan(&returnedRunID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return false, nil
		}
		return false, fmt.Errorf("mark runtime deploy task %s failed: %w", runID, err)
	}
	return strings.TrimSpace(returnedRunID) != "", nil
}

// RenewLease extends running task lease for current owner.
func (r *Repository) RenewLease(ctx context.Context, params domainrepo.RenewLeaseParams) (bool, error) {
	runID := strings.TrimSpace(params.RunID)
	leaseOwner := strings.TrimSpace(params.LeaseOwner)
	leaseTTL := strings.TrimSpace(params.LeaseTTL)
	if runID == "" {
		return false, fmt.Errorf("renew runtime deploy task lease: run_id is required")
	}
	if leaseOwner == "" {
		return false, fmt.Errorf("renew runtime deploy task lease: lease_owner is required")
	}
	if leaseTTL == "" {
		return false, fmt.Errorf("renew runtime deploy task lease: lease_ttl is required")
	}

	var returnedRunID string
	err := r.db.QueryRow(ctx, queryRenewLease, runID, leaseOwner, leaseTTL).Scan(&returnedRunID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return false, nil
		}
		return false, fmt.Errorf("renew runtime deploy task lease for run %s: %w", runID, err)
	}
	return strings.TrimSpace(returnedRunID) != "", nil
}

// Requeue returns one running task lease back to pending for a new reconciler.
func (r *Repository) Requeue(ctx context.Context, params domainrepo.RequeueParams) (bool, error) {
	runID := strings.TrimSpace(params.RunID)
	leaseOwner := strings.TrimSpace(params.LeaseOwner)
	lastError := strings.TrimSpace(params.LastError)
	if runID == "" {
		return false, fmt.Errorf("requeue runtime deploy task: run_id is required")
	}
	if leaseOwner == "" {
		return false, fmt.Errorf("requeue runtime deploy task: lease_owner is required")
	}

	var returnedRunID string
	err := r.db.QueryRow(ctx, queryRequeueRunning, runID, leaseOwner, lastError).Scan(&returnedRunID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return false, nil
		}
		return false, fmt.Errorf("requeue runtime deploy task %s: %w", runID, err)
	}
	return strings.TrimSpace(returnedRunID) != "", nil
}

// RequestAction records one operator cancel/stop request and transitions task to canceled idempotently.
func (r *Repository) RequestAction(ctx context.Context, params domainrepo.RequestActionParams) (domainrepo.RequestActionResult, error) {
	normalized, err := normalizeRequestActionParams(params)
	if err != nil {
		return domainrepo.RequestActionResult{}, err
	}

	tx, err := r.db.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return domainrepo.RequestActionResult{}, fmt.Errorf("begin runtime deploy action transaction: %w", err)
	}
	defer func() {
		_ = tx.Rollback(ctx)
	}()

	task, found, err := getByRunIDForUpdate(ctx, tx, normalized.RunID)
	if err != nil {
		return domainrepo.RequestActionResult{}, err
	}
	if !found {
		return domainrepo.RequestActionResult{}, errs.NotFound{Msg: "run_id: not found"}
	}
	if task.Status.IsTerminal() {
		if err := tx.Commit(ctx); err != nil {
			return domainrepo.RequestActionResult{}, fmt.Errorf("commit runtime deploy action transaction: %w", err)
		}
		return domainrepo.RequestActionResult{
			Task:            task,
			PreviousStatus:  task.Status,
			CurrentStatus:   task.Status,
			AlreadyTerminal: true,
		}, nil
	}
	if normalized.Action == querytypes.RuntimeDeployTaskActionStop {
		if err := validateStopActionTask(task, normalized.RequestedAt); err != nil {
			return domainrepo.RequestActionResult{}, err
		}
	}

	updatedTask, err := requestAction(ctx, tx, normalized)
	if err != nil {
		return domainrepo.RequestActionResult{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domainrepo.RequestActionResult{}, fmt.Errorf("commit runtime deploy action transaction: %w", err)
	}
	return domainrepo.RequestActionResult{
		Task:            updatedTask,
		PreviousStatus:  task.Status,
		CurrentStatus:   updatedTask.Status,
		AlreadyTerminal: false,
	}, nil
}

func validateStopActionTask(task domainrepo.Task, requestedAt time.Time) error {
	if task.Status != entitytypes.RuntimeDeployTaskStatusRunning {
		return errs.FailedPrecondition{Msg: "stop requires running task"}
	}
	if !hasActiveLease(task, requestedAt) {
		return errs.FailedPrecondition{Msg: "running task lease is not active"}
	}
	return nil
}

func hasActiveLease(task domainrepo.Task, requestedAt time.Time) bool {
	if strings.TrimSpace(task.LeaseOwner) == "" {
		return false
	}
	if task.LeaseUntil.IsZero() {
		return false
	}
	return task.LeaseUntil.UTC().After(requestedAt.UTC())
}

// ListRecent returns one runtime deploy task page ordered by updated_at desc.
func (r *Repository) ListRecent(ctx context.Context, filter domainrepo.ListFilter) ([]domainrepo.Task, int, error) {
	page := filter.Page
	if page <= 0 {
		page = 1
	}
	pageSize := filter.PageSize
	if pageSize <= 0 {
		pageSize = 20
	}
	if pageSize > 1000 {
		pageSize = 1000
	}
	offset := (page - 1) * pageSize
	if offset < 0 {
		offset = 0
	}
	totalCount, err := r.countRecent(ctx, filter)
	if err != nil {
		return nil, 0, err
	}
	rows, err := r.db.Query(
		ctx,
		queryListRecent,
		strings.TrimSpace(filter.Status),
		strings.TrimSpace(filter.TargetEnv),
		pageSize,
		offset,
	)
	if err != nil {
		return nil, 0, fmt.Errorf("list runtime deploy tasks: %w", err)
	}
	defer rows.Close()

	items := make([]domainrepo.Task, 0, pageSize)
	for rows.Next() {
		item, scanErr := scanTask(rows)
		if scanErr != nil {
			return nil, 0, fmt.Errorf("scan runtime deploy task list item: %w", scanErr)
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, 0, fmt.Errorf("iterate runtime deploy task list rows: %w", err)
	}
	return items, totalCount, nil
}

func (r *Repository) countRecent(ctx context.Context, filter domainrepo.ListFilter) (int, error) {
	var totalCount int
	if err := r.db.QueryRow(
		ctx,
		queryCountRecent,
		strings.TrimSpace(filter.Status),
		strings.TrimSpace(filter.TargetEnv),
	).Scan(&totalCount); err != nil {
		return 0, fmt.Errorf("count runtime deploy tasks: %w", err)
	}
	return totalCount, nil
}

// AppendLog appends one task log line.
func (r *Repository) AppendLog(ctx context.Context, params domainrepo.AppendLogParams) error {
	runID := strings.TrimSpace(params.RunID)
	if runID == "" {
		return fmt.Errorf("append runtime deploy task log: run_id is required")
	}
	stage := strings.TrimSpace(params.Stage)
	if stage == "" {
		stage = "deploy"
	}
	level := strings.TrimSpace(params.Level)
	if level == "" {
		level = "info"
	}
	message := strings.TrimSpace(params.Message)
	if message == "" {
		return nil
	}
	maxLines := params.MaxLines
	if maxLines <= 0 {
		maxLines = 200
	}
	if maxLines > 5000 {
		maxLines = 5000
	}

	tag, err := r.db.Exec(ctx, queryAppendLog, runID, stage, level, message, maxLines)
	if err != nil {
		return fmt.Errorf("append runtime deploy task log for run %s: %w", runID, err)
	}
	if tag.RowsAffected() == 0 {
		return nil
	}
	return nil
}

// CleanupTaskLogsUpdatedBefore clears heavy logs payloads for old tasks.
func (r *Repository) CleanupTaskLogsUpdatedBefore(ctx context.Context, updatedBefore time.Time) (int64, error) {
	if r == nil || r.db == nil {
		return 0, fmt.Errorf("runtime deploy task repository is not configured")
	}

	cutoff := updatedBefore.UTC()
	if cutoff.IsZero() {
		return 0, fmt.Errorf("updated_before is required")
	}

	tag, err := r.db.Exec(ctx, queryCleanupTaskLogsUpdatedBefore, cutoff)
	if err != nil {
		return 0, fmt.Errorf("cleanup runtime deploy task logs before %s: %w", cutoff.Format(time.RFC3339), err)
	}
	affected := tag.RowsAffected()
	return affected, nil
}

type taskRowScanner interface {
	Scan(dest ...any) error
}

func getByRunIDForUpdate(ctx context.Context, tx pgx.Tx, runID string) (domainrepo.Task, bool, error) {
	row := tx.QueryRow(ctx, querySelectByRunIDForUpdate, runID)
	task, err := scanTask(row)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return domainrepo.Task{}, false, nil
		}
		return domainrepo.Task{}, false, fmt.Errorf("select runtime deploy task run_id=%s for update: %w", runID, err)
	}
	return task, true, nil
}

func insertPending(ctx context.Context, tx pgx.Tx, params domainrepo.UpsertDesiredParams) (domainrepo.Task, error) {
	return applyDesiredStateMutation(ctx, tx, queryInsertPending, params, "insert runtime deploy task")
}

func resetDesiredToPending(ctx context.Context, tx pgx.Tx, params domainrepo.UpsertDesiredParams) (domainrepo.Task, error) {
	return applyDesiredStateMutation(ctx, tx, queryResetDesiredToPending, params, "reset runtime deploy task to pending")
}

func applyDesiredStateMutation(ctx context.Context, tx pgx.Tx, sqlQuery string, params domainrepo.UpsertDesiredParams, action string) (domainrepo.Task, error) {
	row := tx.QueryRow(
		ctx,
		sqlQuery,
		params.RunID,
		params.RuntimeMode,
		params.Namespace,
		params.TargetEnv,
		params.SlotNo,
		params.RepositoryFullName,
		params.ServicesYAMLPath,
		params.BuildRef,
		params.DeployOnly,
	)
	task, err := scanTask(row)
	if err != nil {
		return domainrepo.Task{}, fmt.Errorf("%s run_id=%s: %w", action, params.RunID, err)
	}
	return task, nil
}

func scanTask(row taskRowScanner) (domainrepo.Task, error) {
	var (
		task                 domainrepo.Task
		statusRaw            string
		leaseUntil           pgtype.Timestamptz
		cancelRequestedAt    pgtype.Timestamptz
		stopRequestedAt      pgtype.Timestamptz
		createdAt            time.Time
		updatedAt            time.Time
		startedAt            pgtype.Timestamptz
		finishedAt           pgtype.Timestamptz
		logsRaw              []byte
		leaseOwner           string
		lastError            string
		resultNamespace      string
		resultTargetEnv      string
		cancelRequestedBy    string
		cancelReason         string
		stopRequestedBy      string
		stopReason           string
		terminalStatusSource string
	)

	err := row.Scan(
		&task.RunID,
		&task.RuntimeMode,
		&task.Namespace,
		&task.TargetEnv,
		&task.SlotNo,
		&task.RepositoryFullName,
		&task.ServicesYAMLPath,
		&task.BuildRef,
		&task.DeployOnly,
		&statusRaw,
		&leaseOwner,
		&leaseUntil,
		&task.Attempts,
		&lastError,
		&resultNamespace,
		&resultTargetEnv,
		&cancelRequestedAt,
		&cancelRequestedBy,
		&cancelReason,
		&stopRequestedAt,
		&stopRequestedBy,
		&stopReason,
		&terminalStatusSource,
		&task.TerminalEventSeq,
		&createdAt,
		&updatedAt,
		&startedAt,
		&finishedAt,
		&logsRaw,
	)
	if err != nil {
		return domainrepo.Task{}, err
	}

	status, err := parseRuntimeDeployStatus(statusRaw)
	if err != nil {
		return domainrepo.Task{}, err
	}
	task.Status = status
	task.LeaseOwner = strings.TrimSpace(leaseOwner)
	if leaseUntil.Valid {
		task.LeaseUntil = leaseUntil.Time.UTC()
	}
	task.LastError = strings.TrimSpace(lastError)
	task.ResultNamespace = strings.TrimSpace(resultNamespace)
	task.ResultTargetEnv = strings.TrimSpace(resultTargetEnv)
	if cancelRequestedAt.Valid {
		task.CancelRequestedAt = cancelRequestedAt.Time.UTC()
	}
	task.CancelRequestedBy = strings.TrimSpace(cancelRequestedBy)
	task.CancelReason = strings.TrimSpace(cancelReason)
	if stopRequestedAt.Valid {
		task.StopRequestedAt = stopRequestedAt.Time.UTC()
	}
	task.StopRequestedBy = strings.TrimSpace(stopRequestedBy)
	task.StopReason = strings.TrimSpace(stopReason)
	task.TerminalStatusSource = parseTerminalStatusSource(terminalStatusSource)
	task.CreatedAt = createdAt.UTC()
	task.UpdatedAt = updatedAt.UTC()
	if startedAt.Valid {
		task.StartedAt = startedAt.Time.UTC()
	}
	if finishedAt.Valid {
		task.FinishedAt = finishedAt.Time.UTC()
	}
	task.Logs = parseTaskLogs(logsRaw)

	return task, nil
}

func parseTaskLogs(raw []byte) []entitytypes.RuntimeDeployTaskLogEntry {
	trimmed := strings.TrimSpace(string(raw))
	if trimmed == "" || trimmed == "null" {
		return []entitytypes.RuntimeDeployTaskLogEntry{}
	}
	type dbLogEntry struct {
		Stage     string    `json:"stage"`
		Level     string    `json:"level"`
		Message   string    `json:"message"`
		CreatedAt time.Time `json:"created_at"`
	}
	parsed := make([]dbLogEntry, 0)
	if err := json.Unmarshal(raw, &parsed); err != nil {
		return []entitytypes.RuntimeDeployTaskLogEntry{}
	}
	out := make([]entitytypes.RuntimeDeployTaskLogEntry, 0, len(parsed))
	for _, entry := range parsed {
		out = append(out, entitytypes.RuntimeDeployTaskLogEntry{
			Stage:     strings.TrimSpace(entry.Stage),
			Level:     strings.TrimSpace(entry.Level),
			Message:   strings.TrimSpace(entry.Message),
			CreatedAt: entry.CreatedAt.UTC(),
		})
	}
	return out
}

func parseRuntimeDeployStatus(raw string) (entitytypes.RuntimeDeployTaskStatus, error) {
	status := entitytypes.RuntimeDeployTaskStatus(strings.TrimSpace(raw))
	switch status {
	case entitytypes.RuntimeDeployTaskStatusPending,
		entitytypes.RuntimeDeployTaskStatusRunning,
		entitytypes.RuntimeDeployTaskStatusSucceeded,
		entitytypes.RuntimeDeployTaskStatusFailed,
		entitytypes.RuntimeDeployTaskStatusCanceled:
		return status, nil
	default:
		return "", fmt.Errorf("unknown runtime deploy task status %q", raw)
	}
}

func parseTerminalStatusSource(raw string) entitytypes.RuntimeDeployTaskTerminalStatusSource {
	switch source := entitytypes.RuntimeDeployTaskTerminalStatusSource(strings.TrimSpace(raw)); source {
	case entitytypes.RuntimeDeployTaskTerminalStatusSourceWorker,
		entitytypes.RuntimeDeployTaskTerminalStatusSourceOperator,
		entitytypes.RuntimeDeployTaskTerminalStatusSourceSystem:
		return source
	default:
		return ""
	}
}

func normalizeUpsertParams(params domainrepo.UpsertDesiredParams) (domainrepo.UpsertDesiredParams, error) {
	params.RunID = strings.TrimSpace(params.RunID)
	if params.RunID == "" {
		return domainrepo.UpsertDesiredParams{}, fmt.Errorf("upsert runtime deploy task: run_id is required")
	}
	params.RuntimeMode = strings.TrimSpace(params.RuntimeMode)
	if params.RuntimeMode == "" {
		params.RuntimeMode = "full-env"
	}
	params.Namespace = strings.TrimSpace(params.Namespace)
	params.TargetEnv = strings.TrimSpace(params.TargetEnv)
	if params.TargetEnv == "" {
		params.TargetEnv = "ai"
	}
	if params.SlotNo < 0 {
		params.SlotNo = 0
	}
	params.RepositoryFullName = strings.TrimSpace(params.RepositoryFullName)
	params.ServicesYAMLPath = strings.TrimSpace(params.ServicesYAMLPath)
	params.BuildRef = strings.TrimSpace(params.BuildRef)
	return params, nil
}

func normalizeRequestActionParams(params domainrepo.RequestActionParams) (domainrepo.RequestActionParams, error) {
	params.RunID = strings.TrimSpace(params.RunID)
	if params.RunID == "" {
		return domainrepo.RequestActionParams{}, fmt.Errorf("request runtime deploy action: run_id is required")
	}
	switch params.Action {
	case querytypes.RuntimeDeployTaskActionCancel, querytypes.RuntimeDeployTaskActionStop:
	default:
		return domainrepo.RequestActionParams{}, fmt.Errorf("request runtime deploy action: action must be cancel or stop")
	}
	params.RequestedBy = strings.TrimSpace(params.RequestedBy)
	if params.RequestedBy == "" {
		return domainrepo.RequestActionParams{}, fmt.Errorf("request runtime deploy action: requested_by is required")
	}
	params.Reason = strings.TrimSpace(params.Reason)
	if params.RequestedAt.IsZero() {
		params.RequestedAt = time.Now().UTC()
	} else {
		params.RequestedAt = params.RequestedAt.UTC()
	}
	if len(params.Reason) > 4000 {
		params.Reason = params.Reason[:4000]
	}
	return params, nil
}

func shouldResetDesired(existing domainrepo.Task, params domainrepo.UpsertDesiredParams) bool {
	if existing.Status == entitytypes.RuntimeDeployTaskStatusFailed {
		return true
	}
	return !sameDesired(existing, params)
}

func sameDesired(existing domainrepo.Task, params domainrepo.UpsertDesiredParams) bool {
	if strings.TrimSpace(existing.RuntimeMode) != strings.TrimSpace(params.RuntimeMode) {
		return false
	}
	if strings.TrimSpace(existing.Namespace) != strings.TrimSpace(params.Namespace) {
		return false
	}
	if strings.TrimSpace(existing.TargetEnv) != strings.TrimSpace(params.TargetEnv) {
		return false
	}
	if existing.SlotNo != params.SlotNo {
		return false
	}
	if strings.TrimSpace(existing.RepositoryFullName) != strings.TrimSpace(params.RepositoryFullName) {
		return false
	}
	if strings.TrimSpace(existing.ServicesYAMLPath) != strings.TrimSpace(params.ServicesYAMLPath) {
		return false
	}
	if strings.TrimSpace(existing.BuildRef) != strings.TrimSpace(params.BuildRef) {
		return false
	}
	if existing.DeployOnly != params.DeployOnly {
		return false
	}
	return true
}

func shouldCancelSupersededDeployOnly(task domainrepo.Task, params domainrepo.UpsertDesiredParams) bool {
	if !task.DeployOnly || !params.DeployOnly {
		return false
	}
	if strings.TrimSpace(params.BuildRef) == "" {
		return false
	}
	if strings.TrimSpace(params.TargetEnv) == "" {
		return false
	}
	if strings.TrimSpace(params.RepositoryFullName) == "" {
		return false
	}
	return true
}

func cancelSupersededDeployOnlyTasks(ctx context.Context, tx pgx.Tx, currentRunID string, params domainrepo.UpsertDesiredParams) (int64, error) {
	reason := fmt.Sprintf("superseded by newer deploy task run_id=%s build_ref=%s", currentRunID, strings.TrimSpace(params.BuildRef))
	if len(reason) > 4000 {
		reason = reason[:4000]
	}
	tag, err := tx.Exec(
		ctx,
		queryCancelSupersededDeployOnly,
		currentRunID,
		strings.TrimSpace(params.RepositoryFullName),
		strings.TrimSpace(params.TargetEnv),
		strings.TrimSpace(params.Namespace),
		params.SlotNo,
		strings.TrimSpace(params.BuildRef),
		reason,
	)
	if err != nil {
		return 0, fmt.Errorf("cancel superseded deploy-only tasks for run_id=%s: %w", currentRunID, err)
	}
	return tag.RowsAffected(), nil
}

func requestAction(ctx context.Context, tx pgx.Tx, params domainrepo.RequestActionParams) (domainrepo.Task, error) {
	row := tx.QueryRow(
		ctx,
		queryRequestAction,
		params.RunID,
		string(params.Action),
		params.RequestedAt,
		params.RequestedBy,
		params.Reason,
		params.Reason,
	)
	task, err := scanTask(row)
	if err != nil {
		return domainrepo.Task{}, fmt.Errorf("request runtime deploy action run_id=%s: %w", params.RunID, err)
	}
	return task, nil
}
