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 }