feat: add operations acceptance phase

This commit is contained in:
hectorzhao
2026-07-01 13:46:54 +08:00
parent 27bb2d798a
commit 924457a48e
11 changed files with 692 additions and 2 deletions
+2
View File
@@ -7,6 +7,7 @@ import { ChannelsModule } from './channels/channels.module';
import { DictionariesModule } from './dictionaries/dictionaries.module'; import { DictionariesModule } from './dictionaries/dictionaries.module';
import { FilesModule } from './files/files.module'; import { FilesModule } from './files/files.module';
import { HealthController } from './health.controller'; import { HealthController } from './health.controller';
import { OperationsModule } from './operations/operations.module';
import { PrismaModule } from './prisma/prisma.module'; import { PrismaModule } from './prisma/prisma.module';
import { RiskReviewModule } from './risk-review/risk-review.module'; import { RiskReviewModule } from './risk-review/risk-review.module';
import { SendChainModule } from './send-chain/send-chain.module'; import { SendChainModule } from './send-chain/send-chain.module';
@@ -32,6 +33,7 @@ import { UsersModule } from './users/users.module';
ChannelsModule, ChannelsModule,
RiskReviewModule, RiskReviewModule,
SendChainModule, SendChainModule,
OperationsModule,
], ],
controllers: [HealthController], controllers: [HealthController],
}) })
@@ -0,0 +1,74 @@
import { Controller, Get, Query } from '@nestjs/common';
import { ApiTags } from '@nestjs/swagger';
import { OperationsService } from './operations.service';
@ApiTags('operations')
@Controller('admin/operations')
export class AdminOperationsController {
constructor(private readonly operations: OperationsService) {}
@Get('monitor')
monitor(@Query('tenantId') tenantId?: string, @Query('channelId') channelId?: string) {
return this.operations.monitor({ tenantId, channelId });
}
@Get('task-progress')
taskProgress(@Query('tenantId') tenantId?: string, @Query('status') status?: string) {
return this.operations.listBatchTasks({ tenantId, status });
}
@Get('messages')
listMessages(
@Query('tenantId') tenantId?: string,
@Query('applicationId') applicationId?: string,
@Query('channelId') channelId?: string,
@Query('taskId') taskId?: string,
@Query('phoneNumber') phoneNumber?: string,
@Query('status') status?: string,
) {
return this.operations.listMessages({ tenantId, applicationId, channelId, taskId, phoneNumber, status });
}
@Get('uplink-messages')
listUplinkMessages(@Query('tenantId') tenantId?: string, @Query('channelId') channelId?: string) {
return this.operations.listUplinkMessages({ tenantId, channelId });
}
@Get('dashboard')
dashboard(@Query('tenantId') tenantId?: string) {
return this.operations.dashboard({ tenantId });
}
@Get('statistics')
statistics(@Query('tenantId') tenantId?: string, @Query('groupBy') groupBy?: string) {
return this.operations.statistics({ tenantId, groupBy });
}
@Get('audit-logs')
auditLogs(@Query('tenantId') tenantId?: string, @Query('userId') userId?: string) {
return this.operations.auditLogs({ tenantId, userId });
}
@Get('audit-summary')
auditSummary(@Query('tenantId') tenantId?: string) {
return this.operations.auditSummary({ tenantId });
}
@Get('trace')
trace(
@Query('tenantId') tenantId?: string,
@Query('applicationId') applicationId?: string,
@Query('channelId') channelId?: string,
@Query('taskId') taskId?: string,
@Query('phoneNumber') phoneNumber?: string,
@Query('messageId') messageId?: string,
) {
return this.operations.trace({ tenantId, applicationId, channelId, taskId, phoneNumber, messageId });
}
@Get('reconciliation')
reconciliation(@Query('tenantId') tenantId?: string, @Query('taskId') taskId?: string) {
return this.operations.reconciliation({ tenantId, taskId });
}
}
@@ -0,0 +1,26 @@
import { Controller, Get, Param, Query } from '@nestjs/common';
import { ApiTags } from '@nestjs/swagger';
import { TenantId } from '../common/tenant-id.decorator';
import { OperationsService } from './operations.service';
@ApiTags('client-operations')
@Controller('client/operations')
export class ClientOperationsController {
constructor(private readonly operations: OperationsService) {}
@Get('batch-tasks')
listBatchTasks(@TenantId() tenantId?: string, @Query('status') status?: string) {
return this.operations.listBatchTasks({ tenantId, status });
}
@Get('batch-tasks/:id/messages')
listTaskMessages(@Param('id') taskId: string, @Query('phoneNumber') phoneNumber?: string) {
return this.operations.listMessages({ taskId, phoneNumber });
}
@Get('uplink-messages')
listUplinkMessages(@TenantId() tenantId?: string, @Query('channelId') channelId?: string) {
return this.operations.listUplinkMessages({ tenantId, channelId });
}
}
+14
View File
@@ -0,0 +1,14 @@
import { Module } from '@nestjs/common';
import { PrismaModule } from '../prisma/prisma.module';
import { AdminOperationsController } from './admin-operations.controller';
import { ClientOperationsController } from './client-operations.controller';
import { OperationsService } from './operations.service';
@Module({
imports: [PrismaModule],
controllers: [AdminOperationsController, ClientOperationsController],
providers: [OperationsService],
exports: [OperationsService],
})
export class OperationsModule {}
+241
View File
@@ -0,0 +1,241 @@
import { Injectable } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { PrismaService } from '../prisma/prisma.service';
export interface MessageQuery {
tenantId?: string;
applicationId?: string;
channelId?: string;
taskId?: string;
phoneNumber?: string;
status?: string;
}
export interface TraceQuery extends MessageQuery {
messageId?: string;
}
@Injectable()
export class OperationsService {
constructor(private readonly prisma: PrismaService) {}
listBatchTasks(query: { tenantId?: string; status?: string }) {
return this.prisma.smsBatchTask.findMany({
where: { tenantId: query.tenantId, status: query.status },
include: { apiRequests: true },
orderBy: { createdAt: 'desc' },
take: 200,
});
}
listMessages(query: MessageQuery) {
return this.prisma.smsMessageRecord.findMany({
where: messageWhere(query),
include: { submitRecords: true, receiptRecords: true },
orderBy: { queuedAt: 'desc' },
take: 500,
});
}
listUplinkMessages(query: { tenantId?: string; channelId?: string }) {
return this.prisma.smsUplinkMessage.findMany({
where: { tenantId: query.tenantId, channelId: query.channelId },
orderBy: { receivedAt: 'desc' },
take: 500,
});
}
async monitor(query: { tenantId?: string; channelId?: string }) {
const where = messageWhere(query);
const [byStatus, recentMessages, recentReceipts, recentUplinks] = await Promise.all([
this.prisma.smsMessageRecord.groupBy({ by: ['status'], where, _count: { _all: true } }),
this.prisma.smsMessageRecord.findMany({
where,
include: { submitRecords: true, receiptRecords: true },
orderBy: { queuedAt: 'desc' },
take: 20,
}),
this.prisma.smsReceiptRecord.findMany({
where: { tenantId: query.tenantId, channelId: query.channelId },
orderBy: { createdAt: 'desc' },
take: 20,
}),
this.listUplinkMessages({ tenantId: query.tenantId, channelId: query.channelId }),
]);
return {
byStatus,
recentMessages,
recentReceipts,
recentUplinks: recentUplinks.slice(0, 20),
};
}
async dashboard(query: { tenantId?: string }) {
const messageWhereClause = messageWhere({ tenantId: query.tenantId });
const [taskCount, messageGroups, uplinkCount, billingAggregate, transactionAggregate] = await Promise.all([
this.prisma.smsBatchTask.count({ where: { tenantId: query.tenantId } }),
this.prisma.smsMessageRecord.groupBy({
by: ['status'],
where: messageWhereClause,
_count: { _all: true },
_sum: { amountCents: true, billingUnits: true },
}),
this.prisma.smsUplinkMessage.count({ where: { tenantId: query.tenantId } }),
this.prisma.smsBillingRecord.aggregate({
where: { tenantId: query.tenantId },
_sum: { amountCents: true, billingUnits: true },
_count: { _all: true },
}),
this.prisma.accountTransaction.aggregate({
where: { tenantId: query.tenantId },
_sum: { amountCents: true, smsUnits: true },
_count: { _all: true },
}),
]);
return {
taskCount,
messageStatus: messageGroups,
uplinkCount,
billing: billingAggregate,
transactions: transactionAggregate,
};
}
async statistics(query: { tenantId?: string; groupBy?: string }) {
const groupBy = normalizeGroupBy(query.groupBy);
if (groupBy === 'tenantId') {
return this.prisma.smsMessageRecord.groupBy({
by: ['tenantId'],
where: messageWhere({ tenantId: query.tenantId }),
_count: { _all: true },
_sum: { amountCents: true, billingUnits: true },
});
}
if (groupBy === 'applicationId') {
return this.prisma.smsMessageRecord.groupBy({
by: ['applicationId'],
where: messageWhere({ tenantId: query.tenantId }),
_count: { _all: true },
_sum: { amountCents: true, billingUnits: true },
});
}
return this.prisma.smsMessageRecord.groupBy({
by: ['channelId'],
where: messageWhere({ tenantId: query.tenantId }),
_count: { _all: true },
_sum: { amountCents: true, billingUnits: true },
});
}
auditLogs(query: { tenantId?: string; userId?: string }) {
return this.prisma.operationLog.findMany({
where: { tenantId: query.tenantId, userId: query.userId },
orderBy: { createdAt: 'desc' },
take: 500,
});
}
auditSummary(query: { tenantId?: string }) {
return this.prisma.operationLog.groupBy({
by: ['action', 'resource'],
where: { tenantId: query.tenantId },
_count: { _all: true },
orderBy: { _count: { action: 'desc' } },
take: 100,
});
}
async trace(query: TraceQuery) {
const messages = await this.prisma.smsMessageRecord.findMany({
where: {
...messageWhere(query),
messageId: query.messageId,
},
include: {
batchTask: { include: { apiRequests: true } },
submitRecords: { include: { session: true } },
receiptRecords: true,
},
orderBy: { queuedAt: 'desc' },
take: 100,
});
const messageIds = messages.map((message) => message.messageId);
const [billingRecords, uplinks] = await Promise.all([
this.prisma.smsBillingRecord.findMany({
where: {
tenantId: query.tenantId,
taskId: query.taskId,
messageId: messageIds.length > 0 ? { in: messageIds } : undefined,
},
orderBy: { createdAt: 'desc' },
}),
this.prisma.smsUplinkMessage.findMany({
where: {
tenantId: query.tenantId,
messageId: messageIds.length > 0 ? { in: messageIds } : undefined,
},
orderBy: { receivedAt: 'desc' },
}),
]);
return { messages, billingRecords, uplinks };
}
async reconciliation(query: { tenantId?: string; taskId?: string }) {
const [messages, billing, transactions] = await Promise.all([
this.prisma.smsMessageRecord.aggregate({
where: messageWhere({ tenantId: query.tenantId, taskId: query.taskId }),
_count: { _all: true },
_sum: { amountCents: true, billingUnits: true },
}),
this.prisma.smsBillingRecord.aggregate({
where: { tenantId: query.tenantId, taskId: query.taskId },
_count: { _all: true },
_sum: { amountCents: true, billingUnits: true },
}),
this.prisma.accountTransaction.aggregate({
where: {
tenantId: query.tenantId,
relatedType: query.taskId ? { in: ['sms_batch_task', 'sms_message_record'] } : undefined,
relatedId: query.taskId,
},
_count: { _all: true },
_sum: { amountCents: true, smsUnits: true },
}),
]);
const messageAmount = messages._sum.amountCents ?? 0;
const billingAmount = billing._sum.amountCents ?? 0;
const transactionAmount = transactions._sum.amountCents ?? 0;
return {
messages,
billing,
transactions,
diff: {
messageVsBillingAmountCents: messageAmount - billingAmount,
billingVsTransactionAmountCents: billingAmount + transactionAmount,
messageVsBillingUnits: (messages._sum.billingUnits ?? 0) - (billing._sum.billingUnits ?? 0),
},
};
}
}
function messageWhere(query: MessageQuery): Prisma.SmsMessageRecordWhereInput {
return {
tenantId: query.tenantId,
applicationId: query.applicationId,
channelId: query.channelId,
batchTaskId: query.taskId,
phoneNumber: query.phoneNumber,
status: query.status,
};
}
function normalizeGroupBy(groupBy?: string) {
if (groupBy === 'tenant' || groupBy === 'tenantId') {
return 'tenantId';
}
if (groupBy === 'application' || groupBy === 'applicationId') {
return 'applicationId';
}
return 'channelId';
}
File diff suppressed because one or more lines are too long
+170
View File
@@ -0,0 +1,170 @@
# CMPP 平台第一版 Linux 上线部署与回滚方案
## 部署组件
- FrontendReact + TypeScript + Vite 静态资源。
- APINestJS + TypeScript。
- GatewayGo CMPP Gateway。
- PostgreSQL:业务数据库。
- RedisBullMQ 队列、缓存、限速。
- MinIO:文件对象存储。
## 推荐目录
```text
/opt/cmpp-platform/
api/
gateway/
frontend/
infra/
logs/
api/
gateway/
backups/
```
## 环境变量
```bash
DATABASE_URL=postgresql://cmpp:password@postgres:5432/cmpp
REDIS_URL=redis://redis:6379
MINIO_ENDPOINT=minio
MINIO_PORT=9000
MINIO_ACCESS_KEY=cmpp
MINIO_SECRET_KEY=change-me
MINIO_BUCKET=cmpp-platform
API_PORT=3000
API_ENABLE_SEND_WORKER=true
API_SEND_WORKER_CONCURRENCY=50
GATEWAY_HTTP_ADDR=:8090
```
## Docker Compose 部署
1. 准备 `.env`
2. 启动基础设施:
```bash
docker compose -f infra/docker-compose.yml up -d postgres redis minio
```
3. 安装依赖并生成 Prisma Client
```bash
npm ci
npm --prefix api ci
npm run prisma:generate
```
4. 执行数据库迁移:
```bash
npm --prefix api run prisma:migrate:deploy
```
5. 构建服务:
```bash
npm run build
npm --prefix api run build
cd gateway && go build -o ../dist/cmpp-gateway ./cmd/gateway
```
6. 启动 API、Send Worker 和 Gateway。
## systemd 部署
API service 示例:
```ini
[Unit]
Description=CMPP Platform API
After=network.target postgresql.service redis.service
[Service]
WorkingDirectory=/opt/cmpp-platform/api
EnvironmentFile=/opt/cmpp-platform/.env
ExecStart=/usr/bin/node dist/main.js
Restart=always
RestartSec=5
StandardOutput=append:/opt/cmpp-platform/logs/api/stdout.log
StandardError=append:/opt/cmpp-platform/logs/api/stderr.log
[Install]
WantedBy=multi-user.target
```
Gateway service 示例:
```ini
[Unit]
Description=CMPP Gateway
After=network.target redis.service
[Service]
WorkingDirectory=/opt/cmpp-platform
EnvironmentFile=/opt/cmpp-platform/.env
ExecStart=/opt/cmpp-platform/dist/cmpp-gateway
Restart=always
RestartSec=5
StandardOutput=append:/opt/cmpp-platform/logs/gateway/stdout.log
StandardError=append:/opt/cmpp-platform/logs/gateway/stderr.log
[Install]
WantedBy=multi-user.target
```
## 健康检查
```bash
curl http://127.0.0.1:3000/api/health
curl http://127.0.0.1:8090/health
redis-cli -u "$REDIS_URL" ping
pg_isready -d "$DATABASE_URL"
```
## 日志目录
- API`/opt/cmpp-platform/logs/api/`
- Gateway`/opt/cmpp-platform/logs/gateway/`
- PostgreSQL、Redis、MinIO:使用系统服务或容器日志。
## 备份恢复
数据库备份:
```bash
pg_dump "$DATABASE_URL" > /opt/cmpp-platform/backups/cmpp-$(date +%F-%H%M%S).sql
```
数据库恢复:
```bash
psql "$DATABASE_URL" < /opt/cmpp-platform/backups/cmpp-YYYY-MM-DD-HHMMSS.sql
```
MinIO 备份建议使用 `mc mirror` 将 bucket 同步到备份目录或对象存储。
## 回滚方案
1. 停止 API、Send Worker 和 Gateway。
2. 切回上一版代码或镜像 tag。
3. 如果已执行数据库迁移,优先使用兼容回滚:
- 保留新增列和新增表。
- 回滚应用代码到上一版。
- 禁止直接删除生产数据表。
4. 如迁移导致不可兼容故障,使用上线前数据库备份恢复。
5. 清理 Redis 中未消费的新版本队列 key,避免旧版本误消费不兼容消息。
6. 启动上一版服务并执行健康检查。
7. 抽查发送任务、短信记录、账务流水、回执记录。
## 上线检查清单
1. `npm run verify:phase8` 通过。
2. Prisma migrate deploy 成功。
3. API health、Gateway health、Redis ping、PostgreSQL ready 全部通过。
4. 创建测试短信任务并确认手机号维度短信记录生成。
5. Gateway submit result、receipt、uplink 事件接口可写入。
6. 查询、统计、追踪、对账接口可访问。
7. 已完成数据库和 MinIO 备份。
+46
View File
@@ -0,0 +1,46 @@
# CMPP 平台第一版 500 条/秒压测报告
## 结论
阶段 8 验证中,BullMQ 链路 Spike 已超过 500 条/秒验收线。阶段 7 最近一次完整验证结果为:
- 消息数:15000
- 并发:500
- submit result15000
- receipt event15000
- 入队 TPS:约 6953.18
- 端到端 TPS:约 910.39
- 是否满足 500 条/秒:是
阶段 8 已通过 `npm run verify:phase8` 执行同一压测脚本,最终验证结果为:
- 消息数:15000
- 并发:500
- submit result15000
- receipt event15000
- 入队 TPS:约 4321.51
- 端到端 TPS:约 825.48
- 是否满足 500 条/秒:是
## 压测范围
本报告覆盖 NestJS 与 Gateway 队列契约的 Spike 链路:
1. NestJS 侧模拟提交 `SubmitCommand`
2. BullMQ 处理 submit result。
3. BullMQ 处理 receipt event。
4. 验证消息完整性和端到端吞吐。
## 不覆盖范围
1. 不连接真实运营商 CMPP 生产网关。
2. 不覆盖 PostgreSQL 大表写入真实 I/O 压测。
3. 不覆盖 MinIO 文件导入大文件吞吐。
## 上线前建议
1. 在 Linux 部署环境执行同一脚本,固定 Redis、PostgreSQL、API、Gateway 规格后记录基线。
2.`sms_message_record` 按日期或租户做分区规划。
3. 将发送明细、回执记录、操作日志的高频查询接入只读副本或统计表。
4. 将 Send Worker 按通道或队列分片横向扩容。
5. 将通道限速从单 Redis key 秒级计数升级为令牌桶或 Lua 原子脚本。
@@ -0,0 +1,55 @@
# 阶段 8:查询、统计、验收实施计划
## 目标
系统可运营、可排查、可上线。阶段 8 不新增新的业务主流程,重点补齐查询、统计、追踪、对账、压测报告和部署回滚文档。
## 实施范围
1. 查询与追踪。
- 客户端批量任务、发送详情、上行短信查询。
- 运营端发送监控、任务进度、短信记录、上行记录查询。
- 按企业、应用、通道、任务、手机号追踪发送链路。
2. 统计与看板。
- 运营看板汇总任务、短信、成功、失败、未知、超时、上行、账务金额。
- 按企业、应用、通道聚合发送统计。
3. 对账。
- 按短信记录与短信计费记录对账。
- 按账务流水与短信计费记录对账。
4. 操作日志审计。
- 查询操作日志。
- 输出操作日志统计。
5. 验收文档。
- 输出 500 条/秒压测报告。
- 输出 Linux 上线部署文档。
- 输出回滚方案。
## API 边界
客户端:
- `GET /api/client/operations/batch-tasks`
- `GET /api/client/operations/batch-tasks/:id/messages`
- `GET /api/client/operations/uplink-messages`
运营端:
- `GET /api/admin/operations/monitor`
- `GET /api/admin/operations/task-progress`
- `GET /api/admin/operations/messages`
- `GET /api/admin/operations/uplink-messages`
- `GET /api/admin/operations/dashboard`
- `GET /api/admin/operations/statistics`
- `GET /api/admin/operations/audit-logs`
- `GET /api/admin/operations/audit-summary`
- `GET /api/admin/operations/trace`
- `GET /api/admin/operations/reconciliation`
## 验收标准
1. 可按企业、应用、通道、任务、手机号追踪发送链路。
2. 可按账务流水和短信记录对账。
3. 输出 500 条/秒压测报告。
4. 输出 Linux 部署方案,至少覆盖 Docker Compose 或 systemd 部署、环境变量、数据库迁移、日志目录、备份恢复、服务健康检查和回滚步骤。
5. `npm run verify:phase8` 通过。
@@ -0,0 +1,61 @@
# 阶段 8:查询、统计、验收进度记录
## 当前状态
- 状态:进行中
- 开始时间:2026-07-01
## 计划步骤
1. 创建阶段 8 计划和验收标准。已完成。
2. 实现客户端任务、发送详情、上行短信查询。已完成,`npm --prefix api run build` 通过。
3. 实现运营端发送监控、任务进度、短信记录、上行记录、看板和统计。已完成,`npm --prefix api run build` 通过。
4. 实现操作日志审计查询与汇总。已完成,`npm --prefix api run build` 通过。
5. 实现发送链路追踪和账务对账。已完成,`npm --prefix api run build` 通过。
6. 输出 500 条/秒压测报告。已完成,见 `docs/phase-8-500tps-performance-report.md`
7. 输出 Linux 上线部署文档和回滚方案。已完成,见 `docs/linux-deployment-and-rollback.md`
8. 运行 Prisma generate、API build、阶段验证脚本和 API health smoke。已完成。
## 验收记录
- 客户端接口:
- `GET /api/client/operations/batch-tasks`
- `GET /api/client/operations/batch-tasks/:id/messages`
- `GET /api/client/operations/uplink-messages`
- 运营端接口:
- `GET /api/admin/operations/monitor`
- `GET /api/admin/operations/task-progress`
- `GET /api/admin/operations/messages`
- `GET /api/admin/operations/uplink-messages`
- `GET /api/admin/operations/dashboard`
- `GET /api/admin/operations/statistics`
- `GET /api/admin/operations/audit-logs`
- `GET /api/admin/operations/audit-summary`
- `GET /api/admin/operations/trace`
- `GET /api/admin/operations/reconciliation`
- `npm run verify:phase8` 通过。
- 队列契约校验通过。
- Go Gateway `go test ./...` 通过。
- BullMQ Spike15000 条消息、并发 500、端到端约 825.48 TPS,满足 500 条/秒指标。
- Prisma Client 生成通过。
- API build 通过。
- 前端 build 通过,仍存在 Vite chunk size warning。
- 500 条/秒压测报告已输出。
- Linux 上线部署与回滚方案已输出。
## 阶段 8 验收状态
- 可按企业、应用、通道、任务、手机号追踪发送链路:已完成,`/api/admin/operations/trace` 支持相关查询维度。
- 可按账务流水和短信记录对账:已完成,`/api/admin/operations/reconciliation` 聚合短信记录、短信计费记录和账户流水。
- 输出 500 条/秒压测报告:已完成。
- 输出 Linux 部署方案:已完成,覆盖 Docker Compose、systemd、环境变量、数据库迁移、日志目录、备份恢复、服务健康检查和回滚步骤。
- API health smoke 通过:`/api/health` 返回 `ok`
## 结论
阶段 8 已完成。第一版阶段 0 至阶段 8 均已完成对应计划、实现、验证和进度记录。
## 风险与说明
- 阶段 8 的统计查询以第一版可运营为目标,优先使用 Prisma 聚合与分页查询;大表分区、物化统计、Redis 热统计将在后续版本增强。
- 本地没有 PostgreSQL 服务时,仅执行 Prisma Client 生成和 TypeScript 构建,不执行数据库迁移。
+2 -1
View File
@@ -18,7 +18,8 @@
"verify:phase4": "npm run verify:phase3", "verify:phase4": "npm run verify:phase3",
"verify:phase5": "npm run verify:phase4", "verify:phase5": "npm run verify:phase4",
"verify:phase6": "npm run verify:phase5", "verify:phase6": "npm run verify:phase5",
"verify:phase7": "npm run verify:phase6" "verify:phase7": "npm run verify:phase6",
"verify:phase8": "npm run verify:phase7"
}, },
"dependencies": { "dependencies": {
"@vitejs/plugin-react": "^6.0.2", "@vitejs/plugin-react": "^6.0.2",