fix: 修复上行归属并实现签名质量日报优化

api/src/signature-retirement/signature-retirement.service.ts

@@ -1,3 +1,6 @@
+import { SignatureAnalyticsRead } from '../signature-analytics/analytics-read';
+import { analyticsDate, analyticsPage, todayKey } from '../signature-analytics/analytics-date';
+import { detectRetirement } from '../signature-analytics/retirement-batch';
 import {
   BadRequestException,
   Injectable,
@@ -37,28 +40,6 @@ const shanghaiHourFormatter = new Intl.DateTimeFormat('en-GB', {
 const DAY_MS = 86_400_000;
 const CARRIERS = ['mobile', 'unicom', 'telecom'] as const;
 const DEFAULT_DELIVERY_INTERVAL_MS = 60_000;
-const carrierLabels: Record<string, string> = { mobile: '移动', unicom: '联通', telecom: '电信' };
-
-type RuleRecord = Awaited<ReturnType<PrismaService['signatureRetirementRule']['findFirst']>>;
-type DetectionDimension = {
-  dimensionType: 'enterprise' | 'channel';
-  tenantId: string;
-  applicationId: string | null;
-  signatureId: string;
-  signatureName: string;
-  tenantName: string;
-  channelId: string | null;
-  channelName: string | null;
-  carrier: string;
-  approvedAt: Date;
-  rule: NonNullable<RuleRecord>;
-};
-
-type ActivityCounts = {
-  submittedAttempts: number;
-  acceptedBusinessCount: number;
-  deliveredBusinessCount: number;
-};
 
 @Injectable()
 export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy {
@@ -66,6 +47,8 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy
   private detectionTimer?: ReturnType<typeof setTimeout>;
   private notificationTimer?: ReturnType<typeof setTimeout>;
   private deliveryTimer?: ReturnType<typeof setInterval>;
+  private compensationRunning = false;
+  private publishedDate?: string;
   private startupTimer?: ReturnType<typeof setTimeout>;
 
   constructor(private readonly prisma: PrismaService) {}
@@ -78,7 +61,7 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy
     this.scheduleDetection();
     this.scheduleNotification();
     this.deliveryTimer = setInterval(
-      () => void this.deliverPendingWebhooks(),
+      () => void this.runStartupCompensation(),
       positiveIntegerEnv('SIGNATURE_RETIREMENT_WEBHOOK_INTERVAL_MS', DEFAULT_DELIVERY_INTERVAL_MS),
     );
     this.deliveryTimer.unref?.();
@@ -540,6 +523,30 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy
   }
 
   async unreportedSignatures(query: UnreportedSignatureQuery) {
+    analyticsPage(query.page, query.pageSize);
+    const date = analyticsDate(query.date);
+    if (date !== todayKey()) return new SignatureAnalyticsRead(this.prisma).unreported({ ...query, date });
+    return this.prisma.$transaction(
+      async (tx) => {
+        await tx.$executeRawUnsafe("SET LOCAL statement_timeout='12s'");
+        const data = await new SignatureRetirementService(tx as PrismaService).unreportedSignaturesLive({
+          ...query,
+          date,
+        });
+        return {
+          ...data,
+          dataSource: 'live',
+          reportState: 'ready',
+          frozen: false,
+          sourceAsOf: new Date(),
+          serverBusinessDate: date,
+        };
+      },
+      { isolationLevel: 'RepeatableRead', timeout: 15000 },
+    );
+  }
+
+  async unreportedSignaturesLive(query: UnreportedSignatureQuery) {
     const date = assertDateKey(query.date || shanghaiDateKey());
     const page = positiveInteger(query.page, 1);
     const pageSize = Math.min(100, positiveInteger(query.pageSize, 25));
@@ -630,68 +637,16 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy
   }
 
   async runDetection(date?: string) {
-    const detectionKey = assertDateKey(date || shanghaiDateKey());
-    await this.prisma.signatureRetirementSuppression.updateMany({
-      where: { active: true, mode: 'temporary', muteUntil: { lt: databaseDate(detectionKey) } },
-      data: { active: false },
-    });
-    const [rules, approvedTasks] = await Promise.all([
-      this.prisma.signatureRetirementRule.findMany({ where: { enabled: true } }),
-      this.prisma.channelSignatureReportTask.findMany({
-        where: {
-          reportType: 'signature',
-          status: 'approved',
-          carrier: { not: null },
-          approvalScope: 'carrier_specific',
-          approvedAt: { not: null },
-          signature: { auditStatus: { not: 'deleted' } },
-          channel: { status: { not: 'deleted' } },
-        },
-        include: { signature: { include: { tenant: true, application: true } }, channel: true },
-      }),
-    ]);
-    const dimensions = this.buildDimensions(rules, approvedTasks);
-    let alerted = 0;
-    let healthy = 0;
-    let ineligible = 0;
-    for (const dimension of dimensions) {
-      const { windowDays, threshold } = carrierRule(dimension.rule, dimension.carrier);
-      const windowStartKey = addDays(detectionKey, -windowDays);
-      const windowStart = shanghaiStart(windowStartKey);
-      const activityStart = shanghaiStart(addDays(detectionKey, -1));
-      const activityEnd = shanghaiStart(detectionKey);
-      const effectiveActivityStart = dimension.approvedAt > activityStart ? dimension.approvedAt : activityStart;
-      if (effectiveActivityStart >= activityEnd) {
-        ineligible += 1;
-        continue;
-      }
-      const dailyCounts = await this.activityCounts(dimension, effectiveActivityStart, activityEnd);
-      if (dimension.approvedAt > windowStart) {
-        // 观察期只禁止预警，不能吞掉真实发送快照，否则热力图会错误显示无数据。
-        await this.persistDetection(detectionKey, dimension, windowDays, threshold, dailyCounts, false, true);
-        ineligible += 1;
-        continue;
-      }
-      const windowCounts = await this.activityCounts(dimension, windowStart, activityEnd);
-      const isAlert = windowCounts.acceptedBusinessCount < threshold;
-      await this.persistDetection(
-        detectionKey,
-        dimension,
-        windowDays,
-        threshold,
-        dailyCounts,
-        isAlert,
-        false,
-        windowCounts,
-      );
-      if (isAlert) alerted += 1;
-      else healthy += 1;
-    }
-    return { detectionDate: detectionKey, dimensions: dimensions.length, alerted, healthy, ineligible };
+    return detectRetirement(this.prisma, analyticsDate(date));
   }
 
   async publishNotifications(date?: string) {
-    const notificationKey = assertDateKey(date || shanghaiDateKey());
+    const notificationKey = analyticsDate(date);
+    if (this.publishedDate === notificationKey) return { notificationDate: notificationKey, created: 0 };
+    const completed = await this.prisma.signatureAnalyticsRun.findUnique({
+      where: { scope_businessDate: { scope: 'retirement', businessDate: databaseDate(notificationKey) } },
+    });
+    if (completed?.state !== 'succeeded') throw new Error('签名退网检测尚未完整完成，暂不发布通知');
     const detections = await this.prisma.signatureRetirementDetection.findMany({
       where: {
         detectionDate: databaseDate(notificationKey),
@@ -749,10 +704,13 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy
       }
     }
     await this.enqueueWebhookSummaries(notificationKey);
+    this.publishedDate = notificationKey;
     return { notificationDate: notificationKey, created };
   }
 
   private async runStartupCompensation() {
+    if (this.compensationRunning) return;
+    this.compensationRunning = true;
     const now = new Date();
     const hour = shanghaiHour(now);
     try {
@@ -765,6 +723,8 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy
       this.logger.error(
         `Signature retirement startup compensation failed: ${error instanceof Error ? error.message : String(error)}`,
       );
+    } finally {
+      this.compensationRunning = false;
     }
   }
 
@@ -801,231 +761,6 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy
     this.notificationTimer.unref?.();
   }
 
-  private buildDimensions(
-    rules: Array<NonNullable<RuleRecord>>,
-    tasks: Array<{
-      signatureId: string;
-      channelId: string;
-      carrier: string | null;
-      approvedAt: Date | null;
-      signature: { tenantId: string; applicationId: string | null; name: string; tenant: { name: string } };
-      channel: { name: string };
-    }>,
-  ) {
-    const dimensions: DetectionDimension[] = [];
-    const enterprise = new Map<string, DetectionDimension>();
-    for (const task of tasks) {
-      if (!task.carrier || !task.approvedAt) continue;
-      const channelRule = selectRule(rules, 'channel', task.channelId);
-      if (channelRule)
-        dimensions.push({
-          dimensionType: 'channel',
-          tenantId: task.signature.tenantId,
-          applicationId: task.signature.applicationId,
-          signatureId: task.signatureId,
-          signatureName: task.signature.name,
-          tenantName: task.signature.tenant.name,
-          channelId: task.channelId,
-          channelName: task.channel.name,
-          carrier: task.carrier,
-          approvedAt: task.approvedAt,
-          rule: channelRule,
-        });
-      const enterpriseRule = selectRule(rules, 'enterprise', task.signature.applicationId);
-      if (!enterpriseRule) continue;
-      const key = `${task.signatureId}:${task.carrier}`;
-      const current = enterprise.get(key);
-      if (!current || task.approvedAt < current.approvedAt)
-        enterprise.set(key, {
-          dimensionType: 'enterprise',
-          tenantId: task.signature.tenantId,
-          applicationId: task.signature.applicationId,
-          signatureId: task.signatureId,
-          signatureName: task.signature.name,
-          tenantName: task.signature.tenant.name,
-          channelId: null,
-          channelName: null,
-          carrier: task.carrier,
-          approvedAt: task.approvedAt,
-          rule: enterpriseRule,
-        });
-    }
-    return [...enterprise.values(), ...dimensions];
-  }
-
-  private async activityCounts(dimension: DetectionDimension, startAt: Date, endAt: Date): Promise<ActivityCounts> {
-    const channelFilter = dimension.channelId
-      ? Prisma.sql`AND submit."channelId" = ${dimension.channelId}`
-      : Prisma.empty;
-    const rows = await this.prisma.$queryRaw<Array<ActivityCounts>>(Prisma.sql`
-      WITH attempts AS (
-        SELECT
-          submit.id,
-          submit."messageRecordId" AS message_id,
-          submit."submitStatus" AS submit_status,
-          CASE
-            WHEN EXISTS (SELECT 1 FROM "SmsMessageSegmentAudit" segment WHERE segment."submitRecordId" = submit.id)
-              THEN NOT EXISTS (
-                SELECT 1 FROM "SmsMessageSegmentAudit" segment
-                WHERE segment."submitRecordId" = submit.id AND segment."receiptStatus" IS DISTINCT FROM 'delivered'
-              )
-            ELSE EXISTS (
-              SELECT 1 FROM "SmsReceiptRecord" receipt
-              WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
-                AND receipt."channelId" = submit."channelId"
-                AND receipt."receiptStatus" = 'delivered'
-            )
-          END AS delivery_success
-        FROM "SmsSubmitRecord" submit
-        JOIN "SmsMessageRecord" message ON message.id = submit."messageRecordId"
-        WHERE message."signatureId" = ${dimension.signatureId}
-          AND message.carrier = ${dimension.carrier}
-          AND COALESCE(submit."submittedAt", submit."createdAt") >= ${startAt}
-          AND COALESCE(submit."submittedAt", submit."createdAt") < ${endAt}
-          ${channelFilter}
-      )
-      SELECT
-        COUNT(id)::integer AS "submittedAttempts",
-        COUNT(DISTINCT message_id) FILTER (WHERE submit_status = 'accepted')::integer AS "acceptedBusinessCount",
-        COUNT(DISTINCT message_id) FILTER (WHERE submit_status = 'accepted' AND delivery_success)::integer AS "deliveredBusinessCount"
-      FROM attempts
-    `);
-    return rows[0] ?? { submittedAttempts: 0, acceptedBusinessCount: 0, deliveredBusinessCount: 0 };
-  }
-
-  private async persistDetection(
-    dateKey: string,
-    dimension: DetectionDimension,
-    windowDays: number,
-    threshold: number,
-    counts: ActivityCounts,
-    isAlert: boolean,
-    observing = false,
-    alertCounts = counts,
-  ) {
-    const detectionDate = databaseDate(dateKey);
-    const channelKey = dimension.channelId ?? '';
-    const existingDetection = await this.prisma.signatureRetirementDetection.findUnique({
-      where: {
-        detectionDate_dimensionType_signatureId_channelKey_carrier: {
-          detectionDate,
-          dimensionType: dimension.dimensionType,
-          signatureId: dimension.signatureId,
-          channelKey,
-          carrier: dimension.carrier,
-        },
-      },
-      select: { id: true },
-    });
-    // 同一检测日的结果冻结规则版本；规则变更只在下一检测日生效。
-    if (existingDetection) return;
-    const suppression = await this.prisma.signatureRetirementSuppression.findUnique({
-      where: {
-        dimensionType_signatureId_channelKey_carrier: {
-          dimensionType: dimension.dimensionType,
-          signatureId: dimension.signatureId,
-          channelKey,
-          carrier: dimension.carrier,
-        },
-      },
-    });
-    const suppressed = Boolean(
-      suppression?.active &&
-      (suppression.mode === 'permanent' || !suppression.muteUntil || suppression.muteUntil >= detectionDate),
-    );
-    let cycle = await this.prisma.signatureRetirementCycle.findFirst({
-      where: {
-        dimensionType: dimension.dimensionType,
-        signatureId: dimension.signatureId,
-        channelKey,
-        carrier: dimension.carrier,
-        status: 'open',
-      },
-    });
-    if (observing) {
-      cycle = null;
-    } else if (isAlert) {
-      if (!cycle) {
-        try {
-          cycle = await this.prisma.signatureRetirementCycle.create({
-            data: {
-              dimensionType: dimension.dimensionType,
-              signatureId: dimension.signatureId,
-              channelId: dimension.channelId,
-              channelKey,
-              carrier: dimension.carrier,
-              startedOn: detectionDate,
-              lastDetectedOn: detectionDate,
-            },
-          });
-        } catch (error) {
-          if (!isPrismaUniqueError(error)) throw error;
-          cycle = await this.prisma.signatureRetirementCycle.findFirst({
-            where: {
-              dimensionType: dimension.dimensionType,
-              signatureId: dimension.signatureId,
-              channelKey,
-              carrier: dimension.carrier,
-              status: 'open',
-            },
-          });
-        }
-      } else {
-        cycle = await this.prisma.signatureRetirementCycle.update({
-          where: { id: cycle.id },
-          data: { lastDetectedOn: detectionDate },
-        });
-      }
-    } else if (cycle) {
-      await this.prisma.signatureRetirementCycle.update({
-        where: { id: cycle.id },
-        data: { status: 'resolved', resolvedOn: detectionDate, lastDetectedOn: detectionDate },
-      });
-      cycle = null;
-    }
-    const notificationTitle =
-      isAlert && cycle ? (dimension.dimensionType === 'enterprise' ? '企业签名清退预警' : '通道签名清退预警') : null;
-    const notificationContent =
-      isAlert && cycle
-        ? renderMessage(
-            dimension.rule.messageTemplate,
-            dimension,
-            windowDays,
-            threshold,
-            alertCounts.acceptedBusinessCount,
-          )
-        : null;
-    try {
-      await this.prisma.signatureRetirementDetection.create({
-        data: {
-          detectionDate,
-          dimensionType: dimension.dimensionType,
-          tenantId: dimension.tenantId,
-          applicationId: dimension.applicationId,
-          signatureId: dimension.signatureId,
-          channelId: dimension.channelId,
-          channelKey,
-          carrier: dimension.carrier,
-          windowDays,
-          threshold,
-          ...counts,
-          approvedAt: dimension.approvedAt,
-          ruleId: dimension.rule.id,
-          ruleVersion: dimension.rule.version,
-          status: observing ? 'observing' : isAlert ? 'alert' : 'healthy',
-          cycleId: cycle?.id,
-          suppressed,
-          notificationTitle,
-          notificationContent,
-        },
-      });
-    } catch (error) {
-      // 两个检测实例可能同时越过前置查询；唯一快照已由另一实例生成时直接结束本维度。
-      if (isPrismaUniqueError(error)) return;
-      throw error;
-    }
-  }
-
   private async enqueueWebhookSummaries(dateKey: string) {
     const detectionDate = databaseDate(dateKey);
     const [webhooks, messages] = await Promise.all([
@@ -1119,46 +854,6 @@ export class SignatureRetirementService implements OnModuleInit, OnModuleDestroy
   }
 }
 
-function selectRule(
-  rules: Array<NonNullable<RuleRecord>>,
-  dimension: 'enterprise' | 'channel',
-  targetId: string | null,
-) {
-  const specialType = dimension === 'enterprise' ? 'enterprise_application' : 'channel';
-  const globalType = dimension === 'enterprise' ? 'enterprise_global' : 'channel_global';
-  return (
-    (targetId ? rules.find((rule) => rule.ruleType === specialType && rule.targetId === targetId) : undefined) ??
-    rules.find((rule) => rule.ruleType === globalType && rule.targetKey === '')
-  );
-}
-
-function carrierRule(rule: NonNullable<RuleRecord>, carrier: string) {
-  if (carrier === 'mobile') return { windowDays: rule.mobileWindowDays, threshold: rule.mobileThreshold };
-  if (carrier === 'unicom') return { windowDays: rule.unicomWindowDays, threshold: rule.unicomThreshold };
-  return { windowDays: rule.telecomWindowDays, threshold: rule.telecomThreshold };
-}
-
-function renderMessage(
-  template: string | null,
-  dimension: DetectionDimension,
-  windowDays: number,
-  threshold: number,
-  actual: number,
-) {
-  const fallback =
-    dimension.dimensionType === 'enterprise'
-      ? '请通知 {enterprise}：{signature}在{carrier}过去{days}天发送{actual}条，低于{threshold}条，请及时进行保签名发送。'
-      : '{signature}在通道{channel}的{carrier}过去{days}天发送{actual}条，低于{threshold}条，请及时通知客户或进行保签名发送。';
-  return (template?.trim() || fallback)
-    .replaceAll('{enterprise}', dimension.tenantName)
-    .replaceAll('{signature}', dimension.signatureName)
-    .replaceAll('{channel}', dimension.channelName ?? '-')
-    .replaceAll('{carrier}', carrierLabels[dimension.carrier] ?? dimension.carrier)
-    .replaceAll('{days}', String(windowDays))
-    .replaceAll('{threshold}', String(threshold))
-    .replaceAll('{actual}', String(actual));
-}
-
 function shanghaiDateKey(date = new Date()) {
   return shanghaiDayFormatter.format(date);
 }