package mysql

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

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"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"
)

// readWorkflowPollState 校验当前激活的完整轮次与原预算；时钟和最终信号共用相同检查。
func readWorkflowPollState(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep) (weave.PollState, core.ID, error) {
	p, last, err := readWorkflowPollBudget(ctx, tx, g, row)
	if err == nil && p.ResolvedAt != nil {
		err = errors.New("unsettled poll step has resolved budget")
	}
	return p, last, err
}

// readWorkflowPollBudget 也允许读取已结清预算，供有证据的人工恢复核对；不重新计时。
func readWorkflowPollBudget(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep) (weave.PollState, core.ID, error) {
	var p weave.PollState
	err := tx.QueryRowContext(ctx, `SELECT id,revision,max_rounds,used_rounds,first_round,first_round_attempts_used,max_attempts_per_round,interval_ms,deadline_at,next_poll_at,current_task_id,resolved_at FROM weave_poll_states WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, g.view.NamespaceID, g.view.ID, row.view.ActivationID).Scan(&p.ID, &p.Revision, &p.MaxRounds, &p.UsedRounds, &p.FirstRound, &p.FirstRoundAttemptsUsed, &p.MaxAttemptsPerRound, &p.IntervalMS, &p.DeadlineAt, &p.NextPollAt, &p.CurrentTaskID, &p.ResolvedAt)
	if err != nil {
		return p, "", err
	}
	base, err := workflowActivationOrigin(ctx, tx, g, row)
	if err != nil {
		return p, "", err
	}
	node, _ := g.program.Step(row.view.Key)
	policy := node.Poll
	if policy == nil || p.MaxRounds != policy.MaxRounds || p.MaxAttemptsPerRound != policy.Task.Retry.MaxAttempts || p.IntervalMS != policy.IntervalMS || p.UsedRounds < 1 || p.UsedRounds > p.MaxRounds || p.FirstRound < 1 || p.FirstRound > p.UsedRounds || p.FirstRoundAttemptsUsed >= p.MaxAttemptsPerRound || !p.DeadlineAt.Equal(minimumTime(g.view.DeadlineAt, base.started.Add(time.Duration(policy.TimeoutMS)*time.Millisecond))) {
		return p, "", errors.New("workflow poll policy mismatch")
	}
	if base.sequence == 1 && (p.FirstRound != 1 || p.FirstRoundAttemptsUsed != 0) {
		return p, "", errors.New("initial poll recovery budget invalid")
	}
	if base.previous != nil {
		var used uint32
		var deadline time.Time
		var settled *time.Time
		if err := tx.QueryRowContext(ctx, `SELECT used_rounds,deadline_at,resolved_at FROM weave_poll_states WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, g.view.NamespaceID, g.view.ID, base.previous).Scan(&used, &deadline, &settled); err != nil {
			return p, "", err
		}
		if used != p.FirstRound || !deadline.Equal(p.DeadlineAt) || settled == nil {
			return p, "", errors.New("poll recovery parent budget invalid")
		}
		if _, err := readWorkflowPollRoundHistory(ctx, tx, g, row, p, p.FirstRound); err != nil {
			return p, "", err
		}
	}
	var count, minRound, maxRound uint32
	if err := tx.QueryRowContext(ctx, "SELECT COUNT(*),COALESCE(MIN(`round`),0),COALESCE(MAX(`round`),0) FROM weave_poll_rounds WHERE namespace_id=? AND workflow_id=? AND activation_id=?", g.view.NamespaceID, g.view.ID, row.view.ActivationID).Scan(&count, &minRound, &maxRound); err != nil {
		return p, "", err
	}
	if count != p.UsedRounds-p.FirstRound+1 || minRound != p.FirstRound || maxRound != p.UsedRounds {
		return p, "", errors.New("workflow poll rounds incomplete")
	}
	var lastTask core.ID
	if err := tx.QueryRowContext(ctx, "SELECT task_id FROM weave_poll_rounds WHERE namespace_id=? AND workflow_id=? AND activation_id=? AND `round`=?", g.view.NamespaceID, g.view.ID, row.view.ActivationID, p.UsedRounds).Scan(&lastTask); err != nil {
		return p, "", err
	}
	if p.CurrentTaskID != nil && (*p.CurrentTaskID != lastTask || p.NextPollAt != nil) || p.ResolvedAt != nil && (p.CurrentTaskID != nil || p.NextPollAt != nil) {
		return p, "", errors.New("poll budget execution pointers invalid")
	}
	return p, lastTask, nil
}

func (r *WorkflowRuntime) advancePoll(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, c workflowCandidate, allow bool, now time.Time) (bool, error) {
	p, lastTask, err := readWorkflowPollState(ctx, tx, g, row)
	if err != nil {
		return false, err
	}
	node, _ := g.program.Step(row.view.Key)
	policy := node.Poll
	if p.CurrentTaskID != nil {
		if *p.CurrentTaskID != lastTask || p.NextPollAt != nil || row.view.State != weave.StepActive {
			return false, errors.New("workflow in-flight poll mismatch")
		}
		if err := validatePollTaskOwner(ctx, tx, g, row, lastTask); err != nil {
			return false, err
		}
		if now.Before(p.DeadlineAt) {
			return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, p.DeadlineAt)
		}
		task, err := readTask(ctx, tx, c.ns, lastTask, false)
		if err != nil {
			return false, err
		}
		if task.QueueID != policy.Task.QueueID {
			return false, errors.New("workflow in-flight poll queue mismatch")
		}
		if task.State.Terminal() || task.State == "NEEDS_REVIEW" {
			// 原结果已可靠落库时，先让Inbox按事件时间裁决，不能用扫描器的当前时间覆盖它。
			var available time.Time
			err := tx.QueryRowContext(ctx, `SELECT available_at FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND source_activation_id=? AND state='PENDING' ORDER BY source_revision DESC LIMIT 1`, c.ns, c.workflow, lastTask, row.view.ActivationID).Scan(&available)
			if err != nil {
				return false, err
			}
			if available.Before(now.Add(time.Second)) {
				available = now.Add(time.Second)
			}
			return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, available)
		}
		// 到期但仍在途时先关闭门控，保留当前轮次和ACTIVE步骤继续接收Pulse结清结果。
		changed := g.view.State != weave.InstanceNeedsReview
		if changed {
			if err := holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, "POLL_INFLIGHT_DEADLINE", now); err != nil {
				return false, err
			}
			if err := r.audit(ctx, tx, g, c.id, "POLL_INFLIGHT_DEADLINE", now); err != nil {
				return false, err
			}
		}
		return changed, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, now.Add(time.Second))
	}
	if row.view.State != weave.StepWaiting || p.NextPollAt == nil {
		return false, errors.New("workflow poll continuation missing")
	}
	if !now.Before(p.DeadlineAt) || p.UsedRounds >= p.MaxRounds {
		reason := "POLL_DEADLINE"
		if p.UsedRounds >= p.MaxRounds {
			reason = "POLL_ROUNDS_EXHAUSTED"
		}
		if _, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET resolved_at=?,next_poll_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, now, c.ns, p.ID); err != nil {
			return false, err
		}
		if _, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='EXPIRED',resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND workflow_id=? AND activation_id=? AND state='WAITING'`, now, now, c.ns, c.workflow, row.view.ActivationID); err != nil {
			return false, err
		}
		if err := saveWorkflowResultStep(ctx, tx, g, row, weave.StepNeedsReview, nil, now); err != nil {
			return false, err
		}
		if err := deleteWorkflowClock(ctx, tx, g, row); err != nil {
			return false, err
		}
		if err := holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, reason, now); err != nil {
			return false, err
		}
		return true, r.audit(ctx, tx, g, c.id, reason, now)
	}
	if now.Before(*p.NextPollAt) {
		return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, *p.NextPollAt)
	}
	if !allow {
		return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, minimumTime(p.DeadlineAt, now.Add(time.Second)))
	}
	round, err := p.NextRound(now)
	if err != nil {
		return false, err
	}
	// 只有上一轮成功查询并已分类为PENDING才可以消耗新轮次，不能用时间推进绕过失败预算。
	if err := verifyWorkflowPollPending(ctx, tx, g, row, p, lastTask, now); err != nil {
		return false, err
	}
	in, err := pollTaskInput(ctx, tx, g, row, policy.Task)
	if err != nil {
		return false, err
	}
	taskID, err := core.NewID()
	if err != nil {
		return false, err
	}
	task, err := insertAcceptedTaskUntil(ctx, tx, c.ns, taskID, in, nil, &p.DeadlineAt)
	if errors.Is(err, fault.QueueFull) || errors.Is(err, fault.ApplicationFull) {
		return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, minimumTime(p.DeadlineAt, now.Add(time.Second)))
	}
	if err != nil {
		return false, 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, row.view.ActivationID, g.gate, c.ns, taskID)
	if err := exactlyOne(res, err); err != nil {
		return false, err
	}
	roundID, err := core.NewID()
	if err != nil {
		return false, err
	}
	if _, err := tx.ExecContext(ctx, "INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id) VALUES (?,?,?,?,?,?)", c.ns, roundID, c.workflow, row.view.ActivationID, round, taskID); err != nil {
		return false, err
	}
	res, err = tx.ExecContext(ctx, `UPDATE weave_poll_states SET used_rounds=?,current_task_id=?,next_poll_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_task_id IS NULL`, round, taskID, now, c.ns, p.ID, p.Revision)
	if err := exactlyOne(res, err); err != nil {
		return false, err
	}
	if err := saveWorkflowResultStep(ctx, tx, g, row, weave.StepActive, nil, now); err != nil {
		return false, err
	}
	if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, p.DeadlineAt); err != nil {
		return false, err
	}
	if err := emitEvent(ctx, tx, c.ns, taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": taskID, "state": task.State, "workflow_id": c.workflow, "activation_id": row.view.ActivationID, "round": round}); err != nil {
		return false, err
	}
	// 新查询已持久化，实例从纯等待恢复RUNNING；组合推进仍受同一个32步预算。
	if err := r.progress(ctx, tx, g, allow, now); err != nil {
		return false, err
	}
	return true, r.audit(ctx, tx, g, c.id, "POLL_ROUND_ACCEPTED", now)
}

func validatePollTaskOwner(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, id core.ID) error {
	var owner, gate *core.ID
	var kind sql.NullString
	if err := tx.QueryRowContext(ctx, `SELECT owner_kind,owner_id,gate_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, g.view.NamespaceID, id).Scan(&kind, &owner, &gate); err != nil {
		return err
	}
	if kind.String != engine.TaskOwnerKind || owner == nil || row.view.ActivationID == nil || *owner != *row.view.ActivationID || gate == nil || *gate != g.gate {
		return errors.New("poll task ownership invalid")
	}
	return nil
}

// pollTaskInput 复用首轮已经解析并冻结的输入、业务键及互斥键；不重算依赖可能已改变的表达式。
func pollTaskInput(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, spec definition.Task) (service.SubmitInput, error) {
	origin, err := workflowActivationOrigin(ctx, tx, g, row)
	if err != nil {
		return service.SubmitInput{}, err
	}
	originRow := *row
	originRow.view.ActivationID = &origin.id
	var first core.ID
	if err := tx.QueryRowContext(ctx, "SELECT task_id FROM weave_poll_rounds WHERE namespace_id=? AND workflow_id=? AND activation_id=? AND `round`=1", g.view.NamespaceID, g.view.ID, origin.id).Scan(&first); err != nil {
		return service.SubmitInput{}, err
	}
	if err := validatePollTaskOwner(ctx, tx, g, &originRow, first); err != nil {
		return service.SubmitInput{}, err
	}
	base, err := readTask(ctx, tx, g.view.NamespaceID, first, false)
	if err != nil {
		return service.SubmitInput{}, err
	}
	input, err := jsonvalue.Canonical(base.Payload)
	if err != nil {
		return service.SubmitInput{}, err
	}
	activationInput, err := jsonvalue.Canonical(row.input)
	if err != nil {
		return service.SubmitInput{}, err
	}
	if !bytes.Equal(input, activationInput) || definition.ValidateData(spec.InputType, base.Payload) != nil || base.QueueID != spec.QueueID || base.Handler != spec.Handler || base.HandlerVersion != spec.HandlerVersion || base.PayloadVersion != spec.PayloadVersion || base.Retry != spec.Retry || base.Priority != spec.Priority || base.RecoverySafety != spec.RecoverySafety {
		return service.SubmitInput{}, errors.New("poll frozen task parameters mismatch")
	}
	return service.NormalizeSubmit(service.SubmitInput{QueueID: base.QueueID, Handler: base.Handler, HandlerVersion: base.HandlerVersion, PayloadVersion: base.PayloadVersion, Payload: base.Payload, BusinessKey: base.BusinessKey, ConcurrencyKey: base.ConcurrencyKey, Priority: base.Priority, TaskTimeoutMS: spec.TimeoutMS, Retry: base.Retry, RecoverySafety: base.RecoverySafety})
}