/** * webhookProcessor.js * * Core ServiceChannel → Webex webhook handling logic. * * Extracted during the 2026-05-28 cleanup refactor from the original monolith in index.js. * * This module owns: * - Per-workOrderId mutex + queue to safely handle bursty ServiceChannel events * - Room creation / lookup via mappings DB * - Member seeding (first time only) * - Message formatting per EventType (including xAI summarization for new WOs) * - Posting to the correct Webex space * * IMPORTANT DB NOTE (production constraint): * This processor does NOT create its own database connection. * The caller must pass in the active sqlite3 `db` instance. * The physical DB file used by production must never be moved or renamed. * * Usage (during transition): * import { createWebhookProcessor } from './webhookProcessor.js'; * const processor = createWebhookProcessor({ * db, * webex: botClient, * summarizeDescription: mySummarizer, * teamId: process.env.WEBEX_TEAM_ID, * defaultMembers: [...], * }); * * // Then in the webhook route: * await processor.processWebhook(payload); */ import { Mutex } from 'async-mutex'; import { logger } from '../utils/logger.js'; import approvalService from './approvalService.js'; import invoiceApprovalService from './invoiceApprovalService.js'; import shipmentTrackingService from './shipmentTrackingService.js'; import attachmentService from './attachmentService.js'; import { shouldSuppressAttachmentPost, shouldSuppressWebhookNote, } from '../db/uploadEchoSuppress.js'; export function createWebhookProcessor(options = {}) { const { db, webex, // expected to have: createRoom, addMember, sendMarkdown, getRoom? summarizeDescription, // async (rawDescription, xaiToken, opts?) => { summary } teamId, defaultMembers = [], xaiToken, // for the simple initial-description summarizer // secrets are now loaded directly inside the modules that need them } = options; if (!db) { throw new Error('webhookProcessor requires a sqlite3 db instance'); } if (!webex) { throw new Error('webhookProcessor requires a webex client (botClient)'); } // Per-workOrderId concurrency control (private to this processor instance) const mutexes = new Map(); // workOrderId → Mutex const pendingQueues = new Map(); // workOrderId → Array<{payload, startTime}> function scheduleAttachmentPost({ roomId, workOrderId, woNumber, attachmentIds = null }) { attachmentService.postWorkOrderAttachments({ db, webex, roomId, workOrderId, woNumber, attachmentIds, }).catch((err) => { logger('webhookProcessor:attachments', `Post error for WO ${workOrderId}: ${err.message}`, 'warn'); }); } /** * Main entry point — call this for every incoming ServiceChannel webhook payload. * Safe to call concurrently; internal mutex + queue handles ordering per WO. */ async function processWebhook(payload) { const startTime = Date.now(); const { Object: obj } = payload || {}; if (!obj || !obj.Id) { logger('webhookProcessor', 'Invalid payload — missing Object.Id'); return; } const workOrderId = obj.Id; // Ensure we have a mutex and queue for this work order if (!mutexes.has(workOrderId)) { mutexes.set(workOrderId, new Mutex()); } const mutex = mutexes.get(workOrderId); if (!pendingQueues.has(workOrderId)) { pendingQueues.set(workOrderId, []); } const queue = pendingQueues.get(workOrderId); queue.push({ payload, startTime }); let release; try { release = await mutex.acquire(); // 1. Find or create the Webex room for this work order const row = await new Promise((resolve, reject) => { db.get('SELECT roomId FROM mappings WHERE workOrderId = ?', [workOrderId], (err, r) => { err ? reject(err) : resolve(r); }); }); let roomId = row?.roomId; let roomJustCreated = false; if (!roomId) { logger('webhookProcessor', `Creating new room for WO-${workOrderId}`); const storeLabel = obj.LocationStoreId != null && obj.LocationStoreId !== '' ? String(obj.LocationStoreId) : '?'; const title = `ServChan WO-${obj.Number} | Store ${storeLabel} | ${obj.LocationName || ''}`; const created = await webex.createRoom(title, teamId); roomId = created.id; await new Promise((resolve, reject) => { db.run( 'INSERT OR REPLACE INTO mappings (workOrderId, roomId) VALUES (?, ?)', [workOrderId, roomId], (err) => (err ? reject(err) : resolve()) ); }); logger('webhookProcessor', `Created space ${roomId} for WO-${obj.Number}`); roomJustCreated = true; // Seed default members (best effort) if (defaultMembers.length > 0) { await Promise.allSettled( defaultMembers.map(email => webex.addMember(roomId, email)) ).then(results => { results.forEach((r, i) => { if (r.status === 'rejected') { logger('webhookProcessor:addMember', `${defaultMembers[i]} failed: ${r.reason?.message || r.reason}`); } }); }); } } // 2. Drain the queue and post messages for this work order while (queue.length > 0) { const { payload: p, startTime: itemStart } = queue.shift(); let text = ''; let suppressRoomPost = false; if (p.EventType === 'WorkOrderCreated') { let summaryResult; try { summaryResult = summarizeDescription ? await summarizeDescription(p.Object.Description, xaiToken, { maxTokens: 300 }) : { summary: p.Object.Description?.substring(0, 300) || '' }; } catch (sumErr) { logger('webhookProcessor', `Summarization failed for new WO ${p.Object.Id}: ${sumErr.message}`, 'warn'); summaryResult = { summary: p.Object.Description?.substring(0, 400) || 'No description' }; } // LocationStoreId can arrive as a numeric string, a number, or be // missing. Old code did `LocationStoreId * 1` which produced NaN // when the field was absent and dropped leading zeros for strings. const storeLabel = p.Object.LocationStoreId != null && p.Object.LocationStoreId !== '' ? String(p.Object.LocationStoreId) : '?'; text = `## [New Work Order Created](https://www.servicechannel.com/sc/wo/Workorders/index?id=${p.Object.Id})\n\n` + `### ServChan WO-${p.Object.Number} | Store ${storeLabel} | ${p.Object.LocationName || ''}\n` + `**Trade:** ${p.Object.Trade || 'N/A'} | ${p.Object.ProviderName || 'N/A'}\n` + `**Priority:** ${p.Object.Priority || 'N/A'} | ${p.Object.Category || 'N/A'} | ${p.Object.ProblemCode || 'N/A'}\n` + `**Status:** ${p.Object.Status?.Primary || 'N/A'} | ${p.Object.Status?.Extended || 'N/A'}\n` + `**Description:** ${summaryResult?.summary || p.Object.Description?.substring(0, 300) || 'No description'}\n\n`; } else if (p.EventType === 'WorkOrderNoteAdded') { const note = p.Object.Notes?.[0] || {}; if (await shouldSuppressWebhookNote(db, workOrderId, note)) { suppressRoomPost = true; logger('webhookProcessor', `Suppressed note echo for WO ${workOrderId}`); } else { text = `### [Note Added](https://www.servicechannel.com/sc/wo/Workorders/index?id=${p.Object.Id})\n\n` + `${note.NoteData || 'No note content'} — ${note.CreatedBy || 'Unknown'}\n\n` + `**Status:** ${p.Object.Status?.Primary || 'N/A'} | ${p.Object.Status?.Extended || 'N/A'}`; } } else { // Generic fallback for other events text = `**Work Order Update**\n\n` + `**WO Number:** ${p.Object.Number || 'Unknown'}\n` + `**Event:** ${p.EventType || 'Unknown'}\n` + `**Current Status:** ${p.Object.Status?.Primary || 'N/A'}\n` + `**Updated:** ${new Date().toISOString()}`; } if (!suppressRoomPost) { if (!text.trim()) { text = `Webhook received for Work Order ${workOrderId} (${p.EventType || 'unknown event'}).`; } try { await webex.sendMarkdown(roomId, text); logger('webhookProcessor', `Posted for ${workOrderId} (${Date.now() - itemStart}ms)`); } catch (postErr) { const details = postErr.response?.data ? JSON.stringify(postErr.response.data) : postErr.message; logger('webhookProcessor', `Post failed for ${workOrderId}: ${details}`, 'error'); } } // Auto-post attachments referenced in this note if (p.EventType === 'WorkOrderNoteAdded') { const note = p.Object.Notes?.[0] || {}; if (note.AttachmentIds?.length) { const suppressAttachments = suppressRoomPost || await shouldSuppressAttachmentPost(db, workOrderId, note.AttachmentIds); if (!suppressAttachments) { scheduleAttachmentPost({ roomId, workOrderId, woNumber: p.Object.Number, attachmentIds: note.AttachmentIds, }); } } } // Auto-remove stale approval cards when resolved in ServiceChannel const noteDataForApproval = (p.EventType === 'WorkOrderNoteAdded' && p.Object.Notes?.[0]?.NoteData) || null; approvalService.removeApprovalCardIfResolved({ db, workOrderId, woObj: p.Object, noteData: noteDataForApproval, }).catch((e) => { logger('webhookProcessor:approval', `Card remove error for ${workOrderId}: ${e.message}`, 'warn'); }); invoiceApprovalService.removeInvoiceCardIfResolved({ db, workOrderId, noteData: noteDataForApproval, }).catch((e) => { logger('webhookProcessor:invoiceApproval', `Card remove error for ${workOrderId}: ${e.message}`, 'warn'); }); if (p.EventType === 'WorkOrderNoteAdded' && noteDataForApproval) { shipmentTrackingService.registerShipmentsFromNote({ db, webex, roomId, workOrderId, woNumber: p.Object.Number, noteText: noteDataForApproval, }).catch((e) => { logger('webhookProcessor:shipmentTracking', `Register error for ${workOrderId}: ${e.message}`, 'warn'); }); } // NEW: Proposal approval card for WAITING FOR APPROVAL status (additive to text note) const ext = (p.Object?.Status?.Extended || '').toUpperCase(); if (ext.includes('WAITING FOR APPROVAL')) { // Capture the note that usually contains "Proposal #12345 has been created" // so we can look up the specific proposal and its itemized parts + labor. const noteData = noteDataForApproval; // fire-and-forget; uses the injected webex (WebexService) or falls back internally approvalService.fetchAndPostApprovalCardIfNeeded(webex, roomId, p.Object, noteData, { db }).catch(e => { logger('webhookProcessor:approval', `Card post error for ${workOrderId}: ${e.message}`, 'warn'); }); } } if (roomJustCreated) { scheduleAttachmentPost({ roomId, workOrderId, woNumber: obj.Number, }); } } catch (err) { logger('webhookProcessor', `Critical error for ${workOrderId}: ${err.message}\n${err.stack}`, 'error'); } finally { if (release) release(); // Clean up empty queues/mutexes if (queue.length === 0) { pendingQueues.delete(workOrderId); mutexes.delete(workOrderId); } } } return { processWebhook, // Expose internal state for debugging / the /cleanup-test style endpoints if needed later _getInternalState: () => ({ mutexes, pendingQueues }), }; } export default createWebhookProcessor;