feat(comm): implement Communication & Lead Layer core

- 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
2026-08-12 10:15:54 +00:00
parent 8cb26cdb15
commit 9e519f5c8f
8 changed files with 1806 additions and 0 deletions
@@ -0,0 +1,453 @@
+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
+}