diff --git a/api/prisma/migrations/20260709112500_default_cmpp_version_20/migration.sql b/api/prisma/migrations/20260709112500_default_cmpp_version_20/migration.sql new file mode 100644 index 0000000..5ccef61 --- /dev/null +++ b/api/prisma/migrations/20260709112500_default_cmpp_version_20/migration.sql @@ -0,0 +1 @@ +ALTER TABLE "SmsChannel" ALTER COLUMN "cmppVersion" SET DEFAULT '2.0'; diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index 3a5150a..e35832b 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -490,7 +490,7 @@ model SmsChannel { account String passwordCipher String srcId String - cmppVersion String @default("3.0") + cmppVersion String @default("2.0") rateLimitPerSecond Int @default(100) unitPrice Int @default(0) status String @default("active") diff --git a/api/src/channels/channels.controller.ts b/api/src/channels/channels.controller.ts index be6f205..762e743 100644 --- a/api/src/channels/channels.controller.ts +++ b/api/src/channels/channels.controller.ts @@ -13,6 +13,7 @@ import { CreateReportMaterialDto, CreateReportTaskDto, CreateRouteRuleDto, + TestChannelDto, UpsertConnectionStateDto, UpdateChannelDto, UpdateChannelGroupDto, @@ -39,8 +40,8 @@ export class ChannelsController { } @Post('channels/:id/test') - testChannel(@Param('id') channelId: string) { - return this.channels.testChannel(channelId); + testChannel(@Param('id') channelId: string, @Body() body: TestChannelDto) { + return this.channels.testChannel(channelId, body); } @Post('channels/:id/status') diff --git a/api/src/channels/channels.service.spec.ts b/api/src/channels/channels.service.spec.ts index e21890b..c8625c0 100644 --- a/api/src/channels/channels.service.spec.ts +++ b/api/src/channels/channels.service.spec.ts @@ -7,6 +7,8 @@ const mockFetch = jest.fn().mockResolvedValue({ status: 200, text: jest.fn().mockResolvedValue(''), }); +const mockRedisXadd = jest.fn().mockResolvedValue('1710000000000-0'); +const mockRedisDisconnect = jest.fn(); jest.mock('bullmq', () => ({ Queue: jest.fn().mockImplementation(() => ({ @@ -15,6 +17,11 @@ jest.mock('bullmq', () => ({ })), })); +jest.mock('ioredis', () => jest.fn().mockImplementation(() => ({ + xadd: mockRedisXadd, + disconnect: mockRedisDisconnect, +}))); + function createPrismaMock() { const reportTask = { id: 'report-task-1', tenantId: 'tenant-1', signatureId: 'sig-1', channelId: 'channel-1', status: 'pending' }; const channel = { @@ -29,12 +36,13 @@ function createPrismaMock() { account: 'sp', passwordCipher: 'secret', srcId: '10690000', - cmppVersion: '3.0', + cmppVersion: '2.0', rateLimitPerSecond: 100, unitPrice: 3, status: 'active', config: { serviceId: 'SMS' }, sendRegion: '山东', + connectionStates: [{ id: 'state-1', connectionId: 'conn-a', status: 'connected', currentConnections: 1 }], reportFields: [{ code: 'license', name: '营业执照', fieldType: 'file', required: true, description: null, sortOrder: 1, status: 'active' }], }; return { @@ -113,6 +121,22 @@ function createPrismaMock() { smsApplication: { findUnique: jest.fn().mockResolvedValue({ id: 'app-1', tenantId: 'tenant-1' }), }, + tenant: { + findFirst: jest.fn().mockResolvedValue({ id: 'tenant-1', createdAt: new Date('2026-07-09T00:00:00.000Z') }), + }, + smsBatchTask: { + create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'batch-1', ...data })), + update: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'batch-1', ...data })), + }, + cmppSubmitSession: { + upsert: jest.fn().mockResolvedValue({ id: 'session-1', channelId: 'channel-1', sessionNo: 'OPEN-channel-1' }), + }, + smsMessageRecord: { + create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'message-record-1', ...data })), + }, + smsSubmitRecord: { + create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'submit-record-1', ...data })), + }, cmppConnectionState: { findMany: jest.fn(), findFirst: jest.fn().mockResolvedValue(null), @@ -132,6 +156,8 @@ describe('ChannelsService', () => { mockQueueAdd.mockClear(); mockQueueClose.mockClear(); mockFetch.mockClear(); + mockRedisXadd.mockClear(); + mockRedisDisconnect.mockClear(); global.fetch = mockFetch as never; }); @@ -164,7 +190,7 @@ describe('ChannelsService', () => { expect(prisma.smsChannel.create).toHaveBeenCalledWith({ data: expect.objectContaining({ protocol: 'CMPP', - cmppVersion: '3.0', + cmppVersion: '2.0', rateLimitPerSecond: 100, sendRegion: '全国', status: 'active', @@ -185,6 +211,7 @@ describe('ChannelsService', () => { channelId: 'channel-1', connectionId: 'channel-1:primary', reason: 'channel_created', + channel: expect.objectContaining({ cmppVersion: '2.0' }), }), { jobId: 'channel-1:primary:connect' }); expect(mockFetch).toHaveBeenCalledWith('http://127.0.0.1:8090/connections/connect', expect.objectContaining({ method: 'POST', @@ -206,6 +233,36 @@ describe('ChannelsService', () => { }); }); + it('preserves explicit CMPP 3.0 and rejects unsupported CMPP versions', async () => { + const prisma = createPrismaMock(); + const service = new ChannelsService(prisma as never); + + await service.createChannel({ + code: 'CMPP-3', + name: '3.0通道', + gatewayHost: '127.0.0.1', + gatewayPort: 17890, + account: 'sp', + passwordCipher: 'secret', + srcId: '10690000', + cmppVersion: '3.0', + }); + + expect(prisma.smsChannel.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ cmppVersion: '3.0' }), + }); + await expect(service.createChannel({ + code: 'BAD', + name: '非法版本', + gatewayHost: '127.0.0.1', + gatewayPort: 17890, + account: 'sp', + passwordCipher: 'secret', + srcId: '10690000', + cmppVersion: '1.0', + })).rejects.toThrow('cmppVersion must be 2.0 or 3.0'); + }); + it('updates CMPP channel configuration without requiring password changes', async () => { const prisma = createPrismaMock(); const service = new ChannelsService(prisma as never); @@ -525,6 +582,89 @@ describe('ChannelsService', () => { expect(prisma.smsChannel.update).toHaveBeenCalledWith({ where: { id: 'channel-1' }, data: { status: 'deleted' } }); }); + it('queues channel test SMS through real message records and gateway stream', async () => { + const prisma = createPrismaMock(); + const service = new ChannelsService(prisma as never); + + const result = await service.testChannel('channel-1', { + phoneNumber: '18821203795', + content: '【安徽航天信息】您的验证码是070926,有效时间30分钟。', + operatorId: 'admin-1', + }); + + expect(result).toEqual(expect.objectContaining({ + channelId: 'channel-1', + status: 'submit_queued', + submitted: 1, + batchTaskId: 'batch-1', + })); + expect(prisma.smsBatchTask.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + tenantId: 'tenant-1', + sourceType: 'admin_channel_test', + phoneTotal: 1, + status: 'submit_queued', + }), + }); + expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + tenantId: 'tenant-1', + batchTaskId: 'batch-1', + phoneNumber: '18821203795', + channelId: 'channel-1', + status: 'submit_queued', + submitStatus: 'queued', + }), + }); + expect(prisma.smsSubmitRecord.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + tenantId: 'tenant-1', + batchTaskId: 'batch-1', + channelId: 'channel-1', + sessionId: 'session-1', + submitStatus: 'queued', + }), + }); + expect(mockQueueAdd).toHaveBeenCalledWith('submit-command', expect.objectContaining({ + messageType: 'SubmitCommand', + channelId: 'channel-1', + phoneNumber: '18821203795', + content: '【安徽航天信息】您的验证码是070926,有效时间30分钟。', + upstream: expect.objectContaining({ cmppVersion: '2.0', gatewayHost: '127.0.0.1' }), + })); + expect(mockRedisXadd).toHaveBeenCalledWith( + 'gateway.submit.commands', + '*', + 'messageType', + 'SubmitCommand', + 'data', + expect.stringContaining('"phoneNumber":"18821203795"'), + ); + expect(prisma.operationLog.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + action: 'sms_channel.test_submit', + resource: 'sms_channel', + resourceId: 'channel-1', + }), + }); + }); + + it('rejects channel test SMS when no CMPP connection is online', async () => { + const prisma = createPrismaMock(); + prisma.smsChannel.findUnique.mockResolvedValueOnce({ + ...(await prisma.smsChannel.findUnique()), + connectionStates: [{ connectionId: 'conn-a', status: 'failed', currentConnections: 0 }], + }); + const service = new ChannelsService(prisma as never); + + await expect(service.testChannel('channel-1', { + phoneNumber: '18821203795', + content: '测试短信', + })).rejects.toThrow('通道当前没有可用 CMPP 连接'); + expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled(); + expect(mockRedisXadd).not.toHaveBeenCalled(); + }); + it('upserts and lists CMPP connection states', async () => { const prisma = createPrismaMock(); const service = new ChannelsService(prisma as never); @@ -546,7 +686,7 @@ describe('ChannelsService', () => { where: { applicationId: 'app-1', channelId: 'channel-1', connectionId: 'conn-a' }, }); expect(prisma.cmppConnectionState.create).toHaveBeenCalledWith({ - data: expect.objectContaining({ tenantId: 'tenant-1', applicationId: 'app-1', channelId: 'channel-1', connectionId: 'conn-a', status: 'connected' }), + data: expect.objectContaining({ tenantId: 'tenant-1', applicationId: 'app-1', channelId: 'channel-1', connectionId: 'conn-a', status: 'connected', lastError: null }), }); expect(prisma.cmppConnectionState.findMany).toHaveBeenCalledWith({ where: { channelId: 'channel-1' }, diff --git a/api/src/channels/channels.service.ts b/api/src/channels/channels.service.ts index 94274b6..a6c79be 100644 --- a/api/src/channels/channels.service.ts +++ b/api/src/channels/channels.service.ts @@ -1,5 +1,6 @@ import { BadRequestException, Injectable, Logger, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; import { Queue } from 'bullmq'; +import IORedis from 'ioredis'; import { Prisma } from '@prisma/client'; import { randomUUID } from 'crypto'; import { PrismaService } from '../prisma/prisma.service'; @@ -144,17 +145,30 @@ export interface CopyChannelDto { operatorId?: string; } +export interface TestChannelDto { + phoneNumber?: string; + phones?: string[] | string; + content?: string; + accessNo?: string; + operatorId?: string; +} + const GATEWAY_CONNECTION_QUEUE = 'gateway.connection.commands'; +const GATEWAY_SUBMIT_QUEUE = 'gateway.submit.queue'; +const GATEWAY_SUBMIT_STREAM = 'gateway.submit.commands'; const DEFAULT_GATEWAY_CONTROL_URL = 'http://127.0.0.1:8090'; const DEFAULT_CHANNEL_CONNECTION_ID = 'primary'; const DEFAULT_CONNECTING_TIMEOUT_MS = 30_000; const DEFAULT_CONNECTING_TIMEOUT_SCAN_MS = 5_000; const CONNECTING_TIMEOUT_ERROR = 'Gateway connection request timed out'; +const DEFAULT_CMPP_VERSION = '2.0'; @Injectable() export class ChannelsService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(ChannelsService.name); private gatewayConnectionQueue?: Queue; + private gatewaySubmitQueue?: Queue; + private redis?: IORedis; private connectionTimeoutTimer?: ReturnType; constructor(private readonly prisma: PrismaService) {} @@ -176,6 +190,8 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { clearInterval(this.connectionTimeoutTimer); } await this.gatewayConnectionQueue?.close(); + await this.gatewaySubmitQueue?.close(); + this.redis?.disconnect(); } listChannels() { @@ -198,6 +214,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { if (!Number.isInteger(gatewayPort) || gatewayPort <= 0 || gatewayPort > 65535) { throw new BadRequestException('gatewayPort must be an integer between 1 and 65535'); } + const cmppVersion = normalizeCmppVersion(data.cmppVersion); const config = normalizeChannelRuntimeConfig(data.config, data.desiredConnections, data.windowSize); const channel = await this.prisma.smsChannel.create({ data: { @@ -212,7 +229,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { account: data.account, passwordCipher: data.passwordCipher, srcId: data.srcId, - cmppVersion: data.cmppVersion ?? '3.0', + cmppVersion, rateLimitPerSecond: data.rateLimitPerSecond ?? 100, unitPrice: data.unitPrice ?? 0, status: data.status ?? 'active', @@ -234,6 +251,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { if (gatewayPort !== undefined && (!Number.isInteger(gatewayPort) || gatewayPort <= 0 || gatewayPort > 65535)) { throw new BadRequestException('gatewayPort must be an integer between 1 and 65535'); } + const cmppVersion = data.cmppVersion === undefined ? undefined : normalizeCmppVersion(data.cmppVersion); const config = data.config !== undefined || data.desiredConnections !== undefined || data.windowSize !== undefined ? normalizeChannelRuntimeConfig(channel.config, data.desiredConnections, data.windowSize) : undefined; @@ -251,7 +269,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { account: data.account, passwordCipher: data.passwordCipher, srcId: data.srcId, - cmppVersion: data.cmppVersion, + cmppVersion, rateLimitPerSecond: data.rateLimitPerSecond, unitPrice: data.unitPrice, status: data.status, @@ -394,11 +412,135 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { return this.changeChannelStatus(channelId, { ...data, status: 'deleted' }); } - testChannel(channelId: string) { + async testChannel(channelId: string, data: TestChannelDto = {}) { + const phoneNumbers = normalizeTestPhones(data); + const content = normalizeTestContent(data.content); + const channel = await this.prisma.smsChannel.findUnique({ + where: { id: channelId }, + include: { connectionStates: true }, + }); + if (!channel) { + throw new NotFoundException('Channel not found'); + } + if (channel.status !== 'active') { + throw new BadRequestException('通道未启用,不能发送测试短信'); + } + const connectedState = channel.connectionStates.find((state) => + normalizeGatewayConnectionStatus(state.status) === 'connected' && (state.currentConnections ?? 0) > 0, + ); + if (!connectedState) { + throw new BadRequestException('通道当前没有可用 CMPP 连接,请先连接成功后再测试发送'); + } + + const tenant = await this.prisma.tenant.findFirst({ orderBy: { createdAt: 'asc' } }); + if (!tenant) { + throw new BadRequestException('未找到可归属测试短信的客户租户'); + } + + const createdAt = new Date(); + const taskNo = `CHTEST-${Date.now()}-${randomUUID().slice(0, 8)}`; + const batchTask = await this.prisma.smsBatchTask.create({ + data: { + tenantId: tenant.id, + taskNo, + sourceType: 'admin_channel_test', + content, + category: 'channel_test', + phoneTotal: phoneNumbers.length, + status: 'submit_queued', + auditStatus: 'approved', + progressTotal: phoneNumbers.length, + submittedTotal: 0, + createdById: data.operatorId, + }, + }); + + const results = []; + for (const [index, phoneNumber] of phoneNumbers.entries()) { + const messageId = `MSG-TEST-${Date.now()}-${randomUUID().slice(0, 8)}`; + const submitId = `SUB-TEST-${Date.now()}-${randomUUID().slice(0, 8)}`; + 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 }, + }); + const messageRecord = await this.prisma.smsMessageRecord.create({ + data: { + tenantId: tenant.id, + batchTaskId: batchTask.id, + messageId, + phoneNumber, + content, + billingUnits: calculateBillingUnits(content), + unitPrice: channel.unitPrice, + amountCents: channel.unitPrice * calculateBillingUnits(content), + queuePriority: 'normal', + channelId: channel.id, + submitId, + status: 'submit_queued', + submitStatus: 'queued', + errorMessage: '运营端通道测试短信', + }, + }); + await this.prisma.smsSubmitRecord.create({ + data: { + tenantId: tenant.id, + batchTaskId: batchTask.id, + messageRecordId: messageRecord.id, + channelId: channel.id, + sessionId: session.id, + submitId, + submitStatus: 'queued', + }, + }); + const command = buildChannelTestSubmitCommand({ + channel, + content, + phoneNumber, + messageId, + submitId, + batchTaskId: batchTask.id, + tenantId: tenant.id, + attempt: index, + accessNo: data.accessNo, + }); + await this.getGatewaySubmitQueue().add('submit-command', command); + const streamMessageId = await this.publishGatewaySubmitCommand(command); + results.push({ + phoneNumber, + messageRecordId: messageRecord.id, + submitId, + streamMessageId, + }); + } + + await this.prisma.smsBatchTask.update({ + where: { id: batchTask.id }, + data: { submittedTotal: phoneNumbers.length }, + }); + await this.prisma.operationLog.create({ + data: { + userId: data.operatorId, + action: 'sms_channel.test_submit', + resource: 'sms_channel', + resourceId: channel.id, + detail: { + batchTaskId: batchTask.id, + phoneTotal: phoneNumbers.length, + messageRecordIds: results.map((item) => item.messageRecordId), + connectionId: connectedState.connectionId, + } as Prisma.InputJsonValue, + }, + }); + return { channelId, - status: 'queued', - message: 'Channel test request accepted as a phase-4 placeholder.', + status: 'submit_queued', + batchTaskId: batchTask.id, + taskNo: batchTask.taskNo, + submitted: results.length, + messages: results, + queuedAt: createdAt, }; } @@ -485,7 +627,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { lastDisconnectedAt: data.lastDisconnectedAt ? new Date(data.lastDisconnectedAt) : undefined, lastHeartbeatAt: data.lastHeartbeatAt ? new Date(data.lastHeartbeatAt) : undefined, reconnectCount: data.reconnectCount ?? 0, - lastError: data.lastError, + lastError: status === 'connected' ? null : data.lastError, }; const existing = await this.prisma.cmppConnectionState.findFirst({ where: { @@ -1038,6 +1180,31 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { return this.gatewayConnectionQueue; } + private getGatewaySubmitQueue() { + this.gatewaySubmitQueue ??= new Queue(GATEWAY_SUBMIT_QUEUE, { connection: bullmqConnection() }); + return this.gatewaySubmitQueue; + } + + private getRedis() { + if (!this.redis) { + this.redis = new IORedis(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', { + maxRetriesPerRequest: null, + }); + } + return this.redis; + } + + private async publishGatewaySubmitCommand(command: unknown) { + return this.getRedis().xadd( + process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM, + '*', + 'messageType', + 'SubmitCommand', + 'data', + JSON.stringify(command), + ); + } + private async notifyGatewayConnect(command: Record) { const baseUrl = (process.env.GATEWAY_CONTROL_URL ?? DEFAULT_GATEWAY_CONTROL_URL).replace(/\/+$/, ''); let response: { ok: boolean; status: number; text: () => Promise }; @@ -1057,6 +1224,131 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { } } +function normalizeTestPhones(data: TestChannelDto) { + const rawPhones = Array.isArray(data.phones) + ? data.phones + : String(data.phoneNumber ?? data.phones ?? '').split(/[,\n,\s]+/u); + const phones = rawPhones.map((phone) => String(phone).trim()).filter(Boolean); + const uniquePhones = Array.from(new Set(phones)); + if (uniquePhones.length === 0) { + throw new BadRequestException('请填写测试手机号'); + } + if (uniquePhones.length > 10) { + throw new BadRequestException('测试手机号最多允许 10 个'); + } + for (const phone of uniquePhones) { + if (!/^1[3-9]\d{9}$/.test(phone)) { + throw new BadRequestException(`手机号格式不正确:${phone}`); + } + } + return uniquePhones; +} + +function normalizeTestContent(content?: string) { + const normalized = (content ?? '').trim(); + if (!normalized) { + throw new BadRequestException('请填写测试短信内容'); + } + if (normalized.length > 1000) { + throw new BadRequestException('测试短信内容不能超过 1000 字符'); + } + return normalized; +} + +function calculateBillingUnits(content: string) { + return Math.max(1, Math.ceil([...content].length / 67)); +} + +function buildChannelTestSubmitCommand({ + channel, + content, + phoneNumber, + messageId, + submitId, + batchTaskId, + tenantId, + attempt, + accessNo, +}: { + channel: { + id: string; + code: string; + gatewayHost: string; + gatewayPort: number; + account: string; + passwordCipher: string; + srcId: string; + cmppVersion: string; + rateLimitPerSecond: number; + config?: Prisma.JsonValue | null; + }; + content: string; + phoneNumber: string; + messageId: string; + submitId: string; + batchTaskId: string; + tenantId: string; + attempt: number; + accessNo?: string; +}) { + const srcId = accessNo?.trim() ? `${channel.srcId}${accessNo.trim()}` : channel.srcId; + return { + schemaVersion: 'v1', + messageType: 'SubmitCommand', + traceId: randomUUID(), + messageId, + channelId: channel.id, + createdAt: new Date().toISOString(), + tenantId, + applicationId: 'admin-channel-test', + taskId: batchTaskId, + submitId, + queuePriority: 'normal', + phoneNumber, + content, + signature: 'CHANNEL_TEST', + templateId: 'admin-channel-test', + billingUnits: calculateBillingUnits(content), + route: { + channelCode: channel.code, + cmppAccountCode: channel.account, + priority: attempt, + rateLimitPerSecond: channel.rateLimitPerSecond, + }, + cmpp: { + serviceId: getStringConfigValue(channel.config, 'serviceId', 'SMS'), + srcId, + registeredDelivery: 1, + msgFmt: 8, + }, + upstream: { + gatewayHost: channel.gatewayHost, + gatewayPort: channel.gatewayPort, + account: channel.account, + passwordCipher: channel.passwordCipher, + cmppVersion: channel.cmppVersion, + desiredConnections: getPositiveRuntimeInteger(getConfigValue(channel.config, 'desiredConnections'), 1, 'desiredConnections'), + windowSize: getPositiveRuntimeInteger(getConfigValue(channel.config, 'windowSize'), 16, 'windowSize'), + }, + retry: { attempt: 0, maxAttempts: 1 }, + }; +} + +function getConfigValue(config: Prisma.JsonValue | null | undefined, key: string) { + if (config && typeof config === 'object' && !Array.isArray(config) && key in config) { + return config[key as keyof typeof config]; + } + return undefined; +} + +function getStringConfigValue(config: Prisma.JsonValue | null | undefined, key: string, fallback: string) { + const value = getConfigValue(config, key); + if (value === undefined || value === null || value === '') { + return fallback; + } + return String(value); +} + function normalizeConnectionAction(status: string) { const normalized = status.toLowerCase(); if (normalized === 'connected') { @@ -1077,6 +1369,14 @@ function normalizeConnectionAction(status: string) { return 'updated'; } +function normalizeCmppVersion(version?: string) { + const normalized = (version ?? DEFAULT_CMPP_VERSION).trim(); + if (normalized === '2.0' || normalized === '3.0') { + return normalized; + } + throw new BadRequestException('cmppVersion must be 2.0 or 3.0'); +} + function normalizeGatewayConnectionStatus(status: string) { const normalized = status.toLowerCase(); if (['online', 'open', 'connected'].includes(normalized)) { diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 779dcd7..2efc785 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -554,6 +554,36 @@ describe('SendChainService', () => { }); }); + it('updates admin channel test message status without business retry routing', async () => { + const { service, prisma, billing } = createService(); + prisma.smsBatchTask.findUnique.mockResolvedValue({ id: 'task-1', tenantId: 'tenant-1', sourceType: 'admin_channel_test' }); + + await service.handleSubmitResult({ + messageId: 'MSG-1', + channelId: 'channel-1', + submitId: 'SUB-1', + sequenceId: 7, + gatewayMessageId: 'GW-1', + submitStatus: 'timeout', + errorCode: 'SUBMIT_TIMEOUT', + errorMessage: 'context deadline exceeded', + submittedAt: '2026-07-09T03:44:15.445Z', + }); + + expect(prisma.channelRouteRule.findFirst).not.toHaveBeenCalled(); + expect(billing.release).not.toHaveBeenCalled(); + expect(prisma.smsMessageRecord.update).toHaveBeenCalledWith({ + where: { id: 'record-1' }, + data: expect.objectContaining({ + gatewayMessageId: 'GW-1', + submitStatus: 'timeout', + status: 'timeout', + errorCode: 'SUBMIT_TIMEOUT', + errorMessage: 'context deadline exceeded', + }), + }); + }); + it('releases reservation for rejected submit result and refunds failed receipts', async () => { const { service, prisma, billing } = createService(); prisma.smsBillingRecord.findFirst diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 9f1c4de..bfcf685 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -619,13 +619,18 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }, }); await this.recordSubmitSegments(message, data, submittedAt); + const batchTask = await this.prisma.smsBatchTask.findUnique({ + where: { id: message.batchTaskId }, + select: { sourceType: true }, + }); + const isAdminChannelTest = batchTask?.sourceType === 'admin_channel_test'; if (data.submitId && message.submitId && data.submitId !== message.submitId) { return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); } const status = data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed'; if (data.submitStatus === 'accepted') { await this.chargeAcceptedMessage(message); - } else { + } else if (!isAdminChannelTest) { const retried = await this.retryMessageIfAllowed(message, data.submitStatus === 'timeout' ? '提交超时补发' : '提交失败补发'); if (retried) { await this.refreshTaskProgress(message.batchTaskId); diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index f6d8348..92c9f62 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -187,6 +187,8 @@ #### 4.8.1 上游通道连接能力 1. Gateway 必须按运营端通道配置连接上游 SMSC,使用通道的 `gatewayHost/gatewayPort/account/passwordCipher/srcId/cmppVersion` 完成 CMPP 2.0/3.0 connect/login。 + - 运营端通道创建/编辑必须提供 CMPP 2.0/3.0 版本选项,默认 CMPP 2.0;保存后 Gateway 连接与 SubmitCommand 均必须使用真实保存的 `cmppVersion`。 + - 运营端通道“发送测试”必须走真实闭环:前端提交手机号和短信内容到 NestJS API,后端校验通道在线后创建 `SmsBatchTask`、`SmsMessageRecord`、`SmsSubmitRecord`,并向 Redis Stream `gateway.submit.commands` 写入真实 `SubmitCommand`;短信记录页面必须能查询到测试短信,不允许只返回占位成功或只关闭弹窗。 2. Gateway 必须校验上游 connect/login 返回码,区分 connected、auth_failed、connect_timeout、network_error、protocol_error 等状态,并回写 NestJS 真实连接状态。 3. Gateway 必须支持每个通道配置期望连接数,建立多条长连接,并按连接维度维护 currentConnections、lastConnectedAt、lastHeartbeatAt、lastError、reconnectCount。 - `desiredConnections`、`windowSize` 是平台对上游通道连接池和提交窗口的运行配置,必须通过运营端通道配置页面保存到真实后端;它们不是 CMPP 标准 PDU 字段,也不是 gocmpp 的原生配置字段。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index af3f63a..cc94f58 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -238,13 +238,17 @@ - 优先级:P0 - 前置条件:运营管理员已登录。 - 步骤: - 1. 创建 CMPP 通道,填写网关地址、端口、账号、密码密文、接入号、限速、期望连接数和提交窗口。 + 1. 创建 CMPP 通道,填写网关地址、端口、账号、密码密文、接入号、CMPP 版本、限速、期望连接数和提交窗口。 2. 查询通道列表。 - 3. 停用通道后创建发送任务。 + 3. 在在线通道上打开“短信测试”,填写真实手机号、短信内容和可选接入号后发送测试短信。 + 4. 查询短信记录和提交记录。 + 5. 停用通道后创建发送任务。 - 预期结果: - - 通道协议默认为 CMPP,版本默认为 3.0。 + - 通道协议默认为 CMPP,CMPP 版本默认 2.0,且可选择 2.0 或 3.0。 - 通道限速保存正确。 - 通道真实保存 `desiredConnections/windowSize`,后续 Gateway `ConnectChannel` 与 `SubmitCommand.upstream` 使用该配置。 + - 通道测试短信必须调用真实 NestJS API,创建 `SmsBatchTask`、`SmsMessageRecord`、`SmsSubmitRecord`,并向 Redis Stream `gateway.submit.commands` 写入 `SubmitCommand`;短信记录页面能查询到该测试短信。 + - 若通道未启用或没有在线 CMPP 连接,测试短信 API 返回明确错误,不能只在前端假提示成功。 - 停用通道不会被路由选中。 ### TC-ADMIN-004 通道组与路由规则 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 70af8db..5529688 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -1,5 +1,20 @@ # 第一版系统化测试进度 +## 2026-07-09 运营端通道测试短信闭环修复 + +- 生产验证发现运营端通道“短信测试”弹窗仅关闭页面,未调用后端;`POST /api/admin/channels/:id/test` 仍返回 phase-4 placeholder,不创建 `SmsMessageRecord/SmsSubmitRecord`,也不写入 Gateway SubmitCommand,因此短信记录页面无记录。 +- 已修复为真实链路:前端提交手机号、内容和可选接入号;NestJS 校验通道 active 且存在在线 CMPP 连接后,创建 `SmsBatchTask`、`SmsMessageRecord`、`SmsSubmitRecord` 和操作日志,并向 BullMQ `gateway.submit.queue` 与 Redis Stream `gateway.submit.commands` 写入真实 `SubmitCommand`。 +- 测试口径同步:`TC-ADMIN-003` 增加通道测试短信闭环要求,必须能从页面/API 发起真实测试短信,短信记录页面可查询到对应记录,Gateway submit worker 按通道真实 CMPP 配置消费发送。 +- 已执行:`npm --prefix api test -- channels.service.spec.ts --runInBand`、`npm --prefix api run build`、`npm run build`。待生产部署后用指定号码做一次真实发送验证,并回查 DB/短信记录。 + +## 2026-07-09 线上通道 CMPP 版本修复 + +- 线上生产验证发现 3 个赛邮行业通道配置均指向 `121.40.172.212:7890`,其中 2 个已触发 Gateway 真实连接并失败,`CmppConnectionState.lastError` 为 `packetWriter.ReadBytes error: ReadBytes reads 14 bytes, not equal to 16 we expected`。 +- 生产机到上游 `121.40.172.212:7890` TCP 可连接,失败不是 API/Gateway 服务不可用,也不是网络完全不通;结合上游确认参数为 CMPP 2.0,根因定位为通道创建默认 CMPP 3.0 且运营端没有版本选择入口。 +- 已修复通道创建/编辑:运营端新增 CMPP 2.0/3.0 版本选择,默认 2.0;NestJS 通道 API 默认 `cmppVersion=2.0`,并只允许 2.0 或 3.0;Prisma `SmsChannel.cmppVersion` 默认值同步改为 2.0。 +- 测试口径同步:`TC-ADMIN-003` 要求通道协议默认 CMPP,CMPP 版本默认 2.0,且可选择 2.0 或 3.0;Gateway 连接命令必须携带真实通道版本。 +- 待复测:部署迁移后将线上赛邮通道 `cmppVersion` 调整为 2.0,并重新触发 Gateway 连接验证。 + ## 2026-07-07 企业应用通用下拉与优先队列需求补充 - 已补充需求文档,明确运营端企业应用新增时选择企业必须使用通用 Select/下拉控件,企业选项来自真实企业 API,支持加载、空态和错误态,不允许静态数组或 localStorage 兜底。 diff --git a/src/api/adminApi.ts b/src/api/adminApi.ts index d7c319f..1a140cc 100644 --- a/src/api/adminApi.ts +++ b/src/api/adminApi.ts @@ -69,6 +69,7 @@ export type AdminChannel = { enterpriseCode?: string | null; account: string; srcId: string; + cmppVersion?: '2.0' | '3.0' | string | null; rateLimitPerSecond: number; unitPrice: number; status: string; @@ -89,6 +90,21 @@ export type ChannelConnectionLogResponse = { }>; }; +export type ChannelTestResponse = { + channelId: string; + status: string; + batchTaskId: string; + taskNo: string; + submitted: number; + messages: Array<{ + phoneNumber: string; + messageRecordId: string; + submitId: string; + streamMessageId?: string; + }>; + queuedAt: string; +}; + export type EnterpriseCertification = { id: string; tenantId: string; @@ -784,6 +800,8 @@ export const adminApi = { method: 'POST', body: JSON.stringify(body), }), + testChannel: (id: string, body: { phoneNumber?: string; phones?: string[] | string; content: string; accessNo?: string }) => + request(`/admin/channels/${id}/test`, { method: 'POST', body: JSON.stringify(body) }), changeChannelStatus: (id: string, status: string, reason?: string) => request(`/admin/channels/${id}/status`, { method: 'POST', body: JSON.stringify({ status, reason }), diff --git a/src/apps/admin/AdminChannelsPage.tsx b/src/apps/admin/AdminChannelsPage.tsx index cb54bd3..015435b 100644 --- a/src/apps/admin/AdminChannelsPage.tsx +++ b/src/apps/admin/AdminChannelsPage.tsx @@ -26,6 +26,7 @@ type SmsChannel = { corpCode: string; account: string; accessNo: string; + cmppVersion: '2.0' | '3.0'; desiredConnections: number; windowSize: number; passwordCipher?: string; @@ -68,6 +69,11 @@ const protocolOptions = [ { label: 'SGIP', value: 'SGIP' }, ]; +const cmppVersionOptions = [ + { label: 'CMPP 2.0', value: '2.0' }, + { label: 'CMPP 3.0', value: '3.0' }, +]; + const regionOptions = [ { label: '全国', value: '全国' }, ...'北京,天津,河北,山西,内蒙古,辽宁,吉林,黑龙江,上海,江苏,浙江,安徽,福建,江西,山东,河南,湖北,湖南,广东,广西,海南,重庆,四川,贵州,云南,西藏,陕西,甘肃,青海,宁夏,新疆,香港,澳门,台湾'.split(',').map((province) => ({ label: province, value: province })), @@ -145,6 +151,7 @@ function mapApiChannel(channel: AdminChannel, connections: CmppConnectionState[] corpCode: channel.enterpriseCode ?? channel.code, account: channel.account, accessNo: channel.srcId, + cmppVersion: channel.cmppVersion === '3.0' ? '3.0' : '2.0', desiredConnections: Number(channel.config?.desiredConnections ?? 1), windowSize: Number(channel.config?.windowSize ?? 16), }; @@ -165,6 +172,7 @@ function buildChannelPayload(channel: SmsChannel, passwordCipher?: string) { account: channel.account, passwordCipher: passwordCipher || undefined, srcId: channel.accessNo, + cmppVersion: channel.cmppVersion, rateLimitPerSecond: 100, unitPrice: Math.round(channel.unitPrice), desiredConnections: channel.desiredConnections, @@ -201,6 +209,7 @@ function ChannelFormModal({ const [gatewayPort, setGatewayPort] = useState(channel?.gatewayPort ?? '17890'); const [corpCode, setCorpCode] = useState(channel?.corpCode ?? ''); const [account, setAccount] = useState(channel?.account ?? ''); + const [cmppVersion, setCmppVersion] = useState<'2.0' | '3.0'>(channel?.cmppVersion ?? '2.0'); const [password, setPassword] = useState(''); const [accessNo, setAccessNo] = useState(channel?.accessNo ?? ''); const [extensionDigits, setExtensionDigits] = useState('0'); @@ -228,6 +237,7 @@ function ChannelFormModal({ corpCode, account, accessNo, + cmppVersion, desiredConnections: Number(desiredConnections) || 1, windowSize: Number(windowSize) || 16, passwordCipher: password || undefined, @@ -276,6 +286,7 @@ function ChannelFormModal({ setCorpCode(event.target.value)} placeholder="请输入企业代码" value={corpCode} /> setAccount(event.target.value)} placeholder="请输入网关账号" value={account} /> + setPassword(event.target.value)} placeholder="请输入网关密码" type="password" value={password} />
setAccessNo(event.target.value)} placeholder="请输入通道接入号" value={accessNo} /> @@ -301,14 +312,45 @@ function SmsTestModal({ const [phones, setPhones] = useState(''); const [content, setContent] = useState(''); const [accessNo, setAccessNo] = useState(''); + const [submitting, setSubmitting] = useState(false); + const [error, setError] = useState(''); + const [result, setResult] = useState(''); const billingCount = Math.max(1, Math.ceil(content.length / 67)); + async function submitTestSms() { + if (!phones.trim()) { + setError('请输入测试手机号'); + return; + } + if (!content.trim()) { + setError('请输入测试短信内容'); + return; + } + setSubmitting(true); + setError(''); + setResult(''); + try { + const response = await adminApi.testChannel(channel.id, { + phones, + content, + accessNo: accessNo.trim() || undefined, + }); + setResult(`已提交 ${response.submitted} 条测试短信,任务号 ${response.taskNo}`); + } catch (failure) { + setError(failure instanceof Error ? failure.message : '测试短信发送失败'); + } finally { + setSubmitting(false); + } + } + return ( - + )} onClose={onClose} @@ -358,6 +400,13 @@ function SmsTestModal({ 如需测试接入号,可在通道接入号后上追加接入号进行测试
+ {error ?

{error}

: null} + {result ? ( +
+ + {result} +
+ ) : null} );