'use strict' const http = require('http') const https = require('https') const fs = require('fs') const { URL } = require('url') const cfg = require('./config') function log (...args) { console.log(`[${new Date().toISOString().slice(11, 19)}]`, ...args) } function json (res, status, data) { res.writeHead(status, { 'Content-Type': 'application/json' }) res.end(JSON.stringify(data)) } function parseBody (req) { if (req._bodyPromise) return req._bodyPromise // idempotent : un 2e appel (vérif signature en amont + handler) ne re-consomme pas un flux déjà lu req._bodyPromise = new Promise((resolve, reject) => { const chunks = [] req.on('data', c => chunks.push(c)) req.on('end', () => { const raw = Buffer.concat(chunks).toString() req._rawBody = raw // exposé pour la vérif de signature (Twilio/Stripe) — octets bruts const ct = (req.headers['content-type'] || '').toLowerCase() if (ct.includes('application/json')) { try { resolve(JSON.parse(raw)) } catch { resolve({}) } } else if (ct.includes('urlencoded')) { resolve(Object.fromEntries(new URLSearchParams(raw))) } else { resolve(raw) } }) req.on('error', reject) }) return req._bodyPromise } function httpRequest (baseUrl, path, { method = 'GET', body, headers = {}, timeout = 15000 } = {}) { return new Promise((resolve, reject) => { const u = new URL(path, baseUrl) const proto = u.protocol === 'https:' ? https : http const req = proto.request({ hostname: u.hostname, port: u.port || (u.protocol === 'https:' ? 443 : 80), path: u.pathname + u.search, method, headers: { 'Content-Type': 'application/json', ...headers }, timeout, }, (resp) => { let data = '' resp.on('data', c => { data += c }) resp.on('end', () => { try { resolve({ status: resp.statusCode, data: data ? JSON.parse(data) : null }) } catch { resolve({ status: resp.statusCode, data }) } }) }) req.on('error', reject) req.on('timeout', () => { req.destroy(); reject(new Error('Request timeout: ' + path)) }) if (body) req.write(typeof body === 'string' ? body : JSON.stringify(body)) req.end() }) } function _erpFetchOnce (path, opts = {}) { const parsed = new URL(cfg.ERP_URL + path) return new Promise((resolve, reject) => { const req = http.request({ hostname: parsed.hostname, port: parsed.port || 8000, path: parsed.pathname + parsed.search, method: opts.method || 'GET', headers: { Host: cfg.ERP_SITE, Authorization: 'token ' + cfg.ERP_TOKEN, 'Content-Type': 'application/json', ...opts.headers }, }, (res) => { let body = '' res.on('data', c => body += c) res.on('end', () => { try { resolve({ status: res.statusCode, data: JSON.parse(body) }) } catch { resolve({ status: res.statusCode, data: body }) } }) }) req.on('error', reject) req.setTimeout(25000, () => req.destroy(new Error('erpFetch timeout (25s)'))) // évite un hang infini (sinon → 502 du proxy) if (opts.body) req.write(typeof opts.body === 'string' ? opts.body : JSON.stringify(opts.body)) req.end() }) } // Réessaie les LECTURES (GET) sur erreur réseau TRANSITOIRE : Frappe (peu de workers) reset des connexions sous rafale // concurrente (ECONNRESET « socket hang up ») → sans réessai, erp.list renvoyait [] / lançait par intermittence // (cf. navigation Planification = ~5 appels roster simultanés). Les ÉCRITURES ne sont PAS réessayées (double-écriture). async function erpFetch (path, opts = {}) { const method = String(opts.method || 'GET').toUpperCase() let lastErr for (let attempt = 0; attempt < 3; attempt++) { try { return await _erpFetchOnce(path, opts) } catch (e) { lastErr = e const transient = /ECONNRESET|socket hang up|ETIMEDOUT|EAI_AGAIN|timeout|ECONNREFUSED|EPIPE/i.test(String((e && (e.code || e.message)) || '')) if (method !== 'GET' || !transient || attempt === 2) throw e await new Promise(r => setTimeout(r, 150 * (attempt + 1))) // backoff: 150 ms, 300 ms } } throw lastErr } function erpRequest (method, path, body) { return erpFetch(path, { method, ...(body && { body }) }) } // Renvoie TOUTES les fiches Customer dont un champ téléphone se termine par les 10 chiffres (multi-champs // incl. mobile_no), dédupliquées. Permet de choisir parmi plusieurs fiches quand le numéro est partagé. async function lookupCustomersByPhone (phone, limit = 6) { const digits = String(phone || '').replace(/\D/g, '').slice(-10) if (digits.length < 7) return [] const fields = JSON.stringify(['name', 'customer_name', 'cell_phone', 'mobile_no', 'tel_home', 'tel_office', 'email_id', 'territory']) const byName = new Map() for (const field of ['cell_phone', 'mobile_no', 'tel_home', 'tel_office']) { const filters = JSON.stringify([[field, 'like', '%' + digits]]) const path = `/api/resource/Customer?filters=${encodeURIComponent(filters)}&fields=${encodeURIComponent(fields)}&limit_page_length=${limit}` try { const res = await erpFetch(path) if (res.status === 200 && res.data?.data?.length) { for (const c of res.data.data) if (!byName.has(c.name)) byName.set(c.name, { name: c.name, customer_name: c.customer_name, matched_field: field, email: c.email_id || '', territory: c.territory || '' }) } } catch (e) { log('lookupCustomersByPhone ' + field + ':', e.message) } if (byName.size >= limit) break } return [...byName.values()].slice(0, limit) } // Compat : 1re fiche (ou null). Les flux existants continuent de marcher. async function lookupCustomerByPhone (phone) { const all = await lookupCustomersByPhone(phone, 1); return all[0] || null } // Fiches clients par COURRIEL (email_id exact prioritaire, puis email_billing qui peut contenir plusieurs adresses). // Même forme de retour que lookupCustomersByPhone → réutilise le sélecteur de fiche (multi-match) côté Ops. async function lookupCustomersByEmail (email, limit = 6) { const e = String(email || '').trim().toLowerCase() if (!/.+@.+\..+/.test(e)) return [] const fields = JSON.stringify(['name', 'customer_name', 'email_id', 'email_billing', 'cell_phone', 'mobile_no', 'territory']) const byName = new Map() for (const [field, op, val] of [['email_id', '=', e], ['email_billing', 'like', '%' + e + '%']]) { const filters = JSON.stringify([[field, op, val]]) const path = `/api/resource/Customer?filters=${encodeURIComponent(filters)}&fields=${encodeURIComponent(fields)}&limit_page_length=${limit}` try { const res = await erpFetch(path) if (res.status === 200 && res.data?.data?.length) { for (const c of res.data.data) if (!byName.has(c.name)) byName.set(c.name, { name: c.name, customer_name: c.customer_name, matched_field: field, email: c.email_id || e, phone: c.cell_phone || c.mobile_no || '', territory: c.territory || '' }) } } catch (err) { log('lookupCustomersByEmail ' + field + ':', err.message) } if (byName.size >= limit) break } return [...byName.values()].slice(0, limit) } function createCommunication (fields) { return erpFetch('/api/resource/Communication', { method: 'POST', body: JSON.stringify(fields) }) } // --- GenieACS NBI rate limiter --- // Prevents overwhelming GenieACS with concurrent requests const NBI_MAX_CONCURRENT = parseInt(process.env.NBI_MAX_CONCURRENT || '3') const NBI_MIN_INTERVAL_MS = parseInt(process.env.NBI_MIN_INTERVAL_MS || '200') let nbiActive = 0 let nbiLastRequest = 0 const nbiQueue = [] function nbiRequest (path, method = 'GET', body = null) { return new Promise((resolve, reject) => { function execute () { nbiActive++ const now = Date.now() const wait = Math.max(0, NBI_MIN_INTERVAL_MS - (now - nbiLastRequest)) setTimeout(() => { nbiLastRequest = Date.now() httpRequest(cfg.GENIEACS_NBI_URL, path, { method, body, timeout: 30000 }) .then(resolve).catch(reject) .finally(() => { nbiActive-- if (nbiQueue.length > 0) nbiQueue.shift()() }) }, wait) } if (nbiActive >= NBI_MAX_CONCURRENT) { nbiQueue.push(execute) } else { execute() } }) } function deepGetValue (obj, path) { const node = path.split('.').reduce((o, k) => o?.[k], obj) return node?._value !== undefined ? node._value : null } // CORS partagé pour les endpoints PUBLICS (données publiques, lecture seule + écritures internes). // Pose les en-têtes via setHeader (fusionnés par writeHead de json()). methods optionnel. function cors (res, methods = 'GET, POST, OPTIONS') { res.setHeader('Access-Control-Allow-Origin', '*') res.setHeader('Access-Control-Allow-Methods', methods) res.setHeader('Access-Control-Allow-Headers', 'Content-Type') } // ── Utilitaires partagés (stores JSON, sets de dédup, data-URL) — regroupés ici pour éviter la duplication ── // Lit un fichier JSON ; renvoie `fallback` (copie) si absent/illisible. Comportement identique aux load*() qu'il remplace. function readJsonFile (path, fallback) { try { return JSON.parse(fs.readFileSync(path, 'utf8')) } catch (e) { return (fallback !== undefined ? fallback : null) } } // Écrit un objet en JSON indenté ; renvoie true/false. Best-effort (log en cas d'échec). function writeJsonFile (path, obj) { try { fs.writeFileSync(path, JSON.stringify(obj, null, 2)); return true } catch (e) { log('writeJsonFile ' + path + ': ' + e.message); return false } } // Set de dédup persistée (ex. ids de messages déjà traités). Lit un tableau JSON → Set. function loadSeenSet (path) { try { return new Set(JSON.parse(fs.readFileSync(path, 'utf8'))) } catch (e) { return new Set() } } // Sauve un Set en tableau JSON, borné aux `cap` derniers (évite la croissance infinie). function saveSeenSet (path, set, cap = 3000) { try { fs.writeFileSync(path, JSON.stringify([...set].slice(-cap))) } catch (e) { /* */ } } // Retire le préfixe data:image/...;base64, → base64 pur (pour l'OCR vision). function stripDataUrl (s) { return String(s || '').replace(/^data:[^;]+;base64,/, '') } // NOS PROPRES domaines (incl. sous-domaines infra) — SOURCE UNIQUE. Une adresse qui matche ici n'est // JAMAIS un client : ni à matcher comme fiche (triage), ni à utiliser comme clé de regroupement (ingest). const OWN_DOMAINS = /@([a-z0-9-]+\.)*(targo\.ca|targointernet\.com|gigafibre\.ca)$/i module.exports = { log, json, parseBody, httpRequest, cors, erpFetch, erpRequest, lookupCustomerByPhone, lookupCustomersByPhone, lookupCustomersByEmail, createCommunication, nbiRequest, deepGetValue, readJsonFile, writeJsonFile, loadSeenSet, saveSeenSet, stripDataUrl, OWN_DOMAINS, }