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

feat: restore credit limits and harden SMS sending

Browse Source
This commit is contained in:
hectorzhao
2026-07-14 18:46:35 +08:00
parent 3e3b7a9d1a
commit c1a17699db
23 changed files with 796 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.ts
+55 -15
View File
@@ -1,4 +1,4 @@
import { BadRequestException, Injectable, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { BadRequestException, Injectable, Logger, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { Queue, Worker } from 'bullmq';
import IORedis from 'ioredis';
@@ -218,6 +218,9 @@ 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_RECEIPT_TIMEOUT_HOURS = 72;
const DEFAULT_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS = 5 * 60_000;
const RECEIPT_TIMEOUT_INITIAL_DELAY_MS = 60_000;
const BULLMQ_PRIORITY: Record<QueuePriority, number> = {
priority: 1,
normal: 100,
@@ -225,10 +228,14 @@ const BULLMQ_PRIORITY: Record<QueuePriority, number> = {
@Injectable()
export class SendChainService implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(SendChainService.name);
private redis?: IORedis;
private sendQueue?: Queue<SendJob, unknown, 'send-message'>;
private gatewayQueue?: Queue;
private worker?: Worker<SendJob>;
private receiptTimeoutInitialTimer?: ReturnType<typeof setTimeout>;
private receiptTimeoutIntervalTimer?: ReturnType<typeof setInterval>;
private receiptTimeoutScanRunning = false;
constructor(
private readonly prisma: PrismaService,
@@ -240,9 +247,20 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (process.env.API_ENABLE_SEND_WORKER === 'true') {
this.startWorker();
}
if (process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED !== 'false') {
this.receiptTimeoutInitialTimer = setTimeout(() => void this.runReceiptTimeoutScan(), RECEIPT_TIMEOUT_INITIAL_DELAY_MS);
this.receiptTimeoutInitialTimer.unref?.();
this.receiptTimeoutIntervalTimer = setInterval(
() => void this.runReceiptTimeoutScan(),
positiveInteger(process.env.SMS_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS, DEFAULT_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS),
);
this.receiptTimeoutIntervalTimer.unref?.();
}
}
async onModuleDestroy() {
if (this.receiptTimeoutInitialTimer) clearTimeout(this.receiptTimeoutInitialTimer);
if (this.receiptTimeoutIntervalTimer) clearInterval(this.receiptTimeoutIntervalTimer);
await this.worker?.close();
await this.sendQueue?.close();
await this.gatewayQueue?.close();
@@ -1807,30 +1825,47 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
async markUnknownTimeout(data: TimeoutUnknownDto) {
const olderThanHours = data.olderThanHours ?? 72;
const olderThanHours = data.olderThanHours ?? positiveInteger(process.env.SMS_RECEIPT_TIMEOUT_HOURS, DEFAULT_RECEIPT_TIMEOUT_HOURS);
const cutoff = new Date(Date.now() - olderThanHours * 60 * 60 * 1000);
const candidates = await this.prisma.smsMessageRecord.findMany({
where: {
status: 'unknown',
deliveredAt: { lte: cutoff },
tenantId: { not: null },
status: { in: ['submitted', 'unknown'] },
submittedAt: { lte: cutoff },
},
select: { id: true, batchTaskId: true },
select: { id: true, tenantId: true, batchTaskId: true, messageId: true, amountCents: true, billingUnits: true },
take: 10000,
});
await this.prisma.smsMessageRecord.updateMany({
where: { id: { in: candidates.map((candidate) => candidate.id) } },
data: { status: 'timeout', timeoutAt: new Date(), errorMessage: '72小时未收到明确回执，自动转超时' },
});
const timedOutTaskIds = new Set<string>();
let timeout = 0;
for (const candidate of candidates) {
const message = await this.prisma.smsMessageRecord.findUnique({ where: { id: candidate.id } });
if (message?.tenantId) {
await this.refundMessage(message as typeof message & { tenantId: string }, '72小时未收到明确回执，自动超时退款');
}
if (!candidate.tenantId) continue;
const transitioned = await this.prisma.smsMessageRecord.updateMany({
where: { id: candidate.id, status: { in: ['submitted', 'unknown'] } },
data: { status: 'timeout', timeoutAt: new Date(), errorMessage: `${olderThanHours}小时未收到明确回执，自动转超时` },
});
if (transitioned.count !== 1) continue;
timeout += 1;
await this.refundMessage(candidate as typeof candidate & { tenantId: string }, `${olderThanHours}小时未收到明确回执，自动超时退款`);
if (candidate.batchTaskId) timedOutTaskIds.add(candidate.batchTaskId);
}
for (const batchTaskId of new Set(candidates.map((candidate) => candidate.batchTaskId).filter((value): value is string => Boolean(value)))) {
for (const batchTaskId of timedOutTaskIds) {
await this.refreshTaskProgress(batchTaskId);
}
return { timeout: candidates.length };
return { timeout };
}
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`);
} catch (error) {
this.logger.error('Receipt timeout scan failed', error instanceof Error ? error.stack : String(error));
} finally {
this.receiptTimeoutScanRunning = false;
}
}
private async submitMessageToGateway(
@@ -2883,6 +2918,11 @@ function isProvinceChannel(item: { province?: string | null; channel: { sendRegi
return itemProvince === target || sendRegion === target;
}
function positiveInteger(value: string | undefined, fallback: number) {
const parsed = Number(value);
return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
}
function bullmqConnection() {
const redisUrl = new URL(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379');
return {