poimen-workflows/examples/service_integration.go

// Example: External service integrating with Poimen RoutingWorkflow
// Shows how to submit a task and wait for completion

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"go.temporal.io/sdk/client"
)

// ---- Types (mirror internal/routing/types.go) ----

type WorkflowSpec struct {
	Name   string                 `json:"name"`
	Input  map[string]interface{} `json:"input,omitempty"`
	States []State                `json:"states"`
}

type State struct {
	Name       string                 `json:"name"`
	Type       string                 `json:"type"` // Task, Pass, Fail
	Resource   string                 `json:"resource,omitempty"`
	Parameters map[string]interface{} `json:"parameters,omitempty"`
	Timeout    string                 `json:"timeout,omitempty"`
	Next       string                 `json:"next,omitempty"`
	End        bool                   `json:"end,omitempty"`
}

type RoutingWorkflowInput struct {
	Spec *WorkflowSpec `json:"spec"`
}

type RoutingWorkflowOutput struct {
	Status      string                            `json:"status"`
	StepResults map[string]map[string]interface{} `json:"stepResults"`
	Error       string                            `json:"error,omitempty"`
}

// ---- Example Service ----

type TaskService struct {
	temporalClient client.Client
	taskQueue      string
}

func NewTaskService(temporalHost, namespace, taskQueue string) (*TaskService, error) {
	c, err := client.Dial(client.Options{
		HostPort:  temporalHost,
		Namespace: namespace,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to connect to Temporal: %w", err)
	}

	return &TaskService{
		temporalClient: c,
		taskQueue:      taskQueue,
	}, nil
}

func (s *TaskService) Close() {
	s.temporalClient.Close()
}

// SubmitAndWait submits a workflow spec and waits for completion
func (s *TaskService) SubmitAndWait(ctx context.Context, spec *WorkflowSpec, timeout time.Duration) (*RoutingWorkflowOutput, error) {
	workflowID := fmt.Sprintf("%s-%d", spec.Name, time.Now().UnixNano())

	// Start workflow
	run, err := s.temporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
		ID:        workflowID,
		TaskQueue: s.taskQueue,
	}, "RoutingWorkflow", RoutingWorkflowInput{Spec: spec})
	if err != nil {
		return nil, fmt.Errorf("failed to start workflow: %w", err)
	}

	log.Printf("Workflow started: ID=%s, RunID=%s", run.GetID(), run.GetRunID())

	// Wait for completion with timeout
	waitCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	var result RoutingWorkflowOutput
	if err := run.Get(waitCtx, &result); err != nil {
		return nil, fmt.Errorf("workflow failed: %w", err)
	}

	return &result, nil
}

// SubmitAsync submits workflow and returns immediately (fire-and-forget)
func (s *TaskService) SubmitAsync(ctx context.Context, spec *WorkflowSpec) (workflowID string, runID string, err error) {
	workflowID = fmt.Sprintf("%s-%d", spec.Name, time.Now().UnixNano())

	run, err := s.temporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
		ID:        workflowID,
		TaskQueue: s.taskQueue,
	}, "RoutingWorkflow", RoutingWorkflowInput{Spec: spec})
	if err != nil {
		return "", "", fmt.Errorf("failed to start workflow: %w", err)
	}

	return run.GetID(), run.GetRunID(), nil
}

// WaitForCompletion waits for an existing workflow to complete
func (s *TaskService) WaitForCompletion(ctx context.Context, workflowID string, timeout time.Duration) (*RoutingWorkflowOutput, error) {
	run := s.temporalClient.GetWorkflow(ctx, workflowID, "")

	waitCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	var result RoutingWorkflowOutput
	if err := run.Get(waitCtx, &result); err != nil {
		return nil, fmt.Errorf("workflow failed: %w", err)
	}

	return &result, nil
}

// GetStatus gets current workflow status without waiting
func (s *TaskService) GetStatus(ctx context.Context, workflowID string) (string, error) {
	desc, err := s.temporalClient.DescribeWorkflowExecution(ctx, workflowID, "")
	if err != nil {
		return "", err
	}
	return desc.WorkflowExecutionInfo.Status.String(), nil
}

// ---- Example Usage ----

func main() {
	// Connect to Temporal
	svc, err := NewTaskService(
		"temporal-frontend.temporal:7233",
		"poimen-harness",
		"poimen-taskqueue",
	)
	if err != nil {
		log.Fatal(err)
	}
	defer svc.Close()

	ctx := context.Background()

	// Example 1: Implement task T0.3 with code analysis
	fmt.Println("=== Example 1: Submit and Wait ===")
	spec := &WorkflowSpec{
		Name: "implement-T0.3",
		Input: map[string]interface{}{
			"taskId":      "T0.3",
			"description": "Implement git worktree management",
			"repo":        "https://github.com/rockliang/poimen",
		},
		States: []State{
			{
				Name:     "Clone",
				Type:     "Task",
				Resource: "CloneRepoActivity",
				Parameters: map[string]interface{}{
					"repo":   "${workflow.input.repo}",
					"branch": "main",
				},
				Timeout: "5m",
				Next:    "Analyze",
			},
			{
				Name:     "Analyze",
				Type:     "Task",
				Resource: "AnalyzeCodeActivity",
				Parameters: map[string]interface{}{
					"path":  "${Clone.output.path}",
					"depth": 3,
				},
				Timeout: "10m",
				Next:    "Report",
			},
			{
				Name:     "Report",
				Type:     "Task",
				Resource: "GenerateReportActivity",
				Parameters: map[string]interface{}{
					"analysisResult": "${Analyze.output}",
					"format":         "markdown",
				},
				Timeout: "2m",
				End:     true,
			},
		},
	}

	result, err := svc.SubmitAndWait(ctx, spec, 30*time.Minute)
	if err != nil {
		log.Printf("Error: %v", err)
	} else {
		fmt.Printf("Status: %s\n", result.Status)
		for step, output := range result.StepResults {
			fmt.Printf("  %s: %v\n", step, output)
		}
	}

	// Example 2: Fire and forget, then poll
	fmt.Println("\n=== Example 2: Async Submit + Poll ===")
	workflowID, runID, err := svc.SubmitAsync(ctx, spec)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Submitted: workflowID=%s, runID=%s\n", workflowID, runID)

	// Poll status
	for i := 0; i < 5; i++ {
		status, _ := svc.GetStatus(ctx, workflowID)
		fmt.Printf("  Poll %d: status=%s\n", i+1, status)
		if status == "WORKFLOW_EXECUTION_STATUS_COMPLETED" {
			break
		}
		time.Sleep(5 * time.Second)
	}

	// Get final result
	result, err = svc.WaitForCompletion(ctx, workflowID, 30*time.Minute)
	if err != nil {
		log.Printf("Error: %v", err)
	} else {
		resultJSON, _ := json.MarshalIndent(result, "", "  ")
		fmt.Printf("Final result:\n%s\n", resultJSON)
	}
}