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/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 +} + +