518 lines
19 KiB
TypeScript
518 lines
19 KiB
TypeScript
import { protocolUint32ToDb } from '../common/protocol-uint32';
|
|
import { completionContext } from './completion-context';
|
|
import { resolveUplinkMatch } from './uplink-matching';
|
|
import { BadRequestException, Logger, NotFoundException } from '@nestjs/common';
|
|
import { Prisma } from '@prisma/client';
|
|
import { createHash, randomUUID } from 'node:crypto';
|
|
import { BillingService } from '../billing/billing.service';
|
|
|
|
import type { OpenApiService } from '../open-api/open-api.service';
|
|
import { PrismaService } from '../prisma/prisma.service';
|
|
import type {
|
|
GatewayUplinkEventDto,
|
|
UplinkMatchCandidateInput,
|
|
GatewayControlDeliveryResult,
|
|
} from './send-chain.contracts';
|
|
import { downstreamControlFailureMessage } from './send-chain.helpers';
|
|
|
|
import type { SendCompletionCallbacks, SendCompletionFacade } from './send-completion.service';
|
|
import { queueFinalReceiptDeliveries, type DownstreamDeliveryQueueRequest } from './downstream-receipt-targets';
|
|
|
|
/**
|
|
* R10 downstreamDelivery implementation.
|
|
* Cross-domain calls return through the stable SendChainService seam to preserve locking and test observability.
|
|
*/
|
|
export class SendDownstreamDeliveryService {
|
|
private readonly logger = new Logger('SendChainService');
|
|
|
|
constructor(
|
|
private readonly prisma: PrismaService,
|
|
private readonly billing: BillingService,
|
|
private readonly openApi: OpenApiService | undefined,
|
|
private readonly facade: SendCompletionFacade,
|
|
private readonly callbacks: SendCompletionCallbacks,
|
|
) {}
|
|
|
|
async handleUplink(data: GatewayUplinkEventDto) {
|
|
protocolUint32ToDb(data.sequenceId);
|
|
if (!completionContext.getStore()) {
|
|
return this.prisma.$transaction(
|
|
(tx) => completionContext.run({ tx, messageRecordId: '' }, () => this.persistUplink(data)),
|
|
{ timeout: 15_000 },
|
|
);
|
|
}
|
|
return this.persistUplink(data);
|
|
}
|
|
|
|
private async persistUplink(data: GatewayUplinkEventDto) {
|
|
if (data.eventId) {
|
|
await this.prisma
|
|
.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`uplink-event:${data.eventId}`},0))`;
|
|
const existing = await this.prisma.smsUplinkMessage.findUnique({ where: { eventId: data.eventId } });
|
|
if (existing) return existing;
|
|
}
|
|
const channel = await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } });
|
|
if (!channel) {
|
|
throw new NotFoundException('SMS channel not found');
|
|
}
|
|
const match = await this.facade.resolveUplinkMatch(data, channel);
|
|
const record = await this.prisma.smsUplinkMessage.create({
|
|
data: {
|
|
eventId: data.eventId,
|
|
tenantId: match.tenantId,
|
|
applicationId: match.applicationId,
|
|
messageRecordId: match.messageRecordId,
|
|
channelId: data.channelId,
|
|
messageId: match.messageId,
|
|
gatewayMessageId: data.gatewayMessageId,
|
|
sequenceId: protocolUint32ToDb(data.sequenceId),
|
|
phoneNumber: data.phoneNumber,
|
|
destId: data.destId,
|
|
content: data.content,
|
|
matchStatus: match.matchStatus,
|
|
matchReason: match.matchReason,
|
|
receivedAt: data.receivedAt ? new Date(data.receivedAt) : new Date(),
|
|
},
|
|
});
|
|
if (match.candidates.length > 0) {
|
|
await this.prisma.smsUplinkMatchCandidate.createMany({
|
|
data: match.candidates.map((candidate) => ({
|
|
uplinkMessageId: record.id,
|
|
tenantId: candidate.tenantId,
|
|
applicationId: candidate.applicationId,
|
|
messageRecordId: candidate.messageRecordId,
|
|
matchSource: candidate.matchSource,
|
|
confidence: candidate.confidence,
|
|
reason: candidate.reason,
|
|
})),
|
|
skipDuplicates: true,
|
|
});
|
|
}
|
|
if (match.tenantId && match.applicationId) {
|
|
await this.facade.queueAndTryDownstreamDelivery({
|
|
tenantId: match.tenantId,
|
|
applicationId: match.applicationId,
|
|
messageRecordId: match.messageRecordId,
|
|
messageId: match.messageId,
|
|
deliveryType: 'uplink',
|
|
payload: {
|
|
messageId: match.messageId,
|
|
applicationId: match.applicationId,
|
|
phoneNumber: data.phoneNumber,
|
|
destId: data.destId,
|
|
content: data.content,
|
|
receivedAt: record.receivedAt.toISOString(),
|
|
uplinkMessageId: record.id,
|
|
},
|
|
});
|
|
}
|
|
return record;
|
|
}
|
|
|
|
async claimUplinkMatchCandidate(uplinkMessageId: string, candidateId: string, operatorId?: string) {
|
|
if (!completionContext.getStore()) {
|
|
return this.prisma.$transaction(
|
|
(tx) =>
|
|
completionContext.run({ tx, messageRecordId: '' }, () =>
|
|
this.persistUplinkClaim(uplinkMessageId, candidateId, operatorId),
|
|
),
|
|
{ timeout: 15_000 },
|
|
);
|
|
}
|
|
return this.persistUplinkClaim(uplinkMessageId, candidateId, operatorId);
|
|
}
|
|
|
|
private async persistUplinkClaim(uplinkMessageId: string, candidateId: string, operatorId?: string) {
|
|
await this.prisma
|
|
.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`uplink-claim:${uplinkMessageId}`},0))`;
|
|
const candidate = await this.prisma.smsUplinkMatchCandidate.findFirst({
|
|
where: { id: candidateId, uplinkMessageId },
|
|
include: {
|
|
application: { select: { id: true, name: true, cmppAccount: true } },
|
|
messageRecord: { select: { id: true, messageId: true, content: true } },
|
|
uplinkMessage: true,
|
|
},
|
|
});
|
|
if (!candidate) {
|
|
throw new NotFoundException('Uplink match candidate not found');
|
|
}
|
|
if (candidate.status === 'rejected') {
|
|
throw new BadRequestException('该候选已被排除，不能认领');
|
|
}
|
|
if (candidate.uplinkMessage.matchStatus === 'matched' && candidate.status !== 'claimed') {
|
|
throw new BadRequestException('该上行记录已完成匹配，不能重复认领');
|
|
}
|
|
if (candidate.status === 'claimed') return candidate.uplinkMessage;
|
|
const claimedAt = new Date();
|
|
const messageId = candidate.messageRecord?.messageId ?? null;
|
|
const updatedUplink = await this.prisma.smsUplinkMessage.update({
|
|
where: { id: uplinkMessageId },
|
|
data: {
|
|
tenantId: candidate.tenantId,
|
|
applicationId: candidate.applicationId,
|
|
messageRecordId: candidate.messageRecordId,
|
|
messageId,
|
|
matchStatus: 'matched',
|
|
matchReason: `人工认领：${candidate.reason ?? candidate.matchSource}`,
|
|
},
|
|
});
|
|
await this.prisma.smsUplinkMatchCandidate.updateMany({
|
|
where: {
|
|
uplinkMessageId,
|
|
id: { not: candidate.id },
|
|
status: 'pending',
|
|
},
|
|
data: { status: 'rejected' },
|
|
});
|
|
await this.prisma.smsUplinkMatchCandidate.update({
|
|
where: { id: candidate.id },
|
|
data: {
|
|
status: 'claimed',
|
|
claimedAt,
|
|
claimedById: operatorId,
|
|
},
|
|
});
|
|
await this.prisma.operationLog.create({
|
|
data: {
|
|
tenantId: candidate.tenantId,
|
|
userId: operatorId,
|
|
action: 'gateway.uplink_manual_claim',
|
|
resource: 'sms_uplink_message',
|
|
resourceId: uplinkMessageId,
|
|
detail: {
|
|
candidateId: candidate.id,
|
|
applicationId: candidate.applicationId,
|
|
applicationName: candidate.application.name,
|
|
messageRecordId: candidate.messageRecordId,
|
|
messageId,
|
|
matchSource: candidate.matchSource,
|
|
phoneNumber: candidate.uplinkMessage.phoneNumber,
|
|
destId: candidate.uplinkMessage.destId,
|
|
},
|
|
},
|
|
});
|
|
|
|
await this.facade.queueAndTryDownstreamDelivery({
|
|
tenantId: candidate.tenantId,
|
|
applicationId: candidate.applicationId,
|
|
messageRecordId: candidate.messageRecordId,
|
|
messageId,
|
|
deliveryType: 'uplink',
|
|
payload: {
|
|
messageId,
|
|
applicationId: candidate.applicationId,
|
|
phoneNumber: candidate.uplinkMessage.phoneNumber,
|
|
destId: candidate.uplinkMessage.destId,
|
|
content: candidate.uplinkMessage.content,
|
|
receivedAt: candidate.uplinkMessage.receivedAt.toISOString(),
|
|
manualClaim: true,
|
|
uplinkMessageId,
|
|
},
|
|
});
|
|
return this.prisma.smsUplinkMessage.findUnique({
|
|
where: { id: updatedUplink.id },
|
|
include: {
|
|
tenant: true,
|
|
application: true,
|
|
channel: true,
|
|
messageRecord: { include: { application: true } },
|
|
matchCandidates: {
|
|
include: {
|
|
tenant: true,
|
|
application: true,
|
|
messageRecord: { include: { application: true, tenant: true, channel: true } },
|
|
},
|
|
orderBy: [{ status: 'asc' }, { confidence: 'desc' }, { createdAt: 'asc' }],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
|
|
async queueAndTryDownstreamDelivery(data: DownstreamDeliveryQueueRequest) {
|
|
if (!data.applicationId) {
|
|
return null;
|
|
}
|
|
const application = await this.prisma.smsApplication.findUnique({
|
|
where: { id: data.applicationId },
|
|
select: {
|
|
cmppAccount: true,
|
|
interfaceEnabled: true,
|
|
status: true,
|
|
downstreamReceiptRetryEnabled: true,
|
|
downstreamUplinkRetryEnabled: true,
|
|
httpConfig: true,
|
|
},
|
|
});
|
|
const deliveryAllowed = application?.status === 'active' || application?.status === 'disabling';
|
|
const cmppDeliveryAllowed = deliveryAllowed || data.allowBusinessRejectionCmppDelivery === true;
|
|
if (deliveryAllowed && data.queueHttpWebhook !== false) {
|
|
try {
|
|
await this.openApi?.queueWebhookEvent(
|
|
{
|
|
tenantId: data.tenantId,
|
|
applicationId: data.applicationId,
|
|
messageRecordId: data.messageRecordId,
|
|
messageId: data.messageId,
|
|
uplinkMessageId:
|
|
typeof data.payload.uplinkMessageId === 'string' ? data.payload.uplinkMessageId : undefined,
|
|
eventType: data.deliveryType,
|
|
payload: data.payload,
|
|
},
|
|
completionContext.getStore()?.tx,
|
|
);
|
|
} catch (error) {
|
|
this.logger.error(
|
|
`HTTP webhook queue failed for ${data.deliveryType}/${data.messageId ?? '-'}: ${error instanceof Error ? error.message : String(error)}`,
|
|
);
|
|
if (data.propagateHttpQueueError || completionContext.getStore()) throw error;
|
|
}
|
|
}
|
|
if (data.queueCmppDelivery === false) {
|
|
return null;
|
|
}
|
|
if (
|
|
!application?.cmppAccount ||
|
|
(application.interfaceEnabled !== true && data.allowBusinessRejectionCmppDelivery !== true)
|
|
) {
|
|
return null;
|
|
}
|
|
const payload = { account: application?.cmppAccount, applicationId: data.applicationId, ...data.payload };
|
|
const dedupeKey =
|
|
data.deliveryType === 'receipt' && data.messageRecordId
|
|
? (data.receiptDedupeKey ?? `receipt:${data.messageRecordId}`)
|
|
: data.deliveryType === 'uplink' && typeof data.payload.uplinkMessageId === 'string'
|
|
? `uplink:${data.payload.uplinkMessageId}`
|
|
: null;
|
|
if (completionContext.getStore()) {
|
|
if (!dedupeKey) throw new Error('completion_notification_identity_missing');
|
|
await this.prisma.cmppDownstreamDelivery.createMany({
|
|
data: [
|
|
{
|
|
tenantId: data.tenantId,
|
|
applicationId: data.applicationId,
|
|
messageRecordId: data.messageRecordId,
|
|
messageId: data.messageId,
|
|
dedupeKey,
|
|
deliveryType: data.deliveryType,
|
|
payload,
|
|
retryEnabled:
|
|
cmppDeliveryAllowed &&
|
|
(data.deliveryType === 'uplink'
|
|
? application.downstreamUplinkRetryEnabled
|
|
: application.downstreamReceiptRetryEnabled),
|
|
status: cmppDeliveryAllowed ? 'pending' : 'abandoned',
|
|
lastError: cmppDeliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送',
|
|
},
|
|
],
|
|
skipDuplicates: true,
|
|
});
|
|
const retained = await this.prisma.cmppDownstreamDelivery.findUniqueOrThrow({ where: { dedupeKey } });
|
|
if (
|
|
(retained.messageRecordId ?? null) !== (data.messageRecordId ?? null) ||
|
|
retained.applicationId !== data.applicationId
|
|
)
|
|
throw new Error('completion_notification_identity_mismatch');
|
|
return retained;
|
|
}
|
|
let delivery;
|
|
try {
|
|
delivery = await this.prisma.cmppDownstreamDelivery.create({
|
|
data: {
|
|
tenantId: data.tenantId,
|
|
applicationId: data.applicationId,
|
|
messageRecordId: data.messageRecordId,
|
|
messageId: data.messageId,
|
|
dedupeKey,
|
|
deliveryType: data.deliveryType,
|
|
payload,
|
|
retryEnabled:
|
|
cmppDeliveryAllowed &&
|
|
(data.deliveryType === 'uplink'
|
|
? (application?.downstreamUplinkRetryEnabled ?? true)
|
|
: (application?.downstreamReceiptRetryEnabled ?? true)),
|
|
status: cmppDeliveryAllowed ? 'pending' : 'abandoned',
|
|
lastError: cmppDeliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送',
|
|
},
|
|
});
|
|
} catch (error) {
|
|
if (dedupeKey && error instanceof Prisma.PrismaClientKnownRequestError && error.code === 'P2002') {
|
|
const existing = await this.prisma.cmppDownstreamDelivery.findUnique({
|
|
where: { dedupeKey },
|
|
});
|
|
if (existing) {
|
|
this.logger.warn(
|
|
`downstream_delivery_deduplicated ${JSON.stringify({
|
|
deliveryType: data.deliveryType,
|
|
messageRecordId: data.messageRecordId,
|
|
messageId: data.messageId,
|
|
dedupeKey,
|
|
deliveryId: existing.id,
|
|
})}`,
|
|
);
|
|
return existing;
|
|
}
|
|
}
|
|
throw error;
|
|
}
|
|
if (!cmppDeliveryAllowed) {
|
|
return delivery;
|
|
}
|
|
const claimId = `api-direct:${process.pid}:${randomUUID()}`;
|
|
const claim = await this.prisma.cmppDownstreamDelivery.updateMany({
|
|
where: { id: delivery.id, status: 'pending' },
|
|
data: {
|
|
status: 'dispatching',
|
|
connectionId: claimId,
|
|
ackDeadlineAt: new Date(Date.now() + 30_000),
|
|
nextRetryAt: null,
|
|
lastError: null,
|
|
},
|
|
});
|
|
if (claim.count !== 1) {
|
|
return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: delivery.id } });
|
|
}
|
|
try {
|
|
const result = (await this.facade.postGatewayControl(
|
|
data.deliveryType === 'receipt' ? '/downstream/receipt' : '/downstream/uplink',
|
|
{ deliveryId: delivery.id, claimId, ...payload },
|
|
)) as GatewayControlDeliveryResult;
|
|
if (result.sent || result.delivered) {
|
|
await this.facade.markDownstreamDeliverySent({ id: delivery.id, claimId, ...result });
|
|
} else if (result.reasonCode === 'SUBMIT_RESPONSE_PENDING') {
|
|
await this.facade.markDownstreamDeliveryFailed(
|
|
delivery.id,
|
|
downstreamControlFailureMessage(result),
|
|
'claim_released',
|
|
{ id: delivery.id, claimId, ...result },
|
|
);
|
|
} else {
|
|
await this.facade.markDownstreamDeliveryFailed(
|
|
delivery.id,
|
|
downstreamControlFailureMessage(result),
|
|
result.retryable === false ? 'unrecoverable' : 'send_failed',
|
|
{ id: delivery.id, claimId, ...result },
|
|
);
|
|
}
|
|
} catch (error) {
|
|
await this.facade.markDownstreamDeliveryFailed(
|
|
delivery.id,
|
|
error instanceof Error ? error.message : 'Gateway control delivery failed',
|
|
);
|
|
}
|
|
return delivery;
|
|
}
|
|
|
|
async resolveUplinkMatch(
|
|
data: GatewayUplinkEventDto,
|
|
channel: { id: string; srcId?: string | null },
|
|
): Promise<{
|
|
tenantId?: string;
|
|
applicationId?: string;
|
|
messageRecordId?: string;
|
|
messageId?: string;
|
|
matchStatus: string;
|
|
matchReason: string;
|
|
candidates: UplinkMatchCandidateInput[];
|
|
}> {
|
|
return resolveUplinkMatch(this.prisma, data, channel);
|
|
}
|
|
|
|
async recordCmppFailureReceipt(
|
|
message: {
|
|
id: string;
|
|
tenantId?: string | null;
|
|
batchTaskId?: string | null;
|
|
applicationId?: string | null;
|
|
messageId: string;
|
|
phoneNumber: string;
|
|
cmppSubmitSequenceId?: string | null;
|
|
cmppSubmitGroupMessageId?: string | null;
|
|
cmppRegisteredDelivery?: boolean | null;
|
|
},
|
|
errorCode: string,
|
|
reason: string,
|
|
) {
|
|
if (!message.tenantId || !message.applicationId) return null;
|
|
const existing = await this.prisma.smsReceiptRecord.findFirst({
|
|
where: { messageRecordId: message.id, gatewayMessageId: `PLATFORM:${message.messageId}` },
|
|
});
|
|
const recoverableRejection = errorCode.startsWith('DRN') || errorCode === 'CSW';
|
|
if (existing && !recoverableRejection) return existing;
|
|
const deliveredAt = new Date();
|
|
await this.prisma.smsMessageRecord.update({
|
|
where: { id: message.id },
|
|
data: {
|
|
status: 'failed',
|
|
receiptStatus: 'undelivered',
|
|
receiptRawStatus: 'REJECTD',
|
|
errorCode,
|
|
errorMessage: reason,
|
|
deliveredAt,
|
|
},
|
|
});
|
|
const gatewayMessageId = `PLATFORM:${message.messageId}`;
|
|
const receiptKey = createHash('sha256')
|
|
.update(
|
|
`platform\u0000${gatewayMessageId}\u0000${message.phoneNumber}\u0000undelivered\u0000REJECTD\u0000${errorCode}`,
|
|
)
|
|
.digest('hex');
|
|
const receiptData = {
|
|
tenantId: message.tenantId,
|
|
batchTaskId: message.batchTaskId,
|
|
messageRecordId: message.id,
|
|
receiptKey,
|
|
messageId: message.messageId,
|
|
gatewayMessageId,
|
|
phoneNumber: message.phoneNumber,
|
|
receiptStatus: 'undelivered',
|
|
rawStatus: 'REJECTD',
|
|
errorCode,
|
|
errorMessage: reason,
|
|
deliveredAt,
|
|
};
|
|
const receipt =
|
|
existing ??
|
|
(recoverableRejection
|
|
? await this.prisma.smsReceiptRecord.upsert({ where: { receiptKey }, update: {}, create: receiptData })
|
|
: await this.prisma.smsReceiptRecord.create({ data: receiptData }));
|
|
await queueFinalReceiptDeliveries(this.prisma, (request) => this.facade.queueAndTryDownstreamDelivery(request), {
|
|
message,
|
|
propagateHttpQueueError: recoverableRejection,
|
|
allowBusinessRejectionCmppDelivery: errorCode === 'ACCOUNT' || errorCode === 'INTERFACE',
|
|
payload: {
|
|
messageId: message.messageId,
|
|
gatewayMessageId: `PLATFORM:${message.messageId}`,
|
|
phoneNumber: message.phoneNumber,
|
|
receiptStatus: 'undelivered',
|
|
rawStatus: 'REJECTD',
|
|
errorCode,
|
|
errorMessage: reason,
|
|
deliveredAt: deliveredAt.toISOString(),
|
|
},
|
|
});
|
|
if (message.batchTaskId) await this.facade.refreshTaskProgress(message.batchTaskId);
|
|
if (errorCode === 'CSW')
|
|
await this.prisma.smsMessageRecord.update({
|
|
where: { id: message.id },
|
|
data: { channelWordFinalizationPending: false },
|
|
});
|
|
if (errorCode.startsWith('DRN'))
|
|
await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { drainageReceiptPending: false } });
|
|
return receipt;
|
|
}
|
|
|
|
async postGatewayControl(path: string, payload: unknown) {
|
|
const baseUrl = (process.env.GATEWAY_CONTROL_URL ?? 'http://127.0.0.1:8090').replace(/\/$/, '');
|
|
const response = await fetch(`${baseUrl}${path}`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify(payload),
|
|
});
|
|
if (!response.ok) {
|
|
const body = await response.text().catch(() => '');
|
|
throw new Error(`Gateway control ${path} returned ${response.status}${body ? `: ${body}` : ''}`);
|
|
}
|
|
return response.json().catch(() => ({}));
|
|
}
|
|
}
|