package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"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/weave/engine"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
	"git.yuntuo.hk/k2safe/orbit/weave/recovery"
)

// processedWorkflowPollResult 只接受当前最后任务的已处理权威结果，不把人工声明当作PENDING事实。
func processedWorkflowPollResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, last core.ID, now time.Time) (*workflowResult, error) {
	task, err := readTask(ctx, tx, g.view.NamespaceID, last, false)
	if err != nil {
		return nil, err
	}
	if task.State != "SUCCEEDED" || task.EffectUncertain {
		return nil, fault.Conflict
	}
	var id core.ID
	if err := tx.QueryRowContext(ctx, `SELECT id FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND source_activation_id=? AND source_revision=? AND state='PROCESSED'`, g.view.NamespaceID, g.view.ID, last, row.view.ActivationID, task.Revision).Scan(&id); err != nil {
		return nil, err
	}
	source, err := readWorkflowResultState(ctx, tx, g, workflowCandidate{ns: g.view.NamespaceID, workflow: g.view.ID, id: id}, now, "PROCESSED")
	if err != nil {
		return nil, err
	}
	if source == nil || source.task != task.ID || source.activation != *row.view.ActivationID || source.step != row || source.revision != task.Revision || source.snapshot.State != "SUCCEEDED" || source.snapshot.EffectUncertain || !sameWorkflowJSON(source.snapshot.Result, task.Result) {
		return nil, errors.New("reviewed poll result proof mismatch")
	}
	kind, err := g.program.ClassifyPoll(row.view.Key, source.snapshot.Result)
	if err != nil || kind != engine.PollPending {
		return nil, fault.Conflict
	}
	return source, nil
}

func noPendingWorkflowSignals(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, sequence uint32) error {
	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, sequence).Scan(&pending); err != nil {
		return err
	}
	if pending != 0 {
		return fault.Conflict
	}
	return nil
}

// ContinuePoll 为已核实的PENDING恢复原等待，不重查同轮、不重置尝试，也不自动解除实例核查。
func (s *WorkflowRecoveryStore) ContinuePoll(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.ContinuePollInput) (recovery.Result, error) {
	return s.mutate(ctx, t, id, key, "workflow.poll.continue", 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
		}
		node, ok := g.program.Step(row.view.Key)
		if !ok || node.Poll == nil {
			return nil, fault.Invalid("continue poll requires a POLL step")
		}
		if g.view.StopKind != nil || g.view.CancelRequestedAt != nil || !now.Before(g.view.DeadlineAt) {
			return nil, fault.Conflict
		}
		if err := workflowStepSettled(ctx, tx, g, row); err != nil {
			return nil, err
		}
		p, last, err := readWorkflowPollBudget(ctx, tx, g, row)
		if err != nil {
			return nil, err
		}
		if p.CurrentTaskID != nil || p.UsedRounds >= p.MaxRounds || !now.Before(p.DeadlineAt) {
			return nil, fault.Conflict
		}
		if _, err := readWorkflowPollRoundHistory(ctx, tx, g, row, p, p.UsedRounds); err != nil {
			return nil, err
		}
		source, err := processedWorkflowPollResult(ctx, tx, g, row, last, now)
		if err != nil {
			return nil, err
		}
		if !source.occurred.Before(p.DeadlineAt) {
			return nil, fault.Conflict
		}
		next := minimumTime(source.occurred.Add(time.Duration(p.IntervalMS)*time.Millisecond), p.DeadlineAt)
		meta := map[string]any{"step_id": row.view.ID, "activation_id": row.view.ActivationID, "previous_state": row.view.State, "previous_output": json.RawMessage(row.output), "previous_resolved_at": p.ResolvedAt, "task_id": last, "source_event_id": source.event, "source_revision": source.revision, "round": p.UsedRounds, "next_poll_at": next, "deadline_at": p.DeadlineAt, "pending_confirmed": true}
		if node.Poll.FinalSignal != nil {
			w, err := readWorkflowSignalWait(ctx, tx, g, row)
			if err != nil {
				return nil, err
			}
			if w.winner != nil || w.state != "WAITING" && w.state != "CANCELLED" || !now.Before(w.deadline) {
				return nil, fault.Conflict
			}
			if err := noPendingWorkflowSignals(ctx, tx, g, row, w.sequence); err != nil {
				return nil, err
			}
			meta["wait_id"], meta["previous_wait_state"] = w.id, w.state
			result, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='WAITING',resolved_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state=? AND winner_signal_id IS NULL`, now, t.NamespaceID, w.id, w.state)
			if err := exactlyOne(result, err); err != nil {
				return nil, err
			}
		}
		result, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET next_poll_at=?,resolved_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_task_id IS NULL`, next, now, t.NamespaceID, p.ID, p.Revision)
		if err := exactlyOne(result, err); err != nil {
			return nil, err
		}
		if err := saveWorkflowResultStep(ctx, tx, g, row, weave.StepWaiting, nil, now); err != nil {
			return nil, err
		}
		if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, next); err != nil {
			return nil, err
		}
		meta["step_revision"] = row.view.Revision
		out.Step = &row.view
		return meta, nil
	})
}

// verifyWorkflowPollPending 保留原Inbox处置；人工接续必须有匹配当前快照的不可变审计证明。
func verifyWorkflowPollPending(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, p weave.PollState, last core.ID, now time.Time) error {
	var disposition string
	if err := tx.QueryRowContext(ctx, `SELECT error_code FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND source_activation_id=? AND state='PROCESSED' ORDER BY source_revision DESC LIMIT 1`, g.view.NamespaceID, g.view.ID, last, row.view.ActivationID).Scan(&disposition); err != nil {
		return err
	}
	if disposition == "POLL_PENDING" {
		return nil
	}
	source, err := processedWorkflowPollResult(ctx, tx, g, row, last, now)
	if err != nil {
		return err
	}
	var raw []byte
	if err := tx.QueryRowContext(ctx, "SELECT `after` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow' AND resource_id=? AND action='workflow.poll.continue' AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.step_id'))=? AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.activation_id'))=? AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.task_id'))=? ORDER BY created_at DESC,id DESC LIMIT 1", g.view.NamespaceID, g.view.ID, string(row.view.ID), string(*row.view.ActivationID), string(last)).Scan(&raw); err != nil {
		return err
	}
	var proof struct {
		StepID           core.ID   `json:"step_id"`
		ActivationID     core.ID   `json:"activation_id"`
		TaskID           core.ID   `json:"task_id"`
		SourceEventID    core.ID   `json:"source_event_id"`
		SourceRevision   uint64    `json:"source_revision"`
		StepRevision     uint64    `json:"step_revision"`
		Round            uint32    `json:"round"`
		NextPollAt       time.Time `json:"next_poll_at"`
		DeadlineAt       time.Time `json:"deadline_at"`
		PendingConfirmed bool      `json:"pending_confirmed"`
	}
	expected := minimumTime(source.occurred.Add(time.Duration(p.IntervalMS)*time.Millisecond), p.DeadlineAt)
	if json.Unmarshal(raw, &proof) != nil || proof.StepID != row.view.ID || proof.ActivationID != *row.view.ActivationID || proof.TaskID != last || proof.SourceEventID != source.event || proof.SourceRevision != source.revision || proof.StepRevision != row.view.Revision || proof.Round != p.UsedRounds || !proof.PendingConfirmed || !proof.DeadlineAt.Equal(p.DeadlineAt) || p.NextPollAt == nil || !proof.NextPollAt.Equal(*p.NextPollAt) || !proof.NextPollAt.Equal(expected) {
		return errors.New("poll continuation audit proof invalid")
	}
	return nil
}