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/platform/auth"
	"git.yuntuo.hk/k2safe/orbit/pulse/service"
	"git.yuntuo.hk/k2safe/orbit/weave/definition"
	"git.yuntuo.hk/k2safe/orbit/weave/engine"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
	"git.yuntuo.hk/k2safe/orbit/weave/recovery"
)

// RetryStep 为TASK或POLL建立新的执行激活；保留原输入、绝对截止和跨激活尝试预算。
func (s *WorkflowRecoveryStore) RetryStep(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.RetryStepInput) (recovery.Result, error) {
	return s.mutate(ctx, t, id, key, "workflow.step.retry", in, in.Decision, func(tx *sql.Tx, g *workflowGraph, _ core.ID, now time.Time, out *recovery.Result) (map[string]any, error) {
		row, err := workflowRecoveryStep(g, in.StepDecision)
		if err != nil {
			return nil, err
		}
		if g.view.StopKind != nil || g.view.CancelRequestedAt != nil || !now.Before(g.view.DeadlineAt) {
			return nil, fault.Conflict
		}
		node, ok := g.program.Step(row.view.Key)
		if !ok || node.Task == nil && node.Poll == nil {
			return nil, fault.Invalid("retry currently supports TASK and POLL steps")
		}
		if err := workflowStepSettled(ctx, tx, g, row); err != nil {
			return nil, err
		}
		if node.Poll != nil {
			return retryWorkflowPoll(ctx, tx, g, row, in, now, out)
		}
		var sequence uint32
		if err := tx.QueryRowContext(ctx, `SELECT sequence FROM weave_activations WHERE namespace_id=? AND id=?`, t.NamespaceID, row.view.ActivationID).Scan(&sequence); err != nil {
			return nil, err
		}
		// 人工取消尚未开始的任务也不能无限创建激活；每个静态TASK的激活数不超过原max_attempts。
		if sequence == 0 || sequence >= node.Task.Retry.MaxAttempts {
			return nil, fault.Conflict
		}
		history, err := tx.QueryContext(ctx, `SELECT id,sequence,task_id FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? ORDER BY sequence LIMIT 101`, t.NamespaceID, id, row.view.ID)
		if err != nil {
			return nil, err
		}
		type prior struct {
			activation, task core.ID
			sequence         uint32
		}
		var previous []prior
		for history.Next() {
			var v prior
			if err := history.Scan(&v.activation, &v.sequence, &v.task); err != nil {
				history.Close()
				return nil, err
			}
			previous = append(previous, v)
		}
		err = history.Err()
		history.Close()
		if err != nil {
			return nil, err
		}
		if len(previous) != int(sequence) || previous[len(previous)-1].activation != in.ActivationID {
			return nil, errors.New("workflow retry activation history incomplete")
		}
		var first, last service.TaskView
		var used uint32
		for i, prior := range previous {
			if prior.sequence != uint32(i+1) {
				return nil, errors.New("workflow retry sequence invalid")
			}
			v, err := readTask(ctx, tx, t.NamespaceID, prior.task, false)
			if err != nil {
				return nil, err
			}
			if (v.State != "FAILED" && v.State != "CANCELLED") || v.EffectUncertain {
				return nil, fault.Conflict
			}
			if i == 0 {
				first = v
				if definition.ValidateData(node.Task.InputType, first.Payload) != nil || !first.DeadlineAt.Equal(minimumTime(first.CreatedAt.Add(time.Duration(node.Task.TimeoutMS)*time.Millisecond), g.view.DeadlineAt)) {
					return nil, errors.New("workflow retry initial contract or deadline mismatch")
				}
			}
			expectedPolicy := node.Task.Retry
			if used >= expectedPolicy.MaxAttempts {
				return nil, fault.Conflict
			}
			expectedPolicy.MaxAttempts -= used
			if v.Retry != expectedPolicy || v.AttemptCount > v.Retry.MaxAttempts || v.QueueID != node.Task.QueueID || v.Handler != node.Task.Handler || v.HandlerVersion != node.Task.HandlerVersion || v.PayloadVersion != node.Task.PayloadVersion || v.RecoverySafety != node.Task.RecoverySafety || v.Priority != node.Task.Priority || !sameWorkflowJSON(v.Payload, row.input) || !v.DeadlineAt.Equal(first.DeadlineAt) || v.BusinessKey != first.BusinessKey || v.ConcurrencyKey != first.ConcurrencyKey {
				return nil, errors.New("workflow retry frozen task history mismatch")
			}
			if i == 0 && v.RetryOf != nil || i > 0 && (v.RetryOf == nil || *v.RetryOf != last.ID) {
				return nil, errors.New("workflow retry task ancestry mismatch")
			}
			var attempts uint32
			if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_attempts WHERE namespace_id=? AND task_id=?`, t.NamespaceID, v.ID).Scan(&attempts); err != nil {
				return nil, err
			}
			if attempts != v.AttemptCount {
				return nil, errors.New("workflow retry attempt history incomplete")
			}
			used += v.AttemptCount
			last = v
		}
		if used >= node.Task.Retry.MaxAttempts || !now.Before(first.DeadlineAt) || first.StartDeadlineAt != nil && !now.Before(*first.StartDeadlineAt) {
			return nil, fault.Conflict
		}
		policy := node.Task.Retry
		policy.MaxAttempts -= used
		input, err := service.NormalizeSubmit(service.SubmitInput{QueueID: first.QueueID, Handler: first.Handler, HandlerVersion: first.HandlerVersion, PayloadVersion: first.PayloadVersion, Payload: first.Payload, BusinessKey: first.BusinessKey, ConcurrencyKey: first.ConcurrencyKey, Priority: first.Priority, RecoverySafety: first.RecoverySafety, Retry: policy, TaskTimeoutMS: node.Task.TimeoutMS, StartDeadlineAt: first.StartDeadlineAt})
		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, t.NamespaceID, taskID, input, &last.ID, &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,task_id,input,created_at,updated_at) VALUES (?,?,?,?,?,'ACTIVE',?,?,?,?)`, t.NamespaceID, activation, id, row.view.ID, sequence+1, taskID, string(row.input), now, now); err != nil {
			return nil, err
		}
		res, 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, t.NamespaceID, taskID)
		if err := exactlyOne(res, err); err != nil {
			return nil, err
		}
		res, 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, t.NamespaceID, row.view.ID, row.view.Revision, row.view.ActivationID)
		if err := exactlyOne(res, err); err != nil {
			return nil, err
		}
		row.view.State, row.view.ActivationID, row.view.TaskID = weave.StepActive, &activation, &taskID
		row.view.ActivationSequence = sequence + 1
		row.view.Revision++
		row.output = nil
		if err := deleteWorkflowClock(ctx, tx, g, row); err != nil {
			return nil, err
		}
		if err := emitEvent(ctx, tx, t.NamespaceID, taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": taskID, "state": task.State, "workflow_id": id, "activation_id": activation, "retry_of": last.ID}); err != nil {
			return nil, err
		}
		if err := emitEvent(ctx, tx, t.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": 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": sequence + 1, "retry_of": last.ID, "task_id": taskID, "safe_to_retry": true, "used_attempts": used, "remaining_attempts": policy.MaxAttempts, "deadline_at": first.DeadlineAt}, nil
	})
}