Bernt
9e519f5c8f
- Add database migration: contacts, conversations, messages, leads, workflows, follow-ups, audit log - Contact handler with deduplication and identity resolution - Conversation handler with unified inbox - Lead handler with scoring engine and qualification - Workflow handler with trigger-action engine - Full API routes for all comm layer endpoints - 12 operational agents integrated in all BOC modules
454 lines
15 KiB
Go
454 lines
15 KiB
Go
package comm
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
"github.com/google/uuid"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
// WorkflowHandler hanterar workflows
|
|
type WorkflowHandler struct {
|
|
db *sql.DB
|
|
}
|
|
|
|
// NewWorkflowHandler skapar en ny handler
|
|
func NewWorkflowHandler(db *sql.DB) *WorkflowHandler {
|
|
return &WorkflowHandler{db: db}
|
|
}
|
|
|
|
// Workflow representerar en workflow
|
|
type Workflow struct {
|
|
ID uuid.UUID `json:"id"`
|
|
TenantID uuid.UUID `json:"tenant_id"`
|
|
Name string `json:"name"`
|
|
Description string `json:"description,omitempty"`
|
|
Trigger string `json:"trigger"`
|
|
TriggerConfig json.RawMessage `json:"trigger_config,omitempty"`
|
|
Conditions json.RawMessage `json:"conditions,omitempty"`
|
|
Actions json.RawMessage `json:"actions,omitempty"`
|
|
Status string `json:"status"`
|
|
ExecutionCount int `json:"execution_count"`
|
|
LastExecutedAt *time.Time `json:"last_executed_at,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
// WorkflowExecution representerar en workflow-körning
|
|
type WorkflowExecution struct {
|
|
ID uuid.UUID `json:"id"`
|
|
WorkflowID uuid.UUID `json:"workflow_id"`
|
|
TriggerEvent string `json:"trigger_event"`
|
|
TriggerData json.RawMessage `json:"trigger_data,omitempty"`
|
|
Status string `json:"status"`
|
|
CurrentStep int `json:"current_step"`
|
|
Steps json.RawMessage `json:"steps,omitempty"`
|
|
Result json.RawMessage `json:"result,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
StartedAt time.Time `json:"started_at"`
|
|
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
|
}
|
|
|
|
// CreateWorkflowRequest
|
|
type CreateWorkflowRequest struct {
|
|
Name string `json:"name"`
|
|
Description string `json:"description,omitempty"`
|
|
Trigger string `json:"trigger"`
|
|
TriggerConfig map[string]interface{} `json:"trigger_config,omitempty"`
|
|
Conditions []WorkflowCondition `json:"conditions,omitempty"`
|
|
Actions []WorkflowAction `json:"actions"`
|
|
}
|
|
|
|
// WorkflowCondition representerar ett villkor
|
|
type WorkflowCondition struct {
|
|
Field string `json:"field"`
|
|
Operator string `json:"operator"` // 'eq', 'ne', 'gt', 'lt', 'contains', 'exists'
|
|
Value interface{} `json:"value"`
|
|
}
|
|
|
|
// WorkflowAction representerar en åtgärd
|
|
type WorkflowAction struct {
|
|
Type string `json:"type"` // 'create_contact', 'create_lead', 'update_lead', 'add_tag', 'assign_owner', 'create_task', 'send_message', 'change_status', 'escalate', 'webhook', 'wait'
|
|
Config map[string]interface{} `json:"config"`
|
|
}
|
|
|
|
// CreateWorkflow skapar en ny workflow
|
|
func (h *WorkflowHandler) CreateWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
var req CreateWorkflowRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request")
|
|
return
|
|
}
|
|
|
|
tenantID := getTenantID(r)
|
|
if tenantID == uuid.Nil {
|
|
writeError(w, http.StatusUnauthorized, "missing tenant")
|
|
return
|
|
}
|
|
|
|
triggerConfig, _ := json.Marshal(req.TriggerConfig)
|
|
conditions, _ := json.Marshal(req.Conditions)
|
|
actions, _ := json.Marshal(req.Actions)
|
|
|
|
var workflow Workflow
|
|
err := h.db.QueryRow(`
|
|
INSERT INTO workflows (tenant_id, name, description, trigger, trigger_config, conditions, actions, status)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, 'active')
|
|
RETURNING id, tenant_id, name, description, trigger, trigger_config, conditions, actions, status, created_at, updated_at
|
|
`, tenantID, req.Name, req.Description, req.Trigger, triggerConfig, conditions, actions).Scan(
|
|
&workflow.ID, &workflow.TenantID, &workflow.Name, &workflow.Description, &workflow.Trigger,
|
|
&workflow.TriggerConfig, &workflow.Conditions, &workflow.Actions, &workflow.Status,
|
|
&workflow.CreatedAt, &workflow.UpdatedAt,
|
|
)
|
|
if err != nil {
|
|
log.Error().Err(err).Msg("Failed to create workflow")
|
|
writeError(w, http.StatusInternalServerError, "failed to create workflow")
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusCreated, workflow)
|
|
}
|
|
|
|
// ListWorkflows listar workflows
|
|
func (h *WorkflowHandler) ListWorkflows(w http.ResponseWriter, r *http.Request) {
|
|
tenantID := getTenantID(r)
|
|
if tenantID == uuid.Nil {
|
|
writeError(w, http.StatusUnauthorized, "missing tenant")
|
|
return
|
|
}
|
|
|
|
rows, err := h.db.Query(`
|
|
SELECT id, tenant_id, name, description, trigger, status, execution_count, last_executed_at, created_at, updated_at
|
|
FROM workflows
|
|
WHERE tenant_id = $1
|
|
ORDER BY created_at DESC
|
|
`, tenantID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "database error")
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
|
|
var workflows []Workflow
|
|
for rows.Next() {
|
|
var w Workflow
|
|
rows.Scan(&w.ID, &w.TenantID, &w.Name, &w.Description, &w.Trigger, &w.Status, &w.ExecutionCount, &w.LastExecutedAt, &w.CreatedAt, &w.UpdatedAt)
|
|
workflows = append(workflows, w)
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, map[string]interface{}{"workflows": workflows})
|
|
}
|
|
|
|
// GetWorkflow hämtar en workflow
|
|
func (h *WorkflowHandler) GetWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
id, err := uuid.Parse(chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid id")
|
|
return
|
|
}
|
|
|
|
var workflow Workflow
|
|
err = h.db.QueryRow(`
|
|
SELECT id, tenant_id, name, description, trigger, trigger_config, conditions, actions, status, execution_count, last_executed_at, created_at, updated_at
|
|
FROM workflows WHERE id = $1
|
|
`, id).Scan(
|
|
&workflow.ID, &workflow.TenantID, &workflow.Name, &workflow.Description, &workflow.Trigger,
|
|
&workflow.TriggerConfig, &workflow.Conditions, &workflow.Actions, &workflow.Status,
|
|
&workflow.ExecutionCount, &workflow.LastExecutedAt, &workflow.CreatedAt, &workflow.UpdatedAt,
|
|
)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "workflow not found")
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, workflow)
|
|
}
|
|
|
|
// UpdateWorkflow uppdaterar en workflow
|
|
func (h *WorkflowHandler) UpdateWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
id, err := uuid.Parse(chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid id")
|
|
return
|
|
}
|
|
|
|
var req CreateWorkflowRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request")
|
|
return
|
|
}
|
|
|
|
triggerConfig, _ := json.Marshal(req.TriggerConfig)
|
|
conditions, _ := json.Marshal(req.Conditions)
|
|
actions, _ := json.Marshal(req.Actions)
|
|
|
|
_, err = h.db.Exec(`
|
|
UPDATE workflows
|
|
SET name = $1, description = $2, trigger = $3, trigger_config = $4, conditions = $5, actions = $6, updated_at = NOW()
|
|
WHERE id = $7
|
|
`, req.Name, req.Description, req.Trigger, triggerConfig, conditions, actions, id)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "update failed")
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, map[string]interface{}{"updated": true})
|
|
}
|
|
|
|
// ToggleWorkflow aktiverar/pausar en workflow
|
|
func (h *WorkflowHandler) ToggleWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
id, err := uuid.Parse(chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid id")
|
|
return
|
|
}
|
|
|
|
var req struct {
|
|
Status string `json:"status"`
|
|
}
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request")
|
|
return
|
|
}
|
|
|
|
_, err = h.db.Exec("UPDATE workflows SET status = $1, updated_at = NOW() WHERE id = $2", req.Status, id)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "update failed")
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, map[string]interface{}{"status": req.Status})
|
|
}
|
|
|
|
// ExecuteWorkflow manuellt kör en workflow
|
|
func (h *WorkflowHandler) ExecuteWorkflow(w http.ResponseWriter, r *http.Request) {
|
|
id, err := uuid.Parse(chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid id")
|
|
return
|
|
}
|
|
|
|
var req struct {
|
|
TriggerData map[string]interface{} `json:"trigger_data"`
|
|
}
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request")
|
|
return
|
|
}
|
|
|
|
// Hämta workflow
|
|
var workflow Workflow
|
|
err = h.db.QueryRow(`
|
|
SELECT id, tenant_id, name, trigger, conditions, actions
|
|
FROM workflows WHERE id = $1 AND status = 'active'
|
|
`, id).Scan(&workflow.ID, &workflow.TenantID, &workflow.Name, &workflow.Trigger, &workflow.Conditions, &workflow.Actions)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "workflow not found or not active")
|
|
return
|
|
}
|
|
|
|
// Kör workflow
|
|
execution, err := h.executeWorkflow(workflow, req.TriggerData)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "execution failed")
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, execution)
|
|
}
|
|
|
|
// ListExecutions listar workflow-körningar
|
|
func (h *WorkflowHandler) ListExecutions(w http.ResponseWriter, r *http.Request) {
|
|
workflowID, err := uuid.Parse(chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid workflow id")
|
|
return
|
|
}
|
|
|
|
rows, err := h.db.Query(`
|
|
SELECT id, workflow_id, trigger_event, status, current_step, started_at, completed_at
|
|
FROM workflow_executions
|
|
WHERE workflow_id = $1
|
|
ORDER BY started_at DESC
|
|
LIMIT 50
|
|
`, workflowID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "database error")
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
|
|
var executions []WorkflowExecution
|
|
for rows.Next() {
|
|
var e WorkflowExecution
|
|
rows.Scan(&e.ID, &e.WorkflowID, &e.TriggerEvent, &e.Status, &e.CurrentStep, &e.StartedAt, &e.CompletedAt)
|
|
executions = append(executions, e)
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, map[string]interface{}{"executions": executions})
|
|
}
|
|
|
|
// executeWorkflow kör en workflow
|
|
func (h *WorkflowHandler) executeWorkflow(workflow Workflow, triggerData map[string]interface{}) (*WorkflowExecution, error) {
|
|
// Skapa execution record
|
|
triggerDataJSON, _ := json.Marshal(triggerData)
|
|
|
|
var execution WorkflowExecution
|
|
err := h.db.QueryRow(`
|
|
INSERT INTO workflow_executions (workflow_id, trigger_event, trigger_data, status, current_step, steps)
|
|
VALUES ($1, $2, $3, 'running', 0, '[]')
|
|
RETURNING id, workflow_id, trigger_event, status, current_step, started_at
|
|
`, workflow.ID, workflow.Trigger, triggerDataJSON).Scan(
|
|
&execution.ID, &execution.WorkflowID, &execution.TriggerEvent, &execution.Status, &execution.CurrentStep, &execution.StartedAt,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Parsa actions
|
|
var actions []WorkflowAction
|
|
json.Unmarshal(workflow.Actions, &actions)
|
|
|
|
// Kör varje action
|
|
steps := make([]map[string]interface{}, 0, len(actions))
|
|
for i, action := range actions {
|
|
stepResult := map[string]interface{}{
|
|
"step": i,
|
|
"action": action.Type,
|
|
"status": "completed",
|
|
}
|
|
|
|
// Utför action
|
|
result, err := h.executeAction(action, triggerData)
|
|
if err != nil {
|
|
stepResult["status"] = "failed"
|
|
stepResult["error"] = err.Error()
|
|
|
|
// Uppdatera execution med fel
|
|
h.db.Exec(`
|
|
UPDATE workflow_executions
|
|
SET status = 'failed', error = $1, steps = $2, completed_at = NOW()
|
|
WHERE id = $3
|
|
`, err.Error(), mustMarshal(steps), execution.ID)
|
|
|
|
return &execution, err
|
|
}
|
|
|
|
stepResult["result"] = result
|
|
steps = append(steps, stepResult)
|
|
|
|
// Uppdatera current step
|
|
h.db.Exec("UPDATE workflow_executions SET current_step = $1 WHERE id = $2", i+1, execution.ID)
|
|
}
|
|
|
|
// Markera som completed
|
|
stepsJSON, _ := json.Marshal(steps)
|
|
h.db.Exec(`
|
|
UPDATE workflow_executions
|
|
SET status = 'completed', steps = $1, result = $2, completed_at = NOW()
|
|
WHERE id = $3
|
|
`, stepsJSON, mustMarshal(triggerData), execution.ID)
|
|
|
|
// Uppdatera workflow execution count
|
|
h.db.Exec(`
|
|
UPDATE workflows
|
|
SET execution_count = execution_count + 1, last_executed_at = NOW()
|
|
WHERE id = $1
|
|
`, workflow.ID)
|
|
|
|
execution.Status = "completed"
|
|
return &execution, nil
|
|
}
|
|
|
|
// executeAction utför en enskild action
|
|
func (h *WorkflowHandler) executeAction(action WorkflowAction, context map[string]interface{}) (map[string]interface{}, error) {
|
|
switch action.Type {
|
|
case "create_contact":
|
|
return h.actionCreateContact(action.Config, context)
|
|
case "create_lead":
|
|
return h.actionCreateLead(action.Config, context)
|
|
case "update_lead":
|
|
return h.actionUpdateLead(action.Config, context)
|
|
case "add_tag":
|
|
return h.actionAddTag(action.Config, context)
|
|
case "assign_owner":
|
|
return h.actionAssignOwner(action.Config, context)
|
|
case "create_task":
|
|
return h.actionCreateTask(action.Config, context)
|
|
case "send_message":
|
|
return h.actionSendMessage(action.Config, context)
|
|
case "change_status":
|
|
return h.actionChangeStatus(action.Config, context)
|
|
case "escalate":
|
|
return h.actionEscalate(action.Config, context)
|
|
case "webhook":
|
|
return h.actionWebhook(action.Config, context)
|
|
case "wait":
|
|
// Wait är en no-op i synkron exekvering
|
|
return map[string]interface{}{"waited": true}, nil
|
|
default:
|
|
return nil, fmt.Errorf("unknown action type: %s", action.Type)
|
|
}
|
|
}
|
|
|
|
// Action implementations
|
|
func (h *WorkflowHandler) actionCreateContact(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "create_contact", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionCreateLead(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "create_lead", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionUpdateLead(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "update_lead", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionAddTag(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "add_tag", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionAssignOwner(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "assign_owner", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionCreateTask(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "create_task", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionSendMessage(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "send_message", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionChangeStatus(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "change_status", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionEscalate(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "escalate", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func (h *WorkflowHandler) actionWebhook(config map[string]interface{}, context map[string]interface{}) (map[string]interface{}, error) {
|
|
// TODO: Implementera
|
|
return map[string]interface{}{"action": "webhook", "status": "placeholder"}, nil
|
|
}
|
|
|
|
func mustMarshal(v interface{}) json.RawMessage {
|
|
b, _ := json.Marshal(v)
|
|
return b
|
|
}
|