fix: prevent daily report refresh timeouts

2026-09-08 17:16:43 +08:00
parent 2a9d03be2e
commit ebb185b22b
7 changed files with 1031 additions and 106 deletions
@@ -6,6 +6,8 @@ import { PrismaService } from '../prisma/prisma.service';
 const SHANGHAI_OFFSET_MS = 8 * 60 * 60 * 1000;
 const DAY_MS = 24 * 60 * 60 * 1000;
 const DEFAULT_REFRESH_INTERVAL_MS = 60 * 60 * 1000;
+const DEFAULT_REFRESH_TRANSACTION_TIMEOUT_MS = 30_000;
+const REPORT_LOCK_NAMESPACE = 0x434d5052;

 export type ReportListQuery = {
  dateFrom?: string;
@@ -21,6 +23,7 @@ export type ReportListQuery = {
@Injectable()
 export class ReportsService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(ReportsService.name);
+  private startupTimer?: ReturnType<typeof setTimeout>;
  private refreshTimer?: ReturnType<typeof setInterval>;
  private refreshRunning = false;
  private lastRefreshBusinessDate?: string;
@@ -29,8 +32,8 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {

  onModuleInit() {
    if (process.env.REPORT_DAILY_REFRESH_ENABLED === 'false') return;
-    const startupTimer = setTimeout(() => void this.runScheduledRefresh(), 15_000);
-    startupTimer.unref?.();
+    this.startupTimer = setTimeout(() => void this.runScheduledRefresh(), 15_000);
+    this.startupTimer.unref?.();
    this.refreshTimer = setInterval(
      () => void this.runScheduledRefresh(),
      positiveInteger(process.env.REPORT_REFRESH_INTERVAL_MS, DEFAULT_REFRESH_INTERVAL_MS),
@@ -39,6 +42,7 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
  }

  onModuleDestroy() {
+    if (this.startupTimer) clearTimeout(this.startupTimer);
    if (this.refreshTimer) clearInterval(this.refreshTimer);
  }

@@ -46,7 +50,12 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
    const { page, pageSize, skip } = pagination(query);
    const where = reconciliationWhere(query);
    const [items, total, aggregate] = await Promise.all([
-      this.prisma.dailyReconciliationReport.findMany({ where, orderBy: [{ reportDate: 'desc' }, { tenantName: 'asc' }, { applicationName: 'asc' }], skip, take: pageSize }),
+      this.prisma.dailyReconciliationReport.findMany({
+        where,
+        orderBy: [{ reportDate: 'desc' }, { tenantName: 'asc' }, { applicationName: 'asc' }],
+        skip,
+        take: pageSize,
+      }),
      this.prisma.dailyReconciliationReport.count({ where }),
      this.prisma.dailyReconciliationReport.aggregate({ where, _sum: reportVolumeSumSelection }),
    ]);
@@ -57,7 +66,12 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
    const { page, pageSize, skip } = pagination(query);
    const { dimensionType, where } = profitWhere(query);
    const [storedItems, total, aggregate] = await Promise.all([
-      this.prisma.dailyProfitReport.findMany({ where, orderBy: [{ reportDate: 'desc' }, { dimensionName: 'asc' }], skip, take: pageSize }),
+      this.prisma.dailyProfitReport.findMany({
+        where,
+        orderBy: [{ reportDate: 'desc' }, { dimensionName: 'asc' }],
+        skip,
+        take: pageSize,
+      }),
      this.prisma.dailyProfitReport.count({ where }),
      this.prisma.dailyProfitReport.aggregate({
        where,
@@ -65,7 +79,10 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
      }),
    ]);
    // refundCents 暂留在持久化模型中兼容既有数据和回滚，但利润报表接口不再暴露返还口径。
-    const items = storedItems.map(({ refundCents: _refundCents, ...item }) => item);
+    const items = storedItems.map(({ refundCents, ...item }) => {
+      void refundCents;
+      return item;
+    });
    const summary = {
      ...volumeSummary(aggregate._sum),
      revenueCents: Number(aggregate._sum.revenueCents ?? 0),
@@ -81,7 +98,12 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
    const { page, pageSize, skip } = pagination(query);
    const { dimensionType, where } = qualityWhere(query);
    const [items, total, aggregate] = await Promise.all([
-      this.prisma.dailyQualityReport.findMany({ where, orderBy: [{ sentUnits: 'desc' }, { reportDate: 'desc' }, { dimensionName: 'asc' }], skip, take: pageSize }),
+      this.prisma.dailyQualityReport.findMany({
+        where,
+        orderBy: [{ sentUnits: 'desc' }, { reportDate: 'desc' }, { dimensionName: 'asc' }],
+        skip,
+        take: pageSize,
+      }),
      this.prisma.dailyQualityReport.count({ where }),
      this.prisma.dailyQualityReport.aggregate({ where, _sum: reportVolumeSumSelection }),
    ]);
@@ -94,34 +116,136 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
  }

  async exportReconciliation(query: ReportListQuery) {
-    const items = await this.prisma.dailyReconciliationReport.findMany({ where: reconciliationWhere(query), orderBy: [{ reportDate: 'desc' }, { tenantName: 'asc' }, { applicationName: 'asc' }] });
-    return csvExport('对账单', ['发送日期', '企业', '企业应用', '提交条数', '发送条数', '未知条数', '成功条数', '失败条数', '生成时间'], items.map((item) => [dateKey(item.reportDate), item.tenantName, item.applicationName, item.submittedUnits, item.sentUnits, item.unknownUnits, item.successUnits, item.failedUnits, formatCsvDate(item.generatedAt)]));
+    const items = await this.prisma.dailyReconciliationReport.findMany({
+      where: reconciliationWhere(query),
+      orderBy: [{ reportDate: 'desc' }, { tenantName: 'asc' }, { applicationName: 'asc' }],
+    });
+    return csvExport(
+      '对账单',
+      ['发送日期', '企业', '企业应用', '提交条数', '发送条数', '未知条数', '成功条数', '失败条数', '生成时间'],
+      items.map((item) => [
+        dateKey(item.reportDate),
+        item.tenantName,
+        item.applicationName,
+        item.submittedUnits,
+        item.sentUnits,
+        item.unknownUnits,
+        item.successUnits,
+        item.failedUnits,
+        formatCsvDate(item.generatedAt),
+      ]),
+    );
  }

  async exportProfit(query: ReportListQuery) {
    const { dimensionType, where } = profitWhere(query);
-    const items = await this.prisma.dailyProfitReport.findMany({ where, orderBy: [{ reportDate: 'desc' }, { dimensionName: 'asc' }] });
-    return csvExport(`利润报表-${dimensionType === 'channel' ? '通道' : '企业应用'}`, ['发送日期', '统计维度', '企业', '提交条数', '发送条数', '未知条数', '成功条数', '失败条数', '收入金额(元)', '成本金额(元)', '利润(元)', '利润率(%)', '生成时间'], items.map((item) => [dateKey(item.reportDate), item.dimensionName, item.tenantName ?? '', item.submittedUnits, item.sentUnits, item.unknownUnits, item.successUnits, item.failedUnits, moneyUnitsToFixedYuan(item.revenueCents), moneyUnitsToFixedYuan(item.costCents), moneyUnitsToFixedYuan(item.profitCents), (item.profitRateBps / 100).toFixed(2), formatCsvDate(item.generatedAt)]));
+    const items = await this.prisma.dailyProfitReport.findMany({
+      where,
+      orderBy: [{ reportDate: 'desc' }, { dimensionName: 'asc' }],
+    });
+    return csvExport(
+      `利润报表-${dimensionType === 'channel' ? '通道' : '企业应用'}`,
+      [
+        '发送日期',
+        '统计维度',
+        '企业',
+        '提交条数',
+        '发送条数',
+        '未知条数',
+        '成功条数',
+        '失败条数',
+        '收入金额(元)',
+        '成本金额(元)',
+        '利润(元)',
+        '利润率(%)',
+        '生成时间',
+      ],
+      items.map((item) => [
+        dateKey(item.reportDate),
+        item.dimensionName,
+        item.tenantName ?? '',
+        item.submittedUnits,
+        item.sentUnits,
+        item.unknownUnits,
+        item.successUnits,
+        item.failedUnits,
+        moneyUnitsToFixedYuan(item.revenueCents),
+        moneyUnitsToFixedYuan(item.costCents),
+        moneyUnitsToFixedYuan(item.profitCents),
+        (item.profitRateBps / 100).toFixed(2),
+        formatCsvDate(item.generatedAt),
+      ]),
+    );
  }

  async exportQuality(query: ReportListQuery) {
    const { dimensionType, where } = qualityWhere(query);
-    const items = await this.prisma.dailyQualityReport.findMany({ where, orderBy: [{ sentUnits: 'desc' }, { reportDate: 'desc' }, { dimensionName: 'asc' }] });
-    return csvExport(`发送质量报表-${dimensionType}`, ['发送日期', '统计对象', '企业', '提交条数', '发送条数', '未知条数', '成功条数', '失败条数', '成功率(%)', '平均到达时长(毫秒)', '生成时间'], items.map((item) => [dateKey(item.reportDate), item.dimensionName, item.tenantName ?? '', item.submittedUnits, item.sentUnits, item.unknownUnits, item.successUnits, item.failedUnits, (item.successRateBps / 100).toFixed(2), item.avgArrivalMs ?? '', formatCsvDate(item.generatedAt)]));
+    const items = await this.prisma.dailyQualityReport.findMany({
+      where,
+      orderBy: [{ sentUnits: 'desc' }, { reportDate: 'desc' }, { dimensionName: 'asc' }],
+    });
+    return csvExport(
+      `发送质量报表-${dimensionType}`,
+      [
+        '发送日期',
+        '统计对象',
+        '企业',
+        '提交条数',
+        '发送条数',
+        '未知条数',
+        '成功条数',
+        '失败条数',
+        '成功率(%)',
+        '平均到达时长(毫秒)',
+        '生成时间',
+      ],
+      items.map((item) => [
+        dateKey(item.reportDate),
+        item.dimensionName,
+        item.tenantName ?? '',
+        item.submittedUnits,
+        item.sentUnits,
+        item.unknownUnits,
+        item.successUnits,
+        item.failedUnits,
+        (item.successRateBps / 100).toFixed(2),
+        item.avgArrivalMs ?? '',
+        formatCsvDate(item.generatedAt),
+      ]),
+    );
  }

  async refreshRollingWindow(now = new Date()) {
    const days = completedBusinessDays(now, 4);
-    for (const day of days) await this.refreshBusinessDay(day);
-    return { refreshedDates: days.map((day) => day.key) };
+    const refreshedDates: string[] = [];
+    const failedDates: string[] = [];
+    for (const day of days) {
+      try {
+        await this.refreshBusinessDay(day);
+        refreshedDates.push(day.key);
+      } catch (error) {
+        failedDates.push(day.key);
+        this.logger.error(
+          `Daily report refresh failed for ${day.key}`,
+          error instanceof Error ? error.stack : String(error),
+        );
+      }
+    }
+    if (failedDates.length) {
+      throw new Error(
+        `Daily report refresh incomplete; failed dates: ${failedDates.join(', ')}; refreshed dates: ${refreshedDates.join(', ') || 'none'}`,
+      );
+    }
+    return { refreshedDates };
  }

  private async runScheduledRefresh() {
-    const businessDate = shanghaiDateKey(new Date());
+    const now = new Date();
+    const businessDate = shanghaiDateKey(now);
    if (this.refreshRunning || this.lastRefreshBusinessDate === businessDate) return;
    this.refreshRunning = true;
    try {
-      const result = await this.refreshRollingWindow();
+      const result = await this.refreshRollingWindow(now);
      this.lastRefreshBusinessDate = businessDate;
      this.logger.log(`Daily reports refreshed for ${result.refreshedDates.join(', ')}`);
    } catch (error) {
@@ -132,12 +256,22 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
  }

  private async refreshBusinessDay(day: BusinessDay) {
-    await this.prisma.$transaction(async (tx) => {
-      await tx.dailyReconciliationReport.deleteMany({ where: { reportDate: day.reportDate } });
-      await tx.dailyProfitReport.deleteMany({ where: { reportDate: day.reportDate } });
-      await tx.dailyQualityReport.deleteMany({ where: { reportDate: day.reportDate } });
+    const timeout = Math.min(
+      120_000,
+      positiveInteger(process.env.REPORT_REFRESH_TRANSACTION_TIMEOUT_MS, DEFAULT_REFRESH_TRANSACTION_TIMEOUT_MS),
+    );
+    await this.prisma.$transaction(
+      async (tx) => {
+        await tx.$executeRaw(Prisma.sql`SELECT set_config('statement_timeout', ${`${timeout}ms`}, true)`);
+        const [lock] = await tx.$queryRaw<{ locked: boolean }[]>(Prisma.sql`
+        SELECT pg_try_advisory_xact_lock(${REPORT_LOCK_NAMESPACE}::integer, ${Number(day.key.replaceAll('-', ''))}::integer) AS locked
+      `);
+        if (!lock?.locked) throw new Error(`Daily reports for ${day.key} are being refreshed by another transaction`);
+        await tx.dailyReconciliationReport.deleteMany({ where: { reportDate: day.reportDate } });
+        await tx.dailyProfitReport.deleteMany({ where: { reportDate: day.reportDate } });
+        await tx.dailyQualityReport.deleteMany({ where: { reportDate: day.reportDate } });

-      await tx.$executeRaw(Prisma.sql`
+        await tx.$executeRaw(Prisma.sql`
        INSERT INTO "DailyReconciliationReport" (
          "id", "reportDate", "tenantId", "tenantName", "applicationId", "applicationName",
          "submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "generatedAt", "updatedAt"
@@ -169,7 +303,7 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
        GROUP BY tenant.id, tenant.name, application.id, application.name
      `);

-      await tx.$executeRaw(Prisma.sql`
+        await tx.$executeRaw(Prisma.sql`
        WITH costs AS (
          SELECT
            submit."messageRecordId",
@@ -197,6 +331,9 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
            ) AS delivered
          ) legacy_receipt ON TRUE
          WHERE submit."submitStatus" = 'accepted'
+            -- 应用成本归属于原短信日，仍包含该短信全部跨日补发尝试。
+            AND message."queuedAt" >= ${day.startAt}
+            AND message."queuedAt" < ${day.endAt}
          GROUP BY submit."messageRecordId"
        )
        INSERT INTO "DailyProfitReport" (
@@ -249,7 +386,7 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
        GROUP BY tenant.id, tenant.name, application.id, application.name
      `);

-      await tx.$executeRaw(Prisma.sql`
+        await tx.$executeRaw(Prisma.sql`
        INSERT INTO "DailyProfitReport" (
          "id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
          "tenantId", "tenantName", "applicationId", "channelId",
@@ -348,29 +485,34 @@ export class ReportsService implements OnModuleInit, OnModuleDestroy {
        GROUP BY channel.id, channel.name
      `);

-      await tx.$executeRaw(qualityByMessageDimensionSql(day, 'application'));
-      await tx.$executeRaw(qualityByChannelSql(day));
-      await tx.$executeRaw(qualityByMessageDimensionSql(day, 'signature'));
-      await tx.$executeRaw(qualityByMessageDimensionSql(day, 'drainage'));
-    });
+        await tx.$executeRaw(qualityByMessageDimensionSql(day, 'application'));
+        await tx.$executeRaw(qualityByChannelSql(day));
+        await tx.$executeRaw(qualityByMessageDimensionSql(day, 'signature'));
+        await tx.$executeRaw(qualityByMessageDimensionSql(day, 'drainage'));
+      },
+      { maxWait: 5_000, timeout },
+    );
  }
 }

 function qualityByMessageDimensionSql(day: BusinessDay, dimensionType: 'application' | 'signature' | 'drainage') {
  const dimensionTypeSql = Prisma.raw(`'${dimensionType}'`);
-  const dimensionId = dimensionType === 'application'
-    ? Prisma.sql`application.id`
-    : dimensionType === 'signature'
-      ? Prisma.sql`COALESCE(signature.id, 'unmatched:' || COALESCE(application.id, tenant.id))`
-      : Prisma.sql`COALESCE(drainage.id, 'unmatched:' || COALESCE(application.id, tenant.id))`;
-  const dimensionName = dimensionType === 'application'
-    ? Prisma.sql`application.name`
-    : dimensionType === 'signature'
-      ? Prisma.sql`COALESCE(signature.name, '未关联签名')`
-      : Prisma.sql`COALESCE(drainage."siteName", '未关联引流信息')`;
-  const applicationJoin = dimensionType === 'application'
-    ? Prisma.sql`JOIN "SmsApplication" application ON application.id = message."applicationId"`
-    : Prisma.sql`LEFT JOIN "SmsApplication" application ON application.id = message."applicationId"`;
+  const dimensionId =
+    dimensionType === 'application'
+      ? Prisma.sql`application.id`
+      : dimensionType === 'signature'
+        ? Prisma.sql`COALESCE(signature.id, 'unmatched:' || COALESCE(application.id, tenant.id))`
+        : Prisma.sql`COALESCE(drainage.id, 'unmatched:' || COALESCE(application.id, tenant.id))`;
+  const dimensionName =
+    dimensionType === 'application'
+      ? Prisma.sql`application.name`
+      : dimensionType === 'signature'
+        ? Prisma.sql`COALESCE(signature.name, '未关联签名')`
+        : Prisma.sql`COALESCE(drainage."siteName", '未关联引流信息')`;
+  const applicationJoin =
+    dimensionType === 'application'
+      ? Prisma.sql`JOIN "SmsApplication" application ON application.id = message."applicationId"`
+      : Prisma.sql`LEFT JOIN "SmsApplication" application ON application.id = message."applicationId"`;

  return Prisma.sql`
    WITH base AS (
@@ -548,7 +690,13 @@ const reportVolumeSumSelection = {
  failedUnits: true,
 } as const;

-function volumeSummary(sum: { submittedUnits?: number | null; sentUnits?: number | null; unknownUnits?: number | null; successUnits?: number | null; failedUnits?: number | null }) {
+function volumeSummary(sum: {
+  submittedUnits?: number | null;
+  sentUnits?: number | null;
+  unknownUnits?: number | null;
+  successUnits?: number | null;
+  failedUnits?: number | null;
+}) {
  return {
    submittedUnits: Number(sum.submittedUnits ?? 0),
    sentUnits: Number(sum.sentUnits ?? 0),
@@ -559,11 +707,15 @@ function volumeSummary(sum: { submittedUnits?: number | null; sentUnits?: number
 }

 function ratioBps(numerator: number, denominator: number) {
-  return denominator === 0 ? 0 : Math.round(numerator * 10_000 / denominator);
+  return denominator === 0 ? 0 : Math.round((numerator * 10_000) / denominator);
 }

 function reconciliationWhere(query: ReportListQuery): Prisma.DailyReconciliationReportWhereInput {
-  return { reportDate: dateFilter(query.dateFrom, query.dateTo), tenantId: query.tenantId || undefined, applicationId: query.applicationId || undefined };
+  return {
+    reportDate: dateFilter(query.dateFrom, query.dateTo),
+    tenantId: query.tenantId || undefined,
+    applicationId: query.applicationId || undefined,
+  };
 }

 function profitWhere(query: ReportListQuery) {
@@ -580,7 +732,9 @@ function profitWhere(query: ReportListQuery) {

 function qualityWhere(query: ReportListQuery) {
  const allowedDimensions = new Set(['application', 'channel', 'signature', 'drainage']);
-  const dimensionType = allowedDimensions.has(String(query.dimensionType)) ? String(query.dimensionType) : 'application';
+  const dimensionType = allowedDimensions.has(String(query.dimensionType))
+    ? String(query.dimensionType)
+    : 'application';
  const where: Prisma.DailyQualityReportWhereInput = {
    dimensionType,
    reportDate: dateFilter(query.dateFrom, query.dateTo),