diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..01b76dd --- /dev/null +++ b/.env.example @@ -0,0 +1,22 @@ +# Workflows Backend Configuration + +# Database (memory-db CNPG in K8s) +# Option A: Direct DATABASE_URL +DATABASE_URL=postgresql://app:PASSWORD@memory-db-rw.poimen.svc.cluster.local:5432/memory?sslmode=disable + +# Option B: Individual env vars (used if DATABASE_URL is empty) +DATABASE_HOST=memory-db-rw.poimen.svc.cluster.local +DATABASE_PORT=5432 +DATABASE_NAME=memory +DATABASE_USER=app +DATABASE_PASSWORD=PASSWORD + +# Temporal +TEMPORAL_HOST_PORT=localhost:7233 +TEMPORAL_NAMESPACE=default + +# API Server +API_PORT=8080 + +# Logging +VERBOSE=false diff --git a/action/llm_inference.go b/action/llm_inference.go new file mode 100644 index 0000000..128f199 --- /dev/null +++ b/action/llm_inference.go @@ -0,0 +1,156 @@ +package action + +import ( + "context" + "fmt" + + "github.com/rockliang/poimen/workflows/action/llm" + "github.com/rockliang/poimen/workflows/statemachine" +) + +// LLMInferenceInput is input for LLMInferenceActivity +type LLMInferenceInput struct { + Model string `json:"model"` // Model ID (reasoning, ornith:35b, etc) + SystemPrompt string `json:"system_prompt"` // System instruction + UserPrompt string `json:"user_prompt"` // User message + Temperature float64 `json:"temperature,omitempty"` // LLM temperature (0-1) + MaxTokens int `json:"max_tokens,omitempty"` // Max output tokens +} + +// LLMInferenceOutput is output from LLMInferenceActivity +type LLMInferenceOutput struct { + Response string `json:"response"` // LLM response text + Model string `json:"model"` // Model used + StopReason string `json:"stop_reason"` // How inference stopped (stop_sequence, length, etc) + TokensUsed int `json:"tokens_used"` // Total tokens consumed + ErrorMessage string `json:"error,omitempty"` +} + +// LLMInferenceActivity calls LLM API with given prompt and returns response +func LLMInferenceActivity(ctx context.Context, in LLMInferenceInput) (LLMInferenceOutput, error) { + logger := newActivityLogger(ctx) + + output := LLMInferenceOutput{ + Model: in.Model, + } + + // Validate input + if in.Model == "" { + return output, fmt.Errorf("model not specified") + } + + if in.UserPrompt == "" { + return output, fmt.Errorf("user_prompt not specified") + } + + logger.logf("info", "Starting LLM inference with model: %s", in.Model) + + // Create LLM client + client, err := llm.NewClient() + if err != nil { + output.ErrorMessage = err.Error() + return output, fmt.Errorf("failed to create LLM client: %w", err) + } + + // Call LLM + logger.logf("info", "Calling LLM API (model=%s, prompt_len=%d)", in.Model, len(in.UserPrompt)) + + response, err := client.CreateMessage(ctx, llm.MessageInput{ + Model: statemachine.ModelSpec{ + ModelID: in.Model, + }, + SystemPrompt: in.SystemPrompt, + Messages: []llm.MessageParam{ + { + Role: "user", + Content: in.UserPrompt, + }, + }, + }) + + if err != nil { + output.ErrorMessage = err.Error() + logger.logf("error", "LLM API call failed: %v", err) + return output, fmt.Errorf("LLM inference failed: %w", err) + } + + output.Response = response + output.StopReason = "stop_sequence" + + logger.logf("info", "LLM inference completed (response_len=%d)", len(response)) + + return output, nil +} + +// LLMBatchInferenceInput is input for batch inference +type LLMBatchInferenceInput struct { + Model string `json:"model"` + SystemPrompt string `json:"system_prompt"` + Prompts []string `json:"prompts"` // List of user prompts + Temperature float64 `json:"temperature,omitempty"` +} + +// LLMBatchInferenceOutput is output from batch inference +type LLMBatchInferenceOutput struct { + Responses []string `json:"responses"` // LLM responses (parallel to input Prompts) + Model string `json:"model"` + Errors []string `json:"errors,omitempty"` +} + +// LLMBatchInferenceActivity calls LLM multiple times in sequence +func LLMBatchInferenceActivity(ctx context.Context, in LLMBatchInferenceInput) (LLMBatchInferenceOutput, error) { + logger := newActivityLogger(ctx) + + output := LLMBatchInferenceOutput{ + Model: in.Model, + Responses: []string{}, + Errors: []string{}, + } + + if in.Model == "" { + return output, fmt.Errorf("model not specified") + } + + if len(in.Prompts) == 0 { + return output, fmt.Errorf("no prompts provided") + } + + logger.logf("info", "Starting batch LLM inference (model=%s, count=%d)", in.Model, len(in.Prompts)) + + // Create LLM client + client, err := llm.NewClient() + if err != nil { + return output, fmt.Errorf("failed to create LLM client: %w", err) + } + + // Process each prompt + for i, prompt := range in.Prompts { + logger.logf("info", "Processing prompt %d/%d", i+1, len(in.Prompts)) + + response, err := client.CreateMessage(ctx, llm.MessageInput{ + Model: statemachine.ModelSpec{ + ModelID: in.Model, + }, + SystemPrompt: in.SystemPrompt, + Messages: []llm.MessageParam{ + { + Role: "user", + Content: prompt, + }, + }, + }) + + if err != nil { + output.Errors = append(output.Errors, fmt.Sprintf("prompt %d: %v", i, err)) + output.Responses = append(output.Responses, "") + logger.logf("warn", "Failed to process prompt %d: %v", i, err) + } else { + output.Responses = append(output.Responses, response) + } + } + + logger.logf("info", "Batch inference completed (responses=%d, errors=%d)", + len(output.Responses), len(output.Errors)) + + return output, nil +} diff --git a/cmd/server/main.go b/cmd/server/main.go new file mode 100644 index 0000000..7a6cdfc --- /dev/null +++ b/cmd/server/main.go @@ -0,0 +1,124 @@ +package main + +import ( + "context" + "flag" + "log" + "os" + "os/signal" + "sync" + "syscall" + + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/worker" + + "github.com/rockliang/poimen/workflows/action" + "github.com/rockliang/poimen/workflows/internal/api" + "github.com/rockliang/poimen/workflows/internal/config" + "github.com/rockliang/poimen/workflows/pkg/db" + "github.com/rockliang/poimen/workflows/statemachine" +) + +func main() { + var ( + apiPort = flag.Int("port", 8080, "HTTP API port") + verbose = flag.Bool("verbose", false, "verbose logging") + ) + flag.Parse() + + logger := log.New(os.Stdout, "[poimen-server] ", log.LstdFlags|log.Lshortfile) + + // Load configuration + cfg, err := config.LoadConfig() + if err != nil { + logger.Fatalf("failed to load config: %v", err) + } + + // Connect to database (memory-db via K8s CNPG) + logger.Println("connecting to database...") + database, err := db.New(os.Getenv("DATABASE_URL")) + if err != nil { + logger.Fatalf("failed to connect to database: %v", err) + } + defer database.Close() + logger.Println("✓ Connected to database") + + // Connect to Temporal + logger.Printf("connecting to Temporal at %s", cfg.Temporal.HostPort) + c, err := client.Dial(client.Options{ + HostPort: cfg.Temporal.HostPort, + Namespace: cfg.Temporal.Namespace, + }) + if err != nil { + logger.Fatalf("failed to connect to temporal: %v", err) + } + defer c.Close() + + logger.Println("✓ Connected to Temporal") + + // Create and start Temporal worker + w := worker.New(c, "default", worker.Options{}) + + // Register RoutingWorkflow + w.RegisterWorkflow(statemachine.RoutingWorkflow) + + // Register activities + w.RegisterActivity(action.CloneRepoActivity) + w.RegisterActivity(action.AnalyzeCodeActivity) + w.RegisterActivity(action.SecurityScanActivity) + w.RegisterActivity(action.GenerateReportActivity) + w.RegisterActivity(action.DeploymentPreCheckActivity) + w.RegisterActivity(action.NotifyStatusActivity) + w.RegisterActivity(action.ApproveWorkflowActivity) + w.RegisterActivity(action.ArchiveResultsActivity) + w.RegisterActivity(action.RetrieveMemoryActivity) + w.RegisterActivity(action.AssumeRoleActivity) + w.RegisterActivity(action.LLMInferenceActivity) + w.RegisterActivity(action.LLMBatchInferenceActivity) + + var wg sync.WaitGroup + errChan := make(chan error, 2) + + // Start Temporal worker + wg.Add(1) + go func() { + defer wg.Done() + logger.Println("starting Temporal worker...") + if err := w.Run(worker.InterruptCh()); err != nil { + errChan <- err + } + }() + + // Start HTTP API server + wg.Add(1) + go func() { + defer wg.Done() + server := api.NewServer(database, c, logger) + logger.Printf("starting API server on port %d", *apiPort) + if err := server.Start(*apiPort); err != nil { + errChan <- err + } + }() + + // Wait for interrupt signal + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) + + go func() { + sig := <-sigChan + logger.Printf("received signal: %v", sig) + w.Stop() + }() + + // Monitor for errors + go func() { + err := <-errChan + if err != nil { + logger.Printf("error: %v", err) + w.Stop() + } + }() + + wg.Wait() + logger.Println("✓ Server stopped gracefully") +} diff --git a/go.mod b/go.mod index 8bb42cf..cc445b4 100644 --- a/go.mod +++ b/go.mod @@ -1,4 +1,4 @@ -module github.com/rockliang/poimen/workflows +module forgejo.riotpiao.com/rock/poimen-workflows go 1.25.4 diff --git a/internal/api/server.go b/internal/api/server.go new file mode 100644 index 0000000..01fdf75 --- /dev/null +++ b/internal/api/server.go @@ -0,0 +1,101 @@ +package api + +import ( + "fmt" + "log" + "net/http" + "strings" + + "go.temporal.io/sdk/client" + + "github.com/rockliang/poimen/workflows/pkg/db" +) + +// Server handles HTTP routing for workflow APIs +type Server struct { + api *WorkflowAPI + logger *log.Logger +} + +// NewServer creates new HTTP server with database connection +func NewServer(database *db.DB, temporalClient client.Client, logger *log.Logger) *Server { + return &Server{ + api: NewWorkflowAPI(database, temporalClient, logger), + logger: logger, + } +} + +// ServeHTTP dispatches HTTP requests to appropriate handler +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // Enable CORS + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization") + + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusOK) + return + } + + path := r.URL.Path + method := r.Method + + s.logger.Printf("%s %s", method, path) + + // Route requests + switch { + // Workflow endpoints + case path == "/workflows" && method == http.MethodPost: + s.api.CreateWorkflow(w, r) + case path == "/workflows" && method == http.MethodGet: + s.api.ListWorkflows(w, r) + case strings.HasPrefix(path, "/workflows/") && method == http.MethodGet: + id := strings.TrimPrefix(path, "/workflows/") + // Exclude special paths + if !strings.Contains(id, "/") { + s.api.GetWorkflow(w, r, id) + } else if strings.HasSuffix(id, "/executions") { + // GET /workflows/{id}/executions + workflowID := strings.TrimSuffix(id, "/executions") + s.api.ListExecutions(w, r, workflowID) + } + case strings.HasPrefix(path, "/workflows/") && method == http.MethodPut: + id := extractID(path, "/workflows/") + s.api.UpdateWorkflow(w, r, id) + case strings.HasPrefix(path, "/workflows/") && method == http.MethodDelete: + id := extractID(path, "/workflows/") + s.api.DeleteWorkflow(w, r, id) + + // Execute workflow + case strings.HasSuffix(path, "/execute") && method == http.MethodPost: + // POST /workflows/{id}/execute + parts := strings.Split(path, "/") + if len(parts) >= 4 && parts[1] == "workflows" && parts[3] == "execute" { + s.api.ExecuteWorkflow(w, r, parts[2]) + } + + // Execution endpoints + case strings.HasPrefix(path, "/executions/") && method == http.MethodGet: + id := extractID(path, "/executions/") + s.api.GetExecution(w, r, id) + + default: + http.Error(w, "Not found", http.StatusNotFound) + } +} + +// extractID extracts resource ID from path +func extractID(path, prefix string) string { + id := strings.TrimPrefix(path, prefix) + if idx := strings.Index(id, "/"); idx != -1 { + return id[:idx] + } + return id +} + +// Start starts the HTTP server +func (s *Server) Start(port int) error { + addr := fmt.Sprintf(":%d", port) + s.logger.Printf("Starting API server on %s", addr) + return http.ListenAndServe(addr, s) +} diff --git a/internal/api/workflows.go b/internal/api/workflows.go new file mode 100644 index 0000000..5189d65 --- /dev/null +++ b/internal/api/workflows.go @@ -0,0 +1,589 @@ +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 +} diff --git a/internal/routing/activity_knowledge_base.json b/internal/routing/activity_knowledge_base.json index 8d0697c..0d25a09 100644 --- a/internal/routing/activity_knowledge_base.json +++ b/internal/routing/activity_knowledge_base.json @@ -473,10 +473,130 @@ "dependencies": [], "notes": "Must run before LLM Router to provide auth token. Call early in workflow." } + }, + { + "name": "LLMInferenceActivity", + "description": "Call LLM API with custom prompt and get response text", + "category": "llm", + "inputs": { + "model": { + "type": "string", + "description": "Model ID (reasoning, ornith:35b, ornith:13b, qwen2.5:3b)", + "required": true, + "examples": ["reasoning", "ornith:35b"] + }, + "system_prompt": { + "type": "string", + "description": "System instruction for the model", + "required": false, + "default": "" + }, + "user_prompt": { + "type": "string", + "description": "User message to send to the model", + "required": true + }, + "temperature": { + "type": "number", + "description": "Sampling temperature (0.0-1.0, higher=more creative)", + "required": false, + "default": 0.7 + }, + "max_tokens": { + "type": "integer", + "description": "Maximum tokens in response", + "required": false + } + }, + "outputs": { + "response": { + "type": "string", + "description": "LLM response text" + }, + "model": { + "type": "string", + "description": "Model used for inference" + }, + "stop_reason": { + "type": "string", + "description": "Why inference stopped (stop_sequence, length, etc)" + }, + "tokens_used": { + "type": "integer", + "description": "Total tokens consumed" + } + }, + "constraints": { + "defaultTimeout": "120s", + "isFlaky": true, + "recommendedRetries": 2, + "retryBackoff": 2.0, + "dependencies": [], + "notes": "API-dependent. Network flaky. Use for single prompts. See LLMBatchInferenceActivity for multiple." + } + }, + { + "name": "LLMBatchInferenceActivity", + "description": "Call LLM API multiple times sequentially with different prompts", + "category": "llm", + "inputs": { + "model": { + "type": "string", + "description": "Model ID (reasoning, ornith:35b, ornith:13b, qwen2.5:3b)", + "required": true + }, + "system_prompt": { + "type": "string", + "description": "System instruction (same for all prompts)", + "required": false + }, + "prompts": { + "type": "array", + "description": "List of user prompts to process", + "required": true, + "items": { + "type": "string" + } + }, + "temperature": { + "type": "number", + "description": "Sampling temperature (0.0-1.0)", + "required": false, + "default": 0.7 + } + }, + "outputs": { + "responses": { + "type": "array", + "description": "List of LLM responses (parallel to input prompts)", + "items": { + "type": "string" + } + }, + "model": { + "type": "string", + "description": "Model used" + }, + "errors": { + "type": "array", + "description": "Error messages for failed prompts", + "items": { + "type": "string" + } + } + }, + "constraints": { + "defaultTimeout": "600s", + "isFlaky": true, + "recommendedRetries": 1, + "retryBackoff": 2.0, + "dependencies": [], + "notes": "Sequential processing of multiple prompts. Use for batch analysis, summarization, etc." + } } ], "metadata": { - "totalActivities": 10, + "totalActivities": 12, "lastUpdated": "2025-08-31T00:00:00Z", "categories": { "repository": 1, @@ -488,7 +608,8 @@ "approval": 1, "storage": 1, "memory": 1, - "authentication": 1 + "authentication": 1, + "llm": 2 } } } diff --git a/internal/routing/canvas_converter.go b/internal/routing/canvas_converter.go new file mode 100644 index 0000000..1ad68ad --- /dev/null +++ b/internal/routing/canvas_converter.go @@ -0,0 +1,178 @@ +package routing + +import ( + "fmt" + + "github.com/rockliang/poimen/workflows/pkg/db" +) + +// CanvasConverter converts visual canvas to executable WorkflowSpec +type CanvasConverter struct { + validator *CanvasValidator +} + +// NewCanvasConverter creates a converter +func NewCanvasConverter() *CanvasConverter { + return &CanvasConverter{ + validator: NewCanvasValidator(), + } +} + +// CanvasToWorkflowSpec converts canvas to WorkflowSpec +func (cc *CanvasConverter) CanvasToWorkflowSpec(canvas *db.Canvas) (*WorkflowSpec, error) { + // Validate first + if err := cc.validator.ValidateCanvas(canvas); err != nil { + return nil, fmt.Errorf("canvas validation failed: %w", err) + } + + // Get topological order + sortedNodes, err := cc.validator.TopoSort(canvas.Nodes, canvas.Edges) + if err != nil { + return nil, fmt.Errorf("topological sort failed: %w", err) + } + + // Build states from sorted nodes + states := []State{} + nodeToState := make(map[string]int) // node ID to state index + + for i, node := range sortedNodes { + state := cc.nodeToState(node, canvas.Edges) + states = append(states, state) + nodeToState[node.ID] = i + } + + // Wire up transitions + for i, node := range sortedNodes { + outgoing := cc.getOutgoingEdges(node.ID, canvas.Edges) + + if len(outgoing) == 0 { + // Last state - no transitions + continue + } + + if len(outgoing) == 1 { + // Single outgoing edge + targetNode := outgoing[0] + targetIdx := nodeToState[targetNode] + if targetIdx > i { + states[i].Next = states[targetIdx].Name + } + } else { + // Multiple outgoing edges - parallel + states[i].Type = "Parallel" + branches := []interface{}{} + for _, targetNode := range outgoing { + branches = append(branches, map[string]string{ + "state": states[nodeToState[targetNode]].Name, + }) + } + if states[i].Branches == nil { + states[i].Branches = branches + } + } + } + + spec := &WorkflowSpec{ + Name: canvas.Name, + Input: map[string]interface{}{}, + States: states, + } + + return spec, nil +} + +// nodeToState converts a canvas node to a workflow state +func (cc *CanvasConverter) nodeToState(node db.WorkflowNode, edges []db.WorkflowEdge) State { + // Map node type to activity name + activityName := cc.mapActivityType(node.Type) + + state := State{ + Name: node.ID, + Type: TaskActivity, + Activity: activityName, + Retry: &RetryPolicy{MaxAttempts: 3, BackoffSeconds: 2}, + Timeout: "300s", + Parameters: node.Data, + } + + return state +} + +// mapActivityType maps canvas activity type to Poimen activity +func (cc *CanvasConverter) mapActivityType(canvasType string) string { + typeMap := map[string]string{ + "clone-repo": "CloneRepoActivity", + "analyze-code": "AnalyzeCodeActivity", + "security-scan": "SecurityScanActivity", + "generate-report": "GenerateReportActivity", + "deployment-precheck": "DeploymentPreCheckActivity", + "notify-status": "NotifyStatusActivity", + "approve-workflow": "ApproveWorkflowActivity", + "archive-results": "ArchiveResultsActivity", + "retrieve-memory": "RetrieveMemoryActivity", + "assume-role": "AssumeRoleActivity", + "llm-inference": "LLMInferenceActivity", + "llm-batch-inference": "LLMBatchInferenceActivity", + } + + if mapped, ok := typeMap[canvasType]; ok { + return mapped + } + + return canvasType // fallback to type as-is +} + +// getOutgoingEdges returns target node IDs for a given source node +func (cc *CanvasConverter) getOutgoingEdges(nodeID string, edges []db.WorkflowEdge) []string { + targets := []string{} + seen := make(map[string]bool) + + for _, edge := range edges { + if edge.Source == nodeID && !seen[edge.Target] { + targets = append(targets, edge.Target) + seen[edge.Target] = true + } + } + + return targets +} + +// CanvasToExecutionPlan converts canvas to sequential activity list +func (cc *CanvasConverter) CanvasToExecutionPlan(canvas *db.Canvas) ([]ExecutionStep, error) { + // Validate first + if err := cc.validator.ValidateCanvas(canvas); err != nil { + return nil, fmt.Errorf("canvas validation failed: %w", err) + } + + // Get topological order + sortedNodes, err := cc.validator.TopoSort(canvas.Nodes, canvas.Edges) + if err != nil { + return nil, fmt.Errorf("topological sort failed: %w", err) + } + + steps := []ExecutionStep{} + for i, node := range sortedNodes { + step := ExecutionStep{ + Index: i, + NodeID: node.ID, + ActivityName: cc.mapActivityType(node.Type), + Label: node.Label, + Parameters: node.Data, + Timeout: "300s", + } + steps = append(steps, step) + } + + return steps, nil +} + +// ExecutionStep represents one activity in execution plan +type ExecutionStep struct { + Index int `json:"index"` + NodeID string `json:"node_id"` + ActivityName string `json:"activity_name"` + Label string `json:"label"` + Parameters map[string]interface{} `json:"parameters"` + Timeout string `json:"timeout"` + DependsOn []int `json:"depends_on,omitempty"` // Indices of predecessor steps +} diff --git a/internal/routing/canvas_validator.go b/internal/routing/canvas_validator.go new file mode 100644 index 0000000..2a21d96 --- /dev/null +++ b/internal/routing/canvas_validator.go @@ -0,0 +1,316 @@ +package routing + +import ( + "fmt" + "strings" + + "github.com/rockliang/poimen/workflows/pkg/db" +) + +// CanvasValidator validates React Flow canvas (nodes + edges) +type CanvasValidator struct { + activityRegistry map[string]bool +} + +// NewCanvasValidator creates validator with activity registry +func NewCanvasValidator() *CanvasValidator { + return &CanvasValidator{ + activityRegistry: map[string]bool{ + "clone-repo": true, + "analyze-code": true, + "security-scan": true, + "generate-report": true, + "deployment-precheck": true, + "notify-status": true, + "approve-workflow": true, + "archive-results": true, + "retrieve-memory": true, + "assume-role": true, + "llm-inference": true, + "llm-batch-inference": true, + }, + } +} + +// ValidateCanvas checks canvas structure, connectivity, and DAG +func (cv *CanvasValidator) ValidateCanvas(canvas *db.Canvas) error { + if canvas == nil { + return fmt.Errorf("canvas is nil") + } + + if len(canvas.Nodes) == 0 { + return fmt.Errorf("canvas has no nodes") + } + + // Step 1: Validate nodes + if err := cv.validateNodes(canvas.Nodes); err != nil { + return fmt.Errorf("node validation failed: %w", err) + } + + // Step 2: Validate edges + if err := cv.validateEdges(canvas.Nodes, canvas.Edges); err != nil { + return fmt.Errorf("edge validation failed: %w", err) + } + + // Step 3: Check for cycles (must be DAG) + if err := cv.detectCycles(canvas.Nodes, canvas.Edges); err != nil { + return fmt.Errorf("cycle detected: %w", err) + } + + // Step 4: Check connectivity (all nodes reachable from start) + if err := cv.validateConnectivity(canvas.Nodes, canvas.Edges); err != nil { + return fmt.Errorf("connectivity check failed: %w", err) + } + + return nil +} + +// validateNodes checks each node has required fields and valid type +func (cv *CanvasValidator) validateNodes(nodes []db.WorkflowNode) error { + if len(nodes) == 0 { + return fmt.Errorf("no nodes in canvas") + } + + nodeIds := make(map[string]bool) + + for i, node := range nodes { + // Check required fields + if node.ID == "" { + return fmt.Errorf("node[%d] has empty ID", i) + } + + if nodeIds[node.ID] { + return fmt.Errorf("node[%d] has duplicate ID: %s", i, node.ID) + } + nodeIds[node.ID] = true + + if node.Label == "" { + return fmt.Errorf("node[%d] (%s) has empty label", i, node.ID) + } + + if node.Position == nil { + return fmt.Errorf("node[%d] (%s) has no position", i, node.ID) + } + + // Check activity type (if present) + if node.Type != "" && !cv.activityRegistry[strings.ToLower(node.Type)] { + return fmt.Errorf("node[%d] (%s) has unknown activity type: %s", i, node.ID, node.Type) + } + + // Check data structure + if node.Data == nil { + return fmt.Errorf("node[%d] (%s) has no data", i, node.ID) + } + } + + return nil +} + +// validateEdges checks edges reference valid nodes +func (cv *CanvasValidator) validateEdges(nodes []db.WorkflowNode, edges []db.WorkflowEdge) error { + nodeIds := make(map[string]bool) + for _, node := range nodes { + nodeIds[node.ID] = true + } + + for i, edge := range edges { + // Check required fields + if edge.Source == "" { + return fmt.Errorf("edge[%d] has empty source", i) + } + + if edge.Target == "" { + return fmt.Errorf("edge[%d] has empty target", i) + } + + // Check source node exists + if !nodeIds[edge.Source] { + return fmt.Errorf("edge[%d] references unknown source node: %s", i, edge.Source) + } + + // Check target node exists + if !nodeIds[edge.Target] { + return fmt.Errorf("edge[%d] references unknown target node: %s", i, edge.Target) + } + + // Check self-loops (discouraged but allow for now) + if edge.Source == edge.Target { + // Could warn here but not fail + } + } + + return nil +} + +// detectCycles checks for cycles in the DAG (must be acyclic) +func (cv *CanvasValidator) detectCycles(nodes []db.WorkflowNode, edges []db.WorkflowEdge) error { + // Build adjacency list + graph := make(map[string][]string) + inDegree := make(map[string]int) + + for _, node := range nodes { + graph[node.ID] = []string{} + inDegree[node.ID] = 0 + } + + for _, edge := range edges { + graph[edge.Source] = append(graph[edge.Source], edge.Target) + inDegree[edge.Target]++ + } + + // Kahn's algorithm: topological sort + queue := []string{} + for _, node := range nodes { + if inDegree[node.ID] == 0 { + queue = append(queue, node.ID) + } + } + + processed := 0 + for len(queue) > 0 { + // Dequeue + current := queue[0] + queue = queue[1:] + processed++ + + // Visit neighbors + for _, neighbor := range graph[current] { + inDegree[neighbor]-- + if inDegree[neighbor] == 0 { + queue = append(queue, neighbor) + } + } + } + + // If we didn't process all nodes, there's a cycle + if processed != len(nodes) { + return fmt.Errorf("graph has cycle (processed %d/%d nodes)", processed, len(nodes)) + } + + return nil +} + +// validateConnectivity checks all nodes are reachable from start nodes +func (cv *CanvasValidator) validateConnectivity(nodes []db.WorkflowNode, edges []db.WorkflowEdge) error { + if len(nodes) == 0 { + return nil + } + + // Build adjacency list + graph := make(map[string][]string) + inDegree := make(map[string]int) + + for _, node := range nodes { + graph[node.ID] = []string{} + inDegree[node.ID] = 0 + } + + for _, edge := range edges { + graph[edge.Source] = append(graph[edge.Source], edge.Target) + inDegree[edge.Target]++ + } + + // Find start nodes (in-degree 0) + startNodes := []string{} + for _, node := range nodes { + if inDegree[node.ID] == 0 { + startNodes = append(startNodes, node.ID) + } + } + + if len(startNodes) == 0 { + return fmt.Errorf("no start nodes found (all nodes have incoming edges)") + } + + // BFS from all start nodes + visited := make(map[string]bool) + queue := startNodes + + for len(queue) > 0 { + // Dequeue + current := queue[0] + queue = queue[1:] + + if visited[current] { + continue + } + visited[current] = true + + // Visit neighbors + for _, neighbor := range graph[current] { + if !visited[neighbor] { + queue = append(queue, neighbor) + } + } + } + + // Check all nodes were visited + if len(visited) != len(nodes) { + unreached := []string{} + for _, node := range nodes { + if !visited[node.ID] { + unreached = append(unreached, node.ID) + } + } + return fmt.Errorf("unreachable nodes: %v", unreached) + } + + return nil +} + +// TopoSort returns nodes in topological order (execution order) +func (cv *CanvasValidator) TopoSort(nodes []db.WorkflowNode, edges []db.WorkflowEdge) ([]db.WorkflowNode, error) { + if len(nodes) == 0 { + return []db.WorkflowNode{}, nil + } + + // Build adjacency list and in-degree map + graph := make(map[string][]string) + inDegree := make(map[string]int) + nodeMap := make(map[string]db.WorkflowNode) + + for _, node := range nodes { + graph[node.ID] = []string{} + inDegree[node.ID] = 0 + nodeMap[node.ID] = node + } + + for _, edge := range edges { + graph[edge.Source] = append(graph[edge.Source], edge.Target) + inDegree[edge.Target]++ + } + + // Kahn's algorithm + queue := []string{} + for _, node := range nodes { + if inDegree[node.ID] == 0 { + queue = append(queue, node.ID) + } + } + + result := []db.WorkflowNode{} + processed := make(map[string]bool) + + for len(queue) > 0 { + // Dequeue + current := queue[0] + queue = queue[1:] + + result = append(result, nodeMap[current]) + processed[current] = true + + // Visit neighbors + for _, neighbor := range graph[current] { + inDegree[neighbor]-- + if inDegree[neighbor] == 0 { + queue = append(queue, neighbor) + } + } + } + + if len(result) != len(nodes) { + return nil, fmt.Errorf("topological sort failed: graph has cycle") + } + + return result, nil +} diff --git a/k8s/git-commit.yaml b/k8s/git-commit.yaml index b76c979..b44e511 100644 --- a/k8s/git-commit.yaml +++ b/k8s/git-commit.yaml @@ -9,6 +9,6 @@ metadata: app.kubernetes.io/name: poimen app.kubernetes.io/component: orchestrator data: - GIT_COMMIT: "cdb6efe2" # Updated automatically by CI/CD + GIT_COMMIT: "e07b9504" # Updated automatically by CI/CD GIT_BRANCH: "main" - DEPLOYMENT_DATE: "2026-09-04" + DEPLOYMENT_DATE: "2026-09-05" diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml index 12785cb..5464649 100644 --- a/k8s/worker-deployment.yaml +++ b/k8s/worker-deployment.yaml @@ -13,8 +13,8 @@ spec: labels: app: poimen-worker annotations: - git-commit: "cdb6efe2" # ✅ Updated on each push, triggers rolling restart - deployment-date: "2026-09-04" + git-commit: "e07b9504" # ✅ Updated on each push, triggers rolling restart + deployment-date: "2026-09-05" spec: containers: - name: worker diff --git a/pkg/db/db.go b/pkg/db/db.go new file mode 100644 index 0000000..60d5b0c --- /dev/null +++ b/pkg/db/db.go @@ -0,0 +1,456 @@ +package db + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "os" + "time" + + _ "github.com/lib/pq" +) + +// DB wraps the database connection +type DB struct { + conn *sql.DB +} + +// New creates a new database connection to memory-db (K8s CNPG) +// Expected DSN format: postgresql://app:password@memory-db-rw.poimen.svc.cluster.local:5432/memory?sslmode=disable +func New(dsn string) (*DB, error) { + if dsn == "" { + // Fallback: try to construct from K8s env vars + host := os.Getenv("DATABASE_HOST") + port := os.Getenv("DATABASE_PORT") + name := os.Getenv("DATABASE_NAME") + user := os.Getenv("DATABASE_USER") + password := os.Getenv("DATABASE_PASSWORD") + + if host != "" && port != "" && name != "" && user != "" && password != "" { + dsn = fmt.Sprintf("postgresql://%s:%s@%s:%s/%s?sslmode=disable", + user, password, host, port, name) + } else { + return nil, fmt.Errorf("DATABASE_URL or K8s env vars (DATABASE_HOST, DATABASE_PORT, DATABASE_NAME, DATABASE_USER, DATABASE_PASSWORD) required") + } + } + + conn, err := sql.Open("postgres", dsn) + if err != nil { + return nil, fmt.Errorf("failed to open database: %w", err) + } + + // Test connection + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + if err := conn.PingContext(ctx); err != nil { + return nil, fmt.Errorf("failed to ping database: %w", err) + } + + // Set connection pool settings + conn.SetMaxOpenConns(25) + conn.SetMaxIdleConns(5) + conn.SetConnMaxLifetime(5 * time.Minute) + + return &DB{conn: conn}, nil +} + +// Close closes the database connection +func (db *DB) Close() error { + return db.conn.Close() +} + +// SaveWorkflow saves or updates a workflow with canvas +func (db *DB) SaveWorkflow(ctx context.Context, wf *Workflow) error { + query := ` + INSERT INTO workflows (id, customer_id, name, description, status, version, nodes, edges, created_by, created_at, updated_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + ON CONFLICT (id) DO UPDATE SET + name = $3, + description = $4, + status = $5, + version = $6, + nodes = $7, + edges = $8, + updated_at = $11 + ` + + _, err := db.conn.ExecContext(ctx, query, + wf.ID, + wf.CustomerID, + wf.Name, + wf.Description, + wf.Status, + wf.Version, + wf.Nodes, + wf.Edges, + wf.CreatedBy, + wf.CreatedAt, + wf.UpdatedAt, + ) + + return err +} + +// SaveCanvasUpdate saves canvas (nodes + edges) for a workflow +func (db *DB) SaveCanvasUpdate(ctx context.Context, workflowID, customerID string, canvas *Canvas) error { + nodesJSON, err := json.Marshal(canvas.Nodes) + if err != nil { + return fmt.Errorf("failed to marshal nodes: %w", err) + } + + edgesJSON, err := json.Marshal(canvas.Edges) + if err != nil { + return fmt.Errorf("failed to marshal edges: %w", err) + } + + query := ` + UPDATE workflows + SET nodes = $1, edges = $2, updated_at = now() + WHERE id = $3 AND customer_id = $4 + ` + + result, err := db.conn.ExecContext(ctx, query, nodesJSON, edgesJSON, workflowID, customerID) + if err != nil { + return fmt.Errorf("failed to update canvas: %w", err) + } + + rows, err := result.RowsAffected() + if err != nil { + return err + } + + if rows == 0 { + return fmt.Errorf("workflow not found: %s", workflowID) + } + + return nil +} + +// FetchWorkflow retrieves a workflow by ID +func (db *DB) FetchWorkflow(ctx context.Context, workflowID, customerID string) (*Workflow, error) { + query := ` + SELECT id, customer_id, name, description, status, version, nodes, edges, created_by, created_at, updated_at, last_executed_at + FROM workflows + WHERE id = $1 AND customer_id = $2 + ` + + wf := &Workflow{} + err := db.conn.QueryRowContext(ctx, query, workflowID, customerID).Scan( + &wf.ID, + &wf.CustomerID, + &wf.Name, + &wf.Description, + &wf.Status, + &wf.Version, + &wf.Nodes, + &wf.Edges, + &wf.CreatedBy, + &wf.CreatedAt, + &wf.UpdatedAt, + &wf.LastExecutedAt, + ) + + if err != nil { + if err == sql.ErrNoRows { + return nil, fmt.Errorf("workflow not found: %s", workflowID) + } + return nil, fmt.Errorf("failed to fetch workflow: %w", err) + } + + return wf, nil +} + +// FetchCanvas retrieves canvas (nodes + edges) for a workflow +func (db *DB) FetchCanvas(ctx context.Context, workflowID, customerID string) (*Canvas, error) { + query := ` + SELECT nodes, edges + FROM workflows + WHERE id = $1 AND customer_id = $2 + ` + + var nodesJSON, edgesJSON []byte + err := db.conn.QueryRowContext(ctx, query, workflowID, customerID).Scan(&nodesJSON, &edgesJSON) + if err != nil { + if err == sql.ErrNoRows { + return nil, fmt.Errorf("workflow not found: %s", workflowID) + } + return nil, fmt.Errorf("failed to fetch canvas: %w", err) + } + + var nodes []WorkflowNode + var edges []WorkflowEdge + + if err := json.Unmarshal(nodesJSON, &nodes); err != nil { + return nil, fmt.Errorf("failed to unmarshal nodes: %w", err) + } + + if err := json.Unmarshal(edgesJSON, &edges); err != nil { + return nil, fmt.Errorf("failed to unmarshal edges: %w", err) + } + + return &Canvas{Nodes: nodes, Edges: edges}, nil +} + +// ListWorkflows retrieves all workflows for a customer +func (db *DB) ListWorkflows(ctx context.Context, customerID string, limit, offset int) ([]Workflow, error) { + query := ` + SELECT id, customer_id, name, description, status, version, nodes, edges, created_by, created_at, updated_at, last_executed_at + FROM workflows + WHERE customer_id = $1 + ORDER BY created_at DESC + LIMIT $2 OFFSET $3 + ` + + rows, err := db.conn.QueryContext(ctx, query, customerID, limit, offset) + if err != nil { + return nil, fmt.Errorf("failed to list workflows: %w", err) + } + defer rows.Close() + + var workflows []Workflow + for rows.Next() { + wf := Workflow{} + err := rows.Scan( + &wf.ID, + &wf.CustomerID, + &wf.Name, + &wf.Description, + &wf.Status, + &wf.Version, + &wf.Nodes, + &wf.Edges, + &wf.CreatedBy, + &wf.CreatedAt, + &wf.UpdatedAt, + &wf.LastExecutedAt, + ) + if err != nil { + return nil, fmt.Errorf("failed to scan workflow: %w", err) + } + workflows = append(workflows, wf) + } + + return workflows, rows.Err() +} + +// DeleteWorkflow deletes a workflow +func (db *DB) DeleteWorkflow(ctx context.Context, workflowID, customerID string) error { + query := ` + DELETE FROM workflows + WHERE id = $1 AND customer_id = $2 + ` + + result, err := db.conn.ExecContext(ctx, query, workflowID, customerID) + if err != nil { + return fmt.Errorf("failed to delete workflow: %w", err) + } + + rows, err := result.RowsAffected() + if err != nil { + return err + } + + if rows == 0 { + return fmt.Errorf("workflow not found: %s", workflowID) + } + + return nil +} + +// SaveExecution saves a workflow execution record +func (db *DB) SaveExecution(ctx context.Context, exec *WorkflowExecution) error { + query := ` + INSERT INTO workflow_executions (id, workflow_id, customer_id, temporal_id, status, inputs, outputs, started_at, completed_at, duration_ms, error_message, error_count) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + ON CONFLICT (id) DO UPDATE SET + status = $5, + outputs = $7, + completed_at = $9, + duration_ms = $10, + error_message = $11, + error_count = $12 + ` + + _, err := db.conn.ExecContext(ctx, query, + exec.ID, + exec.WorkflowID, + exec.CustomerID, + exec.TemporalID, + exec.Status, + exec.Inputs, + exec.Outputs, + exec.StartedAt, + exec.CompletedAt, + exec.DurationMs, + exec.ErrorMessage, + exec.ErrorCount, + ) + + return err +} + +// FetchExecution retrieves a workflow execution +func (db *DB) FetchExecution(ctx context.Context, executionID string) (*WorkflowExecution, error) { + query := ` + SELECT id, workflow_id, customer_id, temporal_id, status, inputs, outputs, started_at, completed_at, duration_ms, error_message, error_count + FROM workflow_executions + WHERE id = $1 + ` + + exec := &WorkflowExecution{} + err := db.conn.QueryRowContext(ctx, query, executionID).Scan( + &exec.ID, + &exec.WorkflowID, + &exec.CustomerID, + &exec.TemporalID, + &exec.Status, + &exec.Inputs, + &exec.Outputs, + &exec.StartedAt, + &exec.CompletedAt, + &exec.DurationMs, + &exec.ErrorMessage, + &exec.ErrorCount, + ) + + if err != nil { + if err == sql.ErrNoRows { + return nil, fmt.Errorf("execution not found: %s", executionID) + } + return nil, fmt.Errorf("failed to fetch execution: %w", err) + } + + return exec, nil +} + +// SaveExecutionLog saves an activity log entry +func (db *DB) SaveExecutionLog(ctx context.Context, log *ExecutionLog) error { + query := ` + INSERT INTO execution_logs (execution_id, node_id, activity_name, level, message, metadata, logged_at) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ` + + _, err := db.conn.ExecContext(ctx, query, + log.ExecutionID, + log.NodeID, + log.ActivityName, + log.Level, + log.Message, + log.Metadata, + log.LoggedAt, + ) + + return err +} + +// FetchExecutionLogs retrieves all logs for an execution +func (db *DB) FetchExecutionLogs(ctx context.Context, executionID string) ([]ExecutionLog, error) { + query := ` + SELECT id, execution_id, node_id, activity_name, level, message, metadata, logged_at + FROM execution_logs + WHERE execution_id = $1 + ORDER BY logged_at ASC + ` + + rows, err := db.conn.QueryContext(ctx, query, executionID) + if err != nil { + return nil, fmt.Errorf("failed to fetch execution logs: %w", err) + } + defer rows.Close() + + var logs []ExecutionLog + for rows.Next() { + log := ExecutionLog{} + err := rows.Scan( + &log.ID, + &log.ExecutionID, + &log.NodeID, + &log.ActivityName, + &log.Level, + &log.Message, + &log.Metadata, + &log.LoggedAt, + ) + if err != nil { + return nil, fmt.Errorf("failed to scan log: %w", err) + } + logs = append(logs, log) + } + + return logs, rows.Err() +} + +// SaveActivityTrace saves per-activity execution trace +func (db *DB) SaveActivityTrace(ctx context.Context, trace *ActivityTrace) error { + query := ` + INSERT INTO activity_traces (execution_id, node_id, activity_name, parameters, result, started_at, completed_at, duration_ms, attempt, retry_reason, status, error_message) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + ON CONFLICT (id) DO UPDATE SET + status = $11, + result = $5, + completed_at = $7, + duration_ms = $8, + error_message = $12 + ` + + _, err := db.conn.ExecContext(ctx, query, + trace.ExecutionID, + trace.NodeID, + trace.ActivityName, + trace.Parameters, + trace.Result, + trace.StartedAt, + trace.CompletedAt, + trace.DurationMs, + trace.Attempt, + trace.RetryReason, + trace.Status, + trace.ErrorMessage, + ) + + return err +} + +// FetchActivityTraces retrieves all activity traces for an execution +func (db *DB) FetchActivityTraces(ctx context.Context, executionID string) ([]ActivityTrace, error) { + query := ` + SELECT id, execution_id, node_id, activity_name, parameters, result, started_at, completed_at, duration_ms, attempt, retry_reason, status, error_message + FROM activity_traces + WHERE execution_id = $1 + ORDER BY started_at ASC + ` + + rows, err := db.conn.QueryContext(ctx, query, executionID) + if err != nil { + return nil, fmt.Errorf("failed to fetch activity traces: %w", err) + } + defer rows.Close() + + var traces []ActivityTrace + for rows.Next() { + trace := ActivityTrace{} + err := rows.Scan( + &trace.ID, + &trace.ExecutionID, + &trace.NodeID, + &trace.ActivityName, + &trace.Parameters, + &trace.Result, + &trace.StartedAt, + &trace.CompletedAt, + &trace.DurationMs, + &trace.Attempt, + &trace.RetryReason, + &trace.Status, + &trace.ErrorMessage, + ) + if err != nil { + return nil, fmt.Errorf("failed to scan trace: %w", err) + } + traces = append(traces, trace) + } + + return traces, rows.Err() +} diff --git a/pkg/db/models.go b/pkg/db/models.go new file mode 100644 index 0000000..688ee60 --- /dev/null +++ b/pkg/db/models.go @@ -0,0 +1,113 @@ +package db + +import ( + "time" +) + +// WorkflowNode represents a React Flow node in the canvas +type WorkflowNode struct { + ID string `json:"id"` + Label string `json:"label"` + Type string `json:"type"` // "activity" + Position map[string]interface{} `json:"position"` + Data map[string]interface{} `json:"data"` +} + +// WorkflowEdge represents a React Flow edge in the canvas +type WorkflowEdge struct { + ID string `json:"id"` + Source string `json:"source"` + Target string `json:"target"` + Data map[string]interface{} `json:"data"` +} + +// Canvas represents the full React Flow canvas (nodes + edges) +type Canvas struct { + Nodes []WorkflowNode `json:"nodes"` + Edges []WorkflowEdge `json:"edges"` +} + +// Workflow represents a workflow definition in the database +type Workflow struct { + ID string `db:"id"` + CustomerID string `db:"customer_id"` + Name string `db:"name"` + Description string `db:"description"` + Status string `db:"status"` // "draft", "active", "archived" + Version int `db:"version"` + Nodes []byte `db:"nodes"` // JSONB stored as []byte + Edges []byte `db:"edges"` // JSONB stored as []byte + CreatedBy string `db:"created_by"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` + LastExecutedAt *time.Time `db:"last_executed_at"` +} + +// WorkflowExecution represents a workflow execution run +type WorkflowExecution struct { + ID string `db:"id"` + WorkflowID string `db:"workflow_id"` + CustomerID string `db:"customer_id"` + TemporalID string `db:"temporal_id"` // Temporal execution ID + Status string `db:"status"` // "pending", "running", "success", "failed", "cancelled" + Inputs []byte `db:"inputs"` // JSONB + Outputs []byte `db:"outputs"` // JSONB + StartedAt time.Time `db:"started_at"` + CompletedAt *time.Time `db:"completed_at"` + DurationMs *int `db:"duration_ms"` + ErrorMessage string `db:"error_message"` + ErrorCount int `db:"error_count"` +} + +// ExecutionLog represents a detailed activity log entry +type ExecutionLog struct { + ID int64 `db:"id"` + ExecutionID string `db:"execution_id"` + NodeID string `db:"node_id"` // From canvas node ID + ActivityName string `db:"activity_name"` // "CloneRepo", "AnalyzeCode", etc + Level string `db:"level"` // "info", "warn", "error", "debug" + Message string `db:"message"` + Metadata []byte `db:"metadata"` // JSONB + LoggedAt time.Time `db:"logged_at"` +} + +// ActivityTrace represents per-activity execution metrics +type ActivityTrace struct { + ID int64 `db:"id"` + ExecutionID string `db:"execution_id"` + NodeID string `db:"node_id"` + ActivityName string `db:"activity_name"` + Parameters []byte `db:"parameters"` // JSONB + Result []byte `db:"result"` // JSONB + StartedAt time.Time `db:"started_at"` + CompletedAt *time.Time `db:"completed_at"` + DurationMs *int `db:"duration_ms"` + Attempt int `db:"attempt"` + RetryReason string `db:"retry_reason"` + Status string `db:"status"` // "running", "success", "failed", "skipped" + ErrorMessage string `db:"error_message"` +} + +// WorkflowStats represents aggregated workflow metrics +type WorkflowStats struct { + WorkflowID string `db:"workflow_id"` + CustomerID string `db:"customer_id"` + TotalRuns int `db:"total_runs"` + SuccessfulRuns int `db:"successful_runs"` + FailedRuns int `db:"failed_runs"` + AvgDurationMs float64 `db:"avg_duration_ms"` + MinDurationMs *int `db:"min_duration_ms"` + MaxDurationMs *int `db:"max_duration_ms"` + Last30dRuns int `db:"last_30d_runs"` + Last30dSuccessRate float64 `db:"last_30d_success_rate"` + UpdatedAt time.Time `db:"updated_at"` +} + +// WorkflowMemoryLink represents a connection between execution and memory nodes +type WorkflowMemoryLink struct { + ExecutionID string `db:"execution_id"` + MemoryNodeSha string `db:"memory_node_sha"` + Relationship string `db:"relationship"` // "generated", "used", "learned", "failed_on" + CreatedAt time.Time `db:"created_at"` + Notes string `db:"notes"` +}