475 lines
19 KiB
TypeScript
475 lines
19 KiB
TypeScript
import { Prisma } from '@prisma/client';
|
|
import { moneyToNumber } from '../../common/money';
|
|
import { PrismaService } from '../../prisma/prisma.service';
|
|
|
|
import {
|
|
messageWhere,
|
|
qualityBusinessDay,
|
|
returnedTransactionWhere,
|
|
downstreamAlertWindows,
|
|
stalledPendingWhere,
|
|
clientBatchTaskView,
|
|
clientAccountView,
|
|
clientRechargeView,
|
|
summarizeMessageGroups,
|
|
} from '../operations.helpers';
|
|
|
|
// R2 dashboard query domain. Method bodies are preserved byte-for-byte from the facade baseline.
|
|
export class OperationsDashboardQueries {
|
|
constructor(private readonly prisma: PrismaService) {}
|
|
|
|
async dashboard(query: { tenantId?: string }) {
|
|
const businessDay = qualityBusinessDay();
|
|
const sinceToday = businessDay.startAt;
|
|
const downstreamAlertWindow = downstreamAlertWindows();
|
|
const messageWhereClause = messageWhere({ tenantId: query.tenantId });
|
|
const todayMessageWhereClause = {
|
|
...messageWhereClause,
|
|
queuedAt: { gte: sinceToday, lt: businessDay.endAt },
|
|
};
|
|
const [
|
|
taskCount,
|
|
messageGroups,
|
|
todayMessageGroups,
|
|
uplinkCount,
|
|
billingAggregate,
|
|
transactionAggregate,
|
|
connectionGroups,
|
|
pendingAudits,
|
|
tenantAccounts,
|
|
recentTasks,
|
|
recentRecharges,
|
|
enterpriseSpendRows,
|
|
todayBusinessMetricsRows,
|
|
downstreamPendingCount,
|
|
downstreamFailedCount,
|
|
downstreamDeliveredCount,
|
|
downstreamStalledPendingCount,
|
|
downstreamStalledAckCount,
|
|
downstreamRecentFailedCount,
|
|
hourlySendRows,
|
|
auditSpeedRows,
|
|
] = await Promise.all([
|
|
this.prisma.smsBatchTask.count({ where: { tenantId: query.tenantId } }),
|
|
this.prisma.smsMessageRecord.groupBy({
|
|
by: ['status'],
|
|
where: messageWhereClause,
|
|
_count: { _all: true },
|
|
_sum: { amountCents: true, billingUnits: true },
|
|
}),
|
|
this.prisma.smsMessageRecord.groupBy({
|
|
by: ['status'],
|
|
where: todayMessageWhereClause,
|
|
_count: { _all: true },
|
|
_sum: { amountCents: true, billingUnits: true },
|
|
}),
|
|
this.prisma.smsUplinkMessage.count({ where: { tenantId: query.tenantId } }),
|
|
this.prisma.smsBillingRecord.aggregate({
|
|
where: { tenantId: query.tenantId },
|
|
_sum: { amountCents: true, billingUnits: true },
|
|
_count: { _all: true },
|
|
}),
|
|
this.prisma.accountTransaction.aggregate({
|
|
where: returnedTransactionWhere(sinceToday, query.tenantId),
|
|
_sum: { amountCents: true },
|
|
_count: { _all: true },
|
|
}),
|
|
this.prisma.cmppConnectionState.groupBy({
|
|
by: ['status'],
|
|
where: { tenantId: query.tenantId },
|
|
_count: { _all: true },
|
|
_sum: { currentConnections: true, desiredConnections: true },
|
|
}),
|
|
this.pendingAudits(query.tenantId),
|
|
this.prisma.tenantAccount.findMany({
|
|
where: query.tenantId ? { tenantId: query.tenantId } : undefined,
|
|
include: { tenant: true },
|
|
orderBy: { updatedAt: 'desc' },
|
|
take: 20,
|
|
}),
|
|
this.prisma.smsBatchTask.findMany({
|
|
where: query.tenantId ? { tenantId: query.tenantId } : undefined,
|
|
include: { application: true, messages: { take: 1, include: { channel: true } } },
|
|
orderBy: { createdAt: 'desc' },
|
|
take: 10,
|
|
}),
|
|
this.prisma.rechargeOrder.findMany({
|
|
where: {
|
|
tenantId: query.tenantId,
|
|
payMethod: 'manual_topup',
|
|
},
|
|
include: { tenant: true },
|
|
orderBy: { createdAt: 'desc' },
|
|
take: 10,
|
|
}),
|
|
this.prisma.$queryRaw<
|
|
Array<{
|
|
tenantId: string;
|
|
tenantName: string;
|
|
todaySpendCents: bigint;
|
|
balanceCents: bigint;
|
|
creditCents: bigint;
|
|
}>
|
|
>(Prisma.sql`
|
|
SELECT
|
|
tenant.id AS "tenantId",
|
|
tenant.name AS "tenantName",
|
|
COALESCE(SUM(billing."amountCents") FILTER (WHERE billing."billingStatus" = 'charged'), 0)::bigint AS "todaySpendCents",
|
|
account."balanceCents" AS "balanceCents",
|
|
account."creditCents" AS "creditCents"
|
|
FROM "TenantAccount" account
|
|
JOIN "Tenant" tenant ON tenant.id = account."tenantId"
|
|
LEFT JOIN "SmsBillingRecord" billing
|
|
ON billing."tenantId" = tenant.id
|
|
AND billing."createdAt" >= ${businessDay.startAt}
|
|
AND billing."createdAt" < ${businessDay.endAt}
|
|
WHERE tenant.status <> 'deleted'
|
|
AND (${query.tenantId ?? null}::text IS NULL OR tenant.id = ${query.tenantId ?? null})
|
|
GROUP BY tenant.id, tenant.name, account."balanceCents", account."creditCents"
|
|
ORDER BY "todaySpendCents" DESC, tenant.name ASC
|
|
`),
|
|
this.prisma.$queryRaw<
|
|
Array<{
|
|
segmentCount: bigint;
|
|
deliveredSegmentCount: bigint;
|
|
billedCents: bigint;
|
|
costCents: bigint;
|
|
}>
|
|
>(Prisma.sql`
|
|
WITH segment_metrics AS (
|
|
SELECT
|
|
COUNT(segment.id)::bigint AS "segmentCount",
|
|
COUNT(segment.id) FILTER (WHERE segment."receiptStatus" = 'delivered')::bigint AS "deliveredSegmentCount"
|
|
FROM "SmsMessageSegmentAudit" segment
|
|
JOIN "SmsMessageRecord" message ON message.id = segment."messageRecordId"
|
|
WHERE message."queuedAt" >= ${businessDay.startAt}
|
|
AND message."queuedAt" < ${businessDay.endAt}
|
|
AND (${query.tenantId ?? null}::text IS NULL OR message."tenantId" = ${query.tenantId ?? null})
|
|
),
|
|
message_revenue AS (
|
|
SELECT COALESCE(SUM(
|
|
CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered'
|
|
THEN message."billingUnits" * message."unitPrice" ELSE 0 END
|
|
), 0)::bigint AS "billedCents"
|
|
FROM "SmsMessageRecord" message
|
|
WHERE message."queuedAt" >= ${businessDay.startAt}
|
|
AND message."queuedAt" < ${businessDay.endAt}
|
|
AND (${query.tenantId ?? null}::text IS NULL OR message."tenantId" = ${query.tenantId ?? null})
|
|
),
|
|
submit_cost AS (
|
|
SELECT 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 AS "costCents"
|
|
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" >= ${businessDay.startAt}
|
|
AND message."queuedAt" < ${businessDay.endAt}
|
|
AND (${query.tenantId ?? null}::text IS NULL OR message."tenantId" = ${query.tenantId ?? null})
|
|
)
|
|
SELECT
|
|
segment_metrics."segmentCount",
|
|
segment_metrics."deliveredSegmentCount",
|
|
message_revenue."billedCents",
|
|
submit_cost."costCents"
|
|
FROM segment_metrics, message_revenue, submit_cost
|
|
`),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: { tenantId: query.tenantId, status: 'pending' },
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: { tenantId: query.tenantId, status: 'failed' },
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: { tenantId: query.tenantId, status: 'delivered' },
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: {
|
|
tenantId: query.tenantId,
|
|
...stalledPendingWhere(downstreamAlertWindow.stalledPendingAt),
|
|
},
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: {
|
|
tenantId: query.tenantId,
|
|
status: 'awaiting_ack',
|
|
ackDeadlineAt: { lte: downstreamAlertWindow.now },
|
|
},
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: {
|
|
tenantId: query.tenantId,
|
|
status: { in: ['failed', 'unconfirmed', 'rejected'] },
|
|
updatedAt: { gte: downstreamAlertWindow.recentFailedAt },
|
|
},
|
|
}),
|
|
this.prisma.$queryRaw<
|
|
Array<{
|
|
hour: number;
|
|
submittedCount: bigint;
|
|
successCount: bigint;
|
|
}>
|
|
>(Prisma.sql`
|
|
SELECT
|
|
EXTRACT(
|
|
HOUR FROM (message."queuedAt" AT TIME ZONE 'UTC') AT TIME ZONE 'Asia/Shanghai'
|
|
)::integer AS hour,
|
|
COUNT(*)::bigint AS "submittedCount",
|
|
COUNT(*) FILTER (WHERE message.status = 'delivered')::bigint AS "successCount"
|
|
FROM "SmsMessageRecord" message
|
|
WHERE message."queuedAt" >= ${businessDay.startAt}
|
|
AND message."queuedAt" < ${businessDay.endAt}
|
|
AND (${query.tenantId ?? null}::text IS NULL OR message."tenantId" = ${query.tenantId ?? null})
|
|
GROUP BY 1
|
|
ORDER BY 1
|
|
`),
|
|
// Signature/template tables have no review timestamps, so their latest pending audit is paired with the review audit.
|
|
this.prisma.$queryRaw<
|
|
Array<{
|
|
category: string;
|
|
count: bigint;
|
|
averageProcessingMs: bigint | null;
|
|
}>
|
|
>(Prisma.sql`
|
|
WITH review_samples AS (
|
|
SELECT
|
|
'enterpriseCertifications'::text AS category,
|
|
certification."submittedAt" AS "submittedAt",
|
|
certification."reviewedAt" AS "reviewedAt"
|
|
FROM "EnterpriseCertification" certification
|
|
WHERE certification."reviewedAt" >= ${businessDay.startAt}
|
|
AND certification."reviewedAt" < ${businessDay.endAt}
|
|
AND (${query.tenantId ?? null}::text IS NULL OR certification."tenantId" = ${query.tenantId ?? null})
|
|
|
|
UNION ALL
|
|
|
|
SELECT
|
|
'smsAudits'::text,
|
|
task."createdAt",
|
|
task."reviewedAt"
|
|
FROM "SmsSendTask" task
|
|
WHERE task."reviewedAt" >= ${businessDay.startAt}
|
|
AND task."reviewedAt" < ${businessDay.endAt}
|
|
AND (${query.tenantId ?? null}::text IS NULL OR task."tenantId" = ${query.tenantId ?? null})
|
|
|
|
UNION ALL
|
|
|
|
SELECT
|
|
'drainageInfos'::text,
|
|
drainage."submittedAt",
|
|
drainage."reviewedAt"
|
|
FROM "SmsDrainageInfo" drainage
|
|
WHERE drainage."reviewedAt" >= ${businessDay.startAt}
|
|
AND drainage."reviewedAt" < ${businessDay.endAt}
|
|
AND (${query.tenantId ?? null}::text IS NULL OR drainage."tenantId" = ${query.tenantId ?? null})
|
|
|
|
UNION ALL
|
|
|
|
SELECT
|
|
CASE review."targetType"
|
|
WHEN 'sms_signature' THEN 'signatures'
|
|
WHEN 'sms_template' THEN 'templates'
|
|
END,
|
|
submission."createdAt",
|
|
review."createdAt"
|
|
FROM "AuditRecord" review
|
|
JOIN LATERAL (
|
|
SELECT pending."createdAt"
|
|
FROM "AuditRecord" pending
|
|
WHERE pending."targetType" = review."targetType"
|
|
AND pending."targetId" = review."targetId"
|
|
AND pending."statusAfter" = 'pending'
|
|
AND pending."createdAt" <= review."createdAt"
|
|
ORDER BY pending."createdAt" DESC
|
|
LIMIT 1
|
|
) submission ON true
|
|
WHERE review."targetType" IN ('sms_signature', 'sms_template')
|
|
AND review."statusBefore" = 'pending'
|
|
AND review."statusAfter" IN ('approved', 'rejected')
|
|
AND review."createdAt" >= ${businessDay.startAt}
|
|
AND review."createdAt" < ${businessDay.endAt}
|
|
AND (${query.tenantId ?? null}::text IS NULL OR review."tenantId" = ${query.tenantId ?? null})
|
|
)
|
|
SELECT
|
|
category,
|
|
COUNT(*)::bigint AS count,
|
|
ROUND(AVG(EXTRACT(EPOCH FROM ("reviewedAt" - "submittedAt")) * 1000))::bigint AS "averageProcessingMs"
|
|
FROM review_samples
|
|
WHERE "reviewedAt" >= "submittedAt"
|
|
GROUP BY category
|
|
`),
|
|
]);
|
|
const todayTotals = summarizeMessageGroups(todayMessageGroups);
|
|
const todayBusinessMetrics = todayBusinessMetricsRows[0];
|
|
const supplierSegmentCount = Number(todayBusinessMetrics?.segmentCount ?? 0);
|
|
const segmentCount = todayTotals.billingUnits;
|
|
const deliveredSegmentCount = Number(todayBusinessMetrics?.deliveredSegmentCount ?? 0);
|
|
const billedCents = moneyToNumber(todayBusinessMetrics?.billedCents);
|
|
const costCents = moneyToNumber(todayBusinessMetrics?.costCents);
|
|
const profitCents = billedCents - costCents;
|
|
const hourlyRowsByHour = new Map(hourlySendRows.map((row) => [Number(row.hour), row]));
|
|
// Always return all 24 Shanghai-time buckets so the line chart does not imply missing hours are missing data.
|
|
const hourlySendTrend = Array.from({ length: 24 }, (_, hour) => {
|
|
const row = hourlyRowsByHour.get(hour);
|
|
return {
|
|
hour,
|
|
label: `${String(hour).padStart(2, '0')}:00`,
|
|
submittedCount: Number(row?.submittedCount ?? 0),
|
|
successCount: Number(row?.successCount ?? 0),
|
|
};
|
|
});
|
|
const auditSpeedByCategory = new Map(auditSpeedRows.map((row) => [row.category, row]));
|
|
const auditProcessingSpeed = [
|
|
['enterpriseCertifications', '企业认证'],
|
|
['smsAudits', '短信审核'],
|
|
['templates', '模板'],
|
|
['signatures', '签名'],
|
|
['drainageInfos', '引流信息'],
|
|
].map(([category, label]) => {
|
|
const row = auditSpeedByCategory.get(category);
|
|
return {
|
|
category,
|
|
label,
|
|
count: Number(row?.count ?? 0),
|
|
averageProcessingMs: row?.averageProcessingMs == null ? null : Number(row.averageProcessingMs),
|
|
};
|
|
});
|
|
const downstreamAlertCount =
|
|
downstreamStalledPendingCount + downstreamStalledAckCount + downstreamRecentFailedCount;
|
|
return {
|
|
taskCount,
|
|
messageStatus: messageGroups,
|
|
today: {
|
|
sent: todayTotals.total,
|
|
delivered: todayTotals.delivered,
|
|
failed: todayTotals.failed,
|
|
unknown: todayTotals.unknown,
|
|
successRate: todayTotals.total > 0 ? Number(((todayTotals.delivered / todayTotals.total) * 100).toFixed(1)) : 0,
|
|
spendCents: todayTotals.amountCents,
|
|
returnedCents: moneyToNumber(transactionAggregate._sum.amountCents),
|
|
billingUnits: todayTotals.billingUnits,
|
|
segmentCount,
|
|
deliveredSegmentCount,
|
|
arrivalRate:
|
|
supplierSegmentCount > 0 ? Number(((deliveredSegmentCount / supplierSegmentCount) * 100).toFixed(1)) : 0,
|
|
billedCents,
|
|
profitCents,
|
|
profitRate: billedCents > 0 ? Number(((profitCents / billedCents) * 100).toFixed(1)) : 0,
|
|
},
|
|
uplinkCount,
|
|
billing: billingAggregate,
|
|
transactions: transactionAggregate,
|
|
gatewayConnections: connectionGroups,
|
|
pendingAuditCount: pendingAudits.total,
|
|
pendingAudits,
|
|
hourlySendTrend,
|
|
auditProcessingSpeed,
|
|
downstreamDeliverySummary: {
|
|
pending: downstreamPendingCount,
|
|
failed: downstreamFailedCount,
|
|
delivered: downstreamDeliveredCount,
|
|
stalledPending: downstreamStalledPendingCount,
|
|
stalledAck: downstreamStalledAckCount,
|
|
recentFailed: downstreamRecentFailedCount,
|
|
alertCount: downstreamAlertCount,
|
|
},
|
|
accounts: tenantAccounts,
|
|
enterpriseSpendRanks: enterpriseSpendRows.map((row) => ({
|
|
tenantId: row.tenantId,
|
|
tenantName: row.tenantName,
|
|
todaySpendCents: moneyToNumber(row.todaySpendCents),
|
|
balanceCents: moneyToNumber(row.balanceCents),
|
|
creditCents: moneyToNumber(row.creditCents),
|
|
})),
|
|
recentTasks,
|
|
recentRecharges,
|
|
};
|
|
}
|
|
async clientDashboard(query: { tenantId?: string }) {
|
|
const tenantId = query.tenantId;
|
|
const [dashboard, tenant, approvedCertification, signatureCount, pendingBatchTaskCount] = await Promise.all([
|
|
this.dashboard(query),
|
|
tenantId
|
|
? this.prisma.tenant.findUnique({ where: { id: tenantId }, select: { id: true, name: true } })
|
|
: Promise.resolve(null),
|
|
tenantId
|
|
? this.prisma.enterpriseCertification.findFirst({
|
|
where: { tenantId, status: 'approved' },
|
|
select: { id: true },
|
|
})
|
|
: Promise.resolve(null),
|
|
tenantId
|
|
? this.prisma.smsSignature.count({
|
|
where: { tenantId, auditStatus: { notIn: ['deleted', 'disabled'] } },
|
|
})
|
|
: Promise.resolve(0),
|
|
tenantId
|
|
? this.prisma.smsBatchTask.count({
|
|
where: { tenantId, sourceType: 'client', status: 'pending_review' },
|
|
})
|
|
: Promise.resolve(0),
|
|
]);
|
|
return {
|
|
taskCount: dashboard.taskCount,
|
|
messageStatus: dashboard.messageStatus,
|
|
today: dashboard.today,
|
|
uplinkCount: dashboard.uplinkCount,
|
|
billing: dashboard.billing,
|
|
transactions: dashboard.transactions,
|
|
gatewayConnections: [],
|
|
pendingAuditCount: dashboard.pendingAuditCount,
|
|
pendingAudits: dashboard.pendingAudits,
|
|
hourlySendTrend: dashboard.hourlySendTrend,
|
|
auditProcessingSpeed: dashboard.auditProcessingSpeed,
|
|
downstreamDeliverySummary: dashboard.downstreamDeliverySummary,
|
|
accounts: dashboard.accounts.map(clientAccountView),
|
|
enterpriseSpendRanks: dashboard.enterpriseSpendRanks,
|
|
recentTasks: dashboard.recentTasks.map(clientBatchTaskView),
|
|
recentRecharges: dashboard.recentRecharges.map(clientRechargeView),
|
|
clientOverview: {
|
|
enterpriseName: tenant?.name ?? null,
|
|
certificationStatus: approvedCertification ? 'certified' : 'uncertified',
|
|
signatureCount,
|
|
pendingBatchTaskCount,
|
|
},
|
|
};
|
|
}
|
|
pendingAudits(tenantId?: string) {
|
|
return Promise.all([
|
|
this.prisma.smsTemplate.count({ where: { tenantId, auditStatus: 'pending' } }),
|
|
this.prisma.smsSignature.count({ where: { tenantId, auditStatus: 'pending' } }),
|
|
this.prisma.smsDrainageInfo.count({ where: { tenantId, auditStatus: 'pending' } }),
|
|
this.prisma.enterpriseCertification.count({ where: { tenantId, status: 'pending' } }),
|
|
this.prisma.smsSendTask.count({ where: { tenantId, status: 'pending_review' } }),
|
|
this.prisma.reportMaterialImportItem.count({
|
|
where: { reportType: 'signature', status: 'pending_review', batch: { tenantId } },
|
|
}),
|
|
]).then(([templates, signatures, drainageInfos, enterpriseCertifications, smsAudits, signatureImports]) => ({
|
|
templates,
|
|
signatures,
|
|
drainageInfos,
|
|
enterpriseCertifications,
|
|
smsAudits,
|
|
signatureImports,
|
|
total: templates + signatures + drainageInfos + enterpriseCertifications + smsAudits + signatureImports,
|
|
}));
|
|
}
|
|
}
|