340 lines
11 KiB
JavaScript
340 lines
11 KiB
JavaScript
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 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 groupCampaignRows = (rows) => {
|
|
return Object.values(rows.reduce((acc, row) => {
|
|
const key = `${row.base_product_name}:${row.status}`;
|
|
if (!acc[key]) {
|
|
acc[key] = {
|
|
key,
|
|
baseProductName: row.base_product_name,
|
|
status: row.status,
|
|
totalDelta: 0,
|
|
rowCount: 0,
|
|
attempts: 0,
|
|
lastError: null,
|
|
createdAt: row.created_at,
|
|
updatedAt: row.updated_at,
|
|
sentAt: row.sent_at,
|
|
items: []
|
|
};
|
|
}
|
|
|
|
acc[key].totalDelta += Number(row.delta_estoque || 0);
|
|
acc[key].rowCount += 1;
|
|
acc[key].attempts = Math.max(acc[key].attempts, Number(row.attempts || 0));
|
|
acc[key].lastError = row.last_error || acc[key].lastError;
|
|
acc[key].createdAt = new Date(row.created_at) < new Date(acc[key].createdAt) ? row.created_at : acc[key].createdAt;
|
|
acc[key].updatedAt = new Date(row.updated_at) > new Date(acc[key].updatedAt) ? row.updated_at : acc[key].updatedAt;
|
|
acc[key].sentAt = row.sent_at || acc[key].sentAt;
|
|
acc[key].items.push(row);
|
|
return acc;
|
|
}, {})).sort((a, b) => new Date(b.updatedAt).getTime() - new Date(a.updatedAt).getTime());
|
|
};
|
|
|
|
const getCampaignQueueSummary = async () => {
|
|
const rows = await getCampaignQueueRows();
|
|
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 groups = result.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 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 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,
|
|
getCampaignPreview,
|
|
getCampaignQueueSummary,
|
|
retryCampaignItems,
|
|
processPendingStockCampaigns
|
|
};
|