hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: close sms scheduling and billing gaps

Browse Source
This commit is contained in:
hectorzhao
2026-07-01 18:56:05 +08:00
parent 8ba4ef8a13
commit f8c9b78c21
28 changed files with 1480 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/admin-send-chain.controller.ts
+5 -1
View File
@@ -37,9 +37,13 @@ export class AdminSendChainController {
return this.sendChain.enqueueBatchTask(taskId);
}
@Post('scheduled/dispatch-due')
dispatchDueScheduledTasks() {
return this.sendChain.dispatchDueScheduledTasks();
}
@Post('timeouts/mark-unknown')
markUnknownTimeout(@Body() body: TimeoutUnknownDto) {
return this.sendChain.markUnknownTimeout(body);
}
}
api/src/send-chain/client-send-chain.controller.ts
+16 -2
View File
@@ -1,7 +1,7 @@
import { Body, Controller, Get, Param, Post, Query } from '@nestjs/common';
import { ApiTags } from '@nestjs/swagger';
import { TenantId } from '../common/tenant-id.decorator';
import { CreateBatchTaskDto, SendChainService } from './send-chain.service';
import { ConfirmImportDto, CreateBatchTaskDto, ImportPreviewDto, SendChainService } from './send-chain.service';
@ApiTags('client-send-chain')
@Controller('client/send')
@@ -13,6 +13,16 @@ export class ClientSendChainController {
return this.sendChain.createBatchTask(body);
}
@Post('imports/preview')
previewImport(@Body() body: ImportPreviewDto) {
return this.sendChain.previewImport(body);
}
@Post('imports/confirm')
confirmImport(@Body() body: ConfirmImportDto) {
return this.sendChain.confirmImport(body);
}
@Get('batch-tasks')
listBatchTasks(@TenantId() tenantId?: string, @Query('status') status?: string) {
return this.sendChain.listBatchTasks(tenantId, status);
@@ -27,5 +37,9 @@ export class ClientSendChainController {
listTaskMessages(@Param('id') taskId: string) {
return this.sendChain.listMessages(taskId);
}
}
@Post('batch-tasks/:id/cancel')
cancelBatchTask(@Param('id') taskId: string) {
return this.sendChain.cancelBatchTask(taskId);
}
}
api/src/send-chain/send-chain.service.spec.ts
+161 -3
View File
@@ -14,6 +14,8 @@ function createPrismaMock() {
phoneNumber: '13800000001',
content: 'hello',
billingUnits: 1,
unitPrice: 3,
amountCents: 3,
status: 'queued',
template: { signature: { name: '签名' } },
};
@@ -23,10 +25,26 @@ function createPrismaMock() {
account: 'cmpp-account',
srcId: '10690000',
rateLimitPerSecond: 100,
unitPrice: 3,
status: 'active',
config: { serviceId: 'SMS' },
};
return {
tenant: {
findUnique: jest.fn().mockResolvedValue({ id: 'tenant-1', status: 'active', certificationStatus: 'approved' }),
},
smsApplication: {
findUnique: jest.fn().mockResolvedValue({ id: 'app-1', tenantId: 'tenant-1', status: 'active' }),
},
smsTemplate: {
findUnique: jest.fn().mockResolvedValue({
id: 'tpl-1',
tenantId: 'tenant-1',
applicationId: 'app-1',
auditStatus: 'approved',
signature: { auditStatus: 'approved', reportStatus: 'approved' },
}),
},
smsBatchTask: {
create: jest.fn().mockResolvedValue(task),
findUnique: jest.fn().mockResolvedValue(task),
@@ -69,6 +87,18 @@ function createPrismaMock() {
create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'uplink-1', ...data })),
findMany: jest.fn(),
},
smsBillingRecord: {
findFirst: jest.fn().mockResolvedValue(null),
create: jest.fn().mockResolvedValue({ id: 'bill-1' }),
update: jest.fn().mockResolvedValue({ id: 'bill-1' }),
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
},
enterpriseBlacklist: {
findMany: jest.fn().mockResolvedValue([]),
},
globalBlacklist: {
findMany: jest.fn().mockResolvedValue([]),
},
};
}
@@ -76,9 +106,15 @@ function createService(prisma = createPrismaMock()) {
const billing = {
estimateSmsCost: jest.fn().mockReturnValue({
billingUnitsPerMessage: 1,
totalBillingUnits: 2,
unitPrice: 3,
amountCents: 6,
}),
checkAccount: jest.fn().mockResolvedValue({ canSend: true }),
freeze: jest.fn().mockResolvedValue({ id: 'tx-freeze' }),
release: jest.fn().mockResolvedValue({ id: 'tx-release' }),
charge: jest.fn().mockResolvedValue({ id: 'tx-charge' }),
refund: jest.fn().mockResolvedValue({ id: 'tx-refund' }),
} as unknown as BillingService;
const riskReview = {
evaluateTask: jest.fn().mockResolvedValue({
@@ -92,7 +128,7 @@ function createService(prisma = createPrismaMock()) {
describe('SendChainService', () => {
it('creates batch tasks, deduplicates phones, creates message records, and enqueues approved tasks', async () => {
const { service, prisma, riskReview } = createService();
const { service, prisma, riskReview, billing } = createService();
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 2 });
await service.createBatchTask({
@@ -115,9 +151,105 @@ describe('SendChainService', () => {
expect.objectContaining({ phoneNumber: '13800000002', status: 'queued', billingUnits: 1, amountCents: 3 }),
]),
});
expect(billing.freeze).toHaveBeenCalledWith(expect.objectContaining({ amountCents: 6, smsUnits: 2, relatedId: 'task-1' }));
expect(service.enqueueBatchTask).toHaveBeenCalledWith('task-1');
});
it('creates scheduled tasks without immediate enqueue and dispatches due tasks later', async () => {
const { service, prisma, billing } = createService();
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
const scheduledAt = new Date(Date.now() + 60_000).toISOString();
await service.createBatchTask({
tenantId: 'tenant-1',
applicationId: 'app-1',
templateId: 'tpl-1',
content: 'hello',
phones: ['13800000001'],
sendMode: 'scheduled',
scheduledAt,
});
expect(prisma.smsBatchTask.create).toHaveBeenCalledWith({
data: expect.objectContaining({ status: 'scheduled', scheduledAt: expect.any(Date) }),
});
expect(prisma.smsMessageRecord.createMany).toHaveBeenCalledWith({
data: [expect.objectContaining({ status: 'scheduled' })],
});
expect(billing.freeze).not.toHaveBeenCalled();
expect(service.enqueueBatchTask).not.toHaveBeenCalled();
prisma.smsBatchTask.findMany.mockResolvedValue([{ id: 'task-1', tenantId: 'tenant-1', applicationId: 'app-1', templateId: 'tpl-1' }]);
prisma.smsMessageRecord.findMany.mockResolvedValue([{ id: 'record-1', amountCents: 3, billingUnits: 1 }]);
await expect(service.dispatchDueScheduledTasks(new Date(Date.now() + 120_000))).resolves.toEqual({
dispatched: 1,
results: [{ taskId: 'task-1', status: 'queued', enqueued: 1 }],
});
expect(billing.freeze).toHaveBeenCalledWith(expect.objectContaining({ amountCents: 3, smsUnits: 1, relatedId: 'task-1' }));
expect(prisma.smsMessageRecord.updateMany).toHaveBeenCalledWith({
where: { batchTaskId: 'task-1', status: 'scheduled' },
data: { status: 'queued' },
});
});
it('cancels scheduled tasks before dispatch', async () => {
const { service, prisma } = createService();
prisma.smsBatchTask.findUnique.mockResolvedValue({ id: 'task-1', status: 'scheduled' });
await service.cancelBatchTask('task-1');
expect(prisma.smsMessageRecord.updateMany).toHaveBeenCalledWith({
where: { batchTaskId: 'task-1', status: 'scheduled' },
data: { status: 'canceled', errorMessage: '定时任务已取消' },
});
expect(prisma.smsBatchTask.update).toHaveBeenCalledWith({
where: { id: 'task-1' },
data: { status: 'canceled', canceledAt: expect.any(Date) },
});
});
it('blocks sending when enterprise certification is not approved', async () => {
const { service, prisma } = createService();
prisma.tenant.findUnique.mockResolvedValue({ id: 'tenant-1', status: 'active', certificationStatus: 'rejected' });
await expect(
service.createBatchTask({
tenantId: 'tenant-1',
applicationId: 'app-1',
templateId: 'tpl-1',
content: 'hello',
phones: ['13800000001'],
}),
).rejects.toThrow('企业认证未通过，不能发送短信');
});
it('previews imported phone files with duplicate, invalid, blacklist, and variable errors', async () => {
const { service, prisma } = createService();
prisma.enterpriseBlacklist.findMany.mockResolvedValue([{ phoneNumber: '13800000003' }]);
await expect(
service.previewImport({
tenantId: 'tenant-1',
content: 'phoneNumber,code\n13800000001,1234\n13800000001,1234\nbad,1234\n13800000003,1234\n13900000001,',
requiredVariables: ['code'],
}),
).resolves.toEqual(
expect.objectContaining({
totalRows: 5,
validCount: 1,
errorCount: 4,
phones: ['13800000001'],
errors: expect.arrayContaining([
expect.objectContaining({ reason: '重复号码' }),
expect.objectContaining({ reason: '手机号格式非法' }),
expect.objectContaining({ reason: '命中黑名单' }),
expect.objectContaining({ reason: '变量列缺失：code' }),
]),
}),
);
});
it('adds queued message jobs for a batch task', async () => {
const { service, prisma } = createService();
const add = jest.fn().mockResolvedValue(undefined);
@@ -155,8 +287,8 @@ describe('SendChainService', () => {
);
});
it('updates submit result status and task progress', async () => {
const { service, prisma } = createService();
it('updates submit result status, charges billing, and task progress', async () => {
const { service, prisma, billing } = createService();
await service.handleSubmitResult({
messageId: 'MSG-1',
@@ -176,6 +308,32 @@ describe('SendChainService', () => {
where: { id: 'record-1' },
data: expect.objectContaining({ gatewayMessageId: 'GW-1', status: 'submitted', submitStatus: 'accepted' }),
});
expect(billing.release).toHaveBeenCalledWith(expect.objectContaining({ amountCents: 3, smsUnits: 1, relatedId: 'task-1' }));
expect(billing.charge).toHaveBeenCalledWith(expect.objectContaining({ amountCents: 3, smsUnits: 1, relatedId: 'MSG-1' }));
expect(prisma.smsBillingRecord.create).toHaveBeenCalledWith({
data: expect.objectContaining({ messageId: 'MSG-1', amountCents: 3, billingStatus: 'charged', transactionId: 'tx-charge' }),
});
});
it('releases reservation for rejected submit result and refunds failed receipts', async () => {
const { service, billing } = createService();
await service.handleSubmitResult({
messageId: 'MSG-1',
channelId: 'channel-1',
gatewayMessageId: 'GW-1',
submitStatus: 'rejected',
});
expect(billing.release).toHaveBeenCalledWith(expect.objectContaining({ remark: expect.stringContaining('提交失败释放冻结') }));
await service.handleReceipt({
messageId: 'MSG-1',
channelId: 'channel-1',
gatewayMessageId: 'GW-1',
receiptStatus: 'undelivered',
rawStatus: 'UNDELIV',
});
expect(billing.refund).toHaveBeenCalledWith(expect.objectContaining({ remark: '最终失败退款' }));
});
it('records receipts and uplink messages from gateway events', async () => {
api/src/send-chain/send-chain.service.ts
+388 -4
View File
@@ -1,4 +1,4 @@
import { Injectable, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { BadRequestException, Injectable, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { Queue, Worker } from 'bullmq';
import IORedis from 'ioredis';
import { randomUUID } from 'node:crypto';
@@ -14,6 +14,8 @@ export interface CreateBatchTaskDto {
content: string;
category?: string;
phones: string[];
sendMode?: 'immediate' | 'scheduled';
scheduledAt?: string;
variables?: Record<string, unknown>;
createdById?: string;
sourceIp?: string;
@@ -60,6 +62,20 @@ export interface TimeoutUnknownDto {
olderThanHours?: number;
}
export interface ImportPreviewDto {
tenantId: string;
content: string;
fileName?: string;
encoding?: 'utf8' | 'gbk';
delimiter?: ',' | '\t';
requiredVariables?: string[];
}
export interface ConfirmImportDto extends CreateBatchTaskDto {
importContent: string;
requiredVariables?: string[];
}
interface SendJob {
messageRecordId: string;
}
@@ -95,6 +111,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
async createBatchTask(data: CreateBatchTaskDto) {
const phones = [...new Set(data.phones ?? [])];
const schedule = parseSchedule(data);
await this.validateSendResources(data.tenantId, data.applicationId, data.templateId);
const unitPrice = await this.resolveUnitPrice(data.tenantId, data.applicationId);
const risk = await this.riskReview.evaluateTask({
tenantId: data.tenantId,
applicationId: data.applicationId,
@@ -111,8 +130,20 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
taskId: risk.task?.id,
content: data.content,
phoneCount: phones.length,
unitPrice,
});
const batchStatus = statusFromRisk(risk.status);
const batchStatus = statusFromRisk(risk.status, Boolean(schedule.scheduledAt));
const shouldReserveBalance = batchStatus === 'ready';
if (risk.status === 'approved') {
const accountCheck = await this.billing.checkAccount({
tenantId: data.tenantId,
amountCents: billing.amountCents,
smsUnits: billing.totalBillingUnits,
});
if (!accountCheck.canSend) {
throw new BadRequestException('企业账户余额、套餐余量或授信额度不足');
}
}
const task = await this.prisma.smsBatchTask.create({
data: {
tenantId: data.tenantId,
@@ -129,9 +160,20 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
reviewReason: risk.status === 'pending_review' ? risk.reason : null,
rejectReason: risk.status === 'rejected' ? risk.reason : null,
progressTotal: phones.length,
scheduledAt: schedule.scheduledAt,
createdById: data.createdById,
},
});
if (shouldReserveBalance && billing.amountCents + billing.totalBillingUnits > 0) {
await this.billing.freeze({
tenantId: data.tenantId,
amountCents: billing.amountCents,
smsUnits: billing.totalBillingUnits,
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: '发送任务创建冻结',
});
}
await this.prisma.smsApiRequest.create({
data: {
tenantId: data.tenantId,
@@ -143,6 +185,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
phoneTotal: phones.length,
contentLength: [...data.content].length,
category: data.category,
sendMode: schedule.scheduledAt ? 'scheduled' : 'immediate',
scheduledAt: schedule.scheduledAt?.toISOString(),
},
status: batchStatus === 'rejected' ? 'rejected' : 'accepted',
},
@@ -160,7 +204,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
billingUnits: billing.billingUnitsPerMessage,
unitPrice: billing.unitPrice,
amountCents: billing.billingUnitsPerMessage * billing.unitPrice,
status: batchStatus === 'ready' ? 'queued' : batchStatus,
status: batchStatus === 'ready' ? 'queued' : batchStatus === 'scheduled' ? 'scheduled' : batchStatus,
errorMessage: risk.status === 'rejected' ? risk.reason ?? undefined : undefined,
})),
});
@@ -219,11 +263,81 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
});
}
async previewImport(data: ImportPreviewDto) {
const sizeBytes = Buffer.byteLength(data.content, 'utf8');
if (sizeBytes > 20 * 1024 * 1024) {
throw new BadRequestException('导入文件不能超过 20MB');
}
const rows = parseImportRows(data.content, data.delimiter);
const phones: string[] = [];
const errors: Array<{ rowNumber: number; phoneNumber?: string; reason: string }> = [];
const requiredVariables = data.requiredVariables ?? [];
const enterpriseBlacklist = await this.prisma.enterpriseBlacklist.findMany({
where: { tenantId: data.tenantId, status: 'active' },
select: { phoneNumber: true },
});
const globalBlacklist = await this.prisma.globalBlacklist.findMany({
where: { status: 'active' },
select: { phoneNumber: true },
});
const blacklist = new Set([...enterpriseBlacklist, ...globalBlacklist].map((item) => item.phoneNumber));
const seen = new Set<string>();
for (const row of rows) {
if (!row.phoneNumber) {
errors.push({ rowNumber: row.rowNumber, reason: '缺少手机号' });
continue;
}
if (!/^1[3-9]\d{9}$/.test(row.phoneNumber)) {
errors.push({ rowNumber: row.rowNumber, phoneNumber: row.phoneNumber, reason: '手机号格式非法' });
continue;
}
if (seen.has(row.phoneNumber)) {
errors.push({ rowNumber: row.rowNumber, phoneNumber: row.phoneNumber, reason: '重复号码' });
continue;
}
if (blacklist.has(row.phoneNumber)) {
errors.push({ rowNumber: row.rowNumber, phoneNumber: row.phoneNumber, reason: '命中黑名单' });
continue;
}
const missingVariables = requiredVariables.filter((name) => !row.variables[name]);
if (missingVariables.length > 0) {
errors.push({ rowNumber: row.rowNumber, phoneNumber: row.phoneNumber, reason: `变量列缺失：${missingVariables.join(',')}` });
continue;
}
seen.add(row.phoneNumber);
phones.push(row.phoneNumber);
}
return {
fileName: data.fileName,
encoding: data.encoding ?? 'utf8',
totalRows: rows.length,
validCount: phones.length,
errorCount: errors.length,
phones,
errors,
};
}
async confirmImport(data: ConfirmImportDto) {
const preview = await this.previewImport({
tenantId: data.tenantId,
content: data.importContent,
requiredVariables: data.requiredVariables,
});
if (preview.validCount === 0) {
throw new BadRequestException('导入文件没有可发送号码');
}
return this.createBatchTask({ ...data, phones: preview.phones });
}
async enqueueBatchTask(taskId: string) {
const task = await this.prisma.smsBatchTask.findUnique({ where: { id: taskId } });
if (!task) {
throw new NotFoundException('SMS batch task not found');
}
if (task.status === 'canceled') {
throw new BadRequestException('SMS batch task is canceled');
}
const messages = await this.prisma.smsMessageRecord.findMany({
where: { batchTaskId: taskId, status: 'queued' },
select: { id: true },
@@ -237,6 +351,77 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return { taskId, enqueued: messages.length };
}
async cancelBatchTask(taskId: string) {
const task = await this.prisma.smsBatchTask.findUnique({ where: { id: taskId } });
if (!task) {
throw new NotFoundException('SMS batch task not found');
}
if (task.status !== 'scheduled') {
throw new BadRequestException('Only scheduled SMS batch tasks can be canceled before dispatch');
}
await this.prisma.smsMessageRecord.updateMany({
where: { batchTaskId: taskId, status: 'scheduled' },
data: { status: 'canceled', errorMessage: '定时任务已取消' },
});
return this.prisma.smsBatchTask.update({
where: { id: taskId },
data: { status: 'canceled', canceledAt: new Date() },
});
}
async dispatchDueScheduledTasks(now = new Date()) {
const tasks = await this.prisma.smsBatchTask.findMany({
where: { status: 'scheduled', scheduledAt: { lte: now } },
orderBy: { scheduledAt: 'asc' },
take: 100,
});
const results: Array<{ taskId: string; status: string; enqueued?: number; reason?: string }> = [];
for (const task of tasks) {
try {
await this.validateSendResources(task.tenantId, task.applicationId ?? undefined, task.templateId ?? undefined);
const messages = await this.prisma.smsMessageRecord.findMany({
where: { batchTaskId: task.id, status: 'scheduled' },
select: { id: true, amountCents: true, billingUnits: true },
take: 100000,
});
const amountCents = messages.reduce((sum, message) => sum + message.amountCents, 0);
const smsUnits = messages.reduce((sum, message) => sum + message.billingUnits, 0);
const accountCheck = await this.billing.checkAccount({ tenantId: task.tenantId, amountCents, smsUnits });
if (!accountCheck.canSend) {
throw new BadRequestException('定时任务到点时企业账户余额、套餐余量或授信额度不足');
}
if (amountCents + smsUnits > 0) {
await this.billing.freeze({
tenantId: task.tenantId,
amountCents,
smsUnits,
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: '定时任务到点冻结',
});
}
await this.prisma.smsMessageRecord.updateMany({
where: { batchTaskId: task.id, status: 'scheduled' },
data: { status: 'queued' },
});
const enqueued = await this.enqueueBatchTask(task.id);
results.push({ taskId: task.id, status: 'queued', enqueued: enqueued.enqueued });
} catch (error) {
const reason = error instanceof Error ? error.message : '定时任务到点执行失败';
await this.prisma.smsMessageRecord.updateMany({
where: { batchTaskId: task.id, status: 'scheduled' },
data: { status: 'rejected', errorMessage: reason },
});
await this.prisma.smsBatchTask.update({
where: { id: task.id },
data: { status: 'failed', rejectReason: reason },
});
results.push({ taskId: task.id, status: 'failed', reason });
}
}
return { dispatched: results.filter((result) => result.status === 'queued').length, results };
}
startWorker() {
if (this.worker) {
return { status: 'already_started' };
@@ -332,6 +517,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
},
});
const status = data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed';
if (data.submitStatus === 'accepted') {
await this.chargeAcceptedMessage(message);
} else {
await this.releaseMessageReservation(message, data.submitStatus === 'timeout' ? '提交超时释放冻结' : '提交失败释放冻结');
}
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: {
@@ -353,6 +543,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date();
const status =
data.receiptStatus === 'delivered' ? 'delivered' : data.receiptStatus === 'unknown' ? 'unknown' : 'failed';
if (status === 'failed') {
await this.refundMessage(message, '最终失败退款');
}
await this.prisma.smsReceiptRecord.create({
data: {
tenantId: message.tenantId,
@@ -418,6 +611,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
where: { id: { in: candidates.map((candidate) => candidate.id) } },
data: { status: 'timeout', timeoutAt: new Date(), errorMessage: '72小时未收到明确回执，自动转超时' },
});
for (const candidate of candidates) {
const message = await this.prisma.smsMessageRecord.findUnique({ where: { id: candidate.id } });
if (message) {
await this.refundMessage(message, '72小时未收到明确回执，自动超时退款');
}
}
for (const batchTaskId of new Set(candidates.map((candidate) => candidate.batchTaskId))) {
await this.refreshTaskProgress(batchTaskId);
}
@@ -444,6 +643,135 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return channel;
}
private async resolveUnitPrice(tenantId: string, applicationId?: string) {
try {
const channel = await this.selectChannel(tenantId, applicationId);
return channel.unitPrice ?? 0;
} catch {
return 0;
}
}
private async validateSendResources(tenantId: string, applicationId?: string, templateId?: string) {
const tenant = await this.prisma.tenant.findUnique({ where: { id: tenantId } });
if (!tenant || tenant.status !== 'active') {
throw new BadRequestException('企业客户不存在或已停用');
}
if (tenant.certificationStatus !== 'approved') {
throw new BadRequestException('企业认证未通过，不能发送短信');
}
if (!applicationId) {
return;
}
const application = await this.prisma.smsApplication.findUnique({ where: { id: applicationId } });
if (!application || application.tenantId !== tenantId || application.status !== 'active') {
throw new BadRequestException('短信应用不存在或已停用');
}
if (!templateId) {
return;
}
const template = await this.prisma.smsTemplate.findUnique({
where: { id: templateId },
include: { signature: true },
});
if (!template || template.tenantId !== tenantId || template.applicationId !== applicationId || template.auditStatus !== 'approved') {
throw new BadRequestException('短信模板不存在、未通过审核或不属于当前应用');
}
if (!template.signature || template.signature.auditStatus !== 'approved' || template.signature.reportStatus !== 'approved') {
throw new BadRequestException('短信签名未审核通过或通道报备未通过');
}
}
private async chargeAcceptedMessage(message: {
tenantId: string;
applicationId?: string | null;
batchTaskId: string;
messageId: string;
phoneNumber: string;
content: string;
billingUnits: number;
unitPrice: number;
amountCents: number;
}) {
const amountCents = message.amountCents ?? 0;
const smsUnits = message.billingUnits ?? 0;
if (amountCents + smsUnits > 0) {
await this.billing.release({
tenantId: message.tenantId,
amountCents,
smsUnits,
relatedType: 'sms_batch_task',
relatedId: message.batchTaskId,
remark: `短信 ${message.messageId} 提交成功释放冻结并转扣费`,
});
}
const transaction = await this.billing.charge({
tenantId: message.tenantId,
amountCents,
smsUnits,
relatedType: 'sms_message_record',
relatedId: message.messageId,
remark: '提交成功扣费',
});
const exists = await this.prisma.smsBillingRecord.findFirst({ where: { messageId: message.messageId } });
const data = {
tenantId: message.tenantId,
applicationId: message.applicationId ?? undefined,
taskId: message.batchTaskId,
messageId: message.messageId,
phoneNumber: message.phoneNumber,
contentLength: [...message.content].length,
billingUnits: smsUnits,
unitPrice: message.unitPrice ?? 0,
amountCents,
billingStatus: 'charged',
transactionId: transaction.id,
};
if (exists) {
await this.prisma.smsBillingRecord.update({ where: { id: exists.id }, data });
return;
}
await this.prisma.smsBillingRecord.create({ data });
}
private async releaseMessageReservation(
message: { tenantId: string; batchTaskId: string; messageId: string; amountCents: number; billingUnits: number },
remark: string,
) {
if ((message.amountCents ?? 0) + (message.billingUnits ?? 0) <= 0) {
return;
}
await this.billing.release({
tenantId: message.tenantId,
amountCents: message.amountCents,
smsUnits: message.billingUnits,
relatedType: 'sms_batch_task',
relatedId: message.batchTaskId,
remark: `${remark}: ${message.messageId}`,
});
}
private async refundMessage(
message: { tenantId: string; messageId: string; amountCents: number; billingUnits: number },
remark: string,
) {
if ((message.amountCents ?? 0) + (message.billingUnits ?? 0) <= 0) {
return;
}
const transaction = await this.billing.refund({
tenantId: message.tenantId,
amountCents: message.amountCents,
smsUnits: message.billingUnits,
relatedType: 'sms_message_record',
relatedId: message.messageId,
remark,
});
await this.prisma.smsBillingRecord.updateMany({
where: { messageId: message.messageId },
data: { billingStatus: 'refunded', transactionId: transaction.id },
});
}
private async waitForChannelRateLimit(channelId: string, tps: number) {
const redis = this.getRedis();
for (;;) {
@@ -520,16 +848,72 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
}
function statusFromRisk(status: string) {
function statusFromRisk(status: string, scheduled: boolean) {
if (status === 'rejected') {
return 'rejected';
}
if (status === 'pending_review') {
return 'pending_review';
}
if (scheduled) {
return 'scheduled';
}
return 'ready';
}
function parseSchedule(data: CreateBatchTaskDto) {
if (data.sendMode !== 'scheduled' && !data.scheduledAt) {
return { scheduledAt: null };
}
if (!data.scheduledAt) {
throw new BadRequestException('定时发送必须提供 scheduledAt');
}
const scheduledAt = new Date(data.scheduledAt);
if (Number.isNaN(scheduledAt.getTime())) {
throw new BadRequestException('scheduledAt 时间格式无效');
}
if (scheduledAt.getTime() <= Date.now()) {
throw new BadRequestException('scheduledAt 必须晚于当前时间');
}
return { scheduledAt };
}
function parseImportRows(content: string, delimiter?: ',' | '\t') {
const normalized = content.replace(/^\uFEFF/, '');
const lines = normalized.split(/\r?\n/).filter((line) => line.trim().length > 0);
if (lines.length === 0) {
return [];
}
const firstDelimiter = delimiter ?? (lines[0].includes(',') ? ',' : '\t');
const firstCells = splitImportLine(lines[0], firstDelimiter);
const hasHeader = firstCells.some((cell) => ['phone', 'phoneNumber', 'mobile', '手机号'].includes(cell));
const headers = hasHeader ? firstCells : ['phoneNumber'];
const dataLines = hasHeader ? lines.slice(1) : lines;
return dataLines.map((line, index) => {
const cells = splitImportLine(line, firstDelimiter);
const row: { rowNumber: number; phoneNumber?: string; variables: Record<string, string> } = {
rowNumber: (hasHeader ? index + 2 : index + 1),
phoneNumber: hasHeader ? cellByHeader(headers, cells, ['phone', 'phoneNumber', 'mobile', '手机号']) : cells[0],
variables: {},
};
headers.forEach((header, cellIndex) => {
if (!['phone', 'phoneNumber', 'mobile', '手机号'].includes(header)) {
row.variables[header] = cells[cellIndex] ?? '';
}
});
return row;
});
}
function splitImportLine(line: string, delimiter: ',' | '\t') {
return line.split(delimiter).map((cell) => cell.trim().replace(/^"|"$/g, ''));
}
function cellByHeader(headers: string[], cells: string[], candidates: string[]) {
const index = headers.findIndex((header) => candidates.includes(header));
return index >= 0 ? cells[index] : undefined;
}
function bullmqConnection() {
const redisUrl = new URL(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379');
return {