package engine

import (
	"bytes"
	"encoding/json"
	"errors"
	"reflect"
	"strings"
	"sync"
	"testing"

	core "git.yuntuo.hk/k2safe/orbit/model"
	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
	"git.yuntuo.hk/k2safe/orbit/weave/definition"
	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
)

func val(raw string) definition.Value {
	return definition.Value{Kind: "LITERAL", Literal: json.RawMessage(raw)}
}
func ref(source string, path ...string) definition.Value {
	return definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: source, Path: path}}
}
func block(typ definition.Type, value definition.Value, steps ...definition.Step) definition.Block {
	return definition.Block{OutputType: typ, Output: value, Steps: steps}
}
func task(queue core.ID, input definition.Value) definition.Task {
	return definition.Task{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", InputType: definition.Type{Kind: "NUMBER"}, Input: input, ResultType: definition.Type{Kind: "NUMBER"}, TimeoutMS: 5000, Retry: pulse.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 1000}, RecoverySafety: pulse.RecoveryReconcileOnly}
}
func programSpec() definition.Specification {
	number := definition.Type{Kind: "NUMBER"}
	left, right := ref("input", "n"), val(`9007199254740992`)
	work := task("11111111111111111111111111111111", ref("input", "n"))
	a, b := work, work
	fallback := block(number, val(`0`), definition.Step{Key: "unused", Kind: "TASK", Task: &work})
	choose := definition.Step{Key: "choice", Kind: "BRANCH", Branch: &definition.Branch{Cases: []definition.BranchCase{{When: definition.Predicate{Op: "GT", Left: &left, Right: &right}, Body: block(number, ref("selected"), definition.Step{Key: "selected", Kind: "TASK", Task: &work})}}, Default: &fallback}}
	a.Input = ref("choice")
	b.Input = ref("choice")
	joined := definition.Step{Key: "outputs", Kind: "PARALLEL", Parallel: &definition.Parallel{Branches: []definition.ParallelBranch{{Name: "first", Body: block(number, ref("a"), definition.Step{Key: "a", Kind: "TASK", Task: &a})}, {Name: "second", Body: block(number, ref("b"), definition.Step{Key: "b", Kind: "TASK", Task: &b})}}}}
	out := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"first": number, "second": number}}
	return definition.Specification{SchemaVersion: 1, InputType: definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"n": number}}, TimeoutMS: 30000, FailurePolicy: "REVIEW", Body: block(out, ref("outputs"), choose, joined)}
}
func snapshot() Snapshot {
	return Snapshot{WorkflowID: "22222222222222222222222222222222", BusinessKey: "operation", Input: json.RawMessage(`{"n":9007199254740993}`), Steps: map[string]State{}, AllowNew: true}
}
func compiled(t *testing.T, spec definition.Specification) *Program {
	t.Helper()
	p, err := Compile(spec)
	if err != nil {
		t.Fatal(err)
	}
	return p
}
func persist(t *testing.T, s *Snapshot, d Decision) {
	t.Helper()
	for _, c := range d.Changes {
		s.Steps[c.Key] = c.State
	}
	raw, err := json.Marshal(s.Steps)
	if err != nil {
		t.Fatal(err)
	}
	s.Steps = nil
	if err := json.Unmarshal(raw, &s.Steps); err != nil {
		t.Fatal(err)
	}
}
func succeed(s *Snapshot, key string, raw json.RawMessage) {
	row := s.Steps[key]
	row.Status = weave.StepSucceeded
	row.Output = copyJSON(raw)
	s.Steps[key] = row
}

func TestBudgetedRestartedBranchParallelPlan(t *testing.T) {
	for _, budget := range []int{1, 2, 32} {
		t.Run(string(rune('A'+budget)), func(t *testing.T) {
			spec := programSpec()
			s := snapshot()
			started := map[string]int{}
			var final Decision
			for iteration := 0; iteration < 40; iteration++ {
				// 每轮重新编译并从JSON恢复步骤；模拟存储适配层可丢弃全部进程内状态。
				p := compiled(t, spec)
				before, _ := json.Marshal(s.Steps)
				d, err := p.Plan(s, budget)
				if err != nil {
					t.Fatal(err)
				}
				after, _ := json.Marshal(s.Steps)
				if !bytes.Equal(before, after) {
					t.Fatal("planner mutated caller snapshot")
				}
				if len(d.Changes) > budget {
					t.Fatal("step budget exceeded")
				}
				persist(t, &s, d)
				for _, c := range d.Changes {
					if c.Task != nil {
						started[c.Key]++
						if string(c.Task.Payload) != `9007199254740993` {
							t.Fatal("numeric reference changed", string(c.Task.Payload))
						}
						succeed(&s, c.Key, c.Task.Payload)
					}
				}
				if d.Complete {
					final = d
					break
				}
			}
			if !final.Complete || string(final.Output) != `{"first":9007199254740993,"second":9007199254740993}` || !reflect.DeepEqual(started, map[string]int{"selected": 1, "a": 1, "b": 1}) {
				t.Fatal("incomplete or duplicated work", final, started)
			}
			if s.Steps["unused"].Status != weave.StepSkipped {
				t.Fatal("unselected path was not skipped")
			}
			repeated, err := compiled(t, spec).Plan(s, budget)
			if err != nil || !repeated.Complete || len(repeated.Changes) != 0 {
				t.Fatal("completed replay changed state", err)
			}
		})
	}
}
func TestBudgetExhaustionStillVisitsActiveParallelBranch(t *testing.T) {
	spec := programSpec()
	number := definition.Type{Kind: "NUMBER"}
	work := task("11111111111111111111111111111111", ref("input", "n"))
	left, right := val(`0`), val(`1`)
	fallback := block(number, ref("y"), definition.Step{Key: "y", Kind: "TASK", Task: &work})
	leftChoice := definition.Step{Key: "left_choice", Kind: "BRANCH", Branch: &definition.Branch{
		Cases: []definition.BranchCase{{When: definition.Predicate{Op: "EQ", Left: &left, Right: &right}, Body: block(number, val(`0`))}}, Default: &fallback,
	}}
	joined := definition.Step{Key: "joined", Kind: "PARALLEL", Parallel: &definition.Parallel{Branches: []definition.ParallelBranch{
		{Name: "first", Body: block(number, ref("left_choice"), definition.Step{Key: "x", Kind: "TASK", Task: &work}, leftChoice)},
		{Name: "second", Body: block(number, ref("choice"), spec.Body.Steps[0])},
	}}}
	spec.Body = block(spec.Body.OutputType, ref("joined"), joined)
	s := snapshot()
	d, err := compiled(t, spec).Plan(s, 32)
	if err != nil {
		t.Fatal(err)
	}
	persist(t, &s, d)
	succeed(&s, "x", json.RawMessage(`1`))
	// 左侧激活组合后耗尽预算，右侧已有分支及在途任务仍须遍历，不能误报不可达。
	d, err = compiled(t, spec).Plan(s, 1)
	if err != nil || !d.More || len(d.Changes) != 1 || d.Changes[0].Key != "left_choice" {
		t.Fatal("budget prevented visiting existing work", d, err)
	}
	persist(t, &s, d)
	succeed(&s, "selected", json.RawMessage(`2`))
	for i := 0; i < 12; i++ {
		d, err = compiled(t, spec).Plan(s, 1)
		if err != nil {
			t.Fatal(err)
		}
		persist(t, &s, d)
		for _, change := range d.Changes {
			if change.Task != nil {
				if change.Key != "y" {
					t.Fatal("existing task was activated again", change.Key)
				}
				succeed(&s, "y", json.RawMessage(`3`))
			}
		}
		if d.Complete {
			if string(d.Output) != `{"first":3,"second":2}` {
				t.Fatal("invalid parallel output", string(d.Output))
			}
			return
		}
	}
	t.Fatal("bounded parallel plan did not complete")
}

func TestExactConditionsAndValues(t *testing.T) {
	for _, tc := range []struct {
		a, b, op string
		want     bool
	}{
		{`9007199254740993`, `9007199254740992`, "GT", true},
		{`0.10000000000000000001`, `0.1`, "GT", true},
		{`1e3`, `1000.000`, "EQ", true},
		{`-1e-308`, `0`, "LT", true},
		{`"a "`, `"a"`, "EQ", false}, {`true`, `false`, "NE", true}, {`null`, `null`, "EQ", true},
	} {
		a, b := val(tc.a), val(tc.b)
		got, err := compare(definition.Predicate{Op: tc.op, Left: &a, Right: &b}, values{})
		if err != nil || got != tc.want {
			t.Fatalf("%s %s %s: %v %v", tc.a, tc.op, tc.b, got, err)
		}
	}
	value := definition.Value{Kind: "OBJECT", Fields: map[string]definition.Value{"n": ref("input", "n"), "array": {Kind: "ARRAY", Items: []definition.Value{val(`1.0000000000000000001`), val(`null`)}}}}
	raw, err := resolve(value, definition.Type{Kind: "JSON"}, values{"input": json.RawMessage(`{"n":9007199254740993}`)})
	if err != nil || string(raw) != `{"array":[1.0000000000000000001,null],"n":9007199254740993}` {
		t.Fatal(string(raw), err)
	}
	a, b := val(`true`), val(`true`)
	condition := definition.Predicate{Op: "NOT", Arguments: []definition.Predicate{{Op: "EQ", Left: &a, Right: &b}}}
	ok, err := compare(condition, nil)
	if err != nil || ok {
		t.Fatal("logical comparison", err)
	}
}
func TestSnapshotAndResultFailuresProduceNoChanges(t *testing.T) {
	p := compiled(t, programSpec())
	for _, change := range []func(*Snapshot){
		func(s *Snapshot) { s.Input = json.RawMessage(`{"n":"wrong"}`) },
		func(s *Snapshot) { s.Steps["unknown"] = State{Status: weave.StepBlocked} },
		func(s *Snapshot) {
			s.Steps["selected"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`1`), Output: json.RawMessage(`"wrong"`)}
		},
		func(s *Snapshot) { s.Steps["selected"] = State{Status: weave.StepNeedsReview} },
		func(s *Snapshot) {
			s.Steps["choice"] = State{Status: weave.StepActive, Input: json.RawMessage(`{"case":100}`)}
		},
		func(s *Snapshot) {
			s.Steps["choice"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`{"case":0}`), Output: json.RawMessage(`1`)}
		},
		func(s *Snapshot) { s.Steps["unused"] = State{Status: weave.StepActive, Input: json.RawMessage(`1`)} },
	} {
		s := snapshot()
		change(&s)
		d, err := p.Plan(s, 32)
		if err == nil || len(d.Changes) != 0 || d.Complete {
			t.Fatal("unsafe snapshot advanced", d, err)
		}
	}
	for _, limit := range []int{0, 33} {
		if _, err := p.Plan(snapshot(), limit); err == nil {
			t.Fatal("unbounded iteration accepted")
		}
	}
}
func TestPauseFrozenChoiceAndCompiledIsolation(t *testing.T) {
	spec := programSpec()
	p := compiled(t, spec)
	s := snapshot()
	s.AllowNew = false
	d, err := p.Plan(s, 32)
	if err != nil || len(d.Changes) != 0 || d.Complete {
		t.Fatal("pause activated work", d, err)
	}
	s.AllowNew = true
	d, err = p.Plan(s, 1)
	if err != nil || len(d.Changes) != 1 {
		t.Fatal(err)
	}
	persist(t, &s, d)
	s.AllowNew = false
	d, err = p.Plan(s, 32)
	if err != nil {
		t.Fatal(err)
	}
	for _, c := range d.Changes {
		if c.Operation == "ACTIVATE" {
			t.Fatal("pause activated branch child")
		}
	}
	persist(t, &s, d)
	// 编译之后修改调用方定义和公开副本均不能改变正在解释的固定版本。
	spec.Body.Steps[0].Branch.Cases[0].Body.Steps[0].Task.Handler = "changed"
	copy, ok := p.Step("selected")
	if !ok {
		t.Fatal("missing metadata")
	}
	copy.Task.Handler = "also.changed"
	s.AllowNew = true
	d, err = p.Plan(s, 32)
	if err != nil {
		t.Fatal(err)
	}
	found := false
	for _, c := range d.Changes {
		if c.Task != nil {
			found = true
			if c.Task.Handler != "object.prepare" {
				t.Fatal("compiled program changed")
			}
		}
	}
	if !found {
		t.Fatal("frozen branch did not resume")
	}
	// 已保存的默认分支选择即便重新求值会匹配case，也必须继续原选择。
	s = snapshot()
	s.Steps["choice"] = State{Status: weave.StepActive, Input: json.RawMessage(`{"case":-1}`)}
	d, err = p.Plan(s, 32)
	if err != nil {
		t.Fatal(err)
	}
	for _, c := range d.Changes {
		if c.Task != nil && c.Key != "unused" {
			t.Fatal("branch was reevaluated")
		}
	}
}
func TestWaitAndPollPreparationAndRuntimeLimits(t *testing.T) {
	spec := programSpec()
	str := definition.Type{Kind: "STRING"}
	status := definition.Type{Kind: "STRING", Enum: []string{"PENDING", "DONE"}}
	spec.InputType = str
	signal := definition.Signal{Name: "job.final", PayloadVersion: "v1", Correlation: ref("input"), PayloadType: status, TimeoutMS: 10000}
	query := task("11111111111111111111111111111111", val(`1`))
	query.ResultType = status
	poll := definition.Poll{Task: query, MaxRounds: 2, TimeoutMS: 10000, IntervalMS: 1000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}, FinalSignal: &signal}
	spec.Body = block(status, ref("poll"), definition.Step{Key: "signal", Kind: "SIGNAL", Signal: &signal}, definition.Step{Key: "timer", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 1000}}, definition.Step{Key: "poll", Kind: "POLL", Poll: &poll})
	p := compiled(t, spec)
	s := snapshot()
	s.Input = json.RawMessage(`"operation"`)
	d, err := p.Plan(s, 32)
	if err != nil || len(d.Changes) != 1 || d.Changes[0].Wait.Correlation != "operation" || d.Changes[0].Task != nil {
		t.Fatal("signal", d, err)
	}
	persist(t, &s, d)
	again, err := p.Plan(s, 32)
	if err != nil || len(again.Changes) != 0 {
		t.Fatal("wait restarted", err)
	}
	succeed(&s, "signal", json.RawMessage(`"DONE"`))
	d, err = p.Plan(s, 32)
	if err != nil || d.Changes[0].Wait.TimeoutMS != 1000 {
		t.Fatal("timer", err)
	}
	persist(t, &s, d)
	succeed(&s, "timer", json.RawMessage(`null`))
	d, err = p.Plan(s, 32)
	if err != nil || d.Changes[0].Task == nil || d.Changes[0].Wait.Kind != "POLL" {
		t.Fatal("poll", err)
	}
	s = snapshot()
	s.Input = json.RawMessage(`""`)
	if d, err := p.Plan(s, 32); err == nil || len(d.Changes) != 0 {
		t.Fatal("empty runtime correlation accepted")
	}
	s.Input, _ = json.Marshal(strings.Repeat("x", 513))
	if _, err := p.Plan(s, 32); err == nil {
		t.Fatal("unbounded correlation accepted")
	}
	// 同一引用同时用作两个Pulse键，不能因指针相同漏掉一个键。
	shared := ref("key")
	q := task("11111111111111111111111111111111", val(`1`))
	q.BusinessKey = &shared
	q.ConcurrencyKey = &shared
	prepared, err := prepareTask(q, values{"key": json.RawMessage(`"operation"`)})
	if err != nil || prepared.BusinessKey != "operation" || prepared.ConcurrencyKey != "operation" {
		t.Fatal(prepared, err)
	}
}
func TestParallelCompletionIntegrityAndConcurrentPlanning(t *testing.T) {
	spec := programSpec()
	p := compiled(t, spec)
	s := snapshot()
	for i := 0; i < 15; i++ {
		d, err := p.Plan(s, 32)
		if err != nil {
			t.Fatal(err)
		}
		persist(t, &s, d)
		for _, c := range d.Changes {
			if c.Task != nil {
				succeed(&s, c.Key, c.Task.Payload)
			}
		}
		if d.Complete {
			break
		}
	}
	if !reflect.DeepEqual(s.Steps["outputs"].Status, weave.StepSucceeded) {
		t.Fatal("not completed")
	}
	// 完成的组合不能掩盖未结清子步骤或伪造输出。
	broken := snapshot()
	for k, v := range s.Steps {
		broken.Steps[k] = v
	}
	row := broken.Steps["a"]
	row.Status = weave.StepActive
	row.Output = nil
	broken.Steps["a"] = row
	_, err := p.Plan(broken, 32)
	var issue *Problem
	if !errors.As(err, &issue) || issue.Code != "INCOMPLETE_COMPOSITE_CHILD" {
		t.Fatal("incomplete child hidden", err)
	}
	broken.Steps["a"] = s.Steps["a"]
	row = broken.Steps["outputs"]
	row.Output = json.RawMessage(`{"first":0,"second":0}`)
	broken.Steps["outputs"] = row
	if _, err := p.Plan(broken, 32); err == nil {
		t.Fatal("fabricated composite result accepted")
	}
	var wg sync.WaitGroup
	for i := 0; i < 16; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			d, err := p.Plan(s, 32)
			if err != nil || !d.Complete || len(d.Changes) != 0 {
				t.Error("concurrent immutable plan", err)
			}
		}()
	}
	wg.Wait()
}

func TestUnreachableExecutionsAndAggregateOutputLimit(t *testing.T) {
	p := compiled(t, programSpec())
	s := snapshot()
	s.Steps["outputs"] = State{Status: weave.StepActive, Input: json.RawMessage(`{}`)}
	if d, err := p.Plan(s, 32); err == nil || len(d.Changes) != 0 {
		t.Fatal("future execution was ignored")
	}
	large, _ := json.Marshal(strings.Repeat("x", 140000))
	expr := definition.Value{Kind: "OBJECT", Fields: map[string]definition.Value{"a": ref("value"), "b": ref("value")}}
	if _, err := resolve(expr, definition.Type{Kind: "JSON"}, values{"value": large}); err == nil {
		t.Fatal("aggregate output limit ignored")
	}
}

func TestPollCompletionMustHaveFinalSuccess(t *testing.T) {
	spec := programSpec()
	query := task("11111111111111111111111111111111", val(`1`))
	query.ResultType = definition.Type{Kind: "STRING", Enum: []string{"WORKING", "DONE", "FAILED", "REVIEW"}}
	poll := definition.Poll{Task: query, MaxRounds: 2, TimeoutMS: 10000, IntervalMS: 1000, Pending: []string{"WORKING"}, Succeeded: []string{"DONE"}, Failed: []string{"FAILED"}, Review: []string{"REVIEW"}}
	spec.Body = block(query.ResultType, ref("poll"), definition.Step{Key: "poll", Kind: "POLL", Poll: &poll})
	p := compiled(t, spec)
	for status, want := range map[string]PollOutcome{"WORKING": PollPending, "DONE": PollSucceeded, "FAILED": PollFailed, "REVIEW": PollNeedsReview} {
		raw, _ := json.Marshal(status)
		out, err := p.ClassifyPoll("poll", raw)
		if err != nil || out != want {
			t.Fatal(status, out, err)
		}
		s := snapshot()
		s.Steps["poll"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`1`), Output: raw}
		decision, err := p.Plan(s, 32)
		if status == "DONE" {
			if err != nil || !decision.Complete {
				t.Fatal("final success rejected", err)
			}
		} else if err == nil || len(decision.Changes) != 0 {
			t.Fatal("nonfinal poll result advanced")
		}
	}
	if _, err := p.ClassifyPoll("poll", json.RawMessage(`"unexpected"`)); err == nil {
		t.Fatal("unknown status accepted")
	}
}