package handlers import ( "database/sql" "encoding/json" "net/http" "time" "github.com/go-chi/chi/v5" "boc/automation" ) type AutomationHandler struct { DB *sql.DB Engine *automation.Engine } func NewAutomationHandler(db *sql.DB, engine *automation.Engine) *AutomationHandler { return &AutomationHandler{DB: db, Engine: engine} } type WorkflowRequest struct { Name string `json:"name"` Description string `json:"description"` TriggerType string `json:"trigger_type"` TriggerConfig map[string]interface{} `json:"trigger_config"` Actions []map[string]interface{} `json:"actions"` } type ScheduledJobRequest struct { Name string `json:"name"` Description string `json:"description"` CronExpr string `json:"cron_expr"` Timezone string `json:"timezone"` JobType string `json:"job_type"` JobConfig map[string]interface{} `json:"job_config"` } func (h *AutomationHandler) ListWorkflows(w http.ResponseWriter, r *http.Request) { rows, err := h.DB.Query(` SELECT id, name, description, trigger_type, trigger_config, actions, status, last_run_at, next_run_at, run_count, fail_count, created_at FROM boc_workflows ORDER BY created_at DESC `) if err != nil { writeError(w, http.StatusInternalServerError, "database error") return } defer rows.Close() workflows := []map[string]interface{}{} for rows.Next() { var id, name, description, triggerType, status string var triggerConfig, actions []byte var lastRunAt, nextRunAt *time.Time var runCount, failCount int var createdAt time.Time if err := rows.Scan(&id, &name, &description, &triggerType, &triggerConfig, &actions, &status, &lastRunAt, &nextRunAt, &runCount, &failCount, &createdAt); err != nil { continue } var tc, ac map[string]interface{} json.Unmarshal(triggerConfig, &tc) json.Unmarshal(actions, &ac) workflows = append(workflows, map[string]interface{}{ "id": id, "name": name, "description": description, "trigger_type": triggerType, "trigger_config": tc, "actions": ac, "status": status, "last_run_at": lastRunAt, "next_run_at": nextRunAt, "run_count": runCount, "fail_count": failCount, "created_at": createdAt, }) } writeJSON(w, http.StatusOK, map[string]interface{}{ "workflows": workflows, "total": len(workflows), }) } func (h *AutomationHandler) CreateWorkflow(w http.ResponseWriter, r *http.Request) { var req WorkflowRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "invalid request") return } triggerConfig, _ := json.Marshal(req.TriggerConfig) actions, _ := json.Marshal(req.Actions) var id string err := h.DB.QueryRow(` INSERT INTO boc_workflows (name, description, trigger_type, trigger_config, actions, status) VALUES ($1, $2, $3, $4, $5, 'active') RETURNING id `, req.Name, req.Description, req.TriggerType, triggerConfig, actions).Scan(&id) if err != nil { writeError(w, http.StatusInternalServerError, "failed to create workflow") return } writeJSON(w, http.StatusCreated, map[string]interface{}{ "id": id, "message": "Workflow created", }) } func (h *AutomationHandler) TriggerWorkflow(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "id") var input map[string]interface{} if err := json.NewDecoder(r.Body).Decode(&input); err != nil { input = map[string]interface{}{} } if err := h.Engine.TriggerWorkflow(r.Context(), id, input); err != nil { writeError(w, http.StatusInternalServerError, "failed to trigger workflow") return } writeJSON(w, http.StatusOK, map[string]interface{}{ "message": "Workflow triggered", }) } func (h *AutomationHandler) ListScheduledJobs(w http.ResponseWriter, r *http.Request) { rows, err := h.DB.Query(` SELECT id, name, description, cron_expr, timezone, job_type, job_config, status, last_run_at, next_run_at, run_count, fail_count, created_at FROM boc_scheduled_jobs ORDER BY created_at DESC `) if err != nil { writeError(w, http.StatusInternalServerError, "database error") return } defer rows.Close() jobs := []map[string]interface{}{} for rows.Next() { var id, name, description, cronExpr, timezone, jobType, status string var jobConfig []byte var lastRunAt, nextRunAt *time.Time var runCount, failCount int var createdAt time.Time if err := rows.Scan(&id, &name, &description, &cronExpr, &timezone, &jobType, &jobConfig, &status, &lastRunAt, &nextRunAt, &runCount, &failCount, &createdAt); err != nil { continue } var jc map[string]interface{} json.Unmarshal(jobConfig, &jc) jobs = append(jobs, map[string]interface{}{ "id": id, "name": name, "description": description, "cron_expr": cronExpr, "timezone": timezone, "job_type": jobType, "job_config": jc, "status": status, "last_run_at": lastRunAt, "next_run_at": nextRunAt, "run_count": runCount, "fail_count": failCount, "created_at": createdAt, }) } writeJSON(w, http.StatusOK, map[string]interface{}{ "jobs": jobs, "total": len(jobs), }) } func (h *AutomationHandler) CreateScheduledJob(w http.ResponseWriter, r *http.Request) { var req ScheduledJobRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "invalid request") return } jobConfig, _ := json.Marshal(req.JobConfig) var id string err := h.DB.QueryRow(` INSERT INTO boc_scheduled_jobs (name, description, cron_expr, timezone, job_type, job_config, status) VALUES ($1, $2, $3, $4, $5, $6, 'active') RETURNING id `, req.Name, req.Description, req.CronExpr, req.Timezone, req.JobType, jobConfig).Scan(&id) if err != nil { writeError(w, http.StatusInternalServerError, "failed to create scheduled job") return } writeJSON(w, http.StatusCreated, map[string]interface{}{ "id": id, "message": "Scheduled job created", }) } func (h *AutomationHandler) ListRuns(w http.ResponseWriter, r *http.Request) { workflowID := r.URL.Query().Get("workflow_id") jobID := r.URL.Query().Get("job_id") var rows *sql.Rows var err error if workflowID != "" { rows, err = h.DB.Query(` SELECT id, workflow_id, status, input, output, error, started_at, completed_at FROM boc_workflow_runs WHERE workflow_id = $1 ORDER BY started_at DESC LIMIT 50 `, workflowID) } else if jobID != "" { rows, err = h.DB.Query(` SELECT id, job_id, status, output, error, started_at, completed_at FROM boc_scheduled_job_runs WHERE job_id = $1 ORDER BY started_at DESC LIMIT 50 `, jobID) } else { rows, err = h.DB.Query(` SELECT id, workflow_id, status, input, output, error, started_at, completed_at FROM boc_workflow_runs ORDER BY started_at DESC LIMIT 50 `) } if err != nil { writeError(w, http.StatusInternalServerError, "database error") return } defer rows.Close() runs := []map[string]interface{}{} for rows.Next() { var id, status string var input, output, errorMsg []byte var startedAt time.Time var completedAt *time.Time if workflowID != "" || (!rows.Next() && workflowID == "" && jobID == "") { // Workflow run var workflowID sql.NullString if err := rows.Scan(&id, &workflowID, &status, &input, &output, &errorMsg, &startedAt, &completedAt); err != nil { continue } var inp, out map[string]interface{} json.Unmarshal(input, &inp) json.Unmarshal(output, &out) runs = append(runs, map[string]interface{}{ "id": id, "workflow_id": workflowID.String, "status": status, "input": inp, "output": out, "error": string(errorMsg), "started_at": startedAt, "completed_at": completedAt, }) } } writeJSON(w, http.StatusOK, map[string]interface{}{ "runs": runs, "total": len(runs), }) }