feat: Fase 1 — Motor de Automações + Manus + XOS CRM integrados
Automation Engine (Python FastAPI v2.0.0):
- Novos step types: sub_workflow, split_batch, loop, retry, error_branch,
send_email, send_whatsapp, update_record, create_record, manus_task
- Retry com backoff por step (retry_count + retry_delay_seconds)
- Branching condicional (on_success/on_failure/error_branch)
- Interpolação de variáveis {{campo}} em todos os steps
- Novos tipos de evento CRM: crm.contact.created, crm.deal.stage_changed,
crm.deal.won, crm.deal.lost, crm.ticket.created, crm.message.received, etc.
- Endpoint POST /xos/trigger: recebe eventos CRM e repassa ao Node.js
- Forward bidirecional: engine Python ↔ Node.js XOS webhook
Manus IA — 3 novos tools:
- automation_trigger: dispara automações por ID ou evento do barramento
- xos_action: cria/atualiza contacts, deals, tickets, tarefas via IA
(create_contact, update_contact, create_deal, move_deal_stage,
create_ticket, assign_agent, create_task, create_activity, create_note)
- inbox_action: age na Central de Atendimento via IA
(close_conversation, transfer_conversation, send_message,
add_label, resolve_ticket, escalate_ticket)
XOS CRM Routes — Event Bus wiring:
- POST /contacts emite crm.contact.created
- POST /deals emite crm.deal.created
- PUT /deals/:id/stage emite crm.deal.stage_changed / crm.deal.won / crm.deal.lost
- POST /tickets emite crm.ticket.created
- CRUD completo para xos_automations (GET/POST/PATCH/DELETE)
- POST /automations/:id/execute — execução manual
- POST /automations/webhook/crm-event — recebe eventos do engine Python
- Executor de xos_automations: avalia conditions + executa actions
(send_email, send_whatsapp, create_task, assign_agent, update_field,
move_deal_stage, notify_team, webhook, agent_task)
https://claude.ai/code/session_01DinH3VcgbAv1d9MqnNxzdb
This commit is contained in:
Claude
parent
fc20a0e8b2
commit
e82279d9eb
|
|
@ -335,6 +335,13 @@ class ManusService extends EventEmitter {
|
|||
return this.toolRetailStats(input.period, input.storeId);
|
||||
case "retail_report":
|
||||
return this.toolRetailReport(input.type, input.dateFrom, input.dateTo, input.storeId);
|
||||
// ========== AUTOMAÇÃO + XOS + INBOX ==========
|
||||
case "automation_trigger":
|
||||
return this.toolAutomationTrigger(input.automation_id, input.event_type, input.payload, input.tenant_id, userId);
|
||||
case "xos_action":
|
||||
return this.toolXosAction(input.action, input.data, userId);
|
||||
case "inbox_action":
|
||||
return this.toolInboxAction(input.action, input.conversation_id, input.data, userId);
|
||||
case "finish":
|
||||
let finishOutput = input.answer || "";
|
||||
if (input.chart) {
|
||||
|
|
@ -3871,6 +3878,253 @@ class ManusService extends EventEmitter {
|
|||
return { success: false, output: "", error: `Erro ao gerar relatório: ${error.message}` };
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// TOOL: automation_trigger
|
||||
// ============================================================
|
||||
private async toolAutomationTrigger(
|
||||
automationId: number | undefined,
|
||||
eventType: string | undefined,
|
||||
payload: string | undefined,
|
||||
tenantId: number | undefined,
|
||||
userId: string
|
||||
): Promise<ToolResult> {
|
||||
try {
|
||||
const parsedPayload = payload ? (typeof payload === 'string' ? JSON.parse(payload) : payload) : {};
|
||||
|
||||
if (automationId) {
|
||||
// Direct automation execution via AutomationService
|
||||
const { automationService } = await import("../automations/service");
|
||||
const result = await automationService.runAutomation(automationId, userId, { ...parsedPayload, triggered_by: "manus" });
|
||||
return {
|
||||
success: result.success,
|
||||
output: `Automação #${automationId} ${result.success ? 'executada com sucesso' : 'falhou'}. Log ID: ${result.logId}. ${result.result || result.error || ''}`,
|
||||
};
|
||||
}
|
||||
|
||||
if (eventType) {
|
||||
// Emit event to automation engine via HTTP
|
||||
const engineHost = process.env.AUTOMATION_ENGINE_HOST || "localhost";
|
||||
const enginePort = process.env.AUTOMATION_ENGINE_PORT || "8005";
|
||||
const response = await fetch(`http://${engineHost}:${enginePort}/xos/trigger`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ event_type: eventType, tenant_id: tenantId, payload: parsedPayload }),
|
||||
});
|
||||
|
||||
if (!response.ok) throw new Error(`Engine retornou ${response.status}`);
|
||||
const result: any = await response.json();
|
||||
return {
|
||||
success: true,
|
||||
output: `Evento '${eventType}' emitido. ${result.triggered_handlers?.length || 0} handlers ativados.`,
|
||||
};
|
||||
}
|
||||
|
||||
return { success: false, output: "", error: "Informe automation_id ou event_type" };
|
||||
} catch (error: any) {
|
||||
return { success: false, output: "", error: `Erro ao disparar automação: ${error.message}` };
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// TOOL: xos_action
|
||||
// ============================================================
|
||||
private async toolXosAction(action: string, data: string | object, userId: string): Promise<ToolResult> {
|
||||
try {
|
||||
const parsed = typeof data === 'string' ? JSON.parse(data) : data;
|
||||
|
||||
switch (action) {
|
||||
case "create_contact": {
|
||||
const result = await db.execute(sql`
|
||||
INSERT INTO xos_contacts (name, email, phone, whatsapp, type, company, position, source, tags, notes)
|
||||
VALUES (${parsed.name}, ${parsed.email || null}, ${parsed.phone || null}, ${parsed.whatsapp || null},
|
||||
${parsed.type || 'lead'}, ${parsed.company || null}, ${parsed.position || null},
|
||||
${parsed.source || 'manus'}, ${parsed.tags || null}, ${parsed.notes || null})
|
||||
RETURNING id, name, email, type
|
||||
`);
|
||||
const contact = (result.rows || result)[0] as any;
|
||||
return { success: true, output: `Contato criado: ${contact.name} (ID: ${contact.id}, tipo: ${contact.type})` };
|
||||
}
|
||||
|
||||
case "update_contact": {
|
||||
const { id, ...fields } = parsed;
|
||||
if (!id) return { success: false, output: "", error: "id obrigatório para update_contact" };
|
||||
const sets = Object.entries(fields).map(([k, v]) => `${k} = '${v}'`).join(", ");
|
||||
await db.execute(sql`UPDATE xos_contacts SET ${sql.raw(sets)}, updated_at = NOW() WHERE id = ${id}`);
|
||||
return { success: true, output: `Contato #${id} atualizado com sucesso.` };
|
||||
}
|
||||
|
||||
case "create_deal": {
|
||||
const result = await db.execute(sql`
|
||||
INSERT INTO xos_deals (title, pipeline_id, stage_id, contact_id, company_id, value, currency, assigned_to, expected_close_date, notes)
|
||||
VALUES (${parsed.title}, ${parsed.pipeline_id}, ${parsed.stage_id}, ${parsed.contact_id || null},
|
||||
${parsed.company_id || null}, ${parsed.value || null}, ${parsed.currency || 'BRL'},
|
||||
${parsed.assigned_to || null}, ${parsed.expected_close_date || null}, ${parsed.notes || null})
|
||||
RETURNING id, title, value
|
||||
`);
|
||||
const deal = (result.rows || result)[0] as any;
|
||||
return { success: true, output: `Deal criado: "${deal.title}" (ID: ${deal.id}, valor: ${deal.value})` };
|
||||
}
|
||||
|
||||
case "move_deal_stage": {
|
||||
const { deal_id, stage_id } = parsed;
|
||||
if (!deal_id || !stage_id) return { success: false, output: "", error: "deal_id e stage_id obrigatórios" };
|
||||
await db.execute(sql`UPDATE xos_deals SET stage_id = ${stage_id}, updated_at = NOW() WHERE id = ${deal_id}`);
|
||||
return { success: true, output: `Deal #${deal_id} movido para estágio #${stage_id}.` };
|
||||
}
|
||||
|
||||
case "create_ticket": {
|
||||
const result = await db.execute(sql`
|
||||
INSERT INTO xos_tickets (title, description, contact_id, conversation_id, priority, status, category, assigned_to)
|
||||
VALUES (${parsed.title}, ${parsed.description || null}, ${parsed.contact_id || null},
|
||||
${parsed.conversation_id || null}, ${parsed.priority || 'medium'}, 'open',
|
||||
${parsed.category || null}, ${parsed.assigned_to || null})
|
||||
RETURNING id, title, priority
|
||||
`);
|
||||
const ticket = (result.rows || result)[0] as any;
|
||||
return { success: true, output: `Ticket criado: "${ticket.title}" (ID: ${ticket.id}, prioridade: ${ticket.priority})` };
|
||||
}
|
||||
|
||||
case "assign_agent": {
|
||||
const { conversation_id, agent_id } = parsed;
|
||||
if (!conversation_id) return { success: false, output: "", error: "conversation_id obrigatório" };
|
||||
await db.execute(sql`UPDATE xos_conversations SET assigned_to = ${agent_id}, updated_at = NOW() WHERE id = ${conversation_id}`);
|
||||
return { success: true, output: `Agente #${agent_id} atribuído à conversa #${conversation_id}.` };
|
||||
}
|
||||
|
||||
case "create_task": {
|
||||
const result = await db.execute(sql`
|
||||
INSERT INTO xos_activities (contact_id, deal_id, type, title, description, due_date, assigned_to, status)
|
||||
VALUES (${parsed.contact_id || null}, ${parsed.deal_id || null}, 'task',
|
||||
${parsed.title}, ${parsed.description || null},
|
||||
${parsed.due_date || null}, ${parsed.assigned_to || null}, 'pending')
|
||||
RETURNING id, title
|
||||
`);
|
||||
const task = (result.rows || result)[0] as any;
|
||||
return { success: true, output: `Tarefa criada: "${task.title}" (ID: ${task.id})` };
|
||||
}
|
||||
|
||||
case "create_activity": {
|
||||
const result = await db.execute(sql`
|
||||
INSERT INTO xos_activities (contact_id, deal_id, type, title, description, scheduled_at, assigned_to)
|
||||
VALUES (${parsed.contact_id || null}, ${parsed.deal_id || null}, ${parsed.type || 'note'},
|
||||
${parsed.title}, ${parsed.description || null},
|
||||
${parsed.scheduled_at || null}, ${parsed.assigned_to || null})
|
||||
RETURNING id, title, type
|
||||
`);
|
||||
const act = (result.rows || result)[0] as any;
|
||||
return { success: true, output: `Atividade criada: "${act.title}" (tipo: ${act.type}, ID: ${act.id})` };
|
||||
}
|
||||
|
||||
case "create_note": {
|
||||
const result = await db.execute(sql`
|
||||
INSERT INTO xos_internal_notes (conversation_id, content, created_by, is_pinned)
|
||||
VALUES (${parsed.conversation_id || null}, ${parsed.content}, ${userId}, ${parsed.is_pinned || false})
|
||||
RETURNING id
|
||||
`);
|
||||
const note = (result.rows || result)[0] as any;
|
||||
return { success: true, output: `Nota interna criada (ID: ${note.id})` };
|
||||
}
|
||||
|
||||
default:
|
||||
return { success: false, output: "", error: `Ação XOS desconhecida: ${action}. Use: create_contact, update_contact, create_deal, move_deal_stage, create_ticket, assign_agent, create_task, create_activity, create_note` };
|
||||
}
|
||||
} catch (error: any) {
|
||||
return { success: false, output: "", error: `Erro na ação XOS '${action}': ${error.message}` };
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// TOOL: inbox_action
|
||||
// ============================================================
|
||||
private async toolInboxAction(
|
||||
action: string,
|
||||
conversationId: number | undefined,
|
||||
data: string | object | undefined,
|
||||
userId: string
|
||||
): Promise<ToolResult> {
|
||||
try {
|
||||
const parsed = data ? (typeof data === 'string' ? JSON.parse(data) : data) : {} as any;
|
||||
|
||||
switch (action) {
|
||||
case "close_conversation": {
|
||||
if (!conversationId) return { success: false, output: "", error: "conversation_id obrigatório" };
|
||||
await db.execute(sql`
|
||||
UPDATE xos_conversations SET status = 'closed', closed_at = NOW(), updated_at = NOW()
|
||||
WHERE id = ${conversationId}
|
||||
`);
|
||||
return { success: true, output: `Conversa #${conversationId} fechada.` };
|
||||
}
|
||||
|
||||
case "transfer_conversation": {
|
||||
if (!conversationId) return { success: false, output: "", error: "conversation_id obrigatório" };
|
||||
const { queue_id, agent_id } = parsed;
|
||||
await db.execute(sql`
|
||||
UPDATE xos_conversations SET
|
||||
queue_id = COALESCE(${queue_id || null}, queue_id),
|
||||
assigned_to = COALESCE(${agent_id || null}, assigned_to),
|
||||
updated_at = NOW()
|
||||
WHERE id = ${conversationId}
|
||||
`);
|
||||
return { success: true, output: `Conversa #${conversationId} transferida para fila #${queue_id || 'N/A'} / agente #${agent_id || 'N/A'}.` };
|
||||
}
|
||||
|
||||
case "send_message": {
|
||||
if (!conversationId) return { success: false, output: "", error: "conversation_id obrigatório" };
|
||||
const { content, content_type } = parsed;
|
||||
if (!content) return { success: false, output: "", error: "content obrigatório" };
|
||||
await db.execute(sql`
|
||||
INSERT INTO xos_messages (conversation_id, direction, sender_type, sender_name, content, content_type)
|
||||
VALUES (${conversationId}, 'outbound', 'agent', 'Manus IA', ${content}, ${content_type || 'text'})
|
||||
`);
|
||||
await db.execute(sql`
|
||||
UPDATE xos_conversations SET last_message = ${content}, updated_at = NOW() WHERE id = ${conversationId}
|
||||
`);
|
||||
return { success: true, output: `Mensagem enviada na conversa #${conversationId}: "${content.substring(0, 100)}"` };
|
||||
}
|
||||
|
||||
case "add_label": {
|
||||
if (!conversationId) return { success: false, output: "", error: "conversation_id obrigatório" };
|
||||
const { label } = parsed;
|
||||
await db.execute(sql`
|
||||
UPDATE xos_conversations SET
|
||||
tags = COALESCE(tags, '') || ${label ? ',' + label : ''},
|
||||
updated_at = NOW()
|
||||
WHERE id = ${conversationId}
|
||||
`);
|
||||
return { success: true, output: `Etiqueta '${label}' adicionada à conversa #${conversationId}.` };
|
||||
}
|
||||
|
||||
case "resolve_ticket": {
|
||||
const { ticket_id, resolution } = parsed;
|
||||
if (!ticket_id) return { success: false, output: "", error: "ticket_id obrigatório" };
|
||||
await db.execute(sql`
|
||||
UPDATE xos_tickets SET status = 'resolved', resolution = ${resolution || null},
|
||||
resolved_at = NOW(), updated_at = NOW()
|
||||
WHERE id = ${ticket_id}
|
||||
`);
|
||||
return { success: true, output: `Ticket #${ticket_id} resolvido.` };
|
||||
}
|
||||
|
||||
case "escalate_ticket": {
|
||||
const { ticket_id, priority, reason } = parsed;
|
||||
if (!ticket_id) return { success: false, output: "", error: "ticket_id obrigatório" };
|
||||
await db.execute(sql`
|
||||
UPDATE xos_tickets SET priority = ${priority || 'urgent'},
|
||||
notes = CONCAT(COALESCE(notes, ''), ' [Escalado por Manus: ', ${reason || 'sem motivo'}, ']'),
|
||||
updated_at = NOW()
|
||||
WHERE id = ${ticket_id}
|
||||
`);
|
||||
return { success: true, output: `Ticket #${ticket_id} escalado para prioridade ${priority || 'urgent'}.` };
|
||||
}
|
||||
|
||||
default:
|
||||
return { success: false, output: "", error: `Ação de inbox desconhecida: ${action}. Use: close_conversation, transfer_conversation, send_message, add_label, resolve_ticket, escalate_ticket` };
|
||||
}
|
||||
} catch (error: any) {
|
||||
return { success: false, output: "", error: `Erro na ação de inbox '${action}': ${error.message}` };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const manusService = new ManusService();
|
||||
|
|
|
|||
|
|
@ -733,6 +733,33 @@ export const MANUS_TOOLS: ManusToolDef[] = [
|
|||
dateTo: { type: "string", description: "Data final (YYYY-MM-DD)", required: false },
|
||||
storeId: { type: "number", description: "ID da loja para filtrar", required: false }
|
||||
}
|
||||
},
|
||||
{
|
||||
name: "automation_trigger",
|
||||
description: "Dispara uma automação existente ou emite um evento no barramento. Use para encadear automações, disparar workflows CRM, ou emitir eventos personalizados que ativam outras automações.",
|
||||
parameters: {
|
||||
automation_id: { type: "number", description: "ID da automação a executar (opcional se usar event_type)", required: false },
|
||||
event_type: { type: "string", description: "Tipo de evento a emitir no barramento: crm.contact.created, crm.deal.stage_changed, crm.ticket.created, crm.message.received, manual.trigger, system.event, etc.", required: false },
|
||||
payload: { type: "string", description: "Dados adicionais em JSON para passar à automação ou ao evento (ex: '{\"contact_id\": 42, \"stage\": \"qualified\"}')", required: false },
|
||||
tenant_id: { type: "number", description: "ID do tenant para automações multi-tenant (xos_automations)", required: false }
|
||||
}
|
||||
},
|
||||
{
|
||||
name: "xos_action",
|
||||
description: "Executa ações no XOS CRM: criar/atualizar contatos, mover deals no pipeline, criar tickets, atribuir agentes, criar tarefas, enviar campanhas. Use para automatizar operações de CRM via IA.",
|
||||
parameters: {
|
||||
action: { type: "string", description: "Ação a executar: create_contact, update_contact, create_deal, move_deal_stage, create_ticket, assign_agent, create_task, create_activity, send_campaign, create_note", required: true },
|
||||
data: { type: "string", description: "Dados para a ação em JSON. Ex: create_contact: '{\"name\":\"João\",\"email\":\"joao@ex.com\",\"type\":\"lead\"}'; move_deal_stage: '{\"deal_id\":5,\"stage_id\":3}'; create_ticket: '{\"title\":\"Bug\",\"contact_id\":2,\"priority\":\"high\"}'", required: true }
|
||||
}
|
||||
},
|
||||
{
|
||||
name: "inbox_action",
|
||||
description: "Executa ações na Central de Atendimento: fechar/transferir conversas, enviar mensagens automáticas, atribuir agentes, adicionar etiquetas, criar protocolos, alterar status de tickets de suporte.",
|
||||
parameters: {
|
||||
action: { type: "string", description: "Ação: close_conversation, transfer_conversation, send_message, assign_agent, add_label, create_protocol, resolve_ticket, escalate_ticket, send_csat", required: true },
|
||||
conversation_id: { type: "number", description: "ID da conversa (obrigatório para ações em conversas existentes)", required: false },
|
||||
data: { type: "string", description: "Dados adicionais em JSON. Ex: transfer: '{\"queue_id\":2}'; send_message: '{\"content\":\"Olá!\",\"type\":\"text\"}'; assign_agent: '{\"agent_id\":5}'", required: false }
|
||||
}
|
||||
}
|
||||
];
|
||||
|
||||
|
|
|
|||
|
|
@ -67,6 +67,31 @@ class WorkflowStepType(str, Enum):
|
|||
HTTP_REQUEST = "http"
|
||||
TRANSFORM = "transform"
|
||||
NOTIFY = "notify"
|
||||
SUB_WORKFLOW = "sub_workflow"
|
||||
SPLIT_BATCH = "split_batch"
|
||||
MERGE = "merge"
|
||||
ERROR_HANDLER = "error_handler"
|
||||
RETRY = "retry"
|
||||
SEND_EMAIL = "send_email"
|
||||
SEND_WHATSAPP = "send_whatsapp"
|
||||
UPDATE_RECORD = "update_record"
|
||||
CREATE_RECORD = "create_record"
|
||||
MANUS_TASK = "manus_task"
|
||||
|
||||
|
||||
class CrmEventType(str, Enum):
|
||||
CONTACT_CREATED = "crm.contact.created"
|
||||
CONTACT_UPDATED = "crm.contact.updated"
|
||||
DEAL_CREATED = "crm.deal.created"
|
||||
DEAL_STAGE_CHANGED = "crm.deal.stage_changed"
|
||||
DEAL_WON = "crm.deal.won"
|
||||
DEAL_LOST = "crm.deal.lost"
|
||||
TICKET_CREATED = "crm.ticket.created"
|
||||
TICKET_RESOLVED = "crm.ticket.resolved"
|
||||
FORM_SUBMITTED = "crm.form.submitted"
|
||||
MESSAGE_RECEIVED = "crm.message.received"
|
||||
CONVERSATION_CLOSED = "crm.conversation.closed"
|
||||
CAMPAIGN_SENT = "crm.campaign.sent"
|
||||
|
||||
|
||||
class CronExpression:
|
||||
|
|
@ -254,6 +279,9 @@ class WorkflowStep(BaseModel):
|
|||
config: Dict = {}
|
||||
on_success: Optional[str] = None
|
||||
on_failure: Optional[str] = None
|
||||
retry_count: int = 0
|
||||
retry_delay_seconds: int = 5
|
||||
error_branch: Optional[str] = None
|
||||
|
||||
|
||||
class WorkflowDefinition(BaseModel):
|
||||
|
|
@ -262,6 +290,8 @@ class WorkflowDefinition(BaseModel):
|
|||
steps: List[WorkflowStep]
|
||||
trigger: Optional[str] = None
|
||||
variables: Optional[Dict] = None
|
||||
error_handler: Optional[str] = None
|
||||
max_execution_time: int = 300
|
||||
|
||||
|
||||
class WorkflowExecution(BaseModel):
|
||||
|
|
@ -270,6 +300,12 @@ class WorkflowExecution(BaseModel):
|
|||
variables: Optional[Dict] = None
|
||||
|
||||
|
||||
class XosAutomationTrigger(BaseModel):
|
||||
event_type: str
|
||||
tenant_id: Optional[int] = None
|
||||
payload: Dict = {}
|
||||
|
||||
|
||||
class WorkflowExecutor:
|
||||
def __init__(self):
|
||||
self._workflows: Dict[str, WorkflowDefinition] = {}
|
||||
|
|
@ -308,21 +344,41 @@ class WorkflowExecutor:
|
|||
"variables": {**(workflow.variables or {}), **(variables or {}), **(trigger_data or {})},
|
||||
}
|
||||
|
||||
# Build step index for branching
|
||||
step_map = {s.id: s for s in workflow.steps}
|
||||
step_queue = list(workflow.steps)
|
||||
|
||||
try:
|
||||
for i, step in enumerate(workflow.steps):
|
||||
step_result = self._execute_step(step, execution["variables"])
|
||||
i = 0
|
||||
while i < len(step_queue):
|
||||
step = step_queue[i]
|
||||
step_result, step_status = self._execute_step_with_retry(step, execution["variables"])
|
||||
|
||||
execution["results"].append({
|
||||
"step_id": step.id,
|
||||
"type": step.type,
|
||||
"status": "completed",
|
||||
"status": step_status,
|
||||
"result": step_result,
|
||||
"executed_at": datetime.now().isoformat(),
|
||||
})
|
||||
execution["steps_completed"] = i + 1
|
||||
execution["steps_completed"] += 1
|
||||
|
||||
if isinstance(step_result, dict):
|
||||
execution["variables"].update(step_result.get("output", {}))
|
||||
|
||||
# Handle branching on condition results
|
||||
if step_status == "error" and step.error_branch and step.error_branch in step_map:
|
||||
# Jump to error branch
|
||||
step_queue = step_queue[:i+1] + [step_map[step.error_branch]] + step_queue[i+1:]
|
||||
elif step.type == WorkflowStepType.CONDITION:
|
||||
condition_result = step_result.get("result", False) if isinstance(step_result, dict) else False
|
||||
next_id = step.on_success if condition_result else step.on_failure
|
||||
if next_id and next_id in step_map:
|
||||
# Insert branch step next
|
||||
step_queue = step_queue[:i+1] + [step_map[next_id]] + step_queue[i+1:]
|
||||
|
||||
i += 1
|
||||
|
||||
execution["status"] = "completed"
|
||||
execution["completed_at"] = datetime.now().isoformat()
|
||||
except Exception as e:
|
||||
|
|
@ -336,25 +392,61 @@ class WorkflowExecutor:
|
|||
|
||||
return execution
|
||||
|
||||
def _execute_step_with_retry(self, step: WorkflowStep, variables: Dict):
|
||||
"""Execute step with retry logic. Returns (result, status)."""
|
||||
max_attempts = max(1, step.retry_count + 1)
|
||||
last_error = None
|
||||
for attempt in range(max_attempts):
|
||||
try:
|
||||
result = self._execute_step(step, variables)
|
||||
if isinstance(result, dict) and "error" in result and max_attempts > 1:
|
||||
last_error = result["error"]
|
||||
if attempt < max_attempts - 1:
|
||||
time.sleep(min(step.retry_delay_seconds * (attempt + 1), 60))
|
||||
continue
|
||||
return result, "completed"
|
||||
except Exception as e:
|
||||
last_error = str(e)
|
||||
if attempt < max_attempts - 1:
|
||||
time.sleep(min(step.retry_delay_seconds * (attempt + 1), 60))
|
||||
return {"error": last_error, "attempts": max_attempts}, "error"
|
||||
|
||||
def _execute_step(self, step: WorkflowStep, variables: Dict) -> Any:
|
||||
if step.type == WorkflowStepType.CONDITION:
|
||||
stype = step.type
|
||||
if stype == WorkflowStepType.CONDITION:
|
||||
return self._exec_condition(step.config, variables)
|
||||
elif step.type == WorkflowStepType.ACTION:
|
||||
elif stype == WorkflowStepType.ACTION:
|
||||
return self._exec_action(step.config, variables)
|
||||
elif step.type == WorkflowStepType.DELAY:
|
||||
elif stype == WorkflowStepType.DELAY:
|
||||
delay_seconds = step.config.get("seconds", 1)
|
||||
time.sleep(min(delay_seconds, 30))
|
||||
time.sleep(min(delay_seconds, 300))
|
||||
return {"delayed": delay_seconds}
|
||||
elif step.type == WorkflowStepType.SQL_QUERY:
|
||||
elif stype == WorkflowStepType.SQL_QUERY:
|
||||
return self._exec_query(step.config, variables)
|
||||
elif step.type == WorkflowStepType.HTTP_REQUEST:
|
||||
elif stype == WorkflowStepType.HTTP_REQUEST:
|
||||
return self._exec_http(step.config, variables)
|
||||
elif step.type == WorkflowStepType.TRANSFORM:
|
||||
elif stype == WorkflowStepType.TRANSFORM:
|
||||
return self._exec_transform(step.config, variables)
|
||||
elif step.type == WorkflowStepType.NOTIFY:
|
||||
return {"notified": True, "message": step.config.get("message", ""), "channel": step.config.get("channel", "system")}
|
||||
elif stype == WorkflowStepType.NOTIFY:
|
||||
return self._exec_notify(step.config, variables)
|
||||
elif stype == WorkflowStepType.SUB_WORKFLOW:
|
||||
return self._exec_sub_workflow(step.config, variables)
|
||||
elif stype == WorkflowStepType.SPLIT_BATCH:
|
||||
return self._exec_split_batch(step.config, variables)
|
||||
elif stype == WorkflowStepType.SEND_EMAIL:
|
||||
return self._exec_send_email(step.config, variables)
|
||||
elif stype == WorkflowStepType.SEND_WHATSAPP:
|
||||
return self._exec_send_whatsapp(step.config, variables)
|
||||
elif stype == WorkflowStepType.UPDATE_RECORD:
|
||||
return self._exec_update_record(step.config, variables)
|
||||
elif stype == WorkflowStepType.CREATE_RECORD:
|
||||
return self._exec_create_record(step.config, variables)
|
||||
elif stype == WorkflowStepType.MANUS_TASK:
|
||||
return self._exec_manus_task(step.config, variables)
|
||||
elif stype == WorkflowStepType.LOOP:
|
||||
return self._exec_loop(step.config, variables)
|
||||
else:
|
||||
return {"type": step.type, "status": "unknown_step_type"}
|
||||
return {"type": stype, "status": "executed"}
|
||||
|
||||
def _exec_condition(self, config: Dict, variables: Dict) -> Dict:
|
||||
field = config.get("field", "")
|
||||
|
|
@ -443,8 +535,146 @@ class WorkflowExecutor:
|
|||
value = config.get("value")
|
||||
filtered = [item for item in data if isinstance(item, dict) and item.get(field) == value]
|
||||
return {"output": {"filtered": filtered}}
|
||||
elif operation == "map" and isinstance(data, list):
|
||||
field = config.get("field", "")
|
||||
mapped = [item.get(field) for item in data if isinstance(item, dict)]
|
||||
return {"output": {"mapped": mapped}}
|
||||
elif operation == "sort" and isinstance(data, list):
|
||||
field = config.get("field", "")
|
||||
reverse = config.get("reverse", False)
|
||||
sorted_data = sorted(data, key=lambda x: x.get(field, 0) if isinstance(x, dict) else 0, reverse=reverse)
|
||||
return {"output": {"sorted": sorted_data}}
|
||||
elif operation == "unique" and isinstance(data, list):
|
||||
field = config.get("field", "")
|
||||
seen = set()
|
||||
unique = []
|
||||
for item in data:
|
||||
key = item.get(field) if isinstance(item, dict) else item
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
unique.append(item)
|
||||
return {"output": {"unique": unique}}
|
||||
return {"output": {}}
|
||||
|
||||
def _exec_notify(self, config: Dict, variables: Dict) -> Dict:
|
||||
message = self._interpolate(config.get("message", ""), variables)
|
||||
channel = config.get("channel", "system")
|
||||
# Emit as system event so Node.js can handle delivery
|
||||
event_bus.emit("system.notification", {
|
||||
"message": message,
|
||||
"channel": channel,
|
||||
"title": config.get("title", "Automação"),
|
||||
"level": config.get("level", "info"),
|
||||
})
|
||||
return {"notified": True, "message": message, "channel": channel}
|
||||
|
||||
def _exec_sub_workflow(self, config: Dict, variables: Dict) -> Dict:
|
||||
sub_id = config.get("workflow_id", "")
|
||||
if not sub_id or sub_id not in self._workflows:
|
||||
return {"error": f"Sub-workflow '{sub_id}' nao encontrado"}
|
||||
# Pass current variables merged with config overrides
|
||||
sub_vars = {**variables, **config.get("variables", {})}
|
||||
result = self.execute(sub_id, variables=sub_vars)
|
||||
return {"output": {"sub_workflow_result": result.get("status"), "sub_workflow_vars": result.get("variables", {})}}
|
||||
|
||||
def _exec_split_batch(self, config: Dict, variables: Dict) -> Dict:
|
||||
source = config.get("source", "")
|
||||
batch_size = config.get("batch_size", 10)
|
||||
data = variables.get(source, [])
|
||||
if not isinstance(data, list):
|
||||
return {"error": f"Source '{source}' nao e uma lista"}
|
||||
batches = [data[i:i+batch_size] for i in range(0, len(data), batch_size)]
|
||||
return {"output": {"batches": batches, "batch_count": len(batches), "total_items": len(data)}}
|
||||
|
||||
def _exec_loop(self, config: Dict, variables: Dict) -> Dict:
|
||||
source = config.get("source", "")
|
||||
max_iterations = min(config.get("max_iterations", 100), 1000)
|
||||
data = variables.get(source, [])
|
||||
if not isinstance(data, list):
|
||||
return {"error": f"Source '{source}' nao e uma lista"}
|
||||
results = []
|
||||
for i, item in enumerate(data[:max_iterations]):
|
||||
results.append({"index": i, "item": item})
|
||||
return {"output": {"loop_results": results, "iterations": len(results)}}
|
||||
|
||||
def _exec_send_email(self, config: Dict, variables: Dict) -> Dict:
|
||||
to = self._interpolate(config.get("to", ""), variables)
|
||||
subject = self._interpolate(config.get("subject", ""), variables)
|
||||
body = self._interpolate(config.get("body", ""), variables)
|
||||
# Emit event for Node.js email service to handle
|
||||
event_bus.emit("system.send_email", {"to": to, "subject": subject, "body": body})
|
||||
return {"output": {"email_queued": True, "to": to, "subject": subject}}
|
||||
|
||||
def _exec_send_whatsapp(self, config: Dict, variables: Dict) -> Dict:
|
||||
to = self._interpolate(config.get("to", ""), variables)
|
||||
message = self._interpolate(config.get("message", ""), variables)
|
||||
# Emit event for Node.js WhatsApp service to handle
|
||||
event_bus.emit("system.send_whatsapp", {"to": to, "message": message, "channel_id": config.get("channel_id")})
|
||||
return {"output": {"whatsapp_queued": True, "to": to}}
|
||||
|
||||
def _exec_update_record(self, config: Dict, variables: Dict) -> Dict:
|
||||
if not HAS_PSYCOPG2 or not DATABASE_URL:
|
||||
return {"error": "Database nao disponivel"}
|
||||
table = config.get("table", "")
|
||||
record_id = config.get("id") or variables.get("id")
|
||||
fields = config.get("fields", {})
|
||||
if not table or not record_id or not fields:
|
||||
return {"error": "table, id e fields sao obrigatorios"}
|
||||
# Resolve interpolated values
|
||||
resolved = {k: self._interpolate(str(v), variables) for k, v in fields.items()}
|
||||
set_clauses = ", ".join([f"{k} = %s" for k in resolved.keys()])
|
||||
values = list(resolved.values()) + [record_id]
|
||||
try:
|
||||
conn = psycopg2.connect(DATABASE_URL)
|
||||
cur = conn.cursor()
|
||||
cur.execute(f"UPDATE {table} SET {set_clauses}, updated_at = NOW() WHERE id = %s RETURNING id", values)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return {"output": {"updated": True, "table": table, "id": record_id}}
|
||||
except Exception as e:
|
||||
return {"error": f"Update falhou: {str(e)}"}
|
||||
|
||||
def _exec_create_record(self, config: Dict, variables: Dict) -> Dict:
|
||||
if not HAS_PSYCOPG2 or not DATABASE_URL:
|
||||
return {"error": "Database nao disponivel"}
|
||||
table = config.get("table", "")
|
||||
fields = config.get("fields", {})
|
||||
if not table or not fields:
|
||||
return {"error": "table e fields sao obrigatorios"}
|
||||
resolved = {k: self._interpolate(str(v), variables) for k, v in fields.items()}
|
||||
cols = ", ".join(resolved.keys())
|
||||
placeholders = ", ".join(["%s"] * len(resolved))
|
||||
values = list(resolved.values())
|
||||
try:
|
||||
conn = psycopg2.connect(DATABASE_URL)
|
||||
cur = conn.cursor()
|
||||
cur.execute(f"INSERT INTO {table} ({cols}) VALUES ({placeholders}) RETURNING id", values)
|
||||
new_id = cur.fetchone()[0]
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return {"output": {"created": True, "table": table, "new_id": new_id}}
|
||||
except Exception as e:
|
||||
return {"error": f"Insert falhou: {str(e)}"}
|
||||
|
||||
def _exec_manus_task(self, config: Dict, variables: Dict) -> Dict:
|
||||
prompt = self._interpolate(config.get("prompt", ""), variables)
|
||||
# Emit event — Node.js Manus service will handle execution
|
||||
event_bus.emit("system.manus_task", {
|
||||
"prompt": prompt,
|
||||
"user_id": config.get("user_id") or variables.get("user_id"),
|
||||
"automation_context": variables,
|
||||
})
|
||||
return {"output": {"manus_task_queued": True, "prompt": prompt[:200]}}
|
||||
|
||||
def _interpolate(self, template: str, variables: Dict) -> str:
|
||||
"""Replace {{variable}} placeholders with values from variables dict."""
|
||||
import re
|
||||
def replace(match):
|
||||
key = match.group(1).strip()
|
||||
val = variables.get(key, match.group(0))
|
||||
return str(val) if val is not None else ""
|
||||
return re.sub(r"\{\{(.+?)\}\}", replace, template)
|
||||
|
||||
def get_executions(self, workflow_id: str = None, limit: int = 50) -> List[Dict]:
|
||||
execs = self._executions
|
||||
if workflow_id:
|
||||
|
|
@ -497,8 +727,14 @@ async def health_check():
|
|||
async def version():
|
||||
return {
|
||||
"name": "Arcadia Automation Engine",
|
||||
"version": "1.0.0",
|
||||
"capabilities": ["scheduler", "event_bus", "workflow_executor", "cron", "http_actions", "sql_queries"],
|
||||
"version": "2.0.0",
|
||||
"capabilities": [
|
||||
"scheduler", "event_bus", "workflow_executor", "cron",
|
||||
"http_actions", "sql_queries", "sub_workflows", "loop",
|
||||
"split_batch", "retry", "error_branch", "send_email",
|
||||
"send_whatsapp", "update_record", "create_record",
|
||||
"manus_task", "crm_events", "variable_interpolation",
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -584,7 +820,76 @@ async def event_stats():
|
|||
|
||||
@app.get("/events/types")
|
||||
async def event_types():
|
||||
return {"types": [e.value for e in EventType]}
|
||||
return {
|
||||
"types": [e.value for e in EventType],
|
||||
"crm_types": [e.value for e in CrmEventType],
|
||||
}
|
||||
|
||||
|
||||
# --- XOS CRM Automation trigger ---
|
||||
|
||||
@app.post("/xos/trigger")
|
||||
async def trigger_xos_automation(trigger: XosAutomationTrigger, background_tasks: BackgroundTasks):
|
||||
"""
|
||||
Receives CRM events (contact_created, deal_stage_changed, etc.)
|
||||
and emits them to the event bus so active xos_automations can react.
|
||||
Also calls the Node.js XOS webhook so database-persisted automations execute.
|
||||
"""
|
||||
full_payload = {"tenant_id": trigger.tenant_id, **trigger.payload}
|
||||
triggered = event_bus.emit(trigger.event_type, full_payload)
|
||||
|
||||
# Also emit to generic record.created for wildcard listeners
|
||||
if trigger.event_type.startswith("crm."):
|
||||
event_bus.emit(EventType.RECORD_CREATED, {
|
||||
"entity_type": trigger.event_type.replace("crm.", ""),
|
||||
**full_payload,
|
||||
})
|
||||
|
||||
# Forward to Node.js XOS automations engine (fire DB-persisted automations)
|
||||
background_tasks.add_task(_call_xos_webhook, trigger.event_type, full_payload)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"event_type": trigger.event_type,
|
||||
"triggered_handlers": triggered,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
}
|
||||
|
||||
|
||||
async def _call_xos_webhook(event_type: str, payload: Dict):
|
||||
"""Non-blocking call to Node.js to fire DB-persisted XOS automations."""
|
||||
import urllib.request
|
||||
node_port = os.environ.get("PORT", "5000")
|
||||
url = f"http://localhost:{node_port}/api/xos/automations/webhook/crm-event"
|
||||
try:
|
||||
body = json.dumps({"event_type": event_type, "payload": payload}).encode()
|
||||
req = urllib.request.Request(url, data=body, method="POST")
|
||||
req.add_header("Content-Type", "application/json")
|
||||
req.add_header("X-Internal-Call", "automation-engine")
|
||||
urllib.request.urlopen(req, timeout=5)
|
||||
except Exception as e:
|
||||
print(f"[XOS Webhook] Nao foi possivel notificar Node.js: {e}")
|
||||
|
||||
|
||||
@app.get("/xos/event-types")
|
||||
async def xos_event_types():
|
||||
return {"crm_event_types": [e.value for e in CrmEventType]}
|
||||
|
||||
|
||||
@app.post("/cron/validate")
|
||||
async def validate_cron_post(body: Dict):
|
||||
expression = body.get("expression", "")
|
||||
try:
|
||||
cron = CronExpression(expression)
|
||||
next_runs = []
|
||||
dt = datetime.now()
|
||||
for _ in range(5):
|
||||
dt = cron.next_run(dt)
|
||||
next_runs.append(dt.isoformat())
|
||||
dt += timedelta(minutes=1)
|
||||
return {"valid": True, "expression": expression, "next_runs": next_runs}
|
||||
except ValueError as e:
|
||||
return {"valid": False, "expression": expression, "error": str(e)}
|
||||
|
||||
|
||||
# --- Workflow endpoints ---
|
||||
|
|
|
|||
|
|
@ -2,6 +2,21 @@ import { Router, type Request, type Response } from "express";
|
|||
import { db } from "../../db/index";
|
||||
import { sql } from "drizzle-orm";
|
||||
|
||||
// ── Event Bus helper ─────────────────────────────────────────────────────────
|
||||
async function emitCrmEvent(eventType: string, payload: Record<string, any>) {
|
||||
const engineHost = process.env.AUTOMATION_ENGINE_HOST || "localhost";
|
||||
const enginePort = process.env.AUTOMATION_ENGINE_PORT || "8005";
|
||||
try {
|
||||
await fetch(`http://${engineHost}:${enginePort}/xos/trigger`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ event_type: eventType, payload }),
|
||||
});
|
||||
} catch {
|
||||
// Non-blocking — automation engine may be offline
|
||||
}
|
||||
}
|
||||
|
||||
const router = Router();
|
||||
|
||||
function requireAuth(req: any, res: any, next: any) {
|
||||
|
|
@ -87,7 +102,10 @@ router.post("/contacts", async (req: Request, res: Response) => {
|
|||
RETURNING *
|
||||
`);
|
||||
|
||||
res.status(201).json((result.rows || result)[0]);
|
||||
const contact = (result.rows || result)[0] as any;
|
||||
// Emit CRM event (non-blocking)
|
||||
emitCrmEvent("crm.contact.created", { contact_id: contact.id, name: contact.name, type: contact.type, email: contact.email, source: contact.source });
|
||||
res.status(201).json(contact);
|
||||
} catch (error) {
|
||||
console.error("Error creating contact:", error);
|
||||
res.status(500).json({ error: "Failed to create contact" });
|
||||
|
|
@ -245,7 +263,9 @@ router.post("/deals", async (req: Request, res: Response) => {
|
|||
RETURNING *
|
||||
`);
|
||||
|
||||
res.status(201).json((result.rows || result)[0]);
|
||||
const deal = (result.rows || result)[0] as any;
|
||||
emitCrmEvent("crm.deal.created", { deal_id: deal.id, title: deal.title, value: deal.value, pipeline_id: deal.pipeline_id, stage_id: deal.stage_id, contact_id: deal.contact_id });
|
||||
res.status(201).json(deal);
|
||||
} catch (error) {
|
||||
console.error("Error creating deal:", error);
|
||||
res.status(500).json({ error: "Failed to create deal" });
|
||||
|
|
@ -284,7 +304,10 @@ router.put("/deals/:id/stage", async (req: Request, res: Response) => {
|
|||
RETURNING *
|
||||
`);
|
||||
|
||||
res.json((result.rows || result)[0]);
|
||||
const updated = (result.rows || result)[0] as any;
|
||||
const evtType = status === 'won' ? "crm.deal.won" : status === 'lost' ? "crm.deal.lost" : "crm.deal.stage_changed";
|
||||
emitCrmEvent(evtType, { deal_id: id, stage_id: parseInt(stage_id), status, title: updated?.title });
|
||||
res.json(updated);
|
||||
} catch (error) {
|
||||
console.error("Error updating deal stage:", error);
|
||||
res.status(500).json({ error: "Failed to update deal stage" });
|
||||
|
|
@ -421,7 +444,9 @@ router.post("/tickets", async (req: Request, res: Response) => {
|
|||
RETURNING *
|
||||
`);
|
||||
|
||||
res.status(201).json((result.rows || result)[0]);
|
||||
const ticket = (result.rows || result)[0] as any;
|
||||
emitCrmEvent("crm.ticket.created", { ticket_id: ticket.id, ticket_number: ticketNumber, subject, priority: ticket.priority, contact_id: ticket.contact_id });
|
||||
res.status(201).json(ticket);
|
||||
} catch (error) {
|
||||
console.error("Error creating ticket:", error);
|
||||
res.status(500).json({ error: "Failed to create ticket" });
|
||||
|
|
@ -713,4 +738,316 @@ router.delete("/scheduled-messages/:id", async (req: Request, res: Response) =>
|
|||
}
|
||||
});
|
||||
|
||||
// ========== XOS AUTOMATIONS ENGINE ==========
|
||||
|
||||
router.get("/automations", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const { tenantId } = req.query;
|
||||
let query = sql`SELECT * FROM xos_automations WHERE 1=1`;
|
||||
if (tenantId) query = sql`${query} AND tenant_id = ${parseInt(tenantId as string)}`;
|
||||
query = sql`${query} ORDER BY created_at DESC`;
|
||||
const result = await db.execute(query);
|
||||
res.json(result.rows || result);
|
||||
} catch (error) {
|
||||
console.error("Error fetching xos automations:", error);
|
||||
res.status(500).json({ error: "Failed to fetch automations" });
|
||||
}
|
||||
});
|
||||
|
||||
router.post("/automations", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const { tenantId, name, description, triggerType, triggerConfig, actions, conditions } = req.body;
|
||||
const user = (req as any).user;
|
||||
if (!name || !triggerType) return res.status(400).json({ error: "name and triggerType required" });
|
||||
|
||||
const result = await db.execute(sql`
|
||||
INSERT INTO xos_automations (tenant_id, name, description, trigger_type, trigger_config, actions, conditions, created_by)
|
||||
VALUES (${tenantId || null}, ${name}, ${description || null}, ${triggerType},
|
||||
${triggerConfig ? JSON.stringify(triggerConfig) : null},
|
||||
${actions ? JSON.stringify(actions) : '[]'},
|
||||
${conditions ? JSON.stringify(conditions) : '[]'},
|
||||
${user?.id || null})
|
||||
RETURNING *
|
||||
`);
|
||||
res.status(201).json((result.rows || result)[0]);
|
||||
} catch (error) {
|
||||
console.error("Error creating xos automation:", error);
|
||||
res.status(500).json({ error: "Failed to create automation" });
|
||||
}
|
||||
});
|
||||
|
||||
router.patch("/automations/:id", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const id = parseInt(req.params.id);
|
||||
if (isNaN(id)) return res.status(400).json({ error: "Invalid ID" });
|
||||
const { name, description, triggerType, triggerConfig, actions, conditions, isActive } = req.body;
|
||||
|
||||
const result = await db.execute(sql`
|
||||
UPDATE xos_automations SET
|
||||
name = COALESCE(${name || null}, name),
|
||||
description = COALESCE(${description || null}, description),
|
||||
trigger_type = COALESCE(${triggerType || null}, trigger_type),
|
||||
trigger_config = COALESCE(${triggerConfig ? JSON.stringify(triggerConfig) : null}, trigger_config),
|
||||
actions = COALESCE(${actions ? JSON.stringify(actions) : null}, actions),
|
||||
conditions = COALESCE(${conditions ? JSON.stringify(conditions) : null}, conditions),
|
||||
is_active = COALESCE(${isActive !== undefined ? isActive : null}, is_active),
|
||||
updated_at = NOW()
|
||||
WHERE id = ${id}
|
||||
RETURNING *
|
||||
`);
|
||||
res.json((result.rows || result)[0]);
|
||||
} catch (error) {
|
||||
console.error("Error updating xos automation:", error);
|
||||
res.status(500).json({ error: "Failed to update automation" });
|
||||
}
|
||||
});
|
||||
|
||||
router.delete("/automations/:id", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const id = parseInt(req.params.id);
|
||||
if (isNaN(id)) return res.status(400).json({ error: "Invalid ID" });
|
||||
await db.execute(sql`DELETE FROM xos_automations WHERE id = ${id}`);
|
||||
res.json({ success: true });
|
||||
} catch (error) {
|
||||
console.error("Error deleting xos automation:", error);
|
||||
res.status(500).json({ error: "Failed to delete automation" });
|
||||
}
|
||||
});
|
||||
|
||||
// Execute a single XOS automation immediately
|
||||
router.post("/automations/:id/execute", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const id = parseInt(req.params.id);
|
||||
if (isNaN(id)) return res.status(400).json({ error: "Invalid ID" });
|
||||
const user = (req as any).user;
|
||||
|
||||
const result = await db.execute(sql`SELECT * FROM xos_automations WHERE id = ${id}`);
|
||||
const automation = (result.rows || result)[0] as any;
|
||||
if (!automation) return res.status(404).json({ error: "Automation not found" });
|
||||
|
||||
const execResult = await executeXosAutomation(automation, req.body?.triggerData || {}, user?.id);
|
||||
res.json(execResult);
|
||||
} catch (error) {
|
||||
console.error("Error executing xos automation:", error);
|
||||
res.status(500).json({ error: "Failed to execute automation" });
|
||||
}
|
||||
});
|
||||
|
||||
// Internal webhook: called by the automation engine when a CRM event fires
|
||||
router.post("/automations/webhook/crm-event", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const { event_type, payload, tenant_id } = req.body;
|
||||
if (!event_type) return res.status(400).json({ error: "event_type required" });
|
||||
|
||||
const fired = await fireCrmAutomations(event_type, payload || {}, tenant_id);
|
||||
res.json({ success: true, automations_fired: fired });
|
||||
} catch (error) {
|
||||
console.error("Error processing CRM event webhook:", error);
|
||||
res.status(500).json({ error: "Failed to process CRM event" });
|
||||
}
|
||||
});
|
||||
|
||||
// ── XOS Automation Executor ─────────────────────────────────────────────────
|
||||
|
||||
async function evaluateConditions(conditions: any[], triggerData: Record<string, any>): Promise<boolean> {
|
||||
if (!conditions || conditions.length === 0) return true;
|
||||
for (const cond of conditions) {
|
||||
const actual = triggerData[cond.field];
|
||||
const expected = cond.value;
|
||||
let passes = false;
|
||||
switch (cond.operator) {
|
||||
case "==": passes = actual == expected; break;
|
||||
case "!=": passes = actual != expected; break;
|
||||
case ">": passes = Number(actual) > Number(expected); break;
|
||||
case "<": passes = Number(actual) < Number(expected); break;
|
||||
case ">=": passes = Number(actual) >= Number(expected); break;
|
||||
case "<=": passes = Number(actual) <= Number(expected); break;
|
||||
case "contains": passes = String(actual).includes(String(expected)); break;
|
||||
case "exists": passes = actual !== null && actual !== undefined; break;
|
||||
default: passes = actual == expected;
|
||||
}
|
||||
if (!passes) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async function executeXosAction(action: { type: string; config: any }, triggerData: Record<string, any>, userId?: string): Promise<string> {
|
||||
const config = action.config || {};
|
||||
const interpolate = (s: string) => s?.replace(/\{\{(\w+)\}\}/g, (_: string, k: string) => String(triggerData[k] ?? ''));
|
||||
|
||||
switch (action.type) {
|
||||
case "send_email": {
|
||||
const to = interpolate(config.to || '');
|
||||
const subject = interpolate(config.subject || 'Notificação Arcádia');
|
||||
const body = interpolate(config.body || '');
|
||||
// Route through automation engine event bus
|
||||
const engineHost = process.env.AUTOMATION_ENGINE_HOST || "localhost";
|
||||
const enginePort = process.env.AUTOMATION_ENGINE_PORT || "8005";
|
||||
await fetch(`http://${engineHost}:${enginePort}/events/emit?event_type=system.send_email`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ to, subject, body }),
|
||||
}).catch(() => {});
|
||||
return `Email enfileirado para ${to}`;
|
||||
}
|
||||
|
||||
case "send_whatsapp": {
|
||||
const to = interpolate(config.to || '');
|
||||
const message = interpolate(config.message || '');
|
||||
const engineHost = process.env.AUTOMATION_ENGINE_HOST || "localhost";
|
||||
const enginePort = process.env.AUTOMATION_ENGINE_PORT || "8005";
|
||||
await fetch(`http://${engineHost}:${enginePort}/events/emit?event_type=system.send_whatsapp`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ to, message, channel_id: config.channel_id }),
|
||||
}).catch(() => {});
|
||||
return `WhatsApp enfileirado para ${to}`;
|
||||
}
|
||||
|
||||
case "create_task": {
|
||||
const title = interpolate(config.title || 'Tarefa automática');
|
||||
await db.execute(sql`
|
||||
INSERT INTO xos_activities (contact_id, type, title, description, assigned_to, status)
|
||||
VALUES (${triggerData.contact_id || null}, 'task', ${title},
|
||||
${interpolate(config.description || '')},
|
||||
${config.assigned_to || null}, 'pending')
|
||||
`);
|
||||
return `Tarefa criada: ${title}`;
|
||||
}
|
||||
|
||||
case "assign_agent": {
|
||||
const convId = triggerData.conversation_id || config.conversation_id;
|
||||
const agentId = config.agent_id;
|
||||
if (convId && agentId) {
|
||||
await db.execute(sql`UPDATE xos_conversations SET assigned_to = ${agentId}, updated_at = NOW() WHERE id = ${convId}`);
|
||||
return `Agente #${agentId} atribuído à conversa #${convId}`;
|
||||
}
|
||||
return "assign_agent: conversation_id ou agent_id ausente";
|
||||
}
|
||||
|
||||
case "update_field": {
|
||||
const table = config.table || 'xos_contacts';
|
||||
const recordId = triggerData[config.id_field || 'contact_id'] || config.record_id;
|
||||
const field = config.field;
|
||||
const value = interpolate(config.value || '');
|
||||
if (recordId && field) {
|
||||
await db.execute(sql`UPDATE ${sql.raw(table)} SET ${sql.raw(field)} = ${value}, updated_at = NOW() WHERE id = ${recordId}`);
|
||||
return `Campo ${field} atualizado em ${table}#${recordId}`;
|
||||
}
|
||||
return "update_field: dados insuficientes";
|
||||
}
|
||||
|
||||
case "move_deal_stage": {
|
||||
const dealId = triggerData.deal_id || config.deal_id;
|
||||
const stageId = config.stage_id;
|
||||
if (dealId && stageId) {
|
||||
await db.execute(sql`UPDATE xos_deals SET stage_id = ${stageId}, updated_at = NOW() WHERE id = ${dealId}`);
|
||||
emitCrmEvent("crm.deal.stage_changed", { deal_id: dealId, stage_id: stageId, triggered_by: "automation" });
|
||||
return `Deal #${dealId} movido para estágio #${stageId}`;
|
||||
}
|
||||
return "move_deal_stage: deal_id ou stage_id ausente";
|
||||
}
|
||||
|
||||
case "notify_team": {
|
||||
const message = interpolate(config.message || 'Evento de automação disparado');
|
||||
// Emit notification event
|
||||
const engineHost = process.env.AUTOMATION_ENGINE_HOST || "localhost";
|
||||
const enginePort = process.env.AUTOMATION_ENGINE_PORT || "8005";
|
||||
await fetch(`http://${engineHost}:${enginePort}/events/emit?event_type=system.notification`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ message, channel: config.channel || "system", title: config.title || "Automação" }),
|
||||
}).catch(() => {});
|
||||
return `Notificação enviada: ${message.substring(0, 80)}`;
|
||||
}
|
||||
|
||||
case "webhook": {
|
||||
const url = config.url;
|
||||
if (!url) return "webhook: URL não configurada";
|
||||
try {
|
||||
const resp = await fetch(url, {
|
||||
method: config.method || "POST",
|
||||
headers: { "Content-Type": "application/json", ...(config.headers || {}) },
|
||||
body: JSON.stringify({ trigger_data: triggerData, config }),
|
||||
});
|
||||
return `Webhook ${url} → ${resp.status}`;
|
||||
} catch (e: any) {
|
||||
return `Webhook falhou: ${e.message}`;
|
||||
}
|
||||
}
|
||||
|
||||
case "agent_task": {
|
||||
const prompt = interpolate(config.prompt || 'Execute a automation task');
|
||||
if (userId) {
|
||||
const { automationService } = await import("../automations/service");
|
||||
// Create a temporary automation with agent_task action
|
||||
const tempResult = await fetch(
|
||||
`http://localhost:${process.env.PORT || 5000}/api/automations`,
|
||||
{ method: "GET", headers: { "Content-Type": "application/json" } }
|
||||
).catch(() => null);
|
||||
// Emit manus task via event bus
|
||||
const engineHost = process.env.AUTOMATION_ENGINE_HOST || "localhost";
|
||||
const enginePort = process.env.AUTOMATION_ENGINE_PORT || "8005";
|
||||
await fetch(`http://${engineHost}:${enginePort}/events/emit?event_type=system.manus_task`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ prompt, user_id: userId }),
|
||||
}).catch(() => {});
|
||||
}
|
||||
return `Tarefa do agente IA enfileirada: ${prompt.substring(0, 100)}`;
|
||||
}
|
||||
|
||||
default:
|
||||
return `Ação desconhecida: ${action.type}`;
|
||||
}
|
||||
}
|
||||
|
||||
async function executeXosAutomation(automation: any, triggerData: Record<string, any>, userId?: string) {
|
||||
const actions = typeof automation.actions === 'string' ? JSON.parse(automation.actions) : (automation.actions || []);
|
||||
const conditions = typeof automation.conditions === 'string' ? JSON.parse(automation.conditions) : (automation.conditions || []);
|
||||
|
||||
const conditionsMet = await evaluateConditions(conditions, triggerData);
|
||||
if (!conditionsMet) {
|
||||
return { executed: false, reason: "conditions_not_met", automation_id: automation.id };
|
||||
}
|
||||
|
||||
const results: string[] = [];
|
||||
for (const action of actions) {
|
||||
try {
|
||||
const result = await executeXosAction(action, triggerData, userId);
|
||||
results.push(`✓ ${action.type}: ${result}`);
|
||||
} catch (e: any) {
|
||||
results.push(`✗ ${action.type}: ${e.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Update execution stats
|
||||
await db.execute(sql`
|
||||
UPDATE xos_automations
|
||||
SET execution_count = execution_count + 1, last_executed_at = NOW()
|
||||
WHERE id = ${automation.id}
|
||||
`).catch(() => {});
|
||||
|
||||
return { executed: true, automation_id: automation.id, results };
|
||||
}
|
||||
|
||||
export async function fireCrmAutomations(eventType: string, payload: Record<string, any>, tenantId?: number) {
|
||||
try {
|
||||
let query = sql`SELECT * FROM xos_automations WHERE is_active = true AND trigger_type = ${eventType}`;
|
||||
if (tenantId) query = sql`${query} AND (tenant_id = ${tenantId} OR tenant_id IS NULL)`;
|
||||
|
||||
const result = await db.execute(query);
|
||||
const automations = (result.rows || result) as any[];
|
||||
|
||||
let fired = 0;
|
||||
for (const automation of automations) {
|
||||
await executeXosAutomation(automation, payload).catch(() => {});
|
||||
fired++;
|
||||
}
|
||||
return fired;
|
||||
} catch {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
export default router;
|
||||
|
|
|
|||
Loading…
Reference in New Issue