feat: 实现发送质量监控与报备状态消息通知

api/src/app.module.ts

@@ -27,6 +27,8 @@ import { UsersModule } from './users/users.module';
 import { SignatureRetirementModule } from './signature-retirement/signature-retirement.module';
 import { SecurityDetectionModule } from './security-detection/security-detection.module';
 import { MetricsModule } from './metrics/metrics.module';
+import { ReportNotificationsModule } from './report-notifications/report-notifications.module';
+import { SendingMonitorModule } from './sending-monitor/sending-monitor.module';
 
 @Module({
   imports: [
@@ -57,12 +59,16 @@ import { MetricsModule } from './metrics/metrics.module';
     SignatureRetirementModule,
     SecurityDetectionModule,
     MetricsModule,
+    ReportNotificationsModule,
+    SendingMonitorModule,
   ],
   controllers: [HealthController],
   providers: [RequestContextMiddleware, SessionValidationMiddleware, ManualOperationAuditMiddleware],
 })
 export class AppModule implements NestModule {
   configure(consumer: MiddlewareConsumer) {
-    consumer.apply(RequestContextMiddleware, SessionValidationMiddleware, ManualOperationAuditMiddleware).forRoutes('*');
+    consumer
+      .apply(RequestContextMiddleware, SessionValidationMiddleware, ManualOperationAuditMiddleware)
+      .forRoutes('*');
   }
 }

api/src/operations/admin-operations.controller.ts

@@ -46,8 +46,10 @@ export class AdminOperationsController {
     @Query('hasDrainage') hasDrainage?: string,
     @Query('page') page?: string,
     @Query('pageSize') pageSize?: string,
+    @Query('monitorSnapshotId') monitorSnapshotId?: string,
   ) {
     return this.operations.listMessagesPage({
+      monitorSnapshotId,
       tenantId,
       applicationId,
       channelId,
@@ -80,8 +82,10 @@ export class AdminOperationsController {
     @Query('status') status: string | undefined,
     @Query('hasDrainage') hasDrainage: string | undefined,
     @Res() response: DownloadResponse,
+    @Query('monitorSnapshotId') monitorSnapshotId?: string,
   ) {
     const exported = await this.operations.exportMessages({
+      monitorSnapshotId,
       tenantId,
       applicationId,
       channelId,
@@ -105,25 +109,37 @@ export class AdminOperationsController {
   }
 
   @Get('message-segment-audits')
-  messageSegmentAudits(
-    @Query('messageId') messageId?: string,
-    @Query('messageRecordId') messageRecordId?: string,
-  ) {
+  messageSegmentAudits(@Query('messageId') messageId?: string, @Query('messageRecordId') messageRecordId?: string) {
     return this.operations.listMessageSegmentAudits({ messageId, messageRecordId });
   }
 
   @Get('uplink-messages')
-  listUplinkMessages(@Query('tenantId') tenantId?: string, @Query('channelId') channelId?: string, @Query('phoneNumber') phoneNumber?: string, @Query('keyword') keyword?: string, @Query('startTime') startTime?: string, @Query('endTime') endTime?: string, @Query('page') page?: string, @Query('pageSize') pageSize?: string) {
+  listUplinkMessages(
+    @Query('tenantId') tenantId?: string,
+    @Query('channelId') channelId?: string,
+    @Query('phoneNumber') phoneNumber?: string,
+    @Query('keyword') keyword?: string,
+    @Query('startTime') startTime?: string,
+    @Query('endTime') endTime?: string,
+    @Query('page') page?: string,
+    @Query('pageSize') pageSize?: string,
+  ) {
     return page || pageSize
-      ? this.operations.listUplinkMessagesPage({ tenantId, channelId, phoneNumber, keyword, startTime, endTime, page: Number(page), pageSize: Number(pageSize) })
+      ? this.operations.listUplinkMessagesPage({
+          tenantId,
+          channelId,
+          phoneNumber,
+          keyword,
+          startTime,
+          endTime,
+          page: Number(page),
+          pageSize: Number(pageSize),
+        })
       : this.operations.listUplinkMessages({ tenantId, channelId, phoneNumber, keyword, startTime, endTime });
   }
 
   @Post('uplink-messages/:id/claim')
-  claimUplinkMatchCandidate(
-    @Param('id') id: string,
-    @Body() body: { candidateId?: string; operatorId?: string },
-  ) {
+  claimUplinkMatchCandidate(@Param('id') id: string, @Body() body: { candidateId?: string; operatorId?: string }) {
     return this.sendChain.claimUplinkMatchCandidate(id, String(body.candidateId ?? ''), body.operatorId);
   }
 
@@ -231,10 +247,7 @@ export class AdminOperationsController {
   }
 
   @Post('gateway-submit-dead-letters/:id/resolve')
-  resolveGatewaySubmitDeadLetter(
-    @Param('id') id: string,
-    @CurrentSessionUserId() operatorId?: string,
-  ) {
+  resolveGatewaySubmitDeadLetter(@Param('id') id: string, @CurrentSessionUserId() operatorId?: string) {
     return this.sendChain.resolveGatewaySubmitDeadLetter(id, operatorId);
   }
 
@@ -381,16 +394,23 @@ export class AdminOperationsController {
     @Body() body: { previewToken?: string; reason?: string; ratePerSecond?: number; consecutiveFailureLimit?: number },
     @CurrentSessionUserId() operatorId?: string,
   ) {
-    return this.sendChain.createDownstreamRequeueTask({
-      previewToken: body.previewToken ?? '',
-      reason: body.reason ?? '',
-      ratePerSecond: body.ratePerSecond,
-      consecutiveFailureLimit: body.consecutiveFailureLimit,
-    }, operatorId);
+    return this.sendChain.createDownstreamRequeueTask(
+      {
+        previewToken: body.previewToken ?? '',
+        reason: body.reason ?? '',
+        ratePerSecond: body.ratePerSecond,
+        consecutiveFailureLimit: body.consecutiveFailureLimit,
+      },
+      operatorId,
+    );
   }
 
   @Get('downstream-requeue-tasks')
-  listDownstreamRequeueTasks(@Query('status') status?: string, @Query('page') page?: string, @Query('pageSize') pageSize?: string) {
+  listDownstreamRequeueTasks(
+    @Query('status') status?: string,
+    @Query('page') page?: string,
+    @Query('pageSize') pageSize?: string,
+  ) {
     return this.sendChain.listDownstreamRequeueTasks({ status, page: Number(page), pageSize: Number(pageSize) });
   }
 
@@ -407,7 +427,12 @@ export class AdminOperationsController {
     @Query('page') page?: string,
     @Query('pageSize') pageSize?: string,
   ) {
-    return this.sendChain.listDownstreamRequeueTaskItems(id, { status, keyword, page: Number(page), pageSize: Number(pageSize) });
+    return this.sendChain.listDownstreamRequeueTaskItems(id, {
+      status,
+      keyword,
+      page: Number(page),
+      pageSize: Number(pageSize),
+    });
   }
 
   @Post('downstream-requeue-tasks/:id/:action')
@@ -441,7 +466,18 @@ export class AdminSystemLogsController {
     @Query('page') page?: string,
     @Query('pageSize') pageSize?: string,
   ) {
-    return this.protocolLogs.list({ protocol, direction, eventType, status, keyword, range, createdAtFrom, createdAtTo, page: Number(page), pageSize: Number(pageSize) });
+    return this.protocolLogs.list({
+      protocol,
+      direction,
+      eventType,
+      status,
+      keyword,
+      range,
+      createdAtFrom,
+      createdAtTo,
+      page: Number(page),
+      pageSize: Number(pageSize),
+    });
   }
 
   @Get()
@@ -457,11 +493,34 @@ export class AdminSystemLogsController {
     @Query('page') page?: string,
     @Query('pageSize') pageSize?: string,
   ) {
-    return this.operations.systemLogs({ tenantId, userId, keyword, level, module, range, createdAtFrom, createdAtTo, page: Number(page), pageSize: Number(pageSize) });
+    return this.operations.systemLogs({
+      tenantId,
+      userId,
+      keyword,
+      level,
+      module,
+      range,
+      createdAtFrom,
+      createdAtTo,
+      page: Number(page),
+      pageSize: Number(pageSize),
+    });
   }
 
   @Post('exports')
-  export(@Body() body: { tenantId?: string; userId?: string; keyword?: string; level?: string; module?: string; range?: string; createdAtFrom?: string; createdAtTo?: string }) {
+  export(
+    @Body()
+    body: {
+      tenantId?: string;
+      userId?: string;
+      keyword?: string;
+      level?: string;
+      module?: string;
+      range?: string;
+      createdAtFrom?: string;
+      createdAtTo?: string;
+    },
+  ) {
     return this.operations.exportSystemLogs(body);
   }
 }

api/src/operations/operations.contracts.ts

@@ -1,6 +1,7 @@
 // Stable controller/query contracts extracted in R2.
 
 export interface MessageQuery {
+  monitorSnapshotId?: string;
   tenantId?: string;
   applicationId?: string;
   channelId?: string;

api/src/operations/queries/messages.queries.ts

@@ -1,27 +1,32 @@
 import { BadRequestException, NotFoundException } from '@nestjs/common';
-import { Prisma } from '@prisma/client';
-import { randomUUID } from 'node:crypto';
+
 import { moneyToNumber } from '../../common/money';
 import { PrismaService } from '../../prisma/prisma.service';
-import type { MessageQuery, TraceQuery, OperationLogQuery, GatewaySubmitDeadLetterQuery, DownstreamDeliveryQuery, DownstreamDeliveryDashboardQuery, DownstreamRecoveryStatusQuery, MessageSegmentAuditQuery, SignatureQualityQuery } from '../operations.contracts';
-import { messageWhere, recognizedCarrierValues, carrierWhere, startOfShanghaiDay, endOfShanghaiDay, qualityBusinessDay, shanghaiDateKey, normalizeGroupBy, returnedTransactionWhere, createdAtRange, downstreamAlertPendingMinutes, downstreamAlertRecentFailedHours, downstreamAlertWindows, downstreamAlertWhere, stalledPendingWhere, downstreamDeliveryScopedWhere, parseDateBoundary, downstreamRecoveryStatusWhere, escapeCsvCell, formatCsvDate, formatExportTimestamp, clientApplicationView, clientReceiptView, clientMessageView, clientBatchTaskView, clientUplinkView, clientAccountView, clientRechargeView, summarizeMessageGroups, groupDownstreamByType, groupDownstreamByApplication, positiveInteger, operationLogLevelWhere, normalizeOperationLog, sanitizeGatewaySubmitException, redactGatewayCommandValue } from '../operations.helpers';
+import type { MessageQuery } from '../operations.contracts';
+import {
+  messageWhere,
+  escapeCsvCell,
+  formatExportTimestamp,
+  clientMessageView,
+  clientBatchTaskView,
+} from '../operations.helpers';
 
-// R2 messages query domain. Method bodies are preserved byte-for-byte from the facade baseline.
+// Message queries share the same bounded monitor sample filter for lists and exports.
 export class OperationsMessageQueries {
   constructor(private readonly prisma: PrismaService) {}
 
-listBatchTasks(query: { tenantId?: string; status?: string }) {
+  listBatchTasks(query: { tenantId?: string; status?: string }) {
     return this.prisma.smsBatchTask.findMany({
       where: { tenantId: query.tenantId, status: query.status, sourceType: 'client' },
       include: { apiRequests: true },
       orderBy: { createdAt: 'desc' },
     });
   }
-async listClientBatchTasks(query: { tenantId?: string; status?: string }) {
+  async listClientBatchTasks(query: { tenantId?: string; status?: string }) {
     const items = await this.listBatchTasks(query);
     return items.map(clientBatchTaskView);
   }
-listMessages(query: MessageQuery) {
+  listMessages(query: MessageQuery) {
     return this.prisma.smsMessageRecord.findMany({
       where: messageWhere(query),
       include: {
@@ -38,10 +43,34 @@ listMessages(query: MessageQuery) {
       orderBy: { queuedAt: 'desc' },
     });
   }
-async listMessagesPage(query: MessageQuery) {
+  private async messageFilter(query: MessageQuery) {
+    const where = messageWhere(query);
+    if (query.monitorSnapshotId) {
+      const snapshot = await this.prisma.sendingMonitorSnapshot.findUnique({ where: { id: query.monitorSnapshotId } });
+      if (
+        !snapshot ||
+        (query.tenantId &&
+          (snapshot.dimensions as { tenantId?: string }).tenantId &&
+          (snapshot.dimensions as { tenantId?: string }).tenantId !== query.tenantId)
+      )
+        throw new NotFoundException('监控快照不存在');
+      if (snapshot.windowFrom.getTime() < Date.now() - 72 * 3600000)
+        throw new BadRequestException('该窗口已超过72小时样本保留期，历史快照仍可查询');
+      where.monitorFacts = {
+        some: {
+          kind: snapshot.type === 'industry' ? 'attempt' : 'business',
+          dimensionKey: snapshot.dimensionKey,
+          submittedAt: { gte: snapshot.windowFrom, lt: snapshot.evaluationAt },
+          ...(snapshot.type === 'verification' ? { verification: true } : {}),
+        },
+      };
+    }
+    return where;
+  }
+  async listMessagesPage(query: MessageQuery) {
     const page = Math.max(1, Math.floor(Number(query.page) || 1));
     const pageSize = Math.min(100, Math.max(1, Math.floor(Number(query.pageSize) || 25)));
-    const where = messageWhere(query);
+    const where = await this.messageFilter(query);
     const [items, total] = await Promise.all([
       this.prisma.smsMessageRecord.findMany({
         where,
@@ -75,7 +104,7 @@ async listMessagesPage(query: MessageQuery) {
     return { items, total, page, pageSize };
   }
 
-async getMessage(id: string) {
+  async getMessage(id: string) {
     const item = await this.prisma.smsMessageRecord.findUnique({
       where: { id },
       include: {
@@ -121,9 +150,10 @@ async getMessage(id: string) {
     if (!item) throw new NotFoundException('Message record not found');
     return item;
   }
-async exportMessages(query: MessageQuery) {
+  async exportMessages(query: MessageQuery) {
+    const where = await this.messageFilter(query);
     const items = await this.prisma.smsMessageRecord.findMany({
-      where: messageWhere(query),
+      where,
       select: {
         messageId: true,
         queuedAt: true,
@@ -144,7 +174,22 @@ async exportMessages(query: MessageQuery) {
       orderBy: [{ queuedAt: 'desc' }, { id: 'desc' }],
     });
     const rows = [
-      ['消息编号', '企业', '应用', '提交时间', '手机号', '地区', '运营商', '计费条数', '金额', '通道', '状态', '是否含引流', '回执时间', '短信内容'],
+      [
+        '消息编号',
+        '企业',
+        '应用',
+        '提交时间',
+        '手机号',
+        '地区',
+        '运营商',
+        '计费条数',
+        '金额',
+        '通道',
+        '状态',
+        '是否含引流',
+        '回执时间',
+        '短信内容',
+      ],
       ...items.map((item) => [
         item.messageId,
         item.tenant?.name ?? '',
@@ -156,7 +201,9 @@ async exportMessages(query: MessageQuery) {
         String(item.billingUnits),
         String(moneyToNumber(item.amountCents)),
         item.channel?.name ?? '',
-        item.status === 'submit_failed' || ['rejected', 'timeout'].includes(item.submitStatus ?? '') ? 'submit_failed' : item.status,
+        item.status === 'submit_failed' || ['rejected', 'timeout'].includes(item.submitStatus ?? '')
+          ? 'submit_failed'
+          : item.status,
         item.hasDrainageContent === true ? '是' : item.hasDrainageContent === false ? '否' : '未检测',
         item.deliveredAt?.toISOString() ?? '',
         item.content,
@@ -167,11 +214,11 @@ async exportMessages(query: MessageQuery) {
       content: rows.map((row) => row.map((cell) => escapeCsvCell(String(cell))).join(',')).join('\n'),
     };
   }
-async listClientMessages(query: MessageQuery) {
+  async listClientMessages(query: MessageQuery) {
     const items = await this.listMessages(query);
     return items.map(clientMessageView);
   }
-async listClientMessagesPage(query: MessageQuery) {
+  async listClientMessagesPage(query: MessageQuery) {
     const result = await this.listMessagesPage(query);
     return { ...result, items: result.items.map(clientMessageView) };
   }

api/src/report-notifications/report-notifications.module.ts

@@ -0,0 +1,133 @@
+import {
+  BadRequestException,
+  Body,
+  Controller,
+  ForbiddenException,
+  Get,
+  Injectable,
+  Module,
+  NotFoundException,
+  Param,
+  Post,
+  Query,
+  Req,
+} from '@nestjs/common';
+import { Prisma } from '@prisma/client';
+import { PrismaService } from '../prisma/prisma.service';
+import { PrismaModule } from '../prisma/prisma.module';
+import type { SessionRequest } from '../auth/session-validation.middleware';
+
+export function positivePage(value: unknown, fallback: number, max = 100000) {
+  if (value === undefined) return fallback;
+  const n = Number(value);
+  if (!Number.isSafeInteger(n) || n < 1 || n > max) throw new BadRequestException('页码、条数或版本无效');
+  return n;
+}
+
+@Injectable()
+export class ReportNotificationsService {
+  constructor(private readonly prisma: PrismaService) {}
+
+  async scope(req: SessionRequest) {
+    if (!req.sessionUserId) throw new ForbiddenException('请先登录');
+    if (req.authSession?.portal === 'client' && req.sessionTenantId) return req.sessionTenantId;
+    const admin =
+      req.authSession?.portal === 'admin' &&
+      (await this.prisma.user.findFirst({
+        where: {
+          id: req.sessionUserId,
+          status: 'active',
+          deletedAt: null,
+          roles: { some: { role: { code: 'platform_admin' } } },
+        },
+        select: { id: true },
+      }));
+    if (!admin) throw new ForbiddenException('无消息查看权限');
+    return null;
+  }
+
+  async list(req: SessionRequest, query: { page?: string; pageSize?: string; unread?: string }) {
+    const tenant = await this.scope(req);
+    const page = positivePage(query.page, 1);
+    const pageSize = positivePage(query.pageSize, 20, 100);
+    const filter = Prisma.sql`WHERE (${tenant}::text IS NULL OR h."tenantId"=${tenant})
+      AND (${query.unread === 'true'}=false OR COALESCE(r.revision,0)<h.revision)`;
+    return this.prisma.$transaction(
+      async (tx) => {
+        const items = await tx.$queryRaw<Array<Record<string, unknown>>>(Prisma.sql`
+        SELECT h.*,COALESCE(r.revision,0)<h.revision AS unread FROM "ReportNotificationHour" h
+        LEFT JOIN "ReportNotificationRead" r ON r."hourId"=h.id AND r."userId"=${req.sessionUserId!}
+        ${filter} ORDER BY h.hour DESC,h.id DESC LIMIT ${pageSize} OFFSET ${(page - 1) * pageSize}`);
+        const counts = await tx.$queryRaw<Array<{ total: number }>>(Prisma.sql`
+        SELECT count(*)::integer AS total FROM "ReportNotificationHour" h
+        LEFT JOIN "ReportNotificationRead" r ON r."hourId"=h.id AND r."userId"=${req.sessionUserId!} ${filter}`);
+        return { items, total: counts[0].total, page, pageSize };
+      },
+      { isolationLevel: 'RepeatableRead' },
+    );
+  }
+
+  async hour(req: SessionRequest, id: string) {
+    const tenantId = await this.scope(req);
+    const item = await this.prisma.reportNotificationHour.findFirst({
+      where: { id, ...(tenantId ? { tenantId } : {}) },
+    });
+    if (!item) throw new NotFoundException('消息不存在');
+    return item;
+  }
+
+  async detail(req: SessionRequest, id: string, query: { page?: string; pageSize?: string }) {
+    const hour = await this.hour(req, id);
+    const page = positivePage(query.page, 1),
+      pageSize = positivePage(query.pageSize, 20, 100);
+    const [items, total] = await this.prisma.$transaction([
+      this.prisma.reportReadinessEvent.findMany({
+        where: { hourId: id },
+        orderBy: [{ createdAt: 'desc' }, { id: 'desc' }],
+        skip: (page - 1) * pageSize,
+        take: pageSize,
+      }),
+      this.prisma.reportReadinessEvent.count({ where: { hourId: id } }),
+    ]);
+    return { hour, items, total, page, pageSize };
+  }
+
+  async read(req: SessionRequest, id: string, revision: unknown) {
+    const hour = await this.hour(req, id);
+    const version = positivePage(revision, 0, 2147483647);
+    if (!version || version > hour.revision) throw new BadRequestException('消息版本无效，请刷新');
+    await this.prisma.$executeRaw`INSERT INTO "ReportNotificationRead" ("userId","hourId",revision)
+      VALUES (${req.sessionUserId!},${id},${version}) ON CONFLICT ("userId","hourId")
+      DO UPDATE SET revision=GREATEST("ReportNotificationRead".revision,EXCLUDED.revision)`;
+    return { success: true };
+  }
+}
+
+@Controller(['admin/report-notifications', 'client/report-notifications'])
+class ReportNotificationsController {
+  constructor(private readonly service: ReportNotificationsService) {}
+  @Get() list(@Req() req: SessionRequest, @Query() query: { page?: string; pageSize?: string; unread?: string }) {
+    return this.service.list(req, query);
+  }
+  @Get('summary') async summary(@Req() req: SessionRequest) {
+    const result = await this.service.list(req, { pageSize: '1', unread: 'true' });
+    return { count: result.total };
+  }
+  @Get(':id') detail(
+    @Req() req: SessionRequest,
+    @Param('id') id: string,
+    @Query() query: { page?: string; pageSize?: string },
+  ) {
+    return this.service.detail(req, id, query);
+  }
+  @Post(':id/read') read(@Req() req: SessionRequest, @Param('id') id: string, @Body() body: { revision?: unknown }) {
+    return this.service.read(req, id, body?.revision);
+  }
+}
+
+@Module({
+  imports: [PrismaModule],
+  controllers: [ReportNotificationsController],
+  providers: [ReportNotificationsService],
+})
+export class ReportNotificationsModule {}

api/src/report-notifications/report-notifications.spec.ts

@@ -0,0 +1,35 @@
+import { ReportNotificationsService, positivePage } from './report-notifications.module';
+import type { PrismaService } from '../prisma/prisma.service';
+import type { SessionRequest } from '../auth/session-validation.middleware';
+
+describe('report notification access and read versions', () => {
+  const db = { reportNotificationHour: { findFirst: jest.fn() }, $executeRaw: jest.fn() };
+  const service = new ReportNotificationsService(db as unknown as PrismaService);
+  const req = {
+    sessionUserId: 'reader',
+    sessionTenantId: 'tenant-a',
+    authSession: { portal: 'client' },
+  } as SessionRequest;
+  beforeEach(() => jest.clearAllMocks());
+  it('rejects anonymous access', async () => {
+    await expect(service.scope({} as SessionRequest)).rejects.toThrow('请先登录');
+  });
+  it('uses the session tenant and hides inaccessible hours', async () => {
+    db.reportNotificationHour.findFirst.mockResolvedValue(null);
+    await expect(service.hour(req, 'tenant-b-hour')).rejects.toThrow('消息不存在');
+    expect(db.reportNotificationHour.findFirst).toHaveBeenCalledWith({
+      where: { id: 'tenant-b-hour', tenantId: 'tenant-a' },
+    });
+  });
+  it('rejects future read revisions without writing and accepts a displayed older revision', async () => {
+    db.reportNotificationHour.findFirst.mockResolvedValue({ id: 'hour', revision: 4 });
+    await expect(service.read(req, 'hour', 5)).rejects.toThrow('消息版本无效');
+    expect(db.$executeRaw).not.toHaveBeenCalled();
+    await expect(service.read(req, 'hour', 3)).resolves.toEqual({ success: true });
+    expect(db.$executeRaw).toHaveBeenCalledTimes(1);
+  });
+  it('bounds pages and rejects fractional or zero values', () => {
+    for (const value of ['0', '-1', '1.5', '101', 'NaN']) expect(() => positivePage(value, 20, 100)).toThrow();
+    expect(positivePage(undefined, 20, 100)).toBe(20);
+  });
+});

api/src/send-chain/gateway-wire-timing.spec.ts

@@ -0,0 +1,23 @@
+import { wireTiming } from './send-gateway-result.service';
+describe('Gateway wire timing compatibility', () => {
+  it('does not invent timestamps for old events or pre-write failures', () => {
+    expect(wireTiming({})).toEqual({});
+    expect(wireTiming({ wireTimeSource: 'not_written', receiptRequested: true })).toEqual({
+      wireTimeSource: 'not_written',
+      receiptRequested: true,
+    });
+  });
+  it('preserves the earliest actual time when replaying aggregate results', () => {
+    const prior = new Date('2026-09-06T04:00:00Z');
+    expect(
+      wireTiming(
+        { firstWireSubmitAt: '2026-09-06T04:00:05Z', wireTimeSource: 'gateway_write_complete', receiptRequested: true },
+        prior,
+      ).firstWireSubmitAt,
+    ).toEqual(prior);
+  });
+  it('rejects invalid dates and unverified sources', () => {
+    expect(() => wireTiming({ firstWireSubmitAt: 'invalid', wireTimeSource: 'gateway_write_complete' })).toThrow();
+    expect(() => wireTiming({ firstWireSubmitAt: '2026-09-06T04:00:00Z', wireTimeSource: 'response_time' })).toThrow();
+  });
+});

api/src/send-chain/send-chain.contracts.ts

@@ -70,6 +70,9 @@ export interface GatewaySubmitResultDto {
   errorCode?: string;
   errorMessage?: string;
   submittedAt?: string;
+  firstWireSubmitAt?: string;
+  wireTimeSource?: string;
+  receiptRequested?: boolean;
   segments?: Array<{
     segmentTotal?: number;
     segmentIndex?: number;
@@ -79,6 +82,9 @@ export interface GatewaySubmitResultDto {
     errorCode?: string;
     errorMessage?: string;
     submittedAt?: string;
+    firstWireSubmitAt?: string;
+    wireTimeSource?: string;
+    receiptRequested?: boolean;
   }>;
 }
 
@@ -96,6 +102,9 @@ export interface GatewaySubmitSegmentResultDto {
   errorCode?: string;
   errorMessage?: string;
   submittedAt?: string;
+  firstWireSubmitAt?: string;
+  wireTimeSource?: string;
+  receiptRequested?: boolean;
 }
 
 export interface GatewayReceiptEventDto {

api/src/send-chain/send-gateway-result.service.ts

@@ -1,20 +1,36 @@
-import { BadRequestException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common';
-import { Prisma } from '@prisma/client';
-import { createHash } from 'node:crypto';
+import { BadRequestException, Logger, NotFoundException } from '@nestjs/common';
+
 import { BillingService } from '../billing/billing.service';
-import { moneyToNumber } from '../common/money';
+
 import type { OpenApiService } from '../open-api/open-api.service';
 import { PrismaService } from '../prisma/prisma.service';
-import type { GatewaySubmitResultDto, GatewaySubmitSegmentResultDto, GatewayReceiptEventDto, GatewayUplinkEventDto, GatewayPendingDeliveryQueryDto, GatewayDownstreamSentDto, GatewayDownstreamAcknowledgedDto, GatewayDownstreamFailureType, GatewayControlDeliveryResult, GatewaySubmitDeadLetterDto, RequeueGatewaySubmitExceptionDto, GatewayDownstreamRecoveryStatusDto, TimeoutUnknownDto } from './send-chain.contracts';
-import { positiveInteger, normalizeReceiptStatus, normalizeCarrier, DEFAULT_DOWNSTREAM_RETRY_DELAY_MS, DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS, DEFAULT_DOWNSTREAM_MAX_RETRIES, DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS, DEFAULT_RECEIPT_TIMEOUT_HOURS, DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS, DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS, DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS, gatewaySubmitRequeueKey, isObjectRecord, asDateOrNull, downstreamRetryDelayMs, downstreamAckTimeoutMs, downstreamRetryBaseDelayMs, downstreamRetryMaxDelayMs, downstreamMaxRetries, downstreamPendingTimeoutHours, downstreamControlFailureMessage, normalizeSubmitStatus, downstreamDeliveryAttemptKey, hasRecoveryAuditStateChanged, normalizeRecoveryFailureCategory, aggregateReceiptSegmentState, isSameUpstreamEndpointIdentity, receiptEventKey } from './send-chain.helpers';
-import type { SendSubmissionService } from './send-submission.service';
-import type { SendCompletionCallbacks, SendCompletionFacade } from './send-completion.service';
+import type { GatewaySubmitResultDto, GatewaySubmitSegmentResultDto } from './send-chain.contracts';
+import { normalizeSubmitStatus } from './send-chain.helpers';
 
+import type { SendCompletionCallbacks, SendCompletionFacade } from './send-completion.service';
 
 /**
  * R10 gatewayResult implementation.
  * Cross-domain calls return through the stable SendChainService seam to preserve locking and test observability.
  */
+export function wireTiming(
+  data: { firstWireSubmitAt?: string; wireTimeSource?: string; receiptRequested?: boolean },
+  previous?: Date | null,
+): { firstWireSubmitAt?: Date; wireTimeSource?: string; receiptRequested?: boolean } {
+  const receipt = typeof data.receiptRequested === 'boolean' ? { receiptRequested: data.receiptRequested } : {};
+  if (!data.firstWireSubmitAt)
+    return data.wireTimeSource ? { ...receipt, wireTimeSource: data.wireTimeSource } : receipt;
+  const at = new Date(data.firstWireSubmitAt);
+  if (!Number.isFinite(at.getTime()) || data.wireTimeSource !== 'gateway_write_complete') {
+    throw new BadRequestException('Invalid Gateway wire timestamp');
+  }
+  return {
+    ...receipt,
+    firstWireSubmitAt: previous && previous < at ? previous : at,
+    wireTimeSource: data.wireTimeSource,
+  };
+}
+
 export class SendGatewayResultService {
   private readonly logger = new Logger('SendChainService');
 
@@ -31,27 +47,39 @@ export class SendGatewayResultService {
     const submitRecord = await this.facade.resolveSubmitRecordForGatewaySegmentResult(message.id, data);
     const effectiveSubmitId = submitRecord.submitId;
     const submittedAt = data.submittedAt ? new Date(data.submittedAt) : new Date();
-    await this.facade.recordSubmitSegments(message, {
-      messageId: data.messageId,
-      channelId: data.channelId,
-      submitId: effectiveSubmitId,
-      sequenceId: data.sequenceId,
-      gatewayMessageId: data.gatewayMessageId ?? '',
-      submitStatus: normalizeSubmitStatus(data.submitStatus),
-      errorCode: data.errorCode,
-      errorMessage: data.errorMessage,
-      submittedAt: submittedAt.toISOString(),
-      segments: [{
-        segmentTotal: data.segmentTotal,
-        segmentIndex: data.segmentIndex,
+    await this.facade.recordSubmitSegments(
+      message,
+      {
+        messageId: data.messageId,
+        channelId: data.channelId,
+        submitId: effectiveSubmitId,
         sequenceId: data.sequenceId,
-        gatewayMessageId: data.gatewayMessageId,
-        submitStatus: data.submitStatus,
+        gatewayMessageId: data.gatewayMessageId ?? '',
+        submitStatus: normalizeSubmitStatus(data.submitStatus),
         errorCode: data.errorCode,
         errorMessage: data.errorMessage,
         submittedAt: submittedAt.toISOString(),
-      }],
-    }, submittedAt);
+        firstWireSubmitAt: data.firstWireSubmitAt,
+        wireTimeSource: data.wireTimeSource,
+        receiptRequested: data.receiptRequested,
+        segments: [
+          {
+            segmentTotal: data.segmentTotal,
+            segmentIndex: data.segmentIndex,
+            sequenceId: data.sequenceId,
+            gatewayMessageId: data.gatewayMessageId,
+            submitStatus: data.submitStatus,
+            errorCode: data.errorCode,
+            errorMessage: data.errorMessage,
+            submittedAt: submittedAt.toISOString(),
+            firstWireSubmitAt: data.firstWireSubmitAt,
+            wireTimeSource: data.wireTimeSource,
+            receiptRequested: data.receiptRequested,
+          },
+        ],
+      },
+      submittedAt,
+    );
     if (data.gatewayMessageId) {
       await this.prisma.smsSubmitRecord.updateMany({
         where: {
@@ -68,10 +96,7 @@ export class SendGatewayResultService {
     return { accepted: true };
   }
 
-  async resolveSubmitRecordForGatewaySegmentResult(
-    messageRecordId: string,
-    data: GatewaySubmitSegmentResultDto,
-  ) {
+  async resolveSubmitRecordForGatewaySegmentResult(messageRecordId: string, data: GatewaySubmitSegmentResultDto) {
     if (data.submitId) {
       const exact = await this.prisma.smsSubmitRecord.findUnique({
         where: { submitId: data.submitId },
@@ -81,13 +106,15 @@ export class SendGatewayResultService {
         (exact.messageRecordId && exact.messageRecordId !== messageRecordId) ||
         (exact.channelId && exact.channelId !== data.channelId)
       ) {
-        this.logger.error(`gateway_submit_segment_result_unmatched ${JSON.stringify({
-          messageId: data.messageId,
-          messageRecordId,
-          submitId: data.submitId,
-          channelId: data.channelId,
-          segmentIndex: data.segmentIndex,
-        })}`);
+        this.logger.error(
+          `gateway_submit_segment_result_unmatched ${JSON.stringify({
+            messageId: data.messageId,
+            messageRecordId,
+            submitId: data.submitId,
+            channelId: data.channelId,
+            segmentIndex: data.segmentIndex,
+          })}`,
+        );
         throw new BadRequestException(
           'Gateway SubmitSegmentResult submitId does not match the SMS message and channel',
         );
@@ -104,24 +131,26 @@ export class SendGatewayResultService {
       take: 2,
     });
     if (candidates.length !== 1) {
-      this.logger.error(`gateway_submit_segment_result_unmatched ${JSON.stringify({
+      this.logger.error(
+        `gateway_submit_segment_result_unmatched ${JSON.stringify({
+          messageId: data.messageId,
+          messageRecordId,
+          channelId: data.channelId,
+          segmentIndex: data.segmentIndex,
+          candidateCount: candidates.length,
+        })}`,
+      );
+      throw new BadRequestException('Gateway SubmitSegmentResult without submitId cannot be matched uniquely');
+    }
+    this.logger.warn(
+      `gateway_submit_segment_result_legacy_match ${JSON.stringify({
         messageId: data.messageId,
         messageRecordId,
         channelId: data.channelId,
         segmentIndex: data.segmentIndex,
-        candidateCount: candidates.length,
-      })}`);
-      throw new BadRequestException(
-        'Gateway SubmitSegmentResult without submitId cannot be matched uniquely',
-      );
-    }
-    this.logger.warn(`gateway_submit_segment_result_legacy_match ${JSON.stringify({
-      messageId: data.messageId,
-      messageRecordId,
-      channelId: data.channelId,
-      segmentIndex: data.segmentIndex,
-      submitId: candidates[0].submitId,
-    })}`);
+        submitId: candidates[0].submitId,
+      })}`,
+    );
     return candidates[0];
   }
 
@@ -148,6 +177,7 @@ export class SendGatewayResultService {
         errorCode: data.errorCode,
         errorMessage: data.errorMessage,
         submittedAt,
+        ...wireTiming(data, submitRecord.firstWireSubmitAt),
       },
     });
     await this.facade.recordSubmitSegments(message, effectiveData, submittedAt);
@@ -156,7 +186,8 @@ export class SendGatewayResultService {
       await this.markSubmitResultProcessed(submitRecord.id, data.eventId, submittedAt);
       return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
     }
-    const status = data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed';
+    const status =
+      data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed';
     if (data.submitStatus === 'accepted' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) {
       const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
       await this.facade.chargeAcceptedMessage(businessMessage);
@@ -164,7 +195,12 @@ export class SendGatewayResultService {
       if (latest?.status === 'failed') {
         await this.facade.refundMessage(businessMessage, '先到失败回执补偿退款');
       }
-    } else if (data.submitStatus !== 'accepted' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) {
+    } else if (
+      data.submitStatus !== 'accepted' &&
+      !isStandaloneChannelTest &&
+      message.tenantId &&
+      message.batchTaskId
+    ) {
       const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
       const retried = await this.facade.retryMessageIfAllowed(
         businessMessage,
@@ -176,13 +212,17 @@ export class SendGatewayResultService {
         await this.markSubmitResultProcessed(submitRecord.id, data.eventId, submittedAt);
         return retried;
       }
-      await this.facade.releaseMessageReservation(businessMessage, data.submitStatus === 'timeout' ? '提交超时释放冻结' : '提交失败释放冻结');
+      await this.facade.releaseMessageReservation(
+        businessMessage,
+        data.submitStatus === 'timeout' ? '提交超时释放冻结' : '提交失败释放冻结',
+      );
     }
     const protectedTerminalStatuses = ['delivered', 'failed', 'unknown'];
     const updated = await this.prisma.smsMessageRecord.updateMany({
-      where: data.submitStatus === 'accepted'
-        ? { id: message.id, status: { notIn: protectedTerminalStatuses } }
-        : { id: message.id, status: { not: 'delivered' } },
+      where:
+        data.submitStatus === 'accepted'
+          ? { id: message.id, status: { notIn: protectedTerminalStatuses } }
+          : { id: message.id, status: { not: 'delivered' } },
       data: {
         gatewayMessageId: data.gatewayMessageId,
         submitStatus: data.submitStatus,
@@ -202,7 +242,12 @@ export class SendGatewayResultService {
         },
       });
     }
-    if (data.submitStatus !== 'accepted' && batchTask?.sourceType === 'cmpp' && message.tenantId && message.applicationId) {
+    if (
+      data.submitStatus !== 'accepted' &&
+      batchTask?.sourceType === 'cmpp' &&
+      message.tenantId &&
+      message.applicationId
+    ) {
       await this.facade.recordCmppFailureReceipt(
         message as typeof message & { tenantId: string; applicationId: string; batchTaskId: string },
         data.errorCode || 'SUBMIT',
@@ -212,10 +257,9 @@ export class SendGatewayResultService {
     await this.prisma.gatewaySubmitDeadLetter.updateMany({
       where: {
         status: { in: ['pending', 'requeueing', 'requeue_recovering', 'requeued'] },
-        OR: [
-          { submitId: effectiveData.submitId },
-          data.messageId ? { messageId: data.messageId } : undefined,
-        ].filter(Boolean) as Array<{ submitId?: string; messageId?: string }>,
+        OR: [{ submitId: effectiveData.submitId }, data.messageId ? { messageId: data.messageId } : undefined].filter(
+          Boolean,
+        ) as Array<{ submitId?: string; messageId?: string }>,
       },
       data: {
         status: 'resolved',
@@ -252,9 +296,11 @@ export class SendGatewayResultService {
   async resolveSubmitRecordForGatewayResult(messageRecordId: string, data: GatewaySubmitResultDto) {
     if (data.submitId) {
       const exact = await this.prisma.smsSubmitRecord.findUnique({ where: { submitId: data.submitId } });
-      if (!exact
-        || (exact.messageRecordId && exact.messageRecordId !== messageRecordId)
-        || (exact.channelId && exact.channelId !== data.channelId)) {
+      if (
+        !exact ||
+        (exact.messageRecordId && exact.messageRecordId !== messageRecordId) ||
+        (exact.channelId && exact.channelId !== data.channelId)
+      ) {
         throw new BadRequestException('Gateway SubmitResult submitId does not match the SMS message and channel');
       }
       return exact;
@@ -272,34 +318,40 @@ export class SendGatewayResultService {
       take: 2,
     });
     if (candidates.length !== 1) {
-      this.logger.error(`gateway_submit_result_unmatched ${JSON.stringify({
+      this.logger.error(
+        `gateway_submit_result_unmatched ${JSON.stringify({
+          messageId: data.messageId,
+          messageRecordId,
+          channelId: data.channelId,
+          gatewayMessageId: data.gatewayMessageId,
+          candidateCount: candidates.length,
+        })}`,
+      );
+      throw new BadRequestException('Gateway SubmitResult without submitId cannot be matched uniquely');
+    }
+    this.logger.warn(
+      `gateway_submit_result_legacy_match ${JSON.stringify({
         messageId: data.messageId,
         messageRecordId,
         channelId: data.channelId,
         gatewayMessageId: data.gatewayMessageId,
-        candidateCount: candidates.length,
-      })}`);
-      throw new BadRequestException('Gateway SubmitResult without submitId cannot be matched uniquely');
-    }
-    this.logger.warn(`gateway_submit_result_legacy_match ${JSON.stringify({
-      messageId: data.messageId,
-      messageRecordId,
-      channelId: data.channelId,
-      gatewayMessageId: data.gatewayMessageId,
-      submitId: candidates[0].submitId,
-    })}`);
+        submitId: candidates[0].submitId,
+      })}`,
+    );
     return candidates[0];
   }
 
   smsMessageSegmentAuditDelegate() {
-    return (this.prisma as PrismaService & {
-      smsMessageSegmentAudit: {
-        upsert: (args: Record<string, unknown>) => Promise<any>;
-        updateMany: (args: Record<string, unknown>) => Promise<{ count: number }>;
-        findFirst: (args: Record<string, unknown>) => Promise<any | null>;
-        findMany: (args: Record<string, unknown>) => Promise<any[]>;
-      };
-    }).smsMessageSegmentAudit;
+    return (
+      this.prisma as PrismaService & {
+        smsMessageSegmentAudit: {
+          upsert: (args: Record<string, unknown>) => Promise<any>;
+          updateMany: (args: Record<string, unknown>) => Promise<{ count: number }>;
+          findFirst: (args: Record<string, unknown>) => Promise<any | null>;
+          findMany: (args: Record<string, unknown>) => Promise<any[]>;
+        };
+      }
+    ).smsMessageSegmentAudit;
   }
 
   async recordSubmitSegments(
@@ -327,74 +379,84 @@ export class SendGatewayResultService {
     });
     const submitId = data.submitId ?? submitRecord?.submitId ?? message.submitId ?? `SUB-AUDIT-${message.id}`;
     const attempt = submitRecord
-      ? Math.max(0, await this.prisma.smsSubmitRecord.count({
-        where: {
-          messageRecordId: message.id,
-          createdAt: { lte: submitRecord.createdAt },
-        },
-      }) - 1)
+      ? Math.max(
+          0,
+          (await this.prisma.smsSubmitRecord.count({
+            where: {
+              messageRecordId: message.id,
+              createdAt: { lte: submitRecord.createdAt },
+            },
+          })) - 1,
+        )
       : 0;
-    const fallbackSegments = [{
-      segmentTotal: Math.max(1, Number(message.billingUnits ?? 1)),
-      segmentIndex: 1,
-      sequenceId: data.sequenceId,
-      gatewayMessageId: data.gatewayMessageId,
-      submitStatus: data.submitStatus,
-      errorCode: data.errorCode,
-      errorMessage: data.errorMessage,
-      submittedAt: data.submittedAt,
-    }];
+    const fallbackSegments = [
+      {
+        firstWireSubmitAt: data.firstWireSubmitAt,
+        wireTimeSource: data.wireTimeSource,
+        receiptRequested: data.receiptRequested,
+        segmentTotal: Math.max(1, Number(message.billingUnits ?? 1)),
+        segmentIndex: 1,
+        sequenceId: data.sequenceId,
+        gatewayMessageId: data.gatewayMessageId,
+        submitStatus: data.submitStatus,
+        errorCode: data.errorCode,
+        errorMessage: data.errorMessage,
+        submittedAt: data.submittedAt,
+      },
+    ];
     const segments = data.segments && data.segments.length > 0 ? data.segments : fallbackSegments;
     const segmentTotal = Math.max(1, ...segments.map((item) => Number(item.segmentTotal ?? segments.length ?? 1)));
-    await Promise.all(segments.map((segment, index) => {
-      const segmentIndex = Math.max(1, Number(segment.segmentIndex ?? index + 1));
-      const status = segment.submitStatus ?? data.submitStatus;
-      return segmentAudits.upsert({
-        where: {
-          messageRecordId_submitId_segmentIndex: {
-            messageRecordId: message.id,
-            submitId,
-            segmentIndex,
+    await Promise.all(
+      segments.map((segment, index) => {
+        const segmentIndex = Math.max(1, Number(segment.segmentIndex ?? index + 1));
+        const status = segment.submitStatus ?? data.submitStatus;
+        return segmentAudits.upsert({
+          where: {
+            messageRecordId_submitId_segmentIndex: {
+              messageRecordId: message.id,
+              submitId,
+              segmentIndex,
+            },
           },
-        },
-        update: {
-          submitRecordId: submitRecord?.id ?? null,
-          channelId: data.channelId ?? message.channelId ?? null,
-          attempt,
-          segmentTotal,
-          sequenceId: segment.sequenceId ?? data.sequenceId ?? null,
-          gatewayMessageId: segment.gatewayMessageId ?? data.gatewayMessageId ?? null,
-          submitStatus: status,
-          errorCode: segment.errorCode ?? data.errorCode ?? null,
-          errorMessage: segment.errorMessage ?? data.errorMessage ?? null,
-          submittedAt: segment.submittedAt ? new Date(segment.submittedAt) : submittedAt,
-        },
-        create: {
-          tenantId: message.tenantId,
-          batchTaskId: message.batchTaskId,
-          messageRecordId: message.id,
-          submitRecordId: submitRecord?.id ?? null,
-          channelId: data.channelId ?? message.channelId ?? null,
-          submitId,
-          attempt,
-          segmentTotal,
-          segmentIndex,
-          sequenceId: segment.sequenceId ?? data.sequenceId ?? null,
-          gatewayMessageId: segment.gatewayMessageId ?? data.gatewayMessageId ?? null,
-          submitStatus: status,
-          compensationType: submitRecord && submitRecord.submitId !== message.submitId ? 'retry_submit' : null,
-          errorCode: segment.errorCode ?? data.errorCode ?? null,
-          errorMessage: segment.errorMessage ?? data.errorMessage ?? null,
-          submittedAt: segment.submittedAt ? new Date(segment.submittedAt) : submittedAt,
-        },
-      });
-    }));
+          update: {
+            ...wireTiming(segment),
+            submitRecordId: submitRecord?.id ?? null,
+            channelId: data.channelId ?? message.channelId ?? null,
+            attempt,
+            segmentTotal,
+            sequenceId: segment.sequenceId ?? data.sequenceId ?? null,
+            gatewayMessageId: segment.gatewayMessageId ?? data.gatewayMessageId ?? null,
+            submitStatus: status,
+            errorCode: segment.errorCode ?? data.errorCode ?? null,
+            errorMessage: segment.errorMessage ?? data.errorMessage ?? null,
+            submittedAt: segment.submittedAt ? new Date(segment.submittedAt) : submittedAt,
+          },
+          create: {
+            ...wireTiming(segment),
+            tenantId: message.tenantId,
+            batchTaskId: message.batchTaskId,
+            messageRecordId: message.id,
+            submitRecordId: submitRecord?.id ?? null,
+            channelId: data.channelId ?? message.channelId ?? null,
+            submitId,
+            attempt,
+            segmentTotal,
+            segmentIndex,
+            sequenceId: segment.sequenceId ?? data.sequenceId ?? null,
+            gatewayMessageId: segment.gatewayMessageId ?? data.gatewayMessageId ?? null,
+            submitStatus: status,
+            compensationType: submitRecord && submitRecord.submitId !== message.submitId ? 'retry_submit' : null,
+            errorCode: segment.errorCode ?? data.errorCode ?? null,
+            errorMessage: segment.errorMessage ?? data.errorMessage ?? null,
+            submittedAt: segment.submittedAt ? new Date(segment.submittedAt) : submittedAt,
+          },
+        });
+      }),
+    );
   }
 
   async findMessageByGatewayEvent(messageId?: string, gatewayMessageId?: string) {
-    const conditions = [{ messageId }, gatewayMessageId ? { gatewayMessageId } : undefined].filter(
-      Boolean,
-    ) as Array<{
+    const conditions = [{ messageId }, gatewayMessageId ? { gatewayMessageId } : undefined].filter(Boolean) as Array<{
       messageId?: string;
       gatewayMessageId?: string;
     }>;

api/src/send-chain/send-receipt.service.ts

@@ -1,17 +1,26 @@
-import { BadRequestException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common';
+import { Logger, NotFoundException } from '@nestjs/common';
 import { Prisma } from '@prisma/client';
-import { createHash } from 'node:crypto';
+
 import { BillingService } from '../billing/billing.service';
-import { moneyToNumber } from '../common/money';
+
 import type { OpenApiService } from '../open-api/open-api.service';
 import { PrismaService } from '../prisma/prisma.service';
-import type { GatewaySubmitResultDto, GatewaySubmitSegmentResultDto, GatewayReceiptEventDto, GatewayUplinkEventDto, GatewayPendingDeliveryQueryDto, GatewayDownstreamSentDto, GatewayDownstreamAcknowledgedDto, GatewayDownstreamFailureType, GatewayControlDeliveryResult, GatewaySubmitDeadLetterDto, RequeueGatewaySubmitExceptionDto, GatewayDownstreamRecoveryStatusDto, TimeoutUnknownDto } from './send-chain.contracts';
-import { positiveInteger, normalizeReceiptStatus, normalizeCarrier, DEFAULT_DOWNSTREAM_RETRY_DELAY_MS, DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS, DEFAULT_DOWNSTREAM_MAX_RETRIES, DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS, DEFAULT_RECEIPT_TIMEOUT_HOURS, DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS, DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS, DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS, gatewaySubmitRequeueKey, isObjectRecord, asDateOrNull, downstreamRetryDelayMs, downstreamAckTimeoutMs, downstreamRetryBaseDelayMs, downstreamRetryMaxDelayMs, downstreamMaxRetries, downstreamPendingTimeoutHours, downstreamControlFailureMessage, normalizeSubmitStatus, downstreamDeliveryAttemptKey, hasRecoveryAuditStateChanged, normalizeRecoveryFailureCategory, aggregateReceiptSegmentState, isSameUpstreamEndpointIdentity, receiptEventKey, longMessageReceiptMode } from './send-chain.helpers';
-import type { SendSubmissionService } from './send-submission.service';
+import type { GatewayReceiptEventDto } from './send-chain.contracts';
+import {
+  positiveInteger,
+  normalizeReceiptStatus,
+  DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS,
+  DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS,
+  DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS,
+  aggregateReceiptSegmentState,
+  isSameUpstreamEndpointIdentity,
+  receiptEventKey,
+  longMessageReceiptMode,
+} from './send-chain.helpers';
+
 import type { SendCompletionCallbacks, SendCompletionFacade } from './send-completion.service';
 import { queueFinalReceiptDeliveries } from './downstream-receipt-targets';
 
-
 /**
  * R10 receipt implementation.
  * Cross-domain calls return through the stable SendChainService seam to preserve locking and test observability.
@@ -68,6 +77,7 @@ export class SendReceiptService {
         errorCode: data.errorCode,
         errorMessage: data.errorMessage,
         deliveredAt,
+        gatewayReceivedAt: data.deliveredAt ? new Date(data.deliveredAt) : null,
         status: 'pending',
         nextRetryAt: new Date(),
       },
@@ -81,8 +91,8 @@ export class SendReceiptService {
   async processPendingUpstreamReceiptInbox(limit = 100) {
     const now = new Date();
     const staleBefore = new Date(
-      now.getTime()
-        - positiveInteger(
+      now.getTime() -
+        positiveInteger(
           process.env.UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS,
           DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS,
         ),
@@ -110,8 +120,8 @@ export class SendReceiptService {
 
   async processUpstreamReceiptInboxRecord(id: string) {
     const staleBefore = new Date(
-      Date.now()
-        - positiveInteger(
+      Date.now() -
+        positiveInteger(
           process.env.UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS,
           DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS,
         ),
@@ -119,10 +129,7 @@ export class SendReceiptService {
     const claimed = await this.prisma.upstreamReceiptInbox.updateMany({
       where: {
         id,
-        OR: [
-          { status: { in: ['pending', 'retrying'] } },
-          { status: 'processing', updatedAt: { lte: staleBefore } },
-        ],
+        OR: [{ status: { in: ['pending', 'retrying'] } }, { status: 'processing', updatedAt: { lte: staleBefore } }],
       },
       data: { status: 'processing', attemptCount: { increment: 1 }, nextRetryAt: null },
     });
@@ -130,25 +137,28 @@ export class SendReceiptService {
     const inbox = await this.prisma.upstreamReceiptInbox.findUnique({ where: { id } });
     if (!inbox) return false;
     try {
-      const message = await this.facade.handleReceipt({
-        messageId: inbox.provisionalMessageId ?? undefined,
-        channelId: inbox.incomingChannelId,
-        connectionId: inbox.incomingConnectionId ?? undefined,
-        sequenceId: inbox.sequenceId ?? undefined,
-        gatewayMessageId: inbox.gatewayMessageId,
-        phoneNumber: inbox.phoneNumber ?? undefined,
-        receiptStatus: normalizeReceiptStatus(inbox.receiptStatus),
-        rawStatus: inbox.rawStatus,
-        errorCode: inbox.errorCode ?? undefined,
-        errorMessage: inbox.errorMessage ?? undefined,
-        deliveredAt: inbox.deliveredAt.toISOString(),
-      }, {
-        account: inbox.upstreamAccount,
-        gatewayHost: inbox.upstreamHost,
-        gatewayPort: inbox.upstreamPort,
-        protocol: inbox.protocol,
-        cmppVersion: inbox.protocolVersion,
-      });
+      const message = await this.facade.handleReceipt(
+        {
+          messageId: inbox.provisionalMessageId ?? undefined,
+          channelId: inbox.incomingChannelId,
+          connectionId: inbox.incomingConnectionId ?? undefined,
+          sequenceId: inbox.sequenceId ?? undefined,
+          gatewayMessageId: inbox.gatewayMessageId,
+          phoneNumber: inbox.phoneNumber ?? undefined,
+          receiptStatus: normalizeReceiptStatus(inbox.receiptStatus),
+          rawStatus: inbox.rawStatus,
+          errorCode: inbox.errorCode ?? undefined,
+          errorMessage: inbox.errorMessage ?? undefined,
+          deliveredAt: inbox.deliveredAt.toISOString(),
+        },
+        {
+          account: inbox.upstreamAccount,
+          gatewayHost: inbox.upstreamHost,
+          gatewayPort: inbox.upstreamPort,
+          protocol: inbox.protocol,
+          cmppVersion: inbox.protocolVersion,
+        },
+      );
       const matchedMessageRecordId = message && 'id' in message ? message.id : message?.messageRecordId;
       const matchedChannelId = message && 'channelId' in message ? message.channelId : undefined;
       await this.prisma.upstreamReceiptInbox.update({
@@ -171,8 +181,8 @@ export class SendReceiptService {
         process.env.UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS,
         DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS,
       );
-      const exhausted = inbox.attemptCount >= maxAttempts
-        || inbox.receivedAt.getTime() <= Date.now() - maxAgeHours * 60 * 60_000;
+      const exhausted =
+        inbox.attemptCount >= maxAttempts || inbox.receivedAt.getTime() <= Date.now() - maxAgeHours * 60 * 60_000;
       const retryDelayMs = Math.min(30 * 60_000, 5_000 * 2 ** Math.min(Math.max(inbox.attemptCount - 1, 0), 8));
       await this.prisma.upstreamReceiptInbox.update({
         where: { id },
@@ -204,7 +214,13 @@ export class SendReceiptService {
 
   async handleReceipt(
     data: GatewayReceiptEventDto,
-    incomingIdentity?: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string },
+    incomingIdentity?: {
+      account: string;
+      gatewayHost: string;
+      gatewayPort: number;
+      protocol: string;
+      cmppVersion: string;
+    },
   ) {
     const resolved = await this.facade.resolveReceiptMessage(data, incomingIdentity);
     const logicalChannelId = resolved.channelId ?? data.channelId;
@@ -263,9 +279,8 @@ export class SendReceiptService {
     }
     const logicalReceipt = { ...data, channelId: logicalChannelId };
     await this.facade.recordReceiptSegment(message, logicalReceipt, deliveredAt, resolved.submitRecordId);
-    const receiptMode = Number(message.billingUnits ?? 1) > 1
-      ? await this.getLongMessageReceiptMode(logicalChannelId)
-      : 'per_segment';
+    const receiptMode =
+      Number(message.billingUnits ?? 1) > 1 ? await this.getLongMessageReceiptMode(logicalChannelId) : 'per_segment';
     if (receiptMode === 'message_level' && data.receiptStatus === 'delivered') {
       await this.applyMessageLevelSuccess(
         message,
@@ -287,12 +302,10 @@ export class SendReceiptService {
     }
     const status = aggregate.status;
     const isCurrentAttempt =
-      (!message.channelId || message.channelId === logicalChannelId)
-      && (
-        !message.gatewayMessageId
-        || message.gatewayMessageId === data.gatewayMessageId
-        || (aggregate.segmentTotal > 1 && (!message.submitId || message.submitId === resolved.submitId))
-      );
+      (!message.channelId || message.channelId === logicalChannelId) &&
+      (!message.gatewayMessageId ||
+        message.gatewayMessageId === data.gatewayMessageId ||
+        (aggregate.segmentTotal > 1 && (!message.submitId || message.submitId === resolved.submitId)));
     if (!isCurrentAttempt) {
       return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
     }
@@ -314,11 +327,7 @@ export class SendReceiptService {
     const isStandaloneChannelTest = !message.tenantId && !message.batchTaskId;
     if (status === 'failed' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) {
       const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
-      const retried = await this.facade.retryMessageIfAllowed(
-        businessMessage,
-        '回执失败补发',
-        resolved.submitRecordId,
-      );
+      const retried = await this.facade.retryMessageIfAllowed(businessMessage, '回执失败补发', resolved.submitRecordId);
       if (retried) {
         await this.facade.refreshTaskProgress(businessMessage.batchTaskId);
         return retried;
@@ -339,32 +348,31 @@ export class SendReceiptService {
       },
     });
     if (!isStandaloneChannelTest && message.tenantId && message.applicationId) {
-      await queueFinalReceiptDeliveries(
-        this.prisma,
-        (request) => this.facade.queueAndTryDownstreamDelivery(request),
-        {
-          message,
-          payload: {
-            messageId: message.messageId,
-            gatewayMessageId: data.gatewayMessageId,
-            phoneNumber: message.phoneNumber,
-            receiptStatus: aggregate.receiptStatus,
-            rawStatus: aggregate.rawStatus,
-            errorCode: aggregate.errorCode,
-            deliveredAt: aggregate.deliveredAt.toISOString(),
-          },
-          segmentPayloads: Object.fromEntries(
-            aggregate.segments
-              .filter((segment) => segment.receiptStatus)
-              .map((segment) => [segment.segmentIndex, {
+      await queueFinalReceiptDeliveries(this.prisma, (request) => this.facade.queueAndTryDownstreamDelivery(request), {
+        message,
+        payload: {
+          messageId: message.messageId,
+          gatewayMessageId: data.gatewayMessageId,
+          phoneNumber: message.phoneNumber,
+          receiptStatus: aggregate.receiptStatus,
+          rawStatus: aggregate.rawStatus,
+          errorCode: aggregate.errorCode,
+          deliveredAt: aggregate.deliveredAt.toISOString(),
+        },
+        segmentPayloads: Object.fromEntries(
+          aggregate.segments
+            .filter((segment) => segment.receiptStatus)
+            .map((segment) => [
+              segment.segmentIndex,
+              {
                 receiptStatus: segment.receiptStatus,
                 rawStatus: segment.rawStatus,
                 errorCode: segment.errorCode,
                 deliveredAt: segment.deliveredAt?.toISOString() ?? aggregate.deliveredAt.toISOString(),
-              }]),
-          ),
-        },
-      );
+              },
+            ]),
+        ),
+      });
     }
     if (message.batchTaskId) {
       await this.facade.refreshTaskProgress(message.batchTaskId);
@@ -388,8 +396,9 @@ export class SendReceiptService {
     submitRecordId?: string,
     submitId?: string,
   ) {
-    const belongsToCurrentAttempt = (!message.channelId || message.channelId === data.channelId)
-      && (!message.submitId || message.submitId === submitId);
+    const belongsToCurrentAttempt =
+      (!message.channelId || message.channelId === data.channelId) &&
+      (!message.submitId || message.submitId === submitId);
     if (!belongsToCurrentAttempt) return;
     const attemptWhere = submitRecordId
       ? { messageRecordId: message.id, submitRecordId }
@@ -402,7 +411,9 @@ export class SendReceiptService {
       select: { id: true, receiptStatus: true },
     });
     if (segments.length <= 1) return;
-    if (segments.some((segment) => segment.receiptStatus && !['delivered', 'unknown'].includes(segment.receiptStatus))) {
+    if (
+      segments.some((segment) => segment.receiptStatus && !['delivered', 'unknown'].includes(segment.receiptStatus))
+    ) {
       return;
     }
     // This supplier contract reports one message-level success for a multipart SMS.
@@ -506,9 +517,9 @@ export class SendReceiptService {
     const submitRecord = submitRecordId
       ? await this.prisma.smsSubmitRecord.findUnique({ where: { id: submitRecordId } })
       : await this.prisma.smsSubmitRecord.findFirst({
-        where: { messageRecordId: message.id, gatewayMessageId: data.gatewayMessageId },
-        orderBy: { createdAt: 'desc' },
-      });
+          where: { messageRecordId: message.id, gatewayMessageId: data.gatewayMessageId },
+          orderBy: { createdAt: 'desc' },
+        });
     await segmentAudits.upsert({
       where: {
         messageRecordId_submitId_segmentIndex: {
@@ -575,7 +586,13 @@ export class SendReceiptService {
 
   async resolveReceiptMessage(
     data: GatewayReceiptEventDto,
-    incomingIdentity?: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string },
+    incomingIdentity?: {
+      account: string;
+      gatewayHost: string;
+      gatewayPort: number;
+      protocol: string;
+      cmppVersion: string;
+    },
   ) {
     const exactMessage = data.messageId
       ? await this.prisma.smsMessageRecord.findUnique({ where: { messageId: data.messageId } })
@@ -641,8 +658,8 @@ export class SendReceiptService {
       throw new NotFoundException('SMS message record not found');
     }
 
-    const incomingChannel = incomingIdentity
-      ?? await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } });
+    const incomingChannel =
+      incomingIdentity ?? (await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } }));
     if (!incomingChannel) {
       throw new NotFoundException('SMS message record not found');
     }
@@ -665,8 +682,9 @@ export class SendReceiptService {
         channelId: exactSegmentMatches[0].channelId,
       };
     }
-    const sameSupplierSegments = segmentMatches.filter((candidate) =>
-      candidate.channel && isSameUpstreamEndpointIdentity(incomingChannel, candidate.channel));
+    const sameSupplierSegments = segmentMatches.filter(
+      (candidate) => candidate.channel && isSameUpstreamEndpointIdentity(incomingChannel, candidate.channel),
+    );
     if (sameSupplierSegments.length === 1 && sameSupplierSegments[0]?.messageRecord) {
       return {
         message: sameSupplierSegments[0].messageRecord,
@@ -685,8 +703,9 @@ export class SendReceiptService {
       orderBy: { createdAt: 'desc' },
       take: 10,
     });
-    const sameSupplierSubmits = crossConnectionSubmits.filter((candidate) =>
-      candidate.channel && isSameUpstreamEndpointIdentity(incomingChannel, candidate.channel));
+    const sameSupplierSubmits = crossConnectionSubmits.filter(
+      (candidate) => candidate.channel && isSameUpstreamEndpointIdentity(incomingChannel, candidate.channel),
+    );
     if (sameSupplierSubmits.length === 1 && sameSupplierSubmits[0]?.messageRecord) {
       return {
         message: sameSupplierSubmits[0].messageRecord,

api/src/sending-monitor-worker.ts

@@ -0,0 +1,201 @@
+import { Pool, PoolClient } from 'pg';
+import { projectMessages } from './sending-monitor/monitor-projection';
+import { evaluateWindow } from './sending-monitor/monitor-evaluation';
+import type { MonitorType } from './sending-monitor/monitor-metrics';
+
+process.env.TZ = 'UTC';
+type Cursor = { from: string; to: string; at: string; id: string; completeAt?: string };
+async function checkpoint(db: PoolClient, id: string, data: unknown) {
+  await db.query(
+    `INSERT INTO "SendingMonitorCheckpoint" (id,data) VALUES($1,$2::jsonb) ON CONFLICT(id) DO UPDATE SET data=EXCLUDED.data,"updatedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')`,
+    [id, JSON.stringify(data)],
+  );
+}
+
+export async function monitorTick(db: PoolClient, now = new Date()) {
+  await db.query(`SET TIME ZONE 'UTC'`);
+  const locked = (await db.query(`SELECT pg_try_advisory_lock(20260906,710) locked`)).rows[0].locked;
+  if (!locked) return;
+  try {
+    await db.query('BEGIN');
+    await db.query(`SET LOCAL statement_timeout='20s'`);
+    const dirtyMinutes = new Set<string>();
+    const cursors: Cursor[] = [];
+    for (const source of ['submit', 'receipt'] as const) {
+      const stored = (await db.query(`SELECT data FROM "SendingMonitorCheckpoint" WHERE id=$1`, [source])).rows[0]
+        ?.data as Cursor | undefined;
+      const from = stored?.completeAt
+        ? new Date(new Date(stored.completeAt).getTime() - 120000).toISOString()
+        : new Date(now.getTime() - 7200000).toISOString();
+      const cursor: Cursor =
+        stored && stored.id
+          ? stored
+          : {
+              from,
+              to: new Date(now.getTime() - 2000).toISOString(),
+              at: from,
+              id: '',
+              completeAt: stored?.completeAt,
+            };
+      const table = source === 'submit' ? 'SmsSubmitRecord' : 'UpstreamReceiptInbox';
+      const field = source === 'submit' ? 'messageRecordId' : 'matchedMessageRecordId';
+      const batch = (
+        await db.query(
+          `SELECT id,"updatedAt","${field}" "messageId" FROM "${table}" WHERE ("updatedAt",id)>($1::timestamp,$2) AND "updatedAt"<$3::timestamp ORDER BY "updatedAt",id LIMIT 1000`,
+          [cursor.at, cursor.id, cursor.to],
+        )
+      ).rows;
+      for (const minute of (await projectMessages(db, [
+        ...new Set(batch.map((b) => b.messageId).filter(Boolean)),
+      ] as string[])) ?? [])
+        dirtyMinutes.add(minute);
+      if (batch.length === 1000) {
+        const last = batch[batch.length - 1];
+        cursor.at = last.updatedAt.toISOString();
+        cursor.id = last.id;
+      } else {
+        cursor.completeAt = cursor.to;
+        cursor.id = '';
+      }
+      await checkpoint(db, source, cursor);
+      cursors.push(cursor);
+    }
+    // A bounded, independent 72-hour reconciliation cursor recovers transactions older than the overlap.
+    const recon = (await db.query(`SELECT data FROM "SendingMonitorCheckpoint" WHERE id='reconcile'`)).rows[0]?.data;
+    if (!recon?.nextAt || new Date(recon.nextAt) <= now) {
+      const start = recon?.id ? recon.at : new Date(now.getTime() - 72 * 3600000).toISOString();
+      const batch = (
+        await db.query(
+          `SELECT id,"createdAt","messageRecordId" FROM "SmsSubmitRecord" WHERE ("createdAt",id)>($1::timestamp,$2) AND "createdAt"<$3 ORDER BY "createdAt",id LIMIT 500`,
+          [start, recon?.id ?? '', new Date(now.getTime() - 120000)],
+        )
+      ).rows;
+      for (const minute of (await projectMessages(db, [...new Set(batch.map((b) => b.messageRecordId))] as string[])) ??
+        [])
+        dirtyMinutes.add(minute);
+      const last = batch[batch.length - 1];
+      await checkpoint(
+        db,
+        'reconcile',
+        batch.length === 500
+          ? { at: last.createdAt.toISOString(), id: last.id }
+          : { id: '', nextAt: new Date(now.getTime() + 1800000).toISOString() },
+      );
+    }
+    const pending = (
+      await db.query(
+        `SELECT count(*)::int count,min("receivedAt") oldest FROM "UpstreamReceiptInbox" WHERE status IN ('pending','processing','retrying') AND "receivedAt">$1`,
+        [new Date(now.getTime() - 72 * 3600000)],
+      )
+    ).rows[0];
+    const complete =
+      cursors.every((c) => c.completeAt && now.getTime() - new Date(c.completeAt).getTime() < 30000) &&
+      pending.count === 0;
+    await checkpoint(db, 'health', {
+      complete,
+      pendingReceipts: pending.count,
+      oldestPending: pending.oldest,
+      submitCompleteAt: cursors[0].completeAt,
+      receiptCompleteAt: cursors[1].completeAt,
+      checkedAt: now.toISOString(),
+      poolMax: 2,
+    });
+    // Persist dirty windows with the projection transaction. Reconciliation and late matching revise only affected windows.
+    const pendingWindows = new Map<string, { type: MonitorType; at: string }>(
+      (await db.query(`SELECT data FROM "SendingMonitorCheckpoint" WHERE id='dirty-windows'`)).rows[0]?.data?.windows ??
+        [],
+    );
+    const schedule =
+      (await db.query(`SELECT data FROM "SendingMonitorCheckpoint" WHERE id='schedule'`)).rows[0]?.data ?? {};
+    for (const type of ['industry', 'verification', 'overall'] as MonitorType[]) {
+      const period = type === 'overall' ? 600000 : 300000;
+      const t = new Date(Math.floor(now.getTime() / period) * period);
+      for (let back = 1; back >= 0; back--) {
+        const boundary = new Date(t.getTime() - back * period);
+        const final = type === 'overall' || now.getTime() >= boundary.getTime() + 90000;
+        const key = `${type}:${boundary.toISOString()}`;
+        const phase = `${final}:${complete}`;
+        if (schedule[key] !== phase) pendingWindows.set(key, { type, at: boundary.toISOString() });
+      }
+      for (const minute of dirtyMinutes) {
+        const first = (Math.floor(new Date(minute).getTime() / period) + 1) * period;
+        for (let n = 0; n < (type === 'overall' ? 3 : 1); n++) {
+          const at = new Date(first + n * period).toISOString();
+          if (new Date(at) <= now) pendingWindows.set(`${type}:${at}`, { type, at });
+        }
+      }
+    }
+    // Newer windows first; older revisions must not replace current alert state. Work stays bounded per tick.
+    for (const [key, window] of [...pendingWindows.entries()]
+      .sort((a, b) => b[1].at.localeCompare(a[1].at))
+      .slice(0, 12)
+      .sort((a, b) => a[1].at.localeCompare(b[1].at))) {
+      const final = window.type === 'overall' || now.getTime() >= new Date(window.at).getTime() + 90000;
+      await evaluateWindow(db, window.type, new Date(window.at), final, complete);
+      schedule[key] = `${final}:${complete}`;
+      pendingWindows.delete(key);
+    }
+    for (const key of Object.keys(schedule)) {
+      if (new Date(key.slice(key.indexOf(':') + 1)).getTime() < now.getTime() - 1200000) delete schedule[key];
+    }
+    await checkpoint(db, 'schedule', schedule);
+    await checkpoint(db, 'dirty-windows', { windows: [...pendingWindows.entries()] });
+    await db.query('COMMIT');
+    // Retention only touches monitor-owned tables. Never delete SMS, receipt, queue or audit business data.
+    await db.query(
+      `DELETE FROM "SendingMonitorFact" WHERE id IN (SELECT id FROM "SendingMonitorFact" WHERE "submittedAt"<(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')-interval '72 hours' LIMIT 1000)`,
+    );
+    await db.query(
+      `DELETE FROM "SendingMonitorMinute" WHERE (kind,"dimensionKey",minute,verification) IN (SELECT kind,"dimensionKey",minute,verification FROM "SendingMonitorMinute" WHERE minute<(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')-interval '7 days' LIMIT 1000)`,
+    );
+    await db.query(
+      `DELETE FROM "SendingMonitorSnapshot" WHERE id IN (SELECT id FROM "SendingMonitorSnapshot" WHERE "evaluationAt"<(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')-interval '30 days' LIMIT 1000)`,
+    );
+    await db.query(
+      `DELETE FROM "SendingMonitorAlertRead" WHERE "alertId" IN (SELECT id FROM "SendingMonitorAlert" WHERE state<>'active' AND "closedAt"<(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')-interval '90 days' LIMIT 1000)`,
+    );
+    await db.query(
+      `DELETE FROM "SendingMonitorAlert" WHERE id IN (SELECT id FROM "SendingMonitorAlert" WHERE state<>'active' AND "closedAt"<(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')-interval '90 days' LIMIT 1000)`,
+    );
+  } catch (error) {
+    await db.query('ROLLBACK');
+    throw error;
+  } finally {
+    await db.query('SELECT pg_advisory_unlock(20260906,710)');
+  }
+}
+
+async function main() {
+  const connectionString = process.env.SENDING_MONITOR_DATABASE_URL || process.env.DATABASE_URL;
+  if (!connectionString) throw new Error('SENDING_MONITOR_DATABASE_URL or DATABASE_URL is required');
+  const pool = new Pool({
+    connectionString,
+    max: 2,
+    application_name: 'cmpp-sending-monitor',
+    statement_timeout: 25000,
+  });
+  let stopping = false;
+  process.on('SIGTERM', () => {
+    stopping = true;
+  });
+  process.on('SIGINT', () => {
+    stopping = true;
+  });
+  while (!stopping) {
+    const db = await pool.connect();
+    try {
+      await monitorTick(db);
+    } catch (error) {
+      console.error('sending_monitor_tick_failed', error instanceof Error ? error.message : 'unknown');
+    } finally {
+      db.release();
+    }
+    if (!stopping) await new Promise((resolve) => setTimeout(resolve, 5000));
+  }
+  await pool.end();
+}
+if (require.main === module)
+  void main().catch((error) => {
+    console.error('sending_monitor_start_failed', error instanceof Error ? error.message : 'unknown');
+    process.exitCode = 1;
+  });

api/src/sending-monitor/monitor-evaluation.ts

@@ -0,0 +1,333 @@
+import { randomUUID } from 'node:crypto';
+import { PoolClient } from 'pg';
+import {
+  horizons,
+  matchRules,
+  metric,
+  type Dimensions,
+  type Metric,
+  type MonitorType,
+  type Rule,
+} from './monitor-metrics';
+import { keyOf } from './monitor-projection';
+
+type Aggregate = {
+  dimensionKey: string;
+  dimensions: Dimensions;
+  total: number;
+  eligible: number;
+  unassessable: number;
+  seconds: number;
+  mature: number;
+  success: number;
+};
+type Measurements = { total: number; unassessable: number; metrics: Metric[] };
+type Evaluation = {
+  id: string;
+  type: MonitorType;
+  dimensionKey: string;
+  dimensions: Dimensions;
+  evaluationAt: string;
+  windowFrom: string;
+  observedUntil: string;
+  stage: string;
+  metrics: Measurements;
+  rule: Rule | null;
+  status: string;
+  completeness: { complete: boolean; reason?: string };
+  computedAt: string;
+};
+
+type CounterState = { bad: number; good: number; t: string; ruleKey: string };
+type StoredAlert = {
+  id: string;
+  dimensionKey: string;
+  ruleKey: string;
+  openedAt: Date;
+  worst: Evaluation & { revision: number };
+};
+export async function evaluateWindow(db: PoolClient, type: MonitorType, t: Date, final: boolean, complete: boolean) {
+  const observed = new Date(t.getTime() + (final && type !== 'overall' ? 60000 : 0));
+  const from = new Date(t.getTime() - (type === 'overall' ? 1800000 : 300000));
+  const kind = type === 'industry' ? 'attempt' : 'business';
+  // Whole mature minutes come from sparse buckets. Only the maturity minute uses exact millisecond facts.
+  const data: Aggregate[] = (
+    await db.query(
+      `WITH buckets AS (
+      SELECT b.* FROM "SendingMonitorMinute" b LEFT JOIN LATERAL (SELECT * FROM "SendingMonitorTargetVersion" v WHERE v."channelId"=b.dimensions->>'channelId' AND v."effectiveFrom"<=$3::timestamp ORDER BY version DESC LIMIT 1) target ON true
+      WHERE b.kind=$1 AND b.minute >= $2::timestamp AND b.minute < $3::timestamp AND ($4<>'verification' OR b.verification)
+      AND ($4<>'industry' OR (target.enabled AND b.minute>=target."effectiveFrom"))
+    ), totals AS (SELECT "dimensionKey",(jsonb_agg(dimensions ORDER BY minute DESC))->0 dimensions,
+      sum((metrics->>'total')::int)::int total,sum((metrics->>'eligible')::int)::int eligible,
+      sum((metrics->>'unassessable')::int)::int unassessable FROM buckets GROUP BY "dimensionKey"),
+    hs AS (SELECT unnest($6::int[]) seconds), full_minutes AS (
+      SELECT b."dimensionKey",h.seconds,sum((b.metrics->>'eligible')::int)::int mature,
+      sum((b.metrics->>('s'||h.seconds))::int)::int success FROM buckets b CROSS JOIN hs h
+      WHERE b.minute<date_trunc('minute',$5::timestamp-h.seconds*interval '1 second') GROUP BY b."dimensionKey",h.seconds
+    ), boundary AS (
+      SELECT f."dimensionKey",h.seconds,count(*)::int mature,
+      count(*) FILTER(WHERE f."successAt"-f."submittedAt" BETWEEN interval '0 seconds' AND h.seconds*interval '1 second')::int success
+      FROM "SendingMonitorFact" f CROSS JOIN hs h LEFT JOIN LATERAL (SELECT * FROM "SendingMonitorTargetVersion" v WHERE v."channelId"=f.dimensions->>'channelId' AND v."effectiveFrom"<=$3::timestamp ORDER BY version DESC LIMIT 1) target ON true
+      WHERE f.kind=$1 AND f."submittedAt">=$2::timestamp AND f."submittedAt"<$3::timestamp AND f.reason IS NULL
+      AND ($4<>'verification' OR f.verification) AND ($4<>'industry' OR (target.enabled AND f."submittedAt">=target."effectiveFrom"))
+      AND f."submittedAt">=date_trunc('minute',$5::timestamp-h.seconds*interval '1 second') AND f."submittedAt"<=$5::timestamp-h.seconds*interval '1 second'
+      GROUP BY f."dimensionKey",h.seconds)
+    SELECT totals.*,hs.seconds,COALESCE(m.mature,0)+COALESCE(b.mature,0) mature,COALESCE(m.success,0)+COALESCE(b.success,0) success
+    FROM totals CROSS JOIN hs LEFT JOIN full_minutes m USING("dimensionKey",seconds) LEFT JOIN boundary b USING("dimensionKey",seconds)`,
+      [kind, from, t, type, observed, [...horizons[type]]],
+    )
+  ).rows;
+  const rules: Rule[] = (
+    await db.query(
+      `SELECT DISTINCT ON ("ruleId") * FROM "SendingMonitorRuleVersion" WHERE type=$1 AND "effectiveAt"<=$2 ORDER BY "ruleId",version DESC`,
+      [type, t],
+    )
+  ).rows;
+  const groups = new Map<string, Aggregate[]>();
+  for (const row of data) {
+    const list = groups.get(row.dimensionKey) ?? [];
+    list.push(row);
+    groups.set(row.dimensionKey, list);
+  }
+  if (type === 'industry') {
+    const targets = (
+      await db.query(
+        `SELECT c.id,c.name,CASE WHEN cardinality(c.carriers)>0 THEN c.carriers WHEN c.carrier='all' THEN ARRAY['mobile','unicom','telecom'] ELSE ARRAY[COALESCE(c.carrier,'unknown')] END carriers
+      FROM (SELECT DISTINCT ON ("channelId") * FROM "SendingMonitorTargetVersion" WHERE "effectiveFrom"<=$1 ORDER BY "channelId",version DESC) t JOIN "SmsChannel" c ON c.id=t."channelId" WHERE t.enabled AND t."effectiveFrom"<=$1 AND c.status<>'deleted'`,
+        [t],
+      )
+    ).rows;
+    for (const c of targets)
+      for (const carrier of c.carriers) {
+        const key = keyOf([c.id, carrier]);
+        if (!groups.has(key))
+          groups.set(
+            key,
+            horizons[type].map((seconds) => ({
+              dimensionKey: key,
+              dimensions: { channelId: c.id, channelName: c.name, carrier },
+              total: 0,
+              eligible: 0,
+              unassessable: 0,
+              seconds,
+              mature: 0,
+              success: 0,
+            })),
+          );
+      }
+  }
+  if (type !== 'industry') {
+    const previousDimensions = (
+      await db.query(
+        `SELECT "dimensionKey",dimensions FROM "SendingMonitorSnapshot" WHERE type=$1 AND "evaluationAt"=(SELECT max("evaluationAt") FROM "SendingMonitorSnapshot" WHERE type=$1)`,
+        [type],
+      )
+    ).rows;
+    for (const item of previousDimensions)
+      if (!groups.has(item.dimensionKey))
+        groups.set(
+          item.dimensionKey,
+          horizons[type].map((seconds) => ({
+            dimensionKey: item.dimensionKey,
+            dimensions: item.dimensions,
+            total: 0,
+            eligible: 0,
+            unassessable: 0,
+            seconds,
+            mature: 0,
+            success: 0,
+          })),
+        );
+  }
+  const activeDimensions = (
+    await db.query(`SELECT "dimensionKey",dimensions FROM "SendingMonitorAlert" WHERE type=$1 AND state='active'`, [
+      type,
+    ])
+  ).rows;
+  for (const item of activeDimensions) {
+    if (!groups.has(item.dimensionKey))
+      groups.set(
+        item.dimensionKey,
+        horizons[type].map((seconds) => ({
+          dimensionKey: item.dimensionKey,
+          dimensions: item.dimensions,
+          total: 0,
+          eligible: 0,
+          unassessable: 0,
+          seconds,
+          mature: 0,
+          success: 0,
+        })),
+      );
+  }
+  const activeAlerts = new Map(
+    (
+      await db.query<StoredAlert>(`SELECT * FROM "SendingMonitorAlert" WHERE type=$1 AND state='active'`, [type])
+    ).rows.map((a) => [a.dimensionKey, a]),
+  );
+  const counters = new Map(
+    (
+      await db.query<{ id: string; data: CounterState }>(
+        `SELECT id,data FROM "SendingMonitorCheckpoint" WHERE id=ANY($1::text[])`,
+        [[...groups.keys()].map((key) => `alert-state:${type}:${key}`)],
+      )
+    ).rows.map((c) => [c.id, c.data]),
+  );
+  for (const [key, rows] of groups) {
+    const first = rows[0],
+      rule = matchRules(rules, type, first.dimensions)[0] ?? null;
+    const values = horizons[type].map((seconds, i) => {
+      const row = rows.find((r) => r.seconds === seconds)!;
+      return metric(
+        seconds,
+        row.success,
+        row.mature,
+        Math.max(0, row.eligible - row.mature),
+        rule?.config.thresholds[i] ?? null,
+        rule?.config.minSamples ?? 1,
+      );
+    });
+    const metrics = { total: first.total, unassessable: first.unassessable, metrics: values };
+    const status = !complete
+      ? 'stale'
+      : first.unassessable
+        ? 'unassessable'
+        : !rule
+          ? 'unconfigured'
+          : !rule.config.enabled
+            ? 'disabled'
+            : !first.total
+              ? 'no_data'
+              : values.some((m) => m.bad)
+                ? 'abnormal'
+                : values.some((m) => m.threshold !== null && m.insufficient)
+                  ? 'sample_insufficient'
+                  : 'normal';
+    const snapshot: Evaluation = {
+      id: keyOf([type, key, t.toISOString()]),
+      type,
+      dimensionKey: key,
+      dimensions: first.dimensions,
+      evaluationAt: t.toISOString(),
+      windowFrom: from.toISOString(),
+      observedUntil: observed.toISOString(),
+      stage: final ? 'final' : 'initial',
+      metrics,
+      rule,
+      status,
+      completeness: { complete, reason: complete ? undefined : '采集未完成或回执待匹配，暂停告警判断' },
+      computedAt: new Date().toISOString(),
+    };
+    const saved = await db.query(
+      `INSERT INTO "SendingMonitorSnapshot" (id,type,"dimensionKey",dimensions,"evaluationAt","windowFrom","observedUntil",stage,metrics,rule,status,completeness)
+      VALUES($1,$2,$3,$4::jsonb,$5,$6,$7,$8,$9::jsonb,$10::jsonb,$11,$12::jsonb)
+      ON CONFLICT(id) DO UPDATE SET "observedUntil"=EXCLUDED."observedUntil",stage=EXCLUDED.stage,metrics=EXCLUDED.metrics,rule=EXCLUDED.rule,status=EXCLUDED.status,completeness=EXCLUDED.completeness,
+        revision="SendingMonitorSnapshot".revision+1,"computedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')
+      WHERE ("SendingMonitorSnapshot".metrics,"SendingMonitorSnapshot".stage,"SendingMonitorSnapshot".status,"SendingMonitorSnapshot".rule,"SendingMonitorSnapshot".completeness)
+        IS DISTINCT FROM (EXCLUDED.metrics,EXCLUDED.stage,EXCLUDED.status,EXCLUDED.rule,EXCLUDED.completeness) RETURNING revision`,
+      [
+        snapshot.id,
+        type,
+        key,
+        JSON.stringify(snapshot.dimensions),
+        t,
+        from,
+        observed,
+        snapshot.stage,
+        JSON.stringify(metrics),
+        JSON.stringify(rule),
+        status,
+        JSON.stringify(snapshot.completeness),
+      ],
+    );
+    if (saved.rows.length)
+      await updateAlert(
+        db,
+        snapshot,
+        Number(saved.rows[0].revision),
+        counters.get(`alert-state:${type}:${key}`),
+        activeAlerts.get(key),
+      );
+  }
+}
+
+async function updateAlert(
+  db: PoolClient,
+  s: Evaluation,
+  revision: number,
+  stored?: CounterState,
+  active?: StoredAlert,
+) {
+  const id = `alert-state:${s.type}:${s.dimensionKey}`;
+  const previous = stored ?? { bad: 0, good: 0, t: '', ruleKey: '' };
+  if (previous.t > s.evaluationAt) return;
+  const ruleKey = s.rule ? `${s.rule.ruleId}:${s.rule.version}` : 'none';
+  if (active && active.ruleKey !== ruleKey) {
+    await db.query(
+      `UPDATE "SendingMonitorAlert" SET state='closed',"closedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),"closeReason"=$2 WHERE id=$1`,
+      [active.id, s.rule?.config.enabled ? 'rule_changed' : 'disabled'],
+    );
+    active = undefined;
+  }
+  if (previous.ruleKey !== ruleKey) {
+    previous.bad = 0;
+    previous.good = 0;
+    previous.ruleKey = ruleKey;
+    previous.t = '';
+  }
+  const config = s.rule?.config;
+  const usable = s.completeness.complete && s.metrics.unassessable === 0 && config?.enabled && s.metrics.total > 0;
+  const bad = usable && s.metrics.metrics.some((m) => m.bad);
+  const good = usable && s.metrics.metrics.filter((m) => m.threshold !== null).every((m) => !m.bad && !m.insufficient);
+  const final = s.stage === 'final';
+  if (final && previous.t !== s.evaluationAt && usable) {
+    if (bad) {
+      previous.bad++;
+      previous.good = 0;
+    } else if (good) {
+      previous.good++;
+      previous.bad = 0;
+    }
+  }
+  if (bad && !active && ((final && previous.bad >= config!.consecutiveBad) || config!.consecutiveBad === 1)) {
+    const alertId = randomUUID();
+    await db.query(
+      `INSERT INTO "SendingMonitorAlert" (id,type,"dimensionKey",dimensions,state,"openedAt","lastEvaluatedAt","ruleKey",latest,worst)
+      VALUES($1,$2,$3,$4::jsonb,'active',$5,$5,$6,$7::jsonb,$7::jsonb)`,
+      [
+        alertId,
+        s.type,
+        s.dimensionKey,
+        JSON.stringify(s.dimensions),
+        s.evaluationAt,
+        ruleKey,
+        JSON.stringify({ ...s, revision }),
+      ],
+    );
+  } else if (active) {
+    const corrected = revision > 1 && good && active.openedAt.toISOString() === s.evaluationAt;
+    const recovered = final && good && previous.good >= config!.consecutiveGood;
+    const worst =
+      bad &&
+      s.metrics.metrics.some((m, i) => m.rate !== null && (active.worst.metrics.metrics[i]?.rate ?? 101) > m.rate)
+        ? { ...s, revision }
+        : active.worst;
+    await db.query(
+      `UPDATE "SendingMonitorAlert" SET latest=$2::jsonb,worst=$3::jsonb,"lastEvaluatedAt"=$4,"updatedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),
+      state=$5,"closedAt"=CASE WHEN $5<>'active' THEN (CURRENT_TIMESTAMP AT TIME ZONE 'UTC') ELSE NULL END,"closeReason"=$6 WHERE id=$1`,
+      [
+        active.id,
+        JSON.stringify({ ...s, revision }),
+        JSON.stringify(worst),
+        s.evaluationAt,
+        corrected ? 'closed' : recovered ? 'recovered' : 'active',
+        corrected ? 'data_corrected' : recovered ? 'recovered' : null,
+      ],
+    );
+  }
+  if (final && usable) previous.t = s.evaluationAt;
+  await db.query(
+    `INSERT INTO "SendingMonitorCheckpoint" (id,data) VALUES($1,$2::jsonb) ON CONFLICT(id) DO UPDATE SET data=EXCLUDED.data,"updatedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')`,
+    [id, JSON.stringify(previous)],
+  );
+}

api/src/sending-monitor/monitor-metrics.spec.ts

@@ -0,0 +1,88 @@
+import { completeAttempt, evaluateSamples, matchRules, validateRule, type Rule } from './monitor-metrics';
+const base = Date.parse('2026-09-06T04:30:00Z');
+const rule: Rule = {
+  ruleId: 'r',
+  version: 1,
+  type: 'industry',
+  scope: {},
+  effectiveAt: new Date(base).toISOString(),
+  config: { enabled: true, minSamples: 1, thresholds: [90, 95, 98], consecutiveBad: 1, consecutiveGood: 2 },
+};
+describe('sending monitor business metrics', () => {
+  it('uses exact milliseconds and does not count early successes before maturity', () => {
+    const samples = [4999, 5000, 5001].map((ms) => ({
+      submittedAt: new Date(base - 60000),
+      successAt: new Date(base - 60000 + ms),
+    }));
+    const result = evaluateSamples(
+      [...samples, { submittedAt: new Date(base - 1000), successAt: new Date(base - 500) }],
+      'industry',
+      new Date(base),
+      rule,
+    );
+    expect(result.metrics[0]).toMatchObject({ success: 2, mature: 3, observing: 1, bad: true });
+    expect(result.metrics[2]).toMatchObject({ success: 3, mature: 3, observing: 1, bad: false });
+  });
+  it('finalizes the last second of a fixed five-minute window once mature', () => {
+    const samples = [{ submittedAt: new Date(base - 1000), successAt: new Date(base + 49000) }];
+    expect(evaluateSamples(samples, 'industry', new Date(base), rule).metrics[2].mature).toBe(0);
+    expect(evaluateSamples(samples, 'industry', new Date(base + 60000), rule).metrics[2]).toMatchObject({
+      mature: 1,
+      success: 1,
+    });
+  });
+  it('keeps unknown times out of denominator and null for no mature samples', () => {
+    const result = evaluateSamples(
+      [{ submittedAt: new Date(base - 60000), successAt: null, reason: 'missing_wire_time' }],
+      'industry',
+      new Date(base),
+      rule,
+    );
+    expect(result.unassessable).toBe(1);
+    expect(result.metrics[0].rate).toBeNull();
+    expect(result.metrics[0].bad).toBe(false);
+  });
+  it('only matures the first ten minutes for the 20-minute overall metric', () => {
+    const samples = Array.from({ length: 30 }, (_, i) => ({
+      submittedAt: new Date(base - 1800000 + i * 60000 + 1),
+      successAt: null,
+    }));
+    expect(evaluateSamples(samples, 'overall', new Date(base)).metrics[2]).toMatchObject({ mature: 10, observing: 20 });
+  });
+  it('requires every unique fragment of the same attempt', () => {
+    expect(completeAttempt([{ index: 1, total: 2, firstSuccess: new Date(base) }])).toBeNull();
+    expect(
+      completeAttempt([
+        { index: 1, total: 2, firstSuccess: new Date(base) },
+        { index: 2, total: 2, firstSuccess: new Date(base + 5000) },
+      ])?.getTime(),
+    ).toBe(base + 5000);
+    expect(
+      completeAttempt([
+        { index: 1, total: 2, firstSuccess: new Date(base) },
+        { index: 1, total: 2, firstSuccess: new Date(base) },
+      ]),
+    ).toBeNull();
+  });
+  it('matches full rule precedence without crossing tenants', () => {
+    const scopes = [
+      {},
+      { tenantId: 't', applicationId: 'a' },
+      { tenantId: 't', signatureId: 's' },
+      { tenantId: 't', applicationId: 'a', signatureId: 's' },
+      { tenantId: 'other', signatureId: 's' },
+    ];
+    const rules = scopes.map((scope, i) => ({ ...rule, ruleId: String(i), type: 'overall' as const, scope }));
+    expect(
+      matchRules(rules, 'overall', { tenantId: 't', applicationId: 'a', signatureId: 's' }).map((r) => r.ruleId),
+    ).toEqual(['3', '2', '1', '0']);
+  });
+  it('validates sample count, threshold order, scope and consecutive counts', () => {
+    expect(validateRule('industry', {}, rule.config)).toBeNull();
+    expect(validateRule('industry', {}, { ...rule.config, minSamples: 0 })).toBeTruthy();
+    expect(validateRule('industry', {}, { ...rule.config, thresholds: [99, 95, null] })).toBeTruthy();
+    expect(validateRule('industry', {}, { ...rule.config, thresholds: [null, null, null] })).toBeTruthy();
+    expect(validateRule('overall', { applicationId: 'a' }, rule.config)).toBeTruthy();
+    expect(validateRule('overall', {}, { ...rule.config, consecutiveGood: 6 })).toBeTruthy();
+  });
+});

api/src/sending-monitor/monitor-metrics.ts

@@ -0,0 +1,147 @@
+export const horizons = { industry: [5, 20, 60], verification: [5, 20, 60], overall: [60, 300, 1200] } as const;
+export type MonitorType = keyof typeof horizons;
+export type Dimensions = {
+  tenantId?: string;
+  tenantName?: string;
+  applicationId?: string;
+  applicationName?: string;
+  signatureId?: string;
+  signatureName?: string;
+  channelId?: string;
+  channelName?: string;
+  carrier?: string;
+};
+export type MonitorScope = { tenantId?: string; applicationId?: string; signatureId?: string };
+export type RuleConfig = {
+  enabled: boolean;
+  minSamples: number;
+  thresholds: (number | null)[];
+  consecutiveBad: number;
+  consecutiveGood: number;
+  deleted?: boolean;
+};
+export type Rule = {
+  ruleId: string;
+  version: number;
+  type: MonitorType;
+  scope: MonitorScope;
+  config: RuleConfig;
+  effectiveAt: string;
+};
+export type Metric = {
+  seconds: number;
+  success: number;
+  mature: number;
+  observing: number;
+  rate: number | null;
+  threshold: number | null;
+  insufficient: boolean;
+  bad: boolean;
+};
+export type Sample = { submittedAt: Date; successAt: Date | null; reason?: string | null };
+
+export function evaluateSamples(samples: Sample[], type: MonitorType, observedUntil: Date, rule?: Rule | null) {
+  const eligible = samples.filter((s) => !s.reason);
+  const metrics: Metric[] = horizons[type].map((seconds, i) => {
+    const mature = eligible.filter((s) => s.submittedAt.getTime() + seconds * 1000 <= observedUntil.getTime());
+    const success = mature.filter(
+      (s) =>
+        s.successAt &&
+        s.successAt >= s.submittedAt &&
+        s.successAt.getTime() - s.submittedAt.getTime() <= seconds * 1000,
+    ).length;
+    return metric(
+      seconds,
+      success,
+      mature.length,
+      eligible.length - mature.length,
+      rule?.config.thresholds[i] ?? null,
+      rule?.config.minSamples ?? 1,
+    );
+  });
+  return { total: samples.length, unassessable: samples.length - eligible.length, metrics };
+}
+
+export function metric(
+  seconds: number,
+  success: number,
+  mature: number,
+  observing: number,
+  threshold: number | null,
+  min: number,
+): Metric {
+  return {
+    seconds,
+    success,
+    mature,
+    observing,
+    threshold,
+    rate: mature ? (100 * success) / mature : null,
+    insufficient: mature < min,
+    bad: threshold !== null && mature >= min && success * 10000 < Math.round(threshold * 100) * mature,
+  };
+}
+
+export function matchRules(rules: Rule[], type: MonitorType, dimension: Dimensions) {
+  return rules
+    .filter(
+      (r) =>
+        r.type === type &&
+        !r.config.deleted &&
+        Object.entries(r.scope).every(([key, value]) => !value || dimension[key as keyof Dimensions] === value),
+    )
+    .sort((a, b) => score(b.scope) - score(a.scope));
+}
+const score = (scope: MonitorScope) => (scope.signatureId ? 2 : 0) + (scope.applicationId ? 1 : 0);
+
+export function validateRule(type: string, scope: MonitorScope, input: RuleConfig): string | null {
+  if (!Object.hasOwn(horizons, type)) return '监控类型无效';
+  if (!scope || Object.keys(scope).some((k) => !['tenantId', 'applicationId', 'signatureId'].includes(k)))
+    return '规则范围无效';
+  if (Object.values(scope).some((v) => typeof v !== 'string' || !v.trim() || v.length > 200)) return '规则范围ID无效';
+  if (type !== 'overall' && Object.keys(scope).length) return '仅整体兜底支持个性规则';
+  if (Object.keys(scope).length && (!scope.tenantId || (!scope.applicationId && !scope.signatureId)))
+    return '个性规则须指定企业和应用或签名';
+  if (
+    !input ||
+    typeof input.enabled !== 'boolean' ||
+    !Number.isSafeInteger(input.minSamples) ||
+    input.minSamples < 1 ||
+    input.minSamples > 100000000
+  )
+    return '最低样本量须为正整数';
+  if (![input.consecutiveBad, input.consecutiveGood].every((n) => Number.isInteger(n) && n >= 1 && n <= 5))
+    return '连续次数须为1至5';
+  if (!Array.isArray(input.thresholds) || input.thresholds.length !== 3 || input.thresholds.every((n) => n === null))
+    return '至少设置一项时效指标';
+  if (input.deleted !== undefined && typeof input.deleted !== 'boolean') return '删除标记无效';
+  let previous = -1;
+  for (const n of input.thresholds) {
+    if (n === null) continue;
+    if (
+      typeof n !== 'number' ||
+      !Number.isFinite(n) ||
+      n < 0 ||
+      n > 100 ||
+      Math.abs(n * 100 - Math.round(n * 100)) > 0.000001 ||
+      n < previous
+    )
+      return '下限须为0至100，最多两位小数，按时限递增';
+    previous = n;
+  }
+  return null;
+}
+
+// Success fragments from different attempts must never be combined.
+export function completeAttempt(segments: { index: number; total: number; firstSuccess: Date | null }[]): Date | null {
+  const total = Math.max(1, ...segments.map((s) => s.total));
+  const unique = new Map(segments.map((s) => [s.index, s]));
+  if (unique.size !== total) return null;
+  let last = 0;
+  for (let i = 1; i <= total; i++) {
+    const at = unique.get(i)?.firstSuccess;
+    if (!at) return null;
+    last = Math.max(last, at.getTime());
+  }
+  return new Date(last);
+}

api/src/sending-monitor/monitor-projection.ts

@@ -0,0 +1,198 @@
+import { createHash } from 'node:crypto';
+import { PoolClient } from 'pg';
+import { completeAttempt, type Dimensions } from './monitor-metrics';
+
+type Attempt = {
+  id: string;
+  messageRecordId: string;
+  channelId: string;
+  createdAt: Date;
+  firstWireSubmitAt: Date | null;
+  wireTimeSource: string | null;
+  submitStatus: string;
+  tenantId: string | null;
+  applicationId: string | null;
+  signatureId: string | null;
+  tenantName: string | null;
+  applicationName: string | null;
+  signatureName: string | null;
+  channelName: string;
+  carrier: string | null;
+  content: string;
+  receiptRequested: boolean | null;
+};
+type Segment = {
+  submitRecordId: string;
+  segmentIndex: number;
+  segmentTotal: number;
+  firstWireSubmitAt: Date | null;
+  wireTimeSource: string | null;
+  firstSuccess: Date | null;
+  missingReceiptTime: boolean;
+  ambiguousReceipt: boolean;
+  receiptRequested: boolean | null;
+};
+type Fact = {
+  id: string;
+  kind: string;
+  sourceId: string;
+  messageRecordId: string;
+  dimensionKey: string;
+  dimensions: Dimensions;
+  submittedAt: string;
+  successAt: string | null;
+  verification: boolean;
+  reason: string | null;
+};
+export const keyOf = (parts: unknown[]) => createHash('sha256').update(JSON.stringify(parts)).digest('hex');
+
+export async function projectMessages(db: PoolClient, ids: string[]) {
+  if (!ids.length) return;
+  const attempts: Attempt[] = (
+    await db.query(
+      `SELECT s.id,s."messageRecordId",s."channelId",s."createdAt",s."firstWireSubmitAt",s."wireTimeSource",s."submitStatus",s."receiptRequested",
+    m."tenantId",m."applicationId",m."signatureId",m.carrier,m.content,m."cmppRegisteredDelivery",t.name "tenantName",a.name "applicationName",g.name "signatureName",c.name "channelName"
+    FROM "SmsSubmitRecord" s JOIN "SmsMessageRecord" m ON m.id=s."messageRecordId" LEFT JOIN "Tenant" t ON t.id=m."tenantId"
+    LEFT JOIN "SmsApplication" a ON a.id=m."applicationId" LEFT JOIN "SmsSignature" g ON g.id=m."signatureId" JOIN "SmsChannel" c ON c.id=s."channelId"
+    WHERE s."messageRecordId"=ANY($1::text[])`,
+      [ids],
+    )
+  ).rows;
+  const segments: Segment[] = (
+    await db.query(
+      `SELECT a."submitRecordId",a."segmentIndex",a."segmentTotal",a."firstWireSubmitAt",a."wireTimeSource",a."receiptRequested",
+      EXISTS (SELECT 1 FROM "SmsMessageSegmentAudit" other WHERE other."messageRecordId"=a."messageRecordId" AND other."gatewayMessageId"=a."gatewayMessageId" AND other."channelId"=a."channelId" AND other."submitId"<>a."submitId") "ambiguousReceipt",
+      min(i."gatewayReceivedAt") FILTER(WHERE i."receiptStatus"='delivered' AND i.status='matched') "firstSuccess",
+      bool_or(i."receiptStatus"='delivered' AND i."gatewayReceivedAt" IS NULL) "missingReceiptTime"
+    FROM "SmsMessageSegmentAudit" a LEFT JOIN "UpstreamReceiptInbox" i ON i."matchedMessageRecordId"=a."messageRecordId" AND i."gatewayMessageId"=a."gatewayMessageId"
+      AND (i."incomingChannelId"=a."channelId" OR i."matchedChannelId"=a."channelId")
+      AND NOT EXISTS (SELECT 1 FROM "SmsMessageSegmentAudit" other WHERE other."messageRecordId"=a."messageRecordId" AND other."gatewayMessageId"=a."gatewayMessageId" AND other."channelId"=a."channelId" AND other."submitId"<>a."submitId")
+    WHERE a."messageRecordId"=ANY($1::text[]) GROUP BY a.id`,
+      [ids],
+    )
+  ).rows;
+  const previous: Fact[] = (
+    await db.query(`SELECT * FROM "SendingMonitorFact" WHERE "sourceId"=ANY($1::text[]) AND kind='business'`, [ids])
+  ).rows;
+  const existing = new Map(previous.map((p) => [p.sourceId, p]));
+  const facts: Fact[] = [],
+    byMessage = new Map<string, { attempt: Attempt; facts: Fact[] }>();
+  for (const a of attempts) {
+    const parts = segments.filter((s) => s.submitRecordId === a.id);
+    const times = [a.firstWireSubmitAt, ...parts.map((p) => p.firstWireSubmitAt)].filter((t): t is Date => Boolean(t));
+    const at = times.length ? new Date(Math.min(...times.map((t) => t.getTime()))) : a.createdAt;
+    const source = times.length ? 'gateway_write_complete' : a.wireTimeSource;
+    // A queued attempt with no result may have crashed after write. Keep it visible as unassessable.
+    if (source === 'not_written' && !times.length) continue;
+    const successAt = completeAttempt(
+      parts.map((p) => ({ index: p.segmentIndex, total: p.segmentTotal, firstSuccess: p.firstSuccess })),
+    );
+    const receiptRequested = a.receiptRequested ?? parts.find((p) => p.receiptRequested !== null)?.receiptRequested;
+    const reason =
+      source !== 'gateway_write_complete'
+        ? source === 'write_uncertain'
+          ? 'write_uncertain'
+          : 'missing_wire_time'
+        : receiptRequested === false
+          ? 'receipt_not_requested'
+          : receiptRequested !== true
+            ? 'missing_receipt_configuration'
+            : !parts.length
+              ? 'missing_segment_snapshot'
+              : parts.some((p) => p.ambiguousReceipt)
+                ? 'ambiguous_receipt'
+                : parts.some((p) => p.firstSuccess && p.firstWireSubmitAt && p.firstSuccess < p.firstWireSubmitAt)
+                  ? 'negative_latency'
+                  : parts.some((p) => p.missingReceiptTime && !p.firstSuccess)
+                    ? 'missing_receipt_time'
+                    : successAt && successAt < at
+                      ? 'negative_latency'
+                      : null;
+    const dim: Dimensions = { channelId: a.channelId, channelName: a.channelName, carrier: a.carrier ?? 'unknown' };
+    const fact: Fact = {
+      id: `attempt:${a.id}`,
+      kind: 'attempt',
+      sourceId: a.id,
+      messageRecordId: a.messageRecordId,
+      dimensionKey: keyOf([a.channelId, dim.carrier]),
+      dimensions: dim,
+      submittedAt: at.toISOString(),
+      successAt: successAt?.toISOString() ?? null,
+      verification: existing.get(a.messageRecordId)?.verification ?? a.content.includes('验证码'),
+      reason,
+    };
+    facts.push(fact);
+    const group = byMessage.get(a.messageRecordId) ?? { attempt: a, facts: [] };
+    group.facts.push(fact);
+    byMessage.set(a.messageRecordId, group);
+  }
+  for (const [id, { attempt: a, facts: tries }] of byMessage) {
+    const prev = existing.get(id);
+    const dimensions = prev?.dimensions ?? {
+      tenantId: a.tenantId ?? '',
+      tenantName: a.tenantName ?? '未识别企业',
+      applicationId: a.applicationId ?? '',
+      applicationName: a.applicationName ?? '未识别应用',
+      signatureId: a.signatureId ?? '',
+      signatureName: a.signatureName ?? '未识别签名',
+    };
+    const at = new Date(Math.min(...tries.map((f) => new Date(f.submittedAt).getTime())));
+    const successes = tries.filter((f) => !f.reason && f.successAt).map((f) => new Date(f.successAt!).getTime());
+    const successAt = successes.length ? new Date(Math.min(...successes)) : null;
+    const reason =
+      !a.tenantId || !a.applicationId || !a.signatureId
+        ? 'missing_dimension'
+        : (tries.find((f) => f.reason === 'missing_wire_time' || f.reason === 'write_uncertain')?.reason ??
+          (tries.every((f) => f.reason) ? tries[0].reason : successAt && successAt < at ? 'negative_latency' : null));
+    facts.push({
+      id: `business:${id}`,
+      kind: 'business',
+      sourceId: id,
+      messageRecordId: id,
+      dimensionKey: keyOf([a.tenantId, a.applicationId, a.signatureId]),
+      dimensions,
+      submittedAt: at.toISOString(),
+      successAt: successAt?.toISOString() ?? null,
+      verification: prev?.verification ?? a.content.includes('验证码'),
+      reason,
+    });
+  }
+  if (!facts.length) return;
+  const old = (
+    await db.query(`SELECT id,"submittedAt" FROM "SendingMonitorFact" WHERE id=ANY($1::text[])`, [
+      facts.map((f) => f.id),
+    ])
+  ).rows;
+  const changed = await db.query(
+    `INSERT INTO "SendingMonitorFact" (id,kind,"sourceId","messageRecordId","dimensionKey",dimensions,"submittedAt","successAt",verification,reason)
+    SELECT id,kind,"sourceId","messageRecordId","dimensionKey",dimensions,"submittedAt","successAt",verification,reason FROM jsonb_to_recordset($1::jsonb)
+      AS f(id text,kind text,"sourceId" text,"messageRecordId" text,"dimensionKey" text,dimensions jsonb,"submittedAt" timestamp,"successAt" timestamp,verification boolean,reason text)
+    ON CONFLICT(id) DO UPDATE SET "submittedAt"=EXCLUDED."submittedAt","successAt"=EXCLUDED."successAt",reason=EXCLUDED.reason,"updatedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')
+    WHERE ("SendingMonitorFact"."submittedAt","SendingMonitorFact"."successAt","SendingMonitorFact".reason) IS DISTINCT FROM (EXCLUDED."submittedAt",EXCLUDED."successAt",EXCLUDED.reason) RETURNING id,"submittedAt"`,
+    [JSON.stringify(facts)],
+  );
+  if (!changed.rows.length) return [];
+  const changedIds = new Set(changed.rows.map((f) => f.id));
+  const minutes = [
+    ...new Set(
+      [...changed.rows, ...old.filter((f) => changedIds.has(f.id))].map((f) =>
+        new Date(Math.floor(new Date(f.submittedAt).getTime() / 60000) * 60000).toISOString(),
+      ),
+    ),
+  ];
+  await db.query(`DELETE FROM "SendingMonitorMinute" WHERE minute=ANY($1::timestamp[])`, [minutes]);
+  await db.query(
+    `INSERT INTO "SendingMonitorMinute" (kind,"dimensionKey",minute,verification,dimensions,metrics)
+    SELECT kind,"dimensionKey",date_trunc('minute',"submittedAt"),verification,(jsonb_agg(dimensions ORDER BY "updatedAt" DESC))->0,
+      jsonb_build_object('total',count(*),'eligible',count(*) FILTER(WHERE reason IS NULL),'unassessable',count(*) FILTER(WHERE reason IS NOT NULL),
+      's5',count(*) FILTER(WHERE reason IS NULL AND "successAt"-"submittedAt" BETWEEN interval '0 seconds' AND interval '5 seconds'),
+      's20',count(*) FILTER(WHERE reason IS NULL AND "successAt"-"submittedAt" BETWEEN interval '0 seconds' AND interval '20 seconds'),
+      's60',count(*) FILTER(WHERE reason IS NULL AND "successAt"-"submittedAt" BETWEEN interval '0 seconds' AND interval '60 seconds'),
+      's300',count(*) FILTER(WHERE reason IS NULL AND "successAt"-"submittedAt" BETWEEN interval '0 seconds' AND interval '300 seconds'),
+      's1200',count(*) FILTER(WHERE reason IS NULL AND "successAt"-"submittedAt" BETWEEN interval '0 seconds' AND interval '1200 seconds'))
+    FROM "SendingMonitorFact" WHERE "submittedAt">=ANY($1::timestamp[]) AND date_trunc('minute',"submittedAt")=ANY($1::timestamp[])
+    GROUP BY kind,"dimensionKey",date_trunc('minute',"submittedAt"),verification`,
+    [minutes],
+  );
+  return minutes;
+}

api/src/sending-monitor/sending-monitor.module.ts

@@ -0,0 +1,397 @@
+import {
+  BadRequestException,
+  Body,
+  ConflictException,
+  Controller,
+  ForbiddenException,
+  Get,
+  Injectable,
+  Module,
+  NotFoundException,
+  Param,
+  Post,
+  Put,
+  Query,
+  Req,
+} from '@nestjs/common';
+import { randomUUID } from 'node:crypto';
+import { PrismaService } from '../prisma/prisma.service';
+import { PrismaModule } from '../prisma/prisma.module';
+import type { SessionRequest } from '../auth/session-validation.middleware';
+import {
+  horizons,
+  matchRules,
+  validateRule,
+  type MonitorScope,
+  type MonitorType,
+  type Rule,
+  type RuleConfig,
+} from './monitor-metrics';
+
+type QueryParams = Record<string, string | undefined>;
+const pageNumber = (value: string | undefined, fallback: number, max = 100000) => {
+  const n = value === undefined ? fallback : Number(value);
+  if (!Number.isSafeInteger(n) || n < 1 || n > max) throw new BadRequestException('分页参数无效');
+  return n;
+};
+const healthy = `EXISTS(SELECT 1 FROM "SendingMonitorCheckpoint" WHERE id='health' AND "updatedAt">(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')-interval '30 seconds' AND data->>'complete'='true')`;
+const monitorType = (value = 'industry'): MonitorType => {
+  if (!Object.hasOwn(horizons, value)) throw new BadRequestException('监控类型无效');
+  return value as MonitorType;
+};
+
+@Injectable()
+export class SendingMonitorService {
+  constructor(private readonly prisma: PrismaService) {}
+  async authorize(req: SessionRequest, permission: 'view' | 'rules' | 'targets' = 'view') {
+    const user =
+      req.authSession?.portal === 'admin' &&
+      req.sessionUserId &&
+      (await this.prisma.user.findFirst({
+        where: {
+          id: req.sessionUserId,
+          status: 'active',
+          deletedAt: null,
+          roles: { some: { role: { code: 'platform_admin' } } },
+        },
+        select: { id: true },
+      }));
+    if (!user) throw new ForbiddenException(`无发送监控${permission === 'view' ? '查看' : '配置'}权限`);
+    return user.id;
+  }
+  async rows(query: QueryParams) {
+    const type = monitorType(query.type),
+      page = pageNumber(query.page, 1),
+      pageSize = pageNumber(query.pageSize, 20, 100);
+    const filter = `s.type=$1 AND s."evaluationAt"=(SELECT max("evaluationAt") FROM "SendingMonitorSnapshot" WHERE type=$1)
+      AND ($2='' OR (CASE WHEN ${healthy} THEN s.status ELSE 'stale' END)=$2) AND ($3='' OR s.dimensions::text ILIKE '%'||$3||'%')
+      AND ($4='' OR s.dimensions->>'tenantId'=$4) AND ($5='' OR s.dimensions->>'applicationId'=$5) AND ($6='' OR s.dimensions->>'signatureId'=$6)`;
+    const args = [
+      type,
+      query.status ?? '',
+      query.keyword?.trim() ?? '',
+      query.tenantId ?? '',
+      query.applicationId ?? '',
+      query.signatureId ?? '',
+    ];
+    const [items, count, checkpoint] = await Promise.all([
+      this.prisma.$queryRawUnsafe(
+        `SELECT s.*,CASE WHEN ${healthy} THEN s.status ELSE 'stale' END status FROM "SendingMonitorSnapshot" s WHERE ${filter} ORDER BY (s.status='abnormal') DESC,s."dimensionKey" LIMIT $7 OFFSET $8`,
+        ...args,
+        pageSize,
+        (page - 1) * pageSize,
+      ),
+      this.prisma.$queryRawUnsafe<Array<{ total: number }>>(
+        `SELECT count(*)::int total FROM "SendingMonitorSnapshot" s WHERE ${filter}`,
+        ...args,
+      ),
+      this.prisma.$queryRawUnsafe<Array<{ data: unknown; updatedAt: Date }>>(
+        `SELECT data,"updatedAt" FROM "SendingMonitorCheckpoint" WHERE id='health'`,
+      ),
+    ]);
+    return { items, total: count[0].total, page, pageSize, health: checkpoint[0] ?? null };
+  }
+  async overview(query: QueryParams) {
+    const type = monitorType(query.type);
+    const rows = await this.prisma.$queryRawUnsafe(
+      `SELECT CASE WHEN ${healthy} THEN status ELSE 'stale' END status,count(*)::int dimensions,sum((metrics->>'total')::bigint)::text total,max("evaluationAt") "evaluationAt",max("computedAt") "computedAt"
+      FROM "SendingMonitorSnapshot" WHERE type=$1 AND "evaluationAt"=(SELECT max("evaluationAt") FROM "SendingMonitorSnapshot" WHERE type=$1) GROUP BY 1`,
+      type,
+    );
+    return { type, rows, permissions: { view: true, rules: true, targets: true } };
+  }
+  async history(query: QueryParams) {
+    const type = monitorType(query.type);
+    if (!query.dimensionId || query.dimensionId.length > 500) throw new BadRequestException('维度无效');
+    const hours = query.range === '24h' ? 24 : 2;
+    return this.prisma.$queryRawUnsafe(
+      `SELECT * FROM "SendingMonitorSnapshot" WHERE type=$1 AND "dimensionKey"=$2 AND "evaluationAt">(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')-($3::int*interval '1 hour') ORDER BY "evaluationAt" LIMIT 300`,
+      type,
+      query.dimensionId,
+      hours,
+    );
+  }
+  async targets() {
+    return this.prisma.$queryRawUnsafe(
+      `SELECT c.id,c.name,c.carrier,c.carriers,c.status,COALESCE(t.enabled,false) enabled,COALESCE(t.version,0) version,t."effectiveFrom" FROM "SmsChannel" c LEFT JOIN "SendingMonitorTarget" t ON t."channelId"=c.id WHERE c.status<>'deleted' ORDER BY c.name,c.id`,
+    );
+  }
+  async target(id: string, body: { enabled?: boolean; version?: number }, actor: string) {
+    if (!body || typeof body.enabled !== 'boolean' || !Number.isSafeInteger(body.version) || body.version! < 0)
+      throw new BadRequestException('监控配置参数无效');
+    return this.prisma.$transaction(async (tx) => {
+      const channel = await tx.smsChannel.findFirst({
+        where: { id, status: { not: 'deleted' } },
+        select: { id: true },
+      });
+      if (!channel) throw new NotFoundException('通道不存在');
+      await tx.$queryRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`monitor-target:${id}`},0))`;
+      const old = await tx.$queryRawUnsafe<Array<{ version: number; enabled: boolean }>>(
+        `SELECT * FROM "SendingMonitorTarget" WHERE "channelId"=$1`,
+        id,
+      );
+      if ((old[0]?.version ?? 0) !== body.version) throw new ConflictException('监控范围已被修改，请刷新');
+      const result = await tx.$queryRawUnsafe(
+        `INSERT INTO "SendingMonitorTarget" ("channelId",enabled,version,"effectiveFrom","updatedBy") VALUES($1,$2,1,to_timestamp((floor(extract(epoch FROM (CURRENT_TIMESTAMP AT TIME ZONE 'UTC'))/300)+1)*300) AT TIME ZONE 'UTC',$3)
+        ON CONFLICT("channelId") DO UPDATE SET enabled=EXCLUDED.enabled,version="SendingMonitorTarget".version+1,"effectiveFrom"=EXCLUDED."effectiveFrom","updatedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),"updatedBy"=$3 RETURNING *`,
+        id,
+        body.enabled,
+        actor,
+      );
+      await tx.$executeRawUnsafe(
+        `INSERT INTO "SendingMonitorTargetVersion" ("channelId",version,enabled,"effectiveFrom","updatedBy") SELECT "channelId",version,enabled,"effectiveFrom","updatedBy" FROM "SendingMonitorTarget" WHERE "channelId"=$1`,
+        id,
+      );
+      await tx.operationLog.create({
+        data: {
+          userId: actor,
+          action: 'sending_monitor.target',
+          resource: 'sending_monitor',
+          resourceId: id,
+          detail: { before: old[0] ?? null, after: { enabled: body.enabled, version: body.version! + 1 } },
+        },
+      });
+      if (!body.enabled)
+        await tx.$executeRawUnsafe(
+          `UPDATE "SendingMonitorAlert" SET state='closed',"closedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),"closeReason"='enrollment_removed' WHERE type='industry' AND dimensions->>'channelId'=$1 AND state='active'`,
+          id,
+        );
+      return result;
+    });
+  }
+  async rules() {
+    return this.prisma.$queryRawUnsafe(`SELECT * FROM "SendingMonitorRule" ORDER BY type,"scopeKey"`);
+  }
+  async options(q: QueryParams) {
+    const page = pageNumber(q.page, 1),
+      take = 20,
+      skip = (page - 1) * take,
+      keyword = q.keyword?.trim() ?? '';
+    if (q.kind === 'tenant')
+      return this.prisma.tenant.findMany({
+        where: { name: { contains: keyword } },
+        select: { id: true, name: true },
+        take,
+        skip,
+        orderBy: { name: 'asc' },
+      });
+    if (!q.tenantId) throw new BadRequestException('请先选择企业');
+    if (q.kind === 'application')
+      return this.prisma.smsApplication.findMany({
+        where: { tenantId: q.tenantId, name: { contains: keyword } },
+        select: { id: true, name: true },
+        take,
+        skip,
+        orderBy: { name: 'asc' },
+      });
+    if (q.kind === 'signature')
+      return this.prisma.smsSignature.findMany({
+        where: { tenantId: q.tenantId, applicationId: q.applicationId || undefined, name: { contains: keyword } },
+        select: { id: true, name: true },
+        take,
+        skip,
+        orderBy: { name: 'asc' },
+      });
+    throw new BadRequestException('选项类型无效');
+  }
+  async effective(query: QueryParams) {
+    const type = monitorType(query.type);
+    const rows = await this.prisma.$queryRawUnsafe<Rule[]>(
+      `SELECT DISTINCT ON ("ruleId") * FROM "SendingMonitorRuleVersion" WHERE type=$1 AND "effectiveAt"<=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC') ORDER BY "ruleId",version DESC`,
+      type,
+    );
+    return matchRules(rows, type, {
+      tenantId: query.tenantId,
+      applicationId: query.applicationId,
+      signatureId: query.signatureId,
+    });
+  }
+  async saveRule(body: { type: string; scope: MonitorScope; config: RuleConfig; version: number }, actor: string) {
+    if (!body) throw new BadRequestException('规则参数无效');
+    const error = validateRule(body.type, body.scope, body.config);
+    if (error) throw new BadRequestException(error);
+    if (!Number.isInteger(body.version) || body.version < 0) throw new BadRequestException('版本无效');
+    const scope = Object.fromEntries(
+      Object.entries(body.scope)
+        .filter(([, v]) => Boolean(v))
+        .sort(),
+    ) as MonitorScope;
+    if (
+      scope.applicationId &&
+      !(await this.prisma.smsApplication.findFirst({
+        where: { id: scope.applicationId, tenantId: scope.tenantId },
+        select: { id: true },
+      }))
+    )
+      throw new BadRequestException('应用不属于该企业');
+    if (
+      scope.signatureId &&
+      !(await this.prisma.smsSignature.findFirst({
+        where: {
+          id: scope.signatureId,
+          tenantId: scope.tenantId,
+          ...(scope.applicationId ? { applicationId: scope.applicationId } : {}),
+        },
+        select: { id: true },
+      }))
+    )
+      throw new BadRequestException('签名不属于该企业或应用');
+    const scopeKey = JSON.stringify(scope),
+      type = monitorType(body.type);
+    return this.prisma.$transaction(async (tx) => {
+      await tx.$queryRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`monitor-rule:${type}:${scopeKey}`},0))`;
+      const old = await tx.$queryRawUnsafe<Array<{ id: string; version: number }>>(
+        `SELECT * FROM "SendingMonitorRule" WHERE type=$1 AND "scopeKey"=$2`,
+        type,
+        scopeKey,
+      );
+      if ((old[0]?.version ?? 0) !== body.version) throw new ConflictException('规则已被修改，请刷新后重试');
+      const id = old[0]?.id ?? randomUUID(),
+        period = type === 'overall' ? 600 : 300;
+      const config = {
+        enabled: body.config.enabled,
+        minSamples: body.config.minSamples,
+        thresholds: body.config.thresholds,
+        consecutiveBad: body.config.consecutiveBad,
+        consecutiveGood: body.config.consecutiveGood,
+        deleted: Boolean(body.config.deleted),
+      };
+      const result = await tx.$queryRawUnsafe(
+        `INSERT INTO "SendingMonitorRule" (id,type,"scopeKey",scope,config,version,"effectiveAt","updatedBy")
+        VALUES($1,$2,$3,$4::jsonb,$5::jsonb,$6,to_timestamp((floor(extract(epoch FROM (CURRENT_TIMESTAMP AT TIME ZONE 'UTC'))/$7::int)+1)*$7::int) AT TIME ZONE 'UTC',$8)
+        ON CONFLICT(type,"scopeKey") DO UPDATE SET config=EXCLUDED.config,version=EXCLUDED.version,"effectiveAt"=EXCLUDED."effectiveAt","updatedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC'),"updatedBy"=$8 RETURNING *`,
+        id,
+        type,
+        scopeKey,
+        JSON.stringify(scope),
+        JSON.stringify(config),
+        body.version + 1,
+        period,
+        actor,
+      );
+      await tx.$executeRawUnsafe(
+        `INSERT INTO "SendingMonitorRuleVersion" ("ruleId",version,type,scope,config,"effectiveAt","createdBy") SELECT id,version,type,scope,config,"effectiveAt","updatedBy" FROM "SendingMonitorRule" WHERE id=$1`,
+        id,
+      );
+      await tx.operationLog.create({
+        data: {
+          userId: actor,
+          action: 'sending_monitor.rule',
+          resource: 'sending_monitor',
+          resourceId: id,
+          detail: { before: old[0] ?? null, after: { scope, config, version: body.version + 1 } },
+        },
+      });
+      return result;
+    });
+  }
+  async alerts(query: QueryParams, user: string) {
+    const page = pageNumber(query.page, 1),
+      size = pageNumber(query.pageSize, 20, 100);
+    const state = query.state ?? '';
+    if (state && !['active', 'recovered', 'closed'].includes(state)) throw new BadRequestException('告警状态无效');
+    const [items, total] = await Promise.all([
+      this.prisma.$queryRawUnsafe(
+        `SELECT a.*,r."readAt" IS NULL unread FROM "SendingMonitorAlert" a LEFT JOIN "SendingMonitorAlertRead" r ON r."alertId"=a.id AND r."userId"=$1 WHERE ($2='' OR a.state=$2) ORDER BY a."openedAt" DESC,a.id LIMIT $3 OFFSET $4`,
+        user,
+        state,
+        size,
+        (page - 1) * size,
+      ),
+      this.prisma.$queryRawUnsafe<Array<{ total: number }>>(
+        `SELECT count(*)::int total FROM "SendingMonitorAlert" WHERE ($1='' OR state=$1)`,
+        state,
+      ),
+    ]);
+    return { items, total: total[0].total, page, pageSize: size };
+  }
+  async summary(user: string) {
+    const rows = await this.prisma.$queryRawUnsafe(
+      `SELECT count(*) FILTER(WHERE r."readAt" IS NULL)::int count,count(*)::int "activeCount",NOT (${healthy}) unavailable FROM "SendingMonitorAlert" a LEFT JOIN "SendingMonitorAlertRead" r ON r."alertId"=a.id AND r."userId"=$1 WHERE a.state='active'`,
+      user,
+    );
+    return (rows as object[])[0];
+  }
+  async alert(id: string, user: string) {
+    const rows = await this.prisma.$queryRawUnsafe<object[]>(
+      `SELECT a.*,r."readAt" IS NULL unread FROM "SendingMonitorAlert" a LEFT JOIN "SendingMonitorAlertRead" r ON r."alertId"=a.id AND r."userId"=$2 WHERE a.id=$1`,
+      id,
+      user,
+    );
+    if (!rows.length) throw new NotFoundException('告警不存在');
+    return rows[0];
+  }
+  async read(id: string, user: string) {
+    const rows = await this.prisma.$queryRawUnsafe<Array<{ id: string }>>(
+      `SELECT id FROM "SendingMonitorAlert" WHERE id=$1`,
+      id,
+    );
+    if (!rows.length) throw new NotFoundException('告警不存在');
+    await this.prisma.$executeRawUnsafe(
+      `INSERT INTO "SendingMonitorAlertRead" ("alertId","userId") VALUES($1,$2) ON CONFLICT DO NOTHING`,
+      id,
+      user,
+    );
+    return { success: true };
+  }
+}
+
+@Controller('admin/sending-monitor')
+class SendingMonitorController {
+  constructor(private readonly service: SendingMonitorService) {}
+  @Get('rows') async rows(@Req() req: SessionRequest, @Query() q: QueryParams) {
+    await this.service.authorize(req);
+    return this.service.rows(q);
+  }
+  @Get('overview') async overview(@Req() req: SessionRequest, @Query() q: QueryParams) {
+    await this.service.authorize(req);
+    return this.service.overview(q);
+  }
+  @Get('history') async history(@Req() req: SessionRequest, @Query() q: QueryParams) {
+    await this.service.authorize(req);
+    return this.service.history(q);
+  }
+  @Get('targets') async targets(@Req() req: SessionRequest) {
+    await this.service.authorize(req);
+    return this.service.targets();
+  }
+  @Put('targets/:id') async target(
+    @Req() req: SessionRequest,
+    @Param('id') id: string,
+    @Body() body: { enabled?: boolean; version?: number },
+  ) {
+    return this.service.target(id, body, await this.service.authorize(req, 'targets'));
+  }
+  @Get('rules') async rules(@Req() req: SessionRequest) {
+    await this.service.authorize(req);
+    return this.service.rules();
+  }
+  @Get('options') async options(@Req() req: SessionRequest, @Query() q: QueryParams) {
+    await this.service.authorize(req);
+    return this.service.options(q);
+  }
+  @Post('rules') async save(
+    @Req() req: SessionRequest,
+    @Body() body: { type: string; scope: MonitorScope; config: RuleConfig; version: number },
+  ) {
+    return this.service.saveRule(body, await this.service.authorize(req, 'rules'));
+  }
+  @Get('effective-rule') async effective(@Req() req: SessionRequest, @Query() q: QueryParams) {
+    await this.service.authorize(req);
+    return this.service.effective(q);
+  }
+  @Get('alerts') async alerts(@Req() req: SessionRequest, @Query() q: QueryParams) {
+    return this.service.alerts(q, await this.service.authorize(req));
+  }
+  @Get('notification-summary') async summary(@Req() req: SessionRequest) {
+    return this.service.summary(await this.service.authorize(req));
+  }
+  @Get('alerts/:id') async alert(@Req() req: SessionRequest, @Param('id') id: string) {
+    return this.service.alert(id, await this.service.authorize(req));
+  }
+  @Post('alerts/:id/read') async read(@Req() req: SessionRequest, @Param('id') id: string) {
+    return this.service.read(id, await this.service.authorize(req));
+  }
+}
+@Module({ imports: [PrismaModule], controllers: [SendingMonitorController], providers: [SendingMonitorService] })
+export class SendingMonitorModule {}