const { pool } = require('../db'); const { N8N_WHATSAPP_TRIGGER_URL } = require('../config'); const { buildWhatsappCampaignPayload, formatProductList, groupCampaignRows, groupCampaignRowsByBaseProduct, isCampaignEligibleProductName, mapCampaignProducts } = require('./campaignFormatter'); const TOP_BUYERS_LIMIT = 100; const TOP_CLIENTS_DEFAULT_DAYS = 30; const TOP_CLIENTS_DEFAULT_LIMIT = 1000; const TOP_CLIENTS_MAX_LIMIT = 5000; const MAX_CAMPAIGN_ATTEMPTS = 3; const CAMPAIGN_DELTA_THRESHOLD = 100; const SAO_PAULO_TIME_ZONE = 'America/Sao_Paulo'; const NORMALIZED_CUSTOMER_NAME_SQL = "NULLIF(LOWER(TRIM(regexp_replace(COALESCE(cliente_nome, ''), '\\s+', ' ', 'g'))), '')"; const NORMALIZED_CUSTOMER_PHONE_SQL = "NULLIF(regexp_replace(COALESCE(cliente_fone, ''), '\\D', '', 'g'), '')"; const WHATSAPP_CUSTOMER_PHONE_SQL = ` CASE WHEN ${NORMALIZED_CUSTOMER_PHONE_SQL} LIKE '55%' THEN ${NORMALIZED_CUSTOMER_PHONE_SQL} WHEN length(${NORMALIZED_CUSTOMER_PHONE_SQL}) IN (10, 11) THEN '55' || ${NORMALIZED_CUSTOMER_PHONE_SQL} ELSE ${NORMALIZED_CUSTOMER_PHONE_SQL} END `; const CANONICAL_CAMPAIGN_CUSTOMER_NAME_SQL = ` NULLIF(TRIM(regexp_replace( regexp_replace( regexp_replace( regexp_replace(LOWER(COALESCE(cliente_nome, '')), '[^[:alnum:][:space:]]+', ' ', 'g'), '(^|[[:space:]])[0-9]{2,14}([[:space:]]|$)', ' ', 'g' ), '(^|[[:space:]])(ltda|me|eireli|epp)([[:space:]]|$)', ' ', 'g' ), '[[:space:]]+', ' ', 'g' )), '') `; const CUSTOMER_IDENTITY_CTE = ` WITH customer_phone_by_name AS ( SELECT ${NORMALIZED_CUSTOMER_NAME_SQL} as normalized_customer_name, (ARRAY_AGG(NULLIF(cliente_fone, '') ORDER BY data_pedido_date DESC NULLS LAST, id DESC) )[1] as canonical_phone FROM orders WHERE NULLIF(cliente_fone, '') IS NOT NULL AND ${NORMALIZED_CUSTOMER_NAME_SQL} IS NOT NULL GROUP BY normalized_customer_name ), identity_orders AS ( SELECT orders.*, ${NORMALIZED_CUSTOMER_NAME_SQL} as normalized_customer_name, COALESCE( NULLIF(orders.cliente_fone, ''), customer_phone_by_name.canonical_phone, 'name:' || COALESCE(NULLIF(orders.cliente_nome, ''), 'Cliente Desconhecido') ) as customer_key FROM orders LEFT JOIN customer_phone_by_name ON customer_phone_by_name.normalized_customer_name = ${NORMALIZED_CUSTOMER_NAME_SQL} ) `; const normalizeDateParam = (value) => { if (!value) return null; const match = String(value).trim().match(/^(\d{4})-(\d{2})-(\d{2})$/); if (!match) return null; const [, yearValue, monthValue, dayValue] = match; const year = Number(yearValue); const month = Number(monthValue); const day = Number(dayValue); const date = new Date(Date.UTC(year, month - 1, day)); if ( date.getUTCFullYear() !== year || date.getUTCMonth() !== month - 1 || date.getUTCDate() !== day ) { return null; } return `${yearValue}-${monthValue}-${dayValue}`; }; const parsePositiveInteger = (value, defaultValue, maxValue) => { const parsed = Number.parseInt(value, 10); if (!Number.isFinite(parsed) || parsed < 1) return defaultValue; return Math.min(parsed, maxValue); }; const getDateStringInTimeZone = (date = new Date(), timeZone = SAO_PAULO_TIME_ZONE) => { const parts = new Intl.DateTimeFormat('en-US', { timeZone, year: 'numeric', month: '2-digit', day: '2-digit' }).formatToParts(date); const partMap = Object.fromEntries(parts.map(part => [part.type, part.value])); return `${partMap.year}-${partMap.month}-${partMap.day}`; }; const subtractDaysFromDateString = (dateString, daysToSubtract) => { const [year, month, day] = dateString.split('-').map(Number); const date = new Date(Date.UTC(year, month - 1, day)); date.setUTCDate(date.getUTCDate() - daysToSubtract); return date.toISOString().slice(0, 10); }; const getTopClientsDateRange = ({ days = TOP_CLIENTS_DEFAULT_DAYS, start, end } = {}) => { const normalizedDays = parsePositiveInteger(days, TOP_CLIENTS_DEFAULT_DAYS, 3650); const normalizedEnd = normalizeDateParam(end) || getDateStringInTimeZone(); const normalizedStart = normalizeDateParam(start) || subtractDaysFromDateString(normalizedEnd, normalizedDays - 1); return { days: normalizedDays, start: normalizedStart, end: normalizedEnd }; }; const enqueueStockCampaignItem = async (client, item) => { if (!isCampaignEligibleProductName(item.baseProductName || item.nome)) { return false; } const query = ` INSERT INTO stock_campaign_queue ( base_product_name, produto_id, nome, saldo, delta_estoque ) VALUES ($1, $2, $3, $4, $5) `; await client.query(query, [ item.baseProductName, item.produtoId, item.nome, item.saldo, item.deltaEstoque ]); return true; }; const getTopBuyersAllTime = async () => { const result = await pool.query(` SELECT MAX(cliente_nome) as nome, cliente_fone as fone, SUM(quantidade * valor_unitario) as total_gasto, SUM(quantidade) as total_comprado FROM orders WHERE cliente_fone IS NOT NULL AND cliente_fone != '' GROUP BY cliente_fone ORDER BY total_gasto DESC LIMIT $1; `, [TOP_BUYERS_LIMIT]); return result.rows; }; const getTopClientsForCampaign = async ({ days, limit, start, end } = {}) => { const range = getTopClientsDateRange({ days, start, end }); const normalizedLimit = parsePositiveInteger(limit, TOP_CLIENTS_DEFAULT_LIMIT, TOP_CLIENTS_MAX_LIMIT); const result = await pool.query(` WITH campaign_orders AS ( SELECT orders.*, ${CANONICAL_CAMPAIGN_CUSTOMER_NAME_SQL} as canonical_customer_name, ${WHATSAPP_CUSTOMER_PHONE_SQL} as whatsapp_phone FROM orders ) SELECT ( ARRAY_AGG(COALESCE(NULLIF(cliente_nome, ''), 'Cliente Desconhecido') ORDER BY data_pedido_date DESC NULLS LAST, id DESC) )[1] as nome, ( ARRAY_AGG(whatsapp_phone ORDER BY data_pedido_date DESC NULLS LAST, id DESC) FILTER (WHERE whatsapp_phone IS NOT NULL) )[1] as fone, COALESCE(SUM(quantidade * valor_unitario), 0) as total_gasto, COALESCE(SUM(quantidade), 0) as total_comprado, COUNT(DISTINCT COALESCE(NULLIF(pedido_id, ''), data_pedido || '_' || valor_pedido::text))::int as total_pedidos, MAX(data_pedido_date) as ultima_compra, ARRAY_REMOVE(ARRAY_AGG(DISTINCT whatsapp_phone), NULL) as telefones FROM campaign_orders WHERE data_pedido_date >= $1::date AND data_pedido_date <= $2::date AND whatsapp_phone IS NOT NULL GROUP BY COALESCE(canonical_customer_name, whatsapp_phone) ORDER BY total_gasto DESC LIMIT $3; `, [range.start, range.end, normalizedLimit]); const customers = result.rows.map(row => ({ nome: row.nome, fone: row.fone, total_gasto: Number(row.total_gasto || 0), total_comprado: Number(row.total_comprado || 0), total_pedidos: Number(row.total_pedidos || 0), ultima_compra: row.ultima_compra, telefones: Array.isArray(row.telefones) ? row.telefones : [] })); return { campaign: 'top_clients', days: range.days, start: range.start, end: range.end, limit: normalizedLimit, count: customers.length, generated_at: new Date().toISOString(), customers }; }; const claimReadyCampaignItems = async () => { const client = await pool.connect(); try { await client.query('BEGIN'); const result = await client.query(` WITH ready_groups AS ( SELECT base_product_name FROM stock_campaign_queue WHERE status IN ('pending', 'failed') AND attempts < $1 GROUP BY base_product_name HAVING SUM(delta_estoque) >= $2 ), ready_items AS ( SELECT queue.id FROM stock_campaign_queue queue JOIN ready_groups ON ready_groups.base_product_name = queue.base_product_name WHERE queue.status IN ('pending', 'failed') AND queue.attempts < $1 ORDER BY queue.created_at ASC FOR UPDATE OF queue SKIP LOCKED ) UPDATE stock_campaign_queue SET status = 'processing', attempts = attempts + 1, updated_at = CURRENT_TIMESTAMP, last_error = NULL WHERE id IN (SELECT id FROM ready_items) RETURNING *; `, [MAX_CAMPAIGN_ATTEMPTS, CAMPAIGN_DELTA_THRESHOLD]); await client.query('COMMIT'); return result.rows; } catch (error) { await client.query('ROLLBACK'); throw error; } finally { client.release(); } }; const countPendingBelowThresholdGroups = async () => { const result = await pool.query(` SELECT COUNT(*)::int as count FROM ( SELECT base_product_name FROM stock_campaign_queue WHERE status IN ('pending', 'failed') AND attempts < $1 GROUP BY base_product_name HAVING SUM(delta_estoque) < $2 ) below_threshold_groups; `, [MAX_CAMPAIGN_ATTEMPTS, CAMPAIGN_DELTA_THRESHOLD]); return result.rows[0]?.count || 0; }; const getCampaignQueueRows = async () => { const result = await pool.query(` SELECT * FROM stock_campaign_queue ORDER BY created_at DESC, id DESC LIMIT 500; `); return result.rows; }; const getCampaignQueueSummary = async () => { const rows = (await getCampaignQueueRows()) .filter(row => isCampaignEligibleProductName(row.base_product_name || row.nome)); return { threshold: CAMPAIGN_DELTA_THRESHOLD, maxAttempts: MAX_CAMPAIGN_ATTEMPTS, groups: groupCampaignRows(rows), rows }; }; const getCampaignPreview = async () => { const result = await pool.query(` SELECT * FROM stock_campaign_queue WHERE status IN ('pending', 'failed') AND attempts < $1 ORDER BY created_at ASC, id ASC; `, [MAX_CAMPAIGN_ATTEMPTS]); const eligibleRows = result.rows.filter(row => isCampaignEligibleProductName(row.base_product_name || row.nome)); const groups = groupCampaignRowsByBaseProduct(eligibleRows); const readyGroups = {}; const belowThresholdGroups = {}; Object.entries(groups).forEach(([baseProductName, items]) => { const totalDelta = items.reduce((sum, item) => sum + Number(item.delta_estoque || 0), 0); if (totalDelta >= CAMPAIGN_DELTA_THRESHOLD) { readyGroups[baseProductName] = items; } else { belowThresholdGroups[baseProductName] = items; } }); const readyProducts = mapCampaignProducts(readyGroups).map(({ itemIds, ...product }) => product); const belowThresholdProducts = mapCampaignProducts(belowThresholdGroups).map(({ itemIds, ...product }) => product); const customers = await getTopBuyersAllTime(); return { threshold: CAMPAIGN_DELTA_THRESHOLD, readyProducts, belowThresholdProducts, productsText: readyProducts.length ? formatProductList(readyProducts.map(product => product.baseProduct)) : '', customerCount: customers.length, customersPreview: customers.slice(0, 10) }; }; const retryCampaignItems = async ({ ids, baseProductName } = {}) => { const params = []; const filters = [`status IN ('failed', 'skipped')`]; if (Array.isArray(ids) && ids.length) { params.push(ids.map(Number)); filters.push(`id = ANY($${params.length}::int[])`); } if (baseProductName) { params.push(baseProductName); filters.push(`base_product_name = $${params.length}`); } const result = await pool.query(` UPDATE stock_campaign_queue SET status = 'pending', attempts = 0, last_error = NULL, sent_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE ${filters.join(' AND ')} RETURNING *; `, params); return { retried: result.rowCount, rows: result.rows }; }; const updateCampaignItemsStatus = async (ids, status, errorMessage = null) => { if (!ids.length) return; await pool.query(` UPDATE stock_campaign_queue SET status = $1::varchar, last_error = $2, updated_at = CURRENT_TIMESTAMP, sent_at = CASE WHEN $1 = 'sent' THEN CURRENT_TIMESTAMP ELSE sent_at END WHERE id = ANY($3::int[]); `, [status, errorMessage, ids]); }; const skipIneligibleCampaignItems = async (rows) => { const ineligibleRows = rows.filter(row => !isCampaignEligibleProductName(row.base_product_name || row.nome)); const ids = ineligibleRows.map(row => row.id); await updateCampaignItemsStatus(ids, 'skipped', 'Produto não elegível para campanha de cliente.'); return new Set(ineligibleRows.map(row => row.base_product_name)).size; }; const sendWhatsappCampaign = async (products, customers) => { const response = await fetch(N8N_WHATSAPP_TRIGGER_URL, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(buildWhatsappCampaignPayload(products, customers)) }); if (!response.ok) { throw new Error(`WhatsApp webhook returned status ${response.status}`); } }; const processPendingStockCampaigns = async () => { const rows = await claimReadyCampaignItems(); const summary = { claimed: rows.length, sentGroups: 0, skippedGroups: 0, failedGroups: 0, pendingBelowThresholdGroups: await countPendingBelowThresholdGroups() }; if (!rows.length) { return summary; } summary.skippedGroups += await skipIneligibleCampaignItems(rows); const eligibleRows = rows.filter(row => isCampaignEligibleProductName(row.base_product_name || row.nome)); const groups = groupCampaignRowsByBaseProduct(eligibleRows); const products = mapCampaignProducts(groups); const ids = products.flatMap(product => product.itemIds); const customers = await getTopBuyersAllTime(); if (!products.length) { return summary; } if (!customers.length) { await updateCampaignItemsStatus(ids, 'skipped', 'No customers with valid phone numbers found.'); summary.skippedGroups += products.length; return summary; } try { await sendWhatsappCampaign(products, customers); await updateCampaignItemsStatus(ids, 'sent'); summary.sentGroups = products.length; console.log(`[Campaign Queue] Sent one campaign with ${products.length} products to ${customers.length} all-time top buyers.`); } catch (error) { await updateCampaignItemsStatus(ids, 'failed', error.message); summary.failedGroups = products.length; console.error('[Campaign Queue] Failed to send product list campaign:', error); } return summary; }; module.exports = { enqueueStockCampaignItem, getCampaignPreview, getCampaignQueueSummary, getTopClientsForCampaign, retryCampaignItems, processPendingStockCampaigns };