package orbit

import (
	"context"
	"encoding/json"
	"testing"
	"time"
)

// 此测试由脚本在原服务退出后单独运行；所有流程操作只走公开HTTP/WSS SDK。
// 两旧节点均Wait确认SIGKILL后才等待原时钟到期并启动两新节点。
func TestIntegrationWorkflowProcessPauseRecovery(t *testing.T) {
	a := startOrbitProcess(t)
	b := startOrbitProcess(t)
	_, queue, token := integrationWorkerClient(t)
	api := processClient(t, a, token)
	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Second)
	defer cancel()
	handler := "pause.process." + workerTestID(t)
	spec := workflowSDKSpec(queue, handler)
	businessKey := WorkflowRef("workflow", "business_key")
	spec.Body.Steps[0].Task.BusinessKey = &businessKey
	initial := *spec.Body.Steps[0].Task
	query := initial
	query.Handler = handler + ".query"
	query.ResultType = WorkflowType{Kind: "STRING", Enum: []string{"PENDING", "DONE"}}
	query.TimeoutMS = 20000
	parallel := WorkflowParallel{}
	for _, branch := range []struct {
		step WorkflowStep
		typ  WorkflowType
	}{
		{WorkflowStep{Key: "initial", Kind: "TASK", Task: &initial}, spec.InputType},
		{WorkflowStep{Key: "signal", Kind: "SIGNAL", Signal: &WorkflowSignal{Name: "external.done", PayloadVersion: "v1", Correlation: WorkflowRef("workflow", "business_key"), PayloadType: spec.InputType, TimeoutMS: 30000}}, spec.InputType},
		{WorkflowStep{Key: "timer", Kind: "TIMER", Timer: &WorkflowTimer{DelayMS: 5000}}, WorkflowType{Kind: "NULL"}},
		{WorkflowStep{Key: "query", Kind: "POLL", Poll: &WorkflowPoll{Task: query, MaxRounds: 2, TimeoutMS: 30000, IntervalMS: 5000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}}}, query.ResultType},
	} {
		parallel.Branches = append(parallel.Branches, WorkflowParallelBranch{Name: branch.step.Key, Body: WorkflowBlock{Steps: []WorkflowStep{branch.step}, OutputType: branch.typ, Output: WorkflowRef(branch.step.Key)}})
	}
	spec.Body.Steps = append([]WorkflowStep{{Key: "join", Kind: "PARALLEL", Parallel: &parallel}}, spec.Body.Steps...)
	in := publishWorkflowSDK(t, ctx, api, spec)
	in.BusinessKey = handler
	startKey := workerTestID(t)
	started, err := api.StartWorkflow(ctx, startKey, in)
	if err != nil {
		t.Fatal(err)
	}
	v := started.Instance
	work := workflowWorkerSDK(t, ctx, api, queue, handler)
	queries := workflowWorkerSDK(t, ctx, api, queue, query.Handler)
	first, err := work.NextAssignment(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = work.Start(ctx, workerTestID(t), first.Attempt()); err != nil {
		t.Fatal(err)
	}
	firstQuery := finishWorkflowSDK(t, ctx, queries, WorkerSucceeded, json.RawMessage(`"PENDING"`))
	pollBefore := waitWorkflowStepSDK(t, ctx, api, v.ID, "query", func(d WorkflowStepDetail) bool {
		return d.Poll != nil && d.Poll.UsedRounds == 1 && d.Poll.NextPollAt != nil
	})
	timerBefore, err := api.GetWorkflowStep(ctx, v.ID, "timer")
	if err != nil || timerBefore.Wait == nil || timerBefore.ActivatedAt == nil || timerBefore.Wait.State != "WAITING" {
		t.Fatal("timer not waiting before pause", err)
	}
	signalBefore, err := api.GetWorkflowStep(ctx, v.ID, "signal")
	if err != nil || signalBefore.Wait == nil {
		t.Fatal(err)
	}
	pauseKey := workerTestID(t)
	pauseInput := WorkflowControlRequest{Operation: "PAUSE", ExpectedControlRevision: v.ControlRevision, Reason: "persist pause across all controller exits"}
	paused, err := api.ControlWorkflow(ctx, v.ID, pauseKey, pauseInput)
	if err != nil || !paused.Instance.Paused {
		t.Fatal(err)
	}
	signalKey := workerTestID(t)
	signalInput := SendWorkflowSignalRequest{Name: "external.done", PayloadVersion: "v1", TargetStepKey: "signal", TargetActivationSequence: signalBefore.Step.ActivationSequence, CorrelationKey: v.BusinessKey, Payload: workflowExact, Reason: "verified result while paused", EvidenceRef: "proof:paused-process"}
	accepted, err := api.SendWorkflowSignal(ctx, v.ID, signalKey, signalInput)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = work.Finish(ctx, workerTestID(t), WorkerFinish{Ref: first.Attempt(), Outcome: WorkerSucceeded, Result: workflowExact}); err != nil {
		t.Fatal(err)
	}
	initialTask, err := api.GetTask(ctx, first.Task.ID)
	if err != nil || initialTask.State != TaskSucceeded || initialTask.AttemptCount != 1 {
		t.Fatal("paused result not committed", err)
	}
	work.Close()
	queries.Close()
	a.kill(t)
	b.kill(t)
	// 以首次激活时间计算原TIMER唤醒，不用进程启动时间重新计时。
	wake := timerBefore.ActivatedAt.Add(5 * time.Second)
	if !time.Now().Before(wake) {
		t.Fatal("all controllers must stop before the original timer wake")
	}
	if pollBefore.Poll.NextPollAt.After(wake) {
		wake = *pollBefore.Poll.NextPollAt
	}
	if wait := time.Until(wake.Add(100 * time.Millisecond)); wait > 0 {
		select {
		case <-time.After(wait):
		case <-ctx.Done():
			t.Fatal(ctx.Err())
		}
	}
	c := startOrbitProcess(t)
	d := startOrbitProcess(t)
	if a.nodeID == b.nodeID || c.nodeID == d.nodeID || c.nodeID == a.nodeID || c.nodeID == b.nodeID || d.nodeID == a.nodeID || d.nodeID == b.nodeID {
		t.Fatal("node identity reused")
	}
	api = processClient(t, c, token)
	other := processClient(t, d, token)
	for _, entry := range []struct {
		key    string
		before WorkflowStepDetail
	}{{"signal", signalBefore}, {"timer", timerBefore}} {
		after := waitWorkflowStepSDK(t, ctx, api, v.ID, entry.key, func(d WorkflowStepDetail) bool { return d.Step.State == "SUCCEEDED" })
		if after.Wait == nil || after.Wait.ID != entry.before.Wait.ID || !after.Wait.DeadlineAt.Equal(entry.before.Wait.DeadlineAt) || after.Step.ActivationSequence != entry.before.Step.ActivationSequence || *after.Step.ActivationID != *entry.before.Step.ActivationID {
			t.Fatal("wait identity or original deadline changed across pause/restart")
		}
	}
	initialAfter := waitWorkflowStepSDK(t, ctx, api, v.ID, "initial", func(d WorkflowStepDetail) bool { return d.Step.State == "SUCCEEDED" })
	if initialAfter.Step.TaskID == nil || *initialAfter.Step.TaskID != first.Task.ID {
		t.Fatal("old result replaced after restart")
	}
	current, err := other.GetWorkflow(ctx, v.ID)
	if err != nil || !current.Paused || current.ControlRevision != paused.Instance.ControlRevision || current.DefinitionVersionID != v.DefinitionVersionID || !current.DeadlineAt.Equal(v.DeadlineAt) {
		t.Fatal("paused workflow changed on new nodes", err)
	}
	tail, err := api.GetWorkflowStep(ctx, v.ID, "work")
	if err != nil || tail.Step.State != "BLOCKED" || tail.Step.TaskID != nil {
		t.Fatal("paused successor created", err)
	}
	pollAfter, err := api.GetWorkflowStep(ctx, v.ID, "query")
	if err != nil || pollAfter.Poll == nil || pollAfter.Poll.UsedRounds != 1 || pollAfter.Poll.LastTaskID != firstQuery.Task.ID || pollAfter.Poll.CurrentTaskID != nil || pollAfter.Poll.NextPollAt == nil || !pollAfter.Poll.NextPollAt.Equal(*pollBefore.Poll.NextPollAt) || !pollAfter.Poll.DeadlineAt.Equal(pollBefore.Poll.DeadlineAt) || pollAfter.Poll.MaxRounds != 2 || pollAfter.Poll.MaxAttemptsPerRound != 2 {
		t.Fatal("paused query used another round or reset original budget", err)
	}
	received, err := api.GetWorkflowSignal(ctx, v.ID, accepted.Signal.ID)
	if err != nil || received.State != "CONSUMED" || !received.AcceptedAt.Equal(accepted.Signal.AcceptedAt) || received.TargetActivationSequence != accepted.Signal.TargetActivationSequence {
		t.Fatal("accepted signal lost", err)
	}
	// 在另一个新节点恢复原PAUSE、SIGNAL和START回执，不能修改当前状态或预算。
	duplicate, err := other.ControlWorkflow(ctx, v.ID, pauseKey, pauseInput)
	if err != nil || !duplicate.Deduplicated || duplicate.Instance.ControlRevision != paused.Instance.ControlRevision {
		t.Fatal("pause receipt lost", err)
	}
	signalDuplicate, err := other.SendWorkflowSignal(ctx, v.ID, signalKey, signalInput)
	if err != nil || !signalDuplicate.Deduplicated || signalDuplicate.Signal.ID != accepted.Signal.ID || signalDuplicate.Signal.State != accepted.Signal.State {
		t.Fatal("signal receipt changed", err)
	}
	startDuplicate, err := other.StartWorkflow(ctx, startKey, in)
	if err != nil || !startDuplicate.Deduplicated || startDuplicate.Instance.ID != v.ID || !startDuplicate.Instance.DeadlineAt.Equal(v.DeadlineAt) {
		t.Fatal("start receipt changed", err)
	}
	resumeKey := workerTestID(t)
	resumeInput := WorkflowControlRequest{Operation: "RESUME", ExpectedControlRevision: paused.Instance.ControlRevision, Reason: "resume persisted pause through another node"}
	resumed, err := other.ControlWorkflow(ctx, v.ID, resumeKey, resumeInput)
	if err != nil || resumed.Instance.Paused || resumed.Instance.ControlRevision != 3 || !resumed.Instance.DeadlineAt.Equal(v.DeadlineAt) {
		t.Fatal("resume changed fixed state", err)
	}
	queries = workflowWorkerSDK(t, ctx, api, queue, query.Handler)
	lastQuery := finishWorkflowSDK(t, ctx, queries, WorkerSucceeded, json.RawMessage(`"DONE"`))
	work = workflowWorkerSDK(t, ctx, other, queue, handler)
	last := finishWorkflowSDK(t, ctx, work, WorkerSucceeded, workflowExact)
	finished := waitWorkflowSDK(t, ctx, api, v.ID, "SUCCEEDED")
	if finished.Paused || finished.ControlRevision != 3 || !finished.DeadlineAt.Equal(v.DeadlineAt) || finished.DefinitionVersionID != v.DefinitionVersionID || string(finished.Output) != string(workflowExact) {
		t.Fatal("final output or original version/deadline changed")
	}
	finalPoll, err := api.GetWorkflowStep(ctx, v.ID, "query")
	if err != nil || finalPoll.Poll == nil || finalPoll.Poll.UsedRounds != 2 || !finalPoll.Poll.DeadlineAt.Equal(pollBefore.Poll.DeadlineAt) || finalPoll.Poll.CurrentTaskID != nil {
		t.Fatal("query did not preserve the two-round budget", err)
	}
	rounds, err := api.ListWorkflowPollRounds(ctx, v.ID, "query", *finalPoll.Step.ActivationID, HistoryPageRequest{})
	if err != nil || len(rounds.Items) != 2 || rounds.NextCursor != "" || rounds.Items[0].Round != 2 || rounds.Items[0].TaskID != lastQuery.Task.ID || rounds.Items[1].Round != 1 || rounds.Items[1].TaskID != firstQuery.Task.ID {
		t.Fatal("query history duplicated or reset after paused downtime", err)
	}
	tasks, err := api.ListTasks(ctx, ListTasksRequest{QueueID: queue, BusinessKey: handler, Limit: 100})
	if err != nil || len(tasks.Items) != 4 || tasks.NextCursor != "" {
		t.Fatal("unexpected tasks in paused workflow", err)
	}
	for _, assignment := range []WorkerAssignment{first, firstQuery, lastQuery, last} {
		task, err := api.GetTask(ctx, assignment.Task.ID)
		if err != nil || task.State != TaskSucceeded || task.AttemptCount != 1 {
			t.Fatal("duplicated execution after full stop", err)
		}
		history, err := api.ListTaskAttempts(ctx, task.ID, HistoryPageRequest{})
		if err != nil || len(history.Items) != 1 || history.NextCursor != "" {
			t.Fatal("historical attempts lost or duplicated", err)
		}
	}
	unchanged, err := api.GetTask(ctx, initialTask.ID)
	if err != nil || !unchanged.DeadlineAt.Equal(initialTask.DeadlineAt) || unchanged.Retry != initialTask.Retry {
		t.Fatal("original task policy changed", err)
	}
	resumeDuplicate, err := api.ControlWorkflow(ctx, v.ID, resumeKey, resumeInput)
	if err != nil || !resumeDuplicate.Deduplicated || resumeDuplicate.Instance.ControlRevision != 3 {
		t.Fatal("resume receipt changed", err)
	}
	t.Log("both original nodes SIGKILL confirmed; paused task/signal/timer collected by two fresh nodes; public SDK resume preserves version, receipts, deadlines, two query rounds and four single attempts")
}