// workflow-query 发布有限等待流程，按精确业务键和固定版本分页，并按控制版本结清示例实例。
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"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"`
	}
	if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" {
		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
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	operation, err := orbit.NewWorkerMessageID()
	if err != nil {
		return err
	}
	spec := orbit.WorkflowSpecification{SchemaVersion: 1, InputType: orbit.WorkflowType{Kind: "NULL"}, TimeoutMS: 120000, FailurePolicy: "REVIEW", Body: orbit.WorkflowBlock{
		Steps:      []orbit.WorkflowStep{{Key: "receipt", Kind: "SIGNAL", Signal: &orbit.WorkflowSignal{Name: "object.ready", PayloadVersion: "v1", Correlation: orbit.WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`"` + operation + `"`)}, PayloadType: orbit.WorkflowType{Kind: "BOOLEAN"}, TimeoutMS: 120000}}},
		OutputType: orbit.WorkflowType{Kind: "BOOLEAN"}, Output: orbit.WorkflowRef("receipt"),
	}}
	d, err := client.CreateWorkflowDefinition(ctx, operation, orbit.CreateWorkflowDefinitionRequest{Name: "query." + operation})
	if err != nil {
		return err
	}
	if _, err = client.PublishWorkflowVersion(ctx, d.Definition.ID, operation, orbit.PublishWorkflowVersionRequest{Version: "v1", Specification: spec, Reason: "publish query example"}); err != nil {
		return err
	}
	key := "对象 /+% é " + operation + " "
	var instances []orbit.WorkflowInstance
	for i, business := range []string{key, key, key[:len(key)-1]} {
		out, err := client.StartWorkflow(ctx, fmt.Sprintf("%s-%d", operation, i), orbit.StartWorkflowRequest{DefinitionID: d.Definition.ID, Version: "v1", Input: json.RawMessage(`null`), BusinessKey: business})
		if err != nil {
			return err
		}
		instances = append(instances, out.Instance)
	}
	from, before := instances[0].CreatedAt, instances[2].CreatedAt.Add(time.Microsecond)
	query := orbit.ListWorkflowsRequest{DefinitionID: d.Definition.ID, Version: "v1", BusinessKey: key, State: "WAITING", CreatedFrom: &from, CreatedBefore: &before, Limit: 1}
	seen := map[string]bool{}
	pages := 0
	for {
		page, err := client.ListWorkflows(ctx, query)
		if err != nil {
			return err
		}
		pages++
		for _, v := range page.Items {
			if v.BusinessKey != key || v.DefinitionID != d.Definition.ID || v.Version != "v1" || seen[v.ID] {
				return errors.New("workflow query changed filters or repeated an item")
			}
			seen[v.ID] = true
		}
		if page.NextCursor == "" {
			break
		}
		query.Cursor = page.NextCursor
	}
	if len(seen) != 2 || pages != 2 || !seen[instances[0].ID] || !seen[instances[1].ID] {
		return errors.New("expected exactly two matching workflow instances")
	}
	fmt.Printf("workflow_query_verified=true matched=%d pages=%d exact_trailing_space=true\n", len(seen), pages)
	// 取消记录保留，固定定义版本和审计也保留；失败时不要直接删除运行记录。
	for _, v := range instances {
		current, err := client.GetWorkflow(ctx, v.ID)
		if err != nil {
			return err
		}
		if _, err = client.ControlWorkflow(ctx, v.ID, operation+"-cleanup", orbit.WorkflowControlRequest{Operation: "CANCEL", ExpectedControlRevision: current.ControlRevision, Reason: "query example cleanup"}); err != nil {
			return err
		}
		for {
			current, err = client.GetWorkflow(ctx, v.ID)
			if err != nil {
				return err
			}
			if current.State == "CANCELLED" {
				break
			}
			if current.State == "NEEDS_REVIEW" {
				return errors.New("cleanup requires review")
			}
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(50 * time.Millisecond):
			}
		}
	}
	fmt.Println("workflow_query_cleanup=CANCELLED count=3")
	return nil
}