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

perf: batch paid CMPP accounting and weighted routes

Browse Source
This commit is contained in:
hectorzhao
2026-08-21 10:25:40 +08:00
parent 0176aa6952
commit 487b5282a6
12 changed files with 219 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/billing/billing.service.spec.ts
+28 -3
View File
@@ -13,10 +13,12 @@ function createPrismaMock() {
tenantAccount: {
findMany: jest.fn(),
findUnique: jest.fn().mockImplementation(() => Promise.resolve({ ...accountState })),
findUniqueOrThrow: jest.fn().mockImplementation(() => Promise.resolve({ ...accountState })),
create: jest.fn(),
upsert: jest.fn().mockImplementation(() => Promise.resolve({ ...accountState })),
updateMany: jest.fn().mockImplementation(({ data }) => {
if (data.balanceCents?.increment !== undefined) accountState.balanceCents += data.balanceCents.increment;
else if (data.balanceCents?.decrement !== undefined) accountState.balanceCents -= data.balanceCents.decrement;
accountState.updatedAt = new Date(accountState.updatedAt.getTime() + 1);
return Promise.resolve({ count: 1 });
}),
@@ -31,7 +33,9 @@ function createPrismaMock() {
findMany: jest.fn(),
findFirst: jest.fn(),
findUnique: jest.fn().mockResolvedValue(null),
findUniqueOrThrow: jest.fn().mockImplementation(({ where }) => Promise.resolve({ id: 'tx-charged', idempotencyKey: where.idempotencyKey })),
create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: `tx-${data.transactionType}`, ...data })),
createMany: jest.fn().mockResolvedValue({ count: 2 }),
},
rechargeOrder: {
findMany: jest.fn(),
@@ -94,7 +98,7 @@ describe('BillingService', () => {
);
});
it('allows sending only when cash balance plus credit is greater than zero', async () => {
it('allows sending only when cash balance plus credit covers the required amount', async () => {
const prisma = createPrismaMock();
const service = new BillingService(prisma as never);
@@ -102,7 +106,7 @@ describe('BillingService', () => {
expect.objectContaining({ availableAmount: 1000, canSend: true }),
);
await expect(service.checkAccount({ tenantId: 'tenant-1', amountCents: 1001 })).resolves.toEqual(
expect.objectContaining({ availableAmount: 1000, canSend: true }),
expect.objectContaining({ availableAmount: 1000, canSend: false }),
);
await service.updateCreditLimit('tenant-1', { creditCents: -1000, operatorId: 'admin-1' });
await expect(service.checkAccount({ tenantId: 'tenant-1', amountCents: 1 })).resolves.toEqual(
@@ -110,7 +114,7 @@ describe('BillingService', () => {
);
await service.updateCreditLimit('tenant-1', { creditCents: 500, operatorId: 'admin-1' });
await expect(service.checkAccount({ tenantId: 'tenant-1', amountCents: 999999 })).resolves.toEqual(
expect.objectContaining({ availableAmount: 1500, creditCents: 500, canSend: true }),
expect.objectContaining({ availableAmount: 1500, creditCents: 500, canSend: false }),
);
await expect(service.updateCreditLimit('tenant-1', { creditCents: 1.5 })).rejects.toThrow('授信额度最多支持人民币小数点后 4 位');
expect(prisma.operationLog.create).toHaveBeenCalledWith({
@@ -312,6 +316,27 @@ describe('BillingService', () => {
expect(prisma.accountState).toEqual(expect.objectContaining({ balanceCents: 890 }));
});
it('settles a frozen SMS charge with one account lock and an idempotent ledger pair', async () => {
const prisma = createPrismaMock();
const rows = new Map<string, Record<string, unknown>>();
prisma.accountTransaction.findUnique.mockImplementation(({ where }) => Promise.resolve(rows.get(where.idempotencyKey) ?? null));
prisma.accountTransaction.findUniqueOrThrow.mockImplementation(({ where }) => Promise.resolve(rows.get(where.idempotencyKey)));
prisma.accountTransaction.createMany.mockImplementation(({ data }) => {
data.forEach((row: Record<string, unknown>) => rows.set(String(row.idempotencyKey), { id: `tx-${row.transactionType}`, ...row }));
return Promise.resolve({ count: data.length });
});
const service = new BillingService(prisma as never);
const input = { tenantId: 'tenant-1', amountCents: 325, messageId: 'msg-paid-1', taskId: 'task-paid-1' };
const first = await service.settleFrozenCharge(input);
const replay = await service.settleFrozenCharge(input);
expect(first).toEqual(expect.objectContaining({ transactionType: 'charged', amountCents: -325 }));
expect(replay).toEqual(first);
expect(prisma.accountTransaction.createMany).toHaveBeenCalledTimes(1);
expect(prisma.tenantAccount.update).not.toHaveBeenCalled();
});
it('serializes and replays concurrent refunds with one balance mutation', async () => {
const prisma = createPrismaMock();
let transactionChain = Promise.resolve<unknown>(undefined);
api/src/billing/billing.service.ts
+46 -1
View File
@@ -416,7 +416,7 @@ export class BillingService {
availableAmount,
balanceCents,
creditCents,
canSend: availableAmount > 0,
canSend: availableAmount >= requiredAmount,
};
}
@@ -436,6 +436,51 @@ export class BillingService {
});
}
async settleFrozenCharge(data: { tenantId: string; amountCents: number; messageId: string; taskId: string; remark?: string }) {
const amountCents = data.amountCents ?? 0;
if (amountCents <= 0) return null;
const releaseKey = `sms-charge-release:${data.messageId}`;
const chargeKey = `sms-charge:${data.messageId}`;
return this.prisma.$transaction(async (tx) => {
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'tenant-account:' + data.tenantId}, 0))`;
const [release, charge] = await Promise.all([
tx.accountTransaction.findUnique({ where: { idempotencyKey: releaseKey } }),
tx.accountTransaction.findUnique({ where: { idempotencyKey: chargeKey } }),
]);
if (charge) return charge;
const account = await tx.tenantAccount.findUniqueOrThrow({ where: { tenantId: data.tenantId } });
const balance = moneyToNumber(account.balanceCents);
if (release) {
const updated = await tx.tenantAccount.update({
where: { tenantId: data.tenantId },
data: { balanceCents: { decrement: amountCents } },
});
return tx.accountTransaction.create({
data: {
tenantId: data.tenantId, transactionType: 'charged', idempotencyKey: chargeKey,
amountCents: -amountCents, balanceAfter: updated.balanceCents,
relatedType: 'sms_message_record', relatedId: data.messageId, remark: '提交成功扣费',
},
});
}
await tx.accountTransaction.createMany({
data: [
{
tenantId: data.tenantId, transactionType: 'released', idempotencyKey: releaseKey,
amountCents, balanceAfter: balance + amountCents,
relatedType: 'sms_batch_task', relatedId: data.taskId, remark: data.remark,
},
{
tenantId: data.tenantId, transactionType: 'charged', idempotencyKey: chargeKey,
amountCents: -amountCents, balanceAfter: balance,
relatedType: 'sms_message_record', relatedId: data.messageId, remark: '提交成功扣费',
},
],
});
return tx.accountTransaction.findUniqueOrThrow({ where: { idempotencyKey: chargeKey } });
}, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted });
}
release(data: BillingActionDto) {
return this.applyAccountDelta({
...data,
api/src/send-chain/send-accounting.service.ts
+6 -17
View File
@@ -44,24 +44,13 @@ export class SendAccountingService {
if (exists?.billingStatus === 'charged') {
return;
}
if (amountCents > 0) {
await this.billing.release({
tenantId: message.tenantId,
amountCents,
idempotencyKey: `sms-charge-release:${message.messageId}`,
relatedType: 'sms_batch_task',
relatedId: message.batchTaskId,
remark: `短信 ${message.messageId} 提交成功释放冻结并转扣费`,
});
}
const transaction = await this.billing.charge({
const transaction = amountCents > 0 ? await this.billing.settleFrozenCharge({
tenantId: message.tenantId,
amountCents,
idempotencyKey: `sms-charge:${message.messageId}`,
relatedType: 'sms_message_record',
relatedId: message.messageId,
remark: '提交成功扣费',
});
taskId: message.batchTaskId,
messageId: message.messageId,
remark: `短信 ${message.messageId} 提交成功释放冻结并转扣费`,
}) : null;
const data = {
tenantId: message.tenantId,
applicationId: message.applicationId ?? undefined,
@@ -73,7 +62,7 @@ export class SendAccountingService {
unitPrice,
amountCents,
billingStatus: 'charged',
transactionId: transaction.id,
transactionId: transaction?.id,
};
if (exists) {
await this.prisma.smsBillingRecord.update({ where: { id: exists.id }, data });
api/src/send-chain/send-chain.helpers.spec.ts
+14
View File
@@ -49,6 +49,20 @@ describe('send-chain pure policies', () => {
})).toBeUndefined();
});
it('uses a stable weighted choice among equal-priority online primary channels', () => {
const items = [
{ channelId: 'primary-a', carrier: 'mobile', province: null, priority: 1, weight: 1, isBackup: false, channel: { carrier: 'mobile', sendRegion: '全国', status: 'active', connectionStates: connected } },
{ channelId: 'primary-b', carrier: 'mobile', province: null, priority: 1, weight: 1, isBackup: false, channel: { carrier: 'mobile', sendRegion: '全国', status: 'active', connectionStates: connected } },
{ channelId: 'backup', carrier: 'mobile', province: null, priority: 2, weight: 100, isBackup: true, channel: { carrier: 'mobile', sendRegion: '全国', status: 'active', connectionStates: connected } },
];
const selected = new Set(Array.from({ length: 100 }, (_, index) => selectChannelCandidate(items, {
carrier: 'mobile', routingKey: `message-${index}`,
excludedChannelIds: new Set(), approvedChannelIds: new Set(items.map((item) => item.channelId)),
})?.channelId));
expect(selected).toEqual(new Set(['primary-a', 'primary-b']));
});
it('keeps a segmented message non-terminal until all receipts arrive', () => {
const result = aggregateReceiptSegmentState(
[{ segmentTotal: 2, receiptStatus: 'delivered', deliveredAt: new Date('2026-07-31T00:00:00Z') }],
api/src/send-chain/send-chain.helpers.ts
+26 -1
View File
@@ -447,6 +447,9 @@ export type ChannelCandidate = {
channelId: string;
carrier?: string | null;
province?: string | null;
priority?: number | null;
weight?: number | null;
isBackup?: boolean | null;
channel: {
carrier?: string | null;
carriers?: string[] | null;
@@ -477,6 +480,7 @@ export function selectChannelCandidate<T extends ChannelCandidate>(
forceNational?: boolean;
excludedChannelIds: ReadonlySet<string>;
approvedChannelIds: ReadonlySet<string>;
routingKey?: string;
},
) {
const eligible = items.filter((item) =>
@@ -489,7 +493,28 @@ export function selectChannelCandidate<T extends ChannelCandidate>(
? []
: eligible.filter((item) => isProvinceChannel(item, options.province));
const nationalCandidates = eligible.filter((item) => isNationalChannel(item));
return [...provinceCandidates, ...nationalCandidates].find((item) => isChannelSendAvailable(item.channel));
const scope = provinceCandidates.some((item) => isChannelSendAvailable(item.channel))
? provinceCandidates
: nationalCandidates;
const available = scope.filter((item) => isChannelSendAvailable(item.channel));
if (available.length === 0) return undefined;
const priority = Math.min(...available.map((item) => item.priority ?? 100));
const priorityPool = available.filter((item) => (item.priority ?? 100) === priority);
const primaryPool = priorityPool.filter((item) => !item.isBackup);
const pool = primaryPool.length > 0 ? primaryPool : priorityPool;
if (pool.length === 1 || !options.routingKey) return pool[0];
const totalWeight = pool.reduce((sum, item) => sum + Math.max(1, item.weight ?? 1), 0);
let hash = 2166136261;
for (const character of options.routingKey) {
hash ^= character.charCodeAt(0);
hash = Math.imul(hash, 16777619) >>> 0;
}
let slot = hash % totalWeight;
for (const item of pool) {
slot -= Math.max(1, item.weight ?? 1);
if (slot < 0) return item;
}
return pool[0];
}
export type ReceiptSegmentAudit = {
api/src/send-chain/send-chain.service.spec.ts
+3 -4
View File
@@ -387,6 +387,7 @@ function createService(
freeze: jest.fn().mockResolvedValue({ id: 'tx-freeze' }),
release: jest.fn().mockResolvedValue({ id: 'tx-release' }),
charge: jest.fn().mockResolvedValue({ id: 'tx-charge' }),
settleFrozenCharge: jest.fn().mockResolvedValue({ id: 'tx-charge' }),
refund: jest.fn().mockResolvedValue({ id: 'tx-refund' }),
} as unknown as BillingService;
const riskReview = {
@@ -2373,8 +2374,7 @@ describe('SendChainService', () => {
where: { id: 'record-1', status: { notIn: ['delivered', 'failed', 'unknown'] } },
data: expect.objectContaining({ gatewayMessageId: 'GW-1', status: 'submitted', submitStatus: 'accepted' }),
});
expect(billing.release).toHaveBeenCalledWith(expect.objectContaining({ amountCents: 3, relatedId: 'task-1' }));
expect(billing.charge).toHaveBeenCalledWith(expect.objectContaining({ amountCents: 3, relatedId: 'MSG-1' }));
expect(billing.settleFrozenCharge).toHaveBeenCalledWith(expect.objectContaining({ amountCents: 3, taskId: 'task-1', messageId: 'MSG-1' }));
expect(prisma.smsBillingRecord.create).toHaveBeenCalledWith({
data: expect.objectContaining({ messageId: 'MSG-1', amountCents: 3, billingStatus: 'charged', transactionId: 'tx-charge' }),
});
@@ -2466,8 +2466,7 @@ describe('SendChainService', () => {
await service.handleSubmitResult(event);
await service.handleSubmitResult(event);
expect(billing.charge).toHaveBeenCalledTimes(1);
expect(billing.release).toHaveBeenCalledTimes(1);
expect(billing.settleFrozenCharge).toHaveBeenCalledTimes(1);
expect(prisma.smsSubmitRecord.updateMany).toHaveBeenCalledWith({
where: {
id: 'submit-1',
api/src/send-chain/send-gateway-submit.service.ts
+1
View File
@@ -349,6 +349,7 @@ async selectChannelForMessage(
forceNational: options.forceNational,
excludedChannelIds: excluded,
approvedChannelIds,
routingKey: message.id,
});
if (!selected) {
throw new NotFoundException('无已报备通过且在线的可用通道');
api/src/send-chain/send-inbound-entry.service.ts
+61 -14
View File
@@ -1,4 +1,4 @@
import { BadRequestException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common';
import { BadRequestException, ConflictException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { Queue, Worker } from 'bullmq';
import IORedis from 'ioredis';
@@ -1401,11 +1401,7 @@ startInboundWorkflowWorker() {
|| application.tenant.certificationStatus !== 'approved'
|| !/^1\d{10}$/.test(phoneNumber)
|| globalRejected.has(phoneNumber)
|| enterpriseRejected.has(`${application.tenantId}:${application.id}:${phoneNumber}`)
// Paid messages retain the existing per-message account lock until the
// dedicated batch-ledger migration is introduced; never weaken billing
// correctness merely to increase the benchmark number.
|| moneyToNumber(application.customerUnitPrice) !== 0) continue;
|| enterpriseRejected.has(`${application.tenantId}:${application.id}:${phoneNumber}`)) continue;
try {
if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((allowlist) => allowlist.ipCidr))) continue;
const clientSrcId = validateInboundApplicationSrcId(data.srcId, application);
@@ -1536,9 +1532,63 @@ startInboundWorkflowWorker() {
const messageRecordId = randomUUID();
const content = candidate.payload.data.content;
const drainageDetection = await detectDrainageContent(this.prisma, content);
return { candidate, workflowDigest, taskId, messageRecordId, content, drainageDetection };
const billing = this.billing.estimateSmsCost({
tenantId: candidate.application.tenantId,
applicationId: candidate.application.id,
content,
phoneCount: 1,
unitPrice: moneyToNumber(candidate.application.customerUnitPrice),
});
return { candidate, workflowDigest, taskId, messageRecordId, content, drainageDetection, billing };
}));
await this.measureInboundStage('message_persist', () => this.prisma.$transaction(async (tx) => {
// Lock accounts in a stable order and reserve the whole tenant subtotal once.
// Per-message idempotency rows remain separate so retries, releases and charges
// keep the original accounting contract without one account update per Inbox row.
const tenantIds = [...new Set(prepared.map(({ candidate }) => candidate.application.tenantId))].sort();
for (const tenantId of tenantIds) {
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'tenant-account:' + tenantId}, 0))`;
await tx.tenantAccount.upsert({
where: { tenantId }, update: {},
create: { tenantId, balanceCents: 0, creditCents: 0, status: 'active' },
});
const account = await tx.tenantAccount.findUniqueOrThrow({ where: { tenantId } });
const paid = prepared.filter(({ candidate, billing }) => candidate.application.tenantId === tenantId && billing.amountCents > 0);
const totalAmount = paid.reduce((sum, entry) => sum + entry.billing.amountCents, 0);
if (totalAmount === 0) continue;
const available = moneyToNumber(account.balanceCents) + moneyToNumber(account.creditCents);
if (account.status !== 'active' || available < totalAmount) {
throw new BadRequestException('企业账户余额不足');
}
const existing = await tx.accountTransaction.findMany({
where: { idempotencyKey: { in: paid.map(({ candidate }) => `${candidate.item.requestKey}:freeze`) } },
select: { idempotencyKey: true },
});
if (existing.length > 0) {
throw new ConflictException('批量计费幂等流水已存在，转入逐条恢复');
}
const balanceBefore = moneyToNumber(account.balanceCents);
let reserved = 0;
await tx.accountTransaction.createMany({
data: paid.map(({ candidate, taskId, billing }) => {
reserved += billing.amountCents;
return {
tenantId,
transactionType: 'frozen',
idempotencyKey: `${candidate.item.requestKey}:freeze`,
amountCents: -billing.amountCents,
balanceAfter: balanceBefore - reserved,
relatedType: 'sms_batch_task',
relatedId: taskId,
remark: 'CMPP 入站短信批量冻结',
};
}),
});
await tx.tenantAccount.update({
where: { tenantId },
data: { balanceCents: { decrement: totalAmount } },
});
}
await tx.smsBatchTask.createMany({
data: prepared.map(({ candidate, workflowDigest, taskId, content }) => ({
id: taskId,
@@ -1563,7 +1613,7 @@ startInboundWorkflowWorker() {
})),
});
await tx.smsMessageRecord.createMany({
data: prepared.map(({ candidate, taskId, messageRecordId, content, drainageDetection }) => ({
data: prepared.map(({ candidate, taskId, messageRecordId, content, drainageDetection, billing }) => ({
id: messageRecordId,
tenantId: candidate.application.tenantId,
batchTaskId: taskId,
@@ -1575,12 +1625,9 @@ startInboundWorkflowWorker() {
phoneNumber: candidate.phoneNumber,
content,
...drainageDetection,
billingUnits: this.billing.estimateSmsCost({
tenantId: candidate.application.tenantId, applicationId: candidate.application.id,
content, phoneCount: 1, unitPrice: 0,
}).billingUnitsPerMessage,
unitPrice: 0,
amountCents: 0,
billingUnits: billing.billingUnitsPerMessage,
unitPrice: billing.unitPrice,
amountCents: billing.amountCents,
queuePriority: normalizeQueuePriority(candidate.application.queuePriority),
cmppSubmitSequenceId: candidate.payload.data.sequenceId == null ? null : String(candidate.payload.data.sequenceId),
cmppSubmitGroupMessageId: candidate.payload.submitGroupMessageId,