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

feat: harden platform workflows and UI governance

Browse Source
This commit is contained in:
hectorzhao
2026-07-22 14:14:55 +08:00
parent ef957f7daa
commit 0f223f7f91
80 changed files with 4958 additions and 764 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.spec.ts
+305 -12
View File
@@ -84,8 +84,10 @@ function createPrismaMock() {
id: 'tpl-1',
tenantId: 'tenant-1',
applicationId: 'app-1',
signatureId: 'sig-1',
content: 'hello',
auditStatus: 'approved',
signature: { auditStatus: 'approved', reportStatus: 'reporting' },
signature: { id: 'sig-1', name: '【签名】', auditStatus: 'approved', reportStatus: 'reporting' },
}),
findFirst: jest.fn().mockResolvedValue({
id: 'tpl-1',
@@ -112,6 +114,7 @@ function createPrismaMock() {
findFirst: jest.fn().mockResolvedValue(task),
findMany: jest.fn(),
update: jest.fn().mockResolvedValue(task),
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
},
smsApiRequest: {
create: jest.fn().mockResolvedValue({ id: 'request-1' }),
@@ -214,6 +217,9 @@ function createPrismaMock() {
messageId: 'MSG-1',
deliveryType: 'receipt',
retryCount: 0,
manualRetryCount: 0,
status: 'failed',
updatedAt: new Date('2026-07-21T08:00:00.000Z'),
lastError: null,
payload: { account: '100001', messageId: 'MSG-1', phoneNumber: '13800000001', receiptStatus: 'delivered' },
application: { cmppAccount: '100001' },
@@ -257,6 +263,7 @@ function createPrismaMock() {
}),
update: jest.fn().mockResolvedValue({ id: 'dead-1', tenantId: 'tenant-1', streamMessageId: '1710000000000-0', submitId: 'SUB-1', messageId: 'MSG-1' }),
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
findMany: jest.fn().mockResolvedValue([]),
},
gatewayDownstreamRecoveryStatus: {
findUnique: jest.fn().mockResolvedValue(null),
@@ -364,6 +371,10 @@ describe('SendChainService', () => {
id: 'tpl-http', tenantId: 'tenant-1', applicationId: 'app-1', content: '【签名】验证码${code}',
auditStatus: 'approved', signatureId: 'sig-1', signature: { id: 'sig-1', auditStatus: 'approved' },
});
prisma.smsTemplate.findUnique.mockResolvedValue({
id: 'tpl-http', tenantId: 'tenant-1', applicationId: 'app-1', content: '【签名】验证码${code}',
auditStatus: 'approved', signatureId: 'sig-1', signature: { id: 'sig-1', name: '【签名】', auditStatus: 'approved' },
});
await service.createHttpBatchTask({
tenantId: 'tenant-1', applicationId: 'app-1', content: '【签名】验证码123456', phones: ['13800000001'],
@@ -384,16 +395,17 @@ describe('SendChainService', () => {
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
prisma.smsTemplate.findUnique.mockResolvedValue({
id: 'tpl-1', tenantId: 'tenant-1', applicationId: 'app-1', signatureId: 'sig-1', auditStatus: 'approved',
signature: { id: 'sig-1', auditStatus: 'approved', reportStatus: 'reporting' },
content: '【签名】详情请访问 https://a.example/landing',
signature: { id: 'sig-1', name: '【签名】', auditStatus: 'approved', reportStatus: 'reporting' },
});
prisma.smsDrainageInfo.findMany.mockResolvedValue([
{ id: 'drain-short', url: 'https://a.example', updatedAt: new Date('2026-07-01') },
{ id: 'drain-long', url: 'https://a.example/landing', updatedAt: new Date('2026-07-02') },
{ id: 'drain-short', url: 'https://a.example', auditStatus: 'approved', updatedAt: new Date('2026-07-01') },
{ id: 'drain-long', url: 'https://a.example/landing', auditStatus: 'approved', updatedAt: new Date('2026-07-02') },
]);
await service.createBatchTask({
tenantId: 'tenant-1', applicationId: 'app-1', templateId: 'tpl-1',
content: '详情请访问 https://a.example/landing', phones: ['13800000001'],
content: '【签名】详情请访问 https://a.example/landing', phones: ['13800000001'],
});
expect(prisma.smsMessageRecord.createMany).toHaveBeenCalledWith({
@@ -401,6 +413,81 @@ describe('SendChainService', () => {
});
});
it('rejects a task when the submitted content no longer matches the selected approved template', async () => {
const { service, prisma, riskReview } = createService();
prisma.smsTemplate.findUnique.mockResolvedValue({
id: 'tpl-1', tenantId: 'tenant-1', applicationId: 'app-1', signatureId: 'sig-1',
content: '【签名】验证码${code}', auditStatus: 'approved',
signature: { id: 'sig-1', name: '【签名】', auditStatus: 'approved' },
});
await expect(service.createBatchTask({
tenantId: 'tenant-1', applicationId: 'app-1', templateId: 'tpl-1',
content: '【签名】被篡改的正文', phones: ['13800000001'],
})).rejects.toThrow('短信内容与选定的审核模板不匹配');
expect(riskReview.evaluateTask).not.toHaveBeenCalled();
expect(prisma.smsBatchTask.create).not.toHaveBeenCalled();
});
it('rejects free content without an approved leading signature', async () => {
const { service, prisma } = createService();
prisma.smsApplication.findUnique.mockResolvedValue({
id: 'app-1', tenantId: 'tenant-1', status: 'active', interfaceEnabled: true,
customerUnitPrice: 3, queuePriority: 'normal', templateMismatchMode: 'direct_send',
});
prisma.smsSignature.findFirst.mockResolvedValue(null);
await expect(service.createBatchTask({
tenantId: 'tenant-1', applicationId: 'app-1', content: '没有签名的自由内容', phones: ['13800000001'],
})).rejects.toThrow('短信内容未以当前应用已审核通过的签名开头');
});
it('allows signed free content only when the application explicitly uses direct send', async () => {
const { service, prisma } = createService();
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
prisma.smsApplication.findUnique.mockResolvedValue({
id: 'app-1', tenantId: 'tenant-1', status: 'active', interfaceEnabled: true,
customerUnitPrice: 3, queuePriority: 'normal', templateMismatchMode: 'direct_send',
});
prisma.smsSignature.findFirst.mockResolvedValue({ id: 'sig-1', name: '【签名】', auditStatus: 'approved' });
await expect(service.createBatchTask({
tenantId: 'tenant-1', applicationId: 'app-1', content: '【签名】允许直接发送的自由内容', phones: ['13800000001'],
})).resolves.toBeDefined();
expect(prisma.smsMessageRecord.createMany).toHaveBeenCalledWith({
data: [expect.objectContaining({ signatureId: 'sig-1' })],
});
});
it.each(['pending', 'rejected'])('blocks a matched %s drainage URL and preserves the matched resource on rejected records', async (auditStatus) => {
const { service, prisma, riskReview } = createService();
prisma.smsTemplate.findUnique.mockResolvedValue({
id: 'tpl-1', tenantId: 'tenant-1', applicationId: 'app-1', signatureId: 'sig-1',
content: '【签名】详情 https://blocked.example', auditStatus: 'approved',
signature: { id: 'sig-1', name: '【签名】', auditStatus: 'approved' },
});
prisma.smsDrainageInfo.findMany.mockResolvedValue([
{ id: 'drain-blocked', url: 'https://blocked.example', auditStatus, updatedAt: new Date('2026-07-21') },
]);
await expect(service.createBatchTask({
tenantId: 'tenant-1', applicationId: 'app-1', templateId: 'tpl-1',
content: '【签名】详情 https://blocked.example', phones: ['13800000001'],
})).resolves.toBeDefined();
expect(prisma.smsBatchTask.create).toHaveBeenCalledWith({
data: expect.objectContaining({ status: 'rejected', rejectReason: expect.stringContaining(`drain-blocked 当前为 ${auditStatus}`) }),
});
expect(prisma.smsMessageRecord.createMany).toHaveBeenCalledWith({
data: [expect.objectContaining({
drainageInfoId: 'drain-blocked', status: 'rejected', errorMessage: expect.stringContaining(`drain-blocked 当前为 ${auditStatus}`),
})],
});
expect(riskReview.evaluateTask).not.toHaveBeenCalled();
});
it('creates scheduled tasks without immediate enqueue and dispatches due tasks later', async () => {
const { service, prisma, billing } = createService();
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
@@ -437,6 +524,102 @@ describe('SendChainService', () => {
where: { batchTaskId: 'task-1', status: 'scheduled' },
data: { status: 'queued' },
});
expect(prisma.smsBatchTask.updateMany).toHaveBeenCalledWith({
where: expect.objectContaining({ id: 'task-1', status: 'scheduled' }),
data: { status: 'scheduled_dispatching' },
});
});
it('atomically claims a due scheduled task so concurrent scanners only freeze and enqueue once', async () => {
const { service, prisma, billing } = createService();
const dueTask = {
id: 'task-1', tenantId: 'tenant-1', applicationId: 'app-1', templateId: 'tpl-1',
status: 'scheduled', scheduledAt: new Date(Date.now() - 1_000), updatedAt: new Date(Date.now() - 1_000),
};
prisma.smsBatchTask.findMany.mockResolvedValue([dueTask]);
prisma.smsBatchTask.updateMany
.mockResolvedValueOnce({ count: 1 })
.mockResolvedValueOnce({ count: 0 });
prisma.smsMessageRecord.findMany.mockResolvedValue([{ id: 'record-1', amountCents: 3, billingUnits: 1 }]);
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
const results = await Promise.all([
service.dispatchDueScheduledTasks(new Date()),
service.dispatchDueScheduledTasks(new Date()),
]);
expect(results.map((item) => item.dispatched).sort()).toEqual([0, 1]);
expect(billing.freeze).toHaveBeenCalledTimes(1);
expect(service.enqueueBatchTask).toHaveBeenCalledTimes(1);
});
it('recovers a stale claimed task without freezing its balance twice', async () => {
const { service, prisma, billing } = createService();
const now = new Date();
prisma.smsBatchTask.findMany.mockResolvedValue([{
id: 'task-1', tenantId: 'tenant-1', applicationId: 'app-1', templateId: 'tpl-1',
status: 'scheduled_dispatching', scheduledAt: new Date(now.getTime() - 300_000),
updatedAt: new Date(now.getTime() - 300_000),
}]);
prisma.accountTransaction.findFirst.mockResolvedValue({ id: 'frozen-transaction-1' });
prisma.smsMessageRecord.findMany.mockResolvedValue([{ id: 'record-1', amountCents: 3, billingUnits: 1 }]);
service.enqueueBatchTask = jest.fn().mockResolvedValue({ taskId: 'task-1', enqueued: 1 });
await expect(service.dispatchDueScheduledTasks(now)).resolves.toEqual({
dispatched: 1,
results: [{ taskId: 'task-1', status: 'queued', enqueued: 1 }],
});
expect(prisma.smsBatchTask.updateMany).toHaveBeenCalledWith({
where: expect.objectContaining({ id: 'task-1', status: 'scheduled_dispatching', updatedAt: { lt: expect.any(Date) } }),
data: { status: 'scheduled_recovering' },
});
expect(billing.freeze).not.toHaveBeenCalled();
});
it('keeps a zero-fee task recoverable when queue enqueue fails after preparation', async () => {
const { service, prisma } = createService();
prisma.smsBatchTask.findMany.mockResolvedValue([{
id: 'task-free', tenantId: 'tenant-1', applicationId: 'app-1', templateId: 'tpl-1', status: 'scheduled',
}]);
prisma.smsMessageRecord.findMany.mockResolvedValue([{ id: 'record-free', amountCents: 0, billingUnits: 1 }]);
service.enqueueBatchTask = jest.fn().mockRejectedValue(new Error('Redis unavailable'));
await expect(service.dispatchDueScheduledTasks(new Date())).resolves.toEqual({
dispatched: 0,
results: [{ taskId: 'task-free', status: 'retrying', reason: 'Redis unavailable' }],
});
expect(prisma.smsBatchTask.update).toHaveBeenCalledWith({
where: { id: 'task-free' },
data: { status: 'scheduled_dispatching', rejectReason: '调度将在超时后恢复：Redis unavailable' },
});
});
it('automatically scans and dispatches due scheduled tasks after application startup', async () => {
jest.useFakeTimers();
const previousReceiptEnabled = process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED;
const previousScheduledEnabled = process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED;
const previousScheduledInterval = process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS;
const { service } = createService();
const dispatch = jest.spyOn(service, 'dispatchDueScheduledTasks').mockResolvedValue({ dispatched: 0, results: [] });
try {
process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED = 'false';
process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED = 'true';
process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS = '60000';
service.onModuleInit();
await jest.advanceTimersByTimeAsync(1_000);
expect(dispatch).toHaveBeenCalledTimes(1);
await service.onModuleDestroy();
} finally {
if (previousReceiptEnabled === undefined) delete process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED;
else process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED = previousReceiptEnabled;
if (previousScheduledEnabled === undefined) delete process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED;
else process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED = previousScheduledEnabled;
if (previousScheduledInterval === undefined) delete process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS;
else process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS = previousScheduledInterval;
jest.useRealTimers();
}
});
it('cancels scheduled tasks before dispatch', async () => {
@@ -1155,7 +1338,7 @@ describe('SendChainService', () => {
}));
expect(prisma.gatewaySubmitDeadLetter.updateMany).toHaveBeenCalledWith({
where: {
status: { in: ['pending', 'requeued'] },
status: { in: ['pending', 'requeueing', 'requeue_recovering', 'requeued'] },
OR: [{ submitId: 'SUB-1' }, { messageId: 'MSG-1' }],
},
data: expect.objectContaining({ status: 'resolved', resolvedStatus: 'accepted' }),
@@ -1734,13 +1917,16 @@ describe('SendChainService', () => {
operatorId: 'user-1',
});
expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledWith(expect.objectContaining({ submitId: 'SUB-1' }));
expect(service['publishGatewaySubmitCommand']).toHaveBeenCalledWith(
expect.objectContaining({ submitId: 'SUB-1' }),
'gateway:submit:requeue:dead-1:1',
);
expect(prisma.gatewaySubmitDeadLetter.updateMany).toHaveBeenCalledWith({
where: { id: 'dead-1', status: 'pending' },
data: { status: 'requeueing' },
});
expect(prisma.gatewaySubmitDeadLetter.update).toHaveBeenCalledWith({
where: { id: 'dead-1' },
expect(prisma.gatewaySubmitDeadLetter.updateMany).toHaveBeenCalledWith({
where: { id: 'dead-1', status: 'requeueing' },
data: expect.objectContaining({
status: 'requeued',
manualRetryCount: { increment: 1 },
@@ -1778,6 +1964,103 @@ describe('SendChainService', () => {
expect(service['publishGatewaySubmitCommand']).not.toHaveBeenCalled();
});
it('does not reset a resolved submit exception when Gateway repeats the same dead-letter report', async () => {
const { service, prisma } = createService();
await service.recordGatewaySubmitDeadLetter({
streamMessageId: '1710000000000-0',
failureCode: 'SUBMIT_PROCESSING_FAILED',
failureMessage: 'repeated report',
attempts: 3,
maxAttempts: 3,
});
expect(prisma.gatewaySubmitDeadLetter.upsert).toHaveBeenCalledWith(expect.objectContaining({
update: expect.not.objectContaining({ status: expect.anything(), resolvedAt: expect.anything(), resolvedStatus: expect.anything() }),
}));
});
it('recovers a stale submit requeue with the same Redis idempotency key', async () => {
const { service, prisma } = createService();
const stale = {
...await prisma.gatewaySubmitDeadLetter.findUnique({ where: { id: 'dead-1' } }),
status: 'requeueing',
updatedAt: new Date('2026-07-21T07:00:00.000Z'),
};
prisma.gatewaySubmitDeadLetter.findMany.mockResolvedValue([stale]);
const publish = jest.spyOn(service as any, 'publishGatewaySubmitCommand').mockResolvedValue('1710000001000-0');
await expect(service.recoverStaleGatewaySubmitRequeues(new Date('2026-07-21T08:00:00.000Z'))).resolves.toEqual({ recovered: 1, failed: 0 });
expect(prisma.gatewaySubmitDeadLetter.updateMany).toHaveBeenNthCalledWith(1, {
where: { id: 'dead-1', status: 'requeueing', updatedAt: stale.updatedAt },
data: { status: 'requeue_recovering' },
});
expect(publish).toHaveBeenCalledWith(
stale.commandPayload,
'gateway:submit:requeue:dead-1:1',
);
expect(prisma.gatewaySubmitDeadLetter.updateMany).toHaveBeenNthCalledWith(2, {
where: { id: 'dead-1', status: 'requeue_recovering' },
data: expect.objectContaining({
status: 'requeued',
manualRetryCount: { increment: 1 },
lastRetryStreamId: '1710000001000-0',
}),
});
});
it('atomically claims a downstream manual requeue so concurrent requests only call Gateway once', async () => {
const { service, prisma } = createService();
service['postGatewayControl'] = jest.fn().mockResolvedValue({ sent: true, sequenceId: '11', messageId: '22' });
prisma.cmppDownstreamDelivery.updateMany
.mockResolvedValueOnce({ count: 1 })
.mockResolvedValueOnce({ count: 0 });
const results = await Promise.allSettled([
service.requeueDownstreamDelivery('delivery-1'),
service.requeueDownstreamDelivery('delivery-1'),
]);
expect(results.filter((result) => result.status === 'fulfilled')).toHaveLength(1);
expect(results.filter((result) => result.status === 'rejected')).toHaveLength(1);
expect(service['postGatewayControl']).toHaveBeenCalledTimes(1);
expect(prisma.cmppDownstreamDelivery.updateMany).toHaveBeenCalledWith({
where: {
id: 'delivery-1',
status: 'failed',
updatedAt: new Date('2026-07-21T08:00:00.000Z'),
},
data: expect.objectContaining({
status: 'manual_requeueing',
manualRetryCount: { increment: 1 },
}),
});
});
it('recovers a stale downstream manual-requeue claim into the Gateway pending path', async () => {
const { service, prisma } = createService();
const updatedAt = new Date('2026-07-21T07:00:00.000Z');
prisma.cmppDownstreamDelivery.findMany.mockResolvedValue([{ id: 'delivery-1', updatedAt }]);
await expect(service.recoverStaleDownstreamManualRequeues(new Date('2026-07-21T08:00:00.000Z'))).resolves.toEqual({ recovered: 1 });
expect(prisma.cmppDownstreamDelivery.findMany).toHaveBeenCalledWith({
where: { status: 'manual_requeueing', updatedAt: { lt: expect.any(Date) } },
select: { id: true, updatedAt: true },
orderBy: { updatedAt: 'asc' },
take: 500,
});
expect(prisma.cmppDownstreamDelivery.updateMany).toHaveBeenCalledWith({
where: { id: 'delivery-1', status: 'manual_requeueing', updatedAt },
data: {
status: 'pending',
nextRetryAt: null,
lastError: '人工重投进程中断，已恢复为待投递',
},
});
});
it('records gateway downstream recovery statuses', async () => {
const { service, prisma } = createService();
@@ -2053,6 +2336,7 @@ describe('SendChainService', () => {
status: 'failed',
retryCount: 3,
manualRetryCount: 1,
updatedAt: new Date('2026-07-21T08:00:00.000Z'),
lastError: 'downstream client is not connected',
payload: { account: '100001', messageId: 'MSG-1', phoneNumber: '13800000001', receiptStatus: 'delivered' },
application: { cmppAccount: '100001' },
@@ -2086,9 +2370,14 @@ describe('SendChainService', () => {
resourceId: 'delivery-1',
}),
});
expect(prisma.cmppDownstreamDelivery.update).toHaveBeenCalledWith(expect.objectContaining({
expect(prisma.cmppDownstreamDelivery.updateMany).toHaveBeenCalledWith(expect.objectContaining({
where: expect.objectContaining({
id: 'delivery-1',
status: 'failed',
updatedAt: new Date('2026-07-21T08:00:00.000Z'),
}),
data: expect.objectContaining({
status: 'pending',
status: 'manual_requeueing',
retryCount: 0,
manualRetryCount: { increment: 1 },
lastRetriedAt: expect.any(Date),
@@ -2122,7 +2411,7 @@ describe('SendChainService', () => {
});
await expect(service.requeueDownstreamDelivery('delivery-1')).rejects.toThrow('该记录正在等待客户端确认，不允许并发重投');
expect(prisma.cmppDownstreamDelivery.update).not.toHaveBeenCalled();
expect(prisma.cmppDownstreamDelivery.updateMany).not.toHaveBeenCalled();
expect(service['postGatewayControl']).not.toHaveBeenCalled();
});
@@ -2231,11 +2520,13 @@ describe('SendChainService', () => {
it('starts the automatic receipt-timeout scan after application startup', async () => {
jest.useFakeTimers();
const previousEnabled = process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED;
const previousScheduledEnabled = process.env.SMS_SCHEDULED_DISPATCH_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';
process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED = 'false';
service.onModuleInit();
await jest.advanceTimersByTimeAsync(60_000);
expect(scan).toHaveBeenCalledWith({});
@@ -2244,6 +2535,8 @@ describe('SendChainService', () => {
} finally {
if (previousEnabled === undefined) delete process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED;
else process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED = previousEnabled;
if (previousScheduledEnabled === undefined) delete process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED;
else process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED = previousScheduledEnabled;
jest.useRealTimers();
}
});
api/src/send-chain/send-chain.service.ts
+346 -60
View File
@@ -255,6 +255,12 @@ 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;
const DEFAULT_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS = 5_000;
const DEFAULT_SCHEDULED_DISPATCH_STALE_MS = 2 * 60_000;
const SCHEDULED_DISPATCH_INITIAL_DELAY_MS = 1_000;
const DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS = 2 * 60_000;
const DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS = 2 * 60_000;
const GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS = 30 * 24 * 60 * 60;
const BULLMQ_PRIORITY: Record<QueuePriority, number> = {
priority: 1,
normal: 100,
@@ -270,6 +276,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
private receiptTimeoutInitialTimer?: ReturnType<typeof setTimeout>;
private receiptTimeoutIntervalTimer?: ReturnType<typeof setInterval>;
private receiptTimeoutScanRunning = false;
private scheduledDispatchInitialTimer?: ReturnType<typeof setTimeout>;
private scheduledDispatchIntervalTimer?: ReturnType<typeof setInterval>;
private scheduledDispatchScanRunning = false;
constructor(
private readonly prisma: PrismaService,
@@ -291,11 +300,25 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
);
this.receiptTimeoutIntervalTimer.unref?.();
}
if (process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED !== 'false') {
this.scheduledDispatchInitialTimer = setTimeout(
() => void this.runScheduledDispatchScan(),
SCHEDULED_DISPATCH_INITIAL_DELAY_MS,
);
this.scheduledDispatchInitialTimer.unref?.();
this.scheduledDispatchIntervalTimer = setInterval(
() => void this.runScheduledDispatchScan(),
positiveInteger(process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS, DEFAULT_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS),
);
this.scheduledDispatchIntervalTimer.unref?.();
}
}
async onModuleDestroy() {
if (this.receiptTimeoutInitialTimer) clearTimeout(this.receiptTimeoutInitialTimer);
if (this.receiptTimeoutIntervalTimer) clearInterval(this.receiptTimeoutIntervalTimer);
if (this.scheduledDispatchInitialTimer) clearTimeout(this.scheduledDispatchInitialTimer);
if (this.scheduledDispatchIntervalTimer) clearInterval(this.scheduledDispatchIntervalTimer);
await this.worker?.close();
await this.sendQueue?.close();
await this.gatewayQueue?.close();
@@ -307,21 +330,23 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
const schedule = parseSchedule(data);
await this.validateSendResources(data.tenantId, data.applicationId, data.templateId);
const [messageClassification, unitPrice, queuePriority, accessNumber] = await Promise.all([
this.resolveTemplateMessageClassification(data.templateId, data.content),
this.resolveTemplateMessageClassification(data.tenantId, data.applicationId, data.templateId, data.content),
this.resolveUnitPrice(data.tenantId, data.applicationId),
this.resolveQueuePriority(data.tenantId, data.applicationId),
this.resolveApplicationAccessNumber(data.tenantId, data.applicationId),
]);
const risk = await this.riskReview.evaluateTask({
tenantId: data.tenantId,
applicationId: data.applicationId,
templateId: data.templateId,
content: data.content,
category: data.category,
phones,
variables: data.variables,
createdById: data.createdById,
});
const risk = messageClassification.rejectionReason
? { status: 'rejected', reason: messageClassification.rejectionReason, task: null }
: await this.riskReview.evaluateTask({
tenantId: data.tenantId,
applicationId: data.applicationId,
templateId: data.templateId,
content: data.content,
category: data.category,
phones,
variables: messageClassification.variables ?? data.variables,
createdById: data.createdById,
});
const billing = this.billing.estimateSmsCost({
tenantId: data.tenantId,
applicationId: data.applicationId,
@@ -703,33 +728,64 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
async dispatchDueScheduledTasks(now = new Date()) {
const staleCutoff = new Date(now.getTime() - positiveInteger(
process.env.SMS_SCHEDULED_DISPATCH_STALE_MS,
DEFAULT_SCHEDULED_DISPATCH_STALE_MS,
));
const tasks = await this.prisma.smsBatchTask.findMany({
where: { status: 'scheduled', scheduledAt: { lte: now } },
where: {
OR: [
{ status: 'scheduled', scheduledAt: { lte: now } },
{ status: { in: ['scheduled_dispatching', 'scheduled_recovering'] }, updatedAt: { lt: staleCutoff } },
],
},
orderBy: { scheduledAt: 'asc' },
});
const results: Array<{ taskId: string; status: string; enqueued?: number; reason?: string }> = [];
for (const task of tasks) {
const candidateStatus = task.status || 'scheduled';
const claimedStatus = candidateStatus === 'scheduled_dispatching' ? 'scheduled_recovering' : 'scheduled_dispatching';
const claimed = await this.prisma.smsBatchTask.updateMany({
where: {
id: task.id,
status: candidateStatus,
...(candidateStatus === 'scheduled' ? {} : { updatedAt: { lt: staleCutoff } }),
},
data: { status: claimedStatus },
});
if (claimed.count !== 1) continue;
let reservationEstablished = false;
let dispatchPrepared = false;
try {
await this.validateSendResources(task.tenantId, task.applicationId ?? undefined, task.templateId ?? undefined);
const messages = await this.prisma.smsMessageRecord.findMany({
where: { batchTaskId: task.id, status: 'scheduled' },
where: { batchTaskId: task.id, status: { in: ['scheduled', 'queued'] } },
select: { id: true, amountCents: true, billingUnits: true },
take: 100000,
});
const amountCents = messages.reduce((sum, message) => sum + moneyToNumber(message.amountCents), 0);
const accountCheck = await this.billing.checkAccount({ tenantId: task.tenantId, amountCents });
if (!accountCheck.canSend) {
throw new BadRequestException('定时任务到点时企业账户余额不足');
}
if (amountCents > 0) {
await this.billing.freeze({
tenantId: task.tenantId,
amountCents,
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: '定时任务到点冻结',
});
const existingReservation = await this.prisma.accountTransaction.findFirst({
where: { tenantId: task.tenantId, transactionType: 'frozen', relatedType: 'sms_batch_task', relatedId: task.id },
select: { id: true },
});
reservationEstablished = Boolean(existingReservation);
if (!reservationEstablished) {
const accountCheck = await this.billing.checkAccount({ tenantId: task.tenantId, amountCents });
if (!accountCheck.canSend) {
throw new BadRequestException('定时任务到点时企业账户余额不足');
}
if (amountCents > 0) {
await this.billing.freeze({
tenantId: task.tenantId,
amountCents,
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: '定时任务到点冻结',
});
reservationEstablished = true;
}
}
dispatchPrepared = true;
await this.prisma.smsMessageRecord.updateMany({
where: { batchTaskId: task.id, status: 'scheduled' },
data: { status: 'queued' },
@@ -738,6 +794,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
results.push({ taskId: task.id, status: 'queued', enqueued: enqueued.enqueued });
} catch (error) {
const reason = error instanceof Error ? error.message : '定时任务到点执行失败';
if (reservationEstablished || dispatchPrepared) {
await this.prisma.smsBatchTask.update({
where: { id: task.id },
data: { status: claimedStatus, rejectReason: `调度将在超时后恢复：${reason}` },
});
results.push({ taskId: task.id, status: 'retrying', reason });
continue;
}
await this.prisma.smsMessageRecord.updateMany({
where: { batchTaskId: task.id, status: 'scheduled' },
data: { status: 'rejected', errorMessage: reason },
@@ -752,6 +816,18 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return { dispatched: results.filter((result) => result.status === 'queued').length, results };
}
private async runScheduledDispatchScan() {
if (this.scheduledDispatchScanRunning) return;
this.scheduledDispatchScanRunning = true;
try {
await this.dispatchDueScheduledTasks();
} catch (error) {
this.logger.error(`Scheduled SMS dispatch scan failed: ${error instanceof Error ? error.message : String(error)}`);
} finally {
this.scheduledDispatchScanRunning = false;
}
}
startWorker() {
if (this.worker) {
return { status: 'already_started' };
@@ -852,7 +928,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
await this.prisma.gatewaySubmitDeadLetter.updateMany({
where: {
status: { in: ['pending', 'requeued'] },
status: { in: ['pending', 'requeueing', 'requeue_recovering', 'requeued'] },
OR: [
data.submitId ? { submitId: data.submitId } : undefined,
data.messageId ? { messageId: data.messageId } : undefined,
@@ -1189,15 +1265,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
traceId: data.traceId,
messageId: data.messageId,
submitId: data.submitId,
status: 'pending',
failureCode: data.failureCode,
failureMessage: data.failureMessage,
attempts: data.attempts,
maxAttempts: data.maxAttempts,
commandPayload: data.commandPayload as Prisma.InputJsonValue | undefined,
rawPayload: data.rawPayload,
resolvedAt: null,
resolvedStatus: null,
},
create: {
streamMessageId: data.streamMessageId,
@@ -1369,19 +1442,23 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (claimed.count !== 1) {
throw new BadRequestException('该提交异常已被其他操作处理，请刷新后重试');
}
const requeueKey = gatewaySubmitRequeueKey(deadLetter.id, deadLetter.manualRetryCount + 1);
let retryStreamMessageId: string;
try {
const publishedStreamMessageId = await this.publishGatewaySubmitCommand(deadLetter.commandPayload);
const publishedStreamMessageId = await this.publishGatewaySubmitCommand(deadLetter.commandPayload, requeueKey);
if (!publishedStreamMessageId) {
throw new Error('Gateway提交异常重新入队未返回Stream消息编号');
}
retryStreamMessageId = publishedStreamMessageId;
} catch (error) {
await this.prisma.gatewaySubmitDeadLetter.update({ where: { id }, data: { status: 'pending' } });
await this.prisma.gatewaySubmitDeadLetter.updateMany({
where: { id, status: 'requeueing' },
data: { status: 'pending' },
});
throw error;
}
const updated = await this.prisma.gatewaySubmitDeadLetter.update({
where: { id },
const finalized = await this.prisma.gatewaySubmitDeadLetter.updateMany({
where: { id, status: 'requeueing' },
data: {
status: 'requeued',
manualRetryCount: { increment: 1 },
@@ -1389,6 +1466,13 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
lastRetriedAt: new Date(),
},
});
const updated = await this.prisma.gatewaySubmitDeadLetter.findUnique({ where: { id } });
if (!updated) {
throw new NotFoundException('Gateway提交异常记录不存在');
}
if (finalized.count !== 1 && updated.status !== 'resolved') {
throw new BadRequestException('该提交异常状态已变化，请刷新后确认处理结果');
}
await this.prisma.operationLog.create({
data: {
tenantId: updated.tenantId ?? undefined,
@@ -1409,6 +1493,69 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return updated;
}
async recoverStaleGatewaySubmitRequeues(now = new Date()) {
const staleCutoff = new Date(now.getTime() - positiveInteger(
process.env.GATEWAY_SUBMIT_REQUEUE_STALE_MS,
DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS,
));
const stale = await this.prisma.gatewaySubmitDeadLetter.findMany({
where: { status: 'requeueing', updatedAt: { lt: staleCutoff } },
orderBy: { updatedAt: 'asc' },
take: 100,
});
let recovered = 0;
let failed = 0;
for (const deadLetter of stale) {
if (!deadLetter.commandPayload || !isObjectRecord(deadLetter.commandPayload)) {
await this.prisma.gatewaySubmitDeadLetter.updateMany({
where: { id: deadLetter.id, status: 'requeueing', updatedAt: deadLetter.updatedAt },
data: { status: 'pending' },
});
failed += 1;
continue;
}
const claimed = await this.prisma.gatewaySubmitDeadLetter.updateMany({
where: { id: deadLetter.id, status: 'requeueing', updatedAt: deadLetter.updatedAt },
data: { status: 'requeue_recovering' },
});
if (claimed.count !== 1) continue;
try {
const requeueKey = gatewaySubmitRequeueKey(deadLetter.id, deadLetter.manualRetryCount + 1);
const retryStreamMessageId = await this.publishGatewaySubmitCommand(deadLetter.commandPayload, requeueKey);
if (!retryStreamMessageId) throw new Error('Gateway提交异常恢复未返回Stream消息编号');
const finalized = await this.prisma.gatewaySubmitDeadLetter.updateMany({
where: { id: deadLetter.id, status: 'requeue_recovering' },
data: {
status: 'requeued',
manualRetryCount: { increment: 1 },
lastRetryStreamId: retryStreamMessageId,
lastRetriedAt: new Date(),
},
});
if (finalized.count === 1) {
recovered += 1;
await this.prisma.operationLog.create({
data: {
tenantId: deadLetter.tenantId ?? undefined,
action: 'gateway.submit_dead_letter_requeue_recovered',
resource: 'gateway_submit_dead_letter',
resourceId: deadLetter.id,
detail: { retryStreamMessageId, requeueKey },
},
});
}
} catch (error) {
failed += 1;
await this.prisma.gatewaySubmitDeadLetter.updateMany({
where: { id: deadLetter.id, status: 'requeue_recovering' },
data: { status: 'requeueing' },
});
this.logger.error(`Gateway submit requeue recovery failed for ${deadLetter.id}: ${error instanceof Error ? error.message : String(error)}`);
}
}
return { recovered, failed };
}
async requeueDownstreamDelivery(id: string) {
const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({
where: { id },
@@ -1440,10 +1587,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
...payload,
};
const retriedAt = new Date();
const requeued = await this.prisma.cmppDownstreamDelivery.update({
where: { id: delivery.id },
const claimed = await this.prisma.cmppDownstreamDelivery.updateMany({
where: {
id: delivery.id,
status: delivery.status,
updatedAt: delivery.updatedAt,
},
data: {
status: 'pending',
status: 'manual_requeueing',
retryCount: 0,
manualRetryCount: { increment: 1 },
lastRetriedAt: retriedAt,
@@ -1459,6 +1610,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
lastError: null,
},
});
if (claimed.count !== 1) {
throw new BadRequestException('该下游投递记录已被其他操作处理，请刷新后重试');
}
await this.prisma.operationLog.create({
data: {
tenantId: delivery.tenantId,
@@ -1471,7 +1625,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
messageId: delivery.messageId,
previousStatus: delivery.status,
previousRetryCount: delivery.retryCount,
manualRetryCount: requeued.manualRetryCount,
manualRetryCount: (delivery.manualRetryCount ?? 0) + 1,
lastRetriedAt: retriedAt,
},
},
@@ -1494,6 +1648,32 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
}
async recoverStaleDownstreamManualRequeues(now = new Date()) {
const staleCutoff = new Date(now.getTime() - positiveInteger(
process.env.CMPP_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS,
DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS,
));
const stale = await this.prisma.cmppDownstreamDelivery.findMany({
where: { status: 'manual_requeueing', updatedAt: { lt: staleCutoff } },
select: { id: true, updatedAt: true },
orderBy: { updatedAt: 'asc' },
take: 500,
});
let recovered = 0;
for (const delivery of stale) {
const updated = await this.prisma.cmppDownstreamDelivery.updateMany({
where: { id: delivery.id, status: 'manual_requeueing', updatedAt: delivery.updatedAt },
data: {
status: 'pending',
nextRetryAt: null,
lastError: '人工重投进程中断，已恢复为待投递',
},
});
recovered += updated.count;
}
return { recovered };
}
async batchRequeueDownstreamDeliveries(ids: string[]) {
const uniqueIds = [...new Set(ids.filter(Boolean))];
if (uniqueIds.length === 0) {
@@ -1949,7 +2129,17 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
await this.recordCmppFailureReceipt(message, code, reason);
};
const queueAfterRiskChecks = async (options: { templateId?: string; signatureId?: string }) => {
const drainageInfoId = await this.resolveDrainageInfoId(options.signatureId, data.content);
const drainage = await this.resolveDrainageInfoMatch(options.signatureId, data.content);
const drainageInfoId = drainage?.id;
const drainageReason = drainageRejectionReason(drainage);
if (drainageReason) {
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: { drainageInfoId, signatureId: options.signatureId },
});
await reject('DRAINAGE_NOT_APPROVED', drainageReason);
return;
}
const risk = await this.riskReview.evaluateTask({
tenantId: application.tenantId,
applicationId: application.id,
@@ -2011,6 +2201,24 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (!signature) {
await reject('SIGNATURE', '短信内容未识别到已审核通过的签名');
} else {
const drainage = await this.resolveDrainageInfoMatch(signature.id, data.content);
const drainageReason = drainageRejectionReason(drainage);
if (drainageReason) {
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: { drainageInfoId: drainage?.id, signatureId: signature.id },
});
await reject('DRAINAGE_NOT_APPROVED', drainageReason);
return {
accepted: true,
tenantId: application.tenantId,
applicationId: application.id,
messageId,
messageRecordId: message.id,
taskId: task.id,
status: 'rejected',
};
}
const risk = await this.riskReview.evaluateTask({
tenantId: application.tenantId,
applicationId: application.id,
@@ -2037,7 +2245,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
});
}
const reviewTask = risk.status === 'pending_review' && risk.task
? await this.attachMessageToReviewTask(risk.task.id, message.id, signature.id, await this.resolveDrainageInfoId(signature.id, data.content))
? await this.attachMessageToReviewTask(risk.task.id, message.id, signature.id, drainage?.id)
: await this.riskReview.aggregateTemplateMismatch({
tenantId: application.tenantId,
applicationId: application.id,
@@ -2143,12 +2351,16 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (this.receiptTimeoutScanRunning) return;
this.receiptTimeoutScanRunning = true;
try {
const [receiptResult, downstreamResult] = await Promise.all([
const [receiptResult, downstreamResult, requeueRecoveryResult, downstreamManualRecoveryResult] = await Promise.all([
this.markUnknownTimeout({}),
this.markExpiredDownstreamDeliveries(),
this.recoverStaleGatewaySubmitRequeues(),
this.recoverStaleDownstreamManualRequeues(),
]);
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`);
if (requeueRecoveryResult.recovered > 0) this.logger.log(`Recovered ${requeueRecoveryResult.recovered} stale Gateway submit requeues`);
if (downstreamManualRecoveryResult.recovered > 0) this.logger.log(`Recovered ${downstreamManualRecoveryResult.recovered} stale downstream manual requeues`);
} catch (error) {
this.logger.error('Receipt timeout scan failed', error instanceof Error ? error.stack : String(error));
} finally {
@@ -2510,21 +2722,67 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
});
}
private async resolveTemplateMessageClassification(templateId: string | undefined, content: string) {
if (!templateId) return { signatureId: undefined, drainageInfoId: undefined };
const template = await this.prisma.smsTemplate.findUnique({
where: { id: templateId },
select: { signatureId: true },
});
const signatureId = template?.signatureId ?? undefined;
return { signatureId, drainageInfoId: await this.resolveDrainageInfoId(signatureId, content) };
private async resolveTemplateMessageClassification(
tenantId: string,
applicationId: string | undefined,
templateId: string | undefined,
content: string,
) {
if (templateId) {
const template = await this.prisma.smsTemplate.findUnique({
where: { id: templateId },
include: { signature: true },
});
if (!template || template.tenantId !== tenantId || template.applicationId !== applicationId
|| template.auditStatus !== 'approved' || template.signature?.auditStatus !== 'approved') {
throw new BadRequestException('短信模板不存在、未通过审核或不属于当前应用');
}
const variables = matchTemplateContent(template.content, content);
if (variables === null) {
throw new BadRequestException('短信内容与选定的审核模板不匹配');
}
const drainage = await this.resolveDrainageInfoMatch(template.signatureId, content);
return {
signatureId: template.signatureId,
drainageInfoId: drainage?.id,
variables,
rejectionReason: drainageRejectionReason(drainage),
};
}
if (!applicationId) {
throw new BadRequestException('自由内容短信必须关联企业应用');
}
const [application, signature] = await Promise.all([
this.prisma.smsApplication.findUnique({
where: { id: applicationId },
select: { tenantId: true, templateMismatchMode: true },
}),
this.resolveInboundSignatureCandidate(applicationId, content),
]);
if (!application || application.tenantId !== tenantId) {
throw new BadRequestException('短信应用不存在或不属于当前企业');
}
if (!signature) {
throw new BadRequestException('短信内容未以当前应用已审核通过的签名开头');
}
if (application.templateMismatchMode !== 'direct_send') {
throw new BadRequestException('当前应用未允许无模板自由内容直接发送');
}
const drainage = await this.resolveDrainageInfoMatch(signature.id, content);
return {
signatureId: signature.id,
drainageInfoId: drainage?.id,
variables: undefined,
rejectionReason: drainageRejectionReason(drainage),
};
}
private async resolveDrainageInfoId(signatureId: string | undefined, content: string) {
private async resolveDrainageInfoMatch(signatureId: string | null | undefined, content: string) {
if (!signatureId) return undefined;
const candidates = await this.prisma.smsDrainageInfo.findMany({
where: { signatureId, auditStatus: 'approved' },
select: { id: true, url: true, updatedAt: true },
where: { signatureId, auditStatus: { not: 'deleted' } },
select: { id: true, url: true, auditStatus: true, updatedAt: true },
orderBy: [{ updatedAt: 'desc' }, { id: 'asc' }],
});
const matches = candidates
@@ -2534,7 +2792,15 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (matches.length === 0) return undefined;
const longestLength = matches[0].normalizedUrl.length;
const longestMatches = matches.filter((item) => item.normalizedUrl.length === longestLength);
return longestMatches.length === 1 ? longestMatches[0].id : undefined;
if (longestMatches.length !== 1) {
throw new BadRequestException({
code: 'DRAINAGE_MATCH_AMBIGUOUS',
message: '短信内容同时匹配多条等长引流地址，无法确定报备资料',
drainageInfoIds: longestMatches.map((item) => item.id),
});
}
const matched = longestMatches[0];
return { id: matched.id, auditStatus: matched.auditStatus };
}
private async attachMessageToReviewTask(reviewTaskId: string, messageRecordId: string, signatureId: string, drainageInfoId?: string) {
@@ -3143,19 +3409,39 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return response.json().catch(() => ({}));
}
private async publishGatewaySubmitCommand(command: unknown) {
return this.getRedis().xadd(
process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM,
'*',
'messageType',
'SubmitCommand',
'data',
JSON.stringify(command),
private async publishGatewaySubmitCommand(command: unknown, idempotencyKey?: string) {
const redis = this.getRedis();
const stream = process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM;
const payload = JSON.stringify(command);
if (!idempotencyKey) {
return redis.xadd(stream, '*', 'messageType', 'SubmitCommand', 'data', payload);
}
const result = await redis.eval(
`local existing = redis.call('GET', KEYS[2])
if existing then return existing end
local streamId = redis.call('XADD', KEYS[1], '*', 'messageType', 'SubmitCommand', 'data', ARGV[1])
redis.call('SET', KEYS[2], streamId, 'EX', ARGV[2])
return streamId`,
2,
stream,
idempotencyKey,
payload,
String(GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS),
);
return typeof result === 'string' ? result : String(result ?? '');
}
}
function gatewaySubmitRequeueKey(deadLetterId: string, attempt: number) {
return `gateway:submit:requeue:${deadLetterId}:${attempt}`;
}
function drainageRejectionReason(drainage?: { id: string; auditStatus: string }) {
if (!drainage || drainage.auditStatus === 'approved') return undefined;
return `短信内容匹配的引流资料 ${drainage.id} 当前为 ${drainage.auditStatus}，必须审核通过后才能发送`;
}
function statusFromRisk(status: string, scheduled: boolean) {
if (status === 'rejected') {
return 'rejected';