136 lines
4.8 KiB
JavaScript
136 lines
4.8 KiB
JavaScript
import { Campaign, MailingGroup, GroupSubscriber, Subscriber, SmtpServer, EmailTemplateVersion, DeliveryLog, sequelize } from '../models/index.js';
|
|
import { fillQueueForCampaign } from './queueFiller.js';
|
|
import { Op } from 'sequelize';
|
|
import { topicManager } from './topicManager.js';
|
|
|
|
const BATCH_SIZE = 10000;
|
|
|
|
async function clearKafkaTopics(prefix = 'mail-send-') {
|
|
const deletedCount = await topicManager.clearTopics(prefix);
|
|
console.log(`[queueFillerJob] Cleared ${deletedCount} topics`);
|
|
}
|
|
|
|
export async function processScheduledCampaigns() {
|
|
|
|
const campaigns = await Campaign.findAll({
|
|
where: {
|
|
status: 'scheduled',
|
|
scheduled_at: {
|
|
[Op.lte]: new Date() // Только кампании, запланированные на текущее время или в прошлом
|
|
}
|
|
},
|
|
include: [
|
|
MailingGroup,
|
|
{
|
|
model: EmailTemplateVersion,
|
|
as: 'EmailTemplateVersion'
|
|
}
|
|
],
|
|
});
|
|
|
|
console.log(`[queueFillerJob] Found ${campaigns.length} scheduled campaigns`);
|
|
|
|
if (campaigns.length === 0) {
|
|
console.log(`[queueFillerJob] No scheduled campaigns found`);
|
|
return;
|
|
}
|
|
|
|
for (const campaign of campaigns) {
|
|
console.log(`[queueFillerJob] Processing campaign ${campaign.id}:`, {
|
|
hasEmailTemplateVersion: !!campaign.EmailTemplateVersion,
|
|
templateVersionId: campaign.template_version_id,
|
|
subjectOverride: campaign.subject_override,
|
|
groupId: campaign.group_id
|
|
});
|
|
|
|
if (!campaign.EmailTemplateVersion) {
|
|
console.log(`[queueFillerJob] Campaign ${campaign.id} has no EmailTemplateVersion, skipping`);
|
|
continue;
|
|
}
|
|
|
|
let offset = 0;
|
|
let allSubscriberIds = [];
|
|
|
|
while (true) {
|
|
const groupSubs = await GroupSubscriber.findAll({
|
|
where: { group_id: campaign.group_id },
|
|
attributes: ['subscriber_id'],
|
|
offset,
|
|
limit: BATCH_SIZE,
|
|
raw: true,
|
|
});
|
|
|
|
if (groupSubs.length === 0) break;
|
|
allSubscriberIds.push(...groupSubs.map(gs => gs.subscriber_id));
|
|
if (groupSubs.length < BATCH_SIZE) break;
|
|
offset += BATCH_SIZE;
|
|
}
|
|
|
|
console.log(`[queueFillerJob] Found ${allSubscriberIds.length} subscriber IDs for group ${campaign.group_id}`);
|
|
|
|
if (allSubscriberIds.length === 0) {
|
|
console.log(`[queueFillerJob] No subscribers found for group ${campaign.group_id}, skipping campaign ${campaign.id}`);
|
|
continue;
|
|
}
|
|
|
|
// Проверяем, не завершена ли уже кампания
|
|
if (campaign.status === 'sent' || campaign.status === 'failed') {
|
|
console.log(`[queueFillerJob] Campaign ${campaign.id} already completed with status: ${campaign.status}`);
|
|
continue;
|
|
}
|
|
|
|
// Проверяем, не обрабатывается ли уже кампания
|
|
if (campaign.status === 'sending') {
|
|
console.log(`[queueFillerJob] Campaign ${campaign.id} already being processed`);
|
|
continue;
|
|
}
|
|
|
|
// Проверяем, есть ли уже записи в DeliveryLog для этой кампании
|
|
const campaignLogs = await DeliveryLog.findAll({
|
|
where: { campaign_id: campaign.id },
|
|
limit: 1
|
|
});
|
|
|
|
if (campaignLogs.length > 0) {
|
|
console.log(`[queueFillerJob] Campaign ${campaign.id} already has delivery logs, updating status to 'sent'`);
|
|
await campaign.update({ status: 'sent' });
|
|
continue;
|
|
}
|
|
|
|
for (let i = 0; i < allSubscriberIds.length; i += BATCH_SIZE) {
|
|
const batchIds = allSubscriberIds.slice(i, i + BATCH_SIZE);
|
|
const subscribers = await Subscriber.findAll({
|
|
where: {
|
|
id: { [Op.in]: batchIds },
|
|
status: 'active' // Только активные подписчики
|
|
},
|
|
attributes: ['id', 'email'],
|
|
raw: true,
|
|
});
|
|
|
|
if (subscribers.length === 0) {
|
|
console.log(`[queueFillerJob] No subscribers found for batch IDs:`, batchIds);
|
|
continue;
|
|
}
|
|
|
|
console.log(`[queueFillerJob] Found ${subscribers.length} subscribers for batch:`,
|
|
subscribers.map(s => ({ id: s.id, email: s.email }))
|
|
);
|
|
|
|
const smtpServers = await campaign.getSmtpServers();
|
|
console.log(`[queueFillerJob] Found ${smtpServers.length} SMTP servers for campaign ${campaign.id}:`,
|
|
smtpServers.map(s => ({ id: s.id, name: s.name, host: s.host }))
|
|
);
|
|
|
|
if (smtpServers.length === 0) {
|
|
console.log(`[queueFillerJob] No SMTP servers found for campaign ${campaign.id}`);
|
|
continue;
|
|
}
|
|
|
|
await fillQueueForCampaign(campaign, subscribers, smtpServers);
|
|
}
|
|
|
|
// Обновляем статус кампании на 'sending' только если были отправлены сообщения
|
|
await campaign.update({ status: 'sending' });
|
|
}
|
|
}
|