9c7a383ee2
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 1m22s
448 lines
15 KiB
JavaScript
448 lines
15 KiB
JavaScript
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 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.*,
|
|
${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
|
|
-- A campaign recipient is a WhatsApp destination, so each normalized
|
|
-- phone number must produce exactly one exported customer.
|
|
GROUP BY 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
|
|
};
|