hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修复监控纳管保存并移除运行概况最近记录
CSS quality / css-quality (push) Has been cancelled
Details

Browse Source
This commit is contained in:
hectorzhao
2026-09-06 20:22:48 +08:00
parent f059674852
commit 4c210723cd
9 changed files with 194 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/operations/queries/uplink.queries.spec.ts
+17
View File
@@ -0,0 +1,17 @@
import type { PrismaService } from '../../prisma/prisma.service';
import { OperationsUplinkQueries } from './uplink.queries';
describe('runtime monitor summary', () => {
it('keeps tenant/channel status counts without reading recent business records', async () => {
const byStatus = [{ status: 'delivered', _count: { _all: 3 } }];
const groupBy = jest.fn().mockResolvedValue(byStatus);
// No detail delegates: accessing any removed query fails this test.
const queries = new OperationsUplinkQueries({ smsMessageRecord: { groupBy } } as unknown as PrismaService);
expect(await queries.monitor({ tenantId: 'tenant-a', channelId: 'channel-a' })).toEqual({ byStatus });
expect(groupBy).toHaveBeenCalledWith({
by: ['status'],
where: expect.objectContaining({ tenantId: 'tenant-a', channelId: 'channel-a' }),
_count: { _all: true },
});
});
});
api/src/operations/queries/uplink.queries.ts
+54 -36
View File
@@ -1,16 +1,22 @@
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, DownstreamDeliveryQuery, DownstreamDeliveryDashboardQuery, DownstreamRecoveryStatusQuery, MessageSegmentAuditQuery, SignatureQualityQuery } from '../operations.contracts';
import { messageWhere, 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';
import { messageWhere, clientUplinkView } from '../operations.helpers';
// R2 uplink query domain. Method bodies are preserved byte-for-byte from the facade baseline.
// Uplink record queries and the status-only runtime summary.
export class OperationsUplinkQueries {
constructor(private readonly prisma: PrismaService) {}
listUplinkMessages(query: { tenantId?: string; channelId?: string; applicationId?: string; phoneNumber?: string; keyword?: string; startTime?: string; endTime?: string; page?: number; pageSize?: number }) {
listUplinkMessages(query: {
tenantId?: string;
channelId?: string;
applicationId?: string;
phoneNumber?: string;
keyword?: string;
startTime?: string;
endTime?: string;
page?: number;
pageSize?: number;
}) {
return this.prisma.smsUplinkMessage.findMany({
where: {
tenantId: query.tenantId,
@@ -18,7 +24,13 @@ listUplinkMessages(query: { tenantId?: string; channelId?: string; applicationId
applicationId: query.applicationId,
phoneNumber: query.phoneNumber ? { contains: query.phoneNumber } : undefined,
content: query.keyword ? { contains: query.keyword } : undefined,
receivedAt: query.startTime || query.endTime ? { gte: query.startTime ? new Date(query.startTime) : undefined, lte: query.endTime ? new Date(query.endTime) : undefined } : undefined,
receivedAt:
query.startTime || query.endTime
? {
gte: query.startTime ? new Date(query.startTime) : undefined,
lte: query.endTime ? new Date(query.endTime) : undefined,
}
: undefined,
},
include: {
tenant: true,
@@ -39,11 +51,33 @@ listUplinkMessages(query: { tenantId?: string; channelId?: string; applicationId
take: query.pageSize ?? 500,
});
}
async listClientUplinkMessages(query: { tenantId?: string; applicationId?: string; phoneNumber?: string; keyword?: string; startTime?: string; endTime?: string; page?: number; pageSize?: number }) {
async listClientUplinkMessages(query: {
tenantId?: string;
applicationId?: string;
phoneNumber?: string;
keyword?: string;
startTime?: string;
endTime?: string;
page?: number;
pageSize?: number;
}) {
const items = await this.listUplinkMessages(query);
return items.map(clientUplinkView);
}
async listUplinkMessagesPage(query: { tenantId?: string; channelId?: string; applicationId?: string; phoneNumber?: string; keyword?: string; startTime?: string; endTime?: string; page?: number; pageSize?: number }, clientView = false) {
async listUplinkMessagesPage(
query: {
tenantId?: string;
channelId?: string;
applicationId?: string;
phoneNumber?: string;
keyword?: string;
startTime?: string;
endTime?: string;
page?: number;
pageSize?: number;
},
clientView = false,
) {
const page = Math.max(1, Math.floor(Number(query.page) || 1));
const pageSize = Math.min(100, Math.max(1, Math.floor(Number(query.pageSize) || 10)));
const where: Prisma.SmsUplinkMessageWhereInput = {
@@ -52,10 +86,13 @@ async listUplinkMessagesPage(query: { tenantId?: string; channelId?: string; app
applicationId: query.applicationId,
phoneNumber: query.phoneNumber ? { contains: query.phoneNumber } : undefined,
content: query.keyword ? { contains: query.keyword } : undefined,
receivedAt: query.startTime || query.endTime ? {
gte: query.startTime ? new Date(query.startTime) : undefined,
lte: query.endTime ? new Date(query.endTime) : undefined,
} : undefined,
receivedAt:
query.startTime || query.endTime
? {
gte: query.startTime ? new Date(query.startTime) : undefined,
lte: query.endTime ? new Date(query.endTime) : undefined,
}
: undefined,
};
const [rawItems, total] = await Promise.all([
this.listUplinkMessages({ ...query, page, pageSize }),
@@ -68,28 +105,9 @@ async listUplinkMessagesPage(query: { tenantId?: string; channelId?: string; app
pageSize,
};
}
async monitor(query: { tenantId?: string; channelId?: string }) {
async monitor(query: { tenantId?: string; channelId?: string }) {
const where = messageWhere(query);
const [byStatus, recentMessages, recentReceipts, recentUplinks] = await Promise.all([
this.prisma.smsMessageRecord.groupBy({ by: ['status'], where, _count: { _all: true } }),
this.prisma.smsMessageRecord.findMany({
where,
include: { submitRecords: true, receiptRecords: true },
orderBy: { queuedAt: 'desc' },
take: 20,
}),
this.prisma.smsReceiptRecord.findMany({
where: { tenantId: query.tenantId, channelId: query.channelId },
orderBy: { createdAt: 'desc' },
take: 20,
}),
this.listUplinkMessages({ tenantId: query.tenantId, channelId: query.channelId }),
]);
return {
byStatus,
recentMessages,
recentReceipts,
recentUplinks: recentUplinks.slice(0, 20),
};
const byStatus = await this.prisma.smsMessageRecord.groupBy({ by: ['status'], where, _count: { _all: true } });
return { byStatus };
}
}
api/src/sending-monitor/sending-monitor.module.ts
+2 -2
View File
@@ -125,7 +125,7 @@ export class SendingMonitorService {
select: { id: true },
});
if (!channel) throw new NotFoundException('通道不存在');
await tx.$queryRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`monitor-target:${id}`},0))`;
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`monitor-target:${id}`},0))`;
const old = await tx.$queryRawUnsafe<Array<{ version: number; enabled: boolean }>>(
`SELECT * FROM "SendingMonitorTarget" WHERE "channelId"=$1`,
id,
@@ -239,7 +239,7 @@ export class SendingMonitorService {
const scopeKey = JSON.stringify(scope),
type = monitorType(body.type);
return this.prisma.$transaction(async (tx) => {
await tx.$queryRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`monitor-rule:${type}:${scopeKey}`},0))`;
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`monitor-rule:${type}:${scopeKey}`},0))`;
const old = await tx.$queryRawUnsafe<Array<{ id: string; version: number }>>(
`SELECT * FROM "SendingMonitorRule" WHERE type=$1 AND "scopeKey"=$2`,
type,