From 54617c927ea2087cf2d27f9bf19d58a238d16b45 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Sat, 25 Jul 2026 23:32:13 +0800 Subject: [PATCH] fix: harden upstream and downstream receipt delivery --- .../migration.sql | 86 + api/prisma/schema.prisma | 1517 +++++++++-------- api/src/operations/operations.service.spec.ts | 7 +- api/src/operations/operations.service.ts | 7 +- .../gateway-events.controller.spec.ts | 47 + .../send-chain/gateway-events.controller.ts | 15 +- api/src/send-chain/send-chain.service.spec.ts | 143 +- api/src/send-chain/send-chain.service.ts | 420 ++++- .../first-version-development-requirements.md | 4 + docs/testing-progress.md | 11 + gateway/internal/inbound/server.go | 107 +- gateway/internal/inbound/server_test.go | 38 + gateway/internal/queue/messages.go | 1 + gateway/internal/upstream/deliver_test.go | 11 +- gateway/internal/upstream/manager.go | 89 +- src/api/adminApi.ts | 16 + .../admin/AdminDownstreamDeliveriesPage.tsx | 29 + src/styles/global.css | 28 + 18 files changed, 1797 insertions(+), 779 deletions(-) create mode 100644 api/prisma/migrations/20260725160000_add_reliable_receipt_delivery_tracking/migration.sql diff --git a/api/prisma/migrations/20260725160000_add_reliable_receipt_delivery_tracking/migration.sql b/api/prisma/migrations/20260725160000_add_reliable_receipt_delivery_tracking/migration.sql new file mode 100644 index 0000000..c9c0bce --- /dev/null +++ b/api/prisma/migrations/20260725160000_add_reliable_receipt_delivery_tracking/migration.sql @@ -0,0 +1,86 @@ +CREATE TABLE "CmppDownstreamDeliveryAttempt" ( + "id" TEXT NOT NULL, + "deliveryId" TEXT NOT NULL, + "attemptKey" TEXT NOT NULL, + "attemptNo" INTEGER NOT NULL, + "connectionId" TEXT, + "sequenceId" TEXT, + "messageId" TEXT, + "status" TEXT NOT NULL, + "sentAt" TIMESTAMP(3), + "ackDeadlineAt" TIMESTAMP(3), + "acknowledgedAt" TIMESTAMP(3), + "ackResult" INTEGER, + "failureType" TEXT, + "errorMessage" TEXT, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "CmppDownstreamDeliveryAttempt_pkey" PRIMARY KEY ("id") +); + +CREATE TABLE "UpstreamReceiptInbox" ( + "id" TEXT NOT NULL, + "receiptKey" TEXT NOT NULL, + "incomingChannelId" TEXT NOT NULL, + "incomingConnectionId" TEXT, + "upstreamAccount" TEXT NOT NULL, + "upstreamHost" TEXT NOT NULL, + "upstreamPort" INTEGER NOT NULL, + "protocol" TEXT NOT NULL, + "protocolVersion" TEXT NOT NULL, + "provisionalMessageId" TEXT, + "sequenceId" INTEGER, + "gatewayMessageId" TEXT NOT NULL, + "phoneNumber" TEXT, + "receiptStatus" TEXT NOT NULL, + "rawStatus" TEXT NOT NULL, + "errorCode" TEXT, + "errorMessage" TEXT, + "deliveredAt" TIMESTAMP(3) NOT NULL, + "receivedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "status" TEXT NOT NULL DEFAULT 'pending', + "matchedMessageRecordId" TEXT, + "matchedSubmitRecordId" TEXT, + "matchedChannelId" TEXT, + "attemptCount" INTEGER NOT NULL DEFAULT 0, + "nextRetryAt" TIMESTAMP(3), + "lastError" TEXT, + "processedAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "UpstreamReceiptInbox_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "CmppDownstreamDeliveryAttempt_attemptKey_key" +ON "CmppDownstreamDeliveryAttempt"("attemptKey"); + +CREATE INDEX "CmppDownstreamDeliveryAttempt_deliveryId_attemptNo_idx" +ON "CmppDownstreamDeliveryAttempt"("deliveryId", "attemptNo"); + +CREATE INDEX "CmppDownstreamDeliveryAttempt_status_ackDeadlineAt_idx" +ON "CmppDownstreamDeliveryAttempt"("status", "ackDeadlineAt"); + +CREATE INDEX "CmppDownstreamDeliveryAttempt_connectionId_sequenceId_idx" +ON "CmppDownstreamDeliveryAttempt"("connectionId", "sequenceId"); + +CREATE UNIQUE INDEX "UpstreamReceiptInbox_receiptKey_key" +ON "UpstreamReceiptInbox"("receiptKey"); + +CREATE INDEX "UpstreamReceiptInbox_status_nextRetryAt_receivedAt_idx" +ON "UpstreamReceiptInbox"("status", "nextRetryAt", "receivedAt"); + +CREATE INDEX "UpstreamReceiptInbox_gatewayMessageId_phoneNumber_idx" +ON "UpstreamReceiptInbox"("gatewayMessageId", "phoneNumber"); + +CREATE INDEX "UpstreamReceiptInbox_incomingChannelId_receivedAt_idx" +ON "UpstreamReceiptInbox"("incomingChannelId", "receivedAt"); + +CREATE INDEX "UpstreamReceiptInbox_matchedMessageRecordId_idx" +ON "UpstreamReceiptInbox"("matchedMessageRecordId"); + +ALTER TABLE "CmppDownstreamDeliveryAttempt" +ADD CONSTRAINT "CmppDownstreamDeliveryAttempt_deliveryId_fkey" +FOREIGN KEY ("deliveryId") REFERENCES "CmppDownstreamDelivery"("id") +ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index aade56b..feb22ff 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -7,64 +7,64 @@ datasource db { } model Tenant { - id String @id @default(cuid()) - name String - code String @unique - status String @default("active") - certificationStatus String @default("approved") - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + name String + code String @unique + status String @default("active") + certificationStatus String @default("approved") + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - users User[] - enterpriseCertifications EnterpriseCertification[] - operationLogs OperationLog[] - fileObjects FileObject[] - enterpriseBlacklists EnterpriseBlacklist[] - accounts TenantAccount[] - accountTransactions AccountTransaction[] - rechargeOrders RechargeOrder[] - smsBillingRecords SmsBillingRecord[] - smsApplications SmsApplication[] - smsSignatures SmsSignature[] - smsDrainageInfos SmsDrainageInfo[] - smsTemplates SmsTemplate[] - auditRecords AuditRecord[] - riskRules RiskRule[] - smsSendTasks SmsSendTask[] - riskHitRecords RiskHitRecord[] - smsBatchTasks SmsBatchTask[] - smsMessageRecords SmsMessageRecord[] - smsApiRequests SmsApiRequest[] - smsSubmitRecords SmsSubmitRecord[] - smsReceiptRecords SmsReceiptRecord[] - smsMessageSegmentAudits SmsMessageSegmentAudit[] - smsUplinkMessages SmsUplinkMessage[] - smsUplinkMatchCandidates SmsUplinkMatchCandidate[] - cmppDownstreamDeliveries CmppDownstreamDelivery[] - cmppDownstreamConnections CmppDownstreamConnection[] - cmppConnectionStates CmppConnectionState[] + users User[] + enterpriseCertifications EnterpriseCertification[] + operationLogs OperationLog[] + fileObjects FileObject[] + enterpriseBlacklists EnterpriseBlacklist[] + accounts TenantAccount[] + accountTransactions AccountTransaction[] + rechargeOrders RechargeOrder[] + smsBillingRecords SmsBillingRecord[] + smsApplications SmsApplication[] + smsSignatures SmsSignature[] + smsDrainageInfos SmsDrainageInfo[] + smsTemplates SmsTemplate[] + auditRecords AuditRecord[] + riskRules RiskRule[] + smsSendTasks SmsSendTask[] + riskHitRecords RiskHitRecord[] + smsBatchTasks SmsBatchTask[] + smsMessageRecords SmsMessageRecord[] + smsApiRequests SmsApiRequest[] + smsSubmitRecords SmsSubmitRecord[] + smsReceiptRecords SmsReceiptRecord[] + smsMessageSegmentAudits SmsMessageSegmentAudit[] + smsUplinkMessages SmsUplinkMessage[] + smsUplinkMatchCandidates SmsUplinkMatchCandidate[] + cmppDownstreamDeliveries CmppDownstreamDelivery[] + cmppDownstreamConnections CmppDownstreamConnection[] + cmppConnectionStates CmppConnectionState[] gatewayDownstreamRecoveryStatuses GatewayDownstreamRecoveryStatus[] - gatewaySubmitDeadLetters GatewaySubmitDeadLetter[] - openApiRequests OpenApiRequest[] - httpWebhookEvents HttpWebhookEvent[] - cmppInboundLongMessages CmppInboundLongMessage[] + gatewaySubmitDeadLetters GatewaySubmitDeadLetter[] + openApiRequests OpenApiRequest[] + httpWebhookEvents HttpWebhookEvent[] + cmppInboundLongMessages CmppInboundLongMessage[] } model EnterpriseCertification { - id String @id @default(cuid()) - tenantId String - companyName String - licenseNo String? - contactName String? - contactPhone String? - materials Json? - status String @default("pending") - rejectReason String? - reviewerId String? - submittedAt DateTime @default(now()) - reviewedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String + companyName String + licenseNo String? + contactName String? + contactPhone String? + materials Json? + status String @default("pending") + rejectReason String? + reviewerId String? + submittedAt DateTime @default(now()) + reviewedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt tenant Tenant @relation(fields: [tenantId], references: [id]) @@ -72,28 +72,28 @@ model EnterpriseCertification { } model User { - id String @id @default(cuid()) - tenantId String? - username String @unique - email String? @unique - phone String? @unique - displayName String - passwordHash String - status String @default("active") - sessionVersion Int @default(0) - failedLoginCount Int @default(0) - lockedUntil DateTime? - lastLoginAt DateTime? - deletedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String? + username String @unique + email String? @unique + phone String? @unique + displayName String + passwordHash String + status String @default("active") + sessionVersion Int @default(0) + failedLoginCount Int @default(0) + lockedUntil DateTime? + lastLoginAt DateTime? + deletedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant? @relation(fields: [tenantId], references: [id]) - roles UserRole[] - operationLogs OperationLog[] - auditRecords AuditRecord[] - createdSmsSendTasks SmsSendTask[] @relation("SmsSendTaskCreator") - reviewedSmsSendTasks SmsSendTask[] @relation("SmsSendTaskReviewer") + tenant Tenant? @relation(fields: [tenantId], references: [id]) + roles UserRole[] + operationLogs OperationLog[] + auditRecords AuditRecord[] + createdSmsSendTasks SmsSendTask[] @relation("SmsSendTaskCreator") + reviewedSmsSendTasks SmsSendTask[] @relation("SmsSendTaskReviewer") createdSmsBatchTasks SmsBatchTask[] @relation("SmsBatchTaskCreator") } @@ -162,18 +162,18 @@ model OperationLog { } model OperationLogArchive { - originalId String @id - tenantId String? - userId String? - action String - resource String - resourceId String? - ipAddress String? - userAgent String? - detail Json? - createdAt DateTime + originalId String @id + tenantId String? + userId String? + action String + resource String + resourceId String? + ipAddress String? + userAgent String? + detail Json? + createdAt DateTime archiveMonth String - archivedAt DateTime @default(now()) + archivedAt DateTime @default(now()) @@index([tenantId, createdAt]) @@index([userId, createdAt]) @@ -263,15 +263,15 @@ model SensitiveWord { } model EnterpriseBlacklist { - id String @id @default(cuid()) - tenantId String + id String @id @default(cuid()) + tenantId String applicationId String - phoneNumber String - reason String? - status String @default("active") - createdAt DateTime @default(now()) + phoneNumber String + reason String? + status String @default("active") + createdAt DateTime @default(now()) - tenant Tenant @relation(fields: [tenantId], references: [id]) + tenant Tenant @relation(fields: [tenantId], references: [id]) application SmsApplication @relation(fields: [applicationId], references: [id]) @@unique([applicationId, phoneNumber]) @@ -297,7 +297,7 @@ model DrainageField { updatedAt DateTime @updatedAt channelReportFields ChannelReportField[] - commonReportFields CommonReportField[] + commonReportFields CommonReportField[] } model TenantAccount { @@ -343,17 +343,17 @@ model BillingRule { } model RechargeOrder { - id String @id @default(cuid()) + id String @id @default(cuid()) tenantId String - orderNo String @unique + orderNo String @unique amountCents BigInt - status String @default("created") + status String @default("created") payMethod String? paidAt DateTime? operatorId String? remark String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt tenant Tenant @relation(fields: [tenantId], references: [id]) @@ -361,20 +361,20 @@ model RechargeOrder { } model SmsBillingRecord { - id String @id @default(cuid()) - tenantId String - applicationId String? - taskId String? - messageId String? - phoneNumber String? - contentLength Int - billingUnits Int - unitPrice BigInt - amountCents BigInt - billingStatus String @default("estimated") - transactionId String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String + applicationId String? + taskId String? + messageId String? + phoneNumber String? + contentLength Int + billingUnits Int + unitPrice BigInt + amountCents BigInt + billingStatus String @default("estimated") + transactionId String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt tenant Tenant @relation(fields: [tenantId], references: [id]) @@ -384,57 +384,57 @@ model SmsBillingRecord { } model SmsApplication { - id String @id @default(cuid()) - tenantId String - name String - scene String? - callbackUrl String? - cmppAccount String @unique - cmppEnterpriseCode String - cmppApplicationExtension String? - cmppAccessNumberFillEnabled Boolean @default(false) - cmppAccessNumberFillPrefix String? - cmppClientSrcId String? @unique - secretHash String - interfaceEnabled Boolean @default(true) - interfaceType String @default("cmpp20") - cmppMaxConnections Int @default(1) - cmppWindowSize Int @default(16) - dailyLimit Int @default(100000) - customerUnitPrice BigInt @default(0) - queuePriority String @default("normal") - maxPhonesPerTask Int @default(10000) - templateMismatchMode String @default("reject") - downstreamReceiptRetryEnabled Boolean @default(true) - downstreamUplinkRetryEnabled Boolean @default(true) - status String @default("active") - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String + name String + scene String? + callbackUrl String? + cmppAccount String @unique + cmppEnterpriseCode String + cmppApplicationExtension String? + cmppAccessNumberFillEnabled Boolean @default(false) + cmppAccessNumberFillPrefix String? + cmppClientSrcId String? @unique + secretHash String + interfaceEnabled Boolean @default(true) + interfaceType String @default("cmpp20") + cmppMaxConnections Int @default(1) + cmppWindowSize Int @default(16) + dailyLimit Int @default(100000) + customerUnitPrice BigInt @default(0) + queuePriority String @default("normal") + maxPhonesPerTask Int @default(10000) + templateMismatchMode String @default("reject") + downstreamReceiptRetryEnabled Boolean @default(true) + downstreamUplinkRetryEnabled Boolean @default(true) + status String @default("active") + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant @relation(fields: [tenantId], references: [id]) - ipAllowlist SmsApplicationIpAllowlist[] - signatures SmsSignature[] - drainageInfos SmsDrainageInfo[] - templates SmsTemplate[] - enterpriseBlacklists EnterpriseBlacklist[] - sendTasks SmsSendTask[] - batchTasks SmsBatchTask[] - messageRecords SmsMessageRecord[] - uplinkMessages SmsUplinkMessage[] - uplinkMatchCandidates SmsUplinkMatchCandidate[] - downstreamDeliveries CmppDownstreamDelivery[] - downstreamConnections CmppDownstreamConnection[] - connectionStates CmppConnectionState[] + tenant Tenant @relation(fields: [tenantId], references: [id]) + ipAllowlist SmsApplicationIpAllowlist[] + signatures SmsSignature[] + drainageInfos SmsDrainageInfo[] + templates SmsTemplate[] + enterpriseBlacklists EnterpriseBlacklist[] + sendTasks SmsSendTask[] + batchTasks SmsBatchTask[] + messageRecords SmsMessageRecord[] + uplinkMessages SmsUplinkMessage[] + uplinkMatchCandidates SmsUplinkMatchCandidate[] + downstreamDeliveries CmppDownstreamDelivery[] + downstreamConnections CmppDownstreamConnection[] + connectionStates CmppConnectionState[] gatewayDownstreamRecoveryStatuses GatewayDownstreamRecoveryStatus[] - gatewaySubmitDeadLetters GatewaySubmitDeadLetter[] - httpConfig SmsApplicationHttpConfig? - httpIpAllowlist SmsApplicationHttpIpAllowlist[] - httpApiCredentials HttpApiCredential[] - openApiRequests OpenApiRequest[] - httpWebhookEndpoints HttpWebhookEndpoint[] - httpWebhookEvents HttpWebhookEvent[] - dailyUsages SmsApplicationDailyUsage[] - inboundLongMessages CmppInboundLongMessage[] + gatewaySubmitDeadLetters GatewaySubmitDeadLetter[] + httpConfig SmsApplicationHttpConfig? + httpIpAllowlist SmsApplicationHttpIpAllowlist[] + httpApiCredentials HttpApiCredential[] + openApiRequests OpenApiRequest[] + httpWebhookEndpoints HttpWebhookEndpoint[] + httpWebhookEvents HttpWebhookEvent[] + dailyUsages SmsApplicationDailyUsage[] + inboundLongMessages CmppInboundLongMessage[] @@index([tenantId, status]) } @@ -452,31 +452,31 @@ model SmsApplicationIpAllowlist { } model SmsApplicationHttpConfig { - id String @id @default(cuid()) - applicationId String @unique - enabled Boolean @default(false) - sendEnabled Boolean @default(true) - messageQueryEnabled Boolean @default(true) - receiptWebhookEnabled Boolean @default(true) - uplinkWebhookEnabled Boolean @default(true) - uplinkQueryEnabled Boolean @default(true) - credentialSelfServiceEnabled Boolean @default(true) - qpsLimit Int @default(10) - timestampToleranceSeconds Int @default(300) - maxCredentialCount Int @default(2) - uplinkRetentionDays Int @default(90) - maxQueryRangeDays Int @default(31) - maxPageSize Int @default(100) - receiptDeliveryMode String @default("http") - uplinkDeliveryMode String @default("http") - webhookRetryEnabled Boolean @default(true) - webhookMaxAttempts Int @default(7) - webhookTimeoutSeconds Int @default(10) - requireHttps Boolean @default(true) - allowClientManualRetry Boolean @default(true) - allowClientTest Boolean @default(true) - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + applicationId String @unique + enabled Boolean @default(false) + sendEnabled Boolean @default(true) + messageQueryEnabled Boolean @default(true) + receiptWebhookEnabled Boolean @default(true) + uplinkWebhookEnabled Boolean @default(true) + uplinkQueryEnabled Boolean @default(true) + credentialSelfServiceEnabled Boolean @default(true) + qpsLimit Int @default(10) + timestampToleranceSeconds Int @default(300) + maxCredentialCount Int @default(2) + uplinkRetentionDays Int @default(90) + maxQueryRangeDays Int @default(31) + maxPageSize Int @default(100) + receiptDeliveryMode String @default("http") + uplinkDeliveryMode String @default("http") + webhookRetryEnabled Boolean @default(true) + webhookMaxAttempts Int @default(7) + webhookTimeoutSeconds Int @default(10) + requireHttps Boolean @default(true) + allowClientManualRetry Boolean @default(true) + allowClientTest Boolean @default(true) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) } @@ -508,47 +508,47 @@ model SmsApplicationHttpIpAllowlist { } model HttpApiCredential { - id String @id @default(cuid()) + id String @id @default(cuid()) applicationId String name String - accessKey String @unique + accessKey String @unique secretEncrypted String secretLast4 String - status String @default("active") + status String @default("active") expiresAt DateTime? lastUsedAt DateTime? lastUsedIp String? createdById String? revokedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) + application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) requests OpenApiRequest[] @@index([applicationId, status]) } model OpenApiRequest { - id String @id @default(cuid()) - tenantId String - applicationId String - credentialId String - requestId String @unique - idempotencyKey String - bodyHash String - clientMessageId String? - messageRecordId String? - httpStatus Int? - businessCode String? - responseBody Json? - sourceIp String? - userAgent String? - durationMs Int? - status String @default("processing") - completedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String + applicationId String + credentialId String + requestId String @unique + idempotencyKey String + bodyHash String + clientMessageId String? + messageRecordId String? + httpStatus Int? + businessCode String? + responseBody Json? + sourceIp String? + userAgent String? + durationMs Int? + status String @default("processing") + completedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt tenant Tenant @relation(fields: [tenantId], references: [id]) application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) @@ -559,19 +559,19 @@ model OpenApiRequest { } model HttpWebhookEndpoint { - id String @id @default(cuid()) + id String @id @default(cuid()) applicationId String eventType String url String secretEncrypted String secretLast4 String - status String @default("active") + status String @default("active") lastTestAt DateTime? lastTestStatus String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) + application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) deliveries HttpWebhookDelivery[] @@unique([applicationId, eventType]) @@ -589,29 +589,29 @@ model HttpWebhookEvent { payload Json createdAt DateTime @default(now()) - tenant Tenant @relation(fields: [tenantId], references: [id]) - application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) deliveries HttpWebhookDelivery[] @@index([applicationId, createdAt]) } model HttpWebhookDelivery { - id String @id @default(cuid()) + id String @id @default(cuid()) eventId String endpointId String - status String @default("pending") - attemptCount Int @default(0) + status String @default("pending") + attemptCount Int @default(0) nextRetryAt DateTime? lastHttpStatus Int? lastError String? deliveredAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - event HttpWebhookEvent @relation(fields: [eventId], references: [id], onDelete: Cascade) - endpoint HttpWebhookEndpoint @relation(fields: [endpointId], references: [id], onDelete: Cascade) - attempts HttpWebhookAttempt[] + event HttpWebhookEvent @relation(fields: [eventId], references: [id], onDelete: Cascade) + endpoint HttpWebhookEndpoint @relation(fields: [endpointId], references: [id], onDelete: Cascade) + attempts HttpWebhookAttempt[] @@unique([eventId, endpointId]) @@index([status, nextRetryAt]) @@ -634,60 +634,60 @@ model HttpWebhookAttempt { } model SmsSignature { - id String @id @default(cuid()) - tenantId String - applicationId String? - name String - purpose String? - drainageInfo Json? - auditStatus String @default("draft") - reportStatus String @default("waiting_material") - rejectReason String? - materialVersion Int @default(1) - pendingReport Boolean @default(true) + id String @id @default(cuid()) + tenantId String + applicationId String? + name String + purpose String? + drainageInfo Json? + auditStatus String @default("draft") + reportStatus String @default("waiting_material") + rejectReason String? + materialVersion Int @default(1) + pendingReport Boolean @default(true) reportChangedAt DateTime @default(now()) - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant @relation(fields: [tenantId], references: [id]) - application SmsApplication? @relation(fields: [applicationId], references: [id]) - materials SignatureMaterial[] - templates SmsTemplate[] - reportMaterials SignatureReportMaterial[] + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication? @relation(fields: [applicationId], references: [id]) + materials SignatureMaterial[] + templates SmsTemplate[] + reportMaterials SignatureReportMaterial[] drainageReportMaterials DrainageReportMaterial[] - drainageItems SmsDrainageInfo[] - reportTasks ChannelSignatureReportTask[] - messageRecords SmsMessageRecord[] - reportBatchItems ReportMaterialBatchItem[] + drainageItems SmsDrainageInfo[] + reportTasks ChannelSignatureReportTask[] + messageRecords SmsMessageRecord[] + reportBatchItems ReportMaterialBatchItem[] @@index([tenantId, auditStatus]) @@index([tenantId, reportStatus]) } model SmsDrainageInfo { - id String @id @default(cuid()) - tenantId String - signatureId String - applicationId String? - siteName String - url String - remark String? - reportValues Json? - auditStatus String @default("pending") - rejectReason String? - materialVersion Int @default(1) - pendingReport Boolean @default(true) - reportChangedAt DateTime @default(now()) - submittedAt DateTime @default(now()) - reviewedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String + signatureId String + applicationId String? + siteName String + url String + remark String? + reportValues Json? + auditStatus String @default("pending") + rejectReason String? + materialVersion Int @default(1) + pendingReport Boolean @default(true) + reportChangedAt DateTime @default(now()) + submittedAt DateTime @default(now()) + reviewedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant @relation(fields: [tenantId], references: [id]) - signature SmsSignature @relation(fields: [signatureId], references: [id], onDelete: Cascade) - application SmsApplication? @relation(fields: [applicationId], references: [id]) - reportTasks ChannelSignatureReportTask[] - messageRecords SmsMessageRecord[] + tenant Tenant @relation(fields: [tenantId], references: [id]) + signature SmsSignature @relation(fields: [signatureId], references: [id], onDelete: Cascade) + application SmsApplication? @relation(fields: [applicationId], references: [id]) + reportTasks ChannelSignatureReportTask[] + messageRecords SmsMessageRecord[] reportBatchItems ReportMaterialBatchItem[] @@index([tenantId, auditStatus, updatedAt]) @@ -696,13 +696,13 @@ model SmsDrainageInfo { } model SignatureMaterial { - id String @id @default(cuid()) - signatureId String + id String @id @default(cuid()) + signatureId String fileObjectId String? materialType String - title String - description String? - createdAt DateTime @default(now()) + title String + description String? + createdAt DateTime @default(now()) signature SmsSignature @relation(fields: [signatureId], references: [id], onDelete: Cascade) } @@ -721,12 +721,12 @@ model SmsTemplate { createdAt DateTime @default(now()) updatedAt DateTime @updatedAt - tenant Tenant @relation(fields: [tenantId], references: [id]) - application SmsApplication @relation(fields: [applicationId], references: [id]) - signature SmsSignature? @relation(fields: [signatureId], references: [id]) - variables TemplateVariable[] - sendTasks SmsSendTask[] - batchTasks SmsBatchTask[] + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id]) + signature SmsSignature? @relation(fields: [signatureId], references: [id]) + variables TemplateVariable[] + sendTasks SmsSendTask[] + batchTasks SmsBatchTask[] messageRecords SmsMessageRecord[] @@index([tenantId, auditStatus]) @@ -734,12 +734,12 @@ model SmsTemplate { } model TemplateVariable { - id String @id @default(cuid()) - templateId String - name String - example String? - required Boolean @default(true) - createdAt DateTime @default(now()) + id String @id @default(cuid()) + templateId String + name String + example String? + required Boolean @default(true) + createdAt DateTime @default(now()) template SmsTemplate @relation(fields: [templateId], references: [id], onDelete: Cascade) @@ -758,72 +758,72 @@ model AuditRecord { reviewerId String? createdAt DateTime @default(now()) - tenant Tenant? @relation(fields: [tenantId], references: [id]) - reviewer User? @relation(fields: [reviewerId], references: [id]) + tenant Tenant? @relation(fields: [tenantId], references: [id]) + reviewer User? @relation(fields: [reviewerId], references: [id]) @@index([targetType, targetId]) @@index([tenantId, createdAt]) } model SmsChannel { - id String @id @default(cuid()) - code String @unique - name String - carrier String? - sendRegion String @default("全国") - protocol String @default("CMPP") - gatewayHost String - gatewayPort Int - enterpriseCode String? - account String - passwordCipher String - srcId String - cmppVersion String @default("2.0") - rateLimitPerSecond Int @default(100) - unitPrice BigInt @default(0) - status String @default("active") - config Json? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + code String @unique + name String + carrier String? + sendRegion String @default("全国") + protocol String @default("CMPP") + gatewayHost String + gatewayPort Int + enterpriseCode String? + account String + passwordCipher String + srcId String + cmppVersion String @default("2.0") + rateLimitPerSecond Int @default(100) + unitPrice BigInt @default(0) + status String @default("active") + config Json? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - groupItems SmsChannelGroupItem[] - routeRules ChannelRouteRule[] - healthMetrics ChannelHealthMetric[] - reportFields ChannelReportField[] - drainageReportMaterials DrainageReportMaterial[] - reportTasks ChannelSignatureReportTask[] - reportRecords ChannelSignatureReportRecord[] - messageRecords SmsMessageRecord[] - submitSessions CmppSubmitSession[] - submitRecords SmsSubmitRecord[] - receiptRecords SmsReceiptRecord[] - segmentAudits SmsMessageSegmentAudit[] - uplinkMessages SmsUplinkMessage[] - connectionStates CmppConnectionState[] + groupItems SmsChannelGroupItem[] + routeRules ChannelRouteRule[] + healthMetrics ChannelHealthMetric[] + reportFields ChannelReportField[] + drainageReportMaterials DrainageReportMaterial[] + reportTasks ChannelSignatureReportTask[] + reportRecords ChannelSignatureReportRecord[] + messageRecords SmsMessageRecord[] + submitSessions CmppSubmitSession[] + submitRecords SmsSubmitRecord[] + receiptRecords SmsReceiptRecord[] + segmentAudits SmsMessageSegmentAudit[] + uplinkMessages SmsUplinkMessage[] + connectionStates CmppConnectionState[] gatewaySubmitDeadLetters GatewaySubmitDeadLetter[] @@index([status]) } model CmppConnectionState { - id String @id @default(cuid()) - tenantId String? - applicationId String? - channelId String - connectionId String - status String @default("disconnected") - desiredConnections Int @default(1) - currentConnections Int @default(0) - lastConnectedAt DateTime? - lastDisconnectedAt DateTime? - lastHeartbeatAt DateTime? - reconnectCount Int @default(0) + id String @id @default(cuid()) + tenantId String? + applicationId String? + channelId String + connectionId String + status String @default("disconnected") + desiredConnections Int @default(1) + currentConnections Int @default(0) + lastConnectedAt DateTime? + lastDisconnectedAt DateTime? + lastHeartbeatAt DateTime? + reconnectCount Int @default(0) lastReconnectAttemptAt DateTime? - nextReconnectAt DateTime? - lastErrorCategory String? - lastError String? - updatedAt DateTime @updatedAt - createdAt DateTime @default(now()) + nextReconnectAt DateTime? + lastErrorCategory String? + lastError String? + updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) tenant Tenant? @relation(fields: [tenantId], references: [id]) application SmsApplication? @relation(fields: [applicationId], references: [id]) @@ -837,23 +837,23 @@ model CmppConnectionState { } model CmppDownstreamConnection { - id String @id @default(cuid()) - tenantId String - applicationId String - account String - enterpriseCode String - connectionId String @unique - remoteIp String? - protocol String? - status String @default("connected") - connectedAt DateTime @default(now()) - lastHeartbeatAt DateTime? - lastSubmitAt DateTime? - lastDeliverAt DateTime? - disconnectedAt DateTime? - lastError String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String + applicationId String + account String + enterpriseCode String + connectionId String @unique + remoteIp String? + protocol String? + status String @default("connected") + connectedAt DateTime @default(now()) + lastHeartbeatAt DateTime? + lastSubmitAt DateTime? + lastDeliverAt DateTime? + disconnectedAt DateTime? + lastError String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt tenant Tenant @relation(fields: [tenantId], references: [id]) application SmsApplication @relation(fields: [applicationId], references: [id]) @@ -864,17 +864,17 @@ model CmppDownstreamConnection { } model SmsChannelGroup { - id String @id @default(cuid()) - code String @unique - name String - description String? - carrier String @default("mobile") - status String @default("active") - retryEnabled Boolean @default(true) - retryTimeLimitHours Int @default(72) - retryTimeLimitMinutes Int @default(720) - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + code String @unique + name String + description String? + carrier String @default("mobile") + status String @default("active") + retryEnabled Boolean @default(true) + retryTimeLimitHours Int @default(72) + retryTimeLimitMinutes Int @default(720) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt items SmsChannelGroupItem[] routeRules ChannelRouteRule[] @@ -938,27 +938,27 @@ model ChannelHealthMetric { } model ChannelReportField { - id String @id @default(cuid()) - channelId String + id String @id @default(cuid()) + channelId String drainageFieldId String? - reportType String @default("both") - code String - name String - exportName String? - fieldType String - required Boolean @default(false) - description String? - sortOrder Int @default(100) - columnWidth Int @default(18) - imageWidth Int @default(120) - imageHeight Int @default(80) - defaultValue String? - transform String? - status String @default("active") - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + reportType String @default("both") + code String + name String + exportName String? + fieldType String + required Boolean @default(false) + description String? + sortOrder Int @default(100) + columnWidth Int @default(18) + imageWidth Int @default(120) + imageHeight Int @default(80) + defaultValue String? + transform String? + status String @default("active") + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - channel SmsChannel @relation(fields: [channelId], references: [id], onDelete: Cascade) + channel SmsChannel @relation(fields: [channelId], references: [id], onDelete: Cascade) drainageField DrainageField? @relation(fields: [drainageFieldId], references: [id]) @@unique([channelId, code, reportType]) @@ -982,14 +982,14 @@ model CommonReportField { } model SignatureReportMaterial { - id String @id @default(cuid()) - signatureId String - channelId String - fieldCode String - fieldValue String? + id String @id @default(cuid()) + signatureId String + channelId String + fieldCode String + fieldValue String? fileObjectId String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt signature SmsSignature @relation(fields: [signatureId], references: [id], onDelete: Cascade) @@ -998,15 +998,15 @@ model SignatureReportMaterial { } model DrainageReportMaterial { - id String @id @default(cuid()) - signatureId String - drainageItemId String - channelId String - fieldCode String - fieldValue String? - fileObjectId String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + signatureId String + drainageItemId String + channelId String + fieldCode String + fieldValue String? + fileObjectId String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt signature SmsSignature @relation(fields: [signatureId], references: [id], onDelete: Cascade) channel SmsChannel @relation(fields: [channelId], references: [id], onDelete: Cascade) @@ -1016,25 +1016,25 @@ model DrainageReportMaterial { } model ChannelSignatureReportTask { - id String @id @default(cuid()) - tenantId String - signatureId String - channelId String - reportType String @default("signature") + id String @id @default(cuid()) + tenantId String + signatureId String + channelId String + reportType String @default("signature") drainageItemId String? - status String @default("pending") - reason String? - createdById String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + status String @default("pending") + reason String? + createdById String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - signature SmsSignature @relation(fields: [signatureId], references: [id]) - channel SmsChannel @relation(fields: [channelId], references: [id]) - drainageInfo SmsDrainageInfo? @relation(fields: [drainageItemId], references: [id]) - records ChannelSignatureReportRecord[] - exportFiles ReportExportFile[] + signature SmsSignature @relation(fields: [signatureId], references: [id]) + channel SmsChannel @relation(fields: [channelId], references: [id]) + drainageInfo SmsDrainageInfo? @relation(fields: [drainageItemId], references: [id]) + records ChannelSignatureReportRecord[] + exportFiles ReportExportFile[] receiptImports ReportReceiptImport[] - exportItems ReportExportFileItem[] + exportItems ReportExportFileItem[] @@index([tenantId, status]) @@index([signatureId, channelId]) @@ -1043,16 +1043,16 @@ model ChannelSignatureReportTask { } model ChannelSignatureReportRecord { - id String @id @default(cuid()) - taskId String - channelId String - action String + id String @id @default(cuid()) + taskId String + channelId String + action String statusBefore String? statusAfter String - reason String? - operatorId String? - sourceEntry String @default("system") - createdAt DateTime @default(now()) + reason String? + operatorId String? + sourceEntry String @default("system") + createdAt DateTime @default(now()) task ChannelSignatureReportTask @relation(fields: [taskId], references: [id], onDelete: Cascade) channel SmsChannel @relation(fields: [channelId], references: [id]) @@ -1062,33 +1062,33 @@ model ChannelSignatureReportRecord { } model ReportExportFile { - id String @id @default(cuid()) - taskId String? - batchId String? - channelId String? + id String @id @default(cuid()) + taskId String? + batchId String? + channelId String? fileObjectId String? - fileName String - rowCount Int @default(0) - status String @default("generated") - createdAt DateTime @default(now()) + fileName String + rowCount Int @default(0) + status String @default("generated") + createdAt DateTime @default(now()) - task ChannelSignatureReportTask? @relation(fields: [taskId], references: [id], onDelete: Cascade) - batch ReportMaterialBatch? @relation(fields: [batchId], references: [id], onDelete: Cascade) + task ChannelSignatureReportTask? @relation(fields: [taskId], references: [id], onDelete: Cascade) + batch ReportMaterialBatch? @relation(fields: [batchId], references: [id], onDelete: Cascade) items ReportExportFileItem[] @@index([batchId, channelId]) } model ReportMaterialBatch { - id String @id @default(cuid()) - batchNo String @unique - status String @default("generating") + id String @id @default(cuid()) + batchNo String @unique + status String @default("generating") createdById String? - selectedCount Int @default(0) - channelCount Int @default(0) - fileCount Int @default(0) + selectedCount Int @default(0) + channelCount Int @default(0) + fileCount Int @default(0) errorMessage String? - createdAt DateTime @default(now()) + createdAt DateTime @default(now()) completedAt DateTime? items ReportMaterialBatchItem[] @@ -1107,10 +1107,10 @@ model ReportMaterialBatchItem { snapshot Json createdAt DateTime @default(now()) - batch ReportMaterialBatch @relation(fields: [batchId], references: [id], onDelete: Cascade) - signature SmsSignature @relation(fields: [signatureId], references: [id]) - drainageInfo SmsDrainageInfo? @relation(fields: [drainageItemId], references: [id]) - exportItems ReportExportFileItem[] + batch ReportMaterialBatch @relation(fields: [batchId], references: [id], onDelete: Cascade) + signature SmsSignature @relation(fields: [signatureId], references: [id]) + drainageInfo SmsDrainageInfo? @relation(fields: [drainageItemId], references: [id]) + exportItems ReportExportFileItem[] @@index([batchId, reportType]) @@index([signatureId, drainageItemId]) @@ -1123,8 +1123,8 @@ model ReportExportFileItem { taskId String rowNumber Int - exportFile ReportExportFile @relation(fields: [exportFileId], references: [id], onDelete: Cascade) - batchItem ReportMaterialBatchItem @relation(fields: [batchItemId], references: [id], onDelete: Cascade) + exportFile ReportExportFile @relation(fields: [exportFileId], references: [id], onDelete: Cascade) + batchItem ReportMaterialBatchItem @relation(fields: [batchItemId], references: [id], onDelete: Cascade) task ChannelSignatureReportTask @relation(fields: [taskId], references: [id], onDelete: Cascade) @@unique([exportFileId, batchItemId]) @@ -1171,39 +1171,39 @@ model ReportMaterialImportProfileColumn { } model ReportMaterialImportBatch { - id String @id @default(cuid()) - tenantId String - applicationId String? - profileId String? - fileObjectId String - fileName String - reportType String - status String @default("analyzed") - sheetName String - headerRowCount Int @default(1) - dataStartRow Int @default(2) - mapping Json - preview Json? - result Json? - rowCount Int @default(0) - successCount Int @default(0) - failedCount Int @default(0) - createdAt DateTime @default(now()) - completedAt DateTime? + id String @id @default(cuid()) + tenantId String + applicationId String? + profileId String? + fileObjectId String + fileName String + reportType String + status String @default("analyzed") + sheetName String + headerRowCount Int @default(1) + dataStartRow Int @default(2) + mapping Json + preview Json? + result Json? + rowCount Int @default(0) + successCount Int @default(0) + failedCount Int @default(0) + createdAt DateTime @default(now()) + completedAt DateTime? @@index([tenantId, createdAt]) @@index([status, createdAt]) } model ReportReceiptImport { - id String @id @default(cuid()) - taskId String + id String @id @default(cuid()) + taskId String fileObjectId String? - fileName String - rowCount Int @default(0) - successCount Int @default(0) - failedCount Int @default(0) - status String @default("imported") + fileName String + rowCount Int @default(0) + successCount Int @default(0) + failedCount Int @default(0) + status String @default("imported") result Json? createdAt DateTime @default(now()) @@ -1233,13 +1233,13 @@ model RiskRule { } model SmsSendTask { - id String @id @default(cuid()) + id String @id @default(cuid()) tenantId String applicationId String? templateId String? - taskNo String @unique - sourceType String @default("risk") - aggregationKey String? @unique + taskNo String @unique + sourceType String @default("risk") + aggregationKey String? @unique contentHash String? windowStartedAt DateTime? windowEndsAt DateTime? @@ -1247,26 +1247,26 @@ model SmsSendTask { category String? phoneTotal Int uniquePhoneTotal Int - duplicateRatio Float @default(0) - illegalRatio Float @default(0) - blacklistHitRatio Float @default(0) + duplicateRatio Float @default(0) + illegalRatio Float @default(0) + blacklistHitRatio Float @default(0) variableIssues Json? - status String @default("created") - riskDecision String @default("allow") + status String @default("created") + riskDecision String @default("allow") reviewReason String? rejectReason String? createdById String? reviewedById String? reviewedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant @relation(fields: [tenantId], references: [id]) - application SmsApplication? @relation(fields: [applicationId], references: [id]) - template SmsTemplate? @relation(fields: [templateId], references: [id]) - createdBy User? @relation("SmsSendTaskCreator", fields: [createdById], references: [id]) - reviewedBy User? @relation("SmsSendTaskReviewer", fields: [reviewedById], references: [id]) - riskHits RiskHitRecord[] + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication? @relation(fields: [applicationId], references: [id]) + template SmsTemplate? @relation(fields: [templateId], references: [id]) + createdBy User? @relation("SmsSendTaskCreator", fields: [createdById], references: [id]) + reviewedBy User? @relation("SmsSendTaskReviewer", fields: [reviewedById], references: [id]) + riskHits RiskHitRecord[] messageRecords SmsMessageRecord[] @@index([tenantId, status, createdAt]) @@ -1286,9 +1286,9 @@ model RiskHitRecord { reason String createdAt DateTime @default(now()) - tenant Tenant @relation(fields: [tenantId], references: [id]) - task SmsSendTask @relation(fields: [taskId], references: [id], onDelete: Cascade) - rule RiskRule? @relation(fields: [ruleId], references: [id]) + tenant Tenant @relation(fields: [tenantId], references: [id]) + task SmsSendTask @relation(fields: [taskId], references: [id], onDelete: Cascade) + rule RiskRule? @relation(fields: [ruleId], references: [id]) @@index([tenantId, createdAt]) @@index([taskId]) @@ -1296,41 +1296,41 @@ model RiskHitRecord { } model SmsBatchTask { - id String @id @default(cuid()) + id String @id @default(cuid()) tenantId String applicationId String? templateId String? - taskNo String @unique - sourceType String @default("client") + taskNo String @unique + sourceType String @default("client") content String category String? phoneTotal Int - status String @default("created") + status String @default("created") riskTaskId String? - auditStatus String @default("approved") + auditStatus String @default("approved") reviewReason String? rejectReason String? - progressTotal Int @default(0) - submittedTotal Int @default(0) - successTotal Int @default(0) - failedTotal Int @default(0) - unknownTotal Int @default(0) - timeoutTotal Int @default(0) + progressTotal Int @default(0) + submittedTotal Int @default(0) + successTotal Int @default(0) + failedTotal Int @default(0) + unknownTotal Int @default(0) + timeoutTotal Int @default(0) scheduledAt DateTime? canceledAt DateTime? createdById String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant @relation(fields: [tenantId], references: [id]) - application SmsApplication? @relation(fields: [applicationId], references: [id]) - template SmsTemplate? @relation(fields: [templateId], references: [id]) - createdBy User? @relation("SmsBatchTaskCreator", fields: [createdById], references: [id]) - apiRequests SmsApiRequest[] - messages SmsMessageRecord[] - submitRecords SmsSubmitRecord[] + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication? @relation(fields: [applicationId], references: [id]) + template SmsTemplate? @relation(fields: [templateId], references: [id]) + createdBy User? @relation("SmsBatchTaskCreator", fields: [createdById], references: [id]) + apiRequests SmsApiRequest[] + messages SmsMessageRecord[] + submitRecords SmsSubmitRecord[] receiptRecords SmsReceiptRecord[] - segmentAudits SmsMessageSegmentAudit[] + segmentAudits SmsMessageSegmentAudit[] @@index([tenantId, status, createdAt]) @@index([status, scheduledAt]) @@ -1356,76 +1356,76 @@ model SmsApiRequest { } model SmsMessageRecord { - id String @id @default(cuid()) - tenantId String? - batchTaskId String? - applicationId String? - templateId String? - signatureId String? - drainageInfoId String? - reviewTaskId String? - messageId String @unique - clientMessageId String? - phoneNumber String - carrier String? - province String? - content String - billingUnits Int @default(1) - unitPrice BigInt @default(0) - amountCents BigInt @default(0) - queuePriority String @default("normal") - channelId String? - submitId String? - gatewayMessageId String? - cmppSubmitSequenceId String? + id String @id @default(cuid()) + tenantId String? + batchTaskId String? + applicationId String? + templateId String? + signatureId String? + drainageInfoId String? + reviewTaskId String? + messageId String @unique + clientMessageId String? + phoneNumber String + carrier String? + province String? + content String + billingUnits Int @default(1) + unitPrice BigInt @default(0) + amountCents BigInt @default(0) + queuePriority String @default("normal") + channelId String? + submitId String? + gatewayMessageId String? + cmppSubmitSequenceId String? cmppSubmitGroupMessageId String? - clientSrcId String? - applicationExtension String? - status String @default("queued") - submitStatus String? - receiptStatus String? - receiptRawStatus String? - errorCode String? - errorMessage String? - queuedAt DateTime @default(now()) - submittedAt DateTime? - deliveredAt DateTime? - timeoutAt DateTime? - updatedAt DateTime @updatedAt + clientSrcId String? + applicationExtension String? + status String @default("queued") + submitStatus String? + receiptStatus String? + receiptRawStatus String? + errorCode String? + errorMessage String? + queuedAt DateTime @default(now()) + submittedAt DateTime? + deliveredAt DateTime? + timeoutAt DateTime? + updatedAt DateTime @updatedAt - tenant Tenant? @relation(fields: [tenantId], references: [id]) - batchTask SmsBatchTask? @relation(fields: [batchTaskId], references: [id], onDelete: Cascade) - application SmsApplication? @relation(fields: [applicationId], references: [id]) - template SmsTemplate? @relation(fields: [templateId], references: [id]) - signature SmsSignature? @relation(fields: [signatureId], references: [id]) - drainageInfo SmsDrainageInfo? @relation(fields: [drainageInfoId], references: [id]) - reviewTask SmsSendTask? @relation(fields: [reviewTaskId], references: [id]) - channel SmsChannel? @relation(fields: [channelId], references: [id]) - submitRecords SmsSubmitRecord[] - receiptRecords SmsReceiptRecord[] - segmentAudits SmsMessageSegmentAudit[] - matchedUplinks SmsUplinkMessage[] @relation("SmsUplinkMatchedMessage") + tenant Tenant? @relation(fields: [tenantId], references: [id]) + batchTask SmsBatchTask? @relation(fields: [batchTaskId], references: [id], onDelete: Cascade) + application SmsApplication? @relation(fields: [applicationId], references: [id]) + template SmsTemplate? @relation(fields: [templateId], references: [id]) + signature SmsSignature? @relation(fields: [signatureId], references: [id]) + drainageInfo SmsDrainageInfo? @relation(fields: [drainageInfoId], references: [id]) + reviewTask SmsSendTask? @relation(fields: [reviewTaskId], references: [id]) + channel SmsChannel? @relation(fields: [channelId], references: [id]) + submitRecords SmsSubmitRecord[] + receiptRecords SmsReceiptRecord[] + segmentAudits SmsMessageSegmentAudit[] + matchedUplinks SmsUplinkMessage[] @relation("SmsUplinkMatchedMessage") uplinkMatchCandidates SmsUplinkMatchCandidate[] - downstreamDeliveries CmppDownstreamDelivery[] + downstreamDeliveries CmppDownstreamDelivery[] + @@unique([applicationId, clientMessageId]) @@index([tenantId, status, queuedAt]) @@index([batchTaskId, status]) @@index([reviewTaskId, status]) @@index([phoneNumber]) @@index([gatewayMessageId]) @@index([drainageInfoId, queuedAt]) - @@unique([applicationId, clientMessageId]) } model CmppSubmitSession { - id String @id @default(cuid()) + id String @id @default(cuid()) channelId String - sessionNo String @unique - status String @default("open") - submitTotal Int @default(0) - acceptedTotal Int @default(0) - rejectedTotal Int @default(0) - startedAt DateTime @default(now()) + sessionNo String @unique + status String @default("open") + submitTotal Int @default(0) + acceptedTotal Int @default(0) + rejectedTotal Int @default(0) + startedAt DateTime @default(now()) endedAt DateTime? channel SmsChannel @relation(fields: [channelId], references: [id]) @@ -1435,29 +1435,29 @@ model CmppSubmitSession { } model SmsSubmitRecord { - id String @id @default(cuid()) + id String @id @default(cuid()) tenantId String? batchTaskId String? messageRecordId String channelId String sessionId String? - submitId String @unique + submitId String @unique sequenceId Int? gatewayMessageId String? - submitStatus String @default("queued") - costUnitPrice BigInt @default(0) - costAmountCents BigInt @default(0) + submitStatus String @default("queued") + costUnitPrice BigInt @default(0) + costAmountCents BigInt @default(0) errorCode String? errorMessage String? submittedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant? @relation(fields: [tenantId], references: [id]) - batchTask SmsBatchTask? @relation(fields: [batchTaskId], references: [id], onDelete: Cascade) - messageRecord SmsMessageRecord @relation(fields: [messageRecordId], references: [id], onDelete: Cascade) - channel SmsChannel @relation(fields: [channelId], references: [id]) - session CmppSubmitSession? @relation(fields: [sessionId], references: [id]) + tenant Tenant? @relation(fields: [tenantId], references: [id]) + batchTask SmsBatchTask? @relation(fields: [batchTaskId], references: [id], onDelete: Cascade) + messageRecord SmsMessageRecord @relation(fields: [messageRecordId], references: [id], onDelete: Cascade) + channel SmsChannel @relation(fields: [channelId], references: [id]) + session CmppSubmitSession? @relation(fields: [sessionId], references: [id]) segmentAudits SmsMessageSegmentAudit[] @@index([tenantId, createdAt]) @@ -1488,27 +1488,27 @@ model DailyReconciliationReport { } model DailyProfitReport { - id String @id @default(cuid()) - reportDate DateTime @db.Date - dimensionType String - dimensionId String - dimensionName String - tenantId String? - tenantName String? - applicationId String? - channelId String? - submittedUnits Int @default(0) - sentUnits Int @default(0) - unknownUnits Int @default(0) - successUnits Int @default(0) - failedUnits Int @default(0) - revenueCents BigInt @default(0) - refundCents BigInt @default(0) - costCents BigInt @default(0) - profitCents BigInt @default(0) - profitRateBps Int @default(0) - generatedAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + reportDate DateTime @db.Date + dimensionType String + dimensionId String + dimensionName String + tenantId String? + tenantName String? + applicationId String? + channelId String? + submittedUnits Int @default(0) + sentUnits Int @default(0) + unknownUnits Int @default(0) + successUnits Int @default(0) + failedUnits Int @default(0) + revenueCents BigInt @default(0) + refundCents BigInt @default(0) + costCents BigInt @default(0) + profitCents BigInt @default(0) + profitRateBps Int @default(0) + generatedAt DateTime @default(now()) + updatedAt DateTime @updatedAt @@unique([reportDate, dimensionType, dimensionId]) @@index([dimensionType, reportDate]) @@ -1518,26 +1518,26 @@ model DailyProfitReport { } model DailyQualityReport { - id String @id @default(cuid()) - reportDate DateTime @db.Date - dimensionType String - dimensionId String - dimensionName String - tenantId String? - tenantName String? - applicationId String? - channelId String? - signatureId String? - drainageInfoId String? - submittedUnits Int @default(0) - sentUnits Int @default(0) - unknownUnits Int @default(0) - successUnits Int @default(0) - failedUnits Int @default(0) - successRateBps Int @default(0) - avgArrivalMs Int? - generatedAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + reportDate DateTime @db.Date + dimensionType String + dimensionId String + dimensionName String + tenantId String? + tenantName String? + applicationId String? + channelId String? + signatureId String? + drainageInfoId String? + submittedUnits Int @default(0) + sentUnits Int @default(0) + unknownUnits Int @default(0) + successUnits Int @default(0) + failedUnits Int @default(0) + successRateBps Int @default(0) + avgArrivalMs Int? + generatedAt DateTime @default(now()) + updatedAt DateTime @updatedAt @@unique([reportDate, dimensionType, dimensionId]) @@index([dimensionType, reportDate, sentUnits]) @@ -1549,19 +1549,19 @@ model DailyQualityReport { } model SmsMessageSegmentAudit { - id String @id @default(cuid()) + id String @id @default(cuid()) tenantId String? batchTaskId String? messageRecordId String submitRecordId String? channelId String? submitId String - attempt Int @default(0) - segmentTotal Int @default(1) - segmentIndex Int @default(1) + attempt Int @default(0) + segmentTotal Int @default(1) + segmentIndex Int @default(1) sequenceId Int? gatewayMessageId String? - submitStatus String @default("queued") + submitStatus String @default("queued") receiptStatus String? rawStatus String? compensationType String? @@ -1569,8 +1569,8 @@ model SmsMessageSegmentAudit { errorMessage String? submittedAt DateTime? deliveredAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt tenant Tenant? @relation(fields: [tenantId], references: [id]) batchTask SmsBatchTask? @relation(fields: [batchTaskId], references: [id]) @@ -1587,7 +1587,7 @@ model SmsMessageSegmentAudit { } model CmppInboundLongMessage { - id String @id @default(cuid()) + id String @id @default(cuid()) tenantId String applicationId String groupKey String @@ -1597,16 +1597,16 @@ model CmppInboundLongMessage { concatReference Int segmentTotal Int msgFmt Int - messageId String @unique - status String @default("collecting") + messageId String @unique + status String @default("collecting") response Json? expiresAt DateTime completedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant @relation(fields: [tenantId], references: [id]) - application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Cascade) segments CmppInboundLongMessageSegment[] @@index([groupKey, status, createdAt]) @@ -1614,14 +1614,14 @@ model CmppInboundLongMessage { } model CmppInboundLongMessageSegment { - id String @id @default(cuid()) - groupId String - segmentIndex Int - sequenceId String? - content String - contentHash String - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + groupId String + segmentIndex Int + sequenceId String? + content String + contentHash String + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt group CmppInboundLongMessage @relation(fields: [groupId], references: [id], onDelete: Cascade) @@ -1660,25 +1660,25 @@ model SmsReceiptRecord { } model SmsUplinkMessage { - id String @id @default(cuid()) - tenantId String? - applicationId String? - channelId String - messageRecordId String? - messageId String? - sequenceId Int? - phoneNumber String - destId String - content String - matchStatus String @default("unmatched") - matchReason String? - receivedAt DateTime - createdAt DateTime @default(now()) + id String @id @default(cuid()) + tenantId String? + applicationId String? + channelId String + messageRecordId String? + messageId String? + sequenceId Int? + phoneNumber String + destId String + content String + matchStatus String @default("unmatched") + matchReason String? + receivedAt DateTime + createdAt DateTime @default(now()) - tenant Tenant? @relation(fields: [tenantId], references: [id]) - application SmsApplication? @relation(fields: [applicationId], references: [id]) - messageRecord SmsMessageRecord? @relation("SmsUplinkMatchedMessage", fields: [messageRecordId], references: [id]) - channel SmsChannel @relation(fields: [channelId], references: [id]) + tenant Tenant? @relation(fields: [tenantId], references: [id]) + application SmsApplication? @relation(fields: [applicationId], references: [id]) + messageRecord SmsMessageRecord? @relation("SmsUplinkMatchedMessage", fields: [messageRecordId], references: [id]) + channel SmsChannel @relation(fields: [channelId], references: [id]) matchCandidates SmsUplinkMatchCandidate[] @@index([tenantId, createdAt]) @@ -1690,23 +1690,23 @@ model SmsUplinkMessage { } model SmsUplinkMatchCandidate { - id String @id @default(cuid()) + id String @id @default(cuid()) uplinkMessageId String tenantId String applicationId String messageRecordId String? matchSource String - confidence Int @default(50) + confidence Int @default(50) reason String? - status String @default("pending") + status String @default("pending") claimedAt DateTime? claimedById String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - uplinkMessage SmsUplinkMessage @relation(fields: [uplinkMessageId], references: [id], onDelete: Cascade) - tenant Tenant @relation(fields: [tenantId], references: [id]) - application SmsApplication @relation(fields: [applicationId], references: [id]) + uplinkMessage SmsUplinkMessage @relation(fields: [uplinkMessageId], references: [id], onDelete: Cascade) + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id]) messageRecord SmsMessageRecord? @relation(fields: [messageRecordId], references: [id]) @@unique([uplinkMessageId, applicationId, messageRecordId]) @@ -1717,34 +1717,35 @@ model SmsUplinkMatchCandidate { } model CmppDownstreamDelivery { - id String @id @default(cuid()) - tenantId String - applicationId String - messageRecordId String? - messageId String? - deliveryType String - status String @default("pending") - payload Json - retryEnabled Boolean @default(true) - retryCount Int @default(0) - manualRetryCount Int @default(0) - lastRetriedAt DateTime? - nextRetryAt DateTime? - sentAt DateTime? - acknowledgedAt DateTime? - ackDeadlineAt DateTime? - ackResult Int? - ackSequenceId String? - ackMessageId String? - connectionId String? - deliveredAt DateTime? - lastError String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + tenantId String + applicationId String + messageRecordId String? + messageId String? + deliveryType String + status String @default("pending") + payload Json + retryEnabled Boolean @default(true) + retryCount Int @default(0) + manualRetryCount Int @default(0) + lastRetriedAt DateTime? + nextRetryAt DateTime? + sentAt DateTime? + acknowledgedAt DateTime? + ackDeadlineAt DateTime? + ackResult Int? + ackSequenceId String? + ackMessageId String? + connectionId String? + deliveredAt DateTime? + lastError String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt - tenant Tenant @relation(fields: [tenantId], references: [id]) - application SmsApplication @relation(fields: [applicationId], references: [id]) - messageRecord SmsMessageRecord? @relation(fields: [messageRecordId], references: [id]) + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id]) + messageRecord SmsMessageRecord? @relation(fields: [messageRecordId], references: [id]) + attempts CmppDownstreamDeliveryAttempt[] @@index([tenantId, status, createdAt]) @@index([applicationId, status, createdAt]) @@ -1754,29 +1755,91 @@ model CmppDownstreamDelivery { @@index([status, ackDeadlineAt]) } +model CmppDownstreamDeliveryAttempt { + id String @id @default(cuid()) + deliveryId String + attemptKey String @unique + attemptNo Int + connectionId String? + sequenceId String? + messageId String? + status String + sentAt DateTime? + ackDeadlineAt DateTime? + acknowledgedAt DateTime? + ackResult Int? + failureType String? + errorMessage String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + delivery CmppDownstreamDelivery @relation(fields: [deliveryId], references: [id], onDelete: Cascade) + + @@index([deliveryId, attemptNo]) + @@index([status, ackDeadlineAt]) + @@index([connectionId, sequenceId]) +} + +model UpstreamReceiptInbox { + id String @id @default(cuid()) + receiptKey String @unique + incomingChannelId String + incomingConnectionId String? + upstreamAccount String + upstreamHost String + upstreamPort Int + protocol String + protocolVersion String + provisionalMessageId String? + sequenceId Int? + gatewayMessageId String + phoneNumber String? + receiptStatus String + rawStatus String + errorCode String? + errorMessage String? + deliveredAt DateTime + receivedAt DateTime @default(now()) + status String @default("pending") + matchedMessageRecordId String? + matchedSubmitRecordId String? + matchedChannelId String? + attemptCount Int @default(0) + nextRetryAt DateTime? + lastError String? + processedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@index([status, nextRetryAt, receivedAt]) + @@index([gatewayMessageId, phoneNumber]) + @@index([incomingChannelId, receivedAt]) + @@index([matchedMessageRecordId]) +} + model GatewaySubmitDeadLetter { - id String @id @default(cuid()) - streamMessageId String @unique - tenantId String? - applicationId String? - channelId String? - traceId String? - messageId String? - submitId String? - status String @default("pending") - failureCode String - failureMessage String - attempts Int @default(1) - maxAttempts Int @default(3) - commandPayload Json? - rawPayload String? - manualRetryCount Int @default(0) - lastRetryStreamId String? - lastRetriedAt DateTime? - resolvedAt DateTime? - resolvedStatus String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + streamMessageId String @unique + tenantId String? + applicationId String? + channelId String? + traceId String? + messageId String? + submitId String? + status String @default("pending") + failureCode String + failureMessage String + attempts Int @default(1) + maxAttempts Int @default(3) + commandPayload Json? + rawPayload String? + manualRetryCount Int @default(0) + lastRetryStreamId String? + lastRetriedAt DateTime? + resolvedAt DateTime? + resolvedStatus String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt tenant Tenant? @relation(fields: [tenantId], references: [id]) application SmsApplication? @relation(fields: [applicationId], references: [id]) @@ -1791,8 +1854,8 @@ model GatewaySubmitDeadLetter { } model GatewayDownstreamRecoveryStatus { - id String @id @default(cuid()) - account String @unique + id String @id @default(cuid()) + account String @unique tenantId String? applicationId String? gatewayInstanceId String? @@ -1803,12 +1866,12 @@ model GatewayDownstreamRecoveryStatus { lastSuccessAt DateTime? lastFailureAt DateTime? nextRetryAt DateTime? - attemptCount Int @default(0) + attemptCount Int @default(0) failureCategory String? lastError String? lastSkipReason String? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt tenant Tenant? @relation(fields: [tenantId], references: [id]) application SmsApplication? @relation(fields: [applicationId], references: [id]) diff --git a/api/src/operations/operations.service.spec.ts b/api/src/operations/operations.service.spec.ts index 5f4c404..54aa4d3 100644 --- a/api/src/operations/operations.service.spec.ts +++ b/api/src/operations/operations.service.spec.ts @@ -668,7 +668,12 @@ describe('OperationsService', () => { lte: new Date('2026-07-15T23:59:59.999+08:00'), }, }), - include: { tenant: true, application: true, messageRecord: true }, + include: { + tenant: true, + application: true, + messageRecord: true, + attempts: { orderBy: [{ attemptNo: 'desc' }, { createdAt: 'desc' }] }, + }, orderBy: { createdAt: 'desc' }, skip: 0, take: 10, diff --git a/api/src/operations/operations.service.ts b/api/src/operations/operations.service.ts index 6e67942..1a3dfd2 100644 --- a/api/src/operations/operations.service.ts +++ b/api/src/operations/operations.service.ts @@ -814,7 +814,12 @@ export class OperationsService { const [items, total] = await Promise.all([ this.prisma.cmppDownstreamDelivery.findMany({ where, - include: { tenant: true, application: true, messageRecord: true }, + include: { + tenant: true, + application: true, + messageRecord: true, + attempts: { orderBy: [{ attemptNo: 'desc' }, { createdAt: 'desc' }] }, + }, orderBy: { createdAt: 'desc' }, skip: (page - 1) * pageSize, take: pageSize, diff --git a/api/src/send-chain/gateway-events.controller.spec.ts b/api/src/send-chain/gateway-events.controller.spec.ts index ae2a585..a67f1ce 100644 --- a/api/src/send-chain/gateway-events.controller.spec.ts +++ b/api/src/send-chain/gateway-events.controller.spec.ts @@ -4,7 +4,9 @@ import { GatewayEventsController } from './gateway-events.controller'; describe('GatewayEventsController protocol logging', () => { const sendChain = { handleSubmitResult: jest.fn(), + handleSubmitSegmentResult: jest.fn(), handleReceipt: jest.fn(), + intakeReceipt: jest.fn(), }; const protocolLogs = { record: jest.fn(), @@ -30,6 +32,51 @@ describe('GatewayEventsController protocol logging', () => { expect(protocolLogs.record).not.toHaveBeenCalled(); }); + it('persists each supplier segment response without adding a duplicate protocol log', async () => { + sendChain.handleSubmitSegmentResult.mockResolvedValue({ accepted: true }); + const body = { + messageId: 'MSG-LONG-1', + channelId: 'channel-1', + submitId: 'submit-1', + segmentIndex: 2, + segmentTotal: 3, + gatewayMessageId: '456', + sequenceId: 8, + submitStatus: 'accepted' as const, + }; + + await expect(controller.submitSegmentResult(body)).resolves.toEqual({ accepted: true }); + expect(sendChain.handleSubmitSegmentResult).toHaveBeenCalledWith(body); + expect(protocolLogs.record).not.toHaveBeenCalled(); + }); + + it('logs a supplier receipt only after it is durably accepted by the inbox', async () => { + sendChain.intakeReceipt.mockResolvedValue({ accepted: true, inboxId: 'inbox-1', status: 'pending' }); + const body = { + messageId: 'receipt-456', + channelId: 'channel-1', + connectionId: 'channel-1:0', + gatewayMessageId: '456', + phoneNumber: '13127620092', + receiptStatus: 'undelivered' as const, + rawStatus: 'UNDELIV', + }; + + await expect(controller.receiptIntake(body)).resolves.toEqual({ + accepted: true, + inboxId: 'inbox-1', + status: 'pending', + }); + expect(protocolLogs.record).toHaveBeenCalledWith(expect.objectContaining({ + direction: 'channel_to_platform', + eventType: 'deliver_receipt', + channelId: 'channel-1', + messageId: 'receipt-456', + gatewayMessageId: '456', + status: 'success', + })); + }); + it('accepts only safe outbound Gateway packet events', () => { expect(controller.protocolLog({ protocol: 'cmpp', diff --git a/api/src/send-chain/gateway-events.controller.ts b/api/src/send-chain/gateway-events.controller.ts index 43b8ff6..4646a29 100644 --- a/api/src/send-chain/gateway-events.controller.ts +++ b/api/src/send-chain/gateway-events.controller.ts @@ -11,6 +11,7 @@ import { GatewayReceiptEventDto, GatewaySubmitDeadLetterDto, GatewaySubmitResultDto, + GatewaySubmitSegmentResultDto, GatewayUplinkEventDto, SendChainService, } from './send-chain.service'; @@ -31,6 +32,16 @@ export class GatewayEventsController { return this.sendChain.handleSubmitResult(body); } + @Post('submit-segment-result') + submitSegmentResult(@Body() body: GatewaySubmitSegmentResultDto) { + return this.sendChain.handleSubmitSegmentResult(body); + } + + @Post('receipt/intake') + receiptIntake(@Body() body: GatewayReceiptEventDto) { + return this.trackGatewayEvent('deliver_receipt', body, () => this.sendChain.intakeReceipt(body)); + } + @Post('receipt') receipt(@Body() body: GatewayReceiptEventDto) { return this.trackGatewayEvent('deliver_receipt', body, () => this.sendChain.handleReceipt(body)); @@ -104,8 +115,8 @@ export class GatewayEventsController { } @Post('downstream/failed') - downstreamFailed(@Body() body: { id: string; errorMessage?: string; failureType?: GatewayDownstreamFailureType }) { - return this.sendChain.markDownstreamDeliveryFailed(body.id, body.errorMessage, body.failureType); + downstreamFailed(@Body() body: GatewayDownstreamSentDto & { errorMessage?: string; failureType?: GatewayDownstreamFailureType }) { + return this.sendChain.markDownstreamDeliveryFailed(body.id, body.errorMessage, body.failureType, body); } @Post('downstream/recovery-status') diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index b89c999..a81fff4 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -36,6 +36,7 @@ function createPrismaMock() { sendRegion: '全国', gatewayHost: '127.0.0.1', gatewayPort: 17890, + protocol: 'CMPP', passwordCipher: 'secret', cmppVersion: '3.0', config: { serviceId: 'SMS' }, @@ -240,6 +241,22 @@ function createPrismaMock() { update: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1', deliveryType: 'receipt', ...data })), updateMany: jest.fn().mockResolvedValue({ count: 1 }), }, + cmppDownstreamDeliveryAttempt: { + upsert: jest.fn().mockResolvedValue({ id: 'delivery-attempt-1' }), + findMany: jest.fn().mockResolvedValue([]), + }, + upstreamReceiptInbox: { + upsert: jest.fn().mockImplementation(({ create }) => Promise.resolve({ + id: 'receipt-inbox-1', + attemptCount: 0, + receivedAt: new Date(), + ...create, + })), + findMany: jest.fn().mockResolvedValue([]), + findUnique: jest.fn().mockResolvedValue(null), + update: jest.fn().mockImplementation(({ data }) => Promise.resolve({ id: 'receipt-inbox-1', ...data })), + updateMany: jest.fn().mockResolvedValue({ count: 1 }), + }, gatewaySubmitDeadLetter: { upsert: jest.fn().mockResolvedValue({ id: 'dead-1' }), findUnique: jest.fn().mockResolvedValue({ @@ -1765,8 +1782,8 @@ describe('SendChainService', () => { where: { submitId: 'SUB-1' }, data: expect.objectContaining({ sequenceId: 7, gatewayMessageId: 'GW-1', submitStatus: 'accepted' }), }); - expect(prisma.smsMessageRecord.update).toHaveBeenCalledWith({ - where: { id: 'record-1' }, + expect(prisma.smsMessageRecord.updateMany).toHaveBeenCalledWith({ + 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' })); @@ -1824,8 +1841,8 @@ describe('SendChainService', () => { expect(prisma.channelRouteRule.findFirst).not.toHaveBeenCalled(); expect(billing.release).not.toHaveBeenCalled(); expect(prisma.smsBatchTask.update).not.toHaveBeenCalled(); - expect(prisma.smsMessageRecord.update).toHaveBeenCalledWith({ - where: { id: 'record-1' }, + expect(prisma.smsMessageRecord.updateMany).toHaveBeenCalledWith({ + where: { id: 'record-1', status: { not: 'delivered' } }, data: expect.objectContaining({ gatewayMessageId: 'GW-1', submitStatus: 'timeout', @@ -2680,6 +2697,108 @@ describe('SendChainService', () => { })); }); + it('persists each upstream SubmitResp segment before the aggregate result arrives', async () => { + const { service, prisma } = createService(); + + await service.handleSubmitSegmentResult({ + messageId: 'MSG-1', + channelId: 'channel-1', + submitId: 'SUB-1', + segmentTotal: 3, + segmentIndex: 1, + sequenceId: 71, + gatewayMessageId: 'GW-SEG-1', + submitStatus: 'accepted', + submittedAt: '2026-07-25T15:00:00.000Z', + }); + + expect(prisma.smsMessageSegmentAudit.upsert).toHaveBeenCalledWith(expect.objectContaining({ + where: { + messageRecordId_submitId_segmentIndex: { + messageRecordId: 'record-1', + submitId: 'SUB-1', + segmentIndex: 1, + }, + }, + create: expect.objectContaining({ + segmentTotal: 3, + sequenceId: 71, + gatewayMessageId: 'GW-SEG-1', + submitStatus: 'accepted', + }), + })); + expect(prisma.smsSubmitRecord.updateMany).toHaveBeenCalledWith(expect.objectContaining({ + where: { submitId: 'SUB-1', gatewayMessageId: null }, + data: expect.objectContaining({ sequenceId: 71, gatewayMessageId: 'GW-SEG-1' }), + })); + }); + + it('durably intakes an upstream receipt before asynchronous business matching', async () => { + const { service, prisma } = createService(); + jest.spyOn(service as any, 'processUpstreamReceiptInboxRecord').mockResolvedValue(false); + + await expect(service.intakeReceipt({ + messageId: 'receipt-9001', + channelId: 'channel-1', + connectionId: 'gateway-connection-2', + sequenceId: 81, + gatewayMessageId: '9001', + phoneNumber: '13800000001', + receiptStatus: 'delivered', + rawStatus: 'DELIVRD', + deliveredAt: '2026-07-25T15:01:00.000Z', + })).resolves.toEqual(expect.objectContaining({ + accepted: true, + inboxId: 'receipt-inbox-1', + })); + + expect(prisma.upstreamReceiptInbox.upsert).toHaveBeenCalledWith(expect.objectContaining({ + create: expect.objectContaining({ + incomingChannelId: 'channel-1', + incomingConnectionId: 'gateway-connection-2', + upstreamAccount: 'cmpp-account', + upstreamHost: '127.0.0.1', + upstreamPort: 17890, + protocol: 'CMPP', + protocolVersion: '3.0', + gatewayMessageId: '9001', + status: 'pending', + }), + })); + }); + + it('does not downgrade an early terminal receipt when the aggregate submit result arrives later', async () => { + const { service, prisma } = createService(); + const terminalMessage = { + id: 'record-1', messageId: 'MSG-1', tenantId: null, batchTaskId: null, applicationId: null, + channelId: 'channel-1', submitId: 'SUB-1', gatewayMessageId: 'GW-SEG-1', + phoneNumber: '13800000001', billingUnits: 1, amountCents: 0, status: 'failed', + }; + prisma.smsMessageRecord.findFirst.mockResolvedValue(terminalMessage); + prisma.smsMessageRecord.findUnique.mockResolvedValue(terminalMessage); + prisma.smsMessageRecord.updateMany + .mockResolvedValueOnce({ count: 0 }) + .mockResolvedValueOnce({ count: 1 }); + + await service.handleSubmitResult({ + messageId: 'MSG-1', + channelId: 'channel-1', + submitId: 'SUB-1', + sequenceId: 7, + gatewayMessageId: 'GW-1', + submitStatus: 'accepted', + }); + + expect(prisma.smsMessageRecord.updateMany).toHaveBeenNthCalledWith(1, expect.objectContaining({ + where: { id: 'record-1', status: { notIn: ['delivered', 'failed', 'unknown'] } }, + data: expect.objectContaining({ status: 'submitted' }), + })); + expect(prisma.smsMessageRecord.updateMany).toHaveBeenNthCalledWith(2, { + where: { id: 'record-1', gatewayMessageId: null }, + data: expect.objectContaining({ gatewayMessageId: 'GW-1' }), + }); + }); + it('recovers a stale submit requeue with the same Redis idempotency key', async () => { const { service, prisma } = createService(); const stale = { @@ -2929,6 +3048,15 @@ describe('SendChainService', () => { where: { id: 'delivery-1', status: { not: 'delivered' } }, data: expect.objectContaining({ status: 'awaiting_ack', ackSequenceId: '37', connectionId: 'conn-1' }), })); + expect(prisma.cmppDownstreamDeliveryAttempt.upsert).toHaveBeenLastCalledWith(expect.objectContaining({ + create: expect.objectContaining({ + deliveryId: 'delivery-1', + attemptNo: 1, + connectionId: 'conn-1', + sequenceId: '37', + status: 'awaiting_ack', + }), + })); await service.acknowledgeDownstreamDelivery({ id: 'delivery-1', @@ -2941,6 +3069,13 @@ describe('SendChainService', () => { expect(prisma.cmppDownstreamDelivery.updateMany).toHaveBeenLastCalledWith(expect.objectContaining({ data: expect.objectContaining({ status: 'delivered', ackResult: 0, deliveredAt: new Date('2026-07-14T03:40:18.060Z') }), })); + expect(prisma.cmppDownstreamDeliveryAttempt.upsert).toHaveBeenLastCalledWith(expect.objectContaining({ + update: expect.objectContaining({ + status: 'acknowledged', + acknowledgedAt: new Date('2026-07-14T03:40:18.060Z'), + ackResult: 0, + }), + })); }); it('does not treat Result=0 with Msg_Id=0 as a business acknowledgement', async () => { diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 6308196..7ed2692 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -89,6 +89,21 @@ export interface GatewaySubmitResultDto { }>; } +export interface GatewaySubmitSegmentResultDto { + traceId?: string; + messageId: string; + channelId: string; + submitId?: string; + segmentTotal: number; + segmentIndex: number; + sequenceId?: number; + gatewayMessageId?: string; + submitStatus: 'accepted' | 'rejected' | 'timeout' | string; + errorCode?: string; + errorMessage?: string; + submittedAt?: string; +} + export interface GatewayReceiptEventDto { traceId?: string; messageId?: string; @@ -101,6 +116,7 @@ export interface GatewayReceiptEventDto { errorCode?: string; errorMessage?: string; deliveredAt?: string; + connectionId?: string; } export interface GatewayUplinkEventDto { @@ -269,6 +285,11 @@ const DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS = 2 * 60_000; const DEFAULT_INBOUND_LONG_MESSAGE_SCAN_INTERVAL_MS = 60_000; const INBOUND_LONG_MESSAGE_SCAN_INITIAL_DELAY_MS = 10_000; const DEFAULT_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS = 30; +const DEFAULT_UPSTREAM_RECEIPT_INBOX_SCAN_INTERVAL_MS = 5_000; +const UPSTREAM_RECEIPT_INBOX_INITIAL_DELAY_MS = 1_000; +const DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS = 2 * 60_000; +const DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS = 30; +const DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS = 72; const GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS = 30 * 24 * 60 * 60; const BULLMQ_PRIORITY: Record = { priority: 1, @@ -290,6 +311,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { private scheduledDispatchScanRunning = false; private inboundLongMessageInitialTimer?: ReturnType; private inboundLongMessageIntervalTimer?: ReturnType; + private upstreamReceiptInboxInitialTimer?: ReturnType; + private upstreamReceiptInboxIntervalTimer?: ReturnType; + private upstreamReceiptInboxScanRunning = false; constructor( private readonly prisma: PrismaService, @@ -342,6 +366,21 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { ); this.inboundLongMessageIntervalTimer.unref?.(); } + if (process.env.UPSTREAM_RECEIPT_INBOX_SCAN_ENABLED !== 'false') { + this.upstreamReceiptInboxInitialTimer = setTimeout( + () => void this.runUpstreamReceiptInboxScan(), + UPSTREAM_RECEIPT_INBOX_INITIAL_DELAY_MS, + ); + this.upstreamReceiptInboxInitialTimer.unref?.(); + this.upstreamReceiptInboxIntervalTimer = setInterval( + () => void this.runUpstreamReceiptInboxScan(), + positiveInteger( + process.env.UPSTREAM_RECEIPT_INBOX_SCAN_INTERVAL_MS, + DEFAULT_UPSTREAM_RECEIPT_INBOX_SCAN_INTERVAL_MS, + ), + ); + this.upstreamReceiptInboxIntervalTimer.unref?.(); + } } async onModuleDestroy() { @@ -351,6 +390,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (this.scheduledDispatchIntervalTimer) clearInterval(this.scheduledDispatchIntervalTimer); if (this.inboundLongMessageInitialTimer) clearTimeout(this.inboundLongMessageInitialTimer); if (this.inboundLongMessageIntervalTimer) clearInterval(this.inboundLongMessageIntervalTimer); + if (this.upstreamReceiptInboxInitialTimer) clearTimeout(this.upstreamReceiptInboxInitialTimer); + if (this.upstreamReceiptInboxIntervalTimer) clearInterval(this.upstreamReceiptInboxIntervalTimer); await this.worker?.close(); await this.sendQueue?.close(); await this.gatewayQueue?.close(); @@ -907,6 +948,46 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { } } + async handleSubmitSegmentResult(data: GatewaySubmitSegmentResultDto) { + const message = await this.requireMessageByGatewayEvent(data.messageId, data.gatewayMessageId); + const submittedAt = data.submittedAt ? new Date(data.submittedAt) : new Date(); + await this.recordSubmitSegments(message, { + messageId: data.messageId, + channelId: data.channelId, + submitId: data.submitId, + sequenceId: data.sequenceId, + gatewayMessageId: data.gatewayMessageId ?? '', + submitStatus: normalizeSubmitStatus(data.submitStatus), + errorCode: data.errorCode, + errorMessage: data.errorMessage, + submittedAt: submittedAt.toISOString(), + segments: [{ + segmentTotal: data.segmentTotal, + segmentIndex: data.segmentIndex, + sequenceId: data.sequenceId, + gatewayMessageId: data.gatewayMessageId, + submitStatus: data.submitStatus, + errorCode: data.errorCode, + errorMessage: data.errorMessage, + submittedAt: submittedAt.toISOString(), + }], + }, submittedAt); + if (data.gatewayMessageId) { + await this.prisma.smsSubmitRecord.updateMany({ + where: { + ...(data.submitId ? { submitId: data.submitId } : { messageRecordId: message.id }), + gatewayMessageId: null, + }, + data: { + sequenceId: data.sequenceId, + gatewayMessageId: data.gatewayMessageId, + submittedAt, + }, + }); + } + return { accepted: true }; + } + async handleSubmitResult(data: GatewaySubmitResultDto) { const message = await this.requireMessageByGatewayEvent(data.messageId, data.gatewayMessageId); const batchTask = message.batchTaskId @@ -933,6 +1014,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { if (data.submitStatus === 'accepted' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) { const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string }; await this.chargeAcceptedMessage(businessMessage); + const latest = await this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); + if (latest?.status === 'failed') { + await this.refundMessage(businessMessage, '先到失败回执补偿退款'); + } } else if (data.submitStatus !== 'accepted' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) { const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string }; const retried = await this.retryMessageIfAllowed(businessMessage, data.submitStatus === 'timeout' ? '提交超时补发' : '提交失败补发'); @@ -942,8 +1027,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { } await this.releaseMessageReservation(businessMessage, data.submitStatus === 'timeout' ? '提交超时释放冻结' : '提交失败释放冻结'); } - await this.prisma.smsMessageRecord.update({ - where: { id: message.id }, + const protectedTerminalStatuses = ['delivered', 'failed', 'unknown']; + const updated = await this.prisma.smsMessageRecord.updateMany({ + where: data.submitStatus === 'accepted' + ? { id: message.id, status: { notIn: protectedTerminalStatuses } } + : { id: message.id, status: { not: 'delivered' } }, data: { gatewayMessageId: data.gatewayMessageId, submitStatus: data.submitStatus, @@ -954,6 +1042,15 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { timeoutAt: data.submitStatus === 'timeout' ? submittedAt : undefined, }, }); + if (updated.count === 0 && data.submitStatus === 'accepted') { + await this.prisma.smsMessageRecord.updateMany({ + where: { id: message.id, gatewayMessageId: null }, + data: { + gatewayMessageId: data.gatewayMessageId, + submittedAt, + }, + }); + } if (data.submitStatus !== 'accepted' && batchTask?.sourceType === 'cmpp' && message.tenantId && message.applicationId) { await this.recordCmppFailureReceipt( message as typeof message & { tenantId: string; applicationId: string; batchTaskId: string }, @@ -981,8 +1078,185 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); } - async handleReceipt(data: GatewayReceiptEventDto) { - const resolved = await this.resolveReceiptMessage(data); + async intakeReceipt(data: GatewayReceiptEventDto) { + const channel = await this.prisma.smsChannel.findUnique({ + where: { id: data.channelId }, + select: { + id: true, + account: true, + gatewayHost: true, + gatewayPort: true, + protocol: true, + cmppVersion: true, + }, + }); + if (!channel) { + throw new NotFoundException('SMS channel not found'); + } + const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date(); + const receiptKey = this.receiptEventKey(data, data.channelId); + const inbox = await this.prisma.upstreamReceiptInbox.upsert({ + where: { receiptKey }, + update: { + incomingConnectionId: data.connectionId, + }, + create: { + receiptKey, + incomingChannelId: data.channelId, + incomingConnectionId: data.connectionId, + upstreamAccount: channel.account, + upstreamHost: channel.gatewayHost, + upstreamPort: channel.gatewayPort, + protocol: channel.protocol, + protocolVersion: channel.cmppVersion, + provisionalMessageId: data.messageId, + sequenceId: data.sequenceId, + gatewayMessageId: data.gatewayMessageId, + phoneNumber: data.phoneNumber?.trim() || null, + receiptStatus: data.receiptStatus, + rawStatus: data.rawStatus, + errorCode: data.errorCode, + errorMessage: data.errorMessage, + deliveredAt, + status: 'pending', + nextRetryAt: new Date(), + }, + }); + if (['pending', 'retrying'].includes(inbox.status)) { + setImmediate(() => void this.processUpstreamReceiptInboxRecord(inbox.id)); + } + return { accepted: true, inboxId: inbox.id, status: inbox.status }; + } + + async processPendingUpstreamReceiptInbox(limit = 100) { + const now = new Date(); + const staleBefore = new Date( + now.getTime() + - positiveInteger( + process.env.UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS, + DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS, + ), + ); + const candidates = await this.prisma.upstreamReceiptInbox.findMany({ + where: { + OR: [ + { + status: { in: ['pending', 'retrying'] }, + OR: [{ nextRetryAt: null }, { nextRetryAt: { lte: now } }], + }, + { status: 'processing', updatedAt: { lte: staleBefore } }, + ], + }, + orderBy: [{ receivedAt: 'asc' }, { id: 'asc' }], + take: Math.min(Math.max(limit, 1), 500), + select: { id: true }, + }); + let processed = 0; + for (const candidate of candidates) { + if (await this.processUpstreamReceiptInboxRecord(candidate.id)) processed += 1; + } + return { scanned: candidates.length, processed }; + } + + private async processUpstreamReceiptInboxRecord(id: string) { + const staleBefore = new Date( + Date.now() + - positiveInteger( + process.env.UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS, + DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS, + ), + ); + const claimed = await this.prisma.upstreamReceiptInbox.updateMany({ + where: { + id, + OR: [ + { status: { in: ['pending', 'retrying'] } }, + { status: 'processing', updatedAt: { lte: staleBefore } }, + ], + }, + data: { status: 'processing', attemptCount: { increment: 1 }, nextRetryAt: null }, + }); + if (claimed.count !== 1) return false; + const inbox = await this.prisma.upstreamReceiptInbox.findUnique({ where: { id } }); + if (!inbox) return false; + try { + const message = await this.handleReceipt({ + messageId: inbox.provisionalMessageId ?? undefined, + channelId: inbox.incomingChannelId, + connectionId: inbox.incomingConnectionId ?? undefined, + sequenceId: inbox.sequenceId ?? undefined, + gatewayMessageId: inbox.gatewayMessageId, + phoneNumber: inbox.phoneNumber ?? undefined, + receiptStatus: normalizeReceiptStatus(inbox.receiptStatus), + rawStatus: inbox.rawStatus, + errorCode: inbox.errorCode ?? undefined, + errorMessage: inbox.errorMessage ?? undefined, + deliveredAt: inbox.deliveredAt.toISOString(), + }, { + account: inbox.upstreamAccount, + gatewayHost: inbox.upstreamHost, + gatewayPort: inbox.upstreamPort, + protocol: inbox.protocol, + cmppVersion: inbox.protocolVersion, + }); + const matchedMessageRecordId = message && 'id' in message ? message.id : message?.messageRecordId; + const matchedChannelId = message && 'channelId' in message ? message.channelId : undefined; + await this.prisma.upstreamReceiptInbox.update({ + where: { id }, + data: { + status: 'matched', + matchedMessageRecordId: matchedMessageRecordId ?? null, + matchedChannelId: matchedChannelId ?? null, + lastError: null, + processedAt: new Date(), + }, + }); + return true; + } catch (error) { + const maxAttempts = positiveInteger( + process.env.UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS, + DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS, + ); + const maxAgeHours = positiveInteger( + process.env.UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS, + DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS, + ); + const exhausted = inbox.attemptCount >= maxAttempts + || inbox.receivedAt.getTime() <= Date.now() - maxAgeHours * 60 * 60_000; + const retryDelayMs = Math.min(30 * 60_000, 5_000 * 2 ** Math.min(Math.max(inbox.attemptCount - 1, 0), 8)); + await this.prisma.upstreamReceiptInbox.update({ + where: { id }, + data: { + status: exhausted ? 'unmatched' : 'retrying', + nextRetryAt: exhausted ? null : new Date(Date.now() + retryDelayMs), + lastError: error instanceof Error ? error.message : String(error), + processedAt: exhausted ? new Date() : null, + }, + }); + return false; + } + } + + private async runUpstreamReceiptInboxScan() { + if (this.upstreamReceiptInboxScanRunning) return; + this.upstreamReceiptInboxScanRunning = true; + try { + const result = await this.processPendingUpstreamReceiptInbox(); + if (result.processed > 0) { + this.logger.log(`Matched ${result.processed}/${result.scanned} pending upstream receipts`); + } + } catch (error) { + this.logger.error('Upstream receipt inbox scan failed', error instanceof Error ? error.stack : String(error)); + } finally { + this.upstreamReceiptInboxScanRunning = false; + } + } + + async handleReceipt( + data: GatewayReceiptEventDto, + incomingIdentity?: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string }, + ) { + const resolved = await this.resolveReceiptMessage(data, incomingIdentity); const logicalChannelId = resolved.channelId ?? data.channelId; const receiptKey = this.receiptEventKey(data, logicalChannelId); const existingReceipt = await this.prisma.smsReceiptRecord.findUnique({ @@ -1202,6 +1476,32 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { async markDownstreamDeliverySent(data: GatewayDownstreamSentDto) { const sentAt = asDateOrNull(data.sentAt) ?? new Date(); const ackDeadlineAt = asDateOrNull(data.ackDeadlineAt) ?? new Date(sentAt.getTime() + downstreamAckTimeoutMs()); + const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: data.id } }); + if (!delivery) { + throw new NotFoundException('Downstream delivery not found'); + } + const attemptKey = downstreamDeliveryAttemptKey(data); + await this.prisma.cmppDownstreamDeliveryAttempt.upsert({ + where: { attemptKey }, + update: { + connectionId: data.connectionId, + sequenceId: data.sequenceId, + messageId: data.messageId, + sentAt, + ackDeadlineAt, + }, + create: { + deliveryId: data.id, + attemptKey, + attemptNo: delivery.retryCount + 1, + connectionId: data.connectionId, + sequenceId: data.sequenceId, + messageId: data.messageId, + status: 'awaiting_ack', + sentAt, + ackDeadlineAt, + }, + }); await this.prisma.cmppDownstreamDelivery.updateMany({ where: { id: data.id, status: { not: 'delivered' } }, data: { @@ -1221,7 +1521,48 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { async acknowledgeDownstreamDelivery(data: GatewayDownstreamAcknowledgedDto) { const acknowledgedAt = asDateOrNull(data.acknowledgedAt) ?? new Date(); const acknowledgedMessageId = String(data.messageId ?? '').trim(); - if (data.result === 0 && acknowledgedMessageId !== '' && acknowledgedMessageId !== '0') { + const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: data.id } }); + if (!delivery) { + throw new NotFoundException('Downstream delivery not found'); + } + const attemptKey = downstreamDeliveryAttemptKey(data); + const acknowledgementAccepted = data.result === 0 && acknowledgedMessageId !== '' && acknowledgedMessageId !== '0'; + await this.prisma.cmppDownstreamDeliveryAttempt.upsert({ + where: { attemptKey }, + update: { + status: acknowledgementAccepted ? 'acknowledged' : 'rejected', + connectionId: data.connectionId, + sequenceId: data.sequenceId, + messageId: data.messageId, + acknowledgedAt, + ackResult: data.result, + ackDeadlineAt: null, + failureType: acknowledgementAccepted ? null : data.result === 0 ? 'ack_invalid' : 'ack_rejected', + errorMessage: acknowledgementAccepted + ? null + : data.result === 0 + ? 'CMPP_DELIVER_RESP Msg_Id=0' + : `CMPP_DELIVER_RESP result=${data.result}`, + }, + create: { + deliveryId: data.id, + attemptKey, + attemptNo: delivery.retryCount + 1, + connectionId: data.connectionId, + sequenceId: data.sequenceId, + messageId: data.messageId, + status: acknowledgementAccepted ? 'acknowledged' : 'rejected', + acknowledgedAt, + ackResult: data.result, + failureType: acknowledgementAccepted ? null : data.result === 0 ? 'ack_invalid' : 'ack_rejected', + errorMessage: acknowledgementAccepted + ? null + : data.result === 0 + ? 'CMPP_DELIVER_RESP Msg_Id=0' + : `CMPP_DELIVER_RESP result=${data.result}`, + }, + }); + if (acknowledgementAccepted) { await this.prisma.cmppDownstreamDelivery.updateMany({ where: { id: data.id, status: { not: 'delivered' } }, data: { @@ -1255,7 +1596,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return this.markDownstreamDeliveryFailed(data.id, `downstream CMPP_DELIVER_RESP result=${data.result}`, 'ack_rejected'); } - async markDownstreamDeliveryFailed(id: string, errorMessage?: string, failureType: GatewayDownstreamFailureType = 'send_failed') { + async markDownstreamDeliveryFailed( + id: string, + errorMessage?: string, + failureType: GatewayDownstreamFailureType = 'send_failed', + attempt?: GatewayDownstreamSentDto, + ) { const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({ where: { id } }); if (!delivery) { throw new NotFoundException('Downstream delivery not found'); @@ -1272,6 +1618,33 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { const nonRetryableFailure = failureType === 'unrecoverable' || failureType === 'queue_timeout'; const finalFailure = nonRetryableFailure || !retryAllowed || retryCount >= downstreamMaxRetries(); const finalStatus = failureType === 'ack_rejected' ? 'rejected' : acknowledgementFailure ? 'unconfirmed' : 'failed'; + if (attempt && (attempt.connectionId || attempt.sequenceId || attempt.messageId)) { + const attemptKey = downstreamDeliveryAttemptKey({ ...attempt, id }); + await this.prisma.cmppDownstreamDeliveryAttempt.upsert({ + where: { attemptKey }, + update: { + status: 'failed', + connectionId: attempt.connectionId, + sequenceId: attempt.sequenceId, + messageId: attempt.messageId, + failureType, + errorMessage: errorMessage ?? 'downstream delivery failed', + ackDeadlineAt: null, + }, + create: { + deliveryId: id, + attemptKey, + attemptNo: delivery.retryCount + 1, + connectionId: attempt.connectionId, + sequenceId: attempt.sequenceId, + messageId: attempt.messageId, + status: 'failed', + sentAt: asDateOrNull(attempt.sentAt), + failureType, + errorMessage: errorMessage ?? 'downstream delivery failed', + }, + }); + } const updated = await this.prisma.cmppDownstreamDelivery.update({ where: { id }, data: { @@ -1914,11 +2287,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { ) as GatewayControlDeliveryResult; if (result.sent || result.delivered) { await this.markDownstreamDeliverySent({ id: delivery.id, ...result }); + } else if (result.reasonCode === 'SUBMIT_RESPONSE_PENDING') { + return delivery; } else { await this.markDownstreamDeliveryFailed( delivery.id, downstreamControlFailureMessage(result), result.retryable === false ? 'unrecoverable' : 'send_failed', + { id: delivery.id, ...result }, ); } } catch (error) { @@ -3831,7 +4207,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return message; } - private async resolveReceiptMessage(data: GatewayReceiptEventDto) { + private async resolveReceiptMessage( + data: GatewayReceiptEventDto, + incomingIdentity?: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string }, + ) { const exactMessage = data.messageId ? await this.prisma.smsMessageRecord.findUnique({ where: { messageId: data.messageId } }) : null; @@ -3896,7 +4275,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { throw new NotFoundException('SMS message record not found'); } - const incomingChannel = await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } }); + const incomingChannel = incomingIdentity + ?? await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } }); if (!incomingChannel) { throw new NotFoundException('SMS message record not found'); } @@ -3920,7 +4300,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }; } const sameSupplierSegments = segmentMatches.filter((candidate) => - candidate.channel && this.isSameSupplierConnection(incomingChannel, candidate.channel)); + candidate.channel && this.isSameUpstreamEndpointIdentity(incomingChannel, candidate.channel)); if (sameSupplierSegments.length === 1 && sameSupplierSegments[0]?.messageRecord) { return { message: sameSupplierSegments[0].messageRecord, @@ -3940,7 +4320,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { take: 10, }); const sameSupplierSubmits = crossConnectionSubmits.filter((candidate) => - candidate.channel && this.isSameSupplierConnection(incomingChannel, candidate.channel)); + candidate.channel && this.isSameUpstreamEndpointIdentity(incomingChannel, candidate.channel)); if (sameSupplierSubmits.length === 1 && sameSupplierSubmits[0]?.messageRecord) { return { message: sameSupplierSubmits[0].messageRecord, @@ -3988,7 +4368,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { }; } - private isSameSupplierConnection( + private isSameUpstreamEndpointIdentity( left: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string }, right: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string }, ) { @@ -4341,6 +4721,24 @@ function parseOptionalSequenceId(value: string | null | undefined) { return Number.isInteger(parsed) && parsed >= 0 && parsed <= 0xffffffff ? parsed : undefined; } +function normalizeSubmitStatus(value: string): GatewaySubmitResultDto['submitStatus'] { + return value === 'accepted' || value === 'rejected' || value === 'timeout' ? value : 'timeout'; +} + +function normalizeReceiptStatus(value: string): GatewayReceiptEventDto['receiptStatus'] { + return value === 'delivered' || value === 'undelivered' || value === 'unknown' ? value : 'unknown'; +} + +function downstreamDeliveryAttemptKey(data: GatewayDownstreamSentDto) { + return createHash('sha256').update([ + data.id, + data.connectionId ?? '', + data.sequenceId ?? '', + data.messageId ?? '', + data.sequenceId ? '' : data.sentAt ?? '', + ].join('\u0000')).digest('hex'); +} + function shanghaiDateKey(now = new Date()) { const parts = new Intl.DateTimeFormat('en-CA', { timeZone: 'Asia/Shanghai', diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 2b6cec6..8da57b8 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -311,6 +311,10 @@ - 下游状态必须以客户确认作为终态：Gateway `SendPkt` 成功后只能写 `awaiting_ack`，仅收到匹配连接、`Sequence_Id`、`Msg_Id` 且 `CMPP_DELIVER_RESP.Result=0` 后才能写 `delivered`；超时、非零 Result 和历史未留存 ACK 的记录分别按未确认、拒绝或历史未确认展示，不能再把 TCP 写出冒充客户已收到。 - 企业应用必须分别提供“回执自动重试投递”和“上行短信自动重试投递”开关，默认开启。开关按投递创建时快照保存；关闭只阻止已经写出但未获 ACK/被拒绝后的自动重发，不阻止离线队列在客户首次上线时完成首次投递。手工重投不受开关限制，但必须提示重复业务处理风险并二次确认。 - 客户 Submit 的失败状态回执必须严格晚于对应 `CMPP_SUBMIT_RESP` 写出，且 Deliver 中的业务 `Msg_Id` 必须非 0、与该 SubmitResp 返回的 `Msg_Id` 完全一致；`Result=0` 但 `Msg_Id=0` 只能表示客户端协议栈收包，不能标记业务回执已确认。平台必须持久化原 Submit Sequence_Id，使 Gateway 重启或客户重连后的补投仍可重建相同业务 `Msg_Id`。 +- 每一次客户侧 Deliver 投递都必须单独持久化发送时间、物理连接 ID、`Sequence_Id`、业务 `Msg_Id`、ACK 截止时间和 `DELIVER_RESP` 结果；运营端详情可按尝试查看，不得只保留最后一次结果而覆盖历史。客户连接关闭时必须清理该连接的发送时序状态；无法匹配的 `DELIVER_RESP` 也必须写通讯日志。 +- 及时失败回执必须使用“当前客户物理连接上的 SubmitResp 写出屏障”：从开始处理 Submit 到该包 `CMPP_SUBMIT_RESP` 实际写出前，同一连接不得下发该 Submit 产生的失败 Deliver；长短信以最终分片 SubmitResp 写出为释放时点。屏障只控制协议写出顺序，不修改通道路由、连接配置或连接池。 +- 供应商 Deliver Receipt 必须先持久化到幂等收件箱，再返回成功 `CMPP_DELIVER_RESP`；持久化失败时不得向供应商虚假确认成功。收件箱异步匹配内部短信，支持失败退避、最大尝试/存留时间和进程异常后的 `processing` 超时恢复。 +- 上游长短信必须在每个分片收到 `SubmitResp` 后立即保存该分片的 `Sequence_Id`、供应商 `Msg_Id`、提交状态和时间，不能等待全部分片完成才批量落库；后到的提交成功聚合结果不得覆盖已经由早到失败回执形成的终态。 - 运营端允许对 `delivered`（客户端已确认）记录再次手工重投，但单条和批量入口都必须明确提示可能造成下游重复处理；`awaiting_ack` 状态在确认窗口内不得并发重投。 - 已实现客户侧最终 Deliver 推送的第一版能力：Gateway 在下游 Submit 被接受后记录 messageId 到客户连接的内存映射；NestJS 收到最终 receipt/uplink 并入库后调用 Gateway `/downstream/receipt`、`/downstream/uplink`，Gateway 向仍在线的客户 CMPP 连接下发 Deliver Receipt 或普通 Deliver。 - 已实现客户侧 Deliver 持久化第一版能力：NestJS 收到最终 receipt/uplink 后写入 `CmppDownstreamDelivery` 待投递记录；在线推送成功标记 delivered，客户断线或 Gateway 不可达时保留 pending 并记录 retry 信息；客户重新 bind 后 Gateway 按账号拉取 pending 记录补发。 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 71a00af..5764a5f 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -2440,3 +2440,14 @@ git diff --check - 发布后Gateway、API、Nginx、PostgreSQL和MinIO均active，Redis PONG；`12026/17890/8090/3000/6379/5432/9000`均监听，API/Gateway health正常。`gateway.submit.commands`消费者1、`pending=0`、`lag=0`，3条active供应商通道均为`connected/currentConnections=1/desiredConnections=1`；API、Gateway和Nginx自发布以来关键错误匹配为0。 - 直接调用已部署的真实`OperationsService`及PostgreSQL验证新SQL：`2026-07-24`为总量13、成功2、未知6、失败5、成功率15.4%，返回4个真实企业应用名称、4个通道和5个签名；切换`2026-07-23`为总量8、成功3、未知1、失败4、成功率37.5%，返回2个企业应用、3个通道和1个签名，证明汇总、排行、通道和签名均随所选日期切换。 - 公网首页、运营登录页、客户端登录页和API health均HTTP 200，公网CMPP 17890 TCP连接成功。浏览器运营登录页标题正确、1280px视口无横向溢出、控制台0条error/warn；当前无已登录会话且页面存在图形验证码，未绕过验证码，因此登录后两张全宽签名表和统计图的最终视觉验收保留为人工登录复核项，不将源码/构建结果冒充登录后页面验收。 + +## 2026-07-25 上下游回执可靠性与逐次投递审计（发布前） + +- 下游增加物理连接级 SubmitResp 写出屏障：从客户 Submit 开始处理到对应响应包真正写出前，同一连接产生的及时失败回执保持 pending；长短信在最终分片 SubmitResp 写出后立即补投，避免客户先收到 Deliver、后收到最后一片 SubmitResp。连接关闭会清理屏障，防止异常连接残留。 +- 每次下游 Deliver 单独写入 `CmppDownstreamDeliveryAttempt`，保存投递次数、连接 ID、`Sequence_Id`、业务 `Msg_Id`、发送/ACK 时间、ACK Result、失败类别和错误；主记录继续承载当前状态和退避调度。运营端下游投递详情新增逐次投递记录，便于区分“平台写出、客户 ACK、ACK 超时和重投”。 +- 上游长短信改为每片 `SubmitResp` 到达后立即回传 API 并写 `SmsMessageSegmentAudit`，最终聚合结果只作兜底；若早到供应商失败回执已经形成终态，后到 accepted 聚合不得把主记录倒退到 submitted，并对先扣后退场景保持幂等补偿。 +- 供应商 Deliver Receipt 改为 API `UpstreamReceiptInbox` 幂等持久化成功后才返回 `CMPP_DELIVER_RESP Result=0`。异步工作器基于保存的通道端点身份匹配短信，失败指数退避，默认最多 30 次/72 小时；API 进程在 processing 中重启时，默认 2 分钟后可重新认领，不依赖单个 Gateway 连接的内存映射。 +- 通讯日志覆盖供应商 Submit/SubmitResp、Deliver Receipt/DeliverResp、客户 Submit/SubmitResp、平台 Deliver/客户 DeliverResp；无法匹配当前 ACK tracker 的客户 `DELIVER_RESP` 也记录为失败通讯事件。内部逐分片 HTTP 回调失败另写 Gateway 结构化本地日志，不把内部回调伪装成 CMPP 报文。 +- 新增 migration `20260725160000_add_reliable_receipt_delivery_tracking`，仅新增下游逐次投递表、上游回执收件箱及索引/外键，不改写既有短信、提交、回执或投递历史。回滚必须先停止新版本 API/Gateway，再删除两张新表；回滚会丢失新版本产生的逐次投递和待匹配收件箱证据。 +- 发布前门禁通过：Prisma format/generate/validate，API 定向 3 suites / 119 tests，API 全量 26 suites / 325 tests，API TypeScript build，前端 TypeScript/Vite生产构建及 Gateway `go test ./...`。API 全量仅保留既有 Redis 不可用容错告警和 `--forceExit` 异步句柄提示；前端保留既有约 1.94 MB 单 chunk 警告。`git diff --check`通过。 +- 本节当前为发布前记录；提交、推送、数据库/源码/环境备份、migration、服务重启和预发布只读验收结果在发布完成后追加。未经额外授权不发送真实短信，不改写历史业务记录。 diff --git a/gateway/internal/inbound/server.go b/gateway/internal/inbound/server.go index 6ca4dec..8b5d061 100644 --- a/gateway/internal/inbound/server.go +++ b/gateway/internal/inbound/server.go @@ -209,6 +209,11 @@ var downstreamAckRegistry = struct { items map[string]*downstreamAckTracker }{items: make(map[string]*downstreamAckTracker)} +var downstreamSubmitBarrier = struct { + sync.RWMutex + byConn map[*cmpp.Conn]int +}{byConn: make(map[*cmpp.Conn]int)} + func (s Server) ListenAndServe() error { addr := s.Addr if addr == "" { @@ -336,6 +341,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge } contentHash := fmt.Sprintf("%x", md5.Sum([]byte(content))) startedAt := time.Now() + releaseSubmitBarrier := beginDownstreamSubmitBarrier(packet.Conn) result, err := s.submit(remote, submitRequest{ Account: account, PhoneNumber: phone, @@ -361,7 +367,11 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge clientProtocol, req.protocol, account, remote, req.sequenceID, phone, responseResult, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, reason, ) setInboundSubmitResponse(response.Packer, 0, responseResult) - response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult) + protocolLogger := s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult) + response.AfterSend = func(sendErr error) { + releaseSubmitBarrier() + protocolLogger(sendErr) + } return false, nil } gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID) @@ -401,6 +411,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge go current.report(current, "submit", "") } response.AfterSend = func(sendErr error) { + releaseSubmitBarrier() s.emitProtocolLog(protocolLogEvent{ Protocol: "cmpp", Direction: "platform_to_client", @@ -538,6 +549,9 @@ func (s Server) reportConnectionOrDisconnect(session *downstreamSession, status } func (s Server) handleConnectionClosed(conn *cmpp.Conn) { + downstreamSubmitBarrier.Lock() + delete(downstreamSubmitBarrier.byConn, conn) + downstreamSubmitBarrier.Unlock() session := findSessionByConn(conn) if session == nil { return @@ -705,10 +719,10 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe if err != nil { result.FailedCount++ result.LastError = err.Error() - _ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]string{ - "id": delivery.ID, - "errorMessage": err.Error(), - "failureType": "send_failed", + _ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{ + "id": delivery.ID, "errorMessage": err.Error(), "failureType": "send_failed", + "connectionId": sendResult.ConnectionID, "sequenceId": sendResult.SequenceID, + "messageId": sendResult.MessageID, "sentAt": sendResult.SentAt, }, nil) continue } @@ -716,6 +730,10 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe result.DeliveredCount++ continue } + if sendResult.ReasonCode == "SUBMIT_RESPONSE_PENDING" { + result.WaitingCount++ + continue + } errorMessage := defaultString(sendResult.ErrorMessage, "gateway did not complete downstream delivery") failureType := "unrecoverable" if sendResult.Retryable { @@ -725,10 +743,10 @@ func (s Server) flushPending(account string, logger *log.Logger) (pendingFlushRe 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, + _ = s.post(context.Background(), "/gateway/events/downstream/failed", map[string]any{ + "id": delivery.ID, "errorMessage": errorMessageWithCode(errorMessage, sendResult.ReasonCode), + "failureType": failureType, "connectionId": sendResult.ConnectionID, + "sequenceId": sendResult.SequenceID, "messageId": sendResult.MessageID, "sentAt": sendResult.SentAt, }, nil) } return result, nil @@ -1153,6 +1171,13 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr ErrorMessage: "下游客户端当前未连接，等待自动重试", }, nil } + if downstreamSubmitResponsePending(session.conn) { + return DownstreamSendResult{ + Retryable: true, + ReasonCode: "SUBMIT_RESPONSE_PENDING", + ErrorMessage: "客户 SubmitResp 尚未完成写出，回执已保留并等待响应后投递", + }, nil + } stat := strings.TrimSpace(event.RawStatus) if stat == "" { stat = cmppReceiptStatus(event.ReceiptStatus) @@ -1179,6 +1204,36 @@ func pushReceiptWithResult(event DownstreamReceipt, allowRecovery bool) (Downstr return sendDownstream(session, deliver, event.DeliveryID) } +func beginDownstreamSubmitBarrier(conn *cmpp.Conn) func() { + if conn == nil { + return func() {} + } + downstreamSubmitBarrier.Lock() + downstreamSubmitBarrier.byConn[conn]++ + downstreamSubmitBarrier.Unlock() + var once sync.Once + return func() { + once.Do(func() { + downstreamSubmitBarrier.Lock() + if downstreamSubmitBarrier.byConn[conn] <= 1 { + delete(downstreamSubmitBarrier.byConn, conn) + } else { + downstreamSubmitBarrier.byConn[conn]-- + } + downstreamSubmitBarrier.Unlock() + }) + } +} + +func downstreamSubmitResponsePending(conn *cmpp.Conn) bool { + if conn == nil { + return false + } + downstreamSubmitBarrier.RLock() + defer downstreamSubmitBarrier.RUnlock() + return downstreamSubmitBarrier.byConn[conn] > 0 +} + func findReceiptSession(messageID string, account string) *downstreamSession { downstreamRegistry.RLock() defer downstreamRegistry.RUnlock() @@ -1288,6 +1343,13 @@ func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID sequenceID := <-session.conn.SeqId sentAt := time.Now().UTC() ackDeadlineAt := sentAt.Add(downstreamAckTimeout()) + result := DownstreamSendResult{ + ConnectionID: session.connectionID, + SequenceID: strconv.FormatUint(uint64(sequenceID), 10), + MessageID: strconv.FormatUint(messageID, 10), + SentAt: formatRFC3339Nano(sentAt), + AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt), + } tracker := registerDownstreamAck(session, deliveryID, sequenceID, messageID, ackDeadlineAt) if err := session.conn.SendPkt(deliver, sequenceID); err != nil { removeDownstreamAck(tracker) @@ -1296,14 +1358,13 @@ func sendDownstream(session *downstreamSession, deliver cmpp.Packer, deliveryID go session.report(session, "disconnected", err.Error()) } forgetDownstream(session) - return DownstreamSendResult{}, err + result.Retryable = true + result.ReasonCode = "SEND_FAILED" + result.ErrorMessage = err.Error() + return result, nil } session.recordDownstreamProtocol(deliver, deliveryID, sequenceID, messageID, "success", "", nil) - result := DownstreamSendResult{ - Sent: true, ConnectionID: session.connectionID, - SequenceID: strconv.FormatUint(uint64(sequenceID), 10), MessageID: strconv.FormatUint(messageID, 10), - SentAt: formatRFC3339Nano(sentAt), AckDeadlineAt: formatRFC3339Nano(ackDeadlineAt), - } + result.Sent = true if deliveryID != "" && session.deliveryReport != nil { go session.deliveryReport(downstreamDeliveryLifecycleEvent{ Kind: "sent", DeliveryID: deliveryID, ConnectionID: session.connectionID, @@ -1433,6 +1494,22 @@ func handleDownstreamAcknowledgement(conn *cmpp.Conn, sequenceID uint32, message if logger != nil { logger.Printf("cmpp inbound event=deliver_ack_unmatched seq=%d message_id=%d result=%d", sequenceID, messageID, result) } + if session := findSessionByConn(conn); session != nil && session.protocolLog != nil { + session.protocolLog(protocolLogEvent{ + Protocol: "cmpp", + Direction: "client_to_platform", + EventType: "deliver_resp", + Status: "failed", + TenantID: session.tenantID, + ApplicationID: session.applicationID, + Account: session.account, + MessageID: session.messageID, + GatewayMessageID: strconv.FormatUint(messageID, 10), + Phone: session.phoneNumber, + ResultCode: strconv.FormatUint(uint64(result), 10), + Detail: map[string]any{"sequenceId": sequenceID, "unmatched": true}, + }) + } return } if tracker.messageID != messageID { diff --git a/gateway/internal/inbound/server_test.go b/gateway/internal/inbound/server_test.go index a44f977..4d01dab 100644 --- a/gateway/internal/inbound/server_test.go +++ b/gateway/internal/inbound/server_test.go @@ -549,6 +549,41 @@ func TestRecoverableReceiptWaitsForClientConnection(t *testing.T) { } } +func TestReceiptWaitsUntilCurrentSubmitResponseHasBeenWritten(t *testing.T) { + resetDownstreamRegistry() + defer resetDownstreamRegistry() + + conn := &cmpp.Conn{} + session := &downstreamSession{ + messageID: "MSG-LONG-1", + account: "100001", + conn: conn, + } + downstreamRegistry.Lock() + downstreamRegistry.byMessageID[session.messageID] = session + downstreamRegistry.byAccount[session.account] = session + downstreamRegistry.byConn[conn] = session + downstreamRegistry.Unlock() + + release := beginDownstreamSubmitBarrier(conn) + result, err := PushReceiptWithResult(DownstreamReceipt{ + DeliveryID: "delivery-long-failed", + Account: session.account, + MessageID: session.messageID, + ReceiptStatus: "undelivered", + }) + if err != nil { + t.Fatalf("push guarded receipt: %v", err) + } + if result.Sent || !result.Retryable || result.ReasonCode != "SUBMIT_RESPONSE_PENDING" { + t.Fatalf("unexpected guarded result: %+v", result) + } + release() + if downstreamSubmitResponsePending(conn) { + t.Fatal("submit response barrier remained active after release") + } +} + func TestInboundServerNegotiatesCMPP2AndUsesAuthenticatedAccountForSubmit(t *testing.T) { resetDownstreamRegistry() defer resetDownstreamRegistry() @@ -1058,6 +1093,9 @@ func resetDownstreamRegistry() { } downstreamAckRegistry.items = make(map[string]*downstreamAckTracker) downstreamAckRegistry.Unlock() + downstreamSubmitBarrier.Lock() + downstreamSubmitBarrier.byConn = make(map[*cmpp.Conn]int) + downstreamSubmitBarrier.Unlock() } func reserveTCPAddr(t *testing.T) string { diff --git a/gateway/internal/queue/messages.go b/gateway/internal/queue/messages.go index 48eccb2..a3cb0ba 100644 --- a/gateway/internal/queue/messages.go +++ b/gateway/internal/queue/messages.go @@ -108,6 +108,7 @@ type ReceiptEvent struct { RawStatus string `json:"rawStatus"` ErrorCode string `json:"errorCode,omitempty"` DeliveredAt time.Time `json:"deliveredAt"` + ConnectionID string `json:"connectionId,omitempty"` } type UplinkEvent struct { diff --git a/gateway/internal/upstream/deliver_test.go b/gateway/internal/upstream/deliver_test.go index ce0a4ba..15ef770 100644 --- a/gateway/internal/upstream/deliver_test.go +++ b/gateway/internal/upstream/deliver_test.go @@ -15,7 +15,7 @@ import ( func TestHandleCMPP2DeliverReceiptPostsReceiptEvent(t *testing.T) { events := make(chan queue.ReceiptEvent, 1) api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path != "/gateway/events/receipt" { + if r.URL.Path != "/gateway/events/receipt/intake" { t.Fatalf("unexpected path: %s", r.URL.Path) } var event queue.ReceiptEvent @@ -58,12 +58,14 @@ func TestHandleCMPP2DeliverReceiptPostsReceiptEvent(t *testing.T) { }, }, } - conn.handleDeliver(deliverPacketFromCMPP2(&cmpp.Cmpp2DeliverReqPkt{ + if err := conn.handleDeliver(deliverPacketFromCMPP2(&cmpp.Cmpp2DeliverReqPkt{ SeqId: 7, MsgId: 999, RegisterDelivery: 1, MsgContent: string(payload), - })) + })); err != nil { + t.Fatalf("handle receipt: %v", err) + } select { case event := <-events: @@ -82,6 +84,9 @@ func TestHandleCMPP2DeliverReceiptPostsReceiptEvent(t *testing.T) { if event.ReceiptStatus != "delivered" || event.RawStatus != "DELIVRD" { t.Fatalf("unexpected receipt status: %+v", event) } + if event.ConnectionID != "channel-1-0" { + t.Fatalf("ConnectionID = %q, want channel-1-0", event.ConnectionID) + } case <-time.After(time.Second): t.Fatal("timed out waiting for receipt event") } diff --git a/gateway/internal/upstream/manager.go b/gateway/internal/upstream/manager.go index 6c632e4..c230290 100644 --- a/gateway/internal/upstream/manager.go +++ b/gateway/internal/upstream/manager.go @@ -73,7 +73,26 @@ func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.Su return result, err } - result, err := pool.submit(ctx, cmd) + result, err := pool.submit(ctx, cmd, func(segment queue.SubmitSegmentResult) { + payload := struct { + queue.Envelope + SubmitID string `json:"submitId,omitempty"` + queue.SubmitSegmentResult + }{ + Envelope: cmd.Envelope, + SubmitID: cmd.SubmitID, + SubmitSegmentResult: segment, + } + callbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + postErr := m.post(callbackCtx, "/gateway/events/submit-segment-result", payload) + cancel() + if postErr != nil { + log.Printf( + "protocol_event protocol=cmpp direction=gateway_to_api event=submit_segment_result status=forward_failed channel_id=%s message_id=%s segment=%d/%d error=%q", + cmd.ChannelID, cmd.MessageID, segment.SegmentIndex, segment.SegmentTotal, postErr, + ) + } + }) if err != nil { if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil { return result, postErr @@ -404,7 +423,11 @@ func (p *connectionPool) superviseReconnects() { } } -func (p *connectionPool) submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) { +func (p *connectionPool) submit( + ctx context.Context, + cmd queue.SubmitCommand, + onSegment func(queue.SubmitSegmentResult), +) (queue.SubmitResult, error) { parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content) if err != nil { result := submitResult(cmd, 0, "", "rejected", "ENCODE_FAILED", err.Error()) @@ -423,7 +446,11 @@ func (p *connectionPool) submit(ctx context.Context, cmd queue.SubmitCommand) (q } seq, gatewayMessageID, result, err := conn.submitPart(ctx, cmd, part) release() - segments = append(segments, submitSegmentResult(part, seq, gatewayMessageID, result)) + segment := submitSegmentResult(part, seq, gatewayMessageID, result) + segments = append(segments, segment) + if onSegment != nil { + onSegment(segment) + } if firstSequence == 0 { firstSequence = seq } @@ -878,13 +905,35 @@ func (c *connection) readLoop() { ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: p.Result} } case *cmpp.Cmpp2DeliverReqPkt: - responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) - c.emitDeliverResponse(deliverPacketFromCMPP2(p), responseErr) - c.handleDeliver(deliverPacketFromCMPP2(p)) + deliver := deliverPacketFromCMPP2(p) + if deliver.registerDelivery == 1 { + if err := c.handleDeliver(deliver); err != nil { + log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=intake_failed channel_id=%s sequence_id=%d error=%q", c.channelID, deliver.seqID, err) + c.handleConnectionLoss(fmt.Errorf("persist upstream receipt before DELIVER_RESP: %w", err)) + return + } + responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) + c.emitDeliverResponse(deliver, responseErr) + } else { + responseErr := c.sendResponse(client, &cmpp.Cmpp2DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) + c.emitDeliverResponse(deliver, responseErr) + _ = c.handleDeliver(deliver) + } case *cmpp.Cmpp3DeliverReqPkt: - responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) - c.emitDeliverResponse(deliverPacketFromCMPP3(p), responseErr) - c.handleDeliver(deliverPacketFromCMPP3(p)) + deliver := deliverPacketFromCMPP3(p) + if deliver.registerDelivery == 1 { + if err := c.handleDeliver(deliver); err != nil { + log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=intake_failed channel_id=%s sequence_id=%d error=%q", c.channelID, deliver.seqID, err) + c.handleConnectionLoss(fmt.Errorf("persist upstream receipt before DELIVER_RESP: %w", err)) + return + } + responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) + c.emitDeliverResponse(deliver, responseErr) + } else { + responseErr := c.sendResponse(client, &cmpp.Cmpp3DeliverRspPkt{MsgId: p.MsgId, Result: 0}, p.SeqId) + c.emitDeliverResponse(deliver, responseErr) + _ = c.handleDeliver(deliver) + } case *cmpp.CmppActiveTestReqPkt: _ = c.sendResponse(client, &cmpp.CmppActiveTestRspPkt{}, p.SeqId) _ = c.pool.reportState(context.Background(), "heartbeat", nil) @@ -1069,12 +1118,12 @@ func deliverPacketFromCMPP3(pkt *cmpp.Cmpp3DeliverReqPkt) deliverPacket { } } -func (c *connection) handleDeliver(pkt deliverPacket) { +func (c *connection) handleDeliver(pkt deliverPacket) error { if pkt.registerDelivery == 1 { var receipt cmpp.CmppReceiptPkt if err := receipt.Unpack([]byte(pkt.msgContent)); err != nil { log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_receipt status=parse_failed channel_id=%s sequence_id=%d packet_msg_id=%d error=%q", c.channelID, pkt.seqID, pkt.msgID, err) - return + return err } log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_receipt status=received channel_id=%s sequence_id=%d gateway_message_id=%d raw_status=%s", c.channelID, pkt.seqID, receipt.MsgId, strings.TrimSpace(receipt.Stat)) cmd, ok := c.commandFor(receipt.MsgId) @@ -1104,22 +1153,24 @@ func (c *connection) handleDeliver(pkt deliverPacket) { ReceiptStatus: receiptStatus(receipt.Stat), RawStatus: strings.TrimSpace(receipt.Stat), DeliveredAt: time.Now().UTC(), + ConnectionID: c.identity(), } - if err := postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/receipt", event); err != nil { + if err := postJSON(context.Background(), c.httpClient, c.apiBaseURL, "/gateway/events/receipt/intake", event); err != nil { log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=forward_failed channel_id=%s sequence_id=%d gateway_message_id=%d error=%q", c.channelID, pkt.seqID, receipt.MsgId, err) + return err } else { log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_receipt status=forwarded channel_id=%s sequence_id=%d gateway_message_id=%d", c.channelID, pkt.seqID, receipt.MsgId) } - return + return nil } content, complete, err := c.decodeUplinkContent(pkt) if err != nil { log.Printf("protocol_event protocol=cmpp direction=channel_to_platform event=deliver_uplink status=decode_failed channel_id=%s sequence_id=%d packet_msg_id=%d error=%q", c.channelID, pkt.seqID, pkt.msgID, err) - return + return err } if !complete { - return + return nil } cmd, _ := c.commandFor(pkt.msgID) event := queue.UplinkEvent{ @@ -1142,6 +1193,14 @@ func (c *connection) handleDeliver(pkt deliverPacket) { } else { log.Printf("protocol_event protocol=cmpp direction=gateway_to_api event=deliver_uplink status=forwarded channel_id=%s sequence_id=%d packet_msg_id=%d", c.channelID, pkt.seqID, pkt.msgID) } + return nil +} + +func (c *connection) identity() string { + if c.pool != nil && strings.TrimSpace(c.pool.connectionID) != "" { + return fmt.Sprintf("%s-%d", c.pool.connectionID, c.index) + } + return fmt.Sprintf("%s-%d", c.channelID, c.index) } func (c *connection) decodeUplinkContent(pkt deliverPacket) (string, bool, error) { diff --git a/src/api/adminApi.ts b/src/api/adminApi.ts index 5cdf8e2..c497b55 100644 --- a/src/api/adminApi.ts +++ b/src/api/adminApi.ts @@ -1266,6 +1266,22 @@ export type DownstreamDeliveryRecord = { tenant?: TenantOption | null; application?: EnterpriseApplication | null; messageRecord?: SmsMessageRecord | null; + attempts?: Array<{ + id: string; + attemptNo: number; + connectionId?: string | null; + sequenceId?: string | null; + messageId?: string | null; + status: string; + sentAt?: string | null; + ackDeadlineAt?: string | null; + acknowledgedAt?: string | null; + ackResult?: number | null; + failureType?: string | null; + errorMessage?: string | null; + createdAt: string; + updatedAt: string; + }>; }; export type BatchRequeueResponse = { diff --git a/src/apps/admin/AdminDownstreamDeliveriesPage.tsx b/src/apps/admin/AdminDownstreamDeliveriesPage.tsx index b0bec53..693e418 100644 --- a/src/apps/admin/AdminDownstreamDeliveriesPage.tsx +++ b/src/apps/admin/AdminDownstreamDeliveriesPage.tsx @@ -66,6 +66,35 @@ function DeliveryDetailModal({ record, onClose }: { record: DownstreamDeliveryRe
连接 ID{record.connectionId ?? '-'}
最后错误{record.lastError ?? '-'}
+
+

逐次投递记录

+
+
+ 次数 / 状态发送 / ACK 时间连接 / Sequence / Msg_Id结果 +
+ {(record.attempts ?? []).map((attempt) => ( +
+ 第 {attempt.attemptNo} 次
{attempt.status} + + {attempt.sentAt ? formatDateTime(attempt.sentAt) : '-'} +
+ ACK：{attempt.acknowledgedAt ? formatDateTime(attempt.acknowledgedAt) : '-'} + + + {attempt.connectionId ?? '-'} +
+ Seq：{attempt.sequenceId ?? '-'} / Msg：{attempt.messageId ?? '-'} + + + ACK Result：{attempt.ackResult ?? '-'} +
+ {attempt.errorMessage ?? attempt.failureType ?? '-'} + +
+ ))} + {(record.attempts ?? []).length === 0 ?

暂无逐次投递记录

: null} +
+

Payload

{payloadText}
diff --git a/src/styles/global.css b/src/styles/global.css index 7094d9b..b2c0d85 100644 --- a/src/styles/global.css +++ b/src/styles/global.css @@ -8700,6 +8700,34 @@ h3 { color: var(--color-text-strong); } +.downstream-attempt-table { + border-top: 0; + display: block; + overflow-x: auto; + padding: 0; +} + +.downstream-attempt-table > div { + align-items: start; + border-top: 1px solid var(--color-border); + display: grid; + gap: var(--space-4); + grid-template-columns: minmax(110px, .7fr) minmax(190px, 1.1fr) minmax(260px, 1.5fr) minmax(180px, 1fr); + min-width: 820px; + padding: var(--space-3) 0; +} + +.downstream-attempt-table .downstream-attempt-table__header { + color: var(--color-text-muted); + font-size: var(--font-size-sm); + font-weight: 600; +} + +.downstream-attempt-table > p { + color: var(--color-text-muted); + margin: var(--space-3) 0 0; +} + .admin-task-template-row .ui-inline-text-preview { background: var(--color-bg-subtle); }