590 lines
16 KiB
Go
590 lines
16 KiB
Go
|
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
|
||
|
|
}
|