fix: harden dependencies and downstream delivery
This commit is contained in:
hectorzhao
Generated
+19
-19
@@ -8,10 +8,10 @@
|
||||
"name": "cmpp-platform-api",
|
||||
"version": "0.1.0",
|
||||
"dependencies": {
|
||||
"@nestjs/common": "^11.1.9",
|
||||
"@nestjs/common": "^11.1.28",
|
||||
"@nestjs/config": "^4.0.2",
|
||||
"@nestjs/core": "^11.1.9",
|
||||
"@nestjs/platform-express": "^11.1.9",
|
||||
"@nestjs/core": "^11.1.28",
|
||||
"@nestjs/platform-express": "^11.1.28",
|
||||
"@nestjs/swagger": "^11.2.3",
|
||||
"@prisma/adapter-pg": "^7.8.0",
|
||||
"@prisma/client": "^7.0.1",
|
||||
@@ -649,9 +649,9 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@hono/node-server": {
|
||||
"version": "1.19.11",
|
||||
"resolved": "https://registry.npmjs.org/@hono/node-server/-/node-server-1.19.11.tgz",
|
||||
"integrity": "sha512-dr8/3zEaB+p0D2n/IUrlPF1HZm586qgJNXK1a9fhg/PzdtkK7Ksd5l312tJX2yBuALqDYBlG20QEbayqPyxn+g==",
|
||||
"version": "1.19.13",
|
||||
"resolved": "https://registry.npmjs.org/@hono/node-server/-/node-server-1.19.13.tgz",
|
||||
"integrity": "sha512-TsQLe4i2gvoTtrHje625ngThGBySOgSK3Xo2XRYOdqGN1teR8+I7vchQC46uLJi8OF62YTYA3AhSpumtkhsaKQ==",
|
||||
"devOptional": true,
|
||||
"license": "MIT",
|
||||
"engines": {
|
||||
@@ -1302,9 +1302,9 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@nestjs/common": {
|
||||
"version": "11.1.27",
|
||||
"resolved": "https://registry.npmjs.org/@nestjs/common/-/common-11.1.27.tgz",
|
||||
"integrity": "sha512-kEGSzqM2lWr4whh4Ubflw+oPZSEzxvRMu9WL+LveZploJWTjec5bBlCiRVlVzTPg2kIwBiLwWSvCCW7Wnin1gg==",
|
||||
"version": "11.1.28",
|
||||
"resolved": "https://registry.npmjs.org/@nestjs/common/-/common-11.1.28.tgz",
|
||||
"integrity": "sha512-bRImsxibie+AM7xjdwcrm/gr5YeacI65kSBNzTufa1Ib5iwziaY/lqMtRh9THq6pbV4e1HP9aI2ZxGUumnmaoQ==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"file-type": "21.3.4",
|
||||
@@ -1348,9 +1348,9 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@nestjs/core": {
|
||||
"version": "11.1.27",
|
||||
"resolved": "https://registry.npmjs.org/@nestjs/core/-/core-11.1.27.tgz",
|
||||
"integrity": "sha512-K6DX7hcqmZdeXkv7tsPakKBRCgqL19a4mtbX4FluY0hWtFdtPKp6lbe+lb8gWPfvLdbOWr/CPScn7BSjBX+Ecg==",
|
||||
"version": "11.1.28",
|
||||
"resolved": "https://registry.npmjs.org/@nestjs/core/-/core-11.1.28.tgz",
|
||||
"integrity": "sha512-06m63xIRj8+l8uOeh/8LnYupGubkyu4f+bPKIadaSui6vK9KpXgoz7HveT1yOVLcEt0M0oCOEW5EuEXZkEmBBQ==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"fast-safe-stringify": "2.1.1",
|
||||
@@ -1407,14 +1407,14 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@nestjs/platform-express": {
|
||||
"version": "11.1.27",
|
||||
"resolved": "https://registry.npmjs.org/@nestjs/platform-express/-/platform-express-11.1.27.tgz",
|
||||
"integrity": "sha512-0ZFhz6H6EdGh4xQVbUNwjoAwBuz73P7FvUAl67h9CTdMqQlJDaQYJApBv8pKfVZ1fGjMCbl0m9DcC6pXaZPWSQ==",
|
||||
"version": "11.1.28",
|
||||
"resolved": "https://registry.npmjs.org/@nestjs/platform-express/-/platform-express-11.1.28.tgz",
|
||||
"integrity": "sha512-hU+9Sz4m+onHrR5AmelI59QKmY/Re546bPnygnpqqeQdHDiJpBgjWbL4t6Jr73CBpS60cpyng7WzjgphNB9iwA==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"cors": "2.8.6",
|
||||
"express": "5.2.1",
|
||||
"multer": "2.1.1",
|
||||
"multer": "2.2.0",
|
||||
"path-to-regexp": "8.4.2",
|
||||
"tslib": "2.8.1"
|
||||
},
|
||||
@@ -5712,9 +5712,9 @@
|
||||
}
|
||||
},
|
||||
"node_modules/multer": {
|
||||
"version": "2.1.1",
|
||||
"resolved": "https://registry.npmjs.org/multer/-/multer-2.1.1.tgz",
|
||||
"integrity": "sha512-mo+QTzKlx8R7E5ylSXxWzGoXoZbOsRMpyitcht8By2KHvMbf3tjwosZ/Mu/XYU6UuJ3VZnODIrak5ZrPiPyB6A==",
|
||||
"version": "2.2.0",
|
||||
"resolved": "https://registry.npmjs.org/multer/-/multer-2.2.0.tgz",
|
||||
"integrity": "sha512-6rdyFg2kLrMh9Jee7/BMPuV9lEAd7lLW2YUpF9/YxR7njyoUwwQ0ZPh3TaIY50Sw6vlyD2HW3wGOkTS4P79xrQ==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"append-field": "^1.0.0",
|
||||
|
||||
+6
-3
@@ -14,10 +14,10 @@
|
||||
"prisma:migrate:deploy": "prisma migrate deploy"
|
||||
},
|
||||
"dependencies": {
|
||||
"@nestjs/common": "^11.1.9",
|
||||
"@nestjs/common": "^11.1.28",
|
||||
"@nestjs/config": "^4.0.2",
|
||||
"@nestjs/core": "^11.1.9",
|
||||
"@nestjs/platform-express": "^11.1.9",
|
||||
"@nestjs/core": "^11.1.28",
|
||||
"@nestjs/platform-express": "^11.1.28",
|
||||
"@nestjs/swagger": "^11.2.3",
|
||||
"@prisma/adapter-pg": "^7.8.0",
|
||||
"@prisma/client": "^7.0.1",
|
||||
@@ -38,5 +38,8 @@
|
||||
"ts-jest": "^29.4.11",
|
||||
"ts-node": "^10.9.2",
|
||||
"typescript": "^6.0.3"
|
||||
},
|
||||
"overrides": {
|
||||
"@hono/node-server": "1.19.13"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -48,7 +48,17 @@ export class FilesController {
|
||||
}
|
||||
|
||||
@Post('upload')
|
||||
@UseInterceptors(FileInterceptor('file', { limits: { fileSize: 20 * 1024 * 1024 } }))
|
||||
@UseInterceptors(FileInterceptor('file', {
|
||||
limits: {
|
||||
fileSize: 20 * 1024 * 1024,
|
||||
files: 1,
|
||||
fields: 4,
|
||||
parts: 5,
|
||||
fieldNameSize: 100,
|
||||
fieldSize: 1024,
|
||||
headerPairs: 100,
|
||||
},
|
||||
}))
|
||||
upload(@UploadedFile() file: UploadedMultipartFile, @Body('purpose') purpose: string, @Body('prefix') prefix?: string, @TenantId() tenantId?: string) {
|
||||
if (!file) {
|
||||
throw new BadRequestException('Upload file is required');
|
||||
|
||||
@@ -254,6 +254,7 @@ describe('OperationsService', () => {
|
||||
.mockResolvedValueOnce(1)
|
||||
.mockResolvedValueOnce(1)
|
||||
.mockResolvedValueOnce(2);
|
||||
prisma.accountTransaction.aggregate.mockResolvedValueOnce({ _count: { _all: 2 }, _sum: { amountCents: 10 } });
|
||||
const service = new OperationsService(prisma as never);
|
||||
|
||||
await expect(service.dashboard({ tenantId: 'tenant-1' })).resolves.toEqual(
|
||||
@@ -269,6 +270,7 @@ describe('OperationsService', () => {
|
||||
templates: 1,
|
||||
total: 5,
|
||||
},
|
||||
today: expect.objectContaining({ returnedCents: 10 }),
|
||||
gatewayConnections: [{ status: 'connected', _count: { _all: 1 }, _sum: { currentConnections: 2, desiredConnections: 2 } }],
|
||||
downstreamDeliverySummary: expect.objectContaining({
|
||||
pending: 3,
|
||||
@@ -282,7 +284,14 @@ describe('OperationsService', () => {
|
||||
}),
|
||||
);
|
||||
expect(prisma.cmppDownstreamDelivery.count).toHaveBeenNthCalledWith(4, {
|
||||
where: { tenantId: 'tenant-1', status: 'pending', createdAt: { lte: expect.any(Date) } },
|
||||
where: {
|
||||
tenantId: 'tenant-1',
|
||||
status: 'pending',
|
||||
OR: [
|
||||
{ lastRetriedAt: null, createdAt: { lte: expect.any(Date) } },
|
||||
{ lastRetriedAt: { lte: expect.any(Date) } },
|
||||
],
|
||||
},
|
||||
});
|
||||
expect(prisma.cmppDownstreamDelivery.count).toHaveBeenNthCalledWith(5, {
|
||||
where: { tenantId: 'tenant-1', status: 'awaiting_ack', ackDeadlineAt: { lte: expect.any(Date) } },
|
||||
@@ -294,6 +303,18 @@ describe('OperationsService', () => {
|
||||
updatedAt: { gte: expect.any(Date) },
|
||||
},
|
||||
});
|
||||
expect(prisma.accountTransaction.aggregate).toHaveBeenCalledWith({
|
||||
where: {
|
||||
tenantId: 'tenant-1',
|
||||
createdAt: { gte: expect.any(Date) },
|
||||
OR: [
|
||||
{ transactionType: 'refunded' },
|
||||
{ transactionType: 'released', relatedType: 'sms_message_record' },
|
||||
],
|
||||
},
|
||||
_sum: { amountCents: true },
|
||||
_count: { _all: true },
|
||||
});
|
||||
await service.statistics({ tenantId: 'tenant-1', groupBy: 'application' });
|
||||
|
||||
expect(prisma.smsMessageRecord.groupBy).toHaveBeenCalledWith({
|
||||
@@ -516,10 +537,16 @@ describe('OperationsService', () => {
|
||||
by: ['applicationId'],
|
||||
where: {
|
||||
AND: [
|
||||
{ tenantId: 'tenant-1', applicationId: 'app-1', deliveryType: undefined },
|
||||
{ tenantId: 'tenant-1', applicationId: 'app-1', deliveryType: undefined, createdAt: undefined },
|
||||
{
|
||||
OR: [
|
||||
{ status: 'pending', createdAt: { lte: expect.any(Date) } },
|
||||
{
|
||||
status: 'pending',
|
||||
OR: [
|
||||
{ lastRetriedAt: null, createdAt: { lte: expect.any(Date) } },
|
||||
{ lastRetriedAt: { lte: expect.any(Date) } },
|
||||
],
|
||||
},
|
||||
{ status: 'awaiting_ack', ackDeadlineAt: { lte: expect.any(Date) } },
|
||||
{ status: { in: ['failed', 'unconfirmed', 'rejected'] }, updatedAt: { gte: expect.any(Date) } },
|
||||
],
|
||||
|
||||
@@ -186,7 +186,7 @@ export class OperationsService {
|
||||
_count: { _all: true },
|
||||
}),
|
||||
this.prisma.accountTransaction.aggregate({
|
||||
where: { tenantId: query.tenantId, transactionType: 'refunded', createdAt: { gte: sinceToday } },
|
||||
where: returnedTransactionWhere(sinceToday, query.tenantId),
|
||||
_sum: { amountCents: true },
|
||||
_count: { _all: true },
|
||||
}),
|
||||
@@ -230,8 +230,7 @@ export class OperationsService {
|
||||
this.prisma.cmppDownstreamDelivery.count({
|
||||
where: {
|
||||
tenantId: query.tenantId,
|
||||
status: 'pending',
|
||||
createdAt: { lte: downstreamAlertWindow.stalledPendingAt },
|
||||
...stalledPendingWhere(downstreamAlertWindow.stalledPendingAt),
|
||||
},
|
||||
}),
|
||||
this.prisma.cmppDownstreamDelivery.count({
|
||||
@@ -261,6 +260,7 @@ export class OperationsService {
|
||||
unknown: todayTotals.unknown,
|
||||
successRate: todayTotals.total > 0 ? Number(((todayTotals.delivered / todayTotals.total) * 100).toFixed(1)) : 0,
|
||||
spendCents: todayTotals.amountCents,
|
||||
returnedCents: transactionAggregate._sum.amountCents ?? 0,
|
||||
billingUnits: todayTotals.billingUnits,
|
||||
},
|
||||
uplinkCount,
|
||||
@@ -439,8 +439,7 @@ export class OperationsService {
|
||||
this.prisma.cmppDownstreamDelivery.count({
|
||||
where: {
|
||||
...scopedWhere,
|
||||
status: 'pending',
|
||||
createdAt: { lte: downstreamAlertWindow.stalledPendingAt },
|
||||
...stalledPendingWhere(downstreamAlertWindow.stalledPendingAt),
|
||||
},
|
||||
}),
|
||||
this.prisma.cmppDownstreamDelivery.count({
|
||||
@@ -841,6 +840,17 @@ function startOfToday() {
|
||||
return date;
|
||||
}
|
||||
|
||||
function returnedTransactionWhere(since: Date, tenantId?: string): Prisma.AccountTransactionWhereInput {
|
||||
return {
|
||||
tenantId,
|
||||
createdAt: { gte: since },
|
||||
OR: [
|
||||
{ transactionType: 'refunded' },
|
||||
{ transactionType: 'released', relatedType: 'sms_message_record' },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
function createdAtRange(range?: string): Prisma.DateTimeFilter | undefined {
|
||||
if (!range || range === 'all') {
|
||||
return undefined;
|
||||
@@ -882,7 +892,7 @@ function downstreamAlertWhere(
|
||||
scopedWhere,
|
||||
{
|
||||
OR: [
|
||||
{ status: 'pending', createdAt: { lte: window.stalledPendingAt } },
|
||||
stalledPendingWhere(window.stalledPendingAt),
|
||||
{ status: 'awaiting_ack', ackDeadlineAt: { lte: window.now } },
|
||||
{ status: { in: ['failed', 'unconfirmed', 'rejected'] }, updatedAt: { gte: window.recentFailedAt } },
|
||||
],
|
||||
@@ -891,6 +901,16 @@ function downstreamAlertWhere(
|
||||
};
|
||||
}
|
||||
|
||||
function stalledPendingWhere(cutoff: Date): Prisma.CmppDownstreamDeliveryWhereInput {
|
||||
return {
|
||||
status: 'pending',
|
||||
OR: [
|
||||
{ lastRetriedAt: null, createdAt: { lte: cutoff } },
|
||||
{ lastRetriedAt: { lte: cutoff } },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
function downstreamDeliveryScopedWhere(query: DownstreamDeliveryDashboardQuery): Prisma.CmppDownstreamDeliveryWhereInput {
|
||||
const createdAtFrom = parseDateBoundary(query.createdAtFrom, false);
|
||||
const createdAtTo = parseDateBoundary(query.createdAtTo, true);
|
||||
|
||||
@@ -1574,6 +1574,50 @@ describe('SendChainService', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('immediately terminates an unrecoverable downstream delivery', async () => {
|
||||
const { service, prisma } = createService();
|
||||
prisma.cmppDownstreamDelivery.findUnique.mockResolvedValue({
|
||||
id: 'delivery-1',
|
||||
tenantId: 'tenant-1',
|
||||
applicationId: 'app-1',
|
||||
messageId: 'MSG-1',
|
||||
deliveryType: 'receipt',
|
||||
retryCount: 0,
|
||||
retryEnabled: true,
|
||||
});
|
||||
|
||||
await service.markDownstreamDeliveryFailed(
|
||||
'delivery-1',
|
||||
'历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id',
|
||||
'unrecoverable',
|
||||
);
|
||||
|
||||
expect(prisma.cmppDownstreamDelivery.update).toHaveBeenCalledWith(expect.objectContaining({
|
||||
data: expect.objectContaining({
|
||||
status: 'failed',
|
||||
retryCount: 1,
|
||||
nextRetryAt: null,
|
||||
}),
|
||||
}));
|
||||
});
|
||||
|
||||
it('does not let the pending timeout scan overwrite a delivery that is already awaiting acknowledgement', async () => {
|
||||
const { service, prisma } = createService();
|
||||
const awaitingAck = {
|
||||
id: 'delivery-1', status: 'awaiting_ack', tenantId: 'tenant-1', applicationId: 'app-1',
|
||||
messageId: 'MSG-1', deliveryType: 'receipt', retryCount: 0,
|
||||
};
|
||||
prisma.cmppDownstreamDelivery.findUnique.mockResolvedValue(awaitingAck);
|
||||
|
||||
await expect(service.markDownstreamDeliveryFailed(
|
||||
'delivery-1',
|
||||
'下游投递排队超过 72 小时，系统自动终止重试',
|
||||
'queue_timeout',
|
||||
)).resolves.toEqual(awaitingAck);
|
||||
|
||||
expect(prisma.cmppDownstreamDelivery.update).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('only marks downstream delivery delivered after a successful CMPP_DELIVER_RESP', async () => {
|
||||
const { service, prisma } = createService();
|
||||
|
||||
@@ -1769,6 +1813,35 @@ describe('SendChainService', () => {
|
||||
expect(service['postGatewayControl']).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('terminates a manual requeue when gateway reports it is unrecoverable', async () => {
|
||||
const { service, prisma } = createService();
|
||||
prisma.cmppDownstreamDelivery.findUnique.mockResolvedValueOnce({
|
||||
id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1',
|
||||
deliveryType: 'receipt', status: 'failed', retryCount: 3, manualRetryCount: 0,
|
||||
payload: { account: '100001', messageId: 'MSG-1', receiptStatus: 'delivered' },
|
||||
application: { cmppAccount: '100001' },
|
||||
}).mockResolvedValueOnce({
|
||||
id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1',
|
||||
deliveryType: 'receipt', status: 'pending', retryCount: 0, retryEnabled: true,
|
||||
});
|
||||
service['postGatewayControl'] = jest.fn().mockResolvedValue({
|
||||
sent: false,
|
||||
retryable: false,
|
||||
reasonCode: 'MISSING_SUBMIT_SEQUENCE_ID',
|
||||
errorMessage: '历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id，系统已终止重投',
|
||||
});
|
||||
|
||||
await service.requeueDownstreamDelivery('delivery-1');
|
||||
|
||||
expect(prisma.cmppDownstreamDelivery.update).toHaveBeenLastCalledWith(expect.objectContaining({
|
||||
data: expect.objectContaining({
|
||||
status: 'failed',
|
||||
nextRetryAt: null,
|
||||
lastError: expect.stringContaining('MISSING_SUBMIT_SEQUENCE_ID'),
|
||||
}),
|
||||
}));
|
||||
});
|
||||
|
||||
it('supports batch requeue of downstream deliveries', async () => {
|
||||
const { service } = createService();
|
||||
service.requeueDownstreamDelivery = jest.fn()
|
||||
@@ -1819,16 +1892,41 @@ describe('SendChainService', () => {
|
||||
expect(prisma.smsBatchTask.update).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('terminates downstream deliveries that remain pending for 72 hours after the latest manual retry', async () => {
|
||||
const { service, prisma } = createService();
|
||||
prisma.cmppDownstreamDelivery.findMany.mockResolvedValue([{ id: 'delivery-expired' }]);
|
||||
service.markDownstreamDeliveryFailed = jest.fn().mockResolvedValue({ id: 'delivery-expired', status: 'failed' });
|
||||
|
||||
await expect(service.markExpiredDownstreamDeliveries(72)).resolves.toEqual({ failed: 1 });
|
||||
|
||||
expect(prisma.cmppDownstreamDelivery.findMany).toHaveBeenCalledWith(expect.objectContaining({
|
||||
where: {
|
||||
status: 'pending',
|
||||
OR: [
|
||||
{ lastRetriedAt: null, createdAt: { lte: expect.any(Date) } },
|
||||
{ lastRetriedAt: { lte: expect.any(Date) } },
|
||||
],
|
||||
},
|
||||
}));
|
||||
expect(service.markDownstreamDeliveryFailed).toHaveBeenCalledWith(
|
||||
'delivery-expired',
|
||||
'下游投递排队超过 72 小时，系统自动终止重试',
|
||||
'queue_timeout',
|
||||
);
|
||||
});
|
||||
|
||||
it('starts the automatic receipt-timeout scan after application startup', async () => {
|
||||
jest.useFakeTimers();
|
||||
const previousEnabled = process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED;
|
||||
const { service } = createService();
|
||||
const scan = jest.spyOn(service, 'markUnknownTimeout').mockResolvedValue({ timeout: 0 });
|
||||
const downstreamScan = jest.spyOn(service, 'markExpiredDownstreamDeliveries').mockResolvedValue({ failed: 0 });
|
||||
try {
|
||||
process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED = 'true';
|
||||
service.onModuleInit();
|
||||
await jest.advanceTimersByTimeAsync(60_000);
|
||||
expect(scan).toHaveBeenCalledWith({});
|
||||
expect(downstreamScan).toHaveBeenCalledWith();
|
||||
await service.onModuleDestroy();
|
||||
} finally {
|
||||
if (previousEnabled === undefined) delete process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED;
|
||||
|
||||
@@ -119,11 +119,21 @@ export interface GatewayDownstreamAcknowledgedDto extends GatewayDownstreamSentD
|
||||
acknowledgedAt?: string;
|
||||
}
|
||||
|
||||
export type GatewayDownstreamFailureType = 'send_failed' | 'ack_timeout' | 'ack_rejected' | 'ack_invalid' | 'connection_lost';
|
||||
export type GatewayDownstreamFailureType =
|
||||
| 'send_failed'
|
||||
| 'ack_timeout'
|
||||
| 'ack_rejected'
|
||||
| 'ack_invalid'
|
||||
| 'connection_lost'
|
||||
| 'unrecoverable'
|
||||
| 'queue_timeout';
|
||||
|
||||
type GatewayControlDeliveryResult = {
|
||||
sent?: boolean;
|
||||
delivered?: boolean;
|
||||
retryable?: boolean;
|
||||
reasonCode?: string;
|
||||
errorMessage?: string;
|
||||
connectionId?: string;
|
||||
sequenceId?: string;
|
||||
messageId?: string;
|
||||
@@ -218,6 +228,7 @@ const GATEWAY_SUBMIT_STREAM = 'gateway.submit.commands';
|
||||
const DEFAULT_DOWNSTREAM_RETRY_DELAY_MS = 60_000;
|
||||
const DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS = 30 * 60_000;
|
||||
const DEFAULT_DOWNSTREAM_MAX_RETRIES = 10;
|
||||
const DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS = 72;
|
||||
const DEFAULT_RECEIPT_TIMEOUT_HOURS = 72;
|
||||
const DEFAULT_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS = 5 * 60_000;
|
||||
const RECEIPT_TIMEOUT_INITIAL_DELAY_MS = 60_000;
|
||||
@@ -879,6 +890,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
receiptStatus: data.receiptStatus,
|
||||
rawStatus: data.rawStatus,
|
||||
errorCode: data.errorCode,
|
||||
submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined,
|
||||
deliveredAt: deliveredAt.toISOString(),
|
||||
},
|
||||
});
|
||||
@@ -1044,10 +1056,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
if (delivery.status === 'delivered') {
|
||||
return delivery;
|
||||
}
|
||||
if (failureType === 'queue_timeout' && delivery.status !== 'pending') {
|
||||
return delivery;
|
||||
}
|
||||
const retryCount = (delivery.retryCount ?? 0) + 1;
|
||||
const acknowledgementFailure = failureType === 'ack_timeout' || failureType === 'ack_rejected' || failureType === 'ack_invalid' || failureType === 'connection_lost';
|
||||
const retryAllowed = !acknowledgementFailure || delivery.retryEnabled !== false;
|
||||
const finalFailure = !retryAllowed || retryCount >= downstreamMaxRetries();
|
||||
const nonRetryableFailure = failureType === 'unrecoverable' || failureType === 'queue_timeout';
|
||||
const finalFailure = nonRetryableFailure || !retryAllowed || retryCount >= downstreamMaxRetries();
|
||||
const finalStatus = failureType === 'ack_rejected' ? 'rejected' : acknowledgementFailure ? 'unconfirmed' : 'failed';
|
||||
const updated = await this.prisma.cmppDownstreamDelivery.update({
|
||||
where: { id },
|
||||
@@ -1327,7 +1343,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
if (result.sent || result.delivered) {
|
||||
return this.markDownstreamDeliverySent({ id: delivery.id, ...result });
|
||||
}
|
||||
return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: delivery.id } });
|
||||
return this.markDownstreamDeliveryFailed(
|
||||
delivery.id,
|
||||
downstreamControlFailureMessage(result),
|
||||
result.retryable === false ? 'unrecoverable' : 'send_failed',
|
||||
);
|
||||
} catch (error) {
|
||||
return this.markDownstreamDeliveryFailed(
|
||||
delivery.id,
|
||||
@@ -1504,6 +1524,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
) as GatewayControlDeliveryResult;
|
||||
if (result.sent || result.delivered) {
|
||||
await this.markDownstreamDeliverySent({ id: delivery.id, ...result });
|
||||
} else {
|
||||
await this.markDownstreamDeliveryFailed(
|
||||
delivery.id,
|
||||
downstreamControlFailureMessage(result),
|
||||
result.retryable === false ? 'unrecoverable' : 'send_failed',
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
await this.markDownstreamDeliveryFailed(delivery.id, error instanceof Error ? error.message : 'Gateway control delivery failed');
|
||||
@@ -1884,12 +1910,39 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
|
||||
return { timeout };
|
||||
}
|
||||
|
||||
async markExpiredDownstreamDeliveries(olderThanHours = downstreamPendingTimeoutHours()) {
|
||||
const cutoff = new Date(Date.now() - olderThanHours * 60 * 60_000);
|
||||
const expired = await this.prisma.cmppDownstreamDelivery.findMany({
|
||||
where: {
|
||||
status: 'pending',
|
||||
OR: [
|
||||
{ lastRetriedAt: null, createdAt: { lte: cutoff } },
|
||||
{ lastRetriedAt: { lte: cutoff } },
|
||||
],
|
||||
},
|
||||
select: { id: true },
|
||||
take: 500,
|
||||
});
|
||||
for (const delivery of expired) {
|
||||
await this.markDownstreamDeliveryFailed(
|
||||
delivery.id,
|
||||
`下游投递排队超过 ${olderThanHours} 小时，系统自动终止重试`,
|
||||
'queue_timeout',
|
||||
);
|
||||
}
|
||||
return { failed: expired.length };
|
||||
}
|
||||
|
||||
private async runReceiptTimeoutScan() {
|
||||
if (this.receiptTimeoutScanRunning) return;
|
||||
this.receiptTimeoutScanRunning = true;
|
||||
try {
|
||||
const result = await this.markUnknownTimeout({});
|
||||
if (result.timeout > 0) this.logger.log(`Marked ${result.timeout} messages as receipt timeout and refunded charged messages`);
|
||||
const [receiptResult, downstreamResult] = await Promise.all([
|
||||
this.markUnknownTimeout({}),
|
||||
this.markExpiredDownstreamDeliveries(),
|
||||
]);
|
||||
if (receiptResult.timeout > 0) this.logger.log(`Marked ${receiptResult.timeout} messages as receipt timeout and refunded charged messages`);
|
||||
if (downstreamResult.failed > 0) this.logger.log(`Terminated ${downstreamResult.failed} expired downstream deliveries`);
|
||||
} catch (error) {
|
||||
this.logger.error('Receipt timeout scan failed', error instanceof Error ? error.stack : String(error));
|
||||
} finally {
|
||||
@@ -2863,6 +2916,20 @@ function downstreamMaxRetries() {
|
||||
return Number.isFinite(value) && value > 0 ? Math.floor(value) : DEFAULT_DOWNSTREAM_MAX_RETRIES;
|
||||
}
|
||||
|
||||
function downstreamPendingTimeoutHours() {
|
||||
const value = Number(process.env.CMPP_DOWNSTREAM_PENDING_TIMEOUT_HOURS ?? DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS);
|
||||
return Number.isFinite(value) && value > 0 ? value : DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS;
|
||||
}
|
||||
|
||||
function downstreamControlFailureMessage(result: GatewayControlDeliveryResult) {
|
||||
const reason = String(result.errorMessage ?? '').trim();
|
||||
const code = String(result.reasonCode ?? '').trim();
|
||||
if (reason && code) return `${reason} (${code})`;
|
||||
if (reason) return reason;
|
||||
if (code) return `Gateway 未完成下游投递 (${code})`;
|
||||
return 'Gateway 未完成下游投递，等待自动重试';
|
||||
}
|
||||
|
||||
function parseImportRows(content: string, delimiter?: ',' | '\t') {
|
||||
const normalized = content.replace(/^\uFEFF/, '');
|
||||
const lines = normalized.split(/\r?\n/).filter((line) => line.trim().length > 0);
|
||||
|
||||
@@ -105,7 +105,12 @@ describe('TenantsService', () => {
|
||||
}));
|
||||
expect(prisma.accountTransaction.groupBy).toHaveBeenCalledWith(expect.objectContaining({
|
||||
by: ['tenantId'],
|
||||
where: expect.objectContaining({ transactionType: 'refunded' }),
|
||||
where: expect.objectContaining({
|
||||
OR: [
|
||||
{ transactionType: 'refunded' },
|
||||
{ transactionType: 'released', relatedType: 'sms_message_record' },
|
||||
],
|
||||
}),
|
||||
_sum: { amountCents: true },
|
||||
}));
|
||||
});
|
||||
|
||||
@@ -59,7 +59,7 @@ export class TenantsService {
|
||||
}),
|
||||
this.prisma.accountTransaction.groupBy({
|
||||
by: ['tenantId'],
|
||||
where: { transactionType: 'refunded', createdAt: { gte: sinceToday } },
|
||||
where: returnedTransactionWhere(sinceToday),
|
||||
_sum: { amountCents: true },
|
||||
}),
|
||||
]);
|
||||
@@ -201,6 +201,16 @@ function startOfToday() {
|
||||
return date;
|
||||
}
|
||||
|
||||
function returnedTransactionWhere(since: Date): Prisma.AccountTransactionWhereInput {
|
||||
return {
|
||||
createdAt: { gte: since },
|
||||
OR: [
|
||||
{ transactionType: 'refunded' },
|
||||
{ transactionType: 'released', relatedType: 'sms_message_record' },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
function generateTenantCode(data: CreateTenantDto) {
|
||||
const source = data.creditCode?.trim() || data.name.trim();
|
||||
const normalized = source.replace(/[^\da-zA-Z]/g, '').toLowerCase();
|
||||
|
||||
@@ -279,7 +279,10 @@
|
||||
- 已实现下游投递失败审计与人工重投第一版：运营端后端与页面可分页查看 `CmppDownstreamDelivery` 的 pending/awaiting_ack/failed/unconfirmed/rejected/delivered 记录，支持按状态、类型、应用和关键字筛选，并可对非 `awaiting_ack` 记录执行人工重投，真实调用 Gateway `/downstream/receipt` 或 `/downstream/uplink`。主记录必须分开保存自动重试次数 `retryCount`、人工重投次数 `manualRetryCount` 和最近人工重投时间 `lastRetriedAt`，操作日志保留重投前状态与自动重试次数。
|
||||
- 已实现下游投递批量重投第一版：运营端可在当前页勾选多条 `pending/failed` 下游投递记录，调用真实批量接口逐条重投并返回成功/失败汇总，不允许用前端循环假装成功。
|
||||
- 下游投递的 `pending` 展示必须结合真实尝试字段：自动与人工次数均为 0 时显示“待首次投递”，`retryCount > 0` 时显示“等待自动重试”，`manualRetryCount > 0` 时显示“人工重投排队中”。人工重投可重置新一轮自动重试预算，但不得把记录伪装成从未投递。
|
||||
- 已实现下游投递告警统一口径：运营看板、侧栏通知、下游投递 Dashboard 和应用告警排行必须基于同一组真实 `CmppDownstreamDelivery` 条件统计：`pending` 超过积压阈值、`awaiting_ack` 超过 `ackDeadlineAt`，以及最近失败窗口内的 `failed/unconfirmed/rejected`。默认积压阈值为 10 分钟，最近失败窗口为 1 小时，可分别通过 `CMPP_DOWNSTREAM_ALERT_PENDING_MINUTES` 和 `CMPP_DOWNSTREAM_ALERT_RECENT_FAILED_HOURS` 覆盖。
|
||||
- 下游投递不得无限停留在 `pending`：Gateway 对未写出的结果必须返回明确的 `retryable/reasonCode/errorMessage`。状态回执在原消息映射已经丢失且缺少 `submitSequenceId` 时属于不可恢复错误，立即转为 `failed` 并保留失败原因和操作日志；客户端暂时离线属于可恢复错误，按既有重试次数与指数退避处理。所有 pending 从创建时间或最近人工重投时间起最多保留 72 小时，可通过 `CMPP_DOWNSTREAM_PENDING_TIMEOUT_HOURS` 调整，超时后自动终结为 `failed`。
|
||||
- 新产生的客户侧状态回执投递 payload 必须保存短信原始 `cmppSubmitSequenceId`，使 Gateway 重启后仍能结合账号和平台消息 ID 重建与原 `CMPP_SUBMIT_RESP` 一致的非 0 `Msg_Id`；不得用伪造或为 0 的 `Msg_Id` 冒充成功投递。
|
||||
- 已实现下游投递告警统一口径：运营看板、下游投递 Dashboard 和应用告警排行必须基于同一组真实 `CmppDownstreamDelivery` 条件统计：`pending` 超过积压阈值、`awaiting_ack` 超过 `ackDeadlineAt`，以及最近失败窗口内的 `failed/unconfirmed/rejected`。默认积压阈值为 10 分钟，最近失败窗口为 1 小时，可分别通过 `CMPP_DOWNSTREAM_ALERT_PENDING_MINUTES` 和 `CMPP_DOWNSTREAM_ALERT_RECENT_FAILED_HOURS` 覆盖。右上角“待审核任务”仅汇总审核业务，不显示下游投递告警菜单或数字。
|
||||
- pending 积压告警必须从本轮排队开始时间计算：从未人工重投的记录使用 `createdAt`，人工重投记录使用 `lastRetriedAt`；刚执行人工重投的记录不得因旧 `createdAt` 立即重新计入积压告警。
|
||||
- 已实现下游投递 Dashboard 第一版：运营端“下游投递记录”页面顶部新增真实聚合总览，直接按 `tenantId/applicationId/deliveryType` 统计投递总量、pending/awaiting_ack/delivered/failed/unconfirmed/rejected、积压告警、ACK 超时告警、按类型分布、重试压力分布和应用告警排行，数据源必须来自 `CmppDownstreamDelivery`，不能靠前端本地汇总。应用告警排行只统计满足统一告警时间窗的记录，不得将新创建的 `pending` 或超出最近窗口的历史失败永久累加为告警。
|
||||
- 已实现下游连接映射持久化第一步：Gateway 在客户 CMPP 账号 bind 成功、下游 submit 建链和回执/上行下发时，会把账号在线状态、实例标识、最近活跃时间写入 Redis presence；该状态不再只保留在 Gateway 进程内存中，为后续“Gateway 重启后的 pending 恢复”提供外部状态基础。
|
||||
- 已实现下游连接映射持久化第二步：Gateway 启动时会读取 Redis presence 与当前内存在线账号，形成“恢复候选视图”，并通过控制面 `GET /downstream/recovery-candidates` 暴露候选账号列表，供后续恢复逻辑与运维排查使用；本阶段仍不等同于自动恢复 pending 投递。
|
||||
@@ -521,6 +524,7 @@
|
||||
- 计费规则：支持按短信计费条数和企业应用单价计算费用；发送准入额度为现金余额加授信额度。
|
||||
- 账务流水必须与短信记录形成可追溯关系，支持对账导出。
|
||||
- 最终失败、超时失败需要退费。
|
||||
- “今日返还金额”按当天实际恢复到企业可用余额的消息级流水汇总：包含已扣费后的 `refunded`，以及提交前失败后按 `sms_message_record` 释放的 `released`；任务冻结转扣费过程中按 `sms_batch_task` 产生的内部释放不得计入返还。
|
||||
- 计费口径可配置为按提交成功计费或按回执成功计费。
|
||||
|
||||
### 5.20 数据保存与清理
|
||||
@@ -1346,6 +1350,7 @@
|
||||
- 原“剩余条数”改为“剩余余额”。
|
||||
- 原“近24小时成功率”改为“今日发送条数和今日成功率”。
|
||||
- 新增“今日消费金额”和“今日返还金额”展示。
|
||||
- “今日返还金额”展示当天真实返还到可用余额的金额，包含消息级冻结释放和扣费后退款，排除任务冻结转扣费产生的内部释放流水，金额统一保留三位小数。
|
||||
3. 运营端增加系统日志。
|
||||
- 运营端可查询全平台系统日志。
|
||||
- 日志需支持按企业、模块、级别、操作人、资源 ID 或详情定位。
|
||||
|
||||
@@ -36,6 +36,7 @@
|
||||
| TC-ADMIN-TEMPLATE-0715-06 | P1 | 将光标置于模板中间并插入推荐/自定义变量 | 变量在光标或选区处插入，原选区被替换，光标停在变量后；运营端和客户端一致。 |
|
||||
| TC-ADMIN-RECORD-0715-07 | P1 | 查看桌面/窄屏短信记录及失败详情 | 卡片不产生页面横向滚动，信息分组清晰；失败原因独立突出；详情仍读取真实 submit、receipt 和分片审计 API。 |
|
||||
| TC-ADMIN-MISC-0715-08 | P2 | 查看签名引流信息、零待审核通知和充值弹窗 | 使用“引流信息”标题且无提交时间；0 为黑字灰底；充值弹窗无操作人字段。 |
|
||||
| TC-ADMIN-NOTICE-0715-09 | P1 | 制造下游投递告警后打开右上角“待审核任务” | 待审核总数和菜单仅包含五类真实审核任务，不出现“下游投递告警”；运营看板和下游投递页仍展示下游告警。 |
|
||||
| 客户 | 正常客户、停用客户、欠费客户、未认证客户、跨租户客户、客户联系人和开票资料。 |
|
||||
| 导入文件 | UTF-8 CSV、GBK CSV、TXT、超 20 MB 文件、含空行/重复/非法号码/非法字符文件。 |
|
||||
| 非法内容 | 控制字符、emoji、换行、不可见字符、超长变量、签名外置内容、敏感词内容。 |
|
||||
@@ -1361,14 +1362,14 @@
|
||||
- 优先级：P1
|
||||
- 前置条件：真实 `CmppDownstreamDelivery` 中准备阈值内/外的 `pending`、未超时/已超过 `ackDeadlineAt` 的 `awaiting_ack`、最近窗口内/外的 `failed/unconfirmed/rejected` 及正常 `delivered` 记录，且覆盖多个应用。
|
||||
- 步骤：
|
||||
1. 访问运营端 Dashboard 和右上角通知区域。
|
||||
1. 访问运营端 Dashboard、下游投递记录页和右上角待审核通知区域。
|
||||
2. 调用真实 `/api/admin/operations/dashboard/statistics`，核对返回的下游投递告警聚合。
|
||||
3. 点击“下游投递告警”通知，跳转到下游投递记录页进一步筛查。
|
||||
3. 打开右上角“待审核任务”，核对菜单和总数只包含审核业务。
|
||||
- 预期结果：
|
||||
- Dashboard 返回真实 `downstreamDeliverySummary`，至少包含 `pending/failed/delivered/stalledPending/stalledAck/recentFailed/alertCount`。
|
||||
- `alertCount` 精确等于“超阈值 pending + 超时 awaiting_ack + 最近窗口内 failed/unconfirmed/rejected”，阈值内 pending、未超时 awaiting_ack、历史失败和 delivered 不计入。
|
||||
- 侧栏/首页的“下游投递告警”数量与下游投递 Dashboard 在同一筛选范围下一致，不是前端写死值。
|
||||
- 点击通知后可以进入真实下游投递记录页继续处理。
|
||||
- 运营看板的“下游投递告警”数量与下游投递 Dashboard 在同一筛选范围下一致，不是前端写死值。
|
||||
- 右上角“待审核任务”不出现下游投递告警菜单或数字；下游异常从运营看板或下游投递记录页进入处理。
|
||||
|
||||
### TC-GW-018 下游投递 Dashboard 聚合视图
|
||||
|
||||
@@ -3254,8 +3255,9 @@ npm run verify:phase8
|
||||
| TC-BILLING-009 | 无权限用户、审核员、管理员分别执行充值；大额人工充值不走审批。 | 权限不足被拒绝并写失败日志；有权限用户确认后立即入账；不产生 pending 审批态；充值订单、账户余额、流水和日志同步完成。 |
|
||||
| TC-BILLING-010 | 余额不足发送失败，人工充值后重试发送并模拟 delivered。 | 充值前不扣费；充值后发送成功；冻结、扣费、短信计费记录完整；reconciliation diff 为 0。 |
|
||||
| TC-BILLING-011 | 分别准备 `余额+授信` 为正数、0 和负数的账户，使用相同短信费用发起发送。 | 和为正数时允许发送；和为 0 或负数时提示余额不足。判断公式为 `balanceCents + creditCents > 0`，与本次费用和套餐无关。 |
|
||||
| TC-BILLING-012 | 已扣费短信收到最终失败回执；另一个未提交成功任务只释放冻结。 | 前者只生成一条 refunded 流水并计入“今日返还”；重复回执不重复退款；后者的 released 流水不计入“今日返还”。 |
|
||||
| TC-BILLING-012 | 已扣费短信收到最终失败回执；另一个消息在提交前失败并释放冻结；另准备一笔任务冻结转扣费时的批次级释放。 | 最终失败只生成一条 `refunded` 并计入“今日返还”，重复回执不重复退款；提交前失败生成 `released + relatedType=sms_message_record` 并计入“今日返还”；冻结转扣费的 `released + relatedType=sms_batch_task` 属于内部转换，不计入“今日返还”；客户端和运营端当日金额一致且保留三位小数。 |
|
||||
| TC-BILLING-013 | 准备已提交扣费但 72 小时完全无回执的 `submitted` 短信，以及有 `UNKNOWN` 回执且超过 72 小时的短信；启动 API 定时扫描并模拟重复扫描。 | 两类短信都转为 timeout 并退款；任务进度刷新；同一短信只退款一次；定时扫描默认启用且每 5 分钟执行。 |
|
||||
| TC-SEC-006 | 安装 API 生产依赖并执行 `npm audit`；使用缺文件、多文件、超大文件、超量字段和正常单文件调用认证后的 multipart 上传接口。 | NestJS/Multer/Hono 已升级或锁定到修复版本，生产依赖 audit 为 0；接口只接受一个不超过 20MB 的文件，并限制字段、part、字段名、字段值和 header pair 数量；异常请求返回受控 4xx，正常文件仍写入真实 MinIO 和 `FileObject`。 |
|
||||
|
||||
### 17.6 系统日志细化
|
||||
|
||||
@@ -3370,3 +3372,5 @@ npm run verify:phase8
|
||||
| TC-GW-ACK-003 | 在企业应用中分别关闭“回执自动重试”和“上行短信自动重试”，各制造一次 ACK 超时，再重新开启并创建新投递。 | 关闭只影响对应类型的新投递策略快照；离线后的首次投递仍会在重连时执行；已写出未确认的记录不自动重发；重新开启后新记录按退避策略重试。 |
|
||||
| TC-GW-ACK-004 | 对 `unconfirmed/rejected/failed/delivered` 记录执行单条和批量手工重投。 | 页面提示重复处理风险并二次确认；真实调用 Gateway；`awaiting_ack` 不允许并发重投；重发复用同一业务 Msg_Id。 |
|
||||
| TC-GW-ACK-005 | 客户 Submit 后由业务校验立即生成失败回执，并覆盖在线即时投递、Gateway 重启后恢复投递；另模拟客户端对 `Msg_Id=0` 返回 Result=0。 | 客户收到的第一个响应包必须是对应 `CMPP_SUBMIT_RESP`，之后 Deliver 的 `Msg_Id` 非 0 且与 SubmitResp 完全一致；重启后根据持久化 Submit Sequence_Id 重建同一 Msg_Id；Result=0/Msg_Id=0 不得写为 delivered。 |
|
||||
| TC-GW-ACK-006 | 对一条 Gateway 已丢失原消息映射且 payload 缺少 `submitSequenceId` 的历史状态回执执行人工重投；另对字段完整但客户离线的记录重投。 | 缺少序列号的记录由 Gateway 返回 `retryable=false/MISSING_SUBMIT_SEQUENCE_ID`，NestJS 立即终结为 `failed` 并保存明确原因；客户暂时离线的记录返回 `retryable=true/CLIENT_DISCONNECTED`，按次数上限和指数退避继续处理，不得无限保持 `retryCount=0`。 |
|
||||
| TC-GW-ACK-007 | 构造一条从未重投且 pending 超过 72 小时的记录，以及一条人工重投后尚未满 72 小时但原 `createdAt` 很早的记录，执行自动扫描并查看告警。 | 第一条自动转为 `failed/queue_timeout`；第二条仍保持 pending，终结时间和 10 分钟积压告警均从 `lastRetriedAt` 重新计算，刚重投后不立即告警。 |
|
||||
|
||||
@@ -1,5 +1,27 @@
|
||||
# 第一版系统化测试进度
|
||||
|
||||
## 2026-07-15 依赖安全与今日返还修复（未提交、未部署）
|
||||
|
||||
- 生产数据库只读核查确认目标企业当天共有两笔真实返还：提交前路由失败产生消息级 `released=5` 分，最终失败产生 `refunded=5` 分，正确合计为 10 分（页面应显示 `¥0.100`）。原 Dashboard 和企业列表只聚合 `refunded`，因此少算前一笔并显示 5 分。
|
||||
- 客户端 Dashboard 与运营端企业列表统一按当天 `refunded` 加 `released + relatedType=sms_message_record` 汇总；排除任务冻结转扣费时的 `released + relatedType=sms_batch_task`，避免把内部账务转换误当成返还。
|
||||
- API 依赖升级到 NestJS `11.1.28`、Multer `2.2.0`，并通过 override 将 `@hono/node-server` 固定为 `1.19.13`；`npm --prefix api audit --json` 已由 3 个 moderate、2 个 high 降为 0。
|
||||
- multipart 文件上传补充单文件 20MB、字段数、part 数、字段名/值长度和 header pair 限制，继续使用认证后的真实 NestJS API、MinIO 与 `FileObject` 链路。定向回归 3 suites/21 项通过；完整 API 17 suites/173 项、API build、前端 build、Gateway 全量 Go 测试、Prisma validate/migrate status 和依赖 audit 均通过，43 条 migration 全部齐全。前端仅有既有 Vite chunk size warning。
|
||||
|
||||
## 2026-07-15 下游人工重投无限排队修复（未提交、未部署）
|
||||
|
||||
- 根因是 Gateway 以空的 `DownstreamSendResult` 同时表达“客户暂时离线”和“原消息映射丢失且缺少 Submit Sequence_Id”，NestJS 收到 `sent=false` 后又不登记失败，导致历史状态回执永久保持 `pending/retryCount=0`。
|
||||
- Gateway 控制结果新增 `retryable/reasonCode/errorMessage`：缺少原 `submitSequenceId` 且无法命中内存映射时返回不可恢复的 `MISSING_SUBMIT_SEQUENCE_ID`，NestJS 立即终结为 `failed`；字段完整但客户离线时返回可恢复的 `CLIENT_DISCONNECTED`，失败回调进入真实次数上限与指数退避。
|
||||
- 新状态回执入队补存 `SmsMessageRecord.cmppSubmitSequenceId`，Gateway 控制面允许在当前连接存在时立即恢复原 `Msg_Id`；不生成伪造或为 0 的 Msg_Id。
|
||||
- 自动扫描为 pending 增加默认 72 小时绝对终点（`CMPP_DOWNSTREAM_PENDING_TIMEOUT_HOURS` 可覆盖），从 `createdAt` 或最近 `lastRetriedAt` 计算，超时写 `failed/queue_timeout` 和失败操作日志。积压告警同样以最近人工重投时间重新计时，避免刚重投就因旧创建时间立即告警。
|
||||
- 回归已通过：API 定向 `send-chain/operations` 2 suites、68 项，API 全量 17 suites、173 项，Gateway 全量 Go 测试、Prisma validate、API build、前端 build 和 `git diff --check`；前端仅有既有 Vite chunk size warning。尚未提交、push 或部署，生产原 3 条历史 pending 记录未修改。
|
||||
|
||||
## 2026-07-15 待审核通知移除下游投递告警（未提交、未部署）
|
||||
|
||||
- 生产只读核查确认当前“下游投递告警 3”全部为历史状态回执记录：三条均在人工重投后成为 `pending/manualRetryCount=1/retryCount=0`，因为告警仍按原 `createdAt` 超过 10 分钟判断，立即计入 `stalled_pending`；它们不是待审核任务。
|
||||
- 右上角“待审核任务”移除下游投递告警菜单项及其数字，通知总数只汇总企业认证、短信、模板、签名和引流信息五类审核。运营看板和下游投递记录页面继续保留独立告警展示。
|
||||
- “人工重投排队中”表示记录已由人工重投重置为 `pending` 并累计人工重投次数，正在等待 Gateway 对在线客户写出 Deliver 并取得 `CMPP_DELIVER_RESP`，不代表重投已经成功。
|
||||
- 进一步核查三条历史记录的 `payload.submitSequenceId` 均为空；虽然账号 `910887` 当前 connected 且持续心跳，Gateway 重启后没有原 Submit 的消息映射，也无法安全重建非 0 Msg_Id，因此人工重投没有写出 Deliver，继续保持 pending。前端 build 与 `git diff --check` 通过；应用内 Browser 连接因插件运行时初始化失败，未使用独立 Playwright 或 mock 页面替代。
|
||||
|
||||
## 2026-07-15 当前工作区汇总发布
|
||||
|
||||
- 当前各对话产生的 34 个文件变更已统一提交为 `e47432bc631bf37f4c6a6dcb3576ce3c86370b45` 并 push 到 `origin/main`；提交明确排除 `api/tsconfig.build.tsbuildinfo` 和 `logs/`。发布前确认本地 `main` 与最新 `origin/main` 无分叉，API 全量 17 suites/169 项、API build、前端 build、Gateway 全量 Go 测试、Prisma validate/migrate status 和 `git diff --check` 全部通过，前端仅有既有 Vite chunk size warning。
|
||||
|
||||
@@ -93,6 +93,9 @@ type DownstreamUplink struct {
|
||||
|
||||
type DownstreamSendResult struct {
|
||||
Sent bool `json:"sent"`
|
||||
Retryable bool `json:"retryable"`
|
||||
ReasonCode string `json:"reasonCode,omitempty"`
|
||||
ErrorMessage string `json:"errorMessage,omitempty"`
|
||||
ConnectionID string `json:"connectionId,omitempty"`
|
||||
SequenceID string `json:"sequenceId,omitempty"`
|
||||
MessageID string `json:"messageId,omitempty"`
|
||||
@@ -544,7 +547,20 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe
|
||||
result.DeliveredCount++
|
||||
continue
|
||||
}
|
||||
result.WaitingCount++
|
||||
errorMessage := defaultString(sendResult.ErrorMessage, "gateway did not complete downstream delivery")
|
||||
failureType := "unrecoverable"
|
||||
if sendResult.Retryable {
|
||||
failureType = "send_failed"
|
||||
result.WaitingCount++
|
||||
} else {
|
||||
result.FailedCount++
|
||||
}
|
||||
result.LastError = errorMessage
|
||||
_ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]string{
|
||||
"id": delivery.ID,
|
||||
"errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode),
|
||||
"failureType": failureType,
|
||||
}, nil)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
@@ -776,6 +792,10 @@ func (s Server) recoverPendingCandidates(logger *log.Logger) {
|
||||
status.State = "partial"
|
||||
status.LastError = result.LastError
|
||||
status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
|
||||
case result.FailedCount > 0:
|
||||
status.State = "failed"
|
||||
status.LastError = result.LastError
|
||||
status.FailureCategory = recoveryFailureCategory(status.State, status.LastError, status.LastSkipReason)
|
||||
default:
|
||||
status.State = "success"
|
||||
status.LastError = ""
|
||||
@@ -861,7 +881,7 @@ func PushReceipt(event DownstreamReceipt) (bool, error) {
|
||||
}
|
||||
|
||||
func PushReceiptWithResult(event DownstreamReceipt) (DownstreamSendResult, error) {
|
||||
return pushReceiptWithResult(event, false)
|
||||
return pushReceiptWithResult(event, true)
|
||||
}
|
||||
|
||||
func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (DownstreamSendResult, error) {
|
||||
@@ -870,7 +890,25 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr
|
||||
session = recoverReceiptSession(event)
|
||||
}
|
||||
if session == nil {
|
||||
return DownstreamSendResult{}, nil
|
||||
if event.SubmitSequenceID == 0 {
|
||||
return DownstreamSendResult{
|
||||
Retryable: false,
|
||||
ReasonCode: "MISSING_SUBMIT_SEQUENCE_ID",
|
||||
ErrorMessage: "历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id，系统已终止重投",
|
||||
}, nil
|
||||
}
|
||||
if strings.TrimSpace(event.MessageID) == "" || strings.TrimSpace(event.Account) == "" {
|
||||
return DownstreamSendResult{
|
||||
Retryable: false,
|
||||
ReasonCode: "INVALID_RECEIPT_PAYLOAD",
|
||||
ErrorMessage: "状态回执缺少平台消息 ID 或客户账号，系统已终止重投",
|
||||
}, nil
|
||||
}
|
||||
return DownstreamSendResult{
|
||||
Retryable: true,
|
||||
ReasonCode: "CLIENT_DISCONNECTED",
|
||||
ErrorMessage: "下游客户端当前未连接，等待自动重试",
|
||||
}, nil
|
||||
}
|
||||
stat := strings.TrimSpace(event.RawStatus)
|
||||
if stat == "" {
|
||||
@@ -934,7 +972,11 @@ func PushUplink(event DownstreamUplink) (bool, error) {
|
||||
func PushUplinkWithResult(event DownstreamUplink) (DownstreamSendResult, error) {
|
||||
session := findSession(event.MessageID, event.Account)
|
||||
if session == nil {
|
||||
return DownstreamSendResult{}, nil
|
||||
return DownstreamSendResult{
|
||||
Retryable: true,
|
||||
ReasonCode: "CLIENT_DISCONNECTED",
|
||||
ErrorMessage: "下游客户端当前未连接，等待自动重试",
|
||||
}, nil
|
||||
}
|
||||
content, err := cmpputils.Utf8ToUcs2(event.Content)
|
||||
if err != nil {
|
||||
@@ -952,6 +994,18 @@ func PushUplinkWithResult(event DownstreamUplink) (DownstreamSendResult, error)
|
||||
return sendDownstream(session, deliver, event.DeliveryID)
|
||||
}
|
||||
|
||||
func errorMessageWithCode(message string, code string) string {
|
||||
message = strings.TrimSpace(message)
|
||||
code = strings.TrimSpace(code)
|
||||
if message == "" {
|
||||
message = "gateway did not complete downstream delivery"
|
||||
}
|
||||
if code == "" {
|
||||
return message
|
||||
}
|
||||
return fmt.Sprintf("%s (%s)", message, code)
|
||||
}
|
||||
|
||||
func downstreamDeliverPacket(session *downstreamSession, messageID uint64, destID string, sourceTerminalID string, msgFmt uint8, registerDelivery uint8, content string) cmpp.Packer {
|
||||
if session != nil && (session.protocol == "cmpp20" || session.protocol == "cmpp21") {
|
||||
return &cmpp.Cmpp2DeliverReqPkt{
|
||||
|
||||
@@ -302,6 +302,43 @@ func TestSubmitResponsePrecedesQueuedFailureReceipt(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestReceiptWithoutOriginalSequenceIsUnrecoverable(t *testing.T) {
|
||||
resetDownstreamRegistry()
|
||||
defer resetDownstreamRegistry()
|
||||
|
||||
result, err := PushReceiptWithResult(DownstreamReceipt{
|
||||
DeliveryID: "delivery-history",
|
||||
Account: "100001",
|
||||
MessageID: "MSG-HISTORY",
|
||||
ReceiptStatus: "undelivered",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("push receipt: %v", err)
|
||||
}
|
||||
if result.Sent || result.Retryable || result.ReasonCode != "MISSING_SUBMIT_SEQUENCE_ID" {
|
||||
t.Fatalf("unexpected result: %+v", result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecoverableReceiptWaitsForClientConnection(t *testing.T) {
|
||||
resetDownstreamRegistry()
|
||||
defer resetDownstreamRegistry()
|
||||
|
||||
result, err := PushReceiptWithResult(DownstreamReceipt{
|
||||
DeliveryID: "delivery-retry",
|
||||
Account: "100001",
|
||||
MessageID: "MSG-RETRY",
|
||||
SubmitSequenceID: 77,
|
||||
ReceiptStatus: "delivered",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("push receipt: %v", err)
|
||||
}
|
||||
if result.Sent || !result.Retryable || result.ReasonCode != "CLIENT_DISCONNECTED" {
|
||||
t.Fatalf("unexpected result: %+v", result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *testing.T) {
|
||||
resetDownstreamRegistry()
|
||||
defer resetDownstreamRegistry()
|
||||
@@ -571,8 +608,10 @@ func TestRecoverPendingCandidatesWritesWaitingConnectionStatus(t *testing.T) {
|
||||
_ = json.NewEncoder(w).Encode([]pendingDelivery{{
|
||||
ID: "delivery-1",
|
||||
DeliveryType: "receipt",
|
||||
Payload: json.RawMessage(`{"messageId":"MSG-404","phoneNumber":"13800000001","receiptStatus":"delivered"}`),
|
||||
Payload: json.RawMessage(`{"account":"100010","messageId":"MSG-404","submitSequenceId":77,"phoneNumber":"13800000001","receiptStatus":"delivered"}`),
|
||||
}})
|
||||
case "/api/gateway/events/downstream/failed":
|
||||
w.WriteHeader(http.StatusOK)
|
||||
case "/api/gateway/events/downstream/recovery-status":
|
||||
w.WriteHeader(http.StatusOK)
|
||||
default:
|
||||
|
||||
+1
-1
@@ -240,7 +240,7 @@ export type UserPayload = {
|
||||
export type DashboardResponse = {
|
||||
taskCount: number;
|
||||
messageStatus: Array<{ status: string; _count: { _all: number }; _sum: { amountCents?: number | null; billingUnits?: number | null } }>;
|
||||
today: { sent: number; delivered: number; failed: number; unknown: number; successRate: number; spendCents: number; billingUnits: number };
|
||||
today: { sent: number; delivered: number; failed: number; unknown: number; successRate: number; spendCents: number; returnedCents: number; billingUnits: number };
|
||||
uplinkCount: number;
|
||||
billing: { _count: { _all: number }; _sum: { amountCents?: number | null; billingUnits?: number | null } };
|
||||
transactions: { _count: { _all: number }; _sum: { amountCents?: number | null } };
|
||||
|
||||
@@ -52,7 +52,8 @@ export function ClientHome() {
|
||||
const account = dashboard?.accounts[0];
|
||||
const availableBalance = ((account?.balanceCents ?? 0) + (account?.creditCents ?? 0)) / 100;
|
||||
const todaySpend = (dashboard?.today.spendCents ?? 0) / 100;
|
||||
const todayRefund = Math.abs((dashboard?.transactions._sum.amountCents ?? 0) < 0 ? 0 : dashboard?.transactions._sum.amountCents ?? 0) / 100;
|
||||
const todayRefundCents = Math.max(0, dashboard?.today.returnedCents ?? 0);
|
||||
const todayRefund = todayRefundCents / 100;
|
||||
const balanceBaseline = Math.max(availableBalance + todaySpend - todayRefund, availableBalance, 1);
|
||||
const balancePercent = Math.min(100, Math.round((availableBalance / balanceBaseline) * 100));
|
||||
const recentMessages = useMemo<RecentTaskRow[]>(() => (dashboard?.recentTasks ?? []).map((task) => ({
|
||||
@@ -113,7 +114,7 @@ export function ClientHome() {
|
||||
</div>
|
||||
<div className="surface metric-card">
|
||||
<span>今日返还金额</span>
|
||||
<strong>¥{formatAmount(todayRefund)}</strong>
|
||||
<strong>¥{formatCents(todayRefundCents)}</strong>
|
||||
<small>异常回执与退费返还</small>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -38,16 +38,13 @@ import { AppShell } from '@/layouts/AppShell';
|
||||
export function AdminLayout() {
|
||||
const session = readSession();
|
||||
const [pendingAudits, setPendingAudits] = useState({ enterpriseCertifications: 0, smsAudits: 0, templates: 0, signatures: 0, drainageInfos: 0 });
|
||||
const [downstreamAlertCount, setDownstreamAlertCount] = useState(0);
|
||||
const loadPendingAuditCount = useCallback(() => {
|
||||
adminApi.getDashboard()
|
||||
.then((dashboard) => {
|
||||
setPendingAudits(dashboard.pendingAudits ?? { enterpriseCertifications: 0, smsAudits: 0, templates: 0, signatures: 0, drainageInfos: 0 });
|
||||
setDownstreamAlertCount(dashboard.downstreamDeliverySummary?.alertCount ?? 0);
|
||||
})
|
||||
.catch(() => {
|
||||
setPendingAudits({ enterpriseCertifications: 0, smsAudits: 0, templates: 0, signatures: 0, drainageInfos: 0 });
|
||||
setDownstreamAlertCount(0);
|
||||
});
|
||||
}, []);
|
||||
|
||||
@@ -83,7 +80,6 @@ export function AdminLayout() {
|
||||
{ label: '模板待审', count: pendingAudits.templates, to: '/admin/templates' },
|
||||
{ label: '签名待审', count: pendingAudits.signatures, to: '/admin/signatures' },
|
||||
{ label: '引流信息待审', count: pendingAudits.drainageInfos, to: '/admin/drainage-audits' },
|
||||
{ label: '下游投递告警', count: downstreamAlertCount, to: '/admin/downstream-deliveries' },
|
||||
]}
|
||||
navSections={[
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user