116 lines
3.3 KiB
TypeScript
116 lines
3.3 KiB
TypeScript
import type { Inbox } from '../generated/prisma/client';
|
|
import { InboxStatus, WarmupStatus, prisma } from '../lib/prisma';
|
|
import { withInboxGuard } from '../lib/inboxGuard';
|
|
import { logger } from '../lib/logger';
|
|
import * as aiService from '../services/ai.service';
|
|
import * as emailService from '../services/email.service';
|
|
import { getSettingOrDefault } from '../services/settings.service';
|
|
import { hasAnyCampaign } from '../services/campaign.service';
|
|
|
|
function startOfToday(): Date {
|
|
const d = new Date();
|
|
d.setHours(0, 0, 0, 0);
|
|
return d;
|
|
}
|
|
|
|
function pickRandom<T>(items: T[]): T {
|
|
return items[Math.floor(Math.random() * items.length)]!;
|
|
}
|
|
|
|
async function getTodaySendCount(senderId: string): Promise<number> {
|
|
return prisma.warmupLog.count({
|
|
where: {
|
|
senderId,
|
|
createdAt: { gte: startOfToday() },
|
|
},
|
|
});
|
|
}
|
|
|
|
export async function bumpDailyRamp(): Promise<void> {
|
|
const increment = parseInt(await getSettingOrDefault('ramp_increment'), 10) || 2;
|
|
const inboxes = await prisma.inbox.findMany({ where: { status: InboxStatus.active } });
|
|
|
|
for (const inbox of inboxes) {
|
|
if (inbox.currentRamp < inbox.dailyLimit) {
|
|
await prisma.inbox.update({
|
|
where: { id: inbox.id },
|
|
data: { currentRamp: Math.min(inbox.currentRamp + increment, inbox.dailyLimit) },
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
let lastRampDate: string | null = null;
|
|
|
|
async function maybeBumpRamp(): Promise<void> {
|
|
const today = new Date().toISOString().slice(0, 10);
|
|
if (lastRampDate !== today) {
|
|
lastRampDate = today;
|
|
await bumpDailyRamp();
|
|
}
|
|
}
|
|
|
|
export async function runDispatcherJob(): Promise<void> {
|
|
logger.info('Dispatcher job started');
|
|
|
|
if (await hasAnyCampaign()) {
|
|
logger.info('Legacy flat dispatcher skipped: campaigns exist');
|
|
return;
|
|
}
|
|
|
|
await maybeBumpRamp();
|
|
|
|
const inboxes: Inbox[] = await prisma.inbox.findMany({
|
|
where: { status: InboxStatus.active },
|
|
});
|
|
|
|
if (inboxes.length < 2) {
|
|
logger.warn('Dispatcher skipped: need at least 2 active inboxes');
|
|
return;
|
|
}
|
|
|
|
for (const sender of inboxes) {
|
|
try {
|
|
const sentToday = await getTodaySendCount(sender.id);
|
|
if (sentToday >= sender.currentRamp) {
|
|
logger.debug({ inboxId: sender.id, sentToday }, 'Sender at daily ramp limit');
|
|
continue;
|
|
}
|
|
|
|
const receivers = inboxes.filter((i) => i.id !== sender.id);
|
|
const receiver = pickRandom(receivers);
|
|
|
|
await withInboxGuard(sender, async () => {
|
|
const { subject, body } = await aiService.generateInitialEmail(sender.industry, {
|
|
senderName: sender.senderName,
|
|
companyName: sender.companyName,
|
|
});
|
|
const { messageId } = await emailService.sendMail(sender, {
|
|
to: receiver.email,
|
|
subject,
|
|
body,
|
|
});
|
|
|
|
await prisma.warmupLog.create({
|
|
data: {
|
|
senderId: sender.id,
|
|
receiverId: receiver.id,
|
|
messageId,
|
|
subject,
|
|
body,
|
|
status: WarmupStatus.sent,
|
|
},
|
|
});
|
|
|
|
logger.info(
|
|
{ sender: sender.email, receiver: receiver.email, messageId },
|
|
'Warmup email dispatched',
|
|
);
|
|
});
|
|
} catch (err) {
|
|
logger.error({ err, inboxId: sender.id }, 'Dispatcher failed for inbox');
|
|
}
|
|
}
|
|
|
|
logger.info('Dispatcher job finished');
|
|
}
|