chore(retry): 新增 SMS 限流重试的库侧脚本(MA:0038 映射 + 普通类阶梯)

resources/sql 下两份手工执行迁移脚本(供 DBA 在目标库运行,均幂等):
- provider-error-code-mapping/telecom_sms_MA0038_retryable.sql:
  给 push_provider_error_code_mapping 配 (telecom,sms,MA:0038 ->
  unified_error_code=RATE_LIMIT, should_retry=1)。与 TelecomSmsSender
  的 status=6 改动(c892f71)缺一不可——MA:0038 才能进重试链并被识别为限流类。
- sms_retry_strategy_ladder.sql:
  普通类 sms 失败重试阶梯 retry_intervals='60,120,300,600'、max_retry_count=6
  (TD-3 决策),取代旧 quickfix 的 60s 平铺;限流类由代码 120s/12h 保证,不读此表。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

msgpush-framework/msgpush-common/pom.xml

@@ -72,7 +72,7 @@
         <!-- Swagger 注解,用于 API 文档生成(@Schema、@Operation 等) -->
         <dependency>
             <groupId>io.swagger.core.v3</groupId>
-            <artifactId>swagger-annotations</artifactId>
+            <artifactId>swagger-annotations-jakarta</artifactId>
         </dependency>
 
         <!-- RPC 远程调用相关 -->
@@ -180,4 +180,4 @@
         </dependency>
     </dependencies>
 
-</project>
+</project>

msgpush-framework/msgpush-common/src/main/java/com/njcn/msgpush/framework/common/exception/enums/ServiceErrorCodeRange.java

@@ -1,5 +1,7 @@
 package com.njcn.msgpush.framework.common.exception.enums;
 
+import com.njcn.msgpush.framework.common.exception.ErrorCode;
+
 /**
  * 业务异常的错误码区间，解决：解决各模块错误码定义，避免重复，在此只声明不做实际使用
  *
@@ -27,7 +29,7 @@ package com.njcn.msgpush.framework.common.exception.enums;
  *
  * @author hongawen
  */
-public class ServiceErrorCodeRange {
+public interface ServiceErrorCodeRange {
 
     // 模块 infra 错误码区间 [1-001-000-000 ~ 1-002-000-000)
     // 模块 system 错误码区间 [1-002-000-000 ~ 1-003-000-000)
@@ -43,6 +45,7 @@ public class ServiceErrorCodeRange {
 
     // 模块 crm 错误码区间 [1-020-000-000 ~ 1-021-000-000)
 
-    // 模块 ai 错误码区间 [1-022-000-000 ~ 1-023-000-000)
-
+    // 模块 push 错误码区间 [1-022-000-000 ~ 1-023-000-000)
+    ErrorCode VALIDATE_SYSTEM_SECRET_FAIL = new ErrorCode(1-022-000-000, "校验系统密钥失败");
+    ErrorCode GENERATE_CREDENTIAL_FAIL = new ErrorCode(1-022-000-001, "生成凭证失败");
 }

msgpush-framework/msgpush-spring-boot-starter-web/src/main/java/com/njcn/msgpush/framework/swagger/config/MsgpushSwaggerAutoConfiguration.java

@@ -93,9 +93,10 @@ public class MsgpushSwaggerAutoConfiguration {
     }
 
     public static GroupedOpenApi buildGroupedOpenApi(String group, String path) {
+        String pathSuffix = path == null || path.isBlank() ? "" : "/" + path;
         return GroupedOpenApi.builder()
                 .group(group)
-                .pathsToMatch("/admin-api/" + path + "/**", "/app-api/" + path + "/**")
+                .pathsToMatch("/admin-api" + pathSuffix + "/**", "/app-api" + pathSuffix + "/**")
                 .addOperationCustomizer((operation, handlerMethod) -> operation
                         .addParametersItem(buildSecurityHeaderParameter()))
                 .addOperationCustomizer(buildOperationIdCustomizer())

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/checker/MsgPushGuardChain.java

@@ -4,6 +4,7 @@ import com.njcn.msgpush.module.push.checker.impl.BlacklistChecker;
 import com.njcn.msgpush.module.push.checker.impl.QuotaChecker;
 import com.njcn.msgpush.module.push.checker.impl.RateLimitChecker;
 import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
+import com.njcn.msgpush.module.push.enums.RetrySourceEnum;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -24,9 +25,15 @@ public class MsgPushGuardChain {
     private RateLimitChecker rateLimitChecker;
 
 
-    public void checkAll(List<MessageRecordDO> messageRecordList) {
+    public void checkAll(List<MessageRecordDO> messageRecordList, RetrySourceEnum retrySource) {
+        // 黑名单:首发与重发(含限流自动重试)都必须执行
         blacklistChecker.check(messageRecordList);
-        quotaChecker.check(messageRecordList);
-        rateLimitChecker.check(messageRecordList);
+        // 配额/接收者频率:系统自动重试(限流类无限重试)完全豁免——否则会被守卫置终态、队列记录被删,
+        // 无限重试静默失效。重发节奏由 120s 固定间隔 + 12h 兜底天然封顶,无需配额墙再约束。
+        // 首发(retrySource==null)与人工重试(MANUAL_RETRY)仍正常受这两道守卫。
+        if (retrySource != RetrySourceEnum.AUTO_RETRY) {
+            quotaChecker.check(messageRecordList);
+            rateLimitChecker.check(messageRecordList);
+        }
     }
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/ClientConfiguration.java

@@ -1,60 +0,0 @@
-package com.njcn.msgpush.module.push.client;
-
-import com.njcn.msgpush.module.push.client.factory.MessageProviderFactory;
-import com.njcn.msgpush.module.push.client.factory.impl.AliyunProviderFactory;
-import com.njcn.msgpush.module.push.client.factory.impl.TelecomProviderFactory;
-import com.njcn.msgpush.module.push.client.factory.impl.UniPushProviderFactory;
-import com.njcn.msgpush.module.push.dal.dataobject.channel.ChannelProviderConfigDO;
-import com.njcn.msgpush.module.push.enums.ProviderTypeEnum;
-import com.njcn.msgpush.module.push.service.channel.ChannelProviderConfigService;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * @author caozehui
- * @data 2026-02-10
- */
-@Configuration
-public class ClientConfiguration {
-
-    @Autowired
-    private ChannelProviderConfigService channelProviderConfigService;
-
-    @Bean
-    public Map<String, MessageProviderFactory> messageProviderFactoryMap() throws Exception {
-        List<ChannelProviderConfigDO> activeProviders = channelProviderConfigService.getActiveProviders();
-        Map<String, MessageProviderFactory> messageProviderFactoryMap = new HashMap<>();
-
-        for (ChannelProviderConfigDO config : activeProviders) {
-            switch (ProviderTypeEnum.getByCode(config.getProviderType())) {
-                case ALIYUN: {
-                    MessageProviderFactory orDefault = messageProviderFactoryMap.getOrDefault(ProviderTypeEnum.ALIYUN.getCode(), new AliyunProviderFactory());
-                    messageProviderFactoryMap.put(ProviderTypeEnum.ALIYUN.getCode(), orDefault);
-                }
-                break;
-                case TELECOM: {
-                    MessageProviderFactory orDefault = messageProviderFactoryMap.getOrDefault(ProviderTypeEnum.TELECOM.getCode(), new TelecomProviderFactory());
-                    messageProviderFactoryMap.put(ProviderTypeEnum.TELECOM.getCode(), orDefault);
-                }
-                break;
-                case UNIPUSH: {
-                    MessageProviderFactory orDefault = messageProviderFactoryMap.getOrDefault(ProviderTypeEnum.UNIPUSH.getCode(), new UniPushProviderFactory());
-                    messageProviderFactoryMap.put(ProviderTypeEnum.UNIPUSH.getCode(), orDefault);
-                }
-                break;
-                default:
-                    throw new IllegalArgumentException("该" + config.getProviderType() + "服务商暂不支持");
-            }
-        }
-
-        if (messageProviderFactoryMap.isEmpty()) {
-            throw new Exception("当前没有激活的渠道服务商!");
-        }
-        return messageProviderFactoryMap;
-    }
-}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/factory/MessageProviderFactoryRegistry.java

@@ -0,0 +1,29 @@
+package com.njcn.msgpush.module.push.client.factory;
+
+import com.njcn.msgpush.module.push.client.factory.impl.AliyunProviderFactory;
+import com.njcn.msgpush.module.push.client.factory.impl.TelecomProviderFactory;
+import com.njcn.msgpush.module.push.client.factory.impl.UniPushProviderFactory;
+import com.njcn.msgpush.module.push.enums.ProviderTypeEnum;
+import org.springframework.stereotype.Component;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@Component
+public class MessageProviderFactoryRegistry {
+
+    private final Map<String, MessageProviderFactory> factoryMap;
+
+    public MessageProviderFactoryRegistry() {
+        Map<String, MessageProviderFactory> factories = new HashMap<>();
+        factories.put(ProviderTypeEnum.ALIYUN.getCode(), new AliyunProviderFactory());
+        factories.put(ProviderTypeEnum.TELECOM.getCode(), new TelecomProviderFactory());
+        factories.put(ProviderTypeEnum.UNIPUSH.getCode(), new UniPushProviderFactory());
+        this.factoryMap = Collections.unmodifiableMap(factories);
+    }
+
+    public MessageProviderFactory getFactory(String providerType) {
+        return factoryMap.get(providerType);
+    }
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/factory/impl/AliyunProviderFactory.java

@@ -20,12 +20,12 @@ public class AliyunProviderFactory implements MessageProviderFactory {
     @Override
     public SmsSender createSmsSender(ChannelProviderConfigDO config, Sender sender) {
         // 配置缺失时返回固定结果 sender，由主流程统一落库为配置异常。
-        if (ProviderFactorySupport.hasBlank(config.getAppKey(), config.getAppSecret())) {
+        if (ProviderFactorySupport.hasBlank(config.getAppKey(), config.getSecret())) {
             return new FixedResultSmsSender(SendOutcome.CONFIG_INVALID, "当前服务商短信渠道配置不完整");
         }
         AliYunMailSetting aliYunSmsSetting = AliYunMailSetting.builder()
                 .accessKeyId(config.getAppKey())
-                .accessKeySecret(config.getAppSecret())
+                .accessKeySecret(config.getSecret())
                 .regionId("cn-hangzhou")
                 .endpoint("dysmsapi.aliyuncs.com")
                 .build();
@@ -35,12 +35,12 @@ public class AliyunProviderFactory implements MessageProviderFactory {
     @Override
     public EmailSender createEmailSender(ChannelProviderConfigDO config, Sender sender) {
         // 配置缺失时返回固定结果 sender，由主流程统一落库为配置异常。
-        if (ProviderFactorySupport.hasBlank(config.getAppKey(), config.getAppSecret())) {
+        if (ProviderFactorySupport.hasBlank(config.getAppKey(), config.getSecret())) {
             return new FixedResultEmailSender(SendOutcome.CONFIG_INVALID, "当前服务商邮件渠道配置不完整");
         }
         AliYunMailSetting aliYunMailSetting = AliYunMailSetting.builder()
                 .accessKeyId(config.getAppKey())
-                .accessKeySecret(config.getAppSecret())
+                .accessKeySecret(config.getSecret())
                 .regionId("cn-hangzhou")
                 .endpoint("dm.aliyuncs.com")
                 .build();

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/factory/impl/TelecomProviderFactory.java

@@ -19,12 +19,12 @@ public class TelecomProviderFactory implements MessageProviderFactory {
     @Override
     public SmsSender createSmsSender(ChannelProviderConfigDO config, Sender sender) {
         // 配置缺失时返回固定结果 sender，由主流程统一落库为配置异常。
-        if (ProviderFactorySupport.hasBlank(config.getAppKey(), config.getAppSecret(), config.getApiUrl())) {
+        if (ProviderFactorySupport.hasBlank(config.getAppKey(), config.getSecret(), config.getApiUrl())) {
             return new FixedResultSmsSender(SendOutcome.CONFIG_INVALID, "当前服务商短信渠道配置不完整");
         }
         TelecomSmsSetting telecomSmsSetting = TelecomSmsSetting.builder()
                 .account(config.getAppKey())
-                .password(config.getAppSecret())
+                .password(config.getSecret())
                 .apiUrl(config.getApiUrl())
                 .extno(config.getExtno())
                 .build();

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/factory/impl/UniPushProviderFactory.java

@@ -1,15 +1,8 @@
 package com.njcn.msgpush.module.push.client.factory.impl;
 
-import cn.hutool.core.util.StrUtil;
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
 import com.njcn.msgpush.module.push.client.factory.MessageProviderFactory;
 import com.njcn.msgpush.module.push.client.factory.ProviderFactorySupport;
-import com.njcn.msgpush.module.push.client.sender.AppPushSender;
-import com.njcn.msgpush.module.push.client.sender.EmailSender;
-import com.njcn.msgpush.module.push.client.sender.SendOutcome;
-import com.njcn.msgpush.module.push.client.sender.Sender;
-import com.njcn.msgpush.module.push.client.sender.SmsSender;
+import com.njcn.msgpush.module.push.client.sender.*;
 import com.njcn.msgpush.module.push.client.sender.fallback.FixedResultAppPushSender;
 import com.njcn.msgpush.module.push.client.sender.fallback.FixedResultEmailSender;
 import com.njcn.msgpush.module.push.client.sender.fallback.FixedResultSmsSender;
@@ -19,8 +12,6 @@ import com.njcn.msgpush.module.push.dal.dataobject.channel.ChannelProviderConfig
 
 public class UniPushProviderFactory implements MessageProviderFactory {
 
-    private static final String APP_ID = "appId";
-    private static final String MASTER_SECRET = "masterSecret";
 
     @Override
     public SmsSender createSmsSender(ChannelProviderConfigDO config, Sender sender) {
@@ -36,18 +27,15 @@ public class UniPushProviderFactory implements MessageProviderFactory {
 
     @Override
     public AppPushSender createAppPushSender(ChannelProviderConfigDO config, Sender sender) {
-        JSONObject extraConfig = StrUtil.isBlank(config.getExtraConfig()) ? new JSONObject() : JSON.parseObject(config.getExtraConfig());
-        String appId = extraConfig.getString(APP_ID);
-        String masterSecret = extraConfig.getString(MASTER_SECRET);
         // 配置缺失时返回固定结果 sender，由主流程统一落库为配置异常。
-        if (ProviderFactorySupport.hasBlank(appId, config.getAppKey(), config.getAppSecret(), masterSecret)) {
+        if (ProviderFactorySupport.hasBlank(config.getSecret())) {
             return new FixedResultAppPushSender(SendOutcome.CONFIG_INVALID, "当前服务商 APP 推送渠道配置不完整");
         }
         UniPushAppPushSetting uniPushAppPushSetting = new UniPushAppPushSetting(
-                appId,
+                config.getAppId(),
                 config.getAppKey(),
-                config.getAppSecret(),
-                masterSecret
+                config.getSecret(),
+                config.getApiUrl()
         );
         return new UniPushAppPushSender(uniPushAppPushSetting, sender);
     }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/sender/impl/apppush/UniPushAppPushSender.java

@@ -1,14 +1,22 @@
 package com.njcn.msgpush.module.push.client.sender.impl.apppush;
 
+import cn.hutool.core.util.StrUtil;
 import com.getui.push.v2.sdk.ApiHelper;
 import com.getui.push.v2.sdk.GtApiConfiguration;
 import com.getui.push.v2.sdk.api.PushApi;
 import com.getui.push.v2.sdk.common.ApiResult;
 import com.getui.push.v2.sdk.dto.req.Audience;
 import com.getui.push.v2.sdk.dto.req.Settings;
+import com.getui.push.v2.sdk.dto.req.message.PushChannel;
 import com.getui.push.v2.sdk.dto.req.message.PushDTO;
 import com.getui.push.v2.sdk.dto.req.message.PushMessage;
+import com.getui.push.v2.sdk.dto.req.message.android.AndroidDTO;
 import com.getui.push.v2.sdk.dto.req.message.android.GTNotification;
+import com.getui.push.v2.sdk.dto.req.message.android.ThirdNotification;
+import com.getui.push.v2.sdk.dto.req.message.android.Ups;
+import com.getui.push.v2.sdk.dto.req.message.ios.Alert;
+import com.getui.push.v2.sdk.dto.req.message.ios.Aps;
+import com.getui.push.v2.sdk.dto.req.message.ios.IosDTO;
 import com.njcn.msgpush.module.push.client.sender.AppPushSender;
 import com.njcn.msgpush.module.push.client.sender.SendResult;
 import com.njcn.msgpush.module.push.client.sender.Sender;
@@ -34,7 +42,13 @@ public class UniPushAppPushSender implements AppPushSender {
             gtApiConfiguration.setAppId(uniPushAppPushSetting.getAppId());
             gtApiConfiguration.setAppKey(uniPushAppPushSetting.getAppKey());
             gtApiConfiguration.setMasterSecret(uniPushAppPushSetting.getMasterSecret());
-            gtApiConfiguration.setDomain("https://restapi.getui.com/v2/");
+            // 使用配置的 API 地址，如果未配置则使用默认地址
+            String apiUrl = uniPushAppPushSetting.getApiUrl();
+            if (StrUtil.isNotBlank(apiUrl)) {
+                gtApiConfiguration.setDomain(apiUrl);
+            } else {
+                gtApiConfiguration.setDomain("https://restapi.getui.com/v2/");
+            }
             ApiHelper apiHelper = ApiHelper.build(gtApiConfiguration);
             this.pushApi = apiHelper.creatApi(PushApi.class);
         } catch (Exception e) {
@@ -89,13 +103,48 @@ public class UniPushAppPushSender implements AppPushSender {
         settings.setTtl(3600000);
         pushDTO.setSettings(settings);
 
+        // 创建 PushMessage，设置透传消息（可选）
         PushMessage pushMessage = new PushMessage();
-        GTNotification notification = new GTNotification();
-        notification.setTitle(title);
-        notification.setBody(content);
-        notification.setClickType("startapp");
-        pushMessage.setNotification(notification);
+        pushMessage.setTransmission("{\"title\":\"" + title + "\",\"body\":\"" + content + "\"}");
+
+        // 在线推送 - 使用 GTNotification
+        GTNotification gtNotification = new GTNotification();
+        gtNotification.setTitle(title);
+        gtNotification.setBody(content);
+        gtNotification.setClickType("startapp");
+        pushMessage.setNotification(gtNotification);
+
         pushDTO.setPushMessage(pushMessage);
+
+        // 创建 PushChannel 用于配置各平台的通知
+        PushChannel pushChannel = new PushChannel();
+
+        // Android 离线推送配置 - 使用 ThirdNotification
+        AndroidDTO androidDTO = new AndroidDTO();
+        Ups ups = new Ups();
+        ThirdNotification thirdNotification = new ThirdNotification();
+        thirdNotification.setTitle(title);
+        thirdNotification.setBody(content);
+        thirdNotification.setClickType("startapp");
+        ups.setNotification(thirdNotification);
+        androidDTO.setUps(ups);
+        pushChannel.setAndroid(androidDTO);
+
+        // iOS 通知配置
+        IosDTO iosDTO = new IosDTO();
+        Aps aps = new Aps();
+        Alert alert = new Alert();
+        alert.setTitle(title);
+        alert.setBody(content);
+        aps.setAlert(alert);
+        aps.setSound("default");
+        aps.setContentAvailable(0);
+        iosDTO.setAps(aps);
+        pushChannel.setIos(iosDTO);
+
+        // 将 PushChannel 设置到 PushDTO 中
+        pushDTO.setPushChannel(pushChannel);
+
         return pushDTO;
     }
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/sender/impl/email/AliyunEmailSender.java

@@ -1,11 +1,8 @@
 package com.njcn.msgpush.module.push.client.sender.impl.email;
 
 import cn.hutool.core.util.ObjectUtil;
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
 import com.aliyun.dm20151123.Client;
-import com.aliyun.dm20151123.models.SingleSendMailRequest;
-import com.aliyun.dm20151123.models.SingleSendMailResponse;
+import com.aliyun.dm20151123.models.*;
 import com.aliyun.teaopenapi.models.Config;
 import com.aliyun.teautil.models.RuntimeOptions;
 import com.njcn.msgpush.module.push.client.sender.EmailSender;
@@ -18,6 +15,8 @@ import org.springframework.http.HttpStatus;
 
 import java.time.LocalDateTime;
 import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -25,9 +24,10 @@ import java.util.concurrent.TimeUnit;
 @Slf4j
 public class AliyunEmailSender implements EmailSender {
 
-    private static final String ACCOUNT_NAME = "accountName";
-    private static final String REPLY_TO_ADDRESS = "replyToAddress";
-    private static final String FROM_ALIAS = "fromAlias";
+    private static final String ACCOUNT_NAME = "njcn@shining-electric.cn";
+    private static final Integer ADDRESS_TYPE = 1;
+    private static final Boolean REPLY_TO_ADDRESS = false;
+    private static final String FROM_ALIAS = "南京灿能";
 
     private final Sender sender;
 
@@ -60,16 +60,15 @@ public class AliyunEmailSender implements EmailSender {
             RuntimeOptions runtimeOptions = new RuntimeOptions();
             runtimeOptions.autoretry = true;
 
-            JSONObject jsonObject = JSON.parseObject(message.getExtraInfo());
             SingleSendMailRequest request = new SingleSendMailRequest()
-                    .setAccountName(jsonObject.getString(ACCOUNT_NAME))
-                    .setAddressType(1)
-                    .setReplyToAddress(jsonObject.getBooleanValue(REPLY_TO_ADDRESS))
+                    .setAccountName(ACCOUNT_NAME)
+                    .setAddressType(ADDRESS_TYPE)
+                    .setReplyToAddress(REPLY_TO_ADDRESS)
                     .setToAddress(message.getReceiver())
                     .setSubject(message.getTitle())
                     .setHtmlBody(message.getContent())
                     .setTextBody("")
-                    .setFromAlias(jsonObject.getString(FROM_ALIAS));
+                    .setFromAlias(FROM_ALIAS);
 
             try {
                 LocalDateTime now = LocalDateTime.now();
@@ -81,8 +80,11 @@ public class AliyunEmailSender implements EmailSender {
                 message.setCostTime(costTime);
 
                 if (HttpStatus.OK.value() == response.getStatusCode()) {
+                    message.setThirdPartyId(response.getBody().getEnvId());
+                    // 电信短信同步返回成功同样只代表已受理，最终状态以后续回执为准。
+                    this.getDownInfo(message);
                     // 邮件接口同步返回成功时，当前平台直接认定本次发送成功。
-                    return SendResult.success(now, costTime, null);
+                    return SendResult.accepted(now, costTime, response.getBody().getEnvId());
                 }
 
                 // 邮件服务失败同样统一转换为平台错误码和中文错误信息。
@@ -107,4 +109,50 @@ public class AliyunEmailSender implements EmailSender {
             return this.sender.buildTimeoutResult();
         }
     }
+
+    private void getDownInfo(MessageRecordDO message) {
+        // 回执查询延后执行，给第三方落库和状态变更留出时间。
+        this.sender.MSG_CALLBACK_THREAD_POOL_SCHEDULER.schedule(() -> {
+            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+            SenderStatisticsDetailByParamRequest request = new SenderStatisticsDetailByParamRequest()
+                    .setToAddress(message.getReceiver())
+                    .setLength(1);
+            try {
+                SenderStatisticsDetailByParamResponse detail = this.emailClient.senderStatisticsDetailByParam(request);
+                if (detail.statusCode == 200) {
+                    List<SenderStatisticsDetailByParamResponseBody.SenderStatisticsDetailByParamResponseBodyDataMailDetail> mailDetailList = detail.body.getData().mailDetail;
+
+                    if (mailDetailList.size() > 0) {
+                        SenderStatisticsDetailByParamResponseBody.SenderStatisticsDetailByParamResponseBodyDataMailDetail mailDetail = mailDetailList.get(0);
+                        if (mailDetail.getStatus() == 0) {
+                            // 回执确认成功后，复用统一成功落库逻辑。
+                            this.sender.applyCallbackResult(message, SendResult.success(message.getSendTime(), message.getCostTime(), message.getThirdPartyId()));
+                        } else {
+                            SendResult failedResult = this.sender.buildFailureResult(
+                                    message,
+                                    mailDetail.getErrorClassification(),
+                                    mailDetail.getErrorClassification(),
+                                    "THIRD_PARTY_CALLBACK_FAILED",
+                                    "邮件回执返回失败",
+                                    false
+                            );
+                            this.sender.applyCallbackResult(message, failedResult);
+                        }
+                    } else {
+                        SendResult failedResult = this.sender.buildFailureResult(
+                                message,
+                                null,
+                                null,
+                                "THIRD_PARTY_CALLBACK_FAILED",
+                                "邮件回执返回失败",
+                                false
+                        );
+                        this.sender.applyCallbackResult(message, failedResult);
+                    }
+                }
+            } catch (Exception e) {
+                log.error("电信短信回执查询失败", e);
+            }
+        }, 30, TimeUnit.SECONDS);
+    }
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/sender/impl/sms/TelecomSmsSender.java

@@ -200,33 +200,19 @@ public class TelecomSmsSender implements SmsSender {
                     return;
                 }
                 TelecomSmsSelectDetailRes detailRes = selectResponse.getList().get(0);
-//                if (detailRes.getStatus() == 5) {
-//                    SendResult failedResult = this.sender.buildFailureResult(
-//                            message,
-//                            detailRes.getStat(),
-//                            null,
-//                            "THIRD_PARTY_CALLBACK_FAILED",
-//                            "短信回执返回失败",
-//                            false
-//                    );
-//                    this.sender.applyCallbackResult(message, failedResult);
-//                } else if (detailRes.getStatus() == 4) {
-//                    // 回执确认成功后，复用统一成功落库逻辑。
-//                    this.sender.applyCallbackResult(message, SendResult.success(message.getSendTime(), message.getCostTime(), message.getThirdPartyId()));
-//                }
-                double random = Math.random();
-                System.out.println(random + "  aaaa");
-                if (random > 0.5) {
+                // status=5 为明确失败；status=6 为“短期频繁发送”等场景（stat 如 MA:0038），
+                // 同样走错误码映射，由 push_provider_error_code_mapping.should_retry 决定是否重试。
+                if (detailRes.getStatus() == 5 || detailRes.getStatus() == 6) {
                     SendResult failedResult = this.sender.buildFailureResult(
                             message,
                             detailRes.getStat(),
                             null,
                             "THIRD_PARTY_CALLBACK_FAILED",
                             "短信回执返回失败",
-                            true
+                            false
                     );
                     this.sender.applyCallbackResult(message, failedResult);
-                } else {
+                } else if (detailRes.getStatus() == 4) {
                     // 回执确认成功后，复用统一成功落库逻辑。
                     this.sender.applyCallbackResult(message, SendResult.success(message.getSendTime(), message.getCostTime(), message.getThirdPartyId()));
                 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/client/setting/impl/UniPushAppPushSetting.java

@@ -12,15 +12,17 @@ import lombok.EqualsAndHashCode;
 @Data
 @EqualsAndHashCode(callSuper = true)
 public class UniPushAppPushSetting extends AppPushSetting {
+    // 个推应用ID
     private String appId;
     private String appKey;
-    private String uniAppSecret;
+    // 个推主密钥
     private String masterSecret;
+    private String apiUrl;
 
-    public UniPushAppPushSetting(String appId, String appKey, String uniAppSecret, String masterSecret) {
+    public UniPushAppPushSetting(String appId, String appKey, String masterSecret, String apiUrl) {
         this.appId = appId;
         this.appKey = appKey;
-        this.uniAppSecret = uniAppSecret;
         this.masterSecret = masterSecret;
+        this.apiUrl = apiUrl;
     }
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/blacklist/vo/BlacklistReqVO.java

@@ -19,7 +19,7 @@ public class BlacklistReqVO extends PageParam {
     @Schema(description = "主键 ID", example = "123444")
     private Long id;
 
-    @Schema(description = "渠道类型：sms/email/app_push", example = "sms")
+    @Schema(description = "渠道类型：sms/email/app", example = "sms")
     @InEnum(value = com.njcn.msgpush.module.push.enums.ChannelTypeEnum.class, message = "渠道类型必须是 {value}")
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/channel/ChannelProviderConfigController.java

@@ -3,30 +3,27 @@ package com.njcn.msgpush.module.push.controller.admin.channel;
 import cn.hutool.core.bean.BeanUtil;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.njcn.msgpush.framework.common.pojo.CommonResult;
-import com.njcn.msgpush.module.push.client.factory.MessageProviderFactory;
-import com.njcn.msgpush.module.push.client.factory.impl.AliyunProviderFactory;
-import com.njcn.msgpush.module.push.client.factory.impl.TelecomProviderFactory;
-import com.njcn.msgpush.module.push.client.factory.impl.UniPushProviderFactory;
 import com.njcn.msgpush.module.push.controller.admin.channel.vo.ChannelProviderConfigReqVO;
 import com.njcn.msgpush.module.push.dal.dataobject.channel.ChannelProviderConfigDO;
-import com.njcn.msgpush.module.push.enums.ProviderTypeEnum;
 import com.njcn.msgpush.module.push.service.channel.ChannelProviderConfigService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.validation.annotation.Validated;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
 
 import java.util.List;
-import java.util.Map;
 
 import static com.njcn.msgpush.framework.common.pojo.CommonResult.success;
 
-
 @Tag(name = "管理后台 - 渠道服务商")
 @Slf4j
 @Validated
@@ -37,10 +34,6 @@ public class ChannelProviderConfigController {
     @Autowired
     private ChannelProviderConfigService channelProviderConfigService;
 
-    @Autowired
-    @Qualifier("messageProviderFactoryMap")
-    private Map<String, MessageProviderFactory> messageProviderFactoryMap;
-
     @PostMapping("/page")
     @Operation(summary = "分页查询渠道服务商列表")
     @PreAuthorize("@ss.hasPermission('push:channel:page')")
@@ -67,9 +60,9 @@ public class ChannelProviderConfigController {
     @PostMapping("/delete")
     @Operation(summary = "删除渠道服务商")
     @PreAuthorize("@ss.hasPermission('push:channel:delete')")
-    @Parameter(name = "ids", description = "id列表", required = true)
+    @Parameter(name = "ids", description = "id 列表", required = true)
     public CommonResult<Boolean> delete(@RequestParam("ids") List<String> ids) {
-        boolean res = channelProviderConfigService.removeBatchByIds(ids);
+        boolean res = channelProviderConfigService.deleteByIds(ids);
         return success(res);
     }
 
@@ -78,46 +71,7 @@ public class ChannelProviderConfigController {
     @PreAuthorize("@ss.hasPermission('push:channel:toggle')")
     @Parameter(name = "id", description = "id", required = true)
     public CommonResult<Void> toggleEnableChannelProvider(@RequestParam("id") String id) {
-        ChannelProviderConfigDO channelProviderConfigDO = channelProviderConfigService.toggleEnableField(id);
-
-        if (channelProviderConfigDO.getEnabled() == 1) {
-            registerProviderBean(channelProviderConfigDO.getProviderType());
-        } else {
-            removeProviderBean(channelProviderConfigDO.getProviderType());
-        }
+        channelProviderConfigService.toggleEnableField(id);
         return success(null);
     }
-
-    /**
-     * 添加指定providerType服务商对应的bean
-     *
-     * @param providerType 服务商类型，例如：aliyun\telecom\UniPush
-     */
-    public void registerProviderBean(String providerType) {
-        switch (ProviderTypeEnum.getByCode(providerType)) {
-            case ALIYUN: {
-                messageProviderFactoryMap.put(ProviderTypeEnum.ALIYUN.getCode(), new AliyunProviderFactory());
-            }
-            break;
-            case TELECOM: {
-                messageProviderFactoryMap.put(ProviderTypeEnum.TELECOM.getCode(), new TelecomProviderFactory());
-            }
-            break;
-            case UNIPUSH: {
-                messageProviderFactoryMap.put(ProviderTypeEnum.UNIPUSH.getCode(), new UniPushProviderFactory());
-            }
-            break;
-            default:
-                throw new IllegalArgumentException("该" + providerType + "服务商暂不支持");
-        }
-    }
-
-    /**
-     * 移除指定providerType服务商对应的bean
-     *
-     * @param providerType 服务商类型，例如：aliyun\telecom\UniPush
-     */
-    public void removeProviderBean(String providerType) {
-        messageProviderFactoryMap.remove(providerType);
-    }
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/channel/vo/ChannelProviderConfigReqVO.java

@@ -13,7 +13,7 @@ public class ChannelProviderConfigReqVO extends PageParam {
     @Schema(description = "渠道ID")
     private String id;
 
-    @Schema(description = "渠道类型：sms/email/app_push", example = "sms")
+    @Schema(description = "渠道类型：sms/email/app", example = "sms")
     @InEnum(value = com.njcn.msgpush.module.push.enums.ChannelTypeEnum.class, message = "渠道类型必须是 {value}")
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/credential/CredentialController.java

@@ -0,0 +1,37 @@
+package com.njcn.msgpush.module.push.controller.admin.credential;
+
+import com.njcn.msgpush.framework.common.pojo.CommonResult;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.dto.CredentialReqDTO;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.dto.CredentialRespDTO;
+import com.njcn.msgpush.module.push.service.credential.ICredentialService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.annotation.security.PermitAll;
+import jakarta.validation.Valid;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * 系统凭证 API 接口
+ *
+ * @author msgpush
+ */
+@Tag(name = "系统凭证管理")
+@RestController
+@RequestMapping("/push/credential")
+public class CredentialController {
+
+    @Autowired
+    private ICredentialService credentialService;
+
+    @PermitAll
+    @PostMapping("/generate")
+    @Operation(summary = "生成系统凭证")
+    public CommonResult<CredentialRespDTO> generateCredential(@Valid @RequestBody CredentialReqDTO reqDTO) {
+        CredentialRespDTO respDTO = credentialService.generateCredential(reqDTO);
+        return CommonResult.success(respDTO);
+    }
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/message/MessageRecordController.java

@@ -2,11 +2,15 @@ package com.njcn.msgpush.module.push.controller.admin.message;
 
 import cn.hutool.core.bean.BeanUtil;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.njcn.msgpush.framework.common.exception.enums.GlobalErrorCodeConstants;
 import com.njcn.msgpush.framework.common.pojo.CommonResult;
 import com.njcn.msgpush.framework.idempotent.core.annotation.Idempotent;
 import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageRecordReqVO;
 import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageSendResultVO;
 import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
+import com.njcn.msgpush.module.push.enums.ChannelTypeEnum;
+import com.njcn.msgpush.module.push.service.credential.CredentialServiceImpl;
+import com.njcn.msgpush.module.push.service.credential.ICredentialService;
 import com.njcn.msgpush.module.push.service.message.MessageRecordService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -16,7 +20,12 @@ import jakarta.validation.Valid;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.validation.annotation.Validated;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
 
 import java.util.List;
 
@@ -30,24 +39,55 @@ public class MessageRecordController {
 
     @Autowired
     private MessageRecordService messageRecordService;
+    @Autowired
+    private ICredentialService credentialService;
 
     @PermitAll
-    @PostMapping("/send")
-    @Operation(summary = "消息推送")
+    @PostMapping("/send/sms")
+    @Operation(summary = "短信推送")
     @Idempotent(timeout = 2)
-    public CommonResult<List<MessageSendResultVO>> send(@Valid @RequestBody List<MessageRecordReqVO> reqVOList) {
-        return success(messageRecordService.send(reqVOList));
+    public CommonResult<List<MessageSendResultVO>> sendSms(@Valid @RequestBody List<MessageRecordReqVO> reqVOList,
+                                                           @RequestHeader(value = "X-Credential-Token", required = true) String credentialToken) {
+        return doSend(reqVOList, credentialToken, ChannelTypeEnum.SMS);
+    }
+
+    @PermitAll
+    @PostMapping("/send/email")
+    @Operation(summary = "邮件推送")
+    @Idempotent(timeout = 2)
+    public CommonResult<List<MessageSendResultVO>> sendEmail(@Valid @RequestBody List<MessageRecordReqVO> reqVOList,
+                                                             @RequestHeader(value = "X-Credential-Token", required = true) String credentialToken) {
+        return doSend(reqVOList, credentialToken, ChannelTypeEnum.EMAIL);
+    }
+
+    @PermitAll
+    @PostMapping("/send/app")
+    @Operation(summary = "APP 推送")
+    @Idempotent(timeout = 2)
+    public CommonResult<List<MessageSendResultVO>> sendApp(@Valid @RequestBody List<MessageRecordReqVO> reqVOList,
+                                                           @RequestHeader(value = "X-Credential-Token", required = true) String credentialToken) {
+        return doSend(reqVOList, credentialToken, ChannelTypeEnum.APP);
+    }
+
+    private CommonResult<List<MessageSendResultVO>> doSend(List<MessageRecordReqVO> reqVOList,
+                                                           String credentialToken,
+                                                           ChannelTypeEnum channelTypeEnum) {
+        CredentialServiceImpl.CredentialInfo credentialInfo = credentialService.verifyCredential(credentialToken);
+        if (credentialInfo == null) {
+            return CommonResult.error(GlobalErrorCodeConstants.UNAUTHORIZED.getCode(), "凭证无效或已过期");
+        }
+        return success(messageRecordService.send(reqVOList, channelTypeEnum, credentialInfo.getSystemName()));
     }
 
     @PostMapping("/page")
-    @Operation(summary = "分页查询渠道服务商列表")
+    @Operation(summary = "分页查询消息记录")
     @PreAuthorize("@ss.hasPermission('push:message:page')")
     public CommonResult<Page<MessageRecordDO>> pageChannelProviderConfig(@Validated @RequestBody MessageRecordReqVO reqVO) {
         return success(messageRecordService.getPage(reqVO));
     }
 
     @PostMapping("/add")
-    @Operation(summary = "添加消息记录")
+    @Operation(summary = "新增消息记录")
     @PreAuthorize("@ss.hasPermission('push:message:add')")
     public CommonResult<Boolean> add(@Validated @RequestBody MessageRecordReqVO reqVO) {
         return success(messageRecordService.add(reqVO));

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/message/vo/MessageRecordReqVO.java

@@ -16,15 +16,7 @@ import lombok.Data;
 @Data
 @Schema(description = "管理后台 - 消息记录发送 Request VO")
 public class MessageRecordReqVO extends PageParam {
-    @Schema(description = "主键ID")
-    private Long id;
 
-    @Schema(description = "应用名称/来源系统标识", requiredMode = Schema.RequiredMode.REQUIRED, example = "NPQS-9500")
-    @NotBlank(message = "应用名称/来源系统标识不能为空")
-    private String appName;
-
-    @Schema(description = "渠道类型：sms/email/app_push", example = "sms")
-    @InEnum(value = com.njcn.msgpush.module.push.enums.ChannelTypeEnum.class, message = "渠道类型必须是 {value}")
     private String channel;
 
     @Schema(description = "消息类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "verify_code/order_notify/marketing/system_notify")
@@ -33,7 +25,7 @@ public class MessageRecordReqVO extends PageParam {
 
     @Schema(description = "接收者", requiredMode = Schema.RequiredMode.REQUIRED)
     @NotBlank(message = "接收者不能为空")
-    @Pattern(regexp = RegexPool.EMAIL + "|" + RegexPool.MOBILE, message = "必须是有效的邮箱或手机号格式")
+//    @Pattern(regexp = RegexPool.EMAIL + "|" + RegexPool.MOBILE, message = "必须是有效的邮箱或手机号格式")
     private String receiver;
 
     @Schema(description = "标题", requiredMode = Schema.RequiredMode.REQUIRED)
@@ -47,14 +39,4 @@ public class MessageRecordReqVO extends PageParam {
 
     @Schema(description = "模板参数")
     private String templateParams;
-
-    @Schema(description = "服务商类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "telecom/cmcc/aliyun/twilio")
-    @InEnum(value = com.njcn.msgpush.module.push.enums.ProviderTypeEnum.class, message = "服务商类型必须是 {value}")
-    private String providerType;
-
-    @Schema(description = "第三方消息ID")
-    private String thirdPartyId;
-
-    @Schema(description = "额外信息")
-    private String extraInfo;
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/ratelimit/VO/RateLimitConfigReqVO.java

@@ -18,7 +18,7 @@ public class RateLimitConfigReqVO extends PageParam {
     @Schema(description = "主键 ID")
     private Long id;
 
-    @Schema(description = "渠道类型：sms/email/app_push", example = "sms")
+    @Schema(description = "渠道类型：sms/email/app", example = "sms")
     @InEnum(value = com.njcn.msgpush.module.push.enums.ChannelTypeEnum.class, message = "渠道类型必须是 {value}")
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/retry/vo/MessageRetryQueueReqVO.java

@@ -16,7 +16,7 @@ public class MessageRetryQueueReqVO extends PageParam {
     @Schema(description = "消息ID")
     private Long messageId;
 
-    @Schema(description = "渠道类型", example = "sms/email/app_push")
+    @Schema(description = "渠道类型", example = "sms/email/app")
     @NotBlank(message = "渠道类型不能为空")
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/controller/admin/retry/vo/RetryStrategyConfigReqVO.java

@@ -17,7 +17,7 @@ public class RetryStrategyConfigReqVO {
     @Schema(description = "主键 ID")
     private Long id;
 
-    @Schema(description = "渠道类型：sms/email/app_push", example = "sms")
+    @Schema(description = "渠道类型：sms/email/app", example = "sms")
     @InEnum(value = com.njcn.msgpush.module.push.enums.ChannelTypeEnum.class, message = "渠道类型必须是 {value}")
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/blacklist/BlacklistDO.java

@@ -19,7 +19,7 @@ public class BlacklistDO extends BaseDO {
     private Long id;
 
     /**
-     * 渠道类型：sms/email/app_push
+     * 渠道类型：sms/email/app
      */
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/channel/ChannelProviderConfigDO.java

@@ -22,7 +22,7 @@ public class ChannelProviderConfigDO extends BaseDO {
     private Long id;
 
     /**
-     * 渠道类型：sms/email/app_push
+     * 渠道类型：sms/email/app
      */
     private String channel;
 
@@ -32,30 +32,30 @@ public class ChannelProviderConfigDO extends BaseDO {
     private String providerName;
 
     /**
-     * 服务商类型：telecom/cmcc/aliyun/twilio/unipush
+     * 服务商类型：telecom/aliyun/unipush
      */
     private String providerType;
 
     /**
      * API地址
+     * 对于 UniPush，可配置为自定义API地址，如未配置则使用默认地址 https://restapi.getui.com/v2/
      */
     private String apiUrl;
 
-    /**
-     * AppKey
-     */
     private String appKey;
 
-    /**
-     * AppSecret
-     */
-    private String appSecret;
+    private String secret;
 
     /**
      * 电信sms服务所需接入码
      */
     private String extno;
 
+    /**
+     * app推送服务所需的appId
+     */
+    private String appId;
+
     /**
      * 额外配置（JSON格式）
      */

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/channel/ProviderErrorCodeMappingDO.java

@@ -25,7 +25,7 @@ public class ProviderErrorCodeMappingDO extends BaseDO {
     private String providerType;
 
     /**
-     * 渠道类型：sms/email/app_push
+     * 渠道类型：sms/email/app
      */
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/credential/SystemSecretDO.java

@@ -0,0 +1,16 @@
+package com.njcn.msgpush.module.push.dal.dataobject.credential;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+/**
+ * @author caozehui
+ * @data 2026-04-09
+ */
+@Data
+@TableName("push_sys_secret")
+public class SystemSecretDO {
+    private String id;
+    private String systemName;
+    private String secret;
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/credential/dto/CredentialReqDTO.java

@@ -0,0 +1,35 @@
+package com.njcn.msgpush.module.push.dal.dataobject.credential.dto;
+
+/**
+ * @author caozehui
+ * @data 2026-03-31
+ */
+
+import jakarta.validation.constraints.NotEmpty;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 系统凭证请求 DTO
+ *
+ * @author msgpush
+ */
+@Data
+public class CredentialReqDTO implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 上游系统名称
+     */
+    @NotEmpty(message = "上游系统名称不能为空")
+    private String systemName;
+
+    /**
+     * 密钥（用于生成凭证）
+     */
+    @NotEmpty(message = "密钥不能为空")
+    private String secretKey;
+
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/credential/dto/CredentialRespDTO.java

@@ -0,0 +1,37 @@
+package com.njcn.msgpush.module.push.dal.dataobject.credential.dto;
+
+/**
+ * @author caozehui
+ * @data 2026-03-31
+ */
+import lombok.Data;
+import lombok.experimental.Accessors;
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * 系统凭证响应 DTO
+ *
+ * @author msgpush
+ */
+@Data
+public class CredentialRespDTO implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 凭证令牌
+     */
+    private String credentialToken;
+
+    /**
+     * 上游系统名称
+     */
+    private String systemName;
+
+    /**
+     * 过期时间
+     */
+    private LocalDateTime expiresTime;
+
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/message/MessageRecordDO.java

@@ -1,5 +1,6 @@
 package com.njcn.msgpush.module.push.dal.dataobject.message;
 
+import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.njcn.msgpush.framework.mybatis.core.dataobject.BaseDO;
 import lombok.Data;
@@ -27,7 +28,7 @@ public class MessageRecordDO extends BaseDO {
     private String appName;
 
     /**
-     * 渠道类型：sms/email/app_push
+     * 渠道类型：sms/email/app
      */
     private String channel;
 
@@ -61,10 +62,6 @@ public class MessageRecordDO extends BaseDO {
      */
     private String templateParams;
 
-    /**
-     * 额外信息
-     */
-    private String extraInfo;
 
     /**
      * 状态：pending/sending/success/failed/final_failed/blacklisted/quota_exceeded/rate_limited/abandoned

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/quota/SystemQuotaConfigDO.java

@@ -19,7 +19,7 @@ public class SystemQuotaConfigDO extends BaseDO {
     private Long id;
 
     /**
-     * 渠道类型：sms/email/app_push
+     * 渠道类型：sms/email/app
      */
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/ratelimit/RateLimitConfigDO.java

@@ -17,7 +17,7 @@ public class RateLimitConfigDO extends BaseDO {
     private Long id;
 
     /**
-     * 渠道类型：sms/email/app_push
+     * 渠道类型：sms/email/app
      */
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/retry/MessageRetryHistoryDO.java

@@ -1,5 +1,6 @@
 package com.njcn.msgpush.module.push.dal.dataobject.retry;
 
+import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.njcn.msgpush.framework.mybatis.core.dataobject.BaseDO;
 import lombok.Data;
@@ -61,4 +62,9 @@ public class MessageRetryHistoryDO extends BaseDO {
      * 第三方消息ID
      */
     private String thirdPartyId;
+
+    /**
+     * 重试来源：AUTO_RETRY-系统自动重试，MANUAL_RETRY-人工手动重试
+     */
+    private String retrySource;
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/retry/MessageRetryQueueDO.java

@@ -7,62 +7,32 @@ import lombok.EqualsAndHashCode;
 
 import java.time.LocalDateTime;
 
-/**
- * @author caozehui
- * @data 2026-02-27
- * @description 消息重试队列表对应数据对象
- */
 @Data
 @TableName("push_message_retry_queue")
 @EqualsAndHashCode(callSuper = true)
 public class MessageRetryQueueDO extends BaseDO {
-    /**
-     * 主键ID
-     */
+
     private Long id;
 
-    /**
-     * 关联message_record的message_id
-     */
     private Long messageId;
 
-    /**
-     * 渠道类型
-     */
     private String channel;
 
-    /**
-     * 接收者
-     */
     private String receiver;
 
-    /**
-     * 已重试次数
-     */
     private Integer retryCount;
 
-    /**
-     * 最大重试次数
-     */
     private Integer maxRetry;
 
-    /**
-     * 下次重试时间
-     */
     private LocalDateTime nextRetryTime;
 
-    /**
-     * 首次失败时间
-     */
+    private Integer processStatus;
+
+    private LocalDateTime lockUntil;
+
     private LocalDateTime firstFailTime;
 
-    /**
-     * 最后一次重试时间
-     */
     private LocalDateTime lastRetryTime;
 
-    /**
-     * 最后失败原因
-     */
     private String lastErrorMsg;
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/dataobject/retry/RetryStrategyConfigDO.java

@@ -19,7 +19,7 @@ public class RetryStrategyConfigDO extends BaseDO {
     private Long id;
 
     /**
-     * 渠道类型：sms/email/app_push
+     * 渠道类型：sms/email/app
      */
     private String channel;
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/mysql/channel/ChannelProviderConfigMapper.java

@@ -4,10 +4,6 @@ import com.njcn.msgpush.framework.mybatis.core.mapper.BaseMapperX;
 import com.njcn.msgpush.module.push.dal.dataobject.channel.ChannelProviderConfigDO;
 import org.apache.ibatis.annotations.Mapper;
 
-import java.util.List;
-
 @Mapper
 public interface ChannelProviderConfigMapper extends BaseMapperX<ChannelProviderConfigDO> {
-
-    List<ChannelProviderConfigDO> getActiveProviders();
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/mysql/channel/mapping/ChannelProviderConfigMapper.xml

@@ -1,11 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
-<mapper namespace="com.njcn.msgpush.module.push.dal.mysql.channel.ChannelProviderConfigMapper">
-
-    <select id="getActiveProviders"
-            resultType="com.njcn.msgpush.module.push.dal.dataobject.channel.ChannelProviderConfigDO">
-        select *
-        from push_channel_provider_config
-        where enabled = 1
-    </select>
-</mapper>

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/mysql/credential/SystemSecretMappper.java

@@ -0,0 +1,11 @@
+package com.njcn.msgpush.module.push.dal.mysql.credential;
+
+import com.njcn.msgpush.framework.mybatis.core.mapper.BaseMapperX;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.SystemSecretDO;
+
+/**
+ * @author caozehui
+ * @data 2026-04-09
+ */
+public interface SystemSecretMappper extends BaseMapperX<SystemSecretDO> {
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/mysql/retry/MessageRetryQueueMapper.java

@@ -18,6 +18,12 @@ public interface MessageRetryQueueMapper extends BaseMapperX<MessageRetryQueueDO
      * @param limit       限制数量
      * @return 待重试消息列表
      */
-    List<MessageRetryQueueDO> selectNeedRetryMessages(@Param("currentTime") LocalDateTime currentTime,
+    List<MessageRetryQueueDO> selectNeedRetryMessages(@Param("channel") String channel,
+                                                      @Param("currentTime") LocalDateTime currentTime,
                                                       @Param("limit") int limit);
+
+    int claimRetryMessage(@Param("messageId") Long messageId,
+                          @Param("channel") String channel,
+                          @Param("currentTime") LocalDateTime currentTime,
+                          @Param("lockUntil") LocalDateTime lockUntil);
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/mysql/retry/mapping/MessageRetryQueueMapper.xml

@@ -5,11 +5,32 @@
     <select id="selectNeedRetryMessages" resultType="com.njcn.msgpush.module.push.dal.dataobject.retry.MessageRetryQueueDO">
         SELECT *
         FROM push_message_retry_queue
-        WHERE next_retry_time  <![CDATA[ <= ]]>  #{currentTime}
-          AND deleted = 0
+        WHERE channel = #{channel}
+          AND next_retry_time  <![CDATA[ <= ]]>  #{currentTime}
+          AND (
+            process_status = 0
+            OR process_status IS NULL
+            OR lock_until IS NULL
+            OR lock_until <![CDATA[ < ]]> #{currentTime}
+          )
         ORDER BY next_retry_time ASC
             LIMIT #{limit}
     </select>
 
+    <update id="claimRetryMessage">
+        UPDATE push_message_retry_queue
+        SET process_status = 1,
+            lock_until = #{lockUntil}
+        WHERE message_id = #{messageId}
+          AND channel = #{channel}
+          AND next_retry_time <![CDATA[ <= ]]> #{currentTime}
+          AND (
+            process_status = 0
+            OR process_status IS NULL
+            OR lock_until IS NULL
+            OR lock_until <![CDATA[ < ]]> #{currentTime}
+          )
+    </update>
+
 
 </mapper>

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/redis/MessageRetryRedisDAO.java

@@ -88,6 +88,11 @@ public class MessageRetryRedisDAO {
      * @param message 消息
      */
     public boolean addToRetryQueue(MessageRecordDO message) {
+        // 兜底守卫:nextRetryTime 为空则跳过单条,绝不让一条坏数据把整批/整个定时任务带崩
+        if (message.getNextRetryTime() == null) {
+            log.warn("跳过入队:nextRetryTime 为空, messageId={}, channel={}", message.getId(), message.getChannel());
+            return false;
+        }
         String key = getRetryQueueKey(message.getChannel());
         double score = message.getNextRetryTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/enums/ChannelTypeEnum.java

@@ -18,7 +18,7 @@ import java.util.Arrays;
 public enum ChannelTypeEnum implements ArrayValuable<String> {
     SMS("sms", "短信"),
     EMAIL("email", "邮箱"),
-    APP_PUSH("app_push", "APP 推送"),;
+    APP("app", "APP 推送"),;
 
     public static final String[] ARRAYS = Arrays.stream(values()).map(ChannelTypeEnum::getCode).toArray(String[]::new);
 

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/enums/RetrySourceEnum.java

@@ -0,0 +1,30 @@
+package com.njcn.msgpush.module.push.enums;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * @author caozehui
+ * @data 2026-04-07
+ */
+@Getter
+@AllArgsConstructor
+public enum RetrySourceEnum {
+    AUTO_RETRY("AUTO_RETRY", "系统自动重试"),
+    MANUAL_RETRY("MANUAL_RETRY", "人工手动重试");
+
+    private final String code;
+    private final String desc;
+
+    public static RetrySourceEnum getByCode(String code) {
+        if (code == null) {
+            return null;
+        }
+        for (RetrySourceEnum value : values()) {
+            if (value.getCode().equals(code)) {
+                return value;
+            }
+        }
+        return null;
+    }
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/filter/CredentialAuthenticationFilter.java

@@ -0,0 +1,115 @@
+package com.njcn.msgpush.module.push.filter;
+
+/**
+ * @author caozehui
+ * @data 2026-03-31
+ */
+
+import cn.hutool.core.util.StrUtil;
+import com.njcn.msgpush.framework.common.exception.enums.GlobalErrorCodeConstants;
+import com.njcn.msgpush.framework.common.pojo.CommonResult;
+import com.njcn.msgpush.framework.common.util.servlet.ServletUtils;
+import com.njcn.msgpush.framework.web.config.WebProperties;
+import com.njcn.msgpush.framework.web.core.filter.ApiRequestFilter;
+import com.njcn.msgpush.module.push.service.credential.CredentialServiceImpl;
+import com.njcn.msgpush.module.push.service.credential.ICredentialService;
+import jakarta.servlet.FilterChain;
+import jakarta.servlet.ServletException;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.Ordered;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+/**
+ * 凭证认证过滤器
+ *
+ * @author msgpush
+ */
+@Component
+@Slf4j
+public class CredentialAuthenticationFilter extends ApiRequestFilter implements Ordered {
+
+    @Autowired
+    private ICredentialService credentialService;
+
+    /**
+     * 凭证 Header 名称
+     */
+    private static final String CREDENTIAL_HEADER = "X-Credential-Token";
+
+    /**
+     * 需要凭证认证的接口路径
+     */
+    private static final String[] NOT_IGNORE_PATHS = {
+            "/admin-api/push/message/send/**",
+    };
+
+    private static final String[] IGNORE_PATHS = {
+            "/admin-api/push/credential/generate",
+    };
+
+    public CredentialAuthenticationFilter(WebProperties webProperties) {
+        super(webProperties);
+    }
+
+
+    @Override
+    protected boolean shouldNotFilter(HttpServletRequest request) {
+        // 检查是否在排除列表中
+        String path = request.getRequestURI().substring(request.getContextPath().length());
+        for (String notExcludePath : NOT_IGNORE_PATHS) {
+            if (path.equals(notExcludePath)) {
+                return false;
+            }
+            if (notExcludePath.endsWith("/**") && path.startsWith(notExcludePath.substring(0, notExcludePath.length() - 3))) {
+                return false;
+            }
+        }
+        for (String excludePath : IGNORE_PATHS){
+            if (path.equals(excludePath)) {
+                return true;
+            }
+            if (excludePath.endsWith("/**") && path.startsWith(excludePath.substring(0, excludePath.length() - 3))) {
+                return true;
+            }
+        }
+
+        return super.shouldNotFilter(request);
+    }
+
+    @Override
+    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain) throws ServletException, IOException {
+        // 1. 获取凭证 token
+        String credentialToken = request.getHeader(CREDENTIAL_HEADER);
+        if (StrUtil.isEmpty(credentialToken)) {
+            credentialToken = request.getParameter("credential");
+        }
+
+        // 2. 如果没有凭证，继续过滤链
+        if (StrUtil.isEmpty(credentialToken)) {
+            ServletUtils.writeJSON(response, CommonResult.error(GlobalErrorCodeConstants.UNAUTHORIZED.getCode(), "缺少凭证"));
+            return;
+        }
+
+        // 3. 验证凭证
+        CredentialServiceImpl.CredentialInfo credentialInfo = credentialService.verifyCredential(credentialToken);
+
+        if (credentialInfo == null) {
+            // 校验失败
+            ServletUtils.writeJSON(response, CommonResult.error(GlobalErrorCodeConstants.UNAUTHORIZED.getCode(), "凭证无效或已过期"));
+            return;
+        }
+
+        filterChain.doFilter(request, response);
+    }
+
+
+    @Override
+    public int getOrder() {
+        return 1000;
+    }
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/job/MessageRetryJob.java

@@ -1,5 +1,6 @@
 package com.njcn.msgpush.module.push.job;
 
+import com.njcn.msgpush.module.push.enums.ChannelTypeEnum;
 import com.njcn.msgpush.module.push.service.retry.MessageRetryQueueService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -25,8 +26,7 @@ public class MessageRetryJob {
      */
     @Scheduled(fixedRate = 10000)
     public void processSmsRetryQueue() {
-        log.info("开始处理短信重试队列：{}", LocalDateTime.now());
-        messageRetryQueueService.processRetryBatch("sms");
+        messageRetryQueueService.processRetryBatch(ChannelTypeEnum.SMS.getCode());
     }
 
     /**
@@ -34,7 +34,7 @@ public class MessageRetryJob {
      */
     //@Scheduled(fixedRate = 10000)
     public void processEmailRetryQueue() {
-        messageRetryQueueService.processRetryBatch("email");
+        messageRetryQueueService.processRetryBatch(ChannelTypeEnum.EMAIL.getCode());
     }
 
     /**
@@ -42,7 +42,7 @@ public class MessageRetryJob {
      */
     //@Scheduled(fixedRate = 10000)
     public void processAppPushRetryQueue() {
-        messageRetryQueueService.processRetryBatch("app_push");
+        messageRetryQueueService.processRetryBatch(ChannelTypeEnum.APP.getCode());
     }
 
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/channel/ChannelProviderConfigService.java

@@ -8,28 +8,18 @@ import com.njcn.msgpush.module.push.dal.dataobject.channel.ChannelProviderConfig
 import java.util.List;
 
 public interface ChannelProviderConfigService extends IService<ChannelProviderConfigDO> {
+
     Page<ChannelProviderConfigDO> getPage(ChannelProviderConfigReqVO reqVO);
 
-    List<ChannelProviderConfigDO> getActiveProviders();
-
-    /**
-     * 切换服务提供商enable字段
-     *
-     * @param id 服务提供商id
-     * @return
-     */
     ChannelProviderConfigDO toggleEnableField(String id);
 
-    /**
-     * 根据类型和渠道获取服务提供商
-     *
-     * @param providerType 服务提供商类型
-     * @param channel      渠道
-     * @return
-     */
     ChannelProviderConfigDO getByTypeAndChannel(String providerType, String channel);
 
     void failureUpdate(String providerType, String channel);
 
     void successUpdate(String providerType, String channel);
+
+    List<ChannelProviderConfigDO> getEnabledProviders(String channel);
+
+    boolean deleteByIds(List<String> ids);
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/channel/ChannelProviderConfigServiceImpl.java

@@ -1,5 +1,7 @@
 package com.njcn.msgpush.module.push.service.channel;
 
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.stream.CollectorUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
@@ -12,6 +14,9 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.stream.Collectors;
+
+import static com.njcn.msgpush.framework.common.exception.util.ServiceExceptionUtil.invalidParamException;
 
 @Service
 public class ChannelProviderConfigServiceImpl extends ServiceImpl<ChannelProviderConfigMapper, ChannelProviderConfigDO> implements ChannelProviderConfigService {
@@ -24,14 +29,31 @@ public class ChannelProviderConfigServiceImpl extends ServiceImpl<ChannelProvide
     }
 
     @Override
-    public List<ChannelProviderConfigDO> getActiveProviders() {
-        return this.lambdaQuery().eq(ChannelProviderConfigDO::getEnabled, true).list();
-    }
-
-    @Override
+    @Transactional(rollbackFor = Exception.class)
     public ChannelProviderConfigDO toggleEnableField(String id) {
         ChannelProviderConfigDO channelProviderConfigDO = this.getById(id);
-        channelProviderConfigDO.setEnabled(channelProviderConfigDO.getEnabled() ^ 0X0001);
+        if (channelProviderConfigDO == null) {
+            throw invalidParamException("渠道服务商配置不存在，id={}", id);
+        }
+
+        boolean enable = !Integer.valueOf(1).equals(channelProviderConfigDO.getEnabled());
+        if (enable) {
+            this.lambdaUpdate()
+                    .eq(ChannelProviderConfigDO::getChannel, channelProviderConfigDO.getChannel())
+                    .ne(ChannelProviderConfigDO::getId, channelProviderConfigDO.getId())
+                    .set(ChannelProviderConfigDO::getEnabled, 0)
+                    .update();
+        } else {
+            long enabledCount = this.lambdaQuery()
+                    .eq(ChannelProviderConfigDO::getChannel, channelProviderConfigDO.getChannel())
+                    .eq(ChannelProviderConfigDO::getEnabled, 1)
+                    .count();
+            if (enabledCount <= 1) {
+                throw invalidParamException("渠道 {} 必须保留一个启用服务商，请先启用其它服务商后再禁用当前配置", channelProviderConfigDO.getChannel());
+            }
+        }
+
+        channelProviderConfigDO.setEnabled(enable ? 1 : 0);
         this.updateById(channelProviderConfigDO);
         return channelProviderConfigDO;
     }
@@ -61,4 +83,29 @@ public class ChannelProviderConfigServiceImpl extends ServiceImpl<ChannelProvide
         byTypeAndChannel.setHealthStatus(1);
         this.updateById(byTypeAndChannel);
     }
+
+    @Override
+    public List<ChannelProviderConfigDO> getEnabledProviders(String channel) {
+        return this.lambdaQuery()
+                .eq(ChannelProviderConfigDO::getChannel, channel)
+                .eq(ChannelProviderConfigDO::getEnabled, 1)
+                .list();
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public boolean deleteByIds(List<String> ids) {
+        List<ChannelProviderConfigDO> configs = this.listByIds(ids);
+        List<ChannelProviderConfigDO> enabledConfigs = configs.stream()
+                .filter(config -> Integer.valueOf(1).equals(config.getEnabled()))
+                .collect(Collectors.toList());
+        if (CollectionUtil.isNotEmpty(enabledConfigs)) {
+            String channels = enabledConfigs.stream()
+                    .map(ChannelProviderConfigDO::getChannel)
+                    .distinct()
+                    .collect(Collectors.joining(","));
+            throw invalidParamException("启用中的服务商配置禁止删除，请先禁用后再删除。涉及渠道：{}", channels);
+        }
+        return this.removeBatchByIds(ids);
+    }
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/credential/CredentialServiceImpl.java

@@ -0,0 +1,218 @@
+package com.njcn.msgpush.module.push.service.credential;
+
+import cn.hutool.core.codec.Base64;
+import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.core.util.RandomUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.crypto.Mode;
+import cn.hutool.crypto.Padding;
+import cn.hutool.crypto.symmetric.AES;
+import com.njcn.msgpush.framework.common.exception.ServiceException;
+import com.njcn.msgpush.framework.common.exception.enums.ServiceErrorCodeRange;
+import com.njcn.msgpush.framework.common.util.json.JsonUtils;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.SystemSecretDO;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.dto.CredentialReqDTO;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.dto.CredentialRespDTO;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Service;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author caozehui
+ * @data 2026-03-31
+ */
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class CredentialServiceImpl implements ICredentialService {
+
+    private static final long CREDENTIAL_EXPIRE_HOURS = 24L;
+    private static final long LOCK_WAIT_SECONDS = 3L;
+    private static final long LOCK_LEASE_SECONDS = 5L;
+    private static final String SYSTEM_KEY_PREFIX = "credential:";
+    private static final String LOCK_KEY_PREFIX = "credential:lock:";
+
+    private String credentialSecretKey = "88888888888888888888888888888888"; // 32 瀛楄妭
+
+    private final ISystemSecretService systemSecretService;
+    private final StringRedisTemplate stringRedisTemplate;
+    private final RedissonClient redissonClient;
+
+    @Override
+    public CredentialRespDTO generateCredential(CredentialReqDTO reqDTO) {
+        String systemName = reqDTO.getSystemName();
+
+        boolean validatedSecretRes = validateSecret(systemName, reqDTO.getSecretKey());
+        if (!validatedSecretRes) {
+            throw new ServiceException(ServiceErrorCodeRange.VALIDATE_SYSTEM_SECRET_FAIL);
+        }
+
+        LocalDateTime expiresTime = LocalDateTime.now().plusHours(CREDENTIAL_EXPIRE_HOURS);
+        String credential = encryptCredential(systemName, expiresTime);
+
+        RLock lock = redissonClient.getLock(formatLockKey(systemName));
+        boolean locked = false;
+        try {
+            locked = lock.tryLock(LOCK_WAIT_SECONDS, LOCK_LEASE_SECONDS, TimeUnit.SECONDS);
+            if (!locked) {
+                log.warn("[generateCredential][systemName({}) failed to acquire lock within {} seconds]",
+                        systemName, LOCK_WAIT_SECONDS);
+                throw new ServiceException(ServiceErrorCodeRange.GENERATE_CREDENTIAL_FAIL);
+            }
+
+            deleteCredential(systemName);
+            CredentialInfo credentialInfo = new CredentialInfo(systemName, expiresTime, credential);
+            cacheCredential(systemName, credentialInfo);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("[generateCredential][systemName({}) interrupted while waiting for lock]", systemName, e);
+            throw new ServiceException(ServiceErrorCodeRange.GENERATE_CREDENTIAL_FAIL);
+        } finally {
+            if (locked && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+            }
+        }
+
+        return new CredentialRespDTO()
+                .setCredentialToken(systemName + StrUtil.COLON + credential)
+                .setSystemName(systemName)
+                .setExpiresTime(expiresTime);
+    }
+
+    /**
+     * 验证凭证
+     *
+     * @param credentialToken 凭证 credentialToken
+     * @return
+     */
+    @Override
+    public CredentialInfo verifyCredential(String credentialToken) {
+        if (StrUtil.isEmpty(credentialToken)) {
+            return null;
+        }
+        String[] split = credentialToken.split(StrUtil.COLON);
+        if (split.length != 2) {
+            return null;
+        }
+
+        String systemName = split[0];
+        CredentialInfo credentialInfo = getCredentialFromRedis(systemName);
+        if (ObjectUtil.isNull(credentialInfo)) {
+            return null;
+        }
+
+        if (credentialInfo.getExpiresTime().isBefore(LocalDateTime.now()) || !credentialInfo.getCredential().equals(split[1])) {
+            return null;
+        }
+
+        return credentialInfo;
+    }
+
+
+    /**
+     * 校验系统secretKey
+     *
+     * @param systemName
+     * @param secretKey
+     * @return
+     */
+    private boolean validateSecret(String systemName, String secretKey) {
+        SystemSecretDO systemSecretDO = systemSecretService.getBySystemName(systemName);
+        return systemSecretDO != null && StrUtil.equals(systemSecretDO.getSecret(), secretKey);
+    }
+
+    /**
+     * 获取加密后的凭证credential
+     *
+     * @param systemName  系统名称
+     * @param expiresTime 过期时间
+     * @return
+     */
+    private String encryptCredential(String systemName, LocalDateTime expiresTime) {
+        // 生成随机 IV
+        byte[] iv = RandomUtil.randomBytes(16);
+
+        // 创建 AES 实例
+        AES aes = new AES(
+                Mode.CBC,
+                Padding.PKCS5Padding,
+                credentialSecretKey.getBytes(StandardCharsets.UTF_8),
+                iv
+        );
+        String origin = systemName + StrUtil.C_COLON + expiresTime.getNano();
+        byte[] encrypted = aes.encrypt(origin.getBytes(StandardCharsets.UTF_8));
+        return Base64.encode(encrypted);
+    }
+
+    /**
+     * 缓存系统凭证credential
+     *
+     * @param systemName
+     * @param info
+     */
+    private void cacheCredential(String systemName, CredentialInfo info) {
+        String credentialKey = formatCredentialKey(systemName);
+        String jsonValue = JsonUtils.toJsonString(info);
+
+        long remainingSeconds = Duration.between(LocalDateTime.now(), info.getExpiresTime()).getSeconds();
+        if (remainingSeconds > 0) {
+            stringRedisTemplate.opsForValue().set(credentialKey, jsonValue, remainingSeconds, TimeUnit.SECONDS);
+        }
+    }
+
+    /**
+     * 从redis中获取系统凭证信息
+     *
+     * @param systemName
+     * @return
+     */
+    private CredentialInfo getCredentialFromRedis(String systemName) {
+        String credentialKey = formatCredentialKey(systemName);
+        String jsonValue = stringRedisTemplate.opsForValue().get(credentialKey);
+        if (StrUtil.isEmpty(jsonValue)) {
+            return null;
+        }
+        return JsonUtils.parseObject(jsonValue, CredentialInfo.class);
+    }
+
+
+    private void deleteCredential(String systemName) {
+        String credentialKey = formatCredentialKey(systemName);
+        String oldToken = stringRedisTemplate.opsForValue().get(credentialKey);
+        if (StrUtil.isNotEmpty(oldToken)) {
+            stringRedisTemplate.delete(credentialKey);
+        }
+    }
+
+    private String formatCredentialKey(String systemName) {
+        return SYSTEM_KEY_PREFIX + systemName;
+    }
+
+    /**
+     * 获取锁的key（用于针对同一时刻多个请求，保证多个请求只对同一个key进行加锁）
+     *
+     * @param systemName
+     * @return
+     */
+    private String formatLockKey(String systemName) {
+        return LOCK_KEY_PREFIX + systemName;
+    }
+
+    @Data
+    @AllArgsConstructor
+    public static class CredentialInfo {
+        private String systemName;
+        private LocalDateTime expiresTime;
+        private String credential;
+    }
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/credential/ICredentialService.java

@@ -0,0 +1,26 @@
+package com.njcn.msgpush.module.push.service.credential;
+
+import com.njcn.msgpush.module.push.dal.dataobject.credential.dto.CredentialReqDTO;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.dto.CredentialRespDTO;
+
+/**
+ * @author caozehui
+ * @data 2026-03-31
+ */
+public interface ICredentialService {
+    /**
+     * 生成系统凭证
+     *
+     * @param reqDTO 请求参数
+     * @return 凭证响应
+     */
+    CredentialRespDTO generateCredential(CredentialReqDTO reqDTO);
+
+    /**
+     * 验证凭证是否有效
+     *
+     * @param token 凭证 token
+     * @return 凭证信息
+     */
+    CredentialServiceImpl.CredentialInfo verifyCredential(String token);
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/credential/ISystemSecretService.java

@@ -0,0 +1,12 @@
+package com.njcn.msgpush.module.push.service.credential;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.SystemSecretDO;
+
+/**
+ * @author caozehui
+ * @data 2026-04-09
+ */
+public interface ISystemSecretService extends IService<SystemSecretDO> {
+    SystemSecretDO getBySystemName(String systemName);
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/credential/SystemSecretServiceImpl.java

@@ -0,0 +1,18 @@
+package com.njcn.msgpush.module.push.service.credential;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.njcn.msgpush.module.push.dal.dataobject.credential.SystemSecretDO;
+import com.njcn.msgpush.module.push.dal.mysql.credential.SystemSecretMappper;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author caozehui
+ * @data 2026-04-09
+ */
+@Service
+public class SystemSecretServiceImpl extends ServiceImpl<SystemSecretMappper, SystemSecretDO> implements ISystemSecretService {
+    @Override
+    public SystemSecretDO getBySystemName(String systemName) {
+        return this.lambdaQuery().eq(SystemSecretDO::getSystemName, systemName).one();
+    }
+}

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/message/MessageRecordService.java

@@ -5,29 +5,31 @@ import com.baomidou.mybatisplus.extension.service.IService;
 import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageRecordReqVO;
 import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageSendResultVO;
 import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
+import com.njcn.msgpush.module.push.enums.ChannelTypeEnum;
+import com.njcn.msgpush.module.push.enums.RetrySourceEnum;
 
-import java.io.Serializable;
-import java.time.LocalDateTime;
-import java.util.Collection;
 import java.util.List;
 
 public interface MessageRecordService extends IService<MessageRecordDO> {
 
     /**
-     * 发送消息（包括email、sms、app_push）
+     * 发送消息（包括email、sms、app）
      *
      * @param reqVOList
+     * @param systemName
+     * @return channelTypeEnum
      * @return 发送的结果
      */
-    List<MessageSendResultVO> send(List<MessageRecordReqVO> reqVOList);
+    List<MessageSendResultVO> send(List<MessageRecordReqVO> reqVOList, ChannelTypeEnum channelTypeEnum, String systemName);
 
     /**
      * 处理发送消息
      *
      * @param messageRecordDOList
+     * @param retrySource
      * @return 发送的结果
      */
-    List<MessageSendResultVO> processSendMsg(List<MessageRecordDO> messageRecordDOList);
+    List<MessageSendResultVO> processSendMsg(List<MessageRecordDO> messageRecordDOList, RetrySourceEnum retrySource);
 
     /**
      * 添加消息记录

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/message/MessageRecordServiceImpl.java

@@ -1,6 +1,7 @@
 package com.njcn.msgpush.module.push.service.message;
 
 import cn.hutool.core.bean.BeanUtil;
+import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
@@ -9,6 +10,7 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.njcn.msgpush.framework.common.util.object.PageUtils;
 import com.njcn.msgpush.module.push.checker.MsgPushGuardChain;
 import com.njcn.msgpush.module.push.client.factory.MessageProviderFactory;
+import com.njcn.msgpush.module.push.client.factory.MessageProviderFactoryRegistry;
 import com.njcn.msgpush.module.push.client.sender.SendOutcome;
 import com.njcn.msgpush.module.push.client.sender.SendResult;
 import com.njcn.msgpush.module.push.client.sender.Sender;
@@ -23,11 +25,11 @@ import com.njcn.msgpush.module.push.dal.redis.MessageConfirmRedisDAO;
 import com.njcn.msgpush.module.push.dal.redis.RateLimitRedisDAO;
 import com.njcn.msgpush.module.push.dal.redis.SystemQuotaRedisDAO;
 import com.njcn.msgpush.module.push.enums.ChannelTypeEnum;
+import com.njcn.msgpush.module.push.enums.RetrySourceEnum;
 import com.njcn.msgpush.module.push.service.channel.ChannelProviderConfigService;
 import com.njcn.msgpush.module.push.service.retry.MessageRetryHistoryService;
 import com.njcn.msgpush.module.push.service.retry.MessageRetryQueueService;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -35,7 +37,6 @@ import org.springframework.transaction.annotation.Transactional;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 @Service
 public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, MessageRecordDO> implements MessageRecordService {
@@ -53,8 +54,7 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
     public ChannelProviderConfigService channelProviderConfigService;
 
     @Autowired
-    @Qualifier("messageProviderFactoryMap")
-    private Map<String, MessageProviderFactory> messageProviderFactoryMap;
+    private MessageProviderFactoryRegistry messageProviderFactoryRegistry;
 
     @Autowired
     private MsgPushGuardChain msgPushGuardChain;
@@ -69,16 +69,18 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
     private MessageConfirmRedisDAO messageConfirmRedisDAO;
 
     @Override
-    public List<MessageSendResultVO> send(List<MessageRecordReqVO> reqVOList) {
-        List<MessageRecordDO> messageRecordDOList = this.createMessageRecords(reqVOList);
-        return this.processSendMsg(messageRecordDOList);
+    public List<MessageSendResultVO> send(List<MessageRecordReqVO> reqVOList, ChannelTypeEnum channelTypeEnum, String systemName) {
+        List<MessageRecordDO> messageRecordDOList = this.createMessageRecords(reqVOList, channelTypeEnum, systemName);
+        return this.processSendMsg(messageRecordDOList, null);
     }
 
     @Transactional(rollbackFor = Exception.class)
-    public List<MessageRecordDO> createMessageRecords(List<MessageRecordReqVO> reqVOList) {
+    public List<MessageRecordDO> createMessageRecords(List<MessageRecordReqVO> reqVOList, ChannelTypeEnum channelTypeEnum, String systemName) {
         List<MessageRecordDO> messageRecordDOList = BeanUtil.copyToList(reqVOList, MessageRecordDO.class);
         messageRecordDOList.forEach(messageRecordDO -> {
+            messageRecordDO.setChannel(channelTypeEnum.getCode());
             messageRecordDO.setStatus(MsgStatusConstant.PENDING);
+            messageRecordDO.setAppName(systemName);
 //            messageRecordDO.setRetryCount(0);
         });
         this.saveBatch(messageRecordDOList);
@@ -90,12 +92,12 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
      * 这里统一接收 sender 返回的 SendResult，再决定消息状态、重试队列和服务商健康度如何变化。
      */
     @Override
-    public List<MessageSendResultVO> processSendMsg(List<MessageRecordDO> messageRecordDOList) {
-        msgPushGuardChain.checkAll(messageRecordDOList);
+    public List<MessageSendResultVO> processSendMsg(List<MessageRecordDO> messageRecordDOList, RetrySourceEnum retrySource) {
+        msgPushGuardChain.checkAll(messageRecordDOList, retrySource);
         List<MessageSendResultVO> resultList = new ArrayList<>();
         for (MessageRecordDO messageRecordDO : messageRecordDOList) {
             try {
-                MessageSendResultVO resultVO = this.processSingleMessage(messageRecordDO);
+                MessageSendResultVO resultVO = this.processSingleMessage(messageRecordDO, retrySource);
                 resultList.add(resultVO);
             } catch (Exception e) {
                 MessageSendResultVO failedVO = new MessageSendResultVO();
@@ -108,7 +110,7 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
         return resultList;
     }
 
-    private MessageSendResultVO processSingleMessage(MessageRecordDO messageRecordDO) {
+    private MessageSendResultVO processSingleMessage(MessageRecordDO messageRecordDO, RetrySourceEnum retrySource) {
         MessageSendResultVO resultVO = new MessageSendResultVO();
         resultVO.setMessageId(messageRecordDO.getId());
 
@@ -118,13 +120,24 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
             this.updateById(messageRecordDO);
             return resultVO;
         }
+        List<ChannelProviderConfigDO> enabledProviders = channelProviderConfigService.getEnabledProviders(messageRecordDO.getChannel());
+        SendResult sendResult = null;
+        if (CollectionUtil.isNotEmpty(enabledProviders)) {
+            ChannelProviderConfigDO channelProviderConfigDO = enabledProviders.get(0);
+            MessageProviderFactory messageProviderFactory = messageProviderFactoryRegistry.getFactory(channelProviderConfigDO.getProviderType());
+            sendResult = this.validateProviderAndChannel(messageRecordDO, channelProviderConfigDO, messageProviderFactory);
+
+            messageRecordDO.setProviderType(channelProviderConfigDO.getProviderType());
+            if (sendResult == null) {
+                sendResult = this.sendMessage(messageRecordDO, channelProviderConfigDO, messageProviderFactory);
+            }
+            this.applySendResult(messageRecordDO, sendResult, retrySource);
+            this.recordRetryHistory(messageRecordDO, retrySource);
+        } else {
+            sendResult = this.validateProviderAndChannel(messageRecordDO, null, null);
+            this.applySendResult(messageRecordDO, sendResult, retrySource);
+        }
 
-        ChannelProviderConfigDO channelProviderConfigDO = channelProviderConfigService
-                .getByTypeAndChannel(messageRecordDO.getProviderType(), messageRecordDO.getChannel());
-        MessageProviderFactory messageProviderFactory = messageProviderFactoryMap.get(messageRecordDO.getProviderType());
-        SendResult sendResult = this.sendMessage(messageRecordDO, channelProviderConfigDO, messageProviderFactory);
-        this.applySendResult(messageRecordDO, sendResult);
-        this.recordRetryHistory(messageRecordDO);
 
         // 在这里修改接口返回的是投递结果，还是调用接口结果
         resultVO.setResult(this.isSuccessOutcome(sendResult.getOutcome()));
@@ -132,26 +145,29 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
         return resultVO;
     }
 
-    private SendResult sendMessage(MessageRecordDO messageRecordDO,
-                                   ChannelProviderConfigDO channelProviderConfigDO,
-                                   MessageProviderFactory messageProviderFactory) {
+    private SendResult validateProviderAndChannel(MessageRecordDO messageRecordDO, ChannelProviderConfigDO channelProviderConfigDO, MessageProviderFactory messageProviderFactory) {
+        ChannelTypeEnum channelType = ChannelTypeEnum.getByCode(messageRecordDO.getChannel());
+        if (channelType == null) {
+            return SendResult.unsupported("当前发送渠道类型暂不支持");
+        }
         if (ObjectUtil.isNull(messageProviderFactory)) {
             return SendResult.configInvalid("当前服务商未启用或未注册");
         }
         if (ObjectUtil.isNull(channelProviderConfigDO)) {
             return SendResult.configInvalid("当前服务商渠道配置不存在");
         }
+        return null;
+    }
 
+    private SendResult sendMessage(MessageRecordDO messageRecordDO,
+                                   ChannelProviderConfigDO channelProviderConfigDO,
+                                   MessageProviderFactory messageProviderFactory) {
         ChannelTypeEnum channelType = ChannelTypeEnum.getByCode(messageRecordDO.getChannel());
-        if (channelType == null) {
-            return SendResult.unsupported("当前发送渠道类型暂不支持");
-        }
-
         return switch (channelType) {
             case SMS -> messageProviderFactory.createSmsSender(channelProviderConfigDO, sender).sendSms(messageRecordDO);
             case EMAIL -> messageProviderFactory.createEmailSender(channelProviderConfigDO, sender)
                     .sendEmail(messageRecordDO, new HashMap<>());
-            case APP_PUSH -> messageProviderFactory.createAppPushSender(channelProviderConfigDO, sender).appPush(messageRecordDO);
+            case APP -> messageProviderFactory.createAppPushSender(channelProviderConfigDO, sender).appPush(messageRecordDO);
         };
     }
 
@@ -160,7 +176,7 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
      * 主流程在这里统一做状态落库、重试编排和健康度更新。
      */
     @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
-    public void applySendResult(MessageRecordDO messageRecordDO, SendResult sendResult) {
+    public void applySendResult(MessageRecordDO messageRecordDO, SendResult sendResult, RetrySourceEnum retrySource) {
         messageRecordDO.setStatus(this.mapOutcomeToStatus(sendResult.getOutcome()));
         messageRecordDO.setErrorCode(sendResult.getErrorCode());
         messageRecordDO.setErrorMsg(sendResult.getMessage());
@@ -177,6 +193,13 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
         // 成功或已受理都视为本次调用已正常投递，更新成功健康度并消耗配额。
         if (SendOutcome.ACCEPTED.equals(sendResult.getOutcome())) {
             messageConfirmRedisDAO.addToConfirmQueue(messageRecordDO);
+
+            // 仅首发(retrySource==null)消耗每日配额/接收者频率计数;重发(AUTO_RETRY/MANUAL_RETRY)
+            // 是同一条消息的再次投递,不重复扣额——与守卫豁免配套,避免重发把自己的配额耗尽再被拦。
+            if (retrySource == null) {
+                systemQuotaRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName());
+                rateLimitRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName(), messageRecordDO.getReceiver());
+            }
         } else if (SendOutcome.RETRYABLE_FAILED.equals(sendResult.getOutcome())) {
             messageRetryQueueService.saveOrUpdateRetryMessage(messageRecordDO);
             channelProviderConfigService.failureUpdate(messageRecordDO.getProviderType(), messageRecordDO.getChannel());
@@ -192,11 +215,12 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
     }
 
     @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
-    public void recordRetryHistory(MessageRecordDO messageRecordDO) {
+    public void recordRetryHistory(MessageRecordDO messageRecordDO, RetrySourceEnum retrySource) {
         MessageRetryHistoryDO messageRetryHistoryDO = BeanUtil.copyProperties(messageRecordDO, MessageRetryHistoryDO.class, "id");
         messageRetryHistoryDO.setMessageId(messageRecordDO.getId());
         messageRetryHistoryDO.setStatus(messageRecordDO.getStatus());
         messageRetryHistoryDO.setRetrySequence(messageRetryHistoryService.getMaxRetrySequence(messageRecordDO.getId()));
+        messageRetryHistoryDO.setRetrySource(retrySource != null ? retrySource.getCode() : null);
         messageRetryHistoryService.add(messageRetryHistoryDO);
     }
 
@@ -256,9 +280,7 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
     public Page<MessageRecordDO> getPage(MessageRecordReqVO reqVO) {
         QueryWrapper<MessageRecordDO> wrapper = new QueryWrapper<>();
         wrapper.lambda()
-                .eq(StrUtil.isNotBlank(reqVO.getChannel()), MessageRecordDO::getChannel, reqVO.getChannel())
-                .eq(StrUtil.isNotBlank(reqVO.getProviderType()), MessageRecordDO::getProviderType, reqVO.getProviderType())
-                .eq(StrUtil.isNotBlank(reqVO.getAppName()), MessageRecordDO::getAppName, reqVO.getAppName());
+                .eq(StrUtil.isNotBlank(reqVO.getMessageType()), MessageRecordDO::getMessageType, reqVO.getMessageType());
         return this.page(new Page<>(PageUtils.getPageNum(reqVO), PageUtils.getPageSize(reqVO)), wrapper);
     }
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/MessageRetryQueueServiceImpl.java

@@ -7,15 +7,14 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.njcn.msgpush.framework.common.pojo.PageResult;
 import com.njcn.msgpush.module.push.constant.MsgStatusConstant;
-import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageSendResultVO;
 import com.njcn.msgpush.module.push.controller.admin.retry.vo.MessageRetryQueueReqVO;
 import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
 import com.njcn.msgpush.module.push.dal.dataobject.retry.MessageRetryQueueDO;
 import com.njcn.msgpush.module.push.dal.dataobject.retry.RetryStrategyConfigDO;
 import com.njcn.msgpush.module.push.dal.mysql.retry.MessageRetryQueueMapper;
-import com.njcn.msgpush.module.push.dal.redis.MessageConfirmRedisDAO;
 import com.njcn.msgpush.module.push.dal.redis.MessageRetryRedisDAO;
 import com.njcn.msgpush.module.push.enums.ChannelTypeEnum;
+import com.njcn.msgpush.module.push.enums.RetrySourceEnum;
 import com.njcn.msgpush.module.push.service.message.MessageRecordService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -24,43 +23,52 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 @Slf4j
 @Service
-public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueMapper, MessageRetryQueueDO> implements MessageRetryQueueService, CommandLineRunner {
+public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueMapper, MessageRetryQueueDO>
+        implements MessageRetryQueueService, CommandLineRunner {
+
+    private static final int DEFAULT_BATCH_SIZE = 100;
+    private static final int DEFAULT_MAX_RETRY_COUNT = 5;
+    private static final int PROCESS_STATUS_WAITING = 0;
+    private static final int PROCESS_STATUS_PROCESSING = 1;
+    private static final int DEFAULT_LOCK_SECONDS = 120;
+
+    /** 限流类失败的统一错误码(回执后由 Sender/MessageRecordServiceImpl 写入 message.errorCode) */
+    private static final String RATE_LIMIT_ERROR_CODE = "RATE_LIMIT";
+    /** 限流类固定重试间隔:2 分钟(不走阶梯) */
+    private static final int RATE_LIMIT_INTERVAL_SECONDS = 120;
+    /** 限流类时间兜底:首次失败起满 12 小时未成功才放弃 */
+    private static final long RATE_LIMIT_MAX_DURATION_SECONDS = 43200;
+
+    private static final List<String> CHANNELS = Arrays.asList(
+            ChannelTypeEnum.SMS.getCode(),
+            ChannelTypeEnum.EMAIL.getCode(),
+            ChannelTypeEnum.APP.getCode());
+
     @Autowired
     private MessageRetryRedisDAO messageRetryRedisDAO;
     @Autowired
     private MessageRecordService messageRecordService;
     @Autowired
-    public RetryStrategyConfigService retryStrategyConfigService;
-    @Autowired
-    private MessageConfirmRedisDAO messageConfirmRedisDAO;
-
-    private static final int DEFAULT_BATCH_SIZE = 100;
-
-    private static final int DEFAULT_MAX_RETRY_COUNT = 5;
-
-    private static final List<String> CHANNELS = Arrays.asList(
-            ChannelTypeEnum.SMS.getCode(),
-            ChannelTypeEnum.EMAIL.getCode(),
-            ChannelTypeEnum.APP_PUSH.getCode());
+    private RetryStrategyConfigService retryStrategyConfigService;
 
     private final AtomicBoolean startupRebuildCompleted = new AtomicBoolean(false);
 
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void saveOrUpdateRetryMessage(MessageRecordDO message) {
-        MessageRetryQueueDO existing = super.baseMapper.selectOne(
-                new LambdaQueryWrapper<MessageRetryQueueDO>()
-                        .eq(MessageRetryQueueDO::getMessageId, message.getId())
-        );
+        MessageRetryQueueDO existing = baseMapper.selectOne(new LambdaQueryWrapper<MessageRetryQueueDO>()
+                .eq(MessageRetryQueueDO::getMessageId, message.getId()));
         message.setLastRetryTime(message.getSendTime());
 
         if (ObjectUtil.isNull(existing)) {
@@ -68,48 +76,91 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
             retryRecord.setMessageId(message.getId());
             retryRecord.setChannel(message.getChannel());
             retryRecord.setReceiver(message.getReceiver());
-
-            retryRecord.setRetryCount(1);
+            boolean rateLimit = isRateLimit(message);
+            // I3:首次失败若是限流,普通失败计数从 0 起(限流不占普通次数预算);普通失败则从 1 起
+            retryRecord.setRetryCount(rateLimit ? 0 : 1);
             RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(message.getChannel());
             retryRecord.setMaxRetry(ObjectUtil.isNull(strategyConfig) ? DEFAULT_MAX_RETRY_COUNT : strategyConfig.getMaxRetryCount());
             retryRecord.setFirstFailTime(message.getSendTime());
             retryRecord.setLastRetryTime(message.getSendTime());
             retryRecord.setLastErrorMsg(message.getErrorMsg());
-            LocalDateTime nextRetryTime = this.calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount());
+            LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount(), rateLimit);
             retryRecord.setNextRetryTime(nextRetryTime);
+            retryRecord.setProcessStatus(PROCESS_STATUS_WAITING);
+            retryRecord.setLockUntil(null);
             message.setNextRetryTime(nextRetryTime);
-            this.save(retryRecord);
+            save(retryRecord);
             if (!messageRetryRedisDAO.addToRetryQueue(message)) {
                 log.warn("消息已写入数据库重试队列，但 Redis 入队失败: messageId={}", message.getId());
             }
             return;
         }
 
-        int newRetryCount = existing.getRetryCount() + 1;
+        applyRetryUpdate(message, existing);
+    }
+
+    private void applyRetryUpdate(MessageRecordDO message, MessageRetryQueueDO existing) {
+        boolean rateLimit = isRateLimit(message);
+        // I3:限流重试冻结 retryCount(不消耗普通失败的次数预算),普通失败才累加
+        int newRetryCount = nextRetryCount(existing.getRetryCount(), rateLimit);
         existing.setLastRetryTime(message.getSendTime());
         existing.setLastErrorMsg(message.getErrorMsg());
 
-        if (newRetryCount >= existing.getMaxRetry()) {
+        boolean exhausted = rateLimit
+                ? isRateLimitExhausted(existing.getFirstFailTime(), LocalDateTime.now())
+                : newRetryCount >= existing.getMaxRetry();
+        if (exhausted) {
             message.setStatus(MsgStatusConstant.FINALFAILED);
             message.setNextRetryTime(null);
             existing.setNextRetryTime(null);
-            this.updateById(existing);
-            this.deleteByMessageIds(Collections.singletonList(message.getId()));
+            updateById(existing);
+            deleteByMessageIds(Collections.singletonList(message.getId()));
             messageRetryRedisDAO.removeFromRetryQueue(message.getChannel(), message.getId());
             return;
         }
 
-        LocalDateTime nextRetryTime = this.calculateNextRetryTime(message.getChannel(), newRetryCount);
+        LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), newRetryCount, rateLimit);
         existing.setRetryCount(newRetryCount);
         existing.setNextRetryTime(nextRetryTime);
+        existing.setProcessStatus(PROCESS_STATUS_WAITING);
+        existing.setLockUntil(null);
         message.setNextRetryTime(nextRetryTime);
-        this.updateById(existing);
+        updateById(existing);
         if (!messageRetryRedisDAO.addToRetryQueue(message)) {
             log.warn("消息已更新数据库重试队列，但 Redis 入队失败: messageId={}", message.getId());
         }
     }
 
-    private LocalDateTime calculateNextRetryTime(String channel, int retryCount) {
+    /**
+     * 计算本次失败后队列记录的新 retryCount。
+     * I3:限流重试不计入次数预算 —— 限流时冻结(返回原值),普通失败时累加。
+     * 这样普通失败的 maxRetry 上限只统计普通失败,既不被限流重试撑爆提前 FINALFAILED,
+     * 也不会让普通失败借限流的 12h 兜底绕过次数上限。
+     */
+    static int nextRetryCount(int currentRetryCount, boolean rateLimit) {
+        return rateLimit ? currentRetryCount : currentRetryCount + 1;
+    }
+
+    /** 限流类识别:仅当统一错误码为 RATE_LIMIT */
+    static boolean isRateLimit(MessageRecordDO message) {
+        return message != null && RATE_LIMIT_ERROR_CODE.equals(message.getErrorCode());
+    }
+
+    /**
+     * 限流类时间兜底判定:首次失败起是否已满 12 小时。
+     * firstFailTime 为 null(历史数据)时视为未满,继续重试。
+     */
+    static boolean isRateLimitExhausted(LocalDateTime firstFailTime, LocalDateTime now) {
+        if (firstFailTime == null) {
+            return false;
+        }
+        return Duration.between(firstFailTime, now).getSeconds() >= RATE_LIMIT_MAX_DURATION_SECONDS;
+    }
+
+    LocalDateTime calculateNextRetryTime(String channel, int retryCount, boolean rateLimit) {
+        if (rateLimit) {
+            return LocalDateTime.now().plusSeconds(RATE_LIMIT_INTERVAL_SECONDS);
+        }
         RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(channel);
         long plusSeconds = 0;
         if (ObjectUtil.isNull(strategyConfig)) {
@@ -136,7 +187,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
                         plusSeconds = 60 * 60 * 4;
                     }
                 }
-                case APP_PUSH -> {
+                case APP -> {
                     if (retryCount == 1) {
                         plusSeconds = 60;
                     } else {
@@ -147,11 +198,9 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
         } else {
             String retryIntervals = strategyConfig.getRetryIntervals();
             String[] split = retryIntervals.split(String.valueOf(StrUtil.C_COMMA));
-            if (retryCount >= split.length) {
-                plusSeconds = Long.parseLong(split[split.length - 1]);
-            } else {
-                plusSeconds = Long.parseLong(split[retryCount - 1]);
-            }
+            plusSeconds = retryCount >= split.length
+                    ? Long.parseLong(split[split.length - 1])
+                    : Long.parseLong(split[retryCount - 1]);
         }
         return LocalDateTime.now().plusSeconds(plusSeconds);
     }
@@ -159,7 +208,6 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
     @Override
     public void processRetryBatch(String channel) {
         long epochMilli = System.currentTimeMillis();
-        // 使用原子认领机制，一次性完成读取 + 删除
         List<String> claimedMessageIds = messageRetryRedisDAO.claimAndRemoveDueMessages(channel, epochMilli, DEFAULT_BATCH_SIZE);
 
         if (CollUtil.isEmpty(claimedMessageIds)) {
@@ -168,48 +216,49 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
 
             epochMilli = System.currentTimeMillis();
             claimedMessageIds = messageRetryRedisDAO.claimAndRemoveDueMessages(channel, epochMilli, DEFAULT_BATCH_SIZE);
-
             if (CollUtil.isEmpty(claimedMessageIds)) {
                 return;
             }
         }
 
-        List<MessageRecordDO> messageRecordDOList = messageRecordService.listByIds(claimedMessageIds);
+        LocalDateTime currentTime = LocalDateTime.now();
+        LocalDateTime lockUntil = currentTime.plusSeconds(DEFAULT_LOCK_SECONDS);
+        List<Long> dbClaimedMessageIds = claimedMessageIds.stream()
+                .map(Long::valueOf)
+                .filter(messageId -> claimRetryMessage(messageId, channel, currentTime, lockUntil))
+                .collect(Collectors.toList());
+
+        if (CollUtil.isEmpty(dbClaimedMessageIds)) {
+            return;
+        }
+
+        List<MessageRecordDO> messageRecordDOList = messageRecordService.listByIds(dbClaimedMessageIds);
         for (MessageRecordDO messageRecordDO : messageRecordDOList) {
-            processSingleRetry(messageRecordDO);
+            processSingleRetry(messageRecordDO, RetrySourceEnum.AUTO_RETRY);
         }
     }
 
-    /**
-     * 处理单条消息重试，根据执行结果决定是否重新入队
-     * 成功：不再回 Redis，并逻辑删除数据库队列记录
-     * 失败且继续重试：重新计算 nextRetryTime，重新写入 Redis 和数据库队列表
-     * 失败且终止：不再回 Redis，并逻辑删除数据库队列记录
-     *
-     * @param messageRecordDO 消息记录
-     */
-    private void processSingleRetry(MessageRecordDO messageRecordDO) {
-        messageRecordDO.setStatus(MsgStatusConstant.PENDING);
-        messageRecordService.processSendMsg(Collections.singletonList(messageRecordDO)).get(0);
-        messageRecordService.updateRetryCount(messageRecordDO.getId());
-    }
-
-    /**
-     * 删除重试记录（数据库逻辑删除 + Redis 已提前删除）
-     */
-    private void deleteRetryRecord(MessageRecordDO message) {
-        this.deleteByMessageIds(Collections.singletonList(message.getId()));
-        messageRecordService.updateById(message);
+    private void processSingleRetry(MessageRecordDO messageRecordDO, RetrySourceEnum retrySource) {
+        if (MsgStatusConstant.RETRYABLE_FAILED.equals(messageRecordDO.getStatus())) {
+            messageRecordDO.setStatus(MsgStatusConstant.PENDING);
+            messageRecordService.processSendMsg(Collections.singletonList(messageRecordDO), retrySource);
+            messageRecordService.updateRetryCount(messageRecordDO.getId());
+            finalizeRetryLease(messageRecordDO);
+        } else {
+            // 能被重新认领说明 120s 租约已过期、20s 回执窗口早过 —— SENDING/PENDING 即卡住,与终态一并清理
+            deleteByMessageIds(Collections.singletonList(messageRecordDO.getId()));
+        }
     }
 
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void manualRetry(List<String> messageIds) {
-        if (CollUtil.isNotEmpty(messageIds)) {
-            List<MessageRecordDO> messageRecordDOList = messageRecordService.listByIds(messageIds);
-            for (MessageRecordDO messageRecordDO : messageRecordDOList) {
-                processSingleRetry(messageRecordDO);
-            }
+        if (CollUtil.isEmpty(messageIds)) {
+            return;
+        }
+        List<MessageRecordDO> messageRecordDOList = messageRecordService.listByIds(messageIds);
+        for (MessageRecordDO messageRecordDO : messageRecordDOList) {
+            processSingleRetry(messageRecordDO, RetrySourceEnum.MANUAL_RETRY);
         }
     }
 
@@ -223,7 +272,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
                 .le(ObjectUtil.isNotNull(reqVO.getMaxRetryCount()), MessageRetryQueueDO::getRetryCount, reqVO.getMaxRetryCount())
                 .orderByAsc(MessageRetryQueueDO::getNextRetryTime)
                 .orderByDesc(MessageRetryQueueDO::getId);
-        return this.baseMapper.selectPage(reqVO, queryWrapper);
+        return baseMapper.selectPage(reqVO, queryWrapper);
     }
 
     @Override
@@ -231,14 +280,11 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
         if (CollUtil.isEmpty(messageIds)) {
             return false;
         }
-
-        return this.remove(new LambdaQueryWrapper<MessageRetryQueueDO>()
-                .in(MessageRetryQueueDO::getMessageId, messageIds));
+        return this.baseMapper.delete(new LambdaQueryWrapper<MessageRetryQueueDO>().in(MessageRetryQueueDO::getMessageId, messageIds)) > 0;
     }
 
-
     @Override
-    public void run(String... args) throws Exception {
+    public void run(String... args) {
         rebuildRetryQueueOnStartup();
         startupRebuildCompleted.set(true);
     }
@@ -247,18 +293,13 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
     @Transactional(rollbackFor = Exception.class)
     public void rebuildRetryQueueOnStartup() {
         log.info("========== 开始重建重试队列 ==========");
-
         for (String channel : CHANNELS) {
             try {
                 log.info("正在清空渠道 [{}] 的 Redis 重试队列", channel);
                 messageRetryRedisDAO.clearRetryQueue(channel);
 
                 log.info("正在从数据库加载渠道 [{}] 的待重试消息", channel);
-                List<MessageRetryQueueDO> dbRecords = this.lambdaQuery()
-                        .eq(MessageRetryQueueDO::getChannel, channel)
-                        .isNotNull(MessageRetryQueueDO::getNextRetryTime)
-                        .list();
-
+                List<MessageRetryQueueDO> dbRecords = listEligibleQueueRecords(channel);
                 if (CollUtil.isEmpty(dbRecords)) {
                     log.info("渠道 [{}] 没有待重试的消息", channel);
                     continue;
@@ -266,24 +307,23 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
 
                 List<Long> messageIds = dbRecords.stream().map(MessageRetryQueueDO::getMessageId).collect(Collectors.toList());
                 List<MessageRecordDO> messages = messageRecordService.listByIds(messageIds);
-
                 if (CollUtil.isEmpty(messages)) {
                     log.warn("渠道 [{}] 的消息记录不存在", channel);
                     continue;
                 }
+                // push_message_record.next_retry_time 不保证落库,用队列记录(权威调度时间)回填后再入队,避免 NPE
+                fillNextRetryTimeFromQueue(messages, dbRecords);
 
                 int successCount = messageRetryRedisDAO.batchAddToRetryQueue(messages);
-                log.info("渠道 [{}] 重建完成：数据库记录数={}, 成功入队 Redis 数={}",
-                        channel, dbRecords.size(), successCount);
-
+                log.info("渠道 [{}] 重建完成：数据库记录数={}, 成功入队 Redis 数={}", channel, dbRecords.size(), successCount);
             } catch (Exception e) {
                 log.error("重建渠道 [{}] 的重试队列失败", channel, e);
             }
         }
-
         log.info("========== 重试队列重建完成 ==========");
     }
 
+    // 每60分钟主动从数据库中扫描一次，将消息重试数据同步到 Redis 中
     @Scheduled(fixedRate = 600000)
     @Override
     public void syncRetryQueueConsistency() {
@@ -291,35 +331,31 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
             log.debug("启动重建未完成，跳过本次同步");
             return;
         }
-        log.info("开始同步redis与数据库重试队列一致性");
+        log.info("开始同步 Redis 与数据库重试队列一致性");
 
         for (String channel : CHANNELS) {
-            List<MessageRetryQueueDO> dbRecords = this.lambdaQuery()
-                    .eq(MessageRetryQueueDO::getChannel, channel)
-                    .isNotNull(MessageRetryQueueDO::getNextRetryTime)
-                    .list();
-
+            List<MessageRetryQueueDO> dbRecords = listEligibleQueueRecords(channel);
             if (CollUtil.isEmpty(dbRecords)) {
                 continue;
             }
 
             for (MessageRetryQueueDO dbRecord : dbRecords) {
                 boolean existsInRedis = messageRetryRedisDAO.existsInRetryQueue(channel, dbRecord.getMessageId());
+                if (existsInRedis) {
+                    continue;
+                }
 
-                if (!existsInRedis) {
-                    Long messageId = dbRecord.getMessageId();
-                    MessageRecordDO message = messageRecordService.getById(messageId);
-
-                    if (message != null && message.getNextRetryTime() != null) {
-                        boolean added = messageRetryRedisDAO.addToRetryQueue(message);
-                        if (added) {
-                            log.debug("补回消息到 Redis: channel={}, messageId={}", channel, messageId);
-                        } else {
-                            log.warn("补回消息到 Redis 失败：channel={}, messageId={}", channel, messageId);
-                        }
+                Long messageId = dbRecord.getMessageId();
+                MessageRecordDO message = messageRecordService.getById(messageId);
+                if (message != null && message.getNextRetryTime() != null) {
+                    boolean added = messageRetryRedisDAO.addToRetryQueue(message);
+                    if (added) {
+                        log.debug("补回消息到 Redis: channel={}, messageId={}", channel, messageId);
                     } else {
-                        log.warn("消息记录不存在或 next_retry_time 为空：channel={}, messageId={}", channel, messageId);
+                        log.warn("补回消息到 Redis 失败：channel={}, messageId={}", channel, messageId);
                     }
+                } else {
+                    log.warn("消息记录不存在或 next_retry_time 为空：channel={}, messageId={}", channel, messageId);
                 }
             }
         }
@@ -327,37 +363,115 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
 
     @Override
     public void fallbackToDatabase(String channel) {
-        boolean redisAvailable = messageRetryRedisDAO.isRedisAvailable();
-        if (!redisAvailable) {
+        if (!messageRetryRedisDAO.isRedisAvailable()) {
             log.warn("Redis 不可用，channel={}", channel);
             return;
         }
-
         fallbackToDatabaseDirectly(channel);
     }
 
-
-    /**
-     * 从数据库补拉
-     *
-     * @param channel
-     */
     private void fallbackToDatabaseDirectly(String channel) {
-        List<MessageRetryQueueDO> dbRecords = baseMapper.selectNeedRetryMessages(LocalDateTime.now(), DEFAULT_BATCH_SIZE);
-
+        List<MessageRetryQueueDO> dbRecords = baseMapper.selectNeedRetryMessages(channel, LocalDateTime.now(), DEFAULT_BATCH_SIZE);
         if (CollUtil.isEmpty(dbRecords)) {
             return;
         }
 
         List<Long> messageIds = dbRecords.stream().map(MessageRetryQueueDO::getMessageId).collect(Collectors.toList());
         List<MessageRecordDO> messages = messageRecordService.listByIds(messageIds);
-
         if (CollUtil.isEmpty(messages)) {
             return;
         }
+        // push_message_record.next_retry_time 不保证落库,用队列记录(权威调度时间)回填后再入队,避免 NPE
+        fillNextRetryTimeFromQueue(messages, dbRecords);
 
         if (messageRetryRedisDAO.isRedisAvailable()) {
             messageRetryRedisDAO.batchAddToRetryQueue(messages);
         }
     }
+
+    /**
+     * 用队列记录(push_message_retry_queue,调度权威)的 nextRetryTime 回填消息对象。
+     * push_message_record.next_retry_time 只在内存设置、不保证持久化,批量入队若直接取它会 NPE。
+     * 这些 message 仅用于 Redis 入队、不会回写库,故内存回填安全。
+     */
+    private void fillNextRetryTimeFromQueue(List<MessageRecordDO> messages, List<MessageRetryQueueDO> queueRecords) {
+        Map<Long, LocalDateTime> nextRetryTimeMap = queueRecords.stream()
+                .filter(r -> r.getNextRetryTime() != null)
+                .collect(Collectors.toMap(MessageRetryQueueDO::getMessageId, MessageRetryQueueDO::getNextRetryTime, (a, b) -> a));
+        messages.forEach(m -> {
+            LocalDateTime queueTime = nextRetryTimeMap.get(m.getId());
+            if (queueTime != null) {
+                m.setNextRetryTime(queueTime);
+            }
+        });
+    }
+
+    /**
+     * 通过设置lockUntil时间、process_status=1来从数据库层抢占该条重试消息。
+     */
+    private boolean claimRetryMessage(Long messageId, String channel, LocalDateTime currentTime, LocalDateTime lockUntil) {
+        return baseMapper.claimRetryMessage(messageId, channel, currentTime, lockUntil) > 0;
+    }
+
+    /**
+     * 完成重试租约处理，根据消息状态执行相应的清理或重置操作
+     * - 如果消息状态为可重试失败，直接返回，不进行任何处理
+     * - SENDING(两段式中间态)直接返回，保留记录与租约，等回执推进
+     * - 如果消息状态为待处理，重置队列记录的处理状态和锁时间，并将消息重新加入Redis重试队列
+     * - 其他状态（成功、最终失败等），从Redis重试队列移除并删除数据库中的重试队列记录
+     *
+     * @param messageRecordDO 消息记录对象
+     */
+    private void finalizeRetryLease(MessageRecordDO messageRecordDO) {
+        String status = messageRecordDO.getStatus();
+
+        // 可重试失败:保留记录,等下一轮认领重发 —— 不变
+        if (MsgStatusConstant.RETRYABLE_FAILED.equals(status)) {
+            return;
+        }
+
+        // 两段式发送的中间态:重发已受理、消息在途,保留记录、保持租约锁住、不回 Redis,等 20s 回执推进
+        if (MsgStatusConstant.SENDING.equals(status)) {
+            return;
+        }
+
+        // 处理过程中断但仍停留在 PENDING:主动释放 lease 并回队列(兜底补偿)—— 不变
+        if (MsgStatusConstant.PENDING.equals(status)) {
+            lambdaUpdate()
+                    .eq(MessageRetryQueueDO::getMessageId, messageRecordDO.getId())
+                    .set(MessageRetryQueueDO::getProcessStatus, PROCESS_STATUS_WAITING)
+                    .set(MessageRetryQueueDO::getLockUntil, null)
+                    .update();
+            if (messageRecordDO.getNextRetryTime() != null) {
+                messageRetryRedisDAO.addToRetryQueue(messageRecordDO);
+            }
+            return;
+        }
+
+        // 其余即终态:从 Redis 移除并删除数据库重试队列记录
+        messageRetryRedisDAO.removeFromRetryQueue(messageRecordDO.getChannel(), messageRecordDO.getId());
+        deleteByMessageIds(Collections.singletonList(messageRecordDO.getId()));
+    }
+
+    /**
+     * 查询指定通道下符合重试条件的队列记录
+     * 筛选条件：
+     * - 匹配指定通道
+     * - 下次重试时间不为空
+     * - 处理状态为等待中，或处理状态为空，或锁时间为空，或锁时间已过期
+     *
+     * @param channel 消息通道
+     * @return 符合条件的重试队列记录列表
+     */
+    private List<MessageRetryQueueDO> listEligibleQueueRecords(String channel) {
+        LocalDateTime now = LocalDateTime.now();
+        return lambdaQuery()
+                .eq(MessageRetryQueueDO::getChannel, channel)
+                .isNotNull(MessageRetryQueueDO::getNextRetryTime)
+                .and(wrapper -> wrapper.eq(MessageRetryQueueDO::getProcessStatus, PROCESS_STATUS_WAITING)
+                        .or().isNull(MessageRetryQueueDO::getProcessStatus)
+                        .or().isNull(MessageRetryQueueDO::getLockUntil)
+                        .or().lt(MessageRetryQueueDO::getLockUntil, now))
+                .list();
+    }
 }

msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/RetryStrategyConfigService.java

@@ -34,5 +34,11 @@ public interface RetryStrategyConfigService {
      */
     boolean updateStrategyConfig(RetryStrategyConfigReqVO strategyConfigReqVO);
 
+    /**
+     * 切换重试策略的启用/禁用状态
+     *
+     * @param id 重试策略配置ID
+     * @return 是否切换成功
+     */
     boolean toggleEnableField(String id);
 }

msgpush-module-push/msgpush-module-push-server/src/main/resources/sql/provider-error-code-mapping/telecom_sms_MA0038_retryable.sql

@@ -0,0 +1,29 @@
+-- ============================================================================
+-- 电信短信「短期频繁发送」限流码 MA:0038 配置为可重试
+-- ----------------------------------------------------------------------------
+-- 背景：电信下行回执查询(action=select)对该类消息返回 status=6 / stat=MA:0038。
+--   1) 代码侧已在 TelecomSmsSender.getDownInfo 把 status=6 纳入失败映射分支；
+--   2) 本脚本在错误码映射表为 MA:0038 配置 should_retry=1，
+--      使其经 Sender.buildFailureResult 判定为 RETRYABLE_FAILED，进入重试队列。
+-- 两者缺一不可。SMS 首次重试间隔默认 5 分钟，足以越过服务商 1 分钟限流窗口。
+--
+-- 该脚本幂等：已存在同 (provider_type, channel, original_code) 的有效记录时不重复插入。
+-- ============================================================================
+INSERT INTO push_provider_error_code_mapping
+    (provider_type, channel, original_code, original_message,
+     unified_error_code, unified_error_message, error_category,
+     should_retry, final_status, remark,
+     creator, updater, create_time, update_time, deleted)
+SELECT
+    'telecom', 'sms', 'MA:0038', '短期频繁发送',
+    'RATE_LIMIT', '服务商短期频繁发送限流', 'PROVIDER_ERROR',
+    1, 'retryable_failed', '电信限流码，需自动重试',
+    'system', 'system', NOW(), NOW(), 0
+FROM DUAL
+WHERE NOT EXISTS (
+    SELECT 1 FROM push_provider_error_code_mapping
+    WHERE provider_type = 'telecom'
+      AND channel = 'sms'
+      AND original_code = 'MA:0038'
+      AND deleted = 0
+);

msgpush-module-push/msgpush-module-push-server/src/main/resources/sql/sms_retry_strategy_ladder.sql

@@ -0,0 +1,46 @@
+-- ============================================================================
+-- 普通类短信失败的重试阶梯配置(rootfix 后)
+-- ----------------------------------------------------------------------------
+-- 背景:限流类(电信 MA:0038)已由代码硬编码 120s 间隔 + 12h 兜底,忽略本表。
+--   故 push_retry_strategy_config.retry_intervals 现仅作用于「普通类」sms 可
+--   重试失败(超时 / 通道故障 / 第三方调用失败等),与限流类无关。
+--
+-- 决策(TD-3,2026-06-20):普通类用收紧的阶梯退避,替代旧 quickfix 的 60s 平铺。
+--   retry_intervals = '60,120,300,600'  → 1→2→5→10 分钟,超出后按 10 分钟重复
+--   max_retry_count = 6                  → 第 6 次普通失败转 final_failed(约 28 分钟放弃)
+--
+-- 幂等:可重复执行。⚠ retry_intervals 不能含空格(代码 Long.parseLong 不容忍空格)。
+-- 关系:本文件取代 sms_ratelimit_retry_quickfix.sql 的 ② 段;
+--       ① 错误码映射见 provider-error-code-mapping/telecom_sms_MA0038_retryable.sql。
+-- ============================================================================
+
+-- 1) 已存在启用中的 sms 策略行 → 覆盖为新阶梯
+--    (同时纠正 max_retry_count,覆盖旧 quickfix 写入的 '60' / max 10)
+UPDATE push_retry_strategy_config
+SET retry_intervals = '60,120,300,600',
+    max_retry_count = 6,
+    update_time     = NOW()
+WHERE channel = 'sms'
+  AND enabled = 1
+  AND deleted = 0;
+
+-- 2) 没有启用中的 sms 策略行 → 插一条(否则代码回落硬编码 300/600/1800、max 5)
+INSERT INTO push_retry_strategy_config
+    (channel, max_retry_count, retry_intervals, enabled,
+     creator, updater, create_time, update_time, deleted)
+SELECT 'sms', 6, '60,120,300,600', 1,
+       'system', 'system', NOW(), NOW(), 0
+FROM DUAL
+WHERE NOT EXISTS (
+    SELECT 1 FROM push_retry_strategy_config
+    WHERE channel = 'sms'
+      AND enabled = 1
+      AND deleted = 0
+);
+
+-- ----------------------------------------------------------------------------
+-- 执行后自检(可选):确认普通类阶梯已就位
+-- ----------------------------------------------------------------------------
+-- SELECT channel, max_retry_count, retry_intervals, enabled
+-- FROM push_retry_strategy_config
+-- WHERE channel = 'sms' AND enabled = 1 AND deleted = 0;

msgpush-module-push/msgpush-module-push-server/src/test/java/com/njcn/msgpush/module/push/service/retry/RetryDecisionLogicTest.java

@@ -0,0 +1,79 @@
+package com.njcn.msgpush.module.push.service.retry;
+
+import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * 限流重试治本核心的纯决策逻辑单测（无 Spring / 无 DB / 无 Redis）。
+ */
+class RetryDecisionLogicTest {
+
+    @Test
+    void isRateLimit_trueOnlyForRateLimitErrorCode() {
+        MessageRecordDO rateLimited = new MessageRecordDO();
+        rateLimited.setErrorCode("RATE_LIMIT");
+        assertTrue(MessageRetryQueueServiceImpl.isRateLimit(rateLimited));
+
+        MessageRecordDO other = new MessageRecordDO();
+        other.setErrorCode("SOMETHING_ELSE");
+        assertFalse(MessageRetryQueueServiceImpl.isRateLimit(other));
+
+        MessageRecordDO noCode = new MessageRecordDO();
+        assertFalse(MessageRetryQueueServiceImpl.isRateLimit(noCode));
+
+        assertFalse(MessageRetryQueueServiceImpl.isRateLimit(null));
+    }
+
+    @Test
+    void isRateLimitExhausted_nullFirstFailTimeMeansNotExhausted() {
+        assertFalse(MessageRetryQueueServiceImpl.isRateLimitExhausted(null, LocalDateTime.now()));
+    }
+
+    @Test
+    void isRateLimitExhausted_falseBelow12h_trueAtOrAbove12h() {
+        LocalDateTime now = LocalDateTime.now();
+        // 43199s < 12h → 未满
+        assertFalse(MessageRetryQueueServiceImpl.isRateLimitExhausted(now.minusSeconds(43199), now));
+        // 43200s == 12h → 满（>=）
+        assertTrue(MessageRetryQueueServiceImpl.isRateLimitExhausted(now.minusSeconds(43200), now));
+        // 43201s > 12h → 满
+        assertTrue(MessageRetryQueueServiceImpl.isRateLimitExhausted(now.minusSeconds(43201), now));
+    }
+
+    @Test
+    void calculateNextRetryTime_rateLimitUsesFixed120s() {
+        MessageRetryQueueServiceImpl service = new MessageRetryQueueServiceImpl();
+        LocalDateTime before = LocalDateTime.now();
+        LocalDateTime next = service.calculateNextRetryTime("SMS", 1, true);
+        long seconds = Duration.between(before, next).getSeconds();
+        // rate-limit 分支在使用任何字段前就 return，故 new 裸实例可调；允许执行抖动 ±2s
+        assertTrue(seconds >= 118 && seconds <= 122, "expected ~120s, got " + seconds);
+    }
+
+    @Test
+    void nextRetryCount_rateLimitFreezes_normalIncrements() {
+        // 普通失败:累加,消耗次数预算
+        assertEquals(2, MessageRetryQueueServiceImpl.nextRetryCount(1, false));
+        // 限流失败:冻结,不消耗次数预算(靠 12h 兜底终止)
+        assertEquals(1, MessageRetryQueueServiceImpl.nextRetryCount(1, true));
+    }
+
+    @Test
+    void nextRetryCount_rateLimitRetriesDoNotConsumeNormalBudget() {
+        // I3 回归:首次普通失败后,连续多次限流重试不应吃掉普通失败的次数预算
+        int rc = 1;                                                  // 首次普通失败入队
+        rc = MessageRetryQueueServiceImpl.nextRetryCount(rc, true);  // 限流重试
+        rc = MessageRetryQueueServiceImpl.nextRetryCount(rc, true);  // 限流重试
+        rc = MessageRetryQueueServiceImpl.nextRetryCount(rc, true);  // 限流重试
+        assertEquals(1, rc, "限流重试不应增加普通失败计数");
+        rc = MessageRetryQueueServiceImpl.nextRetryCount(rc, false); // 第 2 次普通失败
+        assertEquals(2, rc, "普通失败才累加,且计数从未被限流污染");
+    }
+}

msgpush-module-push/msgpush-module-push-server/src/test/java/com/njcn/msgpush/module/push/sms/MsgPushClientTest.java

@@ -2,6 +2,7 @@ package com.njcn.msgpush.module.push.sms;
 
 import com.alibaba.fastjson.JSON;
 import com.njcn.msgpush.module.push.client.factory.MessageProviderFactory;
+import com.njcn.msgpush.module.push.client.factory.MessageProviderFactoryRegistry;
 import com.njcn.msgpush.module.push.client.sender.Sender;
 import com.njcn.msgpush.module.push.client.sender.SmsSender;
 import com.njcn.msgpush.module.push.controller.admin.message.vo.MessageRecordReqVO;
@@ -13,15 +14,12 @@ import com.njcn.msgpush.module.push.service.message.MessageRecordService;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
-import java.util.UUID;
 
 /**
  * @author caozehui
@@ -36,8 +34,7 @@ public class MsgPushClientTest {
     private MessageRecordService messageRecordService;
 
     @Autowired
-    @Qualifier("messageProviderFactoryMap")
-    private Map<String, MessageProviderFactory> messageProviderFactoryMap;
+    private MessageProviderFactoryRegistry messageProviderFactoryRegistry;
 
     @Autowired
     private Sender sender;
@@ -57,29 +54,27 @@ public class MsgPushClientTest {
 //            message.setProviderType(MsgPushConstant.PROVIDER_TYPE_ALI_YUN);
 //            messageIdList.add(message);
 //        }
-        for (int i = 0; i < 1; i++) {
-            MessageRecordReqVO message = new MessageRecordReqVO();
-            message.setId(1234567890L);
-            message.setAppName("NPQS-9000");
-            message.setChannel(ChannelTypeEnum.SMS.getMsg());
-            message.setReceiver("18839431215");
-            message.setContent("【南京灿能电力】测试短信" + i + "，请忽略。" + LocalDateTime.now().format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
-            message.setProviderType(ProviderTypeEnum.TELECOM.getCode());
-            messageIdList.add(message);
-        }
-        List<MessageSendResultVO> sendResult = messageRecordService.send(messageIdList);
-
-        Thread.sleep(9000);
-        System.out.println(JSON.toJSONString(sendResult));
+//        for (int i = 0; i < 1; i++) {
+//            MessageRecordReqVO message = new MessageRecordReqVO();
+//            message.setId(1234567890L);
+//            message.setChannel(ChannelTypeEnum.SMS.getMsg());
+//            message.setReceiver("18839431215");
+//            message.setContent("【南京灿能电力】测试短信" + i + "，请忽略。" + LocalDateTime.now().format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+//            messageIdList.add(message);
+//        }
+//        List<MessageSendResultVO> sendResult = messageRecordService.send(messageIdList, ChannelTypeEnum.SMS);
+//
+//        Thread.sleep(9000);
+//        System.out.println(JSON.toJSONString(sendResult));
     }
 
     @Test
     public void templateSelect() {
-        MessageProviderFactory messageProviderFactory = messageProviderFactoryMap.get(ProviderTypeEnum.TELECOM.getCode());
+        MessageProviderFactory messageProviderFactory = messageProviderFactoryRegistry.getFactory(ProviderTypeEnum.TELECOM.getCode());
         ChannelProviderConfigDO config = new ChannelProviderConfigDO();
         config.setApiUrl("https://sms.ymeeting.cn/smsv2");
         config.setAppKey("925631");
-        config.setAppSecret("AMW2pOVrdky");
+        config.setSecret("AMW2pOVrdky");
         config.setExtno("106905631");
         SmsSender smsSender = messageProviderFactory.createSmsSender(config, sender);
         String templateIdentifier = "SMS_481710295";

msgpush-module-system/msgpush-module-system-api/pom.xml

@@ -24,7 +24,7 @@
         <!-- Web 相关 -->
         <dependency>
             <groupId>io.swagger.core.v3</groupId>
-            <artifactId>swagger-annotations</artifactId>
+            <artifactId>swagger-annotations-jakarta</artifactId>
         </dependency>
 
         <!-- 参数校验 -->
@@ -43,4 +43,4 @@
 
     </dependencies>
 
-</project>
+</project>

pom.xml

@@ -27,6 +27,7 @@
         <!-- maven-surefire-plugin 暂时无法通过 bom 的依赖读取（兼容老版本 IDEA 2024 及以前版本） -->
         <lombok.version>1.18.42</lombok.version>
         <spring.boot.version>3.5.9</spring.boot.version>
+        <swagger.jakarta.version>2.2.38</swagger.jakarta.version>
         <mapstruct.version>1.6.3</mapstruct.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
@@ -119,6 +120,12 @@
                 <artifactId>msgpush-spring-boot-starter-websocket</artifactId>
                 <version>${revision}</version>
             </dependency>
+
+            <dependency>
+                <groupId>io.swagger.core.v3</groupId>
+                <artifactId>swagger-annotations-jakarta</artifactId>
+                <version>${swagger.jakarta.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>