feat: add sms send chain phase
This commit is contained in:
hectorzhao
@@ -0,0 +1,542 @@
|
||||
import { Injectable, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
||||
import { Queue, Worker } from 'bullmq';
|
||||
import IORedis from 'ioredis';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { setTimeout as sleep } from 'node:timers/promises';
|
||||
import { BillingService } from '../billing/billing.service';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { RiskReviewService } from '../risk-review/risk-review.service';
|
||||
|
||||
export interface CreateBatchTaskDto {
|
||||
tenantId: string;
|
||||
applicationId?: string;
|
||||
templateId?: string;
|
||||
content: string;
|
||||
category?: string;
|
||||
phones: string[];
|
||||
variables?: Record<string, unknown>;
|
||||
createdById?: string;
|
||||
sourceIp?: string;
|
||||
userAgent?: string;
|
||||
}
|
||||
|
||||
export interface GatewaySubmitResultDto {
|
||||
traceId?: string;
|
||||
messageId: string;
|
||||
channelId: string;
|
||||
submitId?: string;
|
||||
sequenceId?: number;
|
||||
gatewayMessageId: string;
|
||||
submitStatus: 'accepted' | 'rejected' | 'timeout';
|
||||
errorCode?: string;
|
||||
errorMessage?: string;
|
||||
submittedAt?: string;
|
||||
}
|
||||
|
||||
export interface GatewayReceiptEventDto {
|
||||
traceId?: string;
|
||||
messageId: string;
|
||||
channelId: string;
|
||||
sequenceId?: number;
|
||||
gatewayMessageId: string;
|
||||
receiptStatus: 'delivered' | 'undelivered' | 'unknown';
|
||||
rawStatus: string;
|
||||
errorCode?: string;
|
||||
deliveredAt?: string;
|
||||
}
|
||||
|
||||
export interface GatewayUplinkEventDto {
|
||||
traceId?: string;
|
||||
messageId?: string;
|
||||
channelId: string;
|
||||
sequenceId?: number;
|
||||
phoneNumber: string;
|
||||
destId: string;
|
||||
content: string;
|
||||
receivedAt?: string;
|
||||
}
|
||||
|
||||
export interface TimeoutUnknownDto {
|
||||
olderThanHours?: number;
|
||||
}
|
||||
|
||||
interface SendJob {
|
||||
messageRecordId: string;
|
||||
}
|
||||
|
||||
const SEND_QUEUE = 'sms.send.queue';
|
||||
const GATEWAY_SUBMIT_QUEUE = 'gateway.submit.queue';
|
||||
|
||||
@Injectable()
|
||||
export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
private redis?: IORedis;
|
||||
private sendQueue?: Queue<SendJob, unknown, 'send-message'>;
|
||||
private gatewayQueue?: Queue;
|
||||
private worker?: Worker<SendJob>;
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly billing: BillingService,
|
||||
private readonly riskReview: RiskReviewService,
|
||||
) {}
|
||||
|
||||
onModuleInit() {
|
||||
if (process.env.API_ENABLE_SEND_WORKER === 'true') {
|
||||
this.startWorker();
|
||||
}
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
await this.worker?.close();
|
||||
await this.sendQueue?.close();
|
||||
await this.gatewayQueue?.close();
|
||||
this.redis?.disconnect();
|
||||
}
|
||||
|
||||
async createBatchTask(data: CreateBatchTaskDto) {
|
||||
const phones = [...new Set(data.phones ?? [])];
|
||||
const risk = await this.riskReview.evaluateTask({
|
||||
tenantId: data.tenantId,
|
||||
applicationId: data.applicationId,
|
||||
templateId: data.templateId,
|
||||
content: data.content,
|
||||
category: data.category,
|
||||
phones,
|
||||
variables: data.variables,
|
||||
createdById: data.createdById,
|
||||
});
|
||||
const billing = this.billing.estimateSmsCost({
|
||||
tenantId: data.tenantId,
|
||||
applicationId: data.applicationId,
|
||||
taskId: risk.task?.id,
|
||||
content: data.content,
|
||||
phoneCount: phones.length,
|
||||
});
|
||||
const batchStatus = statusFromRisk(risk.status);
|
||||
const task = await this.prisma.smsBatchTask.create({
|
||||
data: {
|
||||
tenantId: data.tenantId,
|
||||
applicationId: data.applicationId,
|
||||
templateId: data.templateId,
|
||||
taskNo: `BT-${Date.now()}-${randomUUID().slice(0, 8)}`,
|
||||
sourceType: 'client',
|
||||
content: data.content,
|
||||
category: data.category,
|
||||
phoneTotal: phones.length,
|
||||
status: batchStatus,
|
||||
riskTaskId: risk.task?.id,
|
||||
auditStatus: risk.status === 'pending_review' ? 'pending' : risk.status === 'rejected' ? 'rejected' : 'approved',
|
||||
reviewReason: risk.status === 'pending_review' ? risk.reason : null,
|
||||
rejectReason: risk.status === 'rejected' ? risk.reason : null,
|
||||
progressTotal: phones.length,
|
||||
createdById: data.createdById,
|
||||
},
|
||||
});
|
||||
await this.prisma.smsApiRequest.create({
|
||||
data: {
|
||||
tenantId: data.tenantId,
|
||||
batchTaskId: task.id,
|
||||
requestId: `REQ-${Date.now()}-${randomUUID().slice(0, 8)}`,
|
||||
sourceIp: data.sourceIp,
|
||||
userAgent: data.userAgent,
|
||||
payloadSummary: {
|
||||
phoneTotal: phones.length,
|
||||
contentLength: [...data.content].length,
|
||||
category: data.category,
|
||||
},
|
||||
status: batchStatus === 'rejected' ? 'rejected' : 'accepted',
|
||||
},
|
||||
});
|
||||
if (phones.length > 0) {
|
||||
await this.prisma.smsMessageRecord.createMany({
|
||||
data: phones.map((phone) => ({
|
||||
tenantId: data.tenantId,
|
||||
batchTaskId: task.id,
|
||||
applicationId: data.applicationId,
|
||||
templateId: data.templateId,
|
||||
messageId: `MSG-${randomUUID()}`,
|
||||
phoneNumber: phone,
|
||||
content: data.content,
|
||||
billingUnits: billing.billingUnitsPerMessage,
|
||||
unitPrice: billing.unitPrice,
|
||||
amountCents: billing.billingUnitsPerMessage * billing.unitPrice,
|
||||
status: batchStatus === 'ready' ? 'queued' : batchStatus,
|
||||
errorMessage: risk.status === 'rejected' ? risk.reason ?? undefined : undefined,
|
||||
})),
|
||||
});
|
||||
}
|
||||
if (batchStatus === 'ready') {
|
||||
await this.enqueueBatchTask(task.id);
|
||||
}
|
||||
return this.getBatchTask(task.id);
|
||||
}
|
||||
|
||||
listBatchTasks(tenantId?: string, status?: string) {
|
||||
return this.prisma.smsBatchTask.findMany({
|
||||
where: { tenantId, status },
|
||||
include: { apiRequests: true },
|
||||
orderBy: { createdAt: 'desc' },
|
||||
take: 100,
|
||||
});
|
||||
}
|
||||
|
||||
getBatchTask(taskId: string) {
|
||||
return this.prisma.smsBatchTask.findUnique({
|
||||
where: { id: taskId },
|
||||
include: { apiRequests: true, messages: { take: 20, orderBy: { queuedAt: 'asc' } } },
|
||||
});
|
||||
}
|
||||
|
||||
listMessages(taskId?: string, phoneNumber?: string) {
|
||||
return this.prisma.smsMessageRecord.findMany({
|
||||
where: { batchTaskId: taskId, phoneNumber },
|
||||
orderBy: { queuedAt: 'desc' },
|
||||
take: 200,
|
||||
});
|
||||
}
|
||||
|
||||
listSubmitRecords(taskId?: string) {
|
||||
return this.prisma.smsSubmitRecord.findMany({
|
||||
where: { batchTaskId: taskId },
|
||||
orderBy: { createdAt: 'desc' },
|
||||
take: 200,
|
||||
});
|
||||
}
|
||||
|
||||
listReceiptRecords(taskId?: string) {
|
||||
return this.prisma.smsReceiptRecord.findMany({
|
||||
where: { batchTaskId: taskId },
|
||||
orderBy: { createdAt: 'desc' },
|
||||
take: 200,
|
||||
});
|
||||
}
|
||||
|
||||
listUplinkMessages(tenantId?: string, channelId?: string) {
|
||||
return this.prisma.smsUplinkMessage.findMany({
|
||||
where: { tenantId, channelId },
|
||||
orderBy: { receivedAt: 'desc' },
|
||||
take: 200,
|
||||
});
|
||||
}
|
||||
|
||||
async enqueueBatchTask(taskId: string) {
|
||||
const task = await this.prisma.smsBatchTask.findUnique({ where: { id: taskId } });
|
||||
if (!task) {
|
||||
throw new NotFoundException('SMS batch task not found');
|
||||
}
|
||||
const messages = await this.prisma.smsMessageRecord.findMany({
|
||||
where: { batchTaskId: taskId, status: 'queued' },
|
||||
select: { id: true },
|
||||
take: 100000,
|
||||
});
|
||||
const queue = this.getSendQueue();
|
||||
for (const message of messages) {
|
||||
await queue.add('send-message', { messageRecordId: message.id }, { jobId: message.id, attempts: 3 });
|
||||
}
|
||||
await this.prisma.smsBatchTask.update({ where: { id: taskId }, data: { status: 'queued' } });
|
||||
return { taskId, enqueued: messages.length };
|
||||
}
|
||||
|
||||
startWorker() {
|
||||
if (this.worker) {
|
||||
return { status: 'already_started' };
|
||||
}
|
||||
const connection = bullmqConnection();
|
||||
this.worker = new Worker<SendJob>(
|
||||
SEND_QUEUE,
|
||||
async (job) => this.processSendJob(job.data),
|
||||
{ connection, concurrency: Number(process.env.API_SEND_WORKER_CONCURRENCY ?? 20) },
|
||||
);
|
||||
return { status: 'started' };
|
||||
}
|
||||
|
||||
async processSendJob(job: SendJob) {
|
||||
const message = await this.prisma.smsMessageRecord.findUnique({
|
||||
where: { id: job.messageRecordId },
|
||||
include: { batchTask: true, template: { include: { signature: true } } },
|
||||
});
|
||||
if (!message || message.status !== 'queued') {
|
||||
return { skipped: true };
|
||||
}
|
||||
const channel = await this.selectChannel(message.tenantId, message.applicationId ?? undefined);
|
||||
await this.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond);
|
||||
const submitId = `SUB-${randomUUID()}`;
|
||||
const session = await this.prisma.cmppSubmitSession.upsert({
|
||||
where: { sessionNo: `OPEN-${channel.id}` },
|
||||
update: { submitTotal: { increment: 1 } },
|
||||
create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 },
|
||||
});
|
||||
await this.prisma.smsSubmitRecord.create({
|
||||
data: {
|
||||
tenantId: message.tenantId,
|
||||
batchTaskId: message.batchTaskId,
|
||||
messageRecordId: message.id,
|
||||
channelId: channel.id,
|
||||
sessionId: session.id,
|
||||
submitId,
|
||||
submitStatus: 'queued',
|
||||
},
|
||||
});
|
||||
await this.prisma.smsMessageRecord.update({
|
||||
where: { id: message.id },
|
||||
data: { channelId: channel.id, submitId, status: 'submit_queued', submitStatus: 'queued' },
|
||||
});
|
||||
await this.getGatewayQueue().add('submit-command', {
|
||||
schemaVersion: 'v1',
|
||||
messageType: 'SubmitCommand',
|
||||
traceId: randomUUID(),
|
||||
messageId: message.messageId,
|
||||
channelId: channel.id,
|
||||
createdAt: new Date().toISOString(),
|
||||
tenantId: message.tenantId,
|
||||
applicationId: message.applicationId ?? 'unknown',
|
||||
taskId: message.batchTaskId,
|
||||
submitId,
|
||||
phoneNumber: message.phoneNumber,
|
||||
content: message.content,
|
||||
signature: message.template?.signature?.name ?? 'SMS',
|
||||
templateId: message.templateId ?? 'unknown',
|
||||
billingUnits: message.billingUnits,
|
||||
route: {
|
||||
channelCode: channel.code,
|
||||
cmppAccountCode: channel.account,
|
||||
priority: 0,
|
||||
rateLimitPerSecond: channel.rateLimitPerSecond,
|
||||
},
|
||||
cmpp: {
|
||||
serviceId: channel.config && typeof channel.config === 'object' && 'serviceId' in channel.config
|
||||
? String(channel.config.serviceId)
|
||||
: 'SMS',
|
||||
srcId: channel.srcId,
|
||||
registeredDelivery: 1,
|
||||
msgFmt: 8,
|
||||
},
|
||||
retry: { attempt: 0, maxAttempts: 3 },
|
||||
});
|
||||
await this.refreshTaskProgress(message.batchTaskId);
|
||||
return { submitted: true, messageRecordId: message.id, channelId: channel.id };
|
||||
}
|
||||
|
||||
async handleSubmitResult(data: GatewaySubmitResultDto) {
|
||||
const message = await this.findMessageByGatewayEvent(data.messageId, data.gatewayMessageId);
|
||||
const submittedAt = data.submittedAt ? new Date(data.submittedAt) : new Date();
|
||||
await this.prisma.smsSubmitRecord.updateMany({
|
||||
where: { OR: [{ submitId: data.submitId }, { messageRecordId: message.id }] },
|
||||
data: {
|
||||
sequenceId: data.sequenceId,
|
||||
gatewayMessageId: data.gatewayMessageId,
|
||||
submitStatus: data.submitStatus,
|
||||
errorCode: data.errorCode,
|
||||
errorMessage: data.errorMessage,
|
||||
submittedAt,
|
||||
},
|
||||
});
|
||||
const status = data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed';
|
||||
await this.prisma.smsMessageRecord.update({
|
||||
where: { id: message.id },
|
||||
data: {
|
||||
gatewayMessageId: data.gatewayMessageId,
|
||||
submitStatus: data.submitStatus,
|
||||
status,
|
||||
errorCode: data.errorCode,
|
||||
errorMessage: data.errorMessage,
|
||||
submittedAt,
|
||||
timeoutAt: data.submitStatus === 'timeout' ? submittedAt : undefined,
|
||||
},
|
||||
});
|
||||
await this.refreshTaskProgress(message.batchTaskId);
|
||||
return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
|
||||
}
|
||||
|
||||
async handleReceipt(data: GatewayReceiptEventDto) {
|
||||
const message = await this.findMessageByGatewayEvent(data.messageId, data.gatewayMessageId);
|
||||
const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date();
|
||||
const status =
|
||||
data.receiptStatus === 'delivered' ? 'delivered' : data.receiptStatus === 'unknown' ? 'unknown' : 'failed';
|
||||
await this.prisma.smsReceiptRecord.create({
|
||||
data: {
|
||||
tenantId: message.tenantId,
|
||||
batchTaskId: message.batchTaskId,
|
||||
messageRecordId: message.id,
|
||||
channelId: data.channelId,
|
||||
messageId: data.messageId,
|
||||
gatewayMessageId: data.gatewayMessageId,
|
||||
sequenceId: data.sequenceId,
|
||||
receiptStatus: data.receiptStatus,
|
||||
rawStatus: data.rawStatus,
|
||||
errorCode: data.errorCode,
|
||||
deliveredAt,
|
||||
},
|
||||
});
|
||||
await this.prisma.smsMessageRecord.update({
|
||||
where: { id: message.id },
|
||||
data: {
|
||||
receiptStatus: data.receiptStatus,
|
||||
status,
|
||||
errorCode: data.errorCode,
|
||||
deliveredAt,
|
||||
},
|
||||
});
|
||||
await this.refreshTaskProgress(message.batchTaskId);
|
||||
return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
|
||||
}
|
||||
|
||||
async handleUplink(data: GatewayUplinkEventDto) {
|
||||
const channel = await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } });
|
||||
if (!channel) {
|
||||
throw new NotFoundException('SMS channel not found');
|
||||
}
|
||||
const message = data.messageId
|
||||
? await this.prisma.smsMessageRecord.findUnique({ where: { messageId: data.messageId } })
|
||||
: null;
|
||||
return this.prisma.smsUplinkMessage.create({
|
||||
data: {
|
||||
tenantId: message?.tenantId,
|
||||
channelId: data.channelId,
|
||||
messageId: data.messageId,
|
||||
sequenceId: data.sequenceId,
|
||||
phoneNumber: data.phoneNumber,
|
||||
destId: data.destId,
|
||||
content: data.content,
|
||||
receivedAt: data.receivedAt ? new Date(data.receivedAt) : new Date(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async markUnknownTimeout(data: TimeoutUnknownDto) {
|
||||
const olderThanHours = data.olderThanHours ?? 72;
|
||||
const cutoff = new Date(Date.now() - olderThanHours * 60 * 60 * 1000);
|
||||
const candidates = await this.prisma.smsMessageRecord.findMany({
|
||||
where: {
|
||||
status: 'unknown',
|
||||
deliveredAt: { lte: cutoff },
|
||||
},
|
||||
select: { id: true, batchTaskId: true },
|
||||
take: 10000,
|
||||
});
|
||||
await this.prisma.smsMessageRecord.updateMany({
|
||||
where: { id: { in: candidates.map((candidate) => candidate.id) } },
|
||||
data: { status: 'timeout', timeoutAt: new Date(), errorMessage: '72小时未收到明确回执，自动转超时' },
|
||||
});
|
||||
for (const batchTaskId of new Set(candidates.map((candidate) => candidate.batchTaskId))) {
|
||||
await this.refreshTaskProgress(batchTaskId);
|
||||
}
|
||||
return { timeout: candidates.length };
|
||||
}
|
||||
|
||||
private async selectChannel(tenantId: string, applicationId?: string) {
|
||||
const route = await this.prisma.channelRouteRule.findFirst({
|
||||
where: {
|
||||
status: 'active',
|
||||
OR: [{ tenantId, applicationId }, { tenantId, applicationId: null }, { tenantId: null, applicationId: null }],
|
||||
},
|
||||
include: { channel: true, group: { include: { items: { include: { channel: true }, orderBy: { priority: 'asc' } } } } },
|
||||
orderBy: { priority: 'asc' },
|
||||
});
|
||||
const routedChannel = route?.channel ?? route?.group.items.find((item) => item.channel.status === 'active')?.channel;
|
||||
if (routedChannel) {
|
||||
return routedChannel;
|
||||
}
|
||||
const channel = await this.prisma.smsChannel.findFirst({ where: { status: 'active' }, orderBy: { createdAt: 'asc' } });
|
||||
if (!channel) {
|
||||
throw new NotFoundException('No active SMS channel available');
|
||||
}
|
||||
return channel;
|
||||
}
|
||||
|
||||
private async waitForChannelRateLimit(channelId: string, tps: number) {
|
||||
const redis = this.getRedis();
|
||||
for (;;) {
|
||||
const bucket = `rate:channel:${channelId}:${Math.floor(Date.now() / 1000)}`;
|
||||
const count = await redis.incr(bucket);
|
||||
if (count === 1) {
|
||||
await redis.expire(bucket, 2);
|
||||
}
|
||||
if (count <= Math.max(1, tps)) {
|
||||
return;
|
||||
}
|
||||
await sleep(100);
|
||||
}
|
||||
}
|
||||
|
||||
private async refreshTaskProgress(batchTaskId: string) {
|
||||
const groups = await this.prisma.smsMessageRecord.groupBy({
|
||||
by: ['status'],
|
||||
where: { batchTaskId },
|
||||
_count: { _all: true },
|
||||
});
|
||||
const count = (statuses: string[]) =>
|
||||
groups.filter((group) => statuses.includes(group.status)).reduce((sum, group) => sum + group._count._all, 0);
|
||||
const progressTotal = groups.reduce((sum, group) => sum + group._count._all, 0);
|
||||
const submittedTotal = count(['submit_queued', 'submitted', 'delivered', 'failed', 'unknown', 'timeout']);
|
||||
const successTotal = count(['delivered']);
|
||||
const failedTotal = count(['submit_failed', 'failed']);
|
||||
const unknownTotal = count(['unknown']);
|
||||
const timeoutTotal = count(['timeout']);
|
||||
const doneTotal = successTotal + failedTotal + timeoutTotal;
|
||||
const status = progressTotal > 0 && doneTotal >= progressTotal ? 'finished' : submittedTotal > 0 ? 'sending' : 'queued';
|
||||
await this.prisma.smsBatchTask.update({
|
||||
where: { id: batchTaskId },
|
||||
data: { progressTotal, submittedTotal, successTotal, failedTotal, unknownTotal, timeoutTotal, status },
|
||||
});
|
||||
}
|
||||
|
||||
private async findMessageByGatewayEvent(messageId: string, gatewayMessageId?: string) {
|
||||
const message = await this.prisma.smsMessageRecord.findFirst({
|
||||
where: {
|
||||
OR: [{ messageId }, gatewayMessageId ? { gatewayMessageId } : undefined].filter(Boolean) as Array<{
|
||||
messageId?: string;
|
||||
gatewayMessageId?: string;
|
||||
}>,
|
||||
},
|
||||
});
|
||||
if (!message) {
|
||||
throw new NotFoundException('SMS message record not found');
|
||||
}
|
||||
return message;
|
||||
}
|
||||
|
||||
private getSendQueue(): Queue<SendJob, unknown, 'send-message'> {
|
||||
if (!this.sendQueue) {
|
||||
this.sendQueue = new Queue<SendJob, unknown, 'send-message'>(SEND_QUEUE, { connection: bullmqConnection() });
|
||||
}
|
||||
return this.sendQueue;
|
||||
}
|
||||
|
||||
private getGatewayQueue(): Queue {
|
||||
if (!this.gatewayQueue) {
|
||||
this.gatewayQueue = new Queue(GATEWAY_SUBMIT_QUEUE, { connection: bullmqConnection() });
|
||||
}
|
||||
return this.gatewayQueue;
|
||||
}
|
||||
|
||||
private getRedis() {
|
||||
if (!this.redis) {
|
||||
this.redis = new IORedis(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', {
|
||||
maxRetriesPerRequest: null,
|
||||
});
|
||||
}
|
||||
return this.redis;
|
||||
}
|
||||
}
|
||||
|
||||
function statusFromRisk(status: string) {
|
||||
if (status === 'rejected') {
|
||||
return 'rejected';
|
||||
}
|
||||
if (status === 'pending_review') {
|
||||
return 'pending_review';
|
||||
}
|
||||
return 'ready';
|
||||
}
|
||||
|
||||
function bullmqConnection() {
|
||||
const redisUrl = new URL(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379');
|
||||
return {
|
||||
host: redisUrl.hostname,
|
||||
port: Number(redisUrl.port || 6379),
|
||||
username: redisUrl.username || undefined,
|
||||
password: redisUrl.password || undefined,
|
||||
maxRetriesPerRequest: null,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user