// AutomationTriggerService.js // Ticket #43305: Fire automation triggers on row creation // Looks up active automations and executes corresponding actions import crypto from 'crypto'; import { dbAll, dbGet, dbRun } from '../database/connection.js'; import { apiLogger } from '../utils/logger.js'; import { enrichSkill } from './SkillEnrichmentService.js'; import { sendMessage, sendAdminAlert } from './TelegramService.js'; import { executeTicketRouting } from './schedule-trigger/pipeline-executors.js'; /** * Find automation tables (automations_list and automation_logs_list) * for the space that contains the given table. * Same pattern as findOrCreateAutomationTables in automations.js, * but read-only — does not create tables if they don't exist. * * @param {number} tableId - The table ID to find automations for * @returns {Promise<{automationsTableId: number, logsTableId: number}|null>} */ async function findAutomationTables(tableId) { // Get project and space for this table const table = await dbGet('SELECT project_id FROM universal_tables WHERE id = ?', [tableId]); if (!table) return null; const project = await dbGet('SELECT space_id FROM projects WHERE id = ?', [table.project_id]); if (!project) return null; // Find System Data project in this space const systemDataProject = await dbGet( "SELECT id FROM projects WHERE space_id = ? AND name = 'System Data'", [project.space_id] ); if (!systemDataProject) return null; // Find automations_list table const automationsTable = await dbGet( "SELECT id FROM universal_tables WHERE project_id = ? AND name = 'automations_list'", [systemDataProject.id] ); if (!automationsTable) return null; // Find automation_logs_list table const logsTable = await dbGet( "SELECT id FROM universal_tables WHERE project_id = ? AND name = 'automation_logs_list'", [systemDataProject.id] ); if (!logsTable) return null; return { automationsTableId: automationsTable.id, logsTableId: logsTable.id }; } /** * Get active automations for a table with a specific trigger type * * @param {number} automationsTableId - The automations_list table ID * @param {number} tableId - The target table ID to filter by * @param {string} triggerType - The trigger type to filter by (e.g., 'row_create') * @returns {Promise} Active automations matching the criteria */ async function getActiveAutomations(automationsTableId, tableId, triggerType) { const rows = await dbAll( 'SELECT id, data FROM table_rows WHERE table_id = ? ORDER BY created_at DESC', [automationsTableId] ); return rows .map(row => { const data = typeof row.data === 'string' ? JSON.parse(row.data) : row.data; return { id: row.id, ...data }; }) .filter(a => a.table_id === tableId && a.trigger_type === triggerType && a.is_active !== false && a.is_active !== 0 ) .map(a => ({ ...a, trigger_config: typeof a.trigger_config === 'string' ? JSON.parse(a.trigger_config || '{}') : (a.trigger_config || {}), action_config: typeof a.action_config === 'string' ? JSON.parse(a.action_config || '{}') : (a.action_config || {}) })); } /** * Log an automation execution result to the automation_logs_list table * * @param {number} logsTableId - The automation_logs_list table ID * @param {Object} logEntry - Log data */ async function logAutomationExecution(logsTableId, logEntry) { try { const now = new Date().toISOString(); const logBaseId = `autolog_${Date.now()}_${crypto.randomBytes(4).toString('hex')}`; const logData = { automation_id: logEntry.automationId, automation_name: logEntry.automationName, row_id: logEntry.rowId, status: logEntry.status, trigger_data: JSON.stringify(logEntry.triggerData || {}), result_data: JSON.stringify(logEntry.resultData || {}), error_message: logEntry.errorMessage || null, duration_ms: logEntry.durationMs, executed_at: now }; await dbRun( 'INSERT INTO table_rows (table_id, base_id, data, created_at, updated_at) VALUES (?, ?, ?, ?, ?)', [logsTableId, logBaseId, JSON.stringify(logData), now, now] ); } catch (err) { apiLogger.error({ err, logEntry }, '[AutomationTrigger] Failed to write automation log'); } } /** * Update automation run count and last_run timestamp * * @param {number} automationRowId - The automation row ID in table_rows * @param {Object} automationData - Current automation data */ async function updateAutomationRunStats(automationRowId, automationData) { try { const now = new Date().toISOString(); automationData.run_count = (automationData.run_count || 0) + 1; automationData.last_run = now; await dbRun( 'UPDATE table_rows SET data = ?, updated_at = ? WHERE id = ?', [JSON.stringify(automationData), now, automationRowId] ); } catch (err) { apiLogger.error({ err, automationRowId }, '[AutomationTrigger] Failed to update automation run stats'); } } /** * Execute the ai_enrich action: call SkillEnrichmentService and update the row * * @param {number} tableId - The table the row belongs to * @param {number} rowId - The row ID to enrich * @param {Object} rowData - The current row data * @returns {Promise} Result with success/error */ async function executeAiEnrich(tableId, rowId, rowData) { const result = await enrichSkill(rowData); if (!result.success) { return { success: false, error: result.error }; } // Merge enrichment into existing row data const enrichedData = { ...rowData, tags: result.enrichment.tags, risk_level: result.enrichment.risk_level, rating: result.enrichment.rating, category: result.enrichment.category, platform: result.enrichment.platform }; // Update the row in the database const now = new Date().toISOString(); await dbRun( 'UPDATE table_rows SET data = ?, updated_at = ? WHERE id = ?', [JSON.stringify(enrichedData), now, rowId] ); return { success: true, enrichment: result.enrichment, durationMs: result.durationMs }; } /** * Execute a webhook action (same pattern as automations.js) * * @param {Object} config - Action config with url, method, headers * @param {Object} rowData - Row data to send * @returns {Promise} Result */ async function executeWebhook(config, rowData) { try { const response = await fetch(config.url, { method: config.method || 'POST', headers: { 'Content-Type': 'application/json', ...(config.headers || {}) }, body: JSON.stringify({ data: rowData, timestamp: new Date().toISOString() }) }); return { success: response.ok, status: response.status, statusText: response.statusText }; } catch (err) { return { success: false, error: err.message }; } } /** * Execute an update_field action * * @param {number} tableId - Table ID * @param {number} rowId - Row ID * @param {Object} config - Action config with column_id and value * @returns {Promise} Result */ async function executeUpdateField(tableId, rowId, config) { try { const { column_id, value } = config; const row = await dbGet('SELECT id, data FROM table_rows WHERE id = ?', [rowId]); if (!row) { return { success: false, error: 'Row not found' }; } const data = typeof row.data === 'string' ? JSON.parse(row.data || '{}') : (row.data || {}); data[column_id] = value; await dbRun( 'UPDATE table_rows SET data = ?, updated_at = ? WHERE id = ?', [JSON.stringify(data), new Date().toISOString(), rowId] ); return { success: true, updated: { [column_id]: value } }; } catch (err) { return { success: false, error: err.message }; } } /** * Evaluate conditions against row data * @param {Array} conditions - Array of { field, operator, value } * @param {Object} rowData - Row data to evaluate * @returns {{ pass: boolean, failedCondition?: string }} */ function evaluateConditions(conditions, rowData) { if (!conditions || !Array.isArray(conditions) || conditions.length === 0) { return { pass: true }; } for (const cond of conditions) { const value = rowData[cond.field]; const expected = cond.value; const strValue = String(value || '').toLowerCase(); const strExpected = String(expected || '').toLowerCase(); let passes = false; switch (cond.operator) { case 'equals': passes = strValue === strExpected; break; case 'not_equals': passes = strValue !== strExpected; break; case 'contains': passes = strValue.includes(strExpected); break; case 'not_contains': passes = !strValue.includes(strExpected); break; case 'is_empty': passes = !value || value === '' || value === null || value === undefined; break; case 'is_not_empty': passes = !!value && value !== '' && value !== null; break; case 'greater_than': passes = Number(value) > Number(expected); break; case 'less_than': passes = Number(value) < Number(expected); break; default: passes = true; } if (!passes) { return { pass: false, failedCondition: `${cond.field} ${cond.operator} ${expected} (actual: ${value})` }; } } return { pass: true }; } /** * Execute a create_row action — supports both flat and array field mapping formats. * * Flat format: { target_table_id, field_mapping: { targetField: "sourceField" } } * Array format: { targetTableId, fieldMappings: [{ sourceColumnId, targetColumnId, staticValue }], conditions: [...] } * * @param {Object} config - Action config * @param {Object} sourceRowData - Source row data * @param {number} [sourceRowId] - Optional source row ID for back-linking * @returns {Promise} Result */ export async function executeCreateRow(config, sourceRowData, sourceRowId) { try { // parseInt guards against JSONB type coercion — action_config may be stored as // a JSON string inside JSONB, causing targetTableId to arrive as a string const rawTargetId = config.targetTableId || config.target_table_id; const targetTableId = rawTargetId ? parseInt(rawTargetId, 10) : null; if (!targetTableId || isNaN(targetTableId)) { return { success: false, error: `No valid target table ID specified (got ${rawTargetId})` }; } // Evaluate conditions first — skip row if conditions don't match const conditions = config.conditions || config.action_conditions; const condResult = evaluateConditions(conditions, sourceRowData); if (!condResult.pass) { return { success: true, skipped: true, reason: condResult.failedCondition }; } // Build new row data from field mappings const newData = {}; // Format 1: Array of { sourceColumnId, targetColumnId, staticValue } const fieldMappings = config.fieldMappings; if (Array.isArray(fieldMappings)) { for (const mapping of fieldMappings) { if (mapping.staticValue !== undefined) { newData[mapping.targetColumnId] = mapping.staticValue; } else if (mapping.sourceColumnId) { newData[mapping.targetColumnId] = sourceRowData[mapping.sourceColumnId]; } } } // Format 2: Flat object { targetField: sourceField } const fieldMapping = config.field_mapping; if (fieldMapping && typeof fieldMapping === 'object' && !Array.isArray(fieldMapping)) { for (const [targetField, sourceField] of Object.entries(fieldMapping)) { newData[targetField] = sourceRowData[sourceField]; } } // Static fields (separate from mappings) if (config.static_fields && typeof config.static_fields === 'object') { Object.assign(newData, config.static_fields); } // Source tracking — link created row back to source if (sourceRowId && config.sourceRowIdField) { newData[config.sourceRowIdField] = sourceRowId; } const now = new Date().toISOString(); // Idempotent upsert (opt-in): when the action declares `upsertKey` and the // new row carries a non-empty value for it, update the existing target row // with the same key instead of inserting a duplicate. This makes re-runs // (e.g. blog → public mirror on every re-publish, or post-publish edits) // propagate into the existing row rather than spawning duplicates. Without // `upsertKey` the action keeps its original insert-only behaviour. const upsertKey = config.upsertKey || config.upsert_key; const upsertVal = upsertKey ? newData[upsertKey] : undefined; if (upsertKey && upsertVal !== undefined && upsertVal !== null && upsertVal !== '') { const existing = await dbGet( 'SELECT id, data FROM table_rows WHERE table_id = ? AND data->>? = ? LIMIT 1', [targetTableId, upsertKey, String(upsertVal)] ); if (existing) { const existingData = typeof existing.data === 'string' ? JSON.parse(existing.data || '{}') : (existing.data || {}); const merged = { ...existingData, ...newData }; await dbRun( 'UPDATE table_rows SET data = ?, updated_at = ? WHERE id = ?', [JSON.stringify(merged), now, existing.id] ); return { success: true, updated_row_id: existing.id, upserted: true, data: merged }; } } const baseId = 'AUTO_' + Math.random().toString(36).substr(2, 8).toUpperCase(); const result = await dbRun( 'INSERT INTO table_rows (table_id, base_id, data, created_at, updated_at) VALUES (?, ?, ?, ?, ?)', [targetTableId, baseId, JSON.stringify(newData), now, now] ); const createdRowId = result.lastID || result.lastInsertRowid; return { success: true, created_row_id: createdRowId, data: newData }; } catch (err) { return { success: false, error: err.message }; } } /** * Execute a notification action — supports Telegram and in-app * * @param {Object} config - Action config with notificationType, recipients, messageTemplate * @param {Object} rowData - Row data for template substitution * @returns {Promise} Result */ async function executeNotification(config, rowData) { try { const { notificationType, recipients, messageTemplate, subject } = config; // Build message from template (replace {{field}} with rowData values) let text = messageTemplate || JSON.stringify(rowData, null, 2); if (messageTemplate) { text = messageTemplate.replace(/\{\{(\w+)\}\}/g, (match, field) => { return rowData[field] !== undefined ? String(rowData[field]) : match; }); } switch (notificationType) { case 'telegram': { // Send to each recipient (chat_id) or admin if none specified const chatIds = recipients && recipients.length > 0 ? recipients : []; if (chatIds.length === 0) { const res = await sendAdminAlert(text); return { success: res.success, type: 'telegram', target: 'admin' }; } const results = []; for (const chatId of chatIds) { const res = await sendMessage(chatId, text); results.push({ chatId, success: res.success }); } return { success: results.every(r => r.success), type: 'telegram', results }; } case 'email': // TODO: Wire to SMTPService when needed return { success: true, type: 'email', message: 'Email notification not yet wired' }; case 'slack': return { success: true, type: 'slack', message: 'Slack notification not yet wired' }; case 'in_app': default: return { success: true, type: notificationType || 'in_app', message: 'In-app notification logged' }; } } catch (err) { return { success: false, error: err.message }; } } /** * Run a single automation: condition gate + action dispatch + log + run-stats. * Shared between row_create and row_update trigger fans-out. */ async function executeAutomation(automation, tableId, rowId, rowData, tables) { const startTime = Date.now(); let result; try { if (automation.action_type !== 'create_row') { const actionConditions = automation.action_config?.conditions || automation.action_config?.action_conditions; const condCheck = evaluateConditions(actionConditions, rowData); if (!condCheck.pass) { result = { success: true, skipped: true, reason: condCheck.failedCondition }; const durationMs = Date.now() - startTime; await logAutomationExecution(tables.logsTableId, { automationId: automation.id, automationName: automation.name, rowId, status: 'success', triggerData: rowData, resultData: result, errorMessage: null, durationMs }); apiLogger.info({ automationId: automation.id, automationName: automation.name, skipped: true, reason: condCheck.failedCondition }, '[AutomationTrigger] Automation skipped (conditions not met)'); return; } } switch (automation.action_type) { case 'ai_enrich': result = await executeAiEnrich(tableId, rowId, rowData); break; case 'webhook': case 'n8n': result = await executeWebhook(automation.action_config, rowData); break; case 'update_field': result = await executeUpdateField(tableId, rowId, automation.action_config); break; case 'create_row': result = await executeCreateRow(automation.action_config, rowData, rowId); break; case 'send_notification': case 'notification': result = await executeNotification(automation.action_config, rowData); break; case 'ticket_routing': result = await executeTicketRouting(automation.action_config, rowData, rowId); break; default: result = { success: false, error: 'Unknown action type: ' + automation.action_type }; } } catch (err) { result = { success: false, error: err.message }; } const durationMs = Date.now() - startTime; await logAutomationExecution(tables.logsTableId, { automationId: automation.id, automationName: automation.name, rowId, status: result.success ? 'success' : 'error', triggerData: rowData, resultData: result, errorMessage: result.error || null, durationMs }); await updateAutomationRunStats(automation.id, automation); apiLogger.info( { automationId: automation.id, automationName: automation.name, success: result.success, durationMs }, '[AutomationTrigger] Automation executed' ); } /** * Fire all row_create automations for a given table and row. * Non-blocking — call without await. */ export async function fireRowCreateTriggers(tableId, rowId, rowData) { try { const tables = await findAutomationTables(tableId); if (!tables) return; const automations = await getActiveAutomations(tables.automationsTableId, tableId, 'row_create'); if (automations.length === 0) return; apiLogger.info( { tableId, rowId, automationCount: automations.length }, '[AutomationTrigger] Firing row_create triggers' ); for (const automation of automations) { await executeAutomation(automation, tableId, rowId, rowData, tables); } } catch (err) { apiLogger.error({ err, tableId, rowId }, '[AutomationTrigger] Error firing row_create triggers'); } } /** * Fire row_update automations for a given table and row. * Watch-field gating prevents action loops: an automation only fires when the * field named by trigger_config.watch_field actually changed (and, if * trigger_config.equals is set, when the new value matches it). An * `update_field` action that mutates a different column won't re-fire, * because the watched field stayed the same on the second pass. * * trigger_config (all optional): * - watch_field {string} — only fire when this field's value changed * - equals {any} — only fire when newValue === equals * * Non-blocking — call without await. */ export async function fireRowUpdateTriggers(tableId, rowId, newData, oldData) { try { const tables = await findAutomationTables(tableId); if (!tables) return; const automations = await getActiveAutomations(tables.automationsTableId, tableId, 'row_update'); if (automations.length === 0) return; const safeOld = oldData || {}; const matched = automations.filter(a => { const cfg = a.trigger_config || {}; const field = cfg.watch_field; if (!field) return true; const before = safeOld[field]; const after = newData ? newData[field] : undefined; if (before === after) return false; if (Object.prototype.hasOwnProperty.call(cfg, 'equals')) { // Loose equality so JSONB number/string round-trips don't miss matches. // eslint-disable-next-line eqeqeq return after == cfg.equals; } return true; }); if (matched.length === 0) return; apiLogger.info( { tableId, rowId, automationCount: matched.length }, '[AutomationTrigger] Firing row_update triggers' ); for (const automation of matched) { await executeAutomation(automation, tableId, rowId, newData, tables); } } catch (err) { apiLogger.error({ err, tableId, rowId }, '[AutomationTrigger] Error firing row_update triggers'); } } export default { fireRowCreateTriggers, fireRowUpdateTriggers };