diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 07dc36c..53e6ca3 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -32,7 +32,7 @@ jobs: run: go vet ./... - name: Go test - run: go test ./... + run: go test -short ./... - name: Get short SHA id: sha diff --git a/cmd/gateway/main.go b/cmd/gateway/main.go index afce3ba..ba826d5 100644 --- a/cmd/gateway/main.go +++ b/cmd/gateway/main.go @@ -108,6 +108,10 @@ func main() { } _ = registry.Add(notifAdapter) + // Add Poimen orchestrator adapter (typed REST for AI orchestrator frontend) + poimenAdapter := serviceadapter.NewPoimenAdapter(temporalHandler) + _ = registry.Add(poimenAdapter) + // Add other adapters from config (skip if already registered in code) for _, a := range cfg.Adapters { if existing := registry.Get(a.ServiceName); existing != nil { diff --git a/internal/serviceadapter/poimen_adapter.go b/internal/serviceadapter/poimen_adapter.go new file mode 100644 index 0000000..269cf2e --- /dev/null +++ b/internal/serviceadapter/poimen_adapter.go @@ -0,0 +1,97 @@ +package serviceadapter + +import ( + "forgejo.riotpiao.com/rock/homelab-frontend/internal/temporal" +) + +// NewPoimenAdapter creates a ServiceAdapter for the Poimen orchestrator service. +func NewPoimenAdapter(h *temporal.Handler) *ServiceAdapter { + return &ServiceAdapter{ + Namespace: "poimen", + ServiceName: "poimen", + Handler: temporal.NewPoimenHandler(h), + Spec: *GetPoimenSpec(), + } +} + +// GetPoimenSpec returns the ServiceAdapter spec for the Poimen service. +func GetPoimenSpec() *Spec { + return &Spec{ + ServiceName: "poimen", + Upstream: Upstream{ + URL: "internal://poimen", + TimeoutSeconds: 60, + }, + Auth: Auth{ + Required: true, + Capability: "workflow:read", + }, + Resources: []Resource{ + { + Name: "plan", + Methods: []Method{{ + Verb: "POST", + UpstreamPath: "/v1/poimen/plan", + Auth: &Auth{Required: true, Capability: "workflow:execute"}, + }}, + }, + { + Name: "execute", + Methods: []Method{{ + Verb: "POST", + UpstreamPath: "/v1/poimen/execute", + Auth: &Auth{Required: true, Capability: "workflow:execute"}, + }}, + }, + { + Name: "executions", + Methods: []Method{ + { + Verb: "GET", + UpstreamPath: "/v1/poimen/executions", + Auth: &Auth{Required: true, Capability: "workflow:read"}, + }, + }, + }, + { + Name: "execution-detail", + Methods: []Method{ + { + Verb: "GET", + UpstreamPath: "/v1/poimen/executions/{id}", + Auth: &Auth{Required: true, Capability: "workflow:read"}, + }, + { + Verb: "DELETE", + UpstreamPath: "/v1/poimen/executions/{id}", + Auth: &Auth{Required: true, Capability: "workflow:execute"}, + }, + }, + }, + { + Name: "execution-history", + Methods: []Method{{ + Verb: "GET", + UpstreamPath: "/v1/poimen/executions/{id}/history", + Auth: &Auth{Required: true, Capability: "workflow:read"}, + }}, + }, + { + Name: "execution-stream", + Methods: []Method{{ + Verb: "GET", + UpstreamPath: "/v1/poimen/executions/{id}/stream", + Auth: &Auth{Required: true, Capability: "workflow:read"}, + }}, + }, + { + Name: "execution-signal", + Methods: []Method{{ + Verb: "POST", + UpstreamPath: "/v1/poimen/executions/{id}/signal", + Auth: &Auth{Required: true, Capability: "workflow:execute"}, + }}, + }, + }, + } +} diff --git a/internal/temporal/handler.go b/internal/temporal/handler.go index 7e9d076..1a452ea 100644 --- a/internal/temporal/handler.go +++ b/internal/temporal/handler.go @@ -10,7 +10,6 @@ import ( "time" "go.temporal.io/api/common/v1" - query "go.temporal.io/api/query/v1" "go.temporal.io/api/taskqueue/v1" update "go.temporal.io/api/update/v1" "go.temporal.io/api/workflowservice/v1" @@ -38,6 +37,8 @@ type ResponsePayload struct { type Handler struct { hostPort string grpcClient *GRPCClient + ops *WorkflowGRPCImpl + searchOps *SearchAttributesGRPCImpl } // NewHandler creates a new Temporal HTTP handler @@ -51,10 +52,15 @@ func NewHandler(hostPort string) *Handler { log.Printf("WARNING: Failed to connect to Temporal at %s: %v", hostPort, err) } - return &Handler{ + h := &Handler{ hostPort: hostPort, grpcClient: grpcClient, } + if grpcClient != nil { + h.ops = NewWorkflowGRPCImpl(grpcClient) + h.searchOps = NewSearchAttributesGRPCImpl(grpcClient) + } + return h } // unavailable returns TEMPORAL_UNAVAILABLE when grpcClient is nil @@ -208,7 +214,7 @@ func (h *Handler) handleMetrics(w http.ResponseWriter, r *http.Request) { // ── Workflow operations ────────────────────────────────────────────────────── func (h *Handler) startWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -223,31 +229,16 @@ func (h *Handler) startWorkflow(ctx context.Context, namespace string, payload m if taskQueue == "" { return nil, "INVALID_REQUEST", "task_queue is required" } - - req := &workflowservice.StartWorkflowExecutionRequest{ - Namespace: namespace, - WorkflowId: workflowID, - WorkflowType: &common.WorkflowType{Name: workflowType}, - TaskQueue: &taskqueue.TaskQueue{Name: taskQueue}, - } - if input := getMap(payload, "input"); len(input) > 0 { - b, _ := json.Marshal(input) - req.Input = &common.Payloads{Payloads: []*common.Payload{{Data: b}}} - } - - resp, err := h.grpcClient.GetWorkflowServiceStub().StartWorkflowExecution(ctx, req) + input := getMap(payload, "input") + result, err := h.ops.StartWorkflowExecution(ctx, namespace, workflowID, workflowType, taskQueue, input) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to start workflow: %v", err) } - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": resp.RunId, - "started_at": time.Now(), - }, "", "" + return result, "", "" } func (h *Handler) describeWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -255,25 +246,11 @@ func (h *Handler) describeWorkflow(ctx context.Context, namespace string, payloa return nil, "INVALID_REQUEST", "workflow_id is required" } runID := getString(payload, "run_id") - - resp, err := h.grpcClient.GetWorkflowServiceStub().DescribeWorkflowExecution(ctx, - &workflowservice.DescribeWorkflowExecutionRequest{ - Namespace: namespace, - Execution: &common.WorkflowExecution{WorkflowId: workflowID, RunId: runID}, - }) + result, err := h.ops.DescribeWorkflowExecution(ctx, namespace, workflowID, runID) if err != nil { return nil, "WORKFLOW_NOT_FOUND", fmt.Sprintf("failed to describe workflow: %v", err) } - - status := "UNKNOWN" - if resp.WorkflowExecutionInfo != nil { - status = resp.WorkflowExecutionInfo.Status.String() - } - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": runID, - "status": status, - }, "", "" + return result, "", "" } func (h *Handler) listWorkflows(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { @@ -319,7 +296,7 @@ func (h *Handler) listWorkflows(ctx context.Context, namespace string, payload m } func (h *Handler) getWorkflowHistory(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -327,79 +304,48 @@ func (h *Handler) getWorkflowHistory(ctx context.Context, namespace string, payl return nil, "INVALID_REQUEST", "workflow_id is required" } runID := getString(payload, "run_id") - - resp, err := h.grpcClient.GetWorkflowServiceStub().GetWorkflowExecutionHistory(ctx, - &workflowservice.GetWorkflowExecutionHistoryRequest{ - Namespace: namespace, - Execution: &common.WorkflowExecution{WorkflowId: workflowID, RunId: runID}, - }) + result, err := h.ops.GetWorkflowExecutionHistory(ctx, namespace, workflowID, runID) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to get history: %v", err) } - - events := make([]interface{}, 0, len(resp.History.GetEvents())) - for _, e := range resp.History.GetEvents() { - events = append(events, map[string]interface{}{ - "event_id": e.EventId, - "event_type": e.EventType.String(), - }) - } - return map[string]interface{}{ - "events": events, - "next_page_token": string(resp.NextPageToken), - }, "", "" + return result, "", "" } func (h *Handler) terminateWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } + runID := getString(payload, "run_id") reason := getString(payload, "reason") - - _, err := h.grpcClient.GetWorkflowServiceStub().TerminateWorkflowExecution(ctx, - &workflowservice.TerminateWorkflowExecutionRequest{ - Namespace: namespace, - WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID}, - Reason: reason, - }) + result, err := h.ops.TerminateWorkflowExecution(ctx, namespace, workflowID, runID, reason) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to terminate workflow: %v", err) } - return map[string]interface{}{ - "workflow_id": workflowID, - "terminated_at": time.Now(), - }, "", "" + return result, "", "" } func (h *Handler) cancelWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") if workflowID == "" { return nil, "INVALID_REQUEST", "workflow_id is required" } - - _, err := h.grpcClient.GetWorkflowServiceStub().RequestCancelWorkflowExecution(ctx, - &workflowservice.RequestCancelWorkflowExecutionRequest{ - Namespace: namespace, - WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID}, - }) + runID := getString(payload, "run_id") + result, err := h.ops.CancelWorkflowExecution(ctx, namespace, workflowID, runID) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to cancel workflow: %v", err) } - return map[string]interface{}{ - "workflow_id": workflowID, - "status": "canceling", - }, "", "" + return result, "", "" } func (h *Handler) signalWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -410,30 +356,17 @@ func (h *Handler) signalWorkflow(ctx context.Context, namespace string, payload if signalName == "" { return nil, "INVALID_REQUEST", "signal_name is required" } - - req := &workflowservice.SignalWorkflowExecutionRequest{ - Namespace: namespace, - WorkflowExecution: &common.WorkflowExecution{WorkflowId: workflowID}, - SignalName: signalName, - } - if data := getMap(payload, "signal_data"); len(data) > 0 { - b, _ := json.Marshal(data) - req.Input = &common.Payloads{Payloads: []*common.Payload{{Data: b}}} - } - - _, err := h.grpcClient.GetWorkflowServiceStub().SignalWorkflowExecution(ctx, req) + runID := getString(payload, "run_id") + signalData := getMap(payload, "signal_data") + result, err := h.ops.SignalWorkflowExecution(ctx, namespace, workflowID, runID, signalName, signalData) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to signal workflow: %v", err) } - return map[string]interface{}{ - "workflow_id": workflowID, - "signal_name": signalName, - "signaled_at": time.Now(), - }, "", "" + return result, "", "" } func (h *Handler) queryWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { - if h.grpcClient == nil { + if h.ops == nil { return h.unavailable() } workflowID := getString(payload, "workflow_id") @@ -444,22 +377,12 @@ func (h *Handler) queryWorkflow(ctx context.Context, namespace string, payload m if queryType == "" { return nil, "INVALID_REQUEST", "query_type is required" } - - resp, err := h.grpcClient.GetWorkflowServiceStub().QueryWorkflow(ctx, - &workflowservice.QueryWorkflowRequest{ - Namespace: namespace, - Execution: &common.WorkflowExecution{WorkflowId: workflowID}, - Query: &query.WorkflowQuery{QueryType: queryType}, - }) + runID := getString(payload, "run_id") + result, err := h.ops.QueryWorkflowExecution(ctx, namespace, workflowID, runID, queryType) if err != nil { return nil, "TEMPORAL_UNAVAILABLE", fmt.Sprintf("failed to query workflow: %v", err) } - - var result interface{} - if resp.QueryResult != nil && len(resp.QueryResult.Payloads) > 0 { - json.Unmarshal(resp.QueryResult.Payloads[0].Data, &result) - } - return map[string]interface{}{"query_result": result}, "", "" + return result, "", "" } func (h *Handler) resetWorkflow(ctx context.Context, namespace string, payload map[string]interface{}) (interface{}, string, string) { diff --git a/internal/temporal/operations.go b/internal/temporal/operations.go deleted file mode 100644 index 43e82d8..0000000 --- a/internal/temporal/operations.go +++ /dev/null @@ -1,95 +0,0 @@ -// Package temporal provides operation wrappers for Temporal operations -package temporal - -import ( - "context" - "fmt" - "time" -) - -// OperationHandler handles Temporal operations -type OperationHandler struct { - grpc *GRPCClient -} - -// NewOperationHandler creates a new operation handler -func NewOperationHandler(grpcClient *GRPCClient) *OperationHandler { - return &OperationHandler{ - grpc: grpcClient, - } -} - -// StartWorkflowExecution starts a new workflow execution -func (oh *OperationHandler) StartWorkflowExecution(ctx context.Context, namespace, workflowID, workflowType, taskQueue string, input map[string]interface{}) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - // This is a placeholder for actual implementation - if oh.grpc == nil { - return nil, fmt.Errorf("gRPC client not initialized") - } - - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": fmt.Sprintf("run_%d", time.Now().UnixNano()), - "start_time": time.Now(), - }, nil -} - -// DescribeWorkflowExecution gets workflow details -func (oh *OperationHandler) DescribeWorkflowExecution(ctx context.Context, namespace, workflowID, runID string) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": runID, - "status": "RUNNING", - "start_time": time.Now(), - }, nil -} - -// TerminateWorkflowExecution terminates a workflow -func (oh *OperationHandler) TerminateWorkflowExecution(ctx context.Context, namespace, workflowID, runID, reason string) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "run_id": runID, - "terminated_at": time.Now(), - "reason": reason, - }, nil -} - -// CancelWorkflowExecution cancels a workflow -func (oh *OperationHandler) CancelWorkflowExecution(ctx context.Context, namespace, workflowID, runID string) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "status": "canceling", - }, nil -} - -// SignalWorkflowExecution sends a signal to a workflow -func (oh *OperationHandler) SignalWorkflowExecution(ctx context.Context, namespace, workflowID, runID, signalName string, input map[string]interface{}) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "signal_name": signalName, - "signaled_at": time.Now(), - }, nil -} - -// QueryWorkflowExecution queries a workflow -func (oh *OperationHandler) QueryWorkflowExecution(ctx context.Context, namespace, workflowID, runID, queryType string) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "workflow_id": workflowID, - "query_type": queryType, - "query_result": map[string]interface{}{}, - }, nil -} - -// ListWorkflowExecutions lists workflows -func (oh *OperationHandler) ListWorkflowExecutions(ctx context.Context, namespace string, pageSize int32) (map[string]interface{}, error) { - // TODO: Implement gRPC call to Temporal - return map[string]interface{}{ - "executions": []interface{}{}, - "next_page_token": "", - }, nil -} diff --git a/internal/temporal/poimen.go b/internal/temporal/poimen.go new file mode 100644 index 0000000..a6b35a3 --- /dev/null +++ b/internal/temporal/poimen.go @@ -0,0 +1,351 @@ +package temporal + +import ( + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +// PoimenHandler provides typed REST endpoints for Poimen workflows. +// Accessed via X-Service: poimen header through the ServiceAdapter dispatcher. +type PoimenHandler struct { + handler *Handler +} + +// NewPoimenHandler creates a new Poimen handler wrapping the Temporal handler. +func NewPoimenHandler(h *Handler) *PoimenHandler { + return &PoimenHandler{handler: h} +} + +// ServeHTTP routes based on X-Upstream-Path set by the dispatcher. +func (p *PoimenHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + path := r.Header.Get("X-Upstream-Path") + if path == "" { + path = r.URL.Path + } + + switch { + case path == "/v1/poimen/plan" && r.Method == http.MethodPost: + p.handlePlan(w, r) + case path == "/v1/poimen/execute" && r.Method == http.MethodPost: + p.handleExecute(w, r) + case path == "/v1/poimen/executions" && r.Method == http.MethodGet: + p.handleListExecutions(w, r) + case strings.HasPrefix(path, "/v1/poimen/executions/") && strings.HasSuffix(path, "/stream") && r.Method == http.MethodGet: + p.handleStream(w, r) + case strings.HasPrefix(path, "/v1/poimen/executions/") && strings.HasSuffix(path, "/signal") && r.Method == http.MethodPost: + p.handleSignal(w, r) + case strings.HasPrefix(path, "/v1/poimen/executions/") && strings.HasSuffix(path, "/history") && r.Method == http.MethodGet: + p.handleHistory(w, r) + case strings.HasPrefix(path, "/v1/poimen/executions/") && r.Method == http.MethodDelete: + p.handleTerminate(w, r) + case strings.HasPrefix(path, "/v1/poimen/executions/") && r.Method == http.MethodGet: + p.handleDescribe(w, r) + default: + writeJSON(w, http.StatusNotFound, map[string]string{"error": "endpoint not found"}) + } +} + +// extractWorkflowID extracts the workflow ID from paths like /v1/poimen/executions/{id}/... +func extractWorkflowID(path string) string { + parts := strings.Split(strings.TrimPrefix(path, "/v1/poimen/executions/"), "/") + if len(parts) > 0 && parts[0] != "" { + return parts[0] + } + return "" +} + +// PoimenWorkflowTypes maps Poimen workflow names to Temporal workflow types and task queues. +var PoimenWorkflowTypes = map[string]struct { + WorkflowType string + TaskQueue string +}{ + "dag": {WorkflowType: "DAGWorkflow", TaskQueue: "poimen-worker"}, + "orchestrator": {WorkflowType: "OrchestratorWorkflow", TaskQueue: "poimen-worker"}, + "routing": {WorkflowType: "RoutingWorkflow", TaskQueue: "poimen-worker"}, +} + +// --- Plan --- + +type PlanRequest struct { + Message string `json:"message"` + Context map[string]interface{} `json:"context,omitempty"` +} + +func (p *PoimenHandler) handlePlan(w http.ResponseWriter, r *http.Request) { + var req PlanRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"}) + return + } + if req.Message == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "message is required"}) + return + } + + // TODO: Call PlannerActivity directly (not via Temporal) for low latency. + // For now, return a placeholder indicating the plan endpoint is wired. + writeJSON(w, http.StatusNotImplemented, map[string]interface{}{ + "error": "PlannerActivity not yet available", + "message": req.Message, + "context": req.Context, + "hint": "Depends on poimen-workflows#23 (PlannerActivity)", + }) +} + +// --- Execute --- + +type ExecuteRequest struct { + WorkflowType string `json:"workflow_type"` // "dag", "orchestrator", "routing" + Namespace string `json:"namespace,omitempty"` + Input map[string]interface{} `json:"input"` +} + +func (p *PoimenHandler) handleExecute(w http.ResponseWriter, r *http.Request) { + var req ExecuteRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"}) + return + } + + if req.Namespace == "" { + req.Namespace = "default" + } + + wfType, ok := PoimenWorkflowTypes[req.WorkflowType] + if !ok { + writeJSON(w, http.StatusBadRequest, map[string]interface{}{ + "error": fmt.Sprintf("unknown workflow_type: %s", req.WorkflowType), + "valid_types": []string{"dag", "orchestrator", "routing"}, + }) + return + } + + if p.handler.ops == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Temporal not available"}) + return + } + + workflowID := fmt.Sprintf("poimen-%s-%d", req.WorkflowType, time.Now().UnixMilli()) + + result, err := p.handler.ops.StartWorkflowExecution( + r.Context(), + req.Namespace, + workflowID, + wfType.WorkflowType, + wfType.TaskQueue, + req.Input, + ) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": fmt.Sprintf("failed to start workflow: %v", err)}) + return + } + + writeJSON(w, http.StatusOK, result) +} + +// --- List Executions --- + +func (p *PoimenHandler) handleListExecutions(w http.ResponseWriter, r *http.Request) { + if p.handler.ops == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Temporal not available"}) + return + } + + namespace := r.URL.Query().Get("namespace") + if namespace == "" { + namespace = "default" + } + + result, err := p.handler.ops.ListWorkflowExecutions(r.Context(), namespace, 100) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": fmt.Sprintf("failed to list: %v", err)}) + return + } + + writeJSON(w, http.StatusOK, result) +} + +// --- Describe --- + +func (p *PoimenHandler) handleDescribe(w http.ResponseWriter, r *http.Request) { + if p.handler.ops == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Temporal not available"}) + return + } + + path := r.Header.Get("X-Upstream-Path") + if path == "" { + path = r.URL.Path + } + workflowID := extractWorkflowID(path) + if workflowID == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "workflow_id required in path"}) + return + } + + namespace := r.URL.Query().Get("namespace") + if namespace == "" { + namespace = "default" + } + runID := r.URL.Query().Get("run_id") + + result, err := p.handler.ops.DescribeWorkflowExecution(r.Context(), namespace, workflowID, runID) + if err != nil { + writeJSON(w, http.StatusNotFound, map[string]string{"error": fmt.Sprintf("workflow not found: %v", err)}) + return + } + + writeJSON(w, http.StatusOK, result) +} + +// --- History --- + +func (p *PoimenHandler) handleHistory(w http.ResponseWriter, r *http.Request) { + if p.handler.ops == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Temporal not available"}) + return + } + + path := r.Header.Get("X-Upstream-Path") + if path == "" { + path = r.URL.Path + } + workflowID := extractWorkflowID(path) + if workflowID == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "workflow_id required in path"}) + return + } + + namespace := r.URL.Query().Get("namespace") + if namespace == "" { + namespace = "default" + } + runID := r.URL.Query().Get("run_id") + + result, err := p.handler.ops.GetWorkflowExecutionHistory(r.Context(), namespace, workflowID, runID) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": fmt.Sprintf("failed to get history: %v", err)}) + return + } + + writeJSON(w, http.StatusOK, result) +} + +// --- Signal --- + +type SignalRequest struct { + SignalName string `json:"signal_name"` + Data map[string]interface{} `json:"data,omitempty"` + Namespace string `json:"namespace,omitempty"` +} + +func (p *PoimenHandler) handleSignal(w http.ResponseWriter, r *http.Request) { + if p.handler.ops == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Temporal not available"}) + return + } + + path := r.Header.Get("X-Upstream-Path") + if path == "" { + path = r.URL.Path + } + // Remove /signal suffix to get workflow ID + cleanPath := strings.TrimSuffix(path, "/signal") + workflowID := extractWorkflowID(cleanPath) + if workflowID == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "workflow_id required in path"}) + return + } + + var req SignalRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"}) + return + } + if req.SignalName == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "signal_name is required"}) + return + } + if req.Namespace == "" { + req.Namespace = "default" + } + + result, err := p.handler.ops.SignalWorkflowExecution(r.Context(), req.Namespace, workflowID, "", req.SignalName, req.Data) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": fmt.Sprintf("failed to signal: %v", err)}) + return + } + + writeJSON(w, http.StatusOK, result) +} + +// --- Terminate --- + +func (p *PoimenHandler) handleTerminate(w http.ResponseWriter, r *http.Request) { + if p.handler.ops == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "Temporal not available"}) + return + } + + path := r.Header.Get("X-Upstream-Path") + if path == "" { + path = r.URL.Path + } + workflowID := extractWorkflowID(path) + if workflowID == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "workflow_id required in path"}) + return + } + + namespace := r.URL.Query().Get("namespace") + if namespace == "" { + namespace = "default" + } + + result, err := p.handler.ops.TerminateWorkflowExecution(r.Context(), namespace, workflowID, "", "terminated via API") + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": fmt.Sprintf("failed to terminate: %v", err)}) + return + } + + writeJSON(w, http.StatusOK, result) +} + +// --- Stream (delegate to existing SSE handler) --- + +func (p *PoimenHandler) handleStream(w http.ResponseWriter, r *http.Request) { + path := r.Header.Get("X-Upstream-Path") + if path == "" { + path = r.URL.Path + } + // Remove /stream suffix + cleanPath := strings.TrimSuffix(path, "/stream") + workflowID := extractWorkflowID(cleanPath) + if workflowID == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "workflow_id required in path"}) + return + } + + // Set query params for the SSE handler + q := r.URL.Query() + q.Set("workflow_id", workflowID) + if q.Get("namespace") == "" { + q.Set("namespace", "default") + } + r.URL.RawQuery = q.Encode() + + p.handler.StreamWorkflowEvents(w, r) +} + +// --- Helpers --- + +func writeJSON(w http.ResponseWriter, status int, data interface{}) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(data) //nolint:errcheck +} + +