feat:手动推送消息

2026-10-19 13:25:56 +08:00
1 parent 8f303f1c63
commit a00f7aa80d
@@ -0,0 +1,45 @@
+package com.bsd.pjcloud.thirdparty.dto.appmessage;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Pattern;
+import jakarta.validation.constraints.Positive;
+import jakarta.validation.constraints.Size;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.List;
+
+/** PC手动批量通知请求，统一文案只用于本次发送，不保存为可复用模板。 */
+@Data
+public class AppManualPushDto implements Serializable {
+
+    /** 单次入箱人数上限，限制事务规模，与PC选择器保持一致。 */
+    public static final int MAX_RECIPIENTS = 500;
+
+    /** 当前来源商户ID，服务端必须复核管理员管理范围。 */
+    @NotNull(message = "商户ID不能为空")
+    @Positive(message = "商户ID无效")
+    private Long mchId;
+
+    /** 当前租户内的接收用户ID，服务端按当前登录人的数据权限再次复核。 */
+    @NotEmpty(message = "请选择接收用户")
+    @Size(max = MAX_RECIPIENTS, message = "单次最多推送500位用户")
+    private List<@NotNull(message = "用户ID不能为空") @Positive(message = "用户ID无效") Long> userIds;
+
+    /** 统一纯文本标题，长度与现有消息标题列宽128字符一致。 */
+    @NotBlank(message = "通知标题不能为空")
+    @Size(max = 128, message = "通知标题最多128字符")
+    private String title;
+
+    /** 统一纯文本正文，长度与现有消息正文列宽2000字符一致。 */
+    @NotBlank(message = "通知正文不能为空")
+    @Size(max = 2000, message = "通知正文最多2000字符")
+    private String content;
+
+    /** 客户端为本次草稿快照生成的幂等标识，网络重试不变，修改草稿后重新生成。 */
+    @NotBlank(message = "提交标识不能为空")
+    @Pattern(regexp = "[A-Za-z0-9_-]{16,64}", message = "提交标识格式无效")
+    private String requestId;
+}
@@ -0,0 +1,22 @@
+package com.bsd.pjcloud.thirdparty.vo.appmessage;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/** 手动推送入箱结果，投递任务已创建不代表通知已经到达设备。 */
+@Data
+public class AppManualPushResultVo implements Serializable {
+
+    /** 本次已确认的客户端提交标识。 */
+    private String requestId;
+
+    /** 已写入App消息列表的接收人数。 */
+    private int recipientCount;
+
+    /** 已加入Push投递流程的人数，重试查询包含已处理的有效任务。 */
+    private int queuedCount;
+
+    /** 因用户关闭Push或无有效设备而跳过通知栏推送的人数，不影响入箱。 */
+    private int skippedCount;
+}
@@ -17,7 +17,7 @@ public class AppMessageVo implements Serializable {
    /** 固定业务事件编码。 */
    private String eventCode;

-    /** 事件展示分组，wallet或meal。 */
+    /** 事件展示分组，wallet钱包、meal用餐、system人工通知。 */
    private String category;

    /** 接收类型，OWNER本人或RELATION关系人。 */
@@ -34,6 +34,15 @@ public class AppMessageConstants {
    /** 用餐核销事件来源服务。 */
    public static final String SOURCE_HARDWARE = "HARDWARE";

+    /** PC人工通知来源服务，区别于自动钱包和用餐事件。 */
+    public static final String SOURCE_THIRD = "THIRD";
+
+    /** 手动推送的稳定事件编码，只用于消息和投递，不加入自动事件策略或关系订阅矩阵。 */
+    public static final String EVENT_MANUAL_PUSH = "MANUAL_PUSH";
+
+    /** 手动通知在App中的展示分类，跨端协议值为system。 */
+    public static final String CATEGORY_SYSTEM = "system";
+
    /** 新建后等待供应商投递的状态。 */
    public static final String DELIVERY_PENDING = "PENDING";

@@ -6,12 +6,15 @@ import com.bsd.pjcloud.common.core.constant.enums.MerchantTypeEnum;
 import com.bsd.pjcloud.common.core.util.R;
 import com.bsd.pjcloud.common.log.annotation.SysLog;
 import com.bsd.pjcloud.common.security.util.SecurityUtils;
+import com.bsd.pjcloud.thirdparty.dto.appmessage.AppManualPushDto;
 import com.bsd.pjcloud.thirdparty.dto.appmessage.AppPushDeliveryRetryDto;
 import com.bsd.pjcloud.thirdparty.dto.appmessage.AppPushPolicySaveDto;
 import com.bsd.pjcloud.thirdparty.dto.appmessage.AppPushTestDto;
 import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdAppMessageSettingService;
+import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdAppManualPushService;
 import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdPushDeliveryService;
 import com.bsd.pjcloud.thirdparty.vo.appmessage.AppMessageEventOptionVo;
+import com.bsd.pjcloud.thirdparty.vo.appmessage.AppManualPushResultVo;
 import com.bsd.pjcloud.thirdparty.vo.appmessage.AppPushDeliveryAttemptVo;
 import com.bsd.pjcloud.thirdparty.vo.appmessage.AppPushDeliveryVo;
 import com.bsd.pjcloud.thirdparty.vo.appmessage.AppPushPolicyVo;
@@ -56,6 +59,24 @@ public class ThirdAppMessagePcController {
    /** Wallet商户内部服务，用于复核当前管理员可管理的商户数据范围。 */
    private final RemoteMchService remoteMchService;

+    /** 手动批量通知入箱服务，复用既有异步Push队列。 */
+    private final ThirdAppManualPushService manualPushService;
+
+    /**
+     * 为当前可管理商户向选定用户发送统一通知，同时写入App消息列表。
+     *
+     * @param dto 接收用户、纯文本内容与幂等提交标识
+     * @return 入箱、加入投递及跳过人数，不表示设备已经收到通知
+     */
+    @Operation(summary = "手动批量发送App通知")
+    @SysLog("手动批量发送App通知")
+    @PreAuthorize("@pms.hasPermission('wallet_mchManage_setting')")
+    @PostMapping("/manual-push")
+    public R<AppManualPushResultVo> manualPush(@Valid @RequestBody AppManualPushDto dto) {
+        requireMerchantAccess(dto.getMchId());
+        return R.ok(manualPushService.send(dto, SecurityUtils.getUser().getId()));
+    }
+
    /**
     * 查询指定商户全事件策略矩阵。
     *
@@ -0,0 +1,17 @@
+package com.bsd.pjcloud.thirdparty.service.appmessage;
+
+import com.bsd.pjcloud.thirdparty.dto.appmessage.AppManualPushDto;
+import com.bsd.pjcloud.thirdparty.vo.appmessage.AppManualPushResultVo;
+
+/** 手动批量通知入箱服务，不参与钱包和用餐的自动事件策略。 */
+public interface ThirdAppManualPushService {
+
+    /**
+     * 复核接收人权限后原子创建消息与Push任务，相同管理员、商户及提交标识保持幂等。
+     *
+     * @param dto 已通过参数校验且商户管理权限已复核的请求
+     * @param operatorId 当前登录管理员ID，不接受客户端提供
+     * @return 已入箱、已加入投递及跳过的人数；任一接收人不可用时整批拒绝
+     */
+    AppManualPushResultVo send(AppManualPushDto dto, Long operatorId);
+}
@@ -0,0 +1,235 @@
+package com.bsd.pjcloud.thirdparty.service.impl.appmessage;
+
+import cn.hutool.crypto.digest.DigestUtil;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.bsd.pjcloud.admin.api.dto.UserDTO;
+import com.bsd.pjcloud.admin.api.feign.RemoteUserService;
+import com.bsd.pjcloud.admin.api.vo.UserVO;
+import com.bsd.pjcloud.common.core.util.R;
+import com.bsd.pjcloud.common.data.tenant.TenantContextHolder;
+import com.bsd.pjcloud.thirdparty.constant.appmessage.AppMessageConstants;
+import com.bsd.pjcloud.thirdparty.dto.appmessage.AppManualPushDto;
+import com.bsd.pjcloud.thirdparty.entity.appmessage.ThirdAppMessageEntity;
+import com.bsd.pjcloud.thirdparty.entity.appmessage.ThirdPushDeliveryEntity;
+import com.bsd.pjcloud.thirdparty.mapper.appmessage.ThirdAppMessageMapper;
+import com.bsd.pjcloud.thirdparty.mapper.appmessage.ThirdPushDeliveryMapper;
+import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdAppManualPushService;
+import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdAppMessageSettingService;
+import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdPushDeliveryService;
+import com.bsd.pjcloud.thirdparty.vo.appmessage.AppManualPushResultVo;
+import com.bsd.pjcloud.wallet.api.feign.RemoteMchService;
+import com.bsd.pjcloud.wallet.api.vo.MchManageVo;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** 手动通知复用现有收件箱和异步投递链路，以数据库唯一约束保证重试不重复入箱。 */
+@Service
+@RequiredArgsConstructor
+public class ThirdAppManualPushServiceImpl implements ThirdAppManualPushService {
+
+    /** 每次按权限回查的用户数，避免单个远程查询携带过多ID。 */
+    private static final int USER_QUERY_BATCH_SIZE = 100;
+
+    /** 当前请求携带登录令牌的UPMS客户端，不使用绕过数据权限的内部用户查询。 */
+    private final RemoteUserService remoteUserService;
+
+    /** 商户详情客户端，仅用于读取已经过管理权限校验的商户名称快照。 */
+    private final RemoteMchService remoteMchService;
+
+    /** 消息持久化入口，自动应用租户和逻辑删除条件。 */
+    private final ThirdAppMessageMapper messageMapper;
+
+    /** 投递任务查询入口，用于统计实际跳过人数。 */
+    private final ThirdPushDeliveryMapper deliveryMapper;
+
+    /** 用户Push总开关查询服务，手动通知不读取自动事件或关系订阅开关。 */
+    private final ThirdAppMessageSettingService settingService;
+
+    /** 既有投递服务，负责设备校验和异步任务创建，不在请求中调用供应商。 */
+    private final ThirdPushDeliveryService deliveryService;
+
+    /** 统一JSON序列化器，用于规范化请求指纹和保存幂等快照。 */
+    private final ObjectMapper objectMapper;
+
+    /** 显式事务边界，唯一键冲突必须先回滚，再读取另一并发请求的完整结果。 */
+    private final TransactionTemplate transactionTemplate;
+
+    /**
+     * 校验全部用户后在一个本地事务内写入消息及Push任务；相同提交快照返回既有结果。
+     *
+     * @param dto 已完成参数和商户权限校验的统一消息请求
+     * @param operatorId 当前登录管理员ID
+     * @return 完整入箱结果，任一写入失败时不保留部分接收人消息
+     */
+    @Override
+    public AppManualPushResultVo send(AppManualPushDto dto, Long operatorId) {
+        if (operatorId == null || TenantContextHolder.getTenantId() == null) {
+            throw new IllegalArgumentException("登录信息无效，请重新登录");
+        }
+        List<Long> userIds = dto.getUserIds().stream().distinct().sorted().toList();
+        if (userIds.isEmpty() || userIds.size() > AppManualPushDto.MAX_RECIPIENTS) {
+            throw new IllegalArgumentException("请选择1至500位接收用户");
+        }
+        requireVisibleRecipients(userIds);
+        String eventId = "MANUAL:" + operatorId + ":" + dto.getMchId() + ":" + dto.getRequestId();
+        String fingerprint = DigestUtil.sha256Hex(serialize(List.of(operatorId, dto.getMchId(), userIds,
+                dto.getTitle().trim(), dto.getContent().trim())));
+        String payloadJson = serialize(Map.of("requestHash", fingerprint));
+        List<ThirdAppMessageEntity> existing = findBatch(eventId);
+        if (!existing.isEmpty()) {
+            return replayResult(dto, userIds, payloadJson, existing);
+        }
+        R<MchManageVo> merchant = remoteMchService.getById(dto.getMchId());
+        if (merchant == null || !merchant.isOk() || merchant.getData() == null) {
+            throw new IllegalArgumentException("来源商户不存在或暂时无法读取");
+        }
+        try {
+            return Objects.requireNonNull(transactionTemplate.execute(status ->
+                    createBatch(dto, userIds, eventId, payloadJson, merchant.getData().getName())));
+        } catch (DuplicateKeyException conflict) {
+            // 两个相同请求同时通过首次查询时，唯一键会让其中一个事务整体回滚；此处已在事务外。
+            List<ThirdAppMessageEntity> committed = findBatch(eventId);
+            if (committed.isEmpty()) throw conflict;
+            return replayResult(dto, userIds, payloadJson, committed);
+        }
+    }
+
+    /**
+     * 按当前登录令牌逐批复核用户，任何不存在、跨租户或不在数据范围内的ID都导致整批失败。
+     *
+     * @param userIds 去重后的完整接收人ID
+     */
+    private void requireVisibleRecipients(List<Long> userIds) {
+        for (int offset = 0; offset < userIds.size(); offset += USER_QUERY_BATCH_SIZE) {
+            List<Long> batch = userIds.subList(offset, Math.min(offset + USER_QUERY_BATCH_SIZE, userIds.size()));
+            UserDTO query = new UserDTO();
+            query.setUserIds(batch);
+            R<Page<UserVO>> response = remoteUserService.getScopeBasicUserPageWithCountOption(1L,
+                    (long) USER_QUERY_BATCH_SIZE, false, query);
+            if (response == null || !response.isOk() || response.getData() == null || response.getData().getRecords() == null) {
+                throw new IllegalStateException("复核接收用户失败，请稍后重试");
+            }
+            Set<Long> visible = response.getData().getRecords().stream().map(UserVO::getUserId).collect(Collectors.toSet());
+            if (!visible.containsAll(batch)) {
+                throw new IllegalArgumentException("部分接收用户已不可用或不在当前权限范围内，请重新选择");
+            }
+        }
+    }
+
+    /**
+     * 创建整个批次的本人消息与投递任务，调用方必须已开启本地事务。
+     *
+     * @param dto 本次通知请求
+     * @param userIds 已经复核权限的接收人ID
+     * @param eventId 管理员、商户与提交标识组成的稳定事件ID
+     * @param payloadJson 请求指纹快照，用于阻止同一标识被不同内容复用
+     * @param merchantName 来源商户名称快照
+     * @return 全部接收人的入箱与投递汇总
+     */
+    private AppManualPushResultVo createBatch(AppManualPushDto dto, List<Long> userIds, String eventId, String payloadJson, String merchantName) {
+        LocalDateTime occurredTime = LocalDateTime.now();
+        List<ThirdAppMessageEntity> messages = new ArrayList<>();
+        for (Long userId : userIds) {
+            ThirdAppMessageEntity message = new ThirdAppMessageEntity();
+            message.setTenantId(TenantContextHolder.getTenantId());
+            message.setEventId(eventId);
+            message.setEventCode(AppMessageConstants.EVENT_MANUAL_PUSH);
+            message.setRecipientUserId(userId);
+            message.setSubjectUserId(userId);
+            message.setRecipientType(AppMessageConstants.RECIPIENT_OWNER);
+            message.setSourceService(AppMessageConstants.SOURCE_THIRD);
+            message.setSourceBizType(AppMessageConstants.EVENT_MANUAL_PUSH);
+            message.setSourceBizId(dto.getRequestId());
+            message.setSourceMchId(dto.getMchId());
+            message.setSourceMchName(merchantName);
+            message.setTitle(dto.getTitle().trim());
+            message.setContent(dto.getContent().trim());
+            message.setPayloadJson(payloadJson);
+            message.setDeepLinkPath(AppMessageConstants.MESSAGE_DETAIL_PATH);
+            message.setReadStatus(AppMessageConstants.UNREAD);
+            message.setOccurredTime(occurredTime);
+            if (messageMapper.insert(message) != 1 || message.getId() == null) {
+                throw new IllegalStateException("写入通知消息失败");
+            }
+            String skipReason = settingService.isUserPushEnabled(userId) ? null : AppMessageConstants.SKIP_USER_DISABLED;
+            deliveryService.createDelivery(message, skipReason);
+            messages.add(message);
+        }
+        return summarize(dto.getRequestId(), messages);
+    }
+
+    /**
+     * 查询当前租户下一个稳定事件的全部消息，供首次判断和并发冲突后回查。
+     *
+     * @param eventId 本次稳定事件ID
+     * @return 已提交的完整消息集合，未提交时为空
+     */
+    private List<ThirdAppMessageEntity> findBatch(String eventId) {
+        return messageMapper.selectList(Wrappers.<ThirdAppMessageEntity>lambdaQuery().eq(ThirdAppMessageEntity::getEventId, eventId));
+    }
+
+    /**
+     * 校验已存在批次与本次快照完全一致，禁止复用同一标识修改内容或追加接收人。
+     *
+     * @param dto 当前请求
+     * @param userIds 本次规范化的接收人ID
+     * @param payloadJson 本次请求指纹JSON
+     * @param messages 已完整提交的消息快照
+     * @return 与首次提交一致的入箱与投递汇总
+     */
+    private AppManualPushResultVo replayResult(AppManualPushDto dto, List<Long> userIds, String payloadJson, List<ThirdAppMessageEntity> messages) {
+        Set<Long> recipients = messages.stream().map(ThirdAppMessageEntity::getRecipientUserId).collect(Collectors.toSet());
+        if (messages.size() != userIds.size() || !recipients.equals(Set.copyOf(userIds))
+                || messages.stream().anyMatch(message -> !payloadJson.equals(message.getPayloadJson()))) {
+            throw new IllegalArgumentException("提交标识已用于其他通知内容，请重新发起发送");
+        }
+        return summarize(dto.getRequestId(), messages);
+    }
+
+    /**
+     * 从实际投递任务统计跳过人数，重放时已处理任务仍计为已加入投递。
+     *
+     * @param requestId 客户端幂等标识
+     * @param messages 已成功入箱的消息集合
+     * @return 不包含设备已送达承诺的汇总结果
+     */
+    private AppManualPushResultVo summarize(String requestId, List<ThirdAppMessageEntity> messages) {
+        List<Long> messageIds = messages.stream().map(ThirdAppMessageEntity::getId).toList();
+        long skipped = deliveryMapper.selectCount(Wrappers.<ThirdPushDeliveryEntity>lambdaQuery()
+                .in(ThirdPushDeliveryEntity::getMessageId, messageIds)
+                .eq(ThirdPushDeliveryEntity::getStatus, AppMessageConstants.DELIVERY_SKIPPED));
+        AppManualPushResultVo result = new AppManualPushResultVo();
+        result.setRequestId(requestId);
+        result.setRecipientCount(messages.size());
+        result.setSkippedCount((int) skipped);
+        result.setQueuedCount(messages.size() - (int) skipped);
+        return result;
+    }
+
+    /**
+     * 使用统一序列化器生成确定性的快照，失败时在写库前终止请求。
+     *
+     * @param value 接收人、正文或单字段指纹快照
+     * @return JSON字符串
+     */
+    private String serialize(Object value) {
+        try {
+            return objectMapper.writeValueAsString(value);
+        } catch (JsonProcessingException exception) {
+            throw new IllegalStateException("生成通知快照失败", exception);
+        }
+    }
+}
@@ -625,7 +625,8 @@ public class ThirdAppMessageServiceImpl implements ThirdAppMessageService {
        AppMessageVo vo = new AppMessageVo();
        BeanUtils.copyProperties(entity, vo);
        AppMessageEventEnum event = AppMessageEventEnum.fromCode(entity.getEventCode());
-        vo.setCategory(event == null ? null : event.getCategory());
+        vo.setCategory(AppMessageConstants.EVENT_MANUAL_PUSH.equals(entity.getEventCode())
+                ? AppMessageConstants.CATEGORY_SYSTEM : event == null ? null : event.getCategory());
        vo.setRead(AppMessageConstants.READ.equals(entity.getReadStatus()));
        vo.setTitle(resolveLocalizedTitle(entity, event));
        vo.setContent(resolveLocalizedContent(entity, event));
@@ -0,0 +1,248 @@
+package com.bsd.pjcloud.thirdparty.service.impl.appmessage;
+
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.bsd.pjcloud.admin.api.dto.UserDTO;
+import com.bsd.pjcloud.admin.api.feign.RemoteUserService;
+import com.bsd.pjcloud.admin.api.vo.UserVO;
+import com.bsd.pjcloud.common.core.util.R;
+import com.bsd.pjcloud.common.data.tenant.TenantContextHolder;
+import com.bsd.pjcloud.thirdparty.constant.appmessage.AppMessageConstants;
+import com.bsd.pjcloud.thirdparty.dto.appmessage.AppManualPushDto;
+import com.bsd.pjcloud.thirdparty.entity.appmessage.ThirdAppMessageEntity;
+import com.bsd.pjcloud.thirdparty.mapper.appmessage.ThirdAppMessageMapper;
+import com.bsd.pjcloud.thirdparty.mapper.appmessage.ThirdPushDeliveryMapper;
+import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdAppMessageSettingService;
+import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdPushDeliveryService;
+import com.bsd.pjcloud.thirdparty.vo.appmessage.AppManualPushResultVo;
+import com.bsd.pjcloud.wallet.api.feign.RemoteMchService;
+import com.bsd.pjcloud.wallet.api.vo.MchManageVo;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.validation.Validation;
+import jakarta.validation.ValidatorFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.transaction.support.SimpleTransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** 手动通知的接收人权限、入箱快照、跳过统计和提交幂等规则单元测试。 */
+class ThirdAppManualPushServiceImplTest {
+
+    /** 测试租户，验证入箱记录继承当前租户上下文。 */
+    private static final Long TENANT_ID = 10L;
+
+    /** 当前管理员ID，参与幂等事件隔离。 */
+    private static final Long OPERATOR_ID = 20L;
+
+    /** 已通过Controller商户管理权限校验的来源商户。 */
+    private static final Long MERCHANT_ID = 30L;
+
+    /** 权限分页客户端Mock，默认返回本批次全部可见用户。 */
+    private final RemoteUserService remoteUserService = mock(RemoteUserService.class);
+
+    /** 来源商户名称快照客户端Mock。 */
+    private final RemoteMchService remoteMchService = mock(RemoteMchService.class);
+
+    /** 消息Mapper Mock，使用内存集合观察是否重复写入。 */
+    private final ThirdAppMessageMapper messageMapper = mock(ThirdAppMessageMapper.class);
+
+    /** 投递统计Mapper Mock，具体测试按需模拟跳过人数。 */
+    private final ThirdPushDeliveryMapper deliveryMapper = mock(ThirdPushDeliveryMapper.class);
+
+    /** 用户Push总开关Mock，默认允许通知栏推送。 */
+    private final ThirdAppMessageSettingService settingService = mock(ThirdAppMessageSettingService.class);
+
+    /** 既有投递服务Mock，用于核对关闭开关时传入的跳过原因。 */
+    private final ThirdPushDeliveryService deliveryService = mock(ThirdPushDeliveryService.class);
+
+    /** 事务模板Mock，仅执行回调；真实数据库回滚和唯一约束仍需集成验证。 */
+    private final TransactionTemplate transactionTemplate = mock(TransactionTemplate.class);
+
+    /** 当前测试观察到的消息快照，模拟完整批次的已提交查询。 */
+    private final List<ThirdAppMessageEntity> messages = new ArrayList<>();
+
+    /** 设置当前租户和最小依赖响应，不启动Spring上下文或调用推送供应商。 */
+    @BeforeEach
+    void setUp() {
+        TenantContextHolder.setTenantId(TENANT_ID);
+        when(remoteUserService.getScopeBasicUserPageWithCountOption(eq(1L), eq(100L), eq(false), any(UserDTO.class)))
+                .thenAnswer(invocation -> visibleUsers(invocation.<UserDTO>getArgument(3).getUserIds()));
+        MchManageVo merchant = new MchManageVo();
+        merchant.setName("测试商户");
+        when(remoteMchService.getById(MERCHANT_ID)).thenReturn(R.ok(merchant));
+        when(messageMapper.selectList(any())).thenAnswer(invocation -> List.copyOf(messages));
+        when(messageMapper.insert(any(ThirdAppMessageEntity.class))).thenAnswer(invocation -> {
+            ThirdAppMessageEntity message = invocation.getArgument(0);
+            message.setId(1000L + messages.size());
+            messages.add(message);
+            return 1;
+        });
+        when(settingService.isUserPushEnabled(any(Long.class))).thenReturn(true);
+        when(deliveryMapper.selectCount(any())).thenReturn(0L);
+        when(transactionTemplate.execute(any())).thenAnswer(invocation -> {
+            TransactionCallback<?> callback = invocation.getArgument(0);
+            return callback.doInTransaction(new SimpleTransactionStatus());
+        });
+    }
+
+    /** 释放线程租户，避免影响同一进程的其他测试。 */
+    @AfterEach
+    void tearDown() {
+        TenantContextHolder.clear();
+    }
+
+    /** 去重接收人后为每人写入一条未读原文消息，且不会扩散为关系消息。 */
+    @Test
+    void sendsOneOwnerMessagePerDistinctVisibleUser() {
+        AppManualPushDto dto = request();
+        dto.setUserIds(List.of(2L, 1L, 1L));
+
+        AppManualPushResultVo result = service().send(dto, OPERATOR_ID);
+
+        assertEquals(2, result.getRecipientCount());
+        assertEquals(2, result.getQueuedCount());
+        assertEquals(0, result.getSkippedCount());
+        assertEquals(List.of(1L, 2L), messages.stream().map(ThirdAppMessageEntity::getRecipientUserId).toList());
+        messages.forEach(message -> {
+            assertEquals(TENANT_ID, message.getTenantId());
+            assertEquals(MERCHANT_ID, message.getSourceMchId());
+            assertEquals(AppMessageConstants.EVENT_MANUAL_PUSH, message.getEventCode());
+            assertEquals(AppMessageConstants.RECIPIENT_OWNER, message.getRecipientType());
+            assertEquals(message.getRecipientUserId(), message.getSubjectUserId());
+            assertEquals(AppMessageConstants.UNREAD, message.getReadStatus());
+            assertEquals("第一行\n第二行", message.getContent());
+            assertNull(message.getRelationId());
+        });
+        verify(deliveryService, times(2)).createDelivery(any(ThirdAppMessageEntity.class), eq(null));
+    }
+
+    /** 任一用户不在当前数据权限内时，在写库之前整批拒绝。 */
+    @Test
+    void rejectsEntireBatchWhenOneRecipientIsNotVisible() {
+        when(remoteUserService.getScopeBasicUserPageWithCountOption(eq(1L), eq(100L), eq(false), any(UserDTO.class)))
+                .thenReturn(visibleUsers(List.of(1L)));
+
+        assertThrows(IllegalArgumentException.class, () -> service().send(request(), OPERATOR_ID));
+
+        verify(messageMapper, never()).insert(any(ThirdAppMessageEntity.class));
+        verify(transactionTemplate, never()).execute(any());
+    }
+
+    /** 用户关闭Push总开关时仍然入箱，统计结果以实际投递跳过任务为准。 */
+    @Test
+    void keepsInboxWhenUserDisabledPushOrHasNoDevice() {
+        when(settingService.isUserPushEnabled(1L)).thenReturn(false);
+        when(deliveryMapper.selectCount(any())).thenReturn(2L);
+
+        AppManualPushResultVo result = service().send(request(), OPERATOR_ID);
+
+        assertEquals(2, messages.size());
+        assertEquals(0, result.getQueuedCount());
+        assertEquals(2, result.getSkippedCount());
+        verify(deliveryService).createDelivery(eq(messages.get(0)), eq(AppMessageConstants.SKIP_USER_DISABLED));
+        verify(deliveryService).createDelivery(eq(messages.get(1)), eq(null));
+    }
+
+    /** 相同提交标识和快照重试时仅返回既有结果，不重复入箱或创建投递任务。 */
+    @Test
+    void replaysSameRequestWithoutDuplicateMessages() {
+        ThirdAppManualPushServiceImpl service = service();
+        service.send(request(), OPERATOR_ID);
+
+        AppManualPushResultVo replay = service.send(request(), OPERATOR_ID);
+
+        assertEquals(2, replay.getRecipientCount());
+        assertEquals(2, messages.size());
+        verify(messageMapper, times(2)).insert(any(ThirdAppMessageEntity.class));
+        verify(transactionTemplate, times(1)).execute(any());
+    }
+
+    /** 同一提交标识不能用于修改正文，否则可能被误当成已成功发送。 */
+    @Test
+    void rejectsChangedContentWithSameRequestId() {
+        ThirdAppManualPushServiceImpl service = service();
+        service.send(request(), OPERATOR_ID);
+        AppManualPushDto changed = request();
+        changed.setContent("另一条通知");
+
+        assertThrows(IllegalArgumentException.class, () -> service.send(changed, OPERATOR_ID));
+        assertEquals(2, messages.size());
+    }
+
+    /** 模拟并发事务已提交后的唯一键冲突，回滚退出后必须回读完整结果。 */
+    @Test
+    void readsCommittedResultAfterConcurrentDuplicateKey() {
+        ThirdAppManualPushServiceImpl service = service();
+        service.send(request(), OPERATOR_ID);
+        List<ThirdAppMessageEntity> committed = List.copyOf(messages);
+        when(messageMapper.selectList(any())).thenReturn(List.of(), committed);
+        when(transactionTemplate.execute(any())).thenThrow(new DuplicateKeyException("concurrent request"));
+
+        AppManualPushResultVo replay = service.send(request(), OPERATOR_ID);
+
+        assertEquals(2, replay.getRecipientCount());
+        assertEquals(2, messages.size());
+    }
+
+    /** API参数校验拒绝超过500人、纯空白标题和超过现有列宽的正文。 */
+    @Test
+    void validatesRecipientLimitAndPlainTextBounds() {
+        AppManualPushDto dto = request();
+        dto.setUserIds(LongStream.rangeClosed(1, 501).boxed().toList());
+        dto.setTitle(" ");
+        dto.setContent("字".repeat(2001));
+        try (ValidatorFactory factory = Validation.buildDefaultValidatorFactory()) {
+            Set<String> fields = factory.getValidator().validate(dto).stream()
+                    .map(violation -> violation.getPropertyPath().toString()).collect(Collectors.toSet());
+            assertTrue(fields.containsAll(Set.of("userIds", "title", "content")));
+        }
+    }
+
+    /** 创建独立服务实例，事务Mock不代替真实数据库集成验证。 */
+    private ThirdAppManualPushServiceImpl service() {
+        return new ThirdAppManualPushServiceImpl(remoteUserService, remoteMchService, messageMapper, deliveryMapper,
+                settingService, deliveryService, new ObjectMapper(), transactionTemplate);
+    }
+
+    /** 构造具备稳定标识、两位用户及多行正文的有效提交快照。 */
+    private AppManualPushDto request() {
+        AppManualPushDto dto = new AppManualPushDto();
+        dto.setMchId(MERCHANT_ID);
+        dto.setUserIds(List.of(1L, 2L));
+        dto.setTitle("统一通知");
+        dto.setContent("第一行\n第二行");
+        dto.setRequestId("manual_push_request_001");
+        return dto;
+    }
+
+    /** 构造当前权限分页响应，仅返回本测试指定的可见用户ID。 */
+    private R<Page<UserVO>> visibleUsers(List<Long> ids) {
+        Page<UserVO> page = new Page<>(1, 100);
+        page.setRecords(ids.stream().map(id -> {
+            UserVO user = new UserVO();
+            user.setUserId(id);
+            return user;
+        }).toList());
+        return R.ok(page);
+    }
+}