perf: batch CMPP inbound workflow processing
This commit is contained in:
hectorzhao
@@ -41,6 +41,20 @@ type PersistedInboundWorkflowRow = {
|
||||
|
||||
type InboundApplication = NonNullable<Awaited<ReturnType<SendSubmissionService['findInboundApplication']>>>;
|
||||
|
||||
type InboundTemplate = Prisma.SmsTemplateGetPayload<{ include: { signature: true } }>;
|
||||
|
||||
type InboundBatchCandidate = {
|
||||
item: ClaimedInboundWorkflow;
|
||||
payload: InboundWorkflowPayload;
|
||||
application: InboundApplication;
|
||||
phoneNumber: string;
|
||||
template?: InboundTemplate;
|
||||
signatureId: string;
|
||||
templateVariables?: Record<string, unknown>;
|
||||
drainageInfoId?: string;
|
||||
clientSrcId: string | null;
|
||||
};
|
||||
|
||||
type InboundWorkflowResponse = {
|
||||
accepted: boolean;
|
||||
tenantId: string;
|
||||
@@ -1227,7 +1241,7 @@ startInboundWorkflowWorker() {
|
||||
this.inboundWorkflowPumping = true;
|
||||
try {
|
||||
await this.refreshInboundWorkflowMetrics();
|
||||
const claimed = await this.claimInboundWorkflows(available);
|
||||
const claimed = await this.measureInboundStage('worker_claim', () => this.claimInboundWorkflows(available));
|
||||
const applications = claimed.length === 0
|
||||
? []
|
||||
: await this.prisma.smsApplication.findMany({
|
||||
@@ -1235,13 +1249,15 @@ startInboundWorkflowWorker() {
|
||||
include: { tenant: true, ipAllowlist: true },
|
||||
});
|
||||
const applicationById = new Map(applications.map((application) => [application.id, application]));
|
||||
for (const item of claimed) {
|
||||
this.inboundWorkflowInFlight += 1;
|
||||
const batchSize = positiveInteger(process.env.API_INBOUND_WORKFLOW_BATCH_SIZE, 64);
|
||||
for (let offset = 0; offset < claimed.length; offset += batchSize) {
|
||||
const batch = claimed.slice(offset, offset + batchSize);
|
||||
this.inboundWorkflowInFlight += batch.length;
|
||||
this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight);
|
||||
const task = this.processClaimedInboundWorkflow(item, applicationById.get(item.applicationId))
|
||||
.catch((error) => this.logger.error(`CMPP inbound workflow ${item.id} failed to settle: ${String(error)}`))
|
||||
const task = this.processClaimedInboundWorkflowBatch(batch, applicationById)
|
||||
.catch((error) => this.logger.error(`CMPP inbound workflow batch failed to settle: ${String(error)}`))
|
||||
.finally(() => {
|
||||
this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - 1);
|
||||
this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - batch.length);
|
||||
this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight);
|
||||
this.inboundWorkflowTasks.delete(task);
|
||||
this.scheduleInboundWorkflowPump(0);
|
||||
@@ -1294,6 +1310,333 @@ startInboundWorkflowWorker() {
|
||||
`);
|
||||
}
|
||||
|
||||
private async processClaimedInboundWorkflowBatch(
|
||||
items: ClaimedInboundWorkflow[],
|
||||
applicationById: Map<string, InboundApplication>,
|
||||
) {
|
||||
if (process.env.API_INBOUND_WORKFLOW_BATCH_ENABLED === 'false' || items.length < 2) {
|
||||
await Promise.all(items.map((item) => this.processClaimedInboundWorkflow(item, applicationById.get(item.applicationId))));
|
||||
return;
|
||||
}
|
||||
let completed = new Set<string>();
|
||||
try {
|
||||
completed = await this.processCommonInboundWorkflowBatch(items, applicationById);
|
||||
} catch (error) {
|
||||
// The common path is wholly idempotent: quota/frequency reservations have
|
||||
// stable keys, records have stable message/task keys, and BullMQ job IDs are
|
||||
// message IDs. Falling back after a partial external failure repairs rather
|
||||
// than duplicates the individual items.
|
||||
this.logger.warn(`CMPP inbound common batch fell back to individual recovery: ${String(error)}`);
|
||||
}
|
||||
await Promise.all(items
|
||||
.filter((item) => !completed.has(item.id))
|
||||
.map((item) => this.processClaimedInboundWorkflow(item, applicationById.get(item.applicationId))));
|
||||
}
|
||||
|
||||
private async processCommonInboundWorkflowBatch(
|
||||
items: ClaimedInboundWorkflow[],
|
||||
applicationById: Map<string, InboundApplication>,
|
||||
) {
|
||||
const parsed = items.flatMap((item) => {
|
||||
try {
|
||||
const payload = parseInboundWorkflowPayload(item.payload);
|
||||
const application = applicationById.get(item.applicationId);
|
||||
return payload.phoneNumbers.length === 1 && application ? [{ item, payload, application }] : [];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
});
|
||||
if (parsed.length < 2) return new Set<string>();
|
||||
const applicationIds = [...new Set(parsed.map((entry) => entry.application.id))];
|
||||
const phones = [...new Set(parsed.map((entry) => entry.payload.phoneNumbers[0]))];
|
||||
const messageIds = parsed.flatMap((entry) => entry.payload.messageIds);
|
||||
const tenantIds = [...new Set(parsed.map((entry) => entry.application.tenantId))];
|
||||
const [templates, signatures, globalBlacklist, enterpriseBlacklist, persisted] = await this.measureInboundStage('reference_preload', () => Promise.all([
|
||||
this.prisma.smsTemplate.findMany({
|
||||
where: { applicationId: { in: applicationIds }, auditStatus: 'approved', signature: { auditStatus: 'approved' } },
|
||||
include: { signature: true },
|
||||
orderBy: { updatedAt: 'desc' },
|
||||
}),
|
||||
this.prisma.smsSignature.findMany({
|
||||
where: { applicationId: { in: applicationIds }, auditStatus: 'approved' },
|
||||
select: { id: true, applicationId: true, name: true, updatedAt: true },
|
||||
orderBy: { updatedAt: 'desc' },
|
||||
}),
|
||||
this.prisma.globalBlacklist.findMany({
|
||||
where: { phoneNumber: { in: phones }, status: 'active' },
|
||||
select: { phoneNumber: true },
|
||||
}),
|
||||
this.prisma.enterpriseBlacklist.findMany({
|
||||
where: { tenantId: { in: tenantIds }, applicationId: { in: applicationIds }, phoneNumber: { in: phones }, status: 'active' },
|
||||
select: { tenantId: true, applicationId: true, phoneNumber: true },
|
||||
}),
|
||||
this.prisma.smsMessageRecord.findMany({ where: { messageId: { in: messageIds } }, select: { messageId: true } }),
|
||||
]));
|
||||
const globalRejected = new Set(globalBlacklist.map((entry) => entry.phoneNumber));
|
||||
const enterpriseRejected = new Set(enterpriseBlacklist.map((entry) => `${entry.tenantId}:${entry.applicationId}:${entry.phoneNumber}`));
|
||||
const persistedIds = new Set(persisted.map((entry) => entry.messageId));
|
||||
const templatesByApplication = new Map<string, InboundTemplate[]>();
|
||||
for (const template of templates) {
|
||||
const values = templatesByApplication.get(template.applicationId) ?? [];
|
||||
values.push(template);
|
||||
templatesByApplication.set(template.applicationId, values);
|
||||
}
|
||||
const signaturesByApplication = new Map<string, typeof signatures>();
|
||||
for (const signature of signatures) {
|
||||
if (!signature.applicationId) continue;
|
||||
const values = signaturesByApplication.get(signature.applicationId) ?? [];
|
||||
values.push(signature);
|
||||
signaturesByApplication.set(signature.applicationId, values);
|
||||
}
|
||||
const candidates: InboundBatchCandidate[] = [];
|
||||
for (const entry of parsed) {
|
||||
const { item, payload, application } = entry;
|
||||
const data = payload.data;
|
||||
const phoneNumber = payload.phoneNumbers[0];
|
||||
if (persistedIds.has(payload.messageIds[0])
|
||||
|| application.cmppAccount !== data.account
|
||||
|| application.status !== 'active'
|
||||
|| application.tenant.status !== 'active'
|
||||
|| !application.interfaceEnabled
|
||||
|| application.tenant.certificationStatus !== 'approved'
|
||||
|| !/^1\d{10}$/.test(phoneNumber)
|
||||
|| globalRejected.has(phoneNumber)
|
||||
|| enterpriseRejected.has(`${application.tenantId}:${application.id}:${phoneNumber}`)
|
||||
// Paid messages retain the existing per-message account lock until the
|
||||
// dedicated batch-ledger migration is introduced; never weaken billing
|
||||
// correctness merely to increase the benchmark number.
|
||||
|| moneyToNumber(application.customerUnitPrice) !== 0) continue;
|
||||
try {
|
||||
if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((allowlist) => allowlist.ipCidr))) continue;
|
||||
const clientSrcId = validateInboundApplicationSrcId(data.srcId, application);
|
||||
const template = (templatesByApplication.get(application.id) ?? []).find((candidate) => (
|
||||
candidate.content === data.content || matchTemplateContent(candidate.content, data.content) !== null
|
||||
));
|
||||
const templateVariables = template ? matchTemplateContent(template.content, data.content) ?? {} : undefined;
|
||||
let signatureId = template?.signature?.id;
|
||||
if (!signatureId && application.templateMismatchMode === 'direct_send') {
|
||||
const signatureName = data.content.match(/^【[^】]+】/)?.[0];
|
||||
signatureId = (signaturesByApplication.get(application.id) ?? []).find((signature) => signature.name === signatureName)?.id;
|
||||
}
|
||||
if (!signatureId || (!template && application.templateMismatchMode !== 'direct_send')) continue;
|
||||
const finalSignatureId = signatureId;
|
||||
candidates.push({ item, payload, application, phoneNumber, template, signatureId: finalSignatureId, templateVariables, clientSrcId });
|
||||
} catch {
|
||||
// Invalid source IDs and other business rejections stay on the established
|
||||
// individual path so their exact failure receipt remains unchanged.
|
||||
}
|
||||
}
|
||||
if (candidates.length < 2) return new Set<string>();
|
||||
|
||||
const quota = await this.measureInboundStage('daily_quota', () => this.reserveDailyQuotaBatch(candidates));
|
||||
const quotaApproved = candidates.filter((candidate) => quota.get(candidate.item.requestKey)?.reserved);
|
||||
if (quotaApproved.length < 2) return new Set<string>();
|
||||
const riskResults = await this.measureInboundStage('risk_frequency', () => this.riskReview.evaluateTasksBatch(quotaApproved.map((candidate) => ({
|
||||
tenantId: candidate.application.tenantId,
|
||||
applicationId: candidate.application.id,
|
||||
templateId: candidate.template?.id,
|
||||
content: candidate.payload.data.content,
|
||||
variables: candidate.templateVariables,
|
||||
phones: [candidate.phoneNumber],
|
||||
sourceType: 'cmpp' as const,
|
||||
}))));
|
||||
const riskApproved = quotaApproved.filter((_, index) => riskResults[index]?.status === 'approved');
|
||||
if (riskApproved.length < 2) return new Set<string>();
|
||||
const frequency = await this.measureInboundStage('risk_frequency', () => this.phoneFrequency.reserveBatch(riskApproved.map((candidate) => ({
|
||||
tenantId: candidate.application.tenantId,
|
||||
applicationId: candidate.application.id,
|
||||
phoneNumber: candidate.phoneNumber,
|
||||
sourceType: 'cmpp',
|
||||
reservationKey: `${candidate.item.requestKey}:message:0:frequency`,
|
||||
}))));
|
||||
const approved = riskApproved.filter((candidate) => (
|
||||
(frequency.get(`${candidate.item.requestKey}:message:0:frequency`)?.size ?? 0) === 0
|
||||
));
|
||||
if (approved.length < 2) return new Set<string>();
|
||||
|
||||
const drainageRows = await this.prisma.smsDrainageInfo.findMany({
|
||||
where: { signatureId: { in: [...new Set(approved.map((candidate) => candidate.signatureId))] }, auditStatus: { not: 'deleted' } },
|
||||
select: { id: true, signatureId: true, url: true, updatedAt: true },
|
||||
orderBy: [{ updatedAt: 'desc' }, { id: 'asc' }],
|
||||
});
|
||||
const finalCandidates = approved.filter((candidate) => {
|
||||
const matches = drainageRows
|
||||
.filter((row) => row.signatureId === candidate.signatureId && row.url.trim() && candidate.payload.data.content.includes(row.url.trim()))
|
||||
.sort((left, right) => right.url.trim().length - left.url.trim().length || right.updatedAt.getTime() - left.updatedAt.getTime());
|
||||
if (matches.length > 1 && matches[0].url.trim().length === matches[1].url.trim().length) return false;
|
||||
candidate.drainageInfoId = matches[0]?.id;
|
||||
return true;
|
||||
});
|
||||
if (finalCandidates.length < 2) return new Set<string>();
|
||||
return this.persistCommonInboundWorkflowBatch(finalCandidates);
|
||||
}
|
||||
|
||||
private async reserveDailyQuotaBatch(candidates: InboundBatchCandidate[]) {
|
||||
const usageDate = shanghaiDateKey();
|
||||
const usageDateValue = new Date(`${usageDate}T00:00:00.000Z`);
|
||||
return this.prisma.$transaction(async (tx) => {
|
||||
const applicationIds = [...new Set(candidates.map((candidate) => candidate.application.id))].sort();
|
||||
await tx.smsApplicationDailyUsage.createMany({
|
||||
data: applicationIds.map((applicationId) => ({ id: randomUUID(), applicationId, usageDate: usageDateValue, usedCount: 0 })),
|
||||
skipDuplicates: true,
|
||||
});
|
||||
const usage = await tx.$queryRaw<Array<{ applicationId: string; tenantId: string; dailyLimit: number; usedCount: number }>>(Prisma.sql`
|
||||
SELECT usage."applicationId", application."tenantId",
|
||||
COALESCE(application."dailyLimit", 100000)::integer AS "dailyLimit",
|
||||
usage."usedCount"
|
||||
FROM "SmsApplicationDailyUsage" usage
|
||||
JOIN "SmsApplication" application ON application.id = usage."applicationId"
|
||||
WHERE usage."usageDate" = ${usageDate}::date
|
||||
AND usage."applicationId" IN (${Prisma.join(applicationIds)})
|
||||
ORDER BY usage."applicationId"
|
||||
FOR UPDATE OF usage
|
||||
`);
|
||||
const state = new Map(usage.map((row) => [row.applicationId, { ...row }]));
|
||||
const keys = candidates.map((candidate) => `${candidate.item.requestKey}:daily-quota`);
|
||||
const existing = await tx.smsApplicationDailyReservation.findMany({ where: { reservationKey: { in: keys } } });
|
||||
const existingByKey = new Map(existing.map((row) => [row.reservationKey, row]));
|
||||
const output = new Map<string, { reserved: boolean; dailyLimit: number; usedCount: number | null }>();
|
||||
const inserts: Prisma.SmsApplicationDailyReservationCreateManyInput[] = [];
|
||||
for (const candidate of candidates) {
|
||||
const reservationKey = `${candidate.item.requestKey}:daily-quota`;
|
||||
const replay = existingByKey.get(reservationKey);
|
||||
if (replay) {
|
||||
if (replay.applicationId !== candidate.application.id || replay.requestedCount !== 1) {
|
||||
throw new Error('CMPP daily quota idempotency key conflicts with another reservation');
|
||||
}
|
||||
output.set(candidate.item.requestKey, { reserved: replay.reserved, dailyLimit: replay.dailyLimit, usedCount: replay.usedCount });
|
||||
continue;
|
||||
}
|
||||
const current = state.get(candidate.application.id);
|
||||
if (!current) throw new Error(`CMPP daily quota application ${candidate.application.id} disappeared`);
|
||||
const reserved = current.usedCount + 1 <= current.dailyLimit;
|
||||
if (reserved) current.usedCount += 1;
|
||||
inserts.push({
|
||||
id: randomUUID(), reservationKey, tenantId: candidate.application.tenantId,
|
||||
applicationId: candidate.application.id, usageDate: usageDateValue, requestedCount: 1,
|
||||
dailyLimit: current.dailyLimit, usedCount: reserved ? current.usedCount : null, reserved,
|
||||
});
|
||||
output.set(candidate.item.requestKey, { reserved, dailyLimit: current.dailyLimit, usedCount: reserved ? current.usedCount : null });
|
||||
}
|
||||
for (const row of state.values()) {
|
||||
await tx.smsApplicationDailyUsage.update({
|
||||
where: { applicationId_usageDate: { applicationId: row.applicationId, usageDate: usageDateValue } },
|
||||
data: { usedCount: row.usedCount },
|
||||
});
|
||||
}
|
||||
if (inserts.length) await tx.smsApplicationDailyReservation.createMany({ data: inserts });
|
||||
return output;
|
||||
}, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted });
|
||||
}
|
||||
|
||||
private async persistCommonInboundWorkflowBatch(candidates: InboundBatchCandidate[]) {
|
||||
const prepared = await Promise.all(candidates.map(async (candidate) => {
|
||||
const workflowDigest = createHash('sha256').update(`${candidate.item.requestKey}:message:0`).digest('hex').slice(0, 32);
|
||||
const taskId = randomUUID();
|
||||
const messageRecordId = randomUUID();
|
||||
const content = candidate.payload.data.content;
|
||||
const drainageDetection = await detectDrainageContent(this.prisma, content);
|
||||
return { candidate, workflowDigest, taskId, messageRecordId, content, drainageDetection };
|
||||
}));
|
||||
await this.measureInboundStage('message_persist', () => this.prisma.$transaction(async (tx) => {
|
||||
await tx.smsBatchTask.createMany({
|
||||
data: prepared.map(({ candidate, workflowDigest, taskId, content }) => ({
|
||||
id: taskId,
|
||||
tenantId: candidate.application.tenantId,
|
||||
applicationId: candidate.application.id,
|
||||
templateId: candidate.template?.id,
|
||||
taskNo: `BT-IN-${workflowDigest}`,
|
||||
sourceType: 'cmpp',
|
||||
content,
|
||||
phoneTotal: 1,
|
||||
status: 'ready',
|
||||
auditStatus: 'approved',
|
||||
progressTotal: 1,
|
||||
})),
|
||||
});
|
||||
await tx.smsApiRequest.createMany({
|
||||
data: prepared.map(({ candidate, workflowDigest, taskId, content }) => ({
|
||||
id: randomUUID(), tenantId: candidate.application.tenantId, batchTaskId: taskId,
|
||||
requestId: `REQ-IN-${workflowDigest}`, sourceIp: candidate.payload.data.remoteIp,
|
||||
userAgent: 'cmpp-gateway', payloadSummary: { phoneTotal: 1, contentLength: [...content].length, account: candidate.payload.data.account },
|
||||
status: 'accepted',
|
||||
})),
|
||||
});
|
||||
await tx.smsMessageRecord.createMany({
|
||||
data: prepared.map(({ candidate, taskId, messageRecordId, content, drainageDetection }) => ({
|
||||
id: messageRecordId,
|
||||
tenantId: candidate.application.tenantId,
|
||||
batchTaskId: taskId,
|
||||
applicationId: candidate.application.id,
|
||||
templateId: candidate.template?.id,
|
||||
signatureId: candidate.signatureId,
|
||||
drainageInfoId: candidate.drainageInfoId,
|
||||
messageId: candidate.payload.messageIds[0],
|
||||
phoneNumber: candidate.phoneNumber,
|
||||
content,
|
||||
...drainageDetection,
|
||||
billingUnits: this.billing.estimateSmsCost({
|
||||
tenantId: candidate.application.tenantId, applicationId: candidate.application.id,
|
||||
content, phoneCount: 1, unitPrice: 0,
|
||||
}).billingUnitsPerMessage,
|
||||
unitPrice: 0,
|
||||
amountCents: 0,
|
||||
queuePriority: normalizeQueuePriority(candidate.application.queuePriority),
|
||||
cmppSubmitSequenceId: candidate.payload.data.sequenceId == null ? null : String(candidate.payload.data.sequenceId),
|
||||
cmppSubmitGroupMessageId: candidate.payload.submitGroupMessageId,
|
||||
cmppRegisteredDelivery: candidate.payload.data.registeredDelivery !== 0,
|
||||
clientSrcId: candidate.clientSrcId,
|
||||
applicationExtension: candidate.application.cmppApplicationExtension,
|
||||
status: 'queued',
|
||||
})),
|
||||
});
|
||||
}));
|
||||
await this.measureInboundStage('queue_publish', () => this.facade.getSendQueue().addBulk(prepared.map(({ candidate, messageRecordId }) => ({
|
||||
name: 'send-message' as const,
|
||||
data: { messageRecordId },
|
||||
opts: {
|
||||
jobId: messageRecordId,
|
||||
attempts: 3,
|
||||
priority: BULLMQ_PRIORITY[normalizeQueuePriority(candidate.application.queuePriority)],
|
||||
},
|
||||
}))));
|
||||
await this.prisma.smsBatchTask.updateMany({
|
||||
where: { id: { in: prepared.map((entry) => entry.taskId) }, status: 'ready' },
|
||||
data: { status: 'queued' },
|
||||
});
|
||||
const results = prepared.map(({ candidate, taskId, messageRecordId }) => ({
|
||||
accepted: true,
|
||||
tenantId: candidate.application.tenantId,
|
||||
applicationId: candidate.application.id,
|
||||
taskId,
|
||||
messageId: candidate.payload.messageIds[0],
|
||||
messageRecordId,
|
||||
status: 'accepted',
|
||||
phoneCount: 1,
|
||||
messages: [{
|
||||
phoneNumber: candidate.phoneNumber,
|
||||
messageId: candidate.payload.messageIds[0],
|
||||
messageRecordId,
|
||||
taskId,
|
||||
status: 'accepted',
|
||||
}],
|
||||
}));
|
||||
const values = prepared.map((entry, index) => Prisma.sql`(${entry.candidate.item.id}, ${JSON.stringify(results[index])}::jsonb)`);
|
||||
const settled = await this.prisma.$queryRaw<Array<{ id: string }>>(Prisma.sql`
|
||||
UPDATE "CmppInboundSubmissionInbox" inbox
|
||||
SET status = 'completed', result = updates.result, "completedAt" = (NOW() AT TIME ZONE 'UTC'),
|
||||
"lockedAt" = NULL, "lockedBy" = NULL, "lastError" = NULL, "updatedAt" = (NOW() AT TIME ZONE 'UTC')
|
||||
FROM (VALUES ${Prisma.join(values)}) AS updates(id, result)
|
||||
WHERE inbox.id = updates.id
|
||||
AND inbox.status = 'processing'
|
||||
AND inbox."lockedBy" = ${this.inboundWorkflowWorkerId}
|
||||
RETURNING inbox.id
|
||||
`);
|
||||
const settledIds = new Set(settled.map((row) => row.id));
|
||||
for (let index = 0; index < settled.length; index += 1) this.metrics?.recordInboundWorkflowResult('completed');
|
||||
return settledIds;
|
||||
}
|
||||
|
||||
private async processClaimedInboundWorkflow(item: ClaimedInboundWorkflow, application?: InboundApplication) {
|
||||
try {
|
||||
const payload = parseInboundWorkflowPayload(item.payload);
|
||||
|
||||
Reference in New Issue
Block a user