433 lines
18 KiB
TypeScript
433 lines
18 KiB
TypeScript
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, ReceiptAnomalyQuery, DownstreamDeliveryQuery, DownstreamDeliveryDashboardQuery, DownstreamRecoveryStatusQuery, MessageSegmentAuditQuery, SignatureQualityQuery } from '../operations.contracts';
|
|
import { 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';
|
|
|
|
// R2 downstream query domain. Method bodies are preserved byte-for-byte from the facade baseline.
|
|
export class OperationsDownstreamQueries {
|
|
constructor(private readonly prisma: PrismaService) {}
|
|
|
|
async listGatewaySubmitDeadLetters(query: GatewaySubmitDeadLetterQuery) {
|
|
const page = Math.max(1, Number(query.page ?? 1));
|
|
const pageSize = Math.min(100, Math.max(1, Number(query.pageSize ?? 10)));
|
|
const baseWhere: Prisma.GatewaySubmitDeadLetterWhereInput = {
|
|
tenantId: query.tenantId,
|
|
applicationId: query.applicationId,
|
|
channelId: query.channelId,
|
|
OR: query.keyword ? [
|
|
{ streamMessageId: { contains: query.keyword } },
|
|
{ traceId: { contains: query.keyword } },
|
|
{ messageId: { contains: query.keyword } },
|
|
{ submitId: { contains: query.keyword } },
|
|
{ failureCode: { contains: query.keyword } },
|
|
{ failureMessage: { contains: query.keyword } },
|
|
] : undefined,
|
|
};
|
|
const where: Prisma.GatewaySubmitDeadLetterWhereInput = {
|
|
...baseWhere,
|
|
status: query.status && query.status !== 'all' ? query.status : undefined,
|
|
};
|
|
const [items, total, statusGroups, oldestPending] = await Promise.all([
|
|
this.prisma.gatewaySubmitDeadLetter.findMany({
|
|
where,
|
|
include: { tenant: true, application: true, channel: true },
|
|
orderBy: { createdAt: 'desc' },
|
|
skip: (page - 1) * pageSize,
|
|
take: pageSize,
|
|
}),
|
|
this.prisma.gatewaySubmitDeadLetter.count({ where }),
|
|
this.prisma.gatewaySubmitDeadLetter.groupBy({
|
|
by: ['status'],
|
|
where: baseWhere,
|
|
_count: { _all: true },
|
|
}),
|
|
this.prisma.gatewaySubmitDeadLetter.findFirst({
|
|
where: { ...baseWhere, status: 'pending' },
|
|
orderBy: { createdAt: 'asc' },
|
|
select: { createdAt: true },
|
|
}),
|
|
]);
|
|
const statusCounts = new Map(statusGroups.map((item) => [item.status, item._count._all]));
|
|
const messageIds = items.map((item) => item.messageId).filter((value): value is string => Boolean(value));
|
|
const messageStates = messageIds.length > 0
|
|
? await this.prisma.smsMessageRecord.findMany({
|
|
where: { messageId: { in: messageIds } },
|
|
select: { messageId: true, status: true, submitStatus: true, receiptStatus: true, phoneNumber: true, content: true },
|
|
})
|
|
: [];
|
|
const messageStateById = new Map(messageStates.map((item) => [item.messageId, item]));
|
|
return {
|
|
items: items.map((item) => sanitizeGatewaySubmitException(item, item.messageId ? messageStateById.get(item.messageId) : undefined)),
|
|
total,
|
|
page,
|
|
pageSize,
|
|
summary: {
|
|
pending: statusCounts.get('pending') ?? 0,
|
|
requeueing: statusCounts.get('requeueing') ?? 0,
|
|
requeued: statusCounts.get('requeued') ?? 0,
|
|
resolved: statusCounts.get('resolved') ?? 0,
|
|
oldestPendingAt: oldestPending?.createdAt ?? null,
|
|
},
|
|
};
|
|
}
|
|
async listReceiptAnomalies(query: ReceiptAnomalyQuery) {
|
|
const page = Math.max(1, Number(query.page ?? 1));
|
|
const pageSize = Math.min(100, Math.max(1, Number(query.pageSize ?? 10)));
|
|
const baseWhere: Prisma.SmsReceiptAnomalyWhereInput = {
|
|
tenantId: query.tenantId,
|
|
applicationId: query.applicationId,
|
|
channelId: query.channelId,
|
|
anomalyType: query.anomalyType && query.anomalyType !== 'all' ? query.anomalyType : undefined,
|
|
OR: query.keyword ? [
|
|
{ anomalyKey: { contains: query.keyword } },
|
|
{ rawStatus: { contains: query.keyword } },
|
|
{ errorCode: { contains: query.keyword } },
|
|
{ messageRecord: { messageId: { contains: query.keyword } } },
|
|
{ submitRecord: { submitId: { contains: query.keyword } } },
|
|
] : undefined,
|
|
};
|
|
const where: Prisma.SmsReceiptAnomalyWhereInput = {
|
|
...baseWhere,
|
|
status: query.status && query.status !== 'all' ? query.status : undefined,
|
|
};
|
|
const [items, total, statusGroups, oldestPending] = await Promise.all([
|
|
this.prisma.smsReceiptAnomaly.findMany({
|
|
where,
|
|
include: {
|
|
tenant: { select: { id: true, name: true } },
|
|
application: { select: { id: true, name: true } },
|
|
channel: { select: { id: true, code: true, name: true, status: true } },
|
|
messageRecord: { select: { messageId: true, phoneNumber: true, status: true } },
|
|
submitRecord: { select: { submitId: true, submitStatus: true } },
|
|
receiptRecord: { select: { gatewayMessageId: true, receiptStatus: true, rawStatus: true, deliveredAt: true } },
|
|
},
|
|
orderBy: [{ lastOccurredAt: 'desc' }, { id: 'desc' }],
|
|
skip: (page - 1) * pageSize,
|
|
take: pageSize,
|
|
}),
|
|
this.prisma.smsReceiptAnomaly.count({ where }),
|
|
this.prisma.smsReceiptAnomaly.groupBy({
|
|
by: ['status'],
|
|
where: baseWhere,
|
|
_count: { _all: true },
|
|
}),
|
|
this.prisma.smsReceiptAnomaly.findFirst({
|
|
where: { ...baseWhere, status: 'pending' },
|
|
orderBy: { firstOccurredAt: 'asc' },
|
|
select: { firstOccurredAt: true },
|
|
}),
|
|
]);
|
|
const statusCounts = new Map(statusGroups.map((item) => [item.status, item._count._all]));
|
|
return {
|
|
items,
|
|
total,
|
|
page,
|
|
pageSize,
|
|
summary: {
|
|
pending: statusCounts.get('pending') ?? 0,
|
|
resolved: statusCounts.get('resolved') ?? 0,
|
|
ignored: statusCounts.get('ignored') ?? 0,
|
|
oldestPendingAt: oldestPending?.firstOccurredAt ?? null,
|
|
},
|
|
};
|
|
}
|
|
async listDownstreamDeliveries(query: DownstreamDeliveryQuery) {
|
|
const page = Math.max(1, Number(query.page ?? 1));
|
|
const pageSize = Math.min(100, Math.max(1, Number(query.pageSize ?? 10)));
|
|
const where: Prisma.CmppDownstreamDeliveryWhereInput = {
|
|
...downstreamDeliveryScopedWhere(query),
|
|
status: query.status && query.status !== 'all' ? query.status : undefined,
|
|
OR: query.keyword ? [
|
|
{ messageId: { contains: query.keyword } },
|
|
{ payload: { path: ['account'], string_contains: query.keyword } },
|
|
{ payload: { path: ['phoneNumber'], string_contains: query.keyword } },
|
|
{ lastError: { contains: query.keyword } },
|
|
] : undefined,
|
|
};
|
|
const [items, total] = await Promise.all([
|
|
this.prisma.cmppDownstreamDelivery.findMany({
|
|
where,
|
|
include: {
|
|
tenant: true,
|
|
application: true,
|
|
messageRecord: true,
|
|
attempts: { orderBy: [{ attemptNo: 'desc' }, { createdAt: 'desc' }] },
|
|
},
|
|
orderBy: { createdAt: 'desc' },
|
|
skip: (page - 1) * pageSize,
|
|
take: pageSize,
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({ where }),
|
|
]);
|
|
return { items, total, page, pageSize };
|
|
}
|
|
async downstreamDeliveryDashboard(query: DownstreamDeliveryDashboardQuery) {
|
|
const scopedWhere = downstreamDeliveryScopedWhere(query);
|
|
const downstreamAlertWindow = downstreamAlertWindows();
|
|
const [total, pending, awaitingAck, delivered, failed, unconfirmed, rejected, stalledPending, stalledAck, recentFailed, typeGroups, applicationGroups, applicationAlertGroups, retryZero, retryLow, retryHigh] = await Promise.all([
|
|
this.prisma.cmppDownstreamDelivery.count({ where: scopedWhere }),
|
|
this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'pending' } }),
|
|
this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'awaiting_ack' } }),
|
|
this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'delivered' } }),
|
|
this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'failed' } }),
|
|
this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'unconfirmed' } }),
|
|
this.prisma.cmppDownstreamDelivery.count({ where: { ...scopedWhere, status: 'rejected' } }),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: {
|
|
...scopedWhere,
|
|
...stalledPendingWhere(downstreamAlertWindow.stalledPendingAt),
|
|
},
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: { ...scopedWhere, status: 'awaiting_ack', ackDeadlineAt: { lte: downstreamAlertWindow.now } },
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: {
|
|
...scopedWhere,
|
|
status: { in: ['failed', 'unconfirmed', 'rejected'] },
|
|
updatedAt: { gte: downstreamAlertWindow.recentFailedAt },
|
|
},
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.groupBy({
|
|
by: ['deliveryType', 'status'],
|
|
where: scopedWhere,
|
|
_count: { _all: true },
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.groupBy({
|
|
by: ['applicationId', 'status'],
|
|
where: scopedWhere,
|
|
_count: { _all: true },
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.groupBy({
|
|
by: ['applicationId'],
|
|
where: downstreamAlertWhere(scopedWhere, downstreamAlertWindow),
|
|
_count: { _all: true },
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: {
|
|
...scopedWhere,
|
|
status: { in: ['pending', 'failed', 'unconfirmed', 'rejected'] },
|
|
retryCount: 0,
|
|
},
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: {
|
|
...scopedWhere,
|
|
status: { in: ['pending', 'failed', 'unconfirmed', 'rejected'] },
|
|
retryCount: { gte: 1, lte: 3 },
|
|
},
|
|
}),
|
|
this.prisma.cmppDownstreamDelivery.count({
|
|
where: {
|
|
...scopedWhere,
|
|
status: { in: ['pending', 'failed', 'unconfirmed', 'rejected'] },
|
|
retryCount: { gte: 4 },
|
|
},
|
|
}),
|
|
]);
|
|
const applicationIds = [...new Set(applicationGroups.map((item) => item.applicationId).filter((value): value is string => Boolean(value)))];
|
|
const applications: Array<{ id: string; name: string }> = applicationIds.length > 0
|
|
? await this.prisma.smsApplication.findMany({
|
|
where: { id: { in: applicationIds } },
|
|
select: { id: true, name: true },
|
|
})
|
|
: [];
|
|
const applicationMap = new Map<string, string>(applications.map((item) => [item.id, item.name]));
|
|
const applicationAlertMap = new Map<string, number>(
|
|
applicationAlertGroups.map((item) => [item.applicationId, item._count._all]),
|
|
);
|
|
const groupedByType = groupDownstreamByType(typeGroups);
|
|
const groupedByApplication = groupDownstreamByApplication(applicationGroups, applicationMap, applicationAlertMap);
|
|
|
|
return {
|
|
summary: {
|
|
total,
|
|
pending,
|
|
awaitingAck,
|
|
delivered,
|
|
failed,
|
|
unconfirmed,
|
|
rejected,
|
|
stalledPending,
|
|
stalledAck,
|
|
recentFailed,
|
|
alertCount: stalledPending + stalledAck + recentFailed,
|
|
},
|
|
typeBreakdown: ['receipt', 'uplink'].map((deliveryType) => ({
|
|
deliveryType,
|
|
total: groupedByType[deliveryType]?.total ?? 0,
|
|
pending: groupedByType[deliveryType]?.pending ?? 0,
|
|
awaitingAck: groupedByType[deliveryType]?.awaitingAck ?? 0,
|
|
delivered: groupedByType[deliveryType]?.delivered ?? 0,
|
|
failed: groupedByType[deliveryType]?.failed ?? 0,
|
|
unconfirmed: groupedByType[deliveryType]?.unconfirmed ?? 0,
|
|
rejected: groupedByType[deliveryType]?.rejected ?? 0,
|
|
})),
|
|
retryBuckets: [
|
|
{ label: '0次', count: retryZero },
|
|
{ label: '1-3次', count: retryLow },
|
|
{ label: '4次及以上', count: retryHigh },
|
|
],
|
|
topApplications: groupedByApplication
|
|
.sort((left, right) => (
|
|
right.alertCount - left.alertCount
|
|
|| right.failed - left.failed
|
|
|| right.pending - left.pending
|
|
|| left.name.localeCompare(right.name, 'zh-CN')
|
|
))
|
|
.slice(0, 5),
|
|
};
|
|
}
|
|
async listDownstreamRecoveryStatuses(query: DownstreamRecoveryStatusQuery) {
|
|
const recoveryStatuses = this.gatewayDownstreamRecoveryStatusDelegate();
|
|
const page = Math.max(1, Number(query.page ?? 1));
|
|
const pageSize = Math.min(100, Math.max(1, Number(query.pageSize ?? 10)));
|
|
const where = downstreamRecoveryStatusWhere(query);
|
|
const now = new Date();
|
|
const [items, total, runningCount, successCount, failedCount, waitingConnectionCount, backoffCount, categoryGroups] = await Promise.all([
|
|
recoveryStatuses.findMany({
|
|
where,
|
|
include: { tenant: true, application: true },
|
|
orderBy: [{ updatedAt: 'desc' }, { account: 'asc' }],
|
|
skip: (page - 1) * pageSize,
|
|
take: pageSize,
|
|
}),
|
|
recoveryStatuses.count({ where }),
|
|
recoveryStatuses.count({ where: { ...where, state: 'running' } }),
|
|
recoveryStatuses.count({ where: { ...where, state: 'success' } }),
|
|
recoveryStatuses.count({ where: { ...where, state: 'failed' } }),
|
|
recoveryStatuses.count({ where: { ...where, state: 'waiting_connection' } }),
|
|
recoveryStatuses.count({
|
|
where: {
|
|
...where,
|
|
nextRetryAt: { gt: now },
|
|
},
|
|
}),
|
|
recoveryStatuses.groupBy({
|
|
by: ['failureCategory'],
|
|
where,
|
|
_count: { _all: true },
|
|
}),
|
|
]);
|
|
return {
|
|
items,
|
|
total,
|
|
page,
|
|
pageSize,
|
|
summary: {
|
|
total,
|
|
running: runningCount,
|
|
success: successCount,
|
|
failed: failedCount,
|
|
waitingConnection: waitingConnectionCount,
|
|
backoff: backoffCount,
|
|
failureCategories: categoryGroups
|
|
.filter((item) => item.failureCategory)
|
|
.map((item) => ({
|
|
category: String(item.failureCategory),
|
|
count: item._count?._all ?? 0,
|
|
}))
|
|
.sort((left, right) => right.count - left.count || left.category.localeCompare(right.category)),
|
|
},
|
|
};
|
|
}
|
|
async listMessageSegmentAudits(query: MessageSegmentAuditQuery) {
|
|
const segmentAudits = (this.prisma as PrismaService & {
|
|
smsMessageSegmentAudit: {
|
|
findMany: (args: Record<string, unknown>) => Promise<any[]>;
|
|
};
|
|
}).smsMessageSegmentAudit;
|
|
if (!query.messageId && !query.messageRecordId) {
|
|
return [];
|
|
}
|
|
return segmentAudits.findMany({
|
|
where: {
|
|
messageRecordId: query.messageRecordId,
|
|
messageRecord: query.messageId ? { messageId: query.messageId } : undefined,
|
|
},
|
|
include: { channel: true, submitRecord: true },
|
|
orderBy: [{ createdAt: 'asc' }, { segmentIndex: 'asc' }, { id: 'asc' }],
|
|
});
|
|
}
|
|
async getDownstreamRecoveryStatus(id: string) {
|
|
const recoveryStatuses = this.gatewayDownstreamRecoveryStatusDelegate();
|
|
const item = await recoveryStatuses.findUnique({
|
|
where: { id },
|
|
include: { tenant: true, application: true },
|
|
});
|
|
if (!item) {
|
|
throw new NotFoundException('Recovery status not found');
|
|
}
|
|
return item;
|
|
}
|
|
async exportDownstreamRecoveryStatuses(query: DownstreamRecoveryStatusQuery) {
|
|
const recoveryStatuses = this.gatewayDownstreamRecoveryStatusDelegate();
|
|
const where = downstreamRecoveryStatusWhere(query);
|
|
const items = await recoveryStatuses.findMany({
|
|
where,
|
|
include: { tenant: true, application: true },
|
|
orderBy: [{ updatedAt: 'desc' }, { account: 'asc' }],
|
|
take: 5000,
|
|
});
|
|
const rows = [
|
|
[
|
|
'账号',
|
|
'企业',
|
|
'应用',
|
|
'Gateway实例',
|
|
'恢复状态',
|
|
'锁持有实例',
|
|
'锁过期时间',
|
|
'失败分类',
|
|
'尝试次数',
|
|
'最后尝试时间',
|
|
'恢复成功时间',
|
|
'恢复失败时间',
|
|
'下次恢复时间',
|
|
'最后错误',
|
|
'最后跳过原因',
|
|
'创建时间',
|
|
'更新时间',
|
|
],
|
|
...items.map((item) => [
|
|
item.account ?? '',
|
|
item.tenant?.name ?? '',
|
|
item.application?.name ?? '',
|
|
item.gatewayInstanceId ?? '',
|
|
item.state ?? '',
|
|
(item as { lockOwner?: string | null }).lockOwner ?? '',
|
|
formatCsvDate((item as { lockExpiresAt?: Date | string | null }).lockExpiresAt),
|
|
(item as { failureCategory?: string | null }).failureCategory ?? '',
|
|
String(item.attemptCount ?? 0),
|
|
formatCsvDate(item.lastAttemptAt),
|
|
formatCsvDate(item.lastSuccessAt),
|
|
formatCsvDate(item.lastFailureAt),
|
|
formatCsvDate(item.nextRetryAt),
|
|
item.lastError ?? '',
|
|
item.lastSkipReason ?? '',
|
|
formatCsvDate(item.createdAt),
|
|
formatCsvDate(item.updatedAt),
|
|
]),
|
|
];
|
|
|
|
return {
|
|
fileName: `gateway-downstream-recovery-statuses-${formatExportTimestamp(new Date())}.csv`,
|
|
content: rows.map((row) => row.map(escapeCsvCell).join(',')).join('\n'),
|
|
total: items.length,
|
|
};
|
|
}
|
|
private gatewayDownstreamRecoveryStatusDelegate() {
|
|
return (this.prisma as PrismaService & {
|
|
gatewayDownstreamRecoveryStatus: {
|
|
findMany: (args: Record<string, unknown>) => Promise<any[]>;
|
|
count: (args: Record<string, unknown>) => Promise<number>;
|
|
findUnique: (args: Record<string, unknown>) => Promise<any | null>;
|
|
groupBy: (args: Record<string, unknown>) => Promise<any[]>;
|
|
};
|
|
}).gatewayDownstreamRecoveryStatus;
|
|
}
|
|
}
|