const { pool } = require('../db'); const { N8N_WHATSAPP_TRIGGER_URL } = require('../config'); const TOP_BUYERS_LIMIT = 100; const MAX_CAMPAIGN_ATTEMPTS = 3; const CAMPAIGN_DELTA_THRESHOLD = 100; const enqueueStockCampaignItem = async (client, item) => { 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 ]); }; 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 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 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 formatProductList = (productNames) => { if (productNames.length <= 2) { return productNames.join(' e '); } return `${productNames.slice(0, -1).join(', ')} e ${productNames[productNames.length - 1]}`; }; const mapCampaignProducts = (groups) => { return Object.entries(groups) .sort(([, aItems], [, bItems]) => { return new Date(aItems[0].created_at).getTime() - new Date(bItems[0].created_at).getTime(); }) .map(([baseProductName, items]) => { const sortedItems = [...items].sort((a, b) => String(a.nome).localeCompare(String(b.nome), 'pt-BR')); return { baseProduct: baseProductName, total_delta: sortedItems.reduce((sum, item) => sum + Number(item.delta_estoque || 0), 0), sizes: sortedItems.map(item => ({ id: item.produto_id, nome: item.nome, delta: item.delta_estoque, saldo: item.saldo })), itemIds: sortedItems.map(item => item.id) }; }); }; const sendWhatsappCampaign = async (products, customers) => { const productNames = products.map(product => product.baseProduct); const productsText = formatProductList(productNames); const allSizes = products.flatMap(product => product.sizes); const totalDelta = products.reduce((sum, product) => sum + product.total_delta, 0); const response = await fetch(N8N_WHATSAPP_TRIGGER_URL, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ baseProduct: productsText, productsText, total_delta: totalDelta, sizes: allSizes, products: products.map(({ itemIds, ...product }) => product), 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; } const groups = rows.reduce((acc, row) => { if (!acc[row.base_product_name]) acc[row.base_product_name] = []; acc[row.base_product_name].push(row); return acc; }, {}); const products = mapCampaignProducts(groups); const ids = products.flatMap(product => product.itemIds); const customers = await getTopBuyersAllTime(); 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, processPendingStockCampaigns };