package api import ( "context" "encoding/json" "fmt" "log" "net/http" "time" "github.com/google/uuid" "go.temporal.io/sdk/client" "github.com/rockliang/poimen/workflows/internal/routing" "github.com/rockliang/poimen/workflows/pkg/db" ) // WorkflowNode matches frontend node type type WorkflowNode struct { ID string `json:"id"` Type string `json:"type"` // "activity", "start", "end" Position map[string]interface{} `json:"position"` Data struct { Label string `json:"label"` Activity string `json:"activity"` Config map[string]interface{} `json:"config"` } `json:"data"` } // WorkflowEdge matches frontend edge type type WorkflowEdge struct { ID string `json:"id"` Source string `json:"source"` Target string `json:"target"` Data map[string]interface{} `json:"data,omitempty"` } // WorkflowDef is the request body for creating/updating workflows type WorkflowDef struct { Name string `json:"name"` Description string `json:"description"` Nodes []WorkflowNode `json:"nodes"` Edges []WorkflowEdge `json:"edges"` Status string `json:"status"` // "draft", "active" } // WorkflowResponse is the workflow with metadata type WorkflowResponse struct { ID string `json:"id"` Name string `json:"name"` Description string `json:"description"` Status string `json:"status"` Version int `json:"version"` Nodes []WorkflowNode `json:"nodes"` Edges []WorkflowEdge `json:"edges"` CreatedAt string `json:"createdAt"` UpdatedAt string `json:"updatedAt"` CreatedBy string `json:"createdBy"` } // ExecutionRequest is the request to execute a workflow type ExecutionRequest struct { Inputs map[string]interface{} `json:"inputs"` } // ExecutionResponse is the execution result type ExecutionResponse struct { ID string `json:"id"` WorkflowID string `json:"workflowId"` Status string `json:"status"` // "pending", "running", "success", "failed" StartedAt string `json:"startedAt"` CompletedAt string `json:"completedAt,omitempty"` Inputs map[string]interface{} `json:"inputs"` Outputs map[string]interface{} `json:"outputs,omitempty"` Errors []string `json:"errors,omitempty"` Logs []ExecutionLog `json:"logs"` } // ExecutionLog is a log entry from execution type ExecutionLog struct { Timestamp string `json:"timestamp"` NodeID string `json:"nodeId"` Level string `json:"level"` // "info", "warn", "error" Message string `json:"message"` } // WorkflowAPI handles workflow endpoints type WorkflowAPI struct { db *db.DB temporalClient client.Client logger *log.Logger customerID string // TODO: Extract from JWT token } // NewWorkflowAPI creates new API handler func NewWorkflowAPI(database *db.DB, tc client.Client, logger *log.Logger) *WorkflowAPI { return &WorkflowAPI{ db: database, temporalClient: tc, logger: logger, customerID: "default-customer", // TODO: From auth context } } // CreateWorkflow handles POST /workflows func (api *WorkflowAPI) CreateWorkflow(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } var req WorkflowDef if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, fmt.Sprintf("Invalid request: %v", err), http.StatusBadRequest) return } if req.Name == "" { http.Error(w, "Workflow name required", http.StatusBadRequest) return } // Create workflow in database id := uuid.New().String() now := time.Now() // Convert nodes and edges to JSONB nodesJSON, err := json.Marshal(req.Nodes) if err != nil { http.Error(w, fmt.Sprintf("Failed to marshal nodes: %v", err), http.StatusBadRequest) return } edgesJSON, err := json.Marshal(req.Edges) if err != nil { http.Error(w, fmt.Sprintf("Failed to marshal edges: %v", err), http.StatusBadRequest) return } status := req.Status if status == "" { status = "draft" } workflow := &db.Workflow{ ID: id, CustomerID: api.customerID, Name: req.Name, Description: req.Description, Status: status, Version: 1, Nodes: nodesJSON, Edges: edgesJSON, CreatedBy: "anonymous", // Use JWT claim in real implementation CreatedAt: now, UpdatedAt: now, } if err := api.db.SaveWorkflow(r.Context(), workflow); err != nil { api.logger.Printf("Failed to save workflow: %v", err) http.Error(w, "Failed to create workflow", http.StatusInternalServerError) return } response := WorkflowResponse{ ID: workflow.ID, Name: workflow.Name, Description: workflow.Description, Status: workflow.Status, Version: workflow.Version, Nodes: req.Nodes, Edges: req.Edges, CreatedAt: workflow.CreatedAt.Format(time.RFC3339), UpdatedAt: workflow.UpdatedAt.Format(time.RFC3339), CreatedBy: workflow.CreatedBy, } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusCreated) json.NewEncoder(w).Encode(response) } // ListWorkflows handles GET /workflows func (api *WorkflowAPI) ListWorkflows(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } page := 1 limit := 10 // Parse pagination params if needed workflows, err := api.db.ListWorkflows(r.Context(), api.customerID, limit, (page-1)*limit) if err != nil { api.logger.Printf("Failed to list workflows: %v", err) http.Error(w, "Failed to list workflows", http.StatusInternalServerError) return } list := make([]WorkflowResponse, 0) for _, wf := range workflows { var nodes []WorkflowNode var edges []WorkflowEdge json.Unmarshal(wf.Nodes, &nodes) json.Unmarshal(wf.Edges, &edges) list = append(list, WorkflowResponse{ ID: wf.ID, Name: wf.Name, Description: wf.Description, Status: wf.Status, Version: wf.Version, Nodes: nodes, Edges: edges, CreatedAt: wf.CreatedAt.Format(time.RFC3339), UpdatedAt: wf.UpdatedAt.Format(time.RFC3339), CreatedBy: wf.CreatedBy, }) } response := map[string]interface{}{ "workflows": list, "total": len(list), "page": page, "limit": limit, } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(response) } // GetWorkflow handles GET /workflows/{id} func (api *WorkflowAPI) GetWorkflow(w http.ResponseWriter, r *http.Request, id string) { if r.Method != http.MethodGet { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } workflow, err := api.db.FetchWorkflow(r.Context(), id, api.customerID) if err != nil { http.Error(w, "Workflow not found", http.StatusNotFound) return } var nodes []WorkflowNode var edges []WorkflowEdge json.Unmarshal(workflow.Nodes, &nodes) json.Unmarshal(workflow.Edges, &edges) response := WorkflowResponse{ ID: workflow.ID, Name: workflow.Name, Description: workflow.Description, Status: workflow.Status, Version: workflow.Version, Nodes: nodes, Edges: edges, CreatedAt: workflow.CreatedAt.Format(time.RFC3339), UpdatedAt: workflow.UpdatedAt.Format(time.RFC3339), CreatedBy: workflow.CreatedBy, } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(response) } // UpdateWorkflow handles PUT /workflows/{id} func (api *WorkflowAPI) UpdateWorkflow(w http.ResponseWriter, r *http.Request, id string) { if r.Method != http.MethodPut { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } // Fetch existing workflow workflow, err := api.db.FetchWorkflow(r.Context(), id, api.customerID) if err != nil { http.Error(w, "Workflow not found", http.StatusNotFound) return } var req WorkflowDef if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, fmt.Sprintf("Invalid request: %v", err), http.StatusBadRequest) return } // Update fields if req.Name != "" { workflow.Name = req.Name } if req.Description != "" { workflow.Description = req.Description } if req.Nodes != nil { nodesJSON, _ := json.Marshal(req.Nodes) workflow.Nodes = nodesJSON } if req.Edges != nil { edgesJSON, _ := json.Marshal(req.Edges) workflow.Edges = edgesJSON } if req.Status != "" { workflow.Status = req.Status } workflow.Version++ workflow.UpdatedAt = time.Now() if err := api.db.SaveWorkflow(r.Context(), workflow); err != nil { api.logger.Printf("Failed to update workflow: %v", err) http.Error(w, "Failed to update workflow", http.StatusInternalServerError) return } var nodes []WorkflowNode var edges []WorkflowEdge json.Unmarshal(workflow.Nodes, &nodes) json.Unmarshal(workflow.Edges, &edges) response := WorkflowResponse{ ID: workflow.ID, Name: workflow.Name, Description: workflow.Description, Status: workflow.Status, Version: workflow.Version, Nodes: nodes, Edges: edges, CreatedAt: workflow.CreatedAt.Format(time.RFC3339), UpdatedAt: workflow.UpdatedAt.Format(time.RFC3339), CreatedBy: workflow.CreatedBy, } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(response) } // DeleteWorkflow handles DELETE /workflows/{id} func (api *WorkflowAPI) DeleteWorkflow(w http.ResponseWriter, r *http.Request, id string) { if r.Method != http.MethodDelete { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } if err := api.db.DeleteWorkflow(r.Context(), id, api.customerID); err != nil { http.Error(w, "Workflow not found", http.StatusNotFound) return } w.WriteHeader(http.StatusNoContent) } // ExecuteWorkflow handles POST /workflows/{id}/execute func (api *WorkflowAPI) ExecuteWorkflow(w http.ResponseWriter, r *http.Request, id string) { if r.Method != http.MethodPost { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } workflow, err := api.db.FetchWorkflow(r.Context(), id, api.customerID) if err != nil { http.Error(w, "Workflow not found", http.StatusNotFound) return } var req ExecutionRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, fmt.Sprintf("Invalid request: %v", err), http.StatusBadRequest) return } // Unmarshal nodes and edges var nodes []WorkflowNode var edges []WorkflowEdge json.Unmarshal(workflow.Nodes, &nodes) json.Unmarshal(workflow.Edges, &edges) // Convert to workflow response for spec conversion workflowResp := &WorkflowResponse{ ID: workflow.ID, Name: workflow.Name, Description: workflow.Description, Status: workflow.Status, Version: workflow.Version, Nodes: nodes, Edges: edges, CreatedBy: workflow.CreatedBy, } // Convert nodes/edges to WorkflowSpec spec := api.nodesToWorkflowSpec(workflowResp, req.Inputs) // Execute via Temporal RoutingWorkflow execID := uuid.New().String() workflowOptions := client.StartWorkflowOptions{ ID: execID, TaskQueue: "default", } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() _, err = api.temporalClient.ExecuteWorkflow(ctx, workflowOptions, "RoutingWorkflow", spec) if err != nil { api.logger.Printf("Failed to execute workflow: %v", err) http.Error(w, fmt.Sprintf("Execution failed: %v", err), http.StatusInternalServerError) return } // Save execution to database inputsJSON, _ := json.Marshal(req.Inputs) now := time.Now() execution := &db.WorkflowExecution{ ID: execID, WorkflowID: id, CustomerID: api.customerID, TemporalID: execID, Status: "running", Inputs: inputsJSON, StartedAt: now, } if err := api.db.SaveExecution(r.Context(), execution); err != nil { api.logger.Printf("Failed to save execution: %v", err) http.Error(w, "Failed to save execution", http.StatusInternalServerError) return } // Create execution response execResp := ExecutionResponse{ ID: execID, WorkflowID: id, Status: "running", StartedAt: now.Format(time.RFC3339), Inputs: req.Inputs, Outputs: make(map[string]interface{}), Logs: []ExecutionLog{}, } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusCreated) json.NewEncoder(w).Encode(execResp) } // GetExecution handles GET /executions/{id} func (api *WorkflowAPI) GetExecution(w http.ResponseWriter, r *http.Request, id string) { if r.Method != http.MethodGet { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } execution, err := api.db.FetchExecution(r.Context(), id) if err != nil { http.Error(w, "Execution not found", http.StatusNotFound) return } // Get logs from database logs, err := api.db.FetchExecutionLogs(r.Context(), id) if err != nil { api.logger.Printf("Failed to fetch logs: %v", err) } execLogs := make([]ExecutionLog, 0) for _, log := range logs { execLogs = append(execLogs, ExecutionLog{ Timestamp: log.LoggedAt.Format(time.RFC3339), NodeID: log.NodeID, Level: log.Level, Message: log.Message, }) } // Parse inputs/outputs var inputs map[string]interface{} var outputs map[string]interface{} json.Unmarshal(execution.Inputs, &inputs) if execution.Outputs != nil { json.Unmarshal(execution.Outputs, &outputs) } // Check Temporal workflow status ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() desc, err := api.temporalClient.DescribeWorkflowExecution(ctx, execution.TemporalID, "") status := execution.Status if err == nil && desc != nil { switch desc.Status.String() { case "RUNNING": status = "running" case "COMPLETED": status = "success" case "FAILED": status = "failed" } } completedAtStr := "" if execution.CompletedAt != nil { completedAtStr = execution.CompletedAt.Format(time.RFC3339) } execResp := ExecutionResponse{ ID: execution.ID, WorkflowID: execution.WorkflowID, Status: status, StartedAt: execution.StartedAt.Format(time.RFC3339), CompletedAt: completedAtStr, Inputs: inputs, Outputs: outputs, Logs: execLogs, } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(execResp) } // ListExecutions handles GET /workflows/{id}/executions func (api *WorkflowAPI) ListExecutions(w http.ResponseWriter, r *http.Request, workflowID string) { if r.Method != http.MethodGet { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } // TODO: Implement query by workflow_id in database // For now, return empty list (needs DB method for filtering by workflow_id) list := make([]ExecutionResponse, 0) w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(list) } // nodesToWorkflowSpec converts frontend nodes/edges to routing.WorkflowSpec func (api *WorkflowAPI) nodesToWorkflowSpec(wf *WorkflowResponse, inputs map[string]interface{}) *routing.WorkflowSpec { spec := &routing.WorkflowSpec{ Name: wf.Name, Input: inputs, States: []routing.State{}, } // Build states from nodes stateMap := make(map[string]*routing.State) // Create all states for _, node := range wf.Nodes { if node.Type == "activity" { state := &routing.State{ Name: node.ID, Type: routing.StateTypeTask, Resource: node.Data.Activity, Parameters: node.Data.Config, End: false, } stateMap[node.ID] = state spec.States = append(spec.States, *state) } } // Wire edges (transitions) for _, edge := range wf.Edges { if state, exists := stateMap[edge.Source]; exists { state.Next = edge.Target } } // Mark last state as End if len(spec.States) > 0 { // Find state with no outgoing edge for i := range spec.States { hasNext := false for _, edge := range wf.Edges { if edge.Source == spec.States[i].Name { hasNext = true break } } if !hasNext { spec.States[i].End = true } } } return spec } // QueryWorkflowGraph handles POST /workflows/{id}/query func (api *WorkflowAPI) QueryWorkflowGraph(w http.ResponseWriter, r *http.Request, workflowID string) { ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second) defer cancel() var req QueryWorkflowGraphRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "Invalid request body", http.StatusBadRequest) return } // Defaults if req.SearchType == "" { req.SearchType = "edges" } if req.ConfidenceFloor == 0 { req.ConfidenceFloor = 0.5 } if req.TopK == 0 { req.TopK = 10 } if req.MaxPathDepth == 0 { req.MaxPathDepth = 3 } if req.RankingProfile == "" { req.RankingProfile = "default" } // Get latest version if not specified if req.Version == 0 { wf, err := api.db.GetWorkflow(ctx, workflowID) if err != nil { http.Error(w, "Workflow not found", http.StatusNotFound) return } req.Version = wf.Version } // Call temporal workflow run, err := api.temporalClient.ExecuteWorkflow( ctx, client.StartWorkflowOptions{ ID: fmt.Sprintf("graph-query-%s-v%d", workflowID, req.Version), TaskQueue: "poimen", }, "WorkflowGraphQuery", map[string]interface{}{ "workflow_id": workflowID, "query": req.Query, "search_type": req.SearchType, "relation_type": req.RelationType, "version": req.Version, "confidence_floor": req.ConfidenceFloor, "top_k": req.TopK, "find_paths": req.FindPaths, "target_node_id": req.TargetNodeID, "max_path_depth": req.MaxPathDepth, "ranking_profile": req.RankingProfile, "include_reasoning": req.IncludeReasoning, }, ) if err != nil { api.logger.Printf("Failed to start workflow: %v", err) http.Error(w, "Failed to start query workflow", http.StatusInternalServerError) return } var result map[string]interface{} if err := run.Get(ctx, &result); err != nil { api.logger.Printf("Workflow execution failed: %v", err) http.Error(w, "Query execution failed", http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(result) } // QueryWorkflowGraphRequest matches frontend payload type QueryWorkflowGraphRequest struct { Query string `json:"query"` SearchType string `json:"search_type"` RelationType string `json:"relation_type"` Version int `json:"version"` ConfidenceFloor float64 `json:"confidence_floor"` TopK int `json:"top_k"` FindPaths bool `json:"find_paths"` TargetNodeID string `json:"target_node_id"` MaxPathDepth int `json:"max_path_depth"` RankingProfile string `json:"ranking_profile"` IncludeReasoning bool `json:"include_reasoning"` } // GetWorkflowRelationVersions handles GET /workflows/{id}/relations/{edge_id}/versions func (api *WorkflowAPI) GetWorkflowRelationVersions(w http.ResponseWriter, r *http.Request, workflowID, edgeID string) { ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second) defer cancel() // Query relation versions from DB versions, err := api.db.GetRelationVersions(ctx, workflowID, edgeID) if err != nil { api.logger.Printf("Failed to get relation versions: %v", err) http.Error(w, "Failed to fetch relation versions", http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{ "workflow_id": workflowID, "edge_id": edgeID, "versions": versions, "total_count": len(versions), }) }