From 4c8f7dd73fed2e7ce7895a328c5f90ed99fcf5d9 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Tue, 7 Jul 2026 14:49:10 +0800 Subject: [PATCH] feat: add application queue priority routing --- .../migration.sql | 2 + api/prisma/schema.prisma | 2 + api/src/send-chain/send-chain.service.spec.ts | 25 +++- api/src/send-chain/send-chain.service.ts | 37 +++++- api/src/sms-config/sms-config.service.spec.ts | 40 +++++- api/src/sms-config/sms-config.service.ts | 18 +++ docs/contracts/examples/submit-command.json | 1 + .../gateway-queue-messages.schema.json | 2 + .../first-version-development-requirements.md | 123 +++++++++++++++--- docs/system-functional-test-cases.md | 63 ++++++++- docs/testing-plan.md | 6 +- docs/testing-progress.md | 17 +++ gateway/internal/queue/messages.go | 1 + gateway/internal/queue/messages_test.go | 50 +++++++ src/api/adminApi.ts | 6 +- .../admin/AdminEnterpriseApplicationsPage.tsx | 22 ++-- .../admin/AdminSmsApplicationFormPage.tsx | 23 +++- 17 files changed, 392 insertions(+), 46 deletions(-) create mode 100644 api/prisma/migrations/20260707143000_add_application_queue_priority/migration.sql create mode 100644 gateway/internal/queue/messages_test.go diff --git a/api/prisma/migrations/20260707143000_add_application_queue_priority/migration.sql b/api/prisma/migrations/20260707143000_add_application_queue_priority/migration.sql new file mode 100644 index 0000000..7b555e4 --- /dev/null +++ b/api/prisma/migrations/20260707143000_add_application_queue_priority/migration.sql @@ -0,0 +1,2 @@ +ALTER TABLE "SmsApplication" ADD COLUMN "queuePriority" TEXT NOT NULL DEFAULT 'normal'; +ALTER TABLE "SmsMessageRecord" ADD COLUMN "queuePriority" TEXT NOT NULL DEFAULT 'normal'; diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index 99ad122..14fa4fb 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -339,6 +339,7 @@ model SmsApplication { secretHash String dailyLimit Int? customerUnitPrice Int @default(0) + queuePriority String @default("normal") maxPhonesPerTask Int @default(1000000) templateMismatchMode String @default("reject") status String @default("active") @@ -852,6 +853,7 @@ model SmsMessageRecord { billingUnits Int @default(1) unitPrice Int @default(0) amountCents Int @default(0) + queuePriority String @default("normal") channelId String? submitId String? gatewayMessageId String? diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 9453b05..fc88652 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -17,6 +17,7 @@ function createPrismaMock() { unitPrice: 3, amountCents: 3, status: 'queued', + queuePriority: 'normal', submitId: 'SUB-1', gatewayMessageId: 'GW-1', channelId: 'channel-1', @@ -58,7 +59,7 @@ function createPrismaMock() { findUnique: jest.fn().mockResolvedValue({ id: 'tenant-1', status: 'active', certificationStatus: 'approved' }), }, smsApplication: { - findUnique: jest.fn().mockResolvedValue({ id: 'app-1', tenantId: 'tenant-1', status: 'active', customerUnitPrice: 3 }), + findUnique: jest.fn().mockResolvedValue({ id: 'app-1', tenantId: 'tenant-1', status: 'active', customerUnitPrice: 3, queuePriority: 'normal' }), }, smsTemplate: { findUnique: jest.fn().mockResolvedValue({ @@ -183,8 +184,8 @@ describe('SendChainService', () => { }); expect(prisma.smsMessageRecord.createMany).toHaveBeenCalledWith({ data: expect.arrayContaining([ - expect.objectContaining({ phoneNumber: '13800000001', status: 'queued', billingUnits: 1, amountCents: 3 }), - expect.objectContaining({ phoneNumber: '13800000002', status: 'queued', billingUnits: 1, amountCents: 3 }), + expect.objectContaining({ phoneNumber: '13800000001', status: 'queued', billingUnits: 1, amountCents: 3, queuePriority: 'normal' }), + expect.objectContaining({ phoneNumber: '13800000002', status: 'queued', billingUnits: 1, amountCents: 3, queuePriority: 'normal' }), ]), }); expect(billing.freeze).toHaveBeenCalledWith(expect.objectContaining({ amountCents: 6, smsUnits: 2, relatedId: 'task-1' })); @@ -309,10 +310,25 @@ describe('SendChainService', () => { service['getSendQueue'] = jest.fn().mockReturnValue({ add }); await expect(service.enqueueBatchTask('task-1')).resolves.toEqual({ taskId: 'task-1', enqueued: 1 }); - expect(add).toHaveBeenCalledWith('send-message', { messageRecordId: 'record-1' }, { jobId: 'record-1', attempts: 3 }); + expect(add).toHaveBeenCalledWith('send-message', { messageRecordId: 'record-1' }, { jobId: 'record-1', attempts: 3, priority: 100 }); expect(prisma.smsBatchTask.update).toHaveBeenCalledWith({ where: { id: 'task-1' }, data: { status: 'queued' } }); }); + it('adds priority message jobs ahead of normal message jobs', async () => { + const { service, prisma } = createService(); + const add = jest.fn().mockResolvedValue(undefined); + prisma.smsMessageRecord.findMany.mockResolvedValue([ + { id: 'record-priority', batchTaskId: 'task-1', queuePriority: 'priority' }, + { id: 'record-normal', batchTaskId: 'task-1', queuePriority: 'normal' }, + ]); + service['getSendQueue'] = jest.fn().mockReturnValue({ add }); + + await expect(service.enqueueBatchTask('task-1')).resolves.toEqual({ taskId: 'task-1', enqueued: 2 }); + + expect(add).toHaveBeenCalledWith('send-message', { messageRecordId: 'record-priority' }, { jobId: 'record-priority', attempts: 3, priority: 1 }); + expect(add).toHaveBeenCalledWith('send-message', { messageRecordId: 'record-normal' }, { jobId: 'record-normal', attempts: 3, priority: 100 }); + }); + it('routes queued messages to gateway submit commands', async () => { const { service, prisma } = createService(); const gatewayAdd = jest.fn().mockResolvedValue(undefined); @@ -333,6 +349,7 @@ describe('SendChainService', () => { messageType: 'SubmitCommand', messageId: 'MSG-1', channelId: 'channel-1', + queuePriority: 'normal', phoneNumber: '13800000001', route: expect.objectContaining({ channelCode: 'CMPP-A', rateLimitPerSecond: 100 }), cmpp: expect.objectContaining({ serviceId: 'SMS', srcId: '10690000' }), diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index a331753..49b3966 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -80,6 +80,8 @@ interface SendJob { messageRecordId: string; } +type QueuePriority = 'normal' | 'priority'; + type RoutedChannel = { channel: { id: string; @@ -101,6 +103,10 @@ type RoutedChannel = { const SEND_QUEUE = 'sms.send.queue'; const GATEWAY_SUBMIT_QUEUE = 'gateway.submit.queue'; +const BULLMQ_PRIORITY: Record = { + priority: 1, + normal: 100, +}; @Injectable() export class SendChainService implements OnModuleInit, OnModuleDestroy { @@ -133,6 +139,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { const schedule = parseSchedule(data); await this.validateSendResources(data.tenantId, data.applicationId, data.templateId); const unitPrice = await this.resolveUnitPrice(data.tenantId, data.applicationId); + const queuePriority = await this.resolveQueuePriority(data.tenantId, data.applicationId); const risk = await this.riskReview.evaluateTask({ tenantId: data.tenantId, applicationId: data.applicationId, @@ -223,6 +230,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { billingUnits: billing.billingUnitsPerMessage, unitPrice: billing.unitPrice, amountCents: billing.billingUnitsPerMessage * billing.unitPrice, + queuePriority, status: batchStatus === 'ready' ? 'queued' : batchStatus === 'scheduled' ? 'scheduled' : batchStatus, errorMessage: risk.status === 'rejected' ? risk.reason ?? undefined : undefined, })), @@ -365,12 +373,17 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { } const messages = await this.prisma.smsMessageRecord.findMany({ where: { batchTaskId: taskId, status: 'queued' }, - select: { id: true }, + select: { id: true, queuePriority: 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 }); + const queuePriority = normalizeQueuePriority(message.queuePriority); + await queue.add('send-message', { messageRecordId: message.id }, { + jobId: message.id, + attempts: 3, + priority: BULLMQ_PRIORITY[queuePriority], + }); } await this.prisma.smsBatchTask.update({ where: { id: taskId }, data: { status: 'queued' } }); return { taskId, enqueued: messages.length }; @@ -653,6 +666,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { phoneNumber: string; content: string; billingUnits: number; + queuePriority?: string | null; template?: { signature?: { id?: string | null; name?: string | null } | null } | null; }, routed: RoutedChannel, @@ -701,6 +715,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { applicationId: message.applicationId ?? 'unknown', taskId: message.batchTaskId, submitId, + queuePriority: normalizeQueuePriority(message.queuePriority), phoneNumber: message.phoneNumber, content: message.content, signature: message.template?.signature?.name ?? 'SMS', @@ -882,6 +897,20 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return application.customerUnitPrice ?? 0; } + private async resolveQueuePriority(tenantId: string, applicationId?: string): Promise { + if (!applicationId) { + return 'normal'; + } + const application = await this.prisma.smsApplication.findUnique({ + where: { id: applicationId }, + select: { tenantId: true, queuePriority: true }, + }); + if (!application || application.tenantId !== tenantId) { + return 'normal'; + } + return normalizeQueuePriority(application.queuePriority); + } + private async validateSendResources(tenantId: string, applicationId?: string, templateId?: string) { const tenant = await this.prisma.tenant.findUnique({ where: { id: tenantId } }); if (!tenant || tenant.status !== 'active') { @@ -1202,6 +1231,10 @@ function normalizeCarrier(carrier?: string | null) { return value || 'mobile'; } +function normalizeQueuePriority(queuePriority?: string | null): QueuePriority { + return queuePriority === 'priority' ? 'priority' : 'normal'; +} + function isCarrierCompatible(channelCarrier: string | null | undefined, targetCarrier: string) { const normalized = normalizeCarrier(channelCarrier); return normalized === 'all' || normalized === targetCarrier; diff --git a/api/src/sms-config/sms-config.service.spec.ts b/api/src/sms-config/sms-config.service.spec.ts index 878f4f4..5620af8 100644 --- a/api/src/sms-config/sms-config.service.spec.ts +++ b/api/src/sms-config/sms-config.service.spec.ts @@ -8,6 +8,7 @@ function createPrismaMock() { tenantId: 'tenant-1', name: '应用A', status: 'active', + queuePriority: 'normal', tenant: { id: 'tenant-1', name: '租户A', code: 'TENANT-A' }, messageRecords: [{ status: 'delivered' }, { status: 'undelivered' }], }]), @@ -16,6 +17,7 @@ function createPrismaMock() { tenantId: 'tenant-1', name: '应用A', status: 'active', + queuePriority: 'normal', secretHash: 'secret-hash', tenant: { id: 'tenant-1', name: '租户A', code: 'TENANT-A' }, }), @@ -147,6 +149,7 @@ describe('SmsConfigService', () => { expect.objectContaining({ id: 'app-1', cmppStatus: 'connected', + queuePriority: 'normal', sentToday: 2, deliveryRate: 50, cmppConnections: [expect.objectContaining({ connectionId: 'conn-a' })], @@ -167,6 +170,40 @@ describe('SmsConfigService', () => { })); }); + it('creates enterprise applications with persisted queue priority', async () => { + const prisma = createPrismaMock(); + const service = new SmsConfigService(prisma as never); + + await expect(service.createApplication({ + tenantId: 'tenant-1', + name: '优先应用', + queuePriority: 'priority', + ipAllowlist: ['10.0.0.1/32'], + })).resolves.toEqual(expect.objectContaining({ id: 'app-new' })); + + expect(prisma.smsApplication.create).toHaveBeenCalledWith(expect.objectContaining({ + data: expect.objectContaining({ + tenantId: 'tenant-1', + name: '优先应用', + queuePriority: 'priority', + ipAllowlist: { create: [{ ipCidr: '10.0.0.1/32' }] }, + }), + })); + }); + + it('rejects invalid enterprise application queue priority', async () => { + const prisma = createPrismaMock(); + const service = new SmsConfigService(prisma as never); + + expect(() => service.createApplication({ + tenantId: 'tenant-1', + name: '异常应用', + queuePriority: 'urgent', + })).toThrow('queuePriority must be normal or priority'); + + expect(prisma.smsApplication.create).not.toHaveBeenCalled(); + }); + it('updates enterprise application profile and allowlist through a transaction', async () => { const prisma = createPrismaMock(); const tx = { @@ -181,7 +218,7 @@ describe('SmsConfigService', () => { prisma.$transaction.mockImplementationOnce((callback: (client: typeof tx) => unknown) => callback(tx)); const service = new SmsConfigService(prisma as never); - await expect(service.updateApplication('app-1', { name: '新应用', customerUnitPrice: 300, ipAllowlist: ['10.0.0.1/32'] })) + await expect(service.updateApplication('app-1', { name: '新应用', customerUnitPrice: 300, queuePriority: 'priority', ipAllowlist: ['10.0.0.1/32'] })) .resolves.toEqual(expect.objectContaining({ id: 'app-1', name: '新应用' })); expect(tx.smsApplicationIpAllowlist.deleteMany).toHaveBeenCalledWith({ where: { applicationId: 'app-1' } }); @@ -190,6 +227,7 @@ describe('SmsConfigService', () => { data: expect.objectContaining({ name: '新应用', customerUnitPrice: 300, + queuePriority: 'priority', ipAllowlist: { create: [{ ipCidr: '10.0.0.1/32' }] }, }), })); diff --git a/api/src/sms-config/sms-config.service.ts b/api/src/sms-config/sms-config.service.ts index 093250c..ea1664b 100644 --- a/api/src/sms-config/sms-config.service.ts +++ b/api/src/sms-config/sms-config.service.ts @@ -10,6 +10,7 @@ export interface CreateSmsApplicationDto { callbackUrl?: string; dailyLimit?: number; customerUnitPrice?: number; + queuePriority?: string; maxPhonesPerTask?: number; templateMismatchMode?: string; ipAllowlist?: string[]; @@ -85,6 +86,9 @@ export interface ApplicationListQuery { includeConnections?: boolean; } +const APPLICATION_QUEUE_PRIORITIES = ['normal', 'priority'] as const; +type ApplicationQueuePriority = typeof APPLICATION_QUEUE_PRIORITIES[number]; + @Injectable() export class SmsConfigService { constructor(private readonly prisma: PrismaService) {} @@ -148,6 +152,7 @@ export class SmsConfigService { createApplication(data: CreateSmsApplicationDto) { const secret = randomBytes(24).toString('hex'); + const queuePriority = normalizeApplicationQueuePriority(data.queuePriority); return this.prisma.smsApplication.create({ data: { tenantId: data.tenantId, @@ -157,6 +162,7 @@ export class SmsConfigService { secretHash: hashSecret(secret), dailyLimit: data.dailyLimit, customerUnitPrice: data.customerUnitPrice ?? 0, + queuePriority, maxPhonesPerTask: data.maxPhonesPerTask ?? 1000000, templateMismatchMode: data.templateMismatchMode ?? 'reject', ipAllowlist: { @@ -172,6 +178,9 @@ export class SmsConfigService { if (!application) { throw new NotFoundException('Application not found'); } + const queuePriority = data.queuePriority === undefined + ? undefined + : normalizeApplicationQueuePriority(data.queuePriority); return this.prisma.$transaction(async (tx) => { if (data.ipAllowlist) { @@ -185,6 +194,7 @@ export class SmsConfigService { callbackUrl: data.callbackUrl, dailyLimit: data.dailyLimit, customerUnitPrice: data.customerUnitPrice, + queuePriority, maxPhonesPerTask: data.maxPhonesPerTask, templateMismatchMode: data.templateMismatchMode, status: data.status, @@ -738,6 +748,14 @@ function startOfToday() { return date; } +function normalizeApplicationQueuePriority(value?: string): ApplicationQueuePriority { + const queuePriority = value ?? 'normal'; + if (!APPLICATION_QUEUE_PRIORITIES.includes(queuePriority as ApplicationQueuePriority)) { + throw new BadRequestException('queuePriority must be normal or priority'); + } + return queuePriority as ApplicationQueuePriority; +} + function normalizeApplicationCmppStatus(connections: Array<{ status: string; currentConnections: number }>, applicationStatus: string) { if (applicationStatus !== 'active') { return 'inactive'; diff --git a/docs/contracts/examples/submit-command.json b/docs/contracts/examples/submit-command.json index 4178c05..cd87030 100644 --- a/docs/contracts/examples/submit-command.json +++ b/docs/contracts/examples/submit-command.json @@ -14,6 +14,7 @@ "signature": "测试平台", "templateId": "tpl-demo-code", "billingUnits": 1, + "queuePriority": "priority", "route": { "channelCode": "CMCC-CMPP-DEMO", "cmppAccountCode": "cmpp-account-demo", diff --git a/docs/contracts/gateway-queue-messages.schema.json b/docs/contracts/gateway-queue-messages.schema.json index b6eb029..0d7ca97 100644 --- a/docs/contracts/gateway-queue-messages.schema.json +++ b/docs/contracts/gateway-queue-messages.schema.json @@ -38,6 +38,7 @@ "signature", "templateId", "billingUnits", + "queuePriority", "route", "cmpp", "retry" @@ -53,6 +54,7 @@ "signature": { "type": "string", "minLength": 1 }, "templateId": { "type": "string", "minLength": 1 }, "billingUnits": { "type": "integer", "minimum": 1 }, + "queuePriority": { "enum": ["normal", "priority"] }, "route": { "type": "object", "required": ["channelCode", "cmppAccountCode", "priority"], diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 2201623..f6e3285 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -102,6 +102,9 @@ 2. 运营端可查看企业应用，并支持按企业名称、应用名称、状态搜索。 3. 运营端可启用、停用、编辑应用。 4. 应用必须归属企业，并关联后续签名、模板和发送任务。 +5. 短信应用必须配置发送队列等级：普通队列或优先队列。未配置时默认普通队列；优先队列用于验证码、登录确认、交易通知等高时效短信，普通队列用于营销、通知等常规短信。 +6. 发送队列等级属于真实业务配置，必须保存到后端数据库，并在客户端、运营端创建/编辑应用时展示和可修改；不得只作为前端展示字段。 +7. 运营端代企业新增短信应用时，第一步选择企业必须使用项目通用 Select/下拉控件，选项来自真实企业 API，支持加载中、空数据、错误态，不允许写死企业列表。 ### 4.3 签名与引流信息 @@ -130,6 +133,9 @@ 9. 平台批量任务只记录客户端创建的发送任务；API 调用和 CMPP 对接发送不进入批量任务。 10. 所有来源的短信，包括平台批量任务、API 调用、CMPP 对接发送，全部按手机号维度进入短信记录。 11. 任务进度、发送详情和短信记录实时或准实时更新。 +12. 发送入队必须按短信应用的队列等级分流到优先队列或普通队列；同等条件下优先队列消息必须先于普通队列消息被 Send Worker 消费并提交 Gateway。 +13. 优先队列只能改变待发送消息的调度顺序，不得绕过企业/应用状态、签名模板审核、通道报备、余额/授信、黑名单、风控、通道组路由、通道限速和 Gateway 连接可用性校验。 +14. 同一队列内部按创建时间、任务顺序和手机号拆分顺序保持 FIFO 或可解释的稳定排序；优先队列插队时必须可在 trace 或任务日志中追踪队列等级和入队时间。 ### 4.6 通道配置与路由 @@ -169,7 +175,65 @@ 8. 报备记录保留每次导出、导入、状态变更和操作人。 9. 发送前必须校验最终选中通道上的签名报备任务为 approved；补发切换到新通道时必须重新按新通道校验报备状态，未通过则该次发送失败。 -### 4.8 回执与上行 +### 4.8 CMPP Gateway 与外部接入 + +第一版发送链路必须区分两类 CMPP 连接，不允许用页面状态、HTTP 占位或模拟器结果替代真实协议能力： + +- 上游通道连接：平台作为 SP 客户端，连接运营商或供应商 SMSC，将平台已路由的短信提交到目标通道。 +- 下游客户接入：企业客户作为外部 SP 客户端，连接平台暴露的 CMPP 监听端口（生产端口 `17890`），通过 CMPP 协议向平台提交短信。 + +#### 4.8.1 上游通道连接能力 + +1. Gateway 必须按运营端通道配置连接上游 SMSC，使用通道的 `gatewayHost/gatewayPort/account/passwordCipher/srcId/cmppVersion` 完成 CMPP 2.0/3.0 connect/login。 +2. Gateway 必须校验上游 connect/login 返回码，区分 connected、auth_failed、connect_timeout、network_error、protocol_error 等状态，并回写 NestJS 真实连接状态。 +3. Gateway 必须支持每个通道配置期望连接数，建立多条长连接，并按连接维度维护 currentConnections、lastConnectedAt、lastHeartbeatAt、lastError、reconnectCount。 +4. Gateway 必须实现 ActiveTest 心跳与超时检测；连续心跳失败后连接进入 heartbeat_timeout/reconnecting，重连成功前该连接不可参与发送。 +5. Gateway 必须支持断线自动重连、指数退避或固定退避、最大重试间隔、重连日志和状态回写。 +6. Gateway 必须维护 CMPP sequenceId 与平台 messageId、submitId、channelId 的映射，submit resp 和 deliver 回执必须能追溯到原短信记录和提交尝试。 +7. Gateway 必须实现真实 CMPP Submit，包括短信内容编码、长短信拆分、RegisteredDelivery、serviceId、srcId、destTerminalId、msgFmt、feeType/feeCode 等字段映射。 +8. Gateway 必须消费 NestJS 投递的 `SubmitCommand` 队列或等价内部接口；提交成功、提交失败、超时均必须回传 `SubmitResult`，不得只停留在 API 侧入队。 +9. Gateway 必须按通道连接和窗口容量控制并发，处理窗口满、SMSC 慢响应、sequence 回绕、连接断开时的在途消息状态。 +10. Gateway 不承担业务审核、计费、签名报备、通道组路由、黑名单或敏感词判断；这些由 NestJS 完成，Gateway 只执行已授权通道提交与协议事件回传。 + +#### 4.8.2 下游客户 CMPP 接入能力 + +1. Gateway 必须监听生产 CMPP 端口 `17890`，作为平台侧 CMPP Server 接收企业客户系统连接；该端口不是 HTTP 健康检查或控制接口。 +2. Gateway 必须按企业应用生成的 CMPP 接入参数校验客户 connect/login，包括企业代码、账号、密码、CMPP 版本、源 IP 白名单、应用状态、企业状态、连接数上限。 +3. 客户端应用的 IP 白名单必须对 CMPP 下游连接生效；未命中白名单、应用停用、企业停用、密码错误、超过连接数上限时必须拒绝连接并记录系统日志。 +4. Gateway 必须维护应用级下游连接状态，回写 applicationId、tenantId、connectionId、currentConnections、desiredConnections、lastHeartbeatAt、lastError，运营端企业应用列表和连接详情必须来自这些真实状态。 +5. Gateway 必须实现下游 ActiveTest、Terminate、异常断开处理；断开后连接数和状态必须及时回写。 +6. Gateway 必须处理客户提交的 CMPP Submit，将手机号、内容、源地址、企业应用、客户消息序号等转换为平台发送请求。 +7. 下游 CMPP Submit 进入平台后，不创建客户端批量任务，但必须按手机号维度创建 `sms_message_record`，source 标记为 `cmpp`，并保留客户侧 sequence/msgId 映射。 +8. 下游 CMPP Submit 必须复用 NestJS 发送前校验：企业/应用状态、IP 白名单、签名/模板报备、模板匹配策略、风控、黑名单、余额/授信、运营商识别、通道组路由。 +9. 对客户 Submit 的响应必须符合 CMPP 协议：参数错误、鉴权失败、余额不足、模板或签名未通过、无可用通道、风控拒绝等应映射为明确失败状态；已接收进入平台发送链路时返回成功并生成可追踪平台 messageId。 +10. Gateway 必须支持平台最终回执向下游客户连接投递 Deliver Receipt；若客户连接已断开，应按策略缓存、重试或记录投递失败，不能丢失平台最终状态。 +11. Gateway 必须支持下游客户上行接入场景：收到运营商上行后，按接入号、手机号、应用、时间窗口匹配并向客户连接推送 Deliver，上行同时入库。 +12. 下游客户连接与上游通道连接必须隔离管理：客户侧账号密码不能用于连接上游通道，上游通道账号密码也不能作为客户接入凭据。 + +#### 4.8.3 回执、上行与幂等 + +1. Gateway 必须解析上游 deliver receipt，将 DELIVRD、UNDELIV、EXPIRED、REJECTD 等供应商状态归一化为平台 delivered、failed、unknown、timeout 等状态。 +2. Gateway 必须解析上游普通 deliver 上行短信，携带手机号、接入号、内容、接收时间、通道和原始报文摘要回传 NestJS。 +3. Gateway 回传 `SubmitResult`、`ReceiptEvent`、`UplinkEvent` 时必须包含 traceId、messageId、submitId 或可映射字段、channelId、gatewayMessageId、sequenceId，保证短信详情能展示完整历史尝试。 +4. 重复 submit resp、重复 receipt、迟到旧通道 receipt 必须交由 NestJS 幂等处理；Gateway 不得因本地缓存丢失而生成无法追踪的重复业务事件。 +5. Gateway 需要保留最小运行日志和指标：连接数、登录失败次数、心跳失败次数、submit TPS、submit resp 延迟、receipt 延迟、队列积压、重连次数、协议错误。 +6. Gateway 控制服务 `/health` 只能表示进程存活；真实验收必须检查上下游连接状态、队列消费、submit/receipt/uplink 事件闭环。 + +#### 4.8.4 当前实现缺口标记 + +截至当前版本，Go Gateway 已有 HTTP 控制服务、健康检查、连接上游 SMSC 的 `ConnectChannel` 控制入口、gocmpp 协议 spike 和队列消息结构；但仍缺少生产验收所需的完整能力： + +- 未实现 `17890` 入站 CMPP Server 监听。 +- 未实现下游客户 connect/login 鉴权、IP 白名单、应用级连接数限制和连接状态回写。 +- 未实现下游 CMPP Submit 到平台发送请求的转换。 +- 未实现客户侧 SubmitResp、最终 Deliver Receipt 和上行 Deliver 投递。 +- 未实现 Gateway 消费 `SubmitCommand` 并真实 submit 到上游通道的 worker。 +- 未实现上游 deliver receipt 和普通上行 deliver 的生产解析与事件回传闭环。 +- 未实现多连接窗口管理、在途消息恢复、断线重连后的消息状态处理。 + +这些缺口未补齐前，不能把 CMPP 对接发送、17890 端口联调、客户账号密码鉴权、客户 IP 白名单或真实网关 submit/receipt/uplink 作为“生产已验收通过”。 + +### 4.9 回执与上行 1. 通道回执接入后更新发送记录状态。 2. 回执状态至少包括提交成功、提交失败、发送成功、发送失败、未知、超时。 @@ -182,7 +246,7 @@ 9. 完全匹配不到的上行短信仍需入库，并展示为未匹配。 10. 客户端可查看本企业上行短信，运营端可查看全平台上行短信。 -### 4.9 账户计费 +### 4.10 账户计费 1. 客户端可查看充值套餐、购买或申请充值套餐、查看账单流水。 2. 发送创建时按短信内容计费条数、企业应用客户单价或套餐规则生成预估费用，计费条数只按 70/67 字规则拆分；不按移动、联通、电信配置不同客户价。 @@ -236,6 +300,7 @@ - 支持新增、编辑、启用、停用应用。 - 支持配置回调地址、IP 白名单、应用密钥、日发送限额。 +- 支持配置发送队列等级：普通队列、优先队列；默认普通队列，变更后只影响新创建或新入队的发送消息。 - 支持配置“不符合模板的短信”处理策略：拒绝发送、人工审核、直接发送。 - 支持配置应用级风控阈值：单任务最大号码数、重复号码比例、非法号码比例、黑名单命中比例、非工作时间大批量发送策略、短时间任务创建频控。 - 默认阈值：重复号码比例 30%、非法号码比例 10%、黑名单命中比例 5%、同一企业/应用 10 分钟内最多创建 10 次任务。 @@ -279,6 +344,8 @@ - 企业应用列表展示 CMPP 连接数；点击连接数打开连接详情弹窗，内容包含连接 id、状态、连接建立时间、最近心跳时间、上次提交时间、窗口占用等。 - 企业应用连接详情支持删除连接；删除连接必须调用真实后端接口或 Gateway 回写接口，并写入系统日志。 - 企业应用列表提供 CMPP 连接参数查看与一键复制能力，参数来源于真实应用/通道配置，不允许只在前端拼接假数据。 +- 企业应用新增/编辑必须展示发送队列等级，并真实保存普通队列或优先队列配置；列表或详情应能看到该配置，便于运营核对高优先级应用。 +- 企业应用新增时选择企业必须使用通用下拉控件和真实企业接口；下拉控件的视觉、尺寸、禁用态、错误态应与系统内其他 Select 保持一致。 ### 5.12 运营端审核 @@ -432,6 +499,7 @@ - Go CMPP Gateway 作为独立服务部署，不和 NestJS 业务后台耦合。 - 第一版支持真实 CMPP 2.0/3.0 长连接能力，后续可扩展 SGIP、SMGP、HTTP 通道。 - 网关负责通道连接、登录认证、长连接保活、submit、deliver、active test、重连、滑动窗口、sequence 管理、回执与上行接收。 +- 网关必须同时覆盖上游通道客户端能力和下游客户服务端能力；两类连接的账号、密码、IP 白名单、连接数、状态回写和消息映射必须隔离管理。 - NestJS 业务后台负责企业、应用、模板、签名、报备、审核、计费、风控、路由决策和发送任务编排。 - NestJS 与 Go Gateway 之间通过 Redis/BullMQ 队列或内部 gRPC/HTTP 通信；第一版建议使用队列提交发送指令、队列回传提交结果和回执事件。 - Go Gateway 不直接承担业务审核、账务扣费、签名报备判断，只执行已被业务后台路由后的通道提交。 @@ -483,24 +551,27 @@ 1. API 创建批量任务。 2. 拆分号码和变量，生成 message_record。 -3. 写入数据库并投递 send.queue。 -4. Send Worker 消费消息，校验黑名单、敏感词、限额。 -5. 使用可配置号码前缀正则识别运营商；识别不出时按移动处理。 -6. 使用手机号段库识别号码省份和城市；识别不出省份时按全国路由处理。 -7. 按企业应用绑定的对应运营商通道组执行路由：通道组只能是移动、联通、电信之一，发送时必须同时满足路由规则运营商、通道组运营商、通道组明细 carrier 与号码识别运营商一致；再校验通道本体 carrier 为对应运营商或三网；最后先匹配省网通道，再匹配全国通道，不得直接绑定或 fallback 到非授权单通道。 -8. 过滤业务 disabled、连接离线、认证失败、心跳超时或无可用连接数的通道。 -9. 通过通道限速器控制 TPS。 -10. 调用 Gateway Adapter 提交短信。 -11. 写入 submit 状态。 -12. Submit rejected、submit timeout、Gateway 连接断开或未提交成功、receipt failed 等场景按通道组策略补发到下一可用全国通道；unknown、超过 72 小时、超过通道组补发时间上限或关闭补发时不再补发。 -13. Receipt Worker 接收回执并更新最终状态。 -14. 补发过程必须保持幂等、计费一致和 trace 可查，最终成功只扣一次客户费率。 -15. 统计任务进度。 +3. 写入数据库并准备投递发送队列。 +4. 根据短信应用的队列等级写入 priority_send.queue 或 normal_send.queue，或在同一 BullMQ 队列中写入可验证的 priority 值；数据库中的 message_record/submit trace 必须保留队列等级。 +5. Send Worker 优先消费优先队列，再消费普通队列；多实例部署时必须避免普通队列持续抢占导致优先队列失效。 +6. Send Worker 消费消息，校验黑名单、敏感词、限额。 +7. 使用可配置号码前缀正则识别运营商；识别不出时按移动处理。 +8. 使用手机号段库识别号码省份和城市；识别不出省份时按全国路由处理。 +9. 按企业应用绑定的对应运营商通道组执行路由：通道组只能是移动、联通、电信之一，发送时必须同时满足路由规则运营商、通道组运营商、通道组明细 carrier 与号码识别运营商一致；再校验通道本体 carrier 为对应运营商或三网；最后先匹配省网通道，再匹配全国通道，不得直接绑定或 fallback 到非授权单通道。 +10. 过滤业务 disabled、连接离线、认证失败、心跳超时或无可用连接数的通道。 +11. 通过通道限速器控制 TPS；优先队列不得突破通道配置的供应商 TPS 和连接窗口上限。 +12. 调用 Gateway Adapter 提交短信。 +13. 写入 submit 状态。 +14. Submit rejected、submit timeout、Gateway 连接断开或未提交成功、receipt failed 等场景按通道组策略补发到下一可用全国通道；unknown、超过 72 小时、超过通道组补发时间上限或关闭补发时不再补发。 +15. Receipt Worker 接收回执并更新最终状态。 +16. 补发过程必须保持幂等、计费一致和 trace 可查，最终成功只扣一次客户费率。 +17. 统计任务进度。 ### 8.3 500 条/秒实现建议 - 创建任务与发送解耦，提交接口只负责入库和投递队列。 - Send Worker 多实例部署，每实例处理固定通道分片或队列分片。 +- 发送调度必须支持优先队列和普通队列：优先队列消息在同等通道和限速条件下先被消费；普通队列不能被永久饿死，应通过批次配额、老化策略或可配置调度比例保证恢复。 - 使用 Redis Token Bucket 或本地令牌桶加 Redis 协调实现通道限速。 - 单条发送记录以 message_id 全局唯一，所有提交和回执处理幂等。 - 批量插入 message_record，避免逐条事务。 @@ -520,6 +591,7 @@ ### 9.2 短信配置 - sms_application：短信应用。 +- sms_application.queue_priority 或等价字段：应用发送队列等级，取值 normal/priority，默认 normal。 - sms_signature：短信签名。 - signature_material：签名证明材料。 - sms_template：短信模板。 @@ -1098,18 +1170,27 @@ 4. Send Worker。 5. 通道路由。 6. Redis 限速。 -7. Go Gateway 提交 CMPP。 -8. Submit Resp 处理。 -9. 回执处理。 -10. 上行短信处理。 -11. 72 小时未知转超时。 -12. 任务进度统计。 +7. 上游通道 CMPP connect/login、ActiveTest、断线重连和连接状态回写。 +8. Gateway 消费 SubmitCommand 并真实 submit 到上游通道。 +9. Submit Resp 解析、sequence/messageId 映射和提交状态回传。 +10. Deliver Receipt 解析、重复/迟到回执幂等回传。 +11. 普通 Deliver 上行解析和上行入库。 +12. 下游客户 CMPP Server 监听 `17890`。 +13. 下游客户 connect/login 鉴权、IP 白名单、应用状态和连接数校验。 +14. 下游客户 CMPP Submit 转平台发送请求，source=cmpp，不创建客户端批量任务。 +15. 平台最终回执和上行向下游客户连接投递。 +16. 72 小时未知转超时。 +17. 任务进度统计。 验收标准： - 平台创建的批量任务只记录客户端任务。 - 平台任务、API 调用、CMPP 对接发送全部按手机号维度进入短信记录。 - 支持 submit resp、deliver 回执、上行短信、超时补偿。 +- Gateway `/health` 只表示进程存活，不等于 CMPP 发送链路验收通过。 +- `17890` 必须是真实 CMPP Server 监听，能完成客户 connect/login 鉴权、IP 白名单校验和 submit 接入。 +- Gateway 必须真实消费 SubmitCommand 并向上游通道 submit，submit resp、receipt、uplink 事件必须进入 NestJS 后端闭环。 +- 客户 CMPP 接入发送不创建客户端批量任务，但所有号码必须进入短信记录并可追踪客户侧 sequence/msgId 与平台 messageId。 ### 阶段 8：查询、统计、验收 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 28ca290..5427f72 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -16,7 +16,7 @@ | --- | --- | | 租户 | `tenant-a` 正常企业，`tenant-b` 用于租户隔离验证。 | | 用户 | 客户端企业管理员、客户端普通用户、运营管理员、运营审核员。 | -| 短信应用 | `app-a`，状态 active，日限额开启，模板不匹配策略分别准备 reject/manual_review/allow 三组。 | +| 短信应用 | `app-a`，状态 active，日限额开启，模板不匹配策略分别准备 reject/manual_review/allow 三组；另准备普通队列应用 `app-normal` 和优先队列应用 `app-priority`。 | | 签名 | `【测试签名】`，审核通过且报备通过；另准备待审核、驳回、报备失败签名。 | | 模板 | 验证码模板 `验证码为 ${code}`，营销模板 `尊敬的${name}，优惠活动开始`，分别准备草稿、待审核、通过、驳回。 | | 通道 | active CMPP 通道、disabled 通道、备用通道；通道组包含主备优先级。 | @@ -34,6 +34,7 @@ | 企业认证闭环 | 客户提交资料、运营审核、客户查看状态、驳回重提、通过后才允许使用受限发送能力、日志可追溯。 | | 客户管理闭环 | 客户创建、编辑、启停、认证、额度/余额、客户下应用/签名/模板/发送记录联动、租户隔离、日志可追溯。 | | 短信应用闭环 | 创建、启停、删除、密钥重置、IP 白名单、模板不匹配策略、变更后对发送实时生效、日志可追溯。 | +| 应用队列等级闭环 | 客户端和运营端创建/编辑应用时保存普通/优先队列配置；发送入队按应用队列等级分流；优先队列在不绕过业务校验和通道限速的前提下插队消费。 | | 签名闭环 | 创建、材料、审核、删除、报备状态变化、发送可用性校验、历史任务不受错误覆盖、日志可追溯。 | | 引流信息闭环 | 字段配置、客户填写、删除字段或删除客户引流信息、报备材料影响、发送阻断或审核原因可见。 | | 模板闭环 | 单变量、多变量、审核、删除、变量缺失/多传/格式异常、计费预估、发送可用性校验。 | @@ -65,12 +66,12 @@ - 优先级：P1 - 前置条件：企业管理员已登录。 - 步骤： - 1. 创建短信应用，填写名称、场景、回调地址、IP 白名单、日限额。 + 1. 创建短信应用，填写名称、场景、回调地址、IP 白名单、日限额，并选择发送队列等级。 2. 查询应用列表和详情。 3. 执行应用密钥重置。 - 预期结果： - 应用创建成功，状态为 active。 - - IP 白名单、日限额、模板不匹配策略保存正确。 + - IP 白名单、日限额、模板不匹配策略和发送队列等级保存正确，刷新后仍来自真实 API/数据库。 - 密钥重置后旧密钥不可见，新密钥或密钥摘要更新。 - 生成操作日志。 @@ -470,6 +471,23 @@ - 删除连接调用真实后端或 Gateway 接口，连接状态刷新并写系统日志。 - CMPP 参数来源于真实应用/通道配置，一键复制内容与 API 返回一致。 +### TC-ADMIN-018A 企业应用新增表单企业选择与队列等级 + +- 优先级：P0 +- 前置条件：存在至少两个真实企业，运营管理员已登录，通道组基础数据可用。 +- 步骤： + 1. 打开运营端企业应用管理，点击新增短信应用。 + 2. 在第一步选择企业下拉框中查看企业选项、加载态和空态。 + 3. 选择企业后进入应用参数表单。 + 4. 配置应用名称、客户单价、IP 白名单、发送队列等级、移动/联通/电信通道组后保存。 + 5. 刷新列表并打开编辑页。 +- 预期结果： + - 企业选择使用项目通用 Select/下拉控件，样式、禁用态、错误态与系统其他下拉一致。 + - 企业选项来自真实企业 API，不使用静态数组、mock 或 localStorage。 + - 请求体包含 tenantId、queuePriority、客户单价、IP 白名单和通道组绑定。 + - 后端真实保存应用队列等级，刷新列表和编辑页后仍显示正确。 + - 不选择任何通道组或缺少必填字段时不能保存，并显示可读提示。 + ### TC-ADMIN-019 通道连接日志展示 - 优先级：P1 @@ -960,6 +978,22 @@ - 第二条记录历史或被幂等处理。 - 不重复扣费或重复变更最终状态。 +### TC-SEND-021 优先队列插队发送 + +- 优先级：P0 +- 前置条件：存在普通队列应用 `app-normal` 和优先队列应用 `app-priority`；两者企业状态、应用状态、签名、模板、报备、余额、通道组、通道连接均可用；Redis/BullMQ 使用真实本地服务。 +- 步骤： + 1. 使用 `app-normal` 连续提交一批普通队列短信，使普通队列形成可观察积压。 + 2. 在普通队列积压未清空时，使用 `app-priority` 提交少量优先队列短信。 + 3. 观察数据库 message_record、BullMQ job、Send Worker 日志、submit trace 和 Gateway SubmitCommand 顺序。 + 4. 等待两类消息处理完成。 +- 预期结果： + - 普通队列和优先队列消息均写入真实数据库，记录包含 queuePriority 或等价可追踪字段。 + - 优先队列消息在普通队列已有积压时更早被 Send Worker 消费并提交 Gateway。 + - 优先队列不绕过模板/签名/报备/余额/风控/通道组路由/通道限速/Gateway 连接校验。 + - 普通队列后续能继续恢复消费，不出现永久积压或重复提交。 + - trace 中可区分队列等级、入队时间、消费时间和 submit 顺序。 + ## 10. 契约与队列用例 ### TC-CONTRACT-001 SubmitCommand Schema 校验 @@ -986,6 +1020,16 @@ - 步骤：运行 `npm run spike:contracts`。 - 预期结果：`uplink-event.json` 通过 schema 校验。 +### TC-CONTRACT-005 SubmitCommand 队列等级字段校验 + +- 优先级：P0 +- 前置条件：SubmitCommand 契约已扩展 queuePriority、priority 或 queueName 字段。 +- 步骤：运行 `npm run spike:contracts`。 +- 预期结果： + - SubmitCommand 示例覆盖普通队列和优先队列。 + - schema 明确允许的队列等级取值，不接受未知等级。 + - NestJS 生产者和 Gateway/Worker 消费者使用同一字段语义。 + ## 11. 端到端 Smoke 用例 ### TC-E2E-001 核心发送闭环 @@ -1083,6 +1127,19 @@ - 无重复最终状态。 - 错误率满足阶段 0/8 性能报告要求。 +### TC-PERF-003 混合优先级队列 Smoke + +- 优先级：P0 +- 前置条件：Redis/BullMQ 可用，准备普通队列和优先队列应用。 +- 步骤： + 1. 先提交一批普通队列消息制造积压。 + 2. 再提交优先队列消息。 + 3. 持续提交少量优先队列消息，同时观察普通队列恢复。 +- 预期结果： + - 优先队列消息消费延迟明显低于已有普通队列积压。 + - 普通队列不会被永久饿死。 + - 端到端吞吐仍满足第一版性能 smoke 口径。 + ## 14. 业务闭环补充用例 ### TC-CERT-001 企业认证资料提交 diff --git a/docs/testing-plan.md b/docs/testing-plan.md index 2d1584c..21fd27a 100644 --- a/docs/testing-plan.md +++ b/docs/testing-plan.md @@ -16,7 +16,7 @@ - 当前重点： - 风控规则评估：最大号码数、重复率、非法号码率、黑名单率、模板变量异常、直接拒绝、进入人工审核。 - 计费：费用预估、余额检查、冻结、提交 accepted 扣费、失败回执退款、短信计费记录、应用级客户费率、重复回调幂等。 - - 发送链路：批量任务创建、手机号拆分、发送入队、运营商前缀正则分流、手机号段归属地识别、应用运营商通道组路由、省网/全国路由、submit result 更新、receipt 更新、迟到旧回执不覆盖最终成功、失败补发、补发停止条件、uplink 记录、72 小时未知转超时。 + - 发送链路：批量任务创建、手机号拆分、按应用队列等级分流普通/优先队列、优先队列插队消费、普通队列防饿死、运营商前缀正则分流、手机号段归属地识别、应用运营商通道组路由、省网/全国路由、submit result 更新、receipt 更新、迟到旧回执不覆盖最终成功、失败补发、补发停止条件、uplink 记录、72 小时未知转超时。 - 通道与报备：通道创建、通道发送地区、三网通道通配、单运营商通道组、通道组明细 carrier 参与发送、省份与通道发送地区一致性、全国通道优先级唯一、企业应用通道组保存校验、禁止单通道发送规则、最终选中通道签名报备校验、签名报备任务、导出、回执导入、签名状态同步。 - 查询统计：发送链路 trace、对账 reconciliation、dashboard/statistics。 - Gateway：SEQID/MSGID 追踪、重连、health、gocmpp submit/resp 模拟器。 @@ -26,7 +26,7 @@ - 单元/轻集成测试可以使用 mock Prisma/BullMQ/Redis 验证 Service 编排和异常分支，但这只代表代码级测试通过。 - 系统功能集成测试必须补充真实服务闭环： - Prisma/PostgreSQL test database 集成测试。 - - BullMQ + Redis 队列消费集成测试。 + - BullMQ + Redis 队列消费集成测试，覆盖普通队列、优先队列、混合积压、优先队列插队和普通队列恢复。 - MinIO 预签名上传集成测试。 - 前端页面调用真实 API 的浏览器 smoke。 @@ -35,6 +35,7 @@ - 继续复用 `docs/contracts/gateway-queue-messages.schema.json`。 - 继续使用 `tools/spike/validate-gateway-queue-contract.mjs` 校验 SubmitCommand、SubmitResult、ReceiptEvent、UplinkEvent 示例。 - 队列消息变更必须先改 schema 和示例，再改 NestJS/Gateway 实现。 +- 若队列消息新增 queuePriority、priority 或队列名称字段，必须同步更新契约 schema、示例、NestJS 生产者、Send Worker 消费者和 Gateway 消费入口。 ### 2.4 端到端 Smoke @@ -52,6 +53,7 @@ - 继续复用 `npm run spike:bullmq`。 - 验证 15000 条消息、并发 500 的入队和端到端队列链路吞吐。 +- 混合优先级压测必须额外验证：普通队列已有积压时，新进入的优先队列消息能更早被消费；优先队列持续进入时，普通队列仍能按配置恢复消费，不出现永久饥饿。 - 性能 smoke 只验证第一版“可稳定入队并调度 500 条短信/秒”的链路能力，不替代生产压测。 ## 3. 执行命令 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 67f64c4..402ae70 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -1,5 +1,22 @@ # 第一版系统化测试进度 +## 2026-07-07 企业应用通用下拉与优先队列需求补充 + +- 已补充需求文档，明确运营端企业应用新增时选择企业必须使用通用 Select/下拉控件，企业选项来自真实企业 API，支持加载、空态和错误态，不允许静态数组或 localStorage 兜底。 +- 已补充应用发送队列等级需求：短信应用必须保存普通队列/优先队列配置，默认普通队列；优先队列用于验证码、登录确认、交易通知等高时效短信。 +- 已补充发送链路要求：发送入队必须按应用队列等级分流，优先队列在同等业务校验、通道组路由、通道限速和 Gateway 连接条件下插队消费；普通队列不能永久饿死。 +- 已补充测试计划和系统功能用例： + - 企业应用新增表单企业选择与队列等级真实保存。 + - 优先队列插队发送。 + - SubmitCommand 队列等级契约校验。 + - 混合优先级队列性能 smoke。 +- 当前状态：仅完成需求和测试口径补充，前后端、Prisma、BullMQ/Send Worker、Gateway 队列契约尚未实现，不能记为功能验收通过。 +- 2026-07-07 追加：Prisma 与 NestJS 企业应用 API 已增加 `queuePriority` 持久化字段，创建/编辑支持 normal/priority 校验，列表/详情随真实应用数据返回；前端表单、发送入队、Send Worker 调度和 Gateway 队列契约仍待后续步骤实现。 +- 2026-07-07 追加：运营端企业应用新增第一步企业选择已改为通用 `Select`；企业应用新增/编辑表单已按设计锚点恢复“发送队列”单选项，并在保存时真实提交 `queuePriority`。发送入队、Send Worker 调度和 Gateway 队列契约仍待后续步骤实现。 +- 2026-07-07 追加：发送链路已将应用 `queuePriority` 固化到 `SmsMessageRecord`，BullMQ 入队按 priority/normal 写入不同 job priority，SubmitCommand 契约、示例和 Go Gateway 队列结构已增加 `queuePriority`；当前实现覆盖优先队列插队的基础能力，持续高优先级流量下普通队列防饥饿策略仍需后续压测和调度增强。 +- 2026-07-07 追加：Gateway 队列契约第 8 步已独立校验，`SubmitCommand` schema/example 要求 `queuePriority`，Go Gateway `SubmitCommand` 结构可反序列化该字段，并通过 `npm run spike:contracts` 与 `npm run spike:gateway`。 +- 2026-07-07 追加：第 9 步收口验证通过：`npm --prefix api run prisma:generate`、`npm --prefix api test -- sms-config.service.spec.ts send-chain.service.spec.ts --runInBand`、`npm run spike:contracts`、`npm run spike:gateway`、`npm --prefix api run build`、`npm run build` 均通过；前端 build 仅保留既有 Vite chunk size warning。 + ## 2026-07-06 企业管理列表字段回归 - 按设计锚点 `131f344a^` 恢复运营端企业管理列表字段：企业 ID、企业名称、当前余额、透支限额、今日消费、企业状态、操作。 diff --git a/gateway/internal/queue/messages.go b/gateway/internal/queue/messages.go index 232af3d..50824d0 100644 --- a/gateway/internal/queue/messages.go +++ b/gateway/internal/queue/messages.go @@ -34,6 +34,7 @@ type SubmitCommand struct { Signature string `json:"signature"` TemplateID string `json:"templateId"` BillingUnits int `json:"billingUnits"` + QueuePriority string `json:"queuePriority"` Route Route `json:"route"` CMPP CMPP `json:"cmpp"` Retry Retry `json:"retry"` diff --git a/gateway/internal/queue/messages_test.go b/gateway/internal/queue/messages_test.go new file mode 100644 index 0000000..68b5d5a --- /dev/null +++ b/gateway/internal/queue/messages_test.go @@ -0,0 +1,50 @@ +package queue + +import ( + "encoding/json" + "testing" +) + +func TestSubmitCommandUnmarshalsQueuePriority(t *testing.T) { + payload := []byte(`{ + "schemaVersion": "v1", + "messageType": "SubmitCommand", + "traceId": "trace-queue-0001", + "messageId": "msg-queue-0001", + "channelId": "channel-1", + "createdAt": "2026-07-07T10:00:00Z", + "tenantId": "tenant-1", + "applicationId": "app-1", + "submitId": "submit-1", + "phoneNumber": "13800138000", + "content": "hello", + "signature": "测试", + "templateId": "tpl-1", + "billingUnits": 1, + "queuePriority": "priority", + "route": { + "channelCode": "CMPP-A", + "cmppAccountCode": "account-a", + "priority": 1, + "rateLimitPerSecond": 100 + }, + "cmpp": { + "serviceId": "SMS", + "srcId": "10690000", + "registeredDelivery": 1, + "msgFmt": 8 + }, + "retry": { + "attempt": 0, + "maxAttempts": 3 + } + }`) + + var command SubmitCommand + if err := json.Unmarshal(payload, &command); err != nil { + t.Fatalf("unmarshal submit command: %v", err) + } + if command.QueuePriority != "priority" { + t.Fatalf("QueuePriority = %q, want priority", command.QueuePriority) + } +} diff --git a/src/api/adminApi.ts b/src/api/adminApi.ts index e201c87..19a7db0 100644 --- a/src/api/adminApi.ts +++ b/src/api/adminApi.ts @@ -187,6 +187,7 @@ export type ClientSmsApplication = { name: string; scene?: string | null; customerUnitPrice?: number | null; + queuePriority?: 'normal' | 'priority' | string | null; status: string; dailyLimit?: number | null; createdAt?: string; @@ -455,6 +456,7 @@ export type EnterpriseApplication = { status: string; dailyLimit?: number | null; customerUnitPrice?: number | null; + queuePriority?: 'normal' | 'priority' | string | null; maxPhonesPerTask?: number | null; templateMismatchMode?: string | null; ipAllowlist?: Array<{ id: string; ipCidr: string; remark?: string | null }>; @@ -550,9 +552,9 @@ export const adminApi = { request(withQuery('/admin/enterprise-applications', query)), getEnterpriseApplication: (id: string) => request(`/admin/enterprise-applications/${id}`), - createEnterpriseApplication: (body: { tenantId: string; name: string; scene?: string; dailyLimit?: number; customerUnitPrice?: number; maxPhonesPerTask?: number; templateMismatchMode?: string; ipAllowlist?: string[] }) => + createEnterpriseApplication: (body: { tenantId: string; name: string; scene?: string; dailyLimit?: number; customerUnitPrice?: number; queuePriority?: 'normal' | 'priority'; maxPhonesPerTask?: number; templateMismatchMode?: string; ipAllowlist?: string[] }) => request('/admin/enterprise-applications', { method: 'POST', body: JSON.stringify(body) }), - updateEnterpriseApplication: (id: string, body: { name?: string; scene?: string; dailyLimit?: number; customerUnitPrice?: number; maxPhonesPerTask?: number; templateMismatchMode?: string; ipAllowlist?: string[] }) => + updateEnterpriseApplication: (id: string, body: { name?: string; scene?: string; dailyLimit?: number; customerUnitPrice?: number; queuePriority?: 'normal' | 'priority'; maxPhonesPerTask?: number; templateMismatchMode?: string; ipAllowlist?: string[] }) => request(`/admin/enterprise-applications/${id}`, { method: 'PUT', body: JSON.stringify(body) }), changeApplicationStatus: (id: string, status: string, reason?: string) => request(`/admin/enterprise-applications/${id}/status`, { diff --git a/src/apps/admin/AdminEnterpriseApplicationsPage.tsx b/src/apps/admin/AdminEnterpriseApplicationsPage.tsx index b129c38..fe7ec4c 100644 --- a/src/apps/admin/AdminEnterpriseApplicationsPage.tsx +++ b/src/apps/admin/AdminEnterpriseApplicationsPage.tsx @@ -1,7 +1,7 @@ import { useEffect, useMemo, useState } from 'react'; import { Copy, Edit3, Plus, Search, Settings2, Trash2 } from 'lucide-react'; import { useNavigate } from 'react-router-dom'; -import { Breadcrumb, Button, Input, Modal, Table, Tabs, Tag, type TableColumn } from '@/components/ui'; +import { Breadcrumb, Button, Input, Modal, Select, Table, Tabs, Tag, type TableColumn } from '@/components/ui'; import { adminApi, type ApplicationCmppParams, type CmppConnectionState, type EnterpriseApplication, type TenantOption } from '@/api/adminApi'; type SmsApp = { @@ -94,15 +94,17 @@ function AddApplicationModal({ title={

新建企业应用

先选择真实企业，再配置短信应用和三网通道组。

} >
- + setScene(event.target.value)} placeholder="行业通知/营销推广/验证码" value={scene} /> setDailyLimit(event.target.value)} placeholder="100000" required value={dailyLimit} /> setCustomerUnitPrice(event.target.value)} placeholder="0.0300" required value={customerUnitPrice} /> +
+ 发送队列 +
+ + +
+
+ + 优先队列会在发送调度中插队处理，但仍必须经过模板、签名、余额、通道组和通道限速校验。 +
+
setPhoneDailyLimit(event.target.value)} placeholder="10" required value={phoneDailyLimit} />