// workflow 通过公共SDK运行并行处理、计时等待、外部回执与后续发布；业务处理器在Worker中执行。
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strings"
	"time"

	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
)

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}
func run() error {
	raw, err := os.ReadFile(os.Getenv("ORBIT_CREDENTIALS_FILE"))
	if err != nil {
		return err
	}
	var credentials struct {
		Token   string `json:"token"`
		QueueID string `json:"queue_id"`
	}
	if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" || credentials.QueueID == "" {
		return errors.New("invalid private credentials file")
	}
	client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: credentials.Token})
	if err != nil {
		return err
	}
	operation, err := orbit.NewWorkerMessageID()
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
	defer cancel()
	handler := "artifact.normalize." + operation
	typ := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"text": {Kind: "STRING"}}}
	task := orbit.WorkflowTask{QueueID: credentials.QueueID, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", InputType: typ, Input: orbit.WorkflowRef("input"), ResultType: typ, TimeoutMS: 20000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 5000}, RecoverySafety: orbit.RecoveryIdempotent}
	pair := orbit.WorkflowParallel{Branches: []orbit.WorkflowParallelBranch{
		{Name: "left", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "left", Kind: "TASK", Task: &task}}, OutputType: typ, Output: orbit.WorkflowRef("left")}},
		{Name: "right", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "right", Kind: "TASK", Task: &task}}, OutputType: typ, Output: orbit.WorkflowRef("right")}},
	}}
	publish := task
	publish.Input = orbit.WorkflowRef("pair", "left")
	spec := orbit.WorkflowSpecification{SchemaVersion: 1, TimeoutMS: 60000, FailurePolicy: "REVIEW", InputType: typ, Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{
		{Key: "pair", Kind: "PARALLEL", Parallel: &pair},
		{Key: "cooldown", Kind: "TIMER", Timer: &orbit.WorkflowTimer{DelayMS: 500}},
		{Key: "receipt", Kind: "SIGNAL", Signal: &orbit.WorkflowSignal{Name: "artifact.confirmed", PayloadVersion: "v1", Correlation: orbit.WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`"` + operation + `"`)}, PayloadType: orbit.WorkflowType{Kind: "BOOLEAN"}, TimeoutMS: 30000}},
		{Key: "publish", Kind: "TASK", Task: &publish},
	}, OutputType: typ, Output: orbit.WorkflowRef("publish")}}
	definition, err := client.CreateWorkflowDefinition(ctx, operation, orbit.CreateWorkflowDefinitionRequest{Name: "artifact." + operation, Description: "并行结果生成与模拟外部回执"})
	if err != nil {
		return err
	}
	if _, err = client.PublishWorkflowVersion(ctx, definition.Definition.ID, operation, orbit.PublishWorkflowVersionRequest{Version: "v1", Specification: spec, Reason: "publish runnable workflow example"}); err != nil {
		return err
	}
	start := orbit.StartWorkflowRequest{DefinitionID: definition.Definition.ID, Version: "v1", Input: json.RawMessage(`{"text":"  hello orbit  "}`), BusinessKey: operation}
	accepted, err := client.StartWorkflow(ctx, operation, start)
	if err != nil {
		return err
	}
	repeated, err := client.StartWorkflow(ctx, operation, start)
	if err != nil || !repeated.Deduplicated || repeated.Instance.ID != accepted.Instance.ID {
		return errors.New("start identity was not preserved")
	}
	worker, err := orbit.NewWorker(client, orbit.WorkerOptions{QueueID: credentials.QueueID, Capacity: 2, ShutdownTimeout: 5 * time.Second, Handlers: []orbit.WorkerHandler{{Capability: orbit.WorkerCapability{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: normalize}}})
	if err != nil {
		return err
	}
	workerCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	go func() { done <- worker.Run(workerCtx) }()
	waited := false
	defer func() {
		stop()
		if !waited {
			<-done
		}
	}()
	var wait orbit.WorkflowStepDetail
	for {
		wait, err = client.GetWorkflowStep(ctx, accepted.Instance.ID, "receipt")
		if err != nil {
			return err
		}
		if wait.Step.State == "WAITING" {
			break
		}
		current, e := client.GetWorkflow(ctx, accepted.Instance.ID)
		if e != nil {
			return e
		}
		if current.State == "NEEDS_REVIEW" || current.State == "FAILED" {
			return fmt.Errorf("workflow requires attention: %s", current.State)
		}
		select {
		case e := <-done:
			waited = true
			return fmt.Errorf("Worker stopped: %v", e)
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(100 * time.Millisecond):
		}
	}
	if wait.Wait == nil || wait.Step.ActivationSequence == 0 {
		return errors.New("wait contract missing")
	}
	// 示例模拟外部系统的已验证回执。实际系统由独立、具备信号权限的调用者确认业务事实。
	signal := orbit.SendWorkflowSignalRequest{Name: *wait.Wait.Name, PayloadVersion: *wait.Wait.PayloadVersion, TargetStepKey: wait.Step.Key, TargetActivationSequence: wait.Step.ActivationSequence, CorrelationKey: wait.Wait.CorrelationKey, Payload: json.RawMessage(`true`), Reason: "simulated verified external receipt"}
	sent, err := client.SendWorkflowSignal(ctx, accepted.Instance.ID, operation, signal)
	if err != nil {
		return err
	}
	again, err := client.SendWorkflowSignal(ctx, accepted.Instance.ID, operation, signal)
	if err != nil || !again.Deduplicated || again.Signal.ID != sent.Signal.ID {
		return errors.New("signal identity was not preserved")
	}
	for {
		current, e := client.GetWorkflow(ctx, accepted.Instance.ID)
		if e != nil {
			return e
		}
		if current.State == "SUCCEEDED" {
			fmt.Printf("workflow=%s state=%s result=%s start_deduplicated=%t signal_deduplicated=%t\n", current.ID, current.State, current.Output, repeated.Deduplicated, again.Deduplicated)
			break
		}
		if current.State == "NEEDS_REVIEW" || current.State == "FAILED" || current.State == "CANCELLED" {
			return fmt.Errorf("workflow requires attention: %s", current.State)
		}
		select {
		case e := <-done:
			waited = true
			return fmt.Errorf("Worker stopped: %v", e)
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(100 * time.Millisecond):
		}
	}
	steps, err := client.GetWorkflowSteps(ctx, accepted.Instance.ID)
	if err != nil {
		return err
	}
	verified := 0
	for _, step := range steps.Steps {
		if step.TaskID == nil {
			continue
		}
		source, err := client.GetTaskSource(ctx, *step.TaskID)
		if err != nil {
			return err
		}
		if source.Workflow == nil || source.Workflow.WorkflowID != accepted.Instance.ID || source.Workflow.DefinitionVersionID != accepted.Instance.DefinitionVersionID || source.Workflow.StepID != step.ID || source.Workflow.ActivationID != *step.ActivationID || source.Workflow.ActivationSequence != 1 {
			return errors.New("workflow task source mismatch")
		}
		verified++
	}
	if verified != 3 {
		return errors.New("incomplete workflow task sources")
	}
	fmt.Printf("workflow_sources_verified=%d fixed_version=%s\n", verified, accepted.Instance.Version)
	history, err := client.ListWorkflowActivations(ctx, accepted.Instance.ID, "receipt", orbit.HistoryPageRequest{Limit: 20})
	if err != nil {
		return err
	}
	if len(history.Items) != 1 {
		return errors.New("unexpected receipt activation history")
	}
	historical, err := client.GetWorkflowActivation(ctx, accepted.Instance.ID, "receipt", history.Items[0].ID)
	if err != nil {
		return err
	}
	if historical.Wait == nil || historical.Wait.WinnerSignalID == nil || *historical.Wait.WinnerSignalID != sent.Signal.ID {
		return errors.New("historical signal winner missing")
	}
	signals, err := client.ListWorkflowSignals(ctx, accepted.Instance.ID, orbit.HistoryPageRequest{Limit: 20})
	if err != nil {
		return err
	}
	if len(signals.Items) != 1 || signals.Items[0].State != "CONSUMED" || signals.Items[0].ID != sent.Signal.ID {
		return errors.New("signal history missing")
	}
	fmt.Println("workflow_history_verified=true signal_state=CONSUMED")
	// 时间线用于审计展示，游标只是倒序翻页位置，不作为可靠消息消费水位。
	cursor, eventCount, consumed := "", 0, false
	for {
		page, err := client.ListWorkflowEvents(ctx, accepted.Instance.ID, orbit.HistoryPageRequest{Limit: 5, Cursor: cursor})
		if err != nil {
			return err
		}
		for _, event := range page.Items {
			eventCount++
			if event.AggregateKind == "workflow_signal" && event.AggregateID == sent.Signal.ID && event.State == "CONSUMED" {
				detail, err := client.GetWorkflowEvent(ctx, accepted.Instance.ID, event.ID)
				if err != nil {
					return err
				}
				var transition struct {
					From           string `json:"from"`
					ConsumedWaitID string `json:"consumed_wait_id"`
				}
				if json.Unmarshal(detail.Payload, &transition) != nil || transition.From != "PENDING" || transition.ConsumedWaitID != historical.Wait.ID {
					return errors.New("signal transition evidence mismatch")
				}
				consumed = true
			}
		}
		if page.NextCursor == "" {
			break
		}
		cursor = page.NextCursor
	}
	if !consumed {
		return errors.New("consumed signal event missing")
	}
	fmt.Printf("workflow_timeline_verified=true events=%d signal_transition=PENDING->CONSUMED\n", eventCount)

	stop()
	err = <-done
	waited = true
	if err != nil {
		return err
	}
	stats := worker.Stats()
	if stats.Started != 3 || stats.ReportsConfirmed != 3 || stats.Running != 0 {
		return errors.New("unexpected task count or incomplete drain")
	}
	fmt.Printf("worker_drained=true started=%d confirmed=%d\n", stats.Started, stats.ReportsConfirmed)
	return nil
}
func normalize(ctx context.Context, execution orbit.WorkerExecution) (json.RawMessage, error) {
	var in struct {
		Text string `json:"text"`
	}
	if json.Unmarshal(execution.Task.Payload, &in) != nil {
		return nil, &orbit.WorkerFailure{Code: "INVALID_TEXT"}
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return json.Marshal(struct {
		Text string `json:"text"`
	}{strings.ToUpper(strings.TrimSpace(in.Text))})
}