770 lines
36 KiB
TypeScript
770 lines
36 KiB
TypeScript
import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
|
import { Prisma } from '@prisma/client';
|
|
import { moneyUnitsToFixedYuan } from '../common/money';
|
|
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;
|
|
dateTo?: string;
|
|
tenantId?: string;
|
|
applicationId?: string;
|
|
channelId?: string;
|
|
dimensionType?: string;
|
|
page?: number;
|
|
pageSize?: number;
|
|
};
|
|
|
|
@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;
|
|
|
|
constructor(private readonly prisma: PrismaService) {}
|
|
|
|
onModuleInit() {
|
|
if (process.env.REPORT_DAILY_REFRESH_ENABLED === 'false') return;
|
|
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),
|
|
);
|
|
this.refreshTimer.unref?.();
|
|
}
|
|
|
|
onModuleDestroy() {
|
|
if (this.startupTimer) clearTimeout(this.startupTimer);
|
|
if (this.refreshTimer) clearInterval(this.refreshTimer);
|
|
}
|
|
|
|
async listReconciliation(query: ReportListQuery) {
|
|
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.count({ where }),
|
|
this.prisma.dailyReconciliationReport.aggregate({ where, _sum: reportVolumeSumSelection }),
|
|
]);
|
|
return { items, total, page, pageSize, summary: volumeSummary(aggregate._sum) };
|
|
}
|
|
|
|
async listProfit(query: ReportListQuery) {
|
|
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.count({ where }),
|
|
this.prisma.dailyProfitReport.aggregate({
|
|
where,
|
|
_sum: { ...reportVolumeSumSelection, revenueCents: true, costCents: true, profitCents: true },
|
|
}),
|
|
]);
|
|
// refundCents 暂留在持久化模型中兼容既有数据和回滚，但利润报表接口不再暴露返还口径。
|
|
const items = storedItems.map(({ refundCents, ...item }) => {
|
|
void refundCents;
|
|
return item;
|
|
});
|
|
const summary = {
|
|
...volumeSummary(aggregate._sum),
|
|
revenueCents: Number(aggregate._sum.revenueCents ?? 0),
|
|
costCents: Number(aggregate._sum.costCents ?? 0),
|
|
profitCents: Number(aggregate._sum.profitCents ?? 0),
|
|
// 利润率必须用全量筛选结果的合计利润/合计收入重新计算，不能对每日百分比求和或简单平均。
|
|
profitRateBps: ratioBps(Number(aggregate._sum.profitCents ?? 0), Number(aggregate._sum.revenueCents ?? 0)),
|
|
};
|
|
return { items, total, page, pageSize, dimensionType, summary };
|
|
}
|
|
|
|
async listQuality(query: ReportListQuery) {
|
|
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.count({ where }),
|
|
this.prisma.dailyQualityReport.aggregate({ where, _sum: reportVolumeSumSelection }),
|
|
]);
|
|
const summary = {
|
|
...volumeSummary(aggregate._sum),
|
|
// 成功率按全量筛选结果的成功量/发送量重新计算，避免分页和分组大小导致失真。
|
|
successRateBps: ratioBps(Number(aggregate._sum.successUnits ?? 0), Number(aggregate._sum.sentUnits ?? 0)),
|
|
};
|
|
return { items, total, page, pageSize, dimensionType, summary };
|
|
}
|
|
|
|
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),
|
|
]),
|
|
);
|
|
}
|
|
|
|
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),
|
|
]),
|
|
);
|
|
}
|
|
|
|
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),
|
|
]),
|
|
);
|
|
}
|
|
|
|
async refreshRollingWindow(now = new Date()) {
|
|
const days = completedBusinessDays(now, 4);
|
|
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 now = new Date();
|
|
const businessDate = shanghaiDateKey(now);
|
|
if (this.refreshRunning || this.lastRefreshBusinessDate === businessDate) return;
|
|
this.refreshRunning = true;
|
|
try {
|
|
const result = await this.refreshRollingWindow(now);
|
|
this.lastRefreshBusinessDate = businessDate;
|
|
this.logger.log(`Daily reports refreshed for ${result.refreshedDates.join(', ')}`);
|
|
} catch (error) {
|
|
this.logger.error('Daily report refresh failed', error instanceof Error ? error.stack : String(error));
|
|
} finally {
|
|
this.refreshRunning = false;
|
|
}
|
|
}
|
|
|
|
private async refreshBusinessDay(day: BusinessDay) {
|
|
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`
|
|
INSERT INTO "DailyReconciliationReport" (
|
|
"id", "reportDate", "tenantId", "tenantName", "applicationId", "applicationName",
|
|
"submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "generatedAt", "updatedAt"
|
|
)
|
|
SELECT
|
|
CONCAT('recon-', MD5(${day.key} || ':' || tenant.id || ':' || application.id)),
|
|
${day.reportDate}::date,
|
|
tenant.id,
|
|
tenant.name,
|
|
application.id,
|
|
application.name,
|
|
COALESCE(SUM(message."billingUnits"), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN COALESCE(message.status, '') <> 'rejected' THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN COALESCE(message.status, '') <> 'rejected'
|
|
AND NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
|
|
AND NOT (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
|
|
THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered' THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
|
|
AND (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
|
|
THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
CURRENT_TIMESTAMP,
|
|
CURRENT_TIMESTAMP
|
|
FROM "SmsMessageRecord" message
|
|
JOIN "Tenant" tenant ON tenant.id = message."tenantId"
|
|
JOIN "SmsApplication" application ON application.id = message."applicationId"
|
|
WHERE message."queuedAt" >= ${day.startAt}
|
|
AND message."queuedAt" < ${day.endAt}
|
|
GROUP BY tenant.id, tenant.name, application.id, application.name
|
|
`);
|
|
|
|
await tx.$executeRaw(Prisma.sql`
|
|
WITH costs AS (
|
|
SELECT
|
|
submit."messageRecordId",
|
|
SUM(submit."costUnitPrice" * CASE
|
|
WHEN segment_receipts.audit_count > 0 THEN segment_receipts.delivered_count
|
|
WHEN legacy_receipt.delivered THEN message."billingUnits"
|
|
ELSE 0
|
|
END)::bigint AS cost
|
|
FROM "SmsSubmitRecord" submit
|
|
JOIN "SmsMessageRecord" message ON message.id = submit."messageRecordId"
|
|
LEFT JOIN LATERAL (
|
|
SELECT
|
|
COUNT(*)::integer AS audit_count,
|
|
COUNT(*) FILTER (WHERE audit."receiptStatus" = 'delivered')::integer AS delivered_count
|
|
FROM "SmsMessageSegmentAudit" audit
|
|
WHERE audit."submitRecordId" = submit.id
|
|
) segment_receipts ON TRUE
|
|
LEFT JOIN LATERAL (
|
|
SELECT EXISTS (
|
|
SELECT 1
|
|
FROM "SmsReceiptRecord" receipt
|
|
WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
|
|
AND receipt."channelId" = submit."channelId"
|
|
AND receipt."receiptStatus" = 'delivered'
|
|
) 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" (
|
|
"id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
|
|
"tenantId", "tenantName", "applicationId", "channelId",
|
|
"submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "revenueCents", "refundCents", "costCents", "profitCents", "profitRateBps",
|
|
"generatedAt", "updatedAt"
|
|
)
|
|
SELECT
|
|
CONCAT('profit-app-', MD5(${day.key} || ':' || application.id)),
|
|
${day.reportDate}::date,
|
|
'application',
|
|
application.id,
|
|
application.name,
|
|
tenant.id,
|
|
tenant.name,
|
|
application.id,
|
|
NULL,
|
|
COALESCE(SUM(message."billingUnits"), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN COALESCE(message.status, '') <> 'rejected' THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN COALESCE(message.status, '') <> 'rejected'
|
|
AND NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
|
|
AND NOT (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
|
|
THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered' THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
|
|
AND (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
|
|
THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
-- 收入按每条最终成功短信的计费条数和发送时客户价快照计算，不能依赖随后可能变为 refunded 的账单状态。
|
|
COALESCE(SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered'
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END), 0)::bigint,
|
|
0::bigint,
|
|
COALESCE(SUM(costs.cost), 0)::bigint,
|
|
(COALESCE(SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered'
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END), 0) - COALESCE(SUM(costs.cost), 0))::bigint,
|
|
CASE WHEN COALESCE(SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered'
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END), 0) = 0 THEN 0
|
|
ELSE ROUND((COALESCE(SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered'
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END), 0) - COALESCE(SUM(costs.cost), 0)) * 10000.0 /
|
|
SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered'
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END))::integer END,
|
|
CURRENT_TIMESTAMP,
|
|
CURRENT_TIMESTAMP
|
|
FROM "SmsMessageRecord" message
|
|
JOIN "Tenant" tenant ON tenant.id = message."tenantId"
|
|
JOIN "SmsApplication" application ON application.id = message."applicationId"
|
|
LEFT JOIN costs ON costs."messageRecordId" = message.id
|
|
WHERE message."queuedAt" >= ${day.startAt}
|
|
AND message."queuedAt" < ${day.endAt}
|
|
GROUP BY tenant.id, tenant.name, application.id, application.name
|
|
`);
|
|
|
|
await tx.$executeRaw(Prisma.sql`
|
|
INSERT INTO "DailyProfitReport" (
|
|
"id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
|
|
"tenantId", "tenantName", "applicationId", "channelId",
|
|
"submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "revenueCents", "refundCents", "costCents", "profitCents", "profitRateBps",
|
|
"generatedAt", "updatedAt"
|
|
)
|
|
SELECT
|
|
CONCAT('profit-channel-', MD5(${day.key} || ':' || channel.id)),
|
|
${day.reportDate}::date,
|
|
'channel',
|
|
channel.id,
|
|
channel.name,
|
|
NULL,
|
|
NULL,
|
|
NULL,
|
|
channel.id,
|
|
COALESCE(SUM(message."billingUnits"), 0)::integer,
|
|
COALESCE(SUM(message."billingUnits"), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN NOT EXISTS (
|
|
SELECT 1 FROM "SmsReceiptRecord" receipt
|
|
WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
|
|
AND receipt."channelId" = submit."channelId"
|
|
AND receipt."receiptStatus" IN ('delivered', 'undelivered')
|
|
) AND submit."submitStatus" NOT IN ('rejected', 'timeout') THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN EXISTS (
|
|
SELECT 1 FROM "SmsReceiptRecord" receipt
|
|
WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
|
|
AND receipt."channelId" = submit."channelId"
|
|
AND receipt."receiptStatus" = 'delivered'
|
|
) THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
COALESCE(SUM(CASE WHEN NOT EXISTS (
|
|
SELECT 1 FROM "SmsReceiptRecord" receipt
|
|
WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
|
|
AND receipt."channelId" = submit."channelId"
|
|
AND receipt."receiptStatus" = 'delivered'
|
|
) AND (EXISTS (
|
|
SELECT 1 FROM "SmsReceiptRecord" receipt
|
|
WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
|
|
AND receipt."channelId" = submit."channelId"
|
|
AND receipt."receiptStatus" = 'undelivered'
|
|
) OR submit."submitStatus" IN ('rejected', 'timeout')) THEN message."billingUnits" ELSE 0 END), 0)::integer,
|
|
-- 补发可能产生多次提交，收入只归属最终成功提交，避免同一短信在多个通道重复计收。
|
|
COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId"
|
|
AND (message.status = 'delivered' OR message."receiptStatus" = 'delivered')
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END), 0)::bigint,
|
|
0::bigint,
|
|
COALESCE(SUM(submit."costUnitPrice" * CASE
|
|
WHEN segment_receipts.audit_count > 0 THEN segment_receipts.delivered_count
|
|
WHEN legacy_receipt.delivered THEN message."billingUnits"
|
|
ELSE 0
|
|
END), 0)::bigint,
|
|
(COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId"
|
|
AND (message.status = 'delivered' OR message."receiptStatus" = 'delivered')
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END), 0) - COALESCE(SUM(submit."costUnitPrice" * CASE
|
|
WHEN segment_receipts.audit_count > 0 THEN segment_receipts.delivered_count
|
|
WHEN legacy_receipt.delivered THEN message."billingUnits"
|
|
ELSE 0
|
|
END), 0))::bigint,
|
|
CASE WHEN COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId"
|
|
AND (message.status = 'delivered' OR message."receiptStatus" = 'delivered')
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END), 0) = 0 THEN 0
|
|
ELSE ROUND((COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId"
|
|
AND (message.status = 'delivered' OR message."receiptStatus" = 'delivered')
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END), 0) - COALESCE(SUM(submit."costUnitPrice" * CASE
|
|
WHEN segment_receipts.audit_count > 0 THEN segment_receipts.delivered_count
|
|
WHEN legacy_receipt.delivered THEN message."billingUnits"
|
|
ELSE 0
|
|
END), 0)) * 10000.0 /
|
|
SUM(CASE WHEN message."submitId" = submit."submitId"
|
|
AND (message.status = 'delivered' OR message."receiptStatus" = 'delivered')
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END))::integer END,
|
|
CURRENT_TIMESTAMP,
|
|
CURRENT_TIMESTAMP
|
|
FROM "SmsSubmitRecord" submit
|
|
JOIN "SmsMessageRecord" message ON message.id = submit."messageRecordId"
|
|
JOIN "SmsChannel" channel ON channel.id = submit."channelId"
|
|
LEFT JOIN LATERAL (
|
|
SELECT
|
|
COUNT(*)::integer AS audit_count,
|
|
COUNT(*) FILTER (WHERE audit."receiptStatus" = 'delivered')::integer AS delivered_count
|
|
FROM "SmsMessageSegmentAudit" audit
|
|
WHERE audit."submitRecordId" = submit.id
|
|
) segment_receipts ON TRUE
|
|
LEFT JOIN LATERAL (
|
|
SELECT EXISTS (
|
|
SELECT 1
|
|
FROM "SmsReceiptRecord" receipt
|
|
WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
|
|
AND receipt."channelId" = submit."channelId"
|
|
AND receipt."receiptStatus" = 'delivered'
|
|
) AS delivered
|
|
) legacy_receipt ON TRUE
|
|
WHERE submit."submitStatus" = 'accepted'
|
|
AND COALESCE(submit."submittedAt", submit."createdAt") >= ${day.startAt}
|
|
AND COALESCE(submit."submittedAt", submit."createdAt") < ${day.endAt}
|
|
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'));
|
|
},
|
|
{ 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"`;
|
|
|
|
return Prisma.sql`
|
|
WITH base AS (
|
|
SELECT
|
|
${dimensionId} AS dimension_id,
|
|
${dimensionName} AS dimension_name,
|
|
tenant.id AS tenant_id,
|
|
tenant.name AS tenant_name,
|
|
application.id AS application_id,
|
|
message."billingUnits" AS billing_units,
|
|
CASE WHEN COALESCE(message.status, '') <> 'rejected' THEN message."billingUnits" ELSE 0 END AS sent_units,
|
|
CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered' THEN message."billingUnits" ELSE 0 END AS success_units,
|
|
CASE WHEN NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
|
|
AND (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
|
|
THEN message."billingUnits" ELSE 0 END AS failed_units,
|
|
CASE WHEN COALESCE(message.status, '') <> 'rejected'
|
|
AND NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
|
|
AND NOT (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
|
|
THEN message."billingUnits" ELSE 0 END AS unknown_units,
|
|
CASE WHEN (message.status = 'delivered' OR message."receiptStatus" = 'delivered')
|
|
AND message."submittedAt" IS NOT NULL AND message."deliveredAt" >= message."submittedAt"
|
|
THEN EXTRACT(EPOCH FROM (message."deliveredAt" - message."submittedAt")) * 1000 END AS arrival_ms
|
|
FROM "SmsMessageRecord" message
|
|
JOIN "Tenant" tenant ON tenant.id = message."tenantId"
|
|
${applicationJoin}
|
|
LEFT JOIN "SmsSignature" signature ON signature.id = message."signatureId"
|
|
LEFT JOIN "SmsDrainageInfo" drainage ON ${dimensionType === 'drainage' ? Prisma.sql`(CASE WHEN message."drainageGate" IS NULL THEN drainage.id = message."drainageInfoId" ELSE EXISTS (SELECT 1 FROM jsonb_array_elements(COALESCE(message."drainageGate"->'targets', '[]'::jsonb)) target WHERE target->'materialIds' ? drainage.id) END)` : Prisma.sql`drainage.id = message."drainageInfoId"`}
|
|
WHERE message."queuedAt" >= ${day.startAt}
|
|
AND message."queuedAt" < ${day.endAt}
|
|
), thresholds AS (
|
|
SELECT dimension_id, PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY arrival_ms) AS p95_ms
|
|
FROM base WHERE arrival_ms IS NOT NULL GROUP BY dimension_id
|
|
)
|
|
INSERT INTO "DailyQualityReport" (
|
|
"id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
|
|
"tenantId", "tenantName", "applicationId", "channelId", "signatureId", "drainageInfoId",
|
|
"submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "successRateBps", "avgArrivalMs", "generatedAt", "updatedAt"
|
|
)
|
|
SELECT
|
|
CONCAT('quality-', ${dimensionTypeSql}, '-', MD5(${day.key} || ':' || base.dimension_id)),
|
|
${day.reportDate}::date,
|
|
${dimensionTypeSql},
|
|
base.dimension_id,
|
|
MAX(base.dimension_name),
|
|
MAX(base.tenant_id),
|
|
MAX(base.tenant_name),
|
|
CASE WHEN ${dimensionTypeSql} = 'application' THEN base.dimension_id ELSE MAX(base.application_id) END,
|
|
NULL,
|
|
CASE WHEN ${dimensionTypeSql} = 'signature' AND base.dimension_id NOT LIKE 'unmatched:%' THEN base.dimension_id ELSE NULL END,
|
|
CASE WHEN ${dimensionTypeSql} = 'drainage' AND base.dimension_id NOT LIKE 'unmatched:%' THEN base.dimension_id ELSE NULL END,
|
|
SUM(base.billing_units)::integer,
|
|
SUM(base.sent_units)::integer,
|
|
SUM(base.unknown_units)::integer,
|
|
SUM(base.success_units)::integer,
|
|
SUM(base.failed_units)::integer,
|
|
CASE WHEN SUM(base.sent_units) = 0 THEN 0 ELSE ROUND(SUM(base.success_units) * 10000.0 / SUM(base.sent_units))::integer END,
|
|
ROUND(AVG(base.arrival_ms) FILTER (WHERE base.arrival_ms IS NOT NULL AND base.arrival_ms <= thresholds.p95_ms))::integer,
|
|
CURRENT_TIMESTAMP,
|
|
CURRENT_TIMESTAMP
|
|
FROM base
|
|
LEFT JOIN thresholds ON thresholds.dimension_id = base.dimension_id
|
|
GROUP BY base.dimension_id
|
|
`;
|
|
}
|
|
|
|
function qualityByChannelSql(day: BusinessDay) {
|
|
return Prisma.sql`
|
|
WITH base AS (
|
|
SELECT
|
|
channel.id AS dimension_id,
|
|
channel.name AS dimension_name,
|
|
message."billingUnits" AS billing_units,
|
|
CASE WHEN receipt."deliveredAt" IS NOT NULL THEN message."billingUnits" ELSE 0 END AS success_units,
|
|
CASE WHEN receipt."deliveredAt" IS NULL AND failed_receipt."failedAt" IS NOT NULL THEN message."billingUnits" ELSE 0 END AS failed_units,
|
|
CASE WHEN receipt."deliveredAt" IS NULL AND failed_receipt."failedAt" IS NULL THEN message."billingUnits" ELSE 0 END AS unknown_units,
|
|
CASE WHEN receipt."deliveredAt" >= COALESCE(submit."submittedAt", submit."createdAt")
|
|
THEN EXTRACT(EPOCH FROM (receipt."deliveredAt" - COALESCE(submit."submittedAt", submit."createdAt"))) * 1000 END AS arrival_ms
|
|
FROM "SmsSubmitRecord" submit
|
|
JOIN "SmsMessageRecord" message ON message.id = submit."messageRecordId"
|
|
JOIN "SmsChannel" channel ON channel.id = submit."channelId"
|
|
LEFT JOIN LATERAL (
|
|
SELECT MIN(receipt."deliveredAt") AS "deliveredAt"
|
|
FROM "SmsReceiptRecord" receipt
|
|
WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
|
|
AND receipt."channelId" = submit."channelId"
|
|
AND receipt."receiptStatus" = 'delivered'
|
|
) receipt ON TRUE
|
|
LEFT JOIN LATERAL (
|
|
SELECT MIN(receipt."deliveredAt") AS "failedAt"
|
|
FROM "SmsReceiptRecord" receipt
|
|
WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
|
|
AND receipt."channelId" = submit."channelId"
|
|
AND receipt."receiptStatus" = 'undelivered'
|
|
) failed_receipt ON TRUE
|
|
WHERE submit."submitStatus" = 'accepted'
|
|
AND COALESCE(submit."submittedAt", submit."createdAt") >= ${day.startAt}
|
|
AND COALESCE(submit."submittedAt", submit."createdAt") < ${day.endAt}
|
|
), thresholds AS (
|
|
SELECT dimension_id, PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY arrival_ms) AS p95_ms
|
|
FROM base WHERE arrival_ms IS NOT NULL GROUP BY dimension_id
|
|
)
|
|
INSERT INTO "DailyQualityReport" (
|
|
"id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
|
|
"tenantId", "tenantName", "applicationId", "channelId", "signatureId", "drainageInfoId",
|
|
"submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "successRateBps", "avgArrivalMs", "generatedAt", "updatedAt"
|
|
)
|
|
SELECT
|
|
CONCAT('quality-channel-', MD5(${day.key} || ':' || base.dimension_id)),
|
|
${day.reportDate}::date,
|
|
'channel',
|
|
base.dimension_id,
|
|
MAX(base.dimension_name),
|
|
NULL, NULL, NULL, base.dimension_id, NULL, NULL,
|
|
SUM(base.billing_units)::integer,
|
|
SUM(base.billing_units)::integer,
|
|
SUM(base.unknown_units)::integer,
|
|
SUM(base.success_units)::integer,
|
|
SUM(base.failed_units)::integer,
|
|
CASE WHEN SUM(base.billing_units) = 0 THEN 0 ELSE ROUND(SUM(base.success_units) * 10000.0 / SUM(base.billing_units))::integer END,
|
|
ROUND(AVG(base.arrival_ms) FILTER (WHERE base.arrival_ms IS NOT NULL AND base.arrival_ms <= thresholds.p95_ms))::integer,
|
|
CURRENT_TIMESTAMP,
|
|
CURRENT_TIMESTAMP
|
|
FROM base
|
|
LEFT JOIN thresholds ON thresholds.dimension_id = base.dimension_id
|
|
GROUP BY base.dimension_id
|
|
`;
|
|
}
|
|
|
|
type BusinessDay = { key: string; reportDate: Date; startAt: Date; endAt: Date };
|
|
|
|
function completedBusinessDays(now: Date, count: number): BusinessDay[] {
|
|
const shifted = new Date(now.getTime() + SHANGHAI_OFFSET_MS);
|
|
const today = Date.UTC(shifted.getUTCFullYear(), shifted.getUTCMonth(), shifted.getUTCDate());
|
|
return Array.from({ length: count }, (_, index) => businessDay(today - (count - index) * DAY_MS));
|
|
}
|
|
|
|
function businessDay(localDateUtc: number): BusinessDay {
|
|
const reportDate = new Date(localDateUtc);
|
|
return {
|
|
key: reportDate.toISOString().slice(0, 10),
|
|
reportDate,
|
|
startAt: new Date(localDateUtc - SHANGHAI_OFFSET_MS),
|
|
endAt: new Date(localDateUtc - SHANGHAI_OFFSET_MS + DAY_MS),
|
|
};
|
|
}
|
|
|
|
function shanghaiDateKey(now: Date) {
|
|
return new Date(now.getTime() + SHANGHAI_OFFSET_MS).toISOString().slice(0, 10);
|
|
}
|
|
|
|
function dateFilter(from?: string, to?: string): Prisma.DateTimeFilter | undefined {
|
|
const gte = parseDate(from);
|
|
const lte = parseDate(to);
|
|
if (!gte && !lte) return undefined;
|
|
return { gte, lte };
|
|
}
|
|
|
|
function parseDate(value?: string) {
|
|
if (!value || !/^\d{4}-\d{2}-\d{2}$/.test(value)) return undefined;
|
|
const date = new Date(`${value}T00:00:00.000Z`);
|
|
return Number.isNaN(date.getTime()) ? undefined : date;
|
|
}
|
|
|
|
function pagination(query: ReportListQuery) {
|
|
const page = Math.max(1, Math.floor(Number(query.page) || 1));
|
|
const pageSize = Math.min(100, Math.max(1, Math.floor(Number(query.pageSize) || 20)));
|
|
return { page, pageSize, skip: (page - 1) * pageSize };
|
|
}
|
|
|
|
const reportVolumeSumSelection = {
|
|
submittedUnits: true,
|
|
sentUnits: true,
|
|
unknownUnits: true,
|
|
successUnits: true,
|
|
failedUnits: true,
|
|
} as const;
|
|
|
|
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),
|
|
unknownUnits: Number(sum.unknownUnits ?? 0),
|
|
successUnits: Number(sum.successUnits ?? 0),
|
|
failedUnits: Number(sum.failedUnits ?? 0),
|
|
};
|
|
}
|
|
|
|
function ratioBps(numerator: number, denominator: number) {
|
|
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,
|
|
};
|
|
}
|
|
|
|
function profitWhere(query: ReportListQuery) {
|
|
const dimensionType = query.dimensionType === 'channel' ? 'channel' : 'application';
|
|
const where: Prisma.DailyProfitReportWhereInput = {
|
|
dimensionType,
|
|
reportDate: dateFilter(query.dateFrom, query.dateTo),
|
|
tenantId: dimensionType === 'application' ? query.tenantId || undefined : undefined,
|
|
applicationId: dimensionType === 'application' ? query.applicationId || undefined : undefined,
|
|
channelId: dimensionType === 'channel' ? query.channelId || undefined : undefined,
|
|
};
|
|
return { dimensionType, where };
|
|
}
|
|
|
|
function qualityWhere(query: ReportListQuery) {
|
|
const allowedDimensions = new Set(['application', 'channel', 'signature', 'drainage']);
|
|
const dimensionType = allowedDimensions.has(String(query.dimensionType))
|
|
? String(query.dimensionType)
|
|
: 'application';
|
|
const where: Prisma.DailyQualityReportWhereInput = {
|
|
dimensionType,
|
|
reportDate: dateFilter(query.dateFrom, query.dateTo),
|
|
tenantId: query.tenantId || undefined,
|
|
applicationId: query.applicationId || undefined,
|
|
channelId: query.channelId || undefined,
|
|
};
|
|
return { dimensionType, where };
|
|
}
|
|
|
|
function csvExport(name: string, headers: string[], rows: Array<Array<string | number>>) {
|
|
const content = [headers, ...rows].map((row) => row.map(csvCell).join(',')).join('\n');
|
|
return { fileName: `${name}-${shanghaiDateKey(new Date())}.csv`, content };
|
|
}
|
|
|
|
function csvCell(value: string | number) {
|
|
const text = String(value);
|
|
return /[",\r\n]/.test(text) ? `"${text.replace(/"/g, '""')}"` : text;
|
|
}
|
|
|
|
function dateKey(value: Date | string) {
|
|
return (value instanceof Date ? value.toISOString() : String(value)).slice(0, 10);
|
|
}
|
|
|
|
function formatCsvDate(value: Date | string) {
|
|
return value instanceof Date ? value.toISOString() : String(value);
|
|
}
|
|
|
|
function positiveInteger(value: string | undefined, fallback: number) {
|
|
const parsed = Number(value);
|
|
return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
|
|
}
|