package mysql

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

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/pulse/service"
	"git.yuntuo.hk/k2safe/orbit/weave/engine"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
	"git.yuntuo.hk/k2safe/orbit/weave/recovery"
)

// retryWorkflowPoll 重新执行当前业务轮；新激活保留旧轮次、同轮尝试、精确输入及两个绝对截止。
func retryWorkflowPoll(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, in recovery.RetryStepInput, now time.Time, out *recovery.Result) (map[string]any, error) {
	p, last, err := readWorkflowPollBudget(ctx, tx, g, row)
	if err != nil {
		return nil, err
	}
	origin, err := workflowActivationOrigin(ctx, tx, g, row)
	if err != nil {
		return nil, err
	}
	if !now.Before(p.DeadlineAt) || origin.sequence >= p.MaxRounds*p.MaxAttemptsPerRound {
		return nil, fault.Conflict
	}
	h, err := readWorkflowPollRoundHistory(ctx, tx, g, row, p, p.UsedRounds)
	if err != nil {
		return nil, err
	}
	if h.last.ID != last || !h.last.State.Terminal() || h.last.EffectUncertain || h.used >= p.MaxAttemptsPerRound || h.count >= p.MaxAttemptsPerRound || !now.Before(h.first.DeadlineAt) {
		return nil, fault.Conflict
	}
	// 已确认的PENDING应按间隔和业务轮次预算推进，不能借人工恢复再查同一轮。
	if h.last.State == "SUCCEEDED" {
		kind, err := g.program.ClassifyPoll(row.view.Key, h.last.Result)
		if err == nil && kind == engine.PollPending {
			return nil, fault.Conflict
		}
	}
	node, _ := g.program.Step(row.view.Key)
	var wait *workflowSignalWait
	if node.Poll.FinalSignal != nil {
		w, err := readWorkflowSignalWait(ctx, tx, g, row)
		if err != nil {
			return nil, err
		}
		if !now.Before(w.deadline) {
			return nil, fault.Conflict
		}
		var pending int
		if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_signals WHERE namespace_id=? AND workflow_id=? AND target_step_key=? AND target_activation_sequence=? AND state='PENDING'`, g.view.NamespaceID, g.view.ID, row.view.Key, w.sequence).Scan(&pending); err != nil {
			return nil, err
		}
		if pending != 0 {
			return nil, fault.Conflict
		}
		if w.winner != nil {
			if _, err := selectedWorkflowSignal(ctx, tx, g, row, w, now); err != nil {
				return nil, err
			}
		}
		wait = &w
	}
	policy := h.first.Retry
	policy.MaxAttempts = p.MaxAttemptsPerRound - h.used
	input, err := service.NormalizeSubmit(service.SubmitInput{QueueID: h.first.QueueID, Handler: h.first.Handler, HandlerVersion: h.first.HandlerVersion, PayloadVersion: h.first.PayloadVersion, Payload: h.first.Payload, BusinessKey: h.first.BusinessKey, ConcurrencyKey: h.first.ConcurrencyKey, Priority: h.first.Priority, RecoverySafety: h.first.RecoverySafety, Retry: policy, TaskTimeoutMS: node.Poll.Task.TimeoutMS})
	if err != nil {
		return nil, err
	}
	activation, err := core.NewID()
	if err != nil {
		return nil, err
	}
	taskID, err := core.NewID()
	if err != nil {
		return nil, err
	}
	task, err := insertAcceptedTaskUntil(ctx, tx, g.view.NamespaceID, taskID, input, &last, &h.first.DeadlineAt)
	if err != nil {
		return nil, err
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,input,created_at,updated_at) VALUES (?,?,?,?,?,'ACTIVE',?,?,?)`, g.view.NamespaceID, activation, g.view.ID, row.view.ID, origin.sequence+1, string(row.input), now, now); err != nil {
		return nil, err
	}
	result, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET owner_kind=?,owner_id=?,gate_id=? WHERE namespace_id=? AND id=? AND owner_kind IS NULL`, engine.TaskOwnerKind, activation, g.gate, g.view.NamespaceID, taskID)
	if err := exactlyOne(result, err); err != nil {
		return nil, err
	}
	// 旧预算仅结清检查点；旧任务、旧激活输出、轮次及已匹配信号不改写。
	if _, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET current_task_id=NULL,next_poll_at=NULL,resolved_at=COALESCE(resolved_at,?),revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, now, g.view.NamespaceID, p.ID); err != nil {
		return nil, err
	}
	pollID, err := core.NewID()
	if err != nil {
		return nil, err
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO weave_poll_states(namespace_id,id,workflow_id,activation_id,max_rounds,used_rounds,first_round,first_round_attempts_used,max_attempts_per_round,interval_ms,deadline_at,current_task_id,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, g.view.NamespaceID, pollID, g.view.ID, activation, p.MaxRounds, p.UsedRounds, p.UsedRounds, h.used, p.MaxAttemptsPerRound, p.IntervalMS, p.DeadlineAt, taskID, now, now); err != nil {
		return nil, err
	}
	roundID, err := core.NewID()
	if err != nil {
		return nil, err
	}
	if _, err := tx.ExecContext(ctx, "INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id) VALUES (?,?,?,?,?,?)", g.view.NamespaceID, roundID, g.view.ID, activation, p.UsedRounds, taskID); err != nil {
		return nil, err
	}
	if wait != nil {
		if _, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='CANCELLED',resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state='WAITING'`, now, now, g.view.NamespaceID, wait.id); err != nil {
			return nil, err
		}
		id, err := core.NewID()
		if err != nil {
			return nil, err
		}
		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at,signal_name,payload_version,created_at,updated_at) VALUES (?,?,?,?,'POLL',?,?,?,?,?,?)`, g.view.NamespaceID, id, g.view.ID, activation, []byte(wait.correlation), wait.deadline, wait.contract.Name, wait.contract.PayloadVersion, now, now); err != nil {
			return nil, err
		}
	}
	result, err = tx.ExecContext(ctx, `UPDATE weave_steps SET state='ACTIVE',current_activation_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_activation_id=?`, activation, now, g.view.NamespaceID, row.view.ID, row.view.Revision, row.view.ActivationID)
	if err := exactlyOne(result, err); err != nil {
		return nil, err
	}
	if row.view.TaskID != nil {
		return nil, errors.New("poll cannot have direct task reference")
	}
	row.view.State, row.view.ActivationID = weave.StepActive, &activation
	row.view.ActivationSequence = origin.sequence + 1
	row.view.Revision++
	row.output = nil
	if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, p.DeadlineAt); err != nil {
		return nil, err
	}
	if err := emitEvent(ctx, tx, g.view.NamespaceID, taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": taskID, "state": task.State, "workflow_id": g.view.ID, "activation_id": activation, "retry_of": last, "round": p.UsedRounds}); err != nil {
		return nil, err
	}
	if err := emitEvent(ctx, tx, g.view.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": g.view.ID, "activation_id": activation, "state": row.view.State}); err != nil {
		return nil, err
	}
	out.Step, out.Task = &row.view, &task
	return map[string]any{"step_id": row.view.ID, "previous_activation_id": in.ActivationID, "activation_id": activation, "sequence": origin.sequence + 1, "retry_of": last, "task_id": taskID, "safe_to_retry": true, "round": p.UsedRounds, "max_rounds": p.MaxRounds, "used_attempts": h.used, "remaining_attempts": policy.MaxAttempts, "deadline_at": p.DeadlineAt, "task_deadline_at": h.first.DeadlineAt}, nil
}