package orbit import ( "context" "encoding/json" "net/http" "net/http/httptest" "sync/atomic" "testing" "time" ) func TestTaskSourceValidationAndPrecision(t *testing.T) { const task = "11111111111111111111111111111111" const parent = "22222222222222222222222222222222" const occurrence = "33333333333333333333333333333333" at := time.Date(2026, 9, 17, 0, 0, 0, 123456000, time.UTC) until := at.Add(time.Minute) var reply atomic.Value var calls atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls.Add(1) if r.Method != "GET" || r.URL.Path != "/v1/tasks/"+task+"/source" || r.Header.Get("Authorization") != "Bearer example-token" || r.Header.Get("Idempotency-Key") != "" { t.Error("invalid source request") } json.NewEncoder(w).Encode(map[string]any{"request_id": task, "data": reply.Load()}) })) defer server.Close() client, _ := NewClient(server.URL, Options{Token: "example-token"}) p := parent valid := []TaskSource{{TaskID: task, Kind: "DIRECT"}, {TaskID: task, Kind: "MANUAL_RETRY", RetryOf: &p}, {TaskID: task, Kind: "SCHEDULE", Schedule: &TaskScheduleSource{ScheduleID: parent, ScheduleVersion: 9007199254740993, OccurrenceID: occurrence, ScheduledAt: at, Decision: "FIRED"}}, {TaskID: task, Kind: "SCHEDULE", Schedule: &TaskScheduleSource{ScheduleID: parent, ScheduleVersion: 1, OccurrenceID: occurrence, ScheduledAt: at, Decision: "COALESCED", CoveredUntil: &until}}} for _, v := range valid { reply.Store(v) out, err := client.GetTaskSource(context.Background(), task) if err != nil { t.Fatal(err) } if v.Schedule != nil && out.Schedule.ScheduleVersion != v.Schedule.ScheduleVersion { t.Fatal("version precision lost") } } workflow := TaskWorkflowSource{WorkflowID: parent, DefinitionID: parent, DefinitionVersionID: occurrence, Version: "v1", StepID: parent, StepKey: "query", StepKind: "TASK", ActivationID: occurrence, ActivationSequence: 1} for _, kind := range []string{"TASK", "POLL"} { for _, sequence := range []uint32{1, 2} { w := workflow w.StepKind, w.ActivationSequence = kind, sequence if sequence == 2 { w.RetryOf = &p } if kind == "POLL" { w.PollRound, w.PollRoundID = 2, &p } reply.Store(TaskSource{TaskID: task, Kind: "WORKFLOW", Workflow: &w}) if _, err := client.GetTaskSource(context.Background(), task); err != nil { t.Fatal("valid workflow source", err) } } } self := task invalid := []TaskSource{{TaskID: parent, Kind: "DIRECT"}, {TaskID: task, Kind: "unknown"}, {TaskID: task, Kind: "DIRECT", RetryOf: &p}, {TaskID: task, Kind: "MANUAL_RETRY"}, {TaskID: task, Kind: "MANUAL_RETRY", RetryOf: &self}, {TaskID: task, Kind: "SCHEDULE"}} invalid = append(invalid, TaskSource{TaskID: task, Kind: "WORKFLOW"}, TaskSource{TaskID: task, Kind: "DIRECT", Workflow: &workflow}, TaskSource{TaskID: task, Kind: "WORKFLOW", Workflow: &workflow, RetryOf: &p}) for _, change := range []func(*TaskWorkflowSource){ func(w *TaskWorkflowSource) { w.WorkflowID = "invalid" }, func(w *TaskWorkflowSource) { w.Version = "../version" }, func(w *TaskWorkflowSource) { w.StepKey = "$root" }, func(w *TaskWorkflowSource) { w.StepKind = "SIGNAL" }, func(w *TaskWorkflowSource) { w.ActivationSequence = 0 }, func(w *TaskWorkflowSource) { w.ActivationSequence = 2 }, func(w *TaskWorkflowSource) { w.RetryOf = &p }, func(w *TaskWorkflowSource) { w.ActivationSequence = 2; w.RetryOf = &self }, func(w *TaskWorkflowSource) { w.PollRound = 1 }, func(w *TaskWorkflowSource) { w.PollRoundID = &p }, func(w *TaskWorkflowSource) { w.StepKind = "POLL"; w.PollRound = 1 }, } { w := workflow change(&w) invalid = append(invalid, TaskSource{TaskID: task, Kind: "WORKFLOW", Workflow: &w}) } for _, change := range []func(*TaskScheduleSource){func(v *TaskScheduleSource) { v.ScheduleVersion = 0 }, func(v *TaskScheduleSource) { v.Decision = "SKIPPED" }, func(v *TaskScheduleSource) { v.Decision = "COALESCED" }, func(v *TaskScheduleSource) { v.CoveredUntil = &until }, func(v *TaskScheduleSource) { v.ScheduledAt = time.Time{} }, func(v *TaskScheduleSource) { v.OccurrenceID = "invalid" }} { v := valid[2] s := *v.Schedule change(&s) v.Schedule = &s invalid = append(invalid, v) } for _, v := range invalid { reply.Store(v) if _, err := client.GetTaskSource(context.Background(), task); err == nil { t.Fatal("invalid source accepted", v) } } before := calls.Load() if _, err := client.GetTaskSource(context.Background(), "../tasks"); err == nil || calls.Load() != before { t.Fatal("unsafe source path sent") } }