feat:聊天后台撤回消息

fix:聊天禁用bug
This commit is contained in:
dzc committed 2026-11-03 19:08:46 +08:00
1 parent 00a0ab327c
commit 73d5a13762
33 files changed
+3101 -55

No files matched your search

@@ -0,0 +1,18 @@
-- MySQL 5.7 / 8.0; run against the IM database during an appropriate maintenance window.
-- Supports bilateral message history and candidate scans. The all-conversation
-- grouping still scans messages; review EXPLAIN with representative data.
-- This script is idempotent and does not change message rows.
SET @private_conversation_index_exists = (
SELECT COUNT(*) FROM information_schema.statistics
WHERE table_schema = DATABASE()
AND table_name = 'im_private_message'
AND index_name = 'idx_private_pair_time'
);
SET @private_conversation_index_sql = IF(
@private_conversation_index_exists = 0,
'ALTER TABLE im_private_message ADD INDEX idx_private_pair_time (send_id, recv_id, send_time, id), ALGORITHM=INPLACE, LOCK=NONE',
'SELECT ''idx_private_pair_time already exists'' AS message'
);
PREPARE private_conversation_index_stmt FROM @private_conversation_index_sql;
EXECUTE private_conversation_index_stmt;
DEALLOCATE PREPARE private_conversation_index_stmt;
+3 -2
View File
@@ -39,7 +39,8 @@ create table `im_private_message`(
`status` tinyint(1) NOT NULL comment '状态 0:未读 1:已读 2:撤回 3:已读',
`send_time` datetime DEFAULT CURRENT_TIMESTAMP comment '发送时间',
key `idx_send_id` (`send_id`),
key `idx_recv_id` (`recv_id`)
key `idx_recv_id` (`recv_id`),
key `idx_private_pair_time` (`send_id`, `recv_id`, `send_time`, `id`)
)ENGINE=InnoDB CHARSET=utf8mb4 comment '私聊消息';
@@ -109,4 +110,4 @@ CREATE TABLE `im_file_info` (
`is_permanent` tinyint DEFAULT 0 comment '是否永久文件',
`md5` VARCHAR(64) NOT NULL comment '文件md5',
KEY `idx_md5` (md5)
) ENGINE = InnoDB CHARSET = utf8mb4 comment '文件';
) ENGINE = InnoDB CHARSET = utf8mb4 comment '文件';
+6 -1
View File
@@ -96,6 +96,11 @@
<artifactId>knife4j-openapi3-jakarta-spring-boot-starter</artifactId>
<version>${knife4j.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>${project.artifactId}</finalName>
@@ -113,4 +118,4 @@
</plugin>
</plugins>
</build>
</project>
</project>
@@ -0,0 +1,27 @@
package com.bx.implatform.controller.third;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageModerateDTO;
import com.bx.implatform.result.Result;
import com.bx.implatform.result.ResultUtils;
import com.bx.implatform.service.PrivateMessageModerationService;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageModerateVO;
import io.swagger.v3.oas.annotations.Operation;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
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;
@RestController
@RequestMapping("/third/privateMsg")
@RequiredArgsConstructor
public class ThirdPrivateMessageModerationController {
private final PrivateMessageModerationService moderationService;
@PostMapping("/moderate")
@Operation(summary = "管理员屏蔽私聊消息,并通知双方隐藏原内容")
public Result<ThirdPrivateMessageModerateVO> moderate(@Valid @RequestBody ThirdPrivateMessageModerateDTO dto) {
return ResultUtils.success(moderationService.moderate(dto));
}
}
@@ -1,6 +1,7 @@
package com.bx.implatform.controller.third;
import com.bx.implatform.dto.MessageTotalDto;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateConversationDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageCountDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageInfoDTO;
@@ -11,6 +12,7 @@ import com.bx.implatform.service.third.ThirdPrivateMsgService;
import com.bx.implatform.vo.MessageTotalVo;
import com.bx.implatform.vo.PrivateMessageVO;
import com.bx.implatform.vo.third.ThirdTableVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageVO;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
@@ -34,6 +36,13 @@ public class ThirdPrivateMsgController {
private final ThirdPrivateMsgService thirdPrivateMsgService;
@PostMapping("/conversation/page")
@Operation(summary = "查询双人私聊会话列表")
public Result<ThirdTableVO<ThirdPrivateConversationVO>> conversationPage(
@Valid @RequestBody ThirdPrivateConversationDTO dto) {
return ResultUtils.success(thirdPrivateMsgService.conversationPage(dto));
}
@PostMapping("/send")
@Operation(summary = "发送私聊消息", description = "发送私聊消息")
public Result<PrivateMessageVO> sendMessage(@Valid @RequestBody ThirdPrivateMessageSendDTO vo) {
@@ -15,6 +15,7 @@ import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import java.util.List;
@@ -41,9 +42,9 @@ public class ThirdUserController {
return ResultUtils.success(thirdUserService.getUserByUserId(dto));
}
@PostMapping("/createOrUpdate")
@PostMapping(value = "/createOrUpdate", produces = MediaType.APPLICATION_JSON_VALUE)
@Operation(summary = "创建或者更新用户", description = "创建或者更新用户")
public Result createOrUpdate(@Valid @RequestBody ThirdCreateOrUpdateUserDTO dto) {
public Result<Void> createOrUpdate(@Valid @RequestBody ThirdCreateOrUpdateUserDTO dto) {
thirdUserService.createOrUpdate(dto);
return ResultUtils.success();
}
@@ -0,0 +1,36 @@
package com.bx.implatform.dto.third.privatemsg;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.Max;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotNull;
import jakarta.validation.constraints.Size;
import lombok.Data;
import java.util.Date;
import java.util.List;
@Data
@Schema(description = "第三方私聊会话分页查询")
public class ThirdPrivateConversationDTO {
@NotNull
@Min(1)
private Integer pageNum = 1;
@NotNull
@Min(1)
@Max(100)
private Integer pageSize = 10;
@Size(max = 100)
@Schema(description = "任意一方的昵称、账号或完整用户ID")
private String userKeyword;
@Size(max = 100)
@Schema(description = "另一方的昵称、账号或完整用户ID")
private String peerKeyword;
@Schema(description = "最近聊天时间范围,包含两端,毫秒时间戳数组")
private List<Date> lastSendTimeList;
}
@@ -0,0 +1,20 @@
package com.bx.implatform.dto.third.privatemsg;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotNull;
import jakarta.validation.constraints.Size;
import lombok.Data;
@Data
public class ThirdPrivateMessageModerateDTO {
@NotNull
@Min(1)
private Long messageId;
@NotNull
@Min(1)
private Long operatorId;
@Size(max = 200)
private String reason;
}
@@ -1,7 +1,7 @@
package com.bx.implatform.dto.third.user;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
import org.hibernate.validator.constraints.Length;
@@ -15,11 +15,11 @@ public class ThirdCreateOrUpdateUserDTO {
private Long userId;
@Length(max = 20, message = "用户名不能大于20字符")
@NotEmpty(message = "用户名不可为空")
@NotBlank(message = "用户名不可为空")
@Schema(description = "用户名")
private String userName;
@NotEmpty(message = "用户手机号不可为空")
@NotBlank(message = "用户手机号不可为空")
@Schema(description = "用户手机号")
private String phone;
@@ -9,6 +9,8 @@ import com.bx.imcommon.enums.IMSendCode;
import com.bx.imcommon.model.IMSendResult;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.enums.MessageStatus;
import com.bx.implatform.enums.MessageType;
import com.bx.implatform.service.PrivateMessageModerationService;
import com.bx.implatform.service.PrivateMessageService;
import com.bx.implatform.vo.PrivateMessageVO;
import lombok.extern.slf4j.Slf4j;
@@ -27,17 +29,23 @@ public class PrivateMessageListener implements MessageListener<PrivateMessageVO>
@Autowired
private PrivateMessageService privateMessageService;
@Lazy
@Autowired
private PrivateMessageModerationService moderationService;
@Override
public void process(List<IMSendResult<PrivateMessageVO>> results) {
Set<Long> messageIds = new HashSet<>();
for (IMSendResult<PrivateMessageVO> result : results) {
if (result == null || result.getData() == null || result.getReceiver() == null) continue;
PrivateMessageVO messageInfo = result.getData();
// 更新消息状态,这里只处理成功消息,失败的消息继续保持未读状态
if (result.getCode().equals(IMSendCode.SUCCESS.code()) && !Objects.isNull(messageInfo.getId())) {
if (result.getReceiver().getId().equals(messageInfo.getRecvId())) {
if (IMSendCode.SUCCESS.code().equals(result.getCode()) && messageInfo.getId() != null
&& messageInfo.getId() > 0 && !MessageType.RECALL.code().equals(messageInfo.getType())) {
if (Objects.equals(result.getReceiver().getId(), messageInfo.getRecvId())) {
messageIds.add(messageInfo.getId());
log.info("消息送达,消息id:{},发送者:{},接收者:{},终端:{}", messageInfo.getId(),
result.getSender().getId(), result.getReceiver().getId(), result.getReceiver().getTerminal());
messageInfo.getSendId(), result.getReceiver().getId(), result.getReceiver().getTerminal());
}
}
}
@@ -48,6 +56,9 @@ public class PrivateMessageListener implements MessageListener<PrivateMessageVO>
.eq(PrivateMessage::getStatus, MessageStatus.PENDING.code())
.set(PrivateMessage::getStatus, MessageStatus.DELIVERED.code());
privateMessageService.update(updateWrapper);
// The source message could have been queued before moderation and delivered
// afterwards. Re-send the persisted tombstone without requesting a callback.
moderationService.replayIfModerated(messageIds);
}
}
}
@@ -2,8 +2,10 @@ package com.bx.implatform.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.bx.implatform.dto.MessageTotalDto;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateConversationDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.vo.MessageTotalVo;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
@@ -12,6 +14,19 @@ import java.util.Map;
public interface PrivateMessageMapper extends BaseMapper<PrivateMessage> {
@Select("SELECT * FROM im_private_message WHERE id = #{id} FOR UPDATE")
PrivateMessage selectForUpdate(@Param("id") Long id);
@Select("SELECT * FROM im_private_message WHERE send_id = #{sendId} AND recv_id = #{recvId} "
+ "AND type = 10 AND BINARY tmp_id = #{marker} ORDER BY id LIMIT 1")
PrivateMessage selectModerationEvent(@Param("marker") String marker, @Param("sendId") Long sendId,
@Param("recvId") Long recvId);
long countConversations(@Param("dto") ThirdPrivateConversationDTO dto);
List<ThirdPrivateConversationVO> selectConversations(@Param("dto") ThirdPrivateConversationDTO dto,
@Param("offset") long offset);
@Select("SELECT DATE(send_time) as date, COUNT(*) as count " +
"FROM im_private_message " +
"WHERE send_time >= DATE_SUB(CURDATE(), INTERVAL #{days} DAY) " +
@@ -0,0 +1,198 @@
package com.bx.implatform.service;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.bx.imclient.IMClient;
import com.bx.imcommon.enums.IMTerminalType;
import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageModerateDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.enums.MessageStatus;
import com.bx.implatform.enums.MessageType;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.util.BeanUtils;
import com.bx.implatform.vo.PrivateMessageVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageModerateVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;
import java.util.*;
import java.util.stream.Collectors;
/** Persists moderation before notifying clients using their existing recall protocol. */
@Slf4j
@Service
public class PrivateMessageModerationService {
public static final String ADMIN_MARKER_PREFIX = "admin-m:";
private final PrivateMessageMapper mapper;
private final IMClient imClient;
private final TransactionTemplate transactionTemplate;
public PrivateMessageModerationService(PrivateMessageMapper mapper, IMClient imClient,
TransactionTemplate transactionTemplate) {
this.mapper = mapper;
this.imClient = imClient;
this.transactionTemplate = new TransactionTemplate(transactionTemplate.getTransactionManager());
// Ensure execute() has really committed before the notification is submitted,
// even if a future caller has an outer transaction.
this.transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
}
public ThirdPrivateMessageModerateVO moderate(ThirdPrivateMessageModerateDTO dto) {
if (dto == null || dto.getMessageId() == null || dto.getMessageId() <= 0
|| dto.getOperatorId() == null || dto.getOperatorId() <= 0
|| (dto.getReason() != null && dto.getReason().length() > 200)) {
throw new GlobalException("消息编号、操作人或屏蔽原因不正确");
}
ModerationChange change = transactionTemplate.execute(transaction -> {
PrivateMessage original = mapper.selectForUpdate(dto.getMessageId());
if (original == null) throw new GlobalException("消息不存在");
if (!isContentMessage(original.getType())) throw new GlobalException("状态或控制消息不能屏蔽");
boolean alreadyHidden = MessageStatus.RECALL.code().equals(original.getStatus());
if (!alreadyHidden) {
PrivateMessage update = new PrivateMessage();
update.setId(original.getId());
update.setStatus(MessageStatus.RECALL.code());
if (mapper.updateById(update) != 1) throw new GlobalException("消息状态更新失败");
}
String marker = ADMIN_MARKER_PREFIX + original.getId();
PrivateMessage event = mapper.selectModerationEvent(marker, original.getSendId(), original.getRecvId());
if (event == null) {
event = new PrivateMessage();
event.setTmpId(marker);
event.setSendId(original.getSendId());
event.setRecvId(original.getRecvId());
event.setContent(original.getId().toString());
event.setType(MessageType.RECALL.code());
event.setStatus(MessageStatus.READED.code());
event.setSendTime(new Date());
if (mapper.insert(event) != 1) throw new GlobalException("撤回同步记录保存失败");
} else if (!isAdminModerationEvent(event) || !original.getId().toString().equals(event.getContent())) {
throw new GlobalException("撤回同步记录不正确");
}
return new ModerationChange(original.getId(), event, alreadyHidden);
});
if (change == null) throw new GlobalException("消息屏蔽失败");
boolean notificationSent = notifyParticipants(change.event());
String reason = dto.getReason() == null ? "" : dto.getReason().replaceAll("[\\r\\n\\t]", " ");
log.info("管理员屏蔽私聊消息,messageId={},operatorId={},reason={},alreadyHidden={},recallMessageId={},notificationSent={}",
change.messageId(), dto.getOperatorId(), reason, change.alreadyHidden(), change.event().getId(), notificationSent);
ThirdPrivateMessageModerateVO result = new ThirdPrivateMessageModerateVO();
result.setMessageId(change.messageId());
result.setStatus(MessageStatus.RECALL.code());
result.setRecallMessageId(change.event().getId());
result.setAlreadyHidden(change.alreadyHidden());
result.setNotificationSent(notificationSent);
return result;
}
/** A delayed delivery acknowledgement may arrive after moderation. Reassert the tombstone. */
public void replayIfModerated(Long sourceMessageId) {
replayIfModerated(Collections.singletonList(sourceMessageId));
}
/** The normal delivery path needs one narrow batch query, never one body lookup per message. */
public void replayIfModerated(Collection<Long> sourceMessageIds) {
if (sourceMessageIds == null || sourceMessageIds.isEmpty()) return;
List<Long> ids = sourceMessageIds.stream().filter(id -> id != null && id > 0).distinct().collect(Collectors.toList());
for (int start = 0; start < ids.size(); start += 500) {
List<Long> batch = ids.subList(start, Math.min(start + 500, ids.size()));
try {
List<PrivateMessage> sources = mapper.selectList(Wrappers.<PrivateMessage>lambdaQuery()
.select(PrivateMessage::getId, PrivateMessage::getSendId, PrivateMessage::getRecvId,
PrivateMessage::getType, PrivateMessage::getStatus)
.in(PrivateMessage::getId, batch).eq(PrivateMessage::getStatus, MessageStatus.RECALL.code()));
for (PrivateMessage source : sources) {
if (!isContentMessage(source.getType())) continue;
PrivateMessage event = mapper.selectModerationEvent(ADMIN_MARKER_PREFIX + source.getId(),
source.getSendId(), source.getRecvId());
if (isAdminModerationEvent(event) && source.getId().toString().equals(event.getContent())) {
notifyParticipants(event);
}
}
} catch (Exception e) {
log.warn("管理员撤回补偿失败,batchSize={},exceptionType={}", batch.size(), e.getClass().getSimpleName());
}
}
}
private boolean notifyParticipants(PrivateMessage event) {
boolean submitted = true;
for (Long participant : new LinkedHashSet<>(Arrays.asList(event.getSendId(), event.getRecvId()))) {
IMPrivateMessage<PrivateMessageVO> push = new IMPrivateMessage<>();
push.setSender(new IMUserInfo(event.getSendId(), IMTerminalType.UNKNOW.code()));
push.setRecvId(participant);
push.setRecvTerminals(IMTerminalType.codes());
push.setSendToSelf(false);
push.setSendResult(false);
push.setData(toMobileVO(event));
try {
imClient.sendPrivateMessage(push);
} catch (Exception e) {
submitted = false;
log.warn("管理员撤回通知提交失败,eventId={},participantId={},exceptionType={}",
event.getId(), participant, e.getClass().getSimpleName());
}
}
return submitted;
}
public PrivateMessageVO toMobileVO(PrivateMessage message) {
PrivateMessageVO vo = BeanUtils.copyProperties(message, PrivateMessageVO.class);
if (MessageType.RECALL.code().equals(message.getType())) {
Long originalId = recallTargetId(message.getContent());
// Old uni-app uses content to find the message and id to find its quotes.
// The database event id is retained in storage/query ordering only.
if (originalId != null) vo.setId(originalId);
if (isAdminModerationEvent(message)) vo.setStatus(MessageStatus.READED.code());
}
return vo;
}
/** Re-read after pending/delivered updates so a stale snapshot cannot re-expose hidden content. */
public List<PrivateMessage> refreshVisibleMessages(List<PrivateMessage> messages) {
if (messages == null || messages.isEmpty()) return Collections.emptyList();
List<Long> ids = messages.stream().map(PrivateMessage::getId).filter(Objects::nonNull).distinct().collect(Collectors.toList());
Map<Long, PrivateMessage> current = new HashMap<>();
for (int start = 0; start < ids.size(); start += 500) {
List<PrivateMessage> rows = mapper.selectBatchIds(ids.subList(start, Math.min(start + 500, ids.size())));
for (PrivateMessage row : rows) current.put(row.getId(), row);
}
return messages.stream().map(message -> current.get(message.getId())).filter(Objects::nonNull)
.filter(message -> !MessageStatus.RECALL.code().equals(message.getStatus())).collect(Collectors.toList());
}
public static boolean isAdminModerationEvent(PrivateMessage message) {
return message != null && MessageType.RECALL.code().equals(message.getType())
&& recallTargetId(message.getContent()) != null
&& (ADMIN_MARKER_PREFIX + message.getContent()).equals(message.getTmpId());
}
public static void rejectReservedTmpId(String tmpId) {
if (tmpId != null && tmpId.toLowerCase(Locale.ROOT).startsWith(ADMIN_MARKER_PREFIX)) {
throw new GlobalException("临时消息编号使用了保留前缀");
}
}
private static Long recallTargetId(String content) {
try {
long id = Long.parseLong(content);
return id > 0 ? id : null;
} catch (RuntimeException e) {
return null;
}
}
private static boolean isContentMessage(Integer type) {
return type != null && ((type >= 0 && type <= 9)
|| MessageType.ACT_RT_VOICE.code().equals(type) || MessageType.ACT_RT_VIDEO.code().equals(type));
}
private record ModerationChange(Long messageId, PrivateMessage event, boolean alreadyHidden) { }
}
@@ -19,9 +19,11 @@ import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.service.FriendService;
import com.bx.implatform.service.PrivateMessageService;
import com.bx.implatform.service.PrivateMessageModerationService;
import com.bx.implatform.session.SessionContext;
import com.bx.implatform.session.UserSession;
import com.bx.implatform.util.BeanUtils;
import com.bx.implatform.util.PrivateMessageSessionControl;
import com.bx.implatform.util.SensitiveFilterUtil;
import com.bx.implatform.vo.PrivateMessageVO;
import lombok.RequiredArgsConstructor;
@@ -46,10 +48,13 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
private final FriendService friendService;
private final IMClient imClient;
private final SensitiveFilterUtil sensitiveFilterUtil;
private final PrivateMessageModerationService moderationService;
private static final ScheduledThreadPoolExecutor EXECUTOR = ThreadPoolExecutorFactory.getThreadPoolExecutor();
@Override
public PrivateMessageVO sendMessage(PrivateMessageDTO dto) {
PrivateMessageModerationService.rejectReservedTmpId(dto.getTmpId());
PrivateMessageSessionControl.rejectReservedKickContent(dto.getContent());
UserSession session = SessionContext.getSession();
Boolean isFriends = friendService.isFriend(session.getUserId(), dto.getRecvId());
if (Boolean.FALSE.equals(isFriends)) {
@@ -86,6 +91,9 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
if (Objects.isNull(msg)) {
throw new GlobalException("消息不存在");
}
if (PrivateMessageModerationService.isAdminModerationEvent(msg)) {
throw new GlobalException("管理员处理记录不能撤回");
}
if (!msg.getSendId().equals(session.getUserId())) {
throw new GlobalException("这条消息不是由您发送,无法撤回");
}
@@ -105,11 +113,13 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
recallMsg.setContent(id.toString());
this.save(recallMsg);
// 推送消息
PrivateMessageVO msgInfo = BeanUtils.copyProperties(recallMsg, PrivateMessageVO.class);
PrivateMessageVO msgInfo = moderationService.toMobileVO(recallMsg);
IMPrivateMessage<PrivateMessageVO> sendMessage = new IMPrivateMessage<>();
sendMessage.setSender(new IMUserInfo(session.getUserId(), session.getTerminal()));
sendMessage.setRecvId(msgInfo.getRecvId());
sendMessage.setData(msgInfo);
// Recall VO ids point to the original message for old client compatibility.
sendMessage.setSendResult(false);
imClient.sendPrivateMessage(sendMessage);
log.info("撤回私聊消息,发送id:{},接收id:{},内容:{}", msg.getSendId(), msg.getRecvId(), msg.getContent());
return msgInfo;
@@ -122,6 +132,7 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
Long userId = SessionContext.getSession().getUserId();
long stIdx = (page - 1) * size;
QueryWrapper<PrivateMessage> wrapper = new QueryWrapper<>();
PrivateMessageSessionControl.applyHistoryFilter(wrapper);
wrapper.lambda().and(
wrap -> wrap.and(wp -> wp.eq(PrivateMessage::getSendId, userId).eq(PrivateMessage::getRecvId, friendId))
.or(wp -> wp.eq(PrivateMessage::getRecvId, userId).eq(PrivateMessage::getSendId, friendId)))
@@ -130,7 +141,9 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
List<PrivateMessage> messages = this.list(wrapper);
List<PrivateMessageVO> messageInfos =
messages.stream().map(m -> BeanUtils.copyProperties(m, PrivateMessageVO.class))
moderationService.refreshVisibleMessages(messages).stream()
.filter(m -> !PrivateMessageSessionControl.isKickMessage(m))
.map(moderationService::toMobileVO)
.collect(Collectors.toList());
log.info("拉取聊天记录,用户id:{},好友id:{},数量:{}", userId, friendId, messageInfos.size());
return messageInfos;
@@ -139,34 +152,29 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
@Override
public void pullOfflineMessage(Long minId) {
UserSession session = SessionContext.getSession();
// 获取当前用户的消息
LambdaQueryWrapper<PrivateMessage> wrapper = Wrappers.lambdaQuery();
// 只能拉取最近1个月的消息
Date minDate = DateUtils.addMonths(new Date(), -1);
wrapper.gt(PrivateMessage::getId, minId);
wrapper.ge(PrivateMessage::getSendTime, minDate);
wrapper.and(wp -> wp.eq(PrivateMessage::getSendId, session.getUserId()).or()
.eq(PrivateMessage::getRecvId, session.getUserId()));
wrapper.orderByAsc(PrivateMessage::getId);
List<PrivateMessage> messages = this.list(wrapper);
List<PrivateMessage> messages = this.list(buildOfflineQuery(session.getUserId(), minId));
// 异步推送消息
EXECUTOR.execute(() -> {
// 开启加载中标志
this.sendLoadingMessage(true, session);
for (PrivateMessage m : messages) {
for (PrivateMessage m : moderationService.refreshVisibleMessages(messages)) {
// Historical disable notices must not terminate a new login session.
if (PrivateMessageSessionControl.isKickMessage(m)) {
continue;
}
// 推送过程如果用户下线了,则不再推送
if (!imClient.isOnline(session.getUserId(), IMTerminalType.fromCode(session.getTerminal()))) {
log.info("用户已下线,停止推送离线私聊消息,用户id:{}", session.getUserId());
return;
}
PrivateMessageVO vo = BeanUtils.copyProperties(m, PrivateMessageVO.class);
PrivateMessageVO vo = moderationService.toMobileVO(m);
IMPrivateMessage<PrivateMessageVO> sendMessage = new IMPrivateMessage<>();
sendMessage.setSender(new IMUserInfo(m.getSendId(), IMTerminalType.WEB.code()));
sendMessage.setRecvId(session.getUserId());
sendMessage.setRecvTerminals(List.of(session.getTerminal()));
sendMessage.setSendToSelf(false);
sendMessage.setData(vo);
sendMessage.setSendResult(true);
sendMessage.setSendResult(!MessageType.RECALL.code().equals(m.getType()));
imClient.sendPrivateMessage(sendMessage);
}
// 关闭加载中标志
@@ -178,16 +186,7 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
@Override
public List<PrivateMessageVO> loadOfflineMessage(Long minId) {
UserSession session = SessionContext.getSession();
// 获取当前用户的消息
LambdaQueryWrapper<PrivateMessage> wrapper = Wrappers.lambdaQuery();
// 只能拉取最近1个月的消息
Date minDate = DateUtils.addMonths(new Date(), -1);
wrapper.gt(PrivateMessage::getId, minId);
wrapper.ge(PrivateMessage::getSendTime, minDate);
wrapper.and(wp -> wp.eq(PrivateMessage::getSendId, session.getUserId()).or()
.eq(PrivateMessage::getRecvId, session.getUserId()));
wrapper.orderByAsc(PrivateMessage::getId);
List<PrivateMessage> messages = this.list(wrapper);
List<PrivateMessage> messages = this.list(buildOfflineQuery(session.getUserId(), minId));
// 更新消息为送达状态
List<Long> messageIds = messages.stream().filter(m -> m.getRecvId().equals(session.getUserId()))
.filter(m -> m.getStatus().equals(MessageStatus.PENDING.code())).map(PrivateMessage::getId)
@@ -195,16 +194,37 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
if (!messageIds.isEmpty()) {
LambdaUpdateWrapper<PrivateMessage> updateWrapper = Wrappers.lambdaUpdate();
updateWrapper.in(PrivateMessage::getId, messageIds);
updateWrapper.eq(PrivateMessage::getStatus, MessageStatus.PENDING.code());
updateWrapper.set(PrivateMessage::getStatus, MessageStatus.DELIVERED.code());
update(updateWrapper);
}
// 转换vo
List<PrivateMessageVO> vos = messages.stream().map(m -> BeanUtils.copyProperties(m, PrivateMessageVO.class))
List<PrivateMessageVO> vos = moderationService.refreshVisibleMessages(messages).stream()
.filter(m -> !PrivateMessageSessionControl.isKickMessage(m))
.map(moderationService::toMobileVO)
.collect(Collectors.toList());
log.info("拉取私聊消息,用户id:{},数量:{},minId:{}", session.getUserId(), messages.size(), minId);
return vos;
}
private LambdaQueryWrapper<PrivateMessage> buildOfflineQuery(Long userId, Long minId) {
Date minDate = DateUtils.addMonths(new Date(), -1);
LambdaQueryWrapper<PrivateMessage> wrapper = Wrappers.lambdaQuery();
wrapper.and(pair -> pair.eq(PrivateMessage::getSendId, userId).or().eq(PrivateMessage::getRecvId, userId));
wrapper.ne(PrivateMessage::getStatus, MessageStatus.RECALL.code());
PrivateMessageSessionControl.applyHistoryFilter(wrapper);
wrapper.and(eligible -> eligible
.and(recent -> recent.gt(PrivateMessage::getId, minId == null ? 0L : minId)
.ge(PrivateMessage::getSendTime, minDate))
// Older clients do not acknowledge recall events. Replay administrator
// tombstones independently of the cursor and one-month message window.
.or(admin -> admin.eq(PrivateMessage::getType, MessageType.RECALL.code())
.eq(PrivateMessage::getStatus, MessageStatus.READED.code())
.apply("BINARY tmp_id = CONCAT({0}, content)", PrivateMessageModerationService.ADMIN_MARKER_PREFIX)));
wrapper.orderByAsc(PrivateMessage::getId);
return wrapper;
}
@Transactional(rollbackFor = Exception.class)
@Override
public void readedMessage(Long friendId) {
@@ -1,6 +1,7 @@
package com.bx.implatform.service.third;
import com.bx.implatform.dto.MessageTotalDto;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateConversationDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageCountDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageInfoDTO;
@@ -8,6 +9,7 @@ import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageSendDTO;
import com.bx.implatform.vo.MessageTotalVo;
import com.bx.implatform.vo.PrivateMessageVO;
import com.bx.implatform.vo.third.ThirdTableVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageVO;
import jakarta.validation.Valid;
@@ -16,6 +18,8 @@ import java.util.Map;
public interface ThirdPrivateMsgService {
ThirdTableVO<ThirdPrivateConversationVO> conversationPage(@Valid ThirdPrivateConversationDTO dto);
/**
* 发送私聊消息
*/
@@ -5,6 +5,7 @@ import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.IterUtil;
import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -14,6 +15,7 @@ import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.implatform.contant.Constant;
import com.bx.implatform.dto.MessageTotalDto;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateConversationDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageCountDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageInfoDTO;
@@ -26,18 +28,22 @@ import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.mapper.UserMapper;
import com.bx.implatform.service.FriendService;
import com.bx.implatform.service.PrivateMessageModerationService;
import com.bx.implatform.service.third.ThirdPrivateMsgService;
import com.bx.implatform.session.SessionContext;
import com.bx.implatform.session.UserSession;
import com.bx.implatform.util.BeanUtils;
import com.bx.implatform.util.PrivateMessageSessionControl;
import com.bx.implatform.util.SensitiveFilterUtil;
import com.bx.implatform.vo.MessageTotalVo;
import com.bx.implatform.vo.PrivateMessageVO;
import com.bx.implatform.vo.third.ThirdTableVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageVO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.*;
@@ -56,11 +62,32 @@ public class ThirdPrivateMsgServiceImpl extends ServiceImpl<PrivateMessageMapper
private final FriendService friendService;
private final SensitiveFilterUtil sensitiveFilterUtil;
@Override
@Transactional(readOnly = true)
public ThirdTableVO<ThirdPrivateConversationVO> conversationPage(ThirdPrivateConversationDTO dto) {
validatePagination(dto.getPageNum(), dto.getPageSize());
validateTimeRange(dto.getLastSendTimeList());
dto.setUserKeyword(StrUtil.trimToNull(dto.getUserKeyword()));
dto.setPeerKeyword(StrUtil.trimToNull(dto.getPeerKeyword()));
long total = privateMessageMapper.countConversations(dto);
long offset = (long) (dto.getPageNum() - 1) * dto.getPageSize();
if (offset >= total) {
return new ThirdTableVO<>(Collections.emptyList(), total);
}
List<ThirdPrivateConversationVO> rows = privateMessageMapper.selectConversations(dto, offset);
return new ThirdTableVO<>(rows == null ? Collections.emptyList() : rows, total);
}
/**
* 发送私聊消息
*/
@Override
public PrivateMessageVO sendMessage(ThirdPrivateMessageSendDTO dto) {
PrivateMessageModerationService.rejectReservedTmpId(dto.getTmpId());
PrivateMessageSessionControl.validateKickMessage(dto.getSendId(), dto.getType(), dto.getContent());
if (PrivateMessageSessionControl.hasKickPrefix(dto.getContent())) {
return sendSessionControl(dto);
}
Boolean isFriends = friendService.isFriend(dto.getSendId(), dto.getRecvId());
if (Boolean.FALSE.equals(isFriends)) {
throw new GlobalException("您已不是对方好友,无法发送消息");
@@ -90,13 +117,34 @@ public class ThirdPrivateMsgServiceImpl extends ServiceImpl<PrivateMessageMapper
return msgInfo;
}
private PrivateMessageVO sendSessionControl(ThirdPrivateMessageSendDTO dto) {
PrivateMessageVO command = BeanUtils.copyProperties(dto, PrivateMessageVO.class);
// Keep the existing response/envelope shape, without creating a replayable message row.
command.setId(0L);
command.setStatus(MessageStatus.READED.code());
command.setSendTime(new Date());
IMPrivateMessage<PrivateMessageVO> push = new IMPrivateMessage<>();
push.setSender(new IMUserInfo(dto.getSendId(), dto.getTerminal()));
push.setRecvId(dto.getRecvId());
push.setSendToSelf(false);
push.setSendResult(false);
push.setData(command);
imClient.sendPrivateMessage(push);
log.info("发送实时会话控制指令,接收id:{}", dto.getRecvId());
return command;
}
/**
* 查询私聊消息列表
*/
@Override
public ThirdTableVO<ThirdPrivateMessageVO> page(ThirdPrivateMessageDTO dto) {
ThirdTableVO<ThirdPrivateMessageVO> result = new ThirdTableVO<ThirdPrivateMessageVO>();
Page<PrivateMessage> page = baseMapper.selectPage(new Page(dto.getPageNum(), dto.getPageSize()), buildQueryWrapper(dto));
if (dto.getPageNum() == null) dto.setPageNum(1);
if (dto.getPageSize() == null) dto.setPageSize(10);
validatePagination(dto.getPageNum(), dto.getPageSize());
validateTimeRange(dto.getSendTimeList());
ThirdTableVO<ThirdPrivateMessageVO> result = new ThirdTableVO<>(Collections.emptyList(), 0);
Page<PrivateMessage> page = baseMapper.selectPage(new Page<>(dto.getPageNum(), dto.getPageSize()), buildQueryWrapper(dto));
result.setTotal(page.getTotal());
if (ObjectUtil.isNotEmpty(page.getRecords())) {
@@ -210,7 +258,7 @@ public class ThirdPrivateMsgServiceImpl extends ServiceImpl<PrivateMessageMapper
}
}
Map<String, Object> params = bo.getParams();
Map<String, Object> params = bo.getParams() == null ? Collections.emptyMap() : bo.getParams();
LambdaQueryWrapper<PrivateMessage> wrapper = Wrappers.lambdaQuery();
if (!Objects.isNull(bo.getSendId()) && !Objects.isNull(bo.getRecvId())) {
// 指定了两个用户,返回这两个用户之间的聊天记录
@@ -241,8 +289,26 @@ public class ThirdPrivateMsgServiceImpl extends ServiceImpl<PrivateMessageMapper
wrapper.between(params.get("beginTime") != null && params.get("endTime") != null, PrivateMessage::getSendTime,
params.get("beginTime"), params.get("endTime"));
wrapper.orderByDesc(PrivateMessage::getId);
if (bo.getSendId() != null && bo.getRecvId() != null) {
wrapper.orderByDesc(PrivateMessage::getSendTime, PrivateMessage::getId);
} else {
wrapper.orderByDesc(PrivateMessage::getId);
}
return wrapper;
}
private void validatePagination(Integer pageNum, Integer pageSize) {
if (pageNum == null || pageNum < 1 || pageSize == null || pageSize < 1 || pageSize > 100) {
throw new GlobalException("页码必须大于0,每页数量须在1到100之间");
}
}
private void validateTimeRange(List<Date> times) {
if (times == null || times.isEmpty()) return;
if (times.size() != 2 || times.get(0) == null || times.get(1) == null
|| times.get(0).after(times.get(1))) {
throw new GlobalException("时间范围必须包含有效的开始时间和结束时间,且开始时间不能晚于结束时间");
}
}
}
@@ -12,6 +12,7 @@ import com.bx.implatform.dto.third.user.ThirdGetUserByUserIdDTO;
import com.bx.implatform.dto.third.user.ThirdLoginDTO;
import com.bx.implatform.dto.third.user.ThirdUserOnLineStatusDto;
import com.bx.implatform.entity.User;
import com.bx.implatform.enums.ResultCode;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.UserMapper;
import com.bx.implatform.service.third.ThirdUserService;
@@ -22,6 +23,7 @@ import com.bx.implatform.vo.third.user.ThirdLoginVO;
import com.bx.implatform.vo.third.user.ThirdUserOnLineStatusVo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.security.crypto.password.PasswordEncoder;
import org.springframework.stereotype.Service;
@@ -30,6 +32,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
/**
@@ -76,22 +79,54 @@ public class ThirdUserServiceImpl extends ServiceImpl<UserMapper, User> implemen
*/
@Override
public void createOrUpdate(ThirdCreateOrUpdateUserDTO dto) {
// 查询用户
User user = baseMapper.selectById(dto.getUserId());
User saveObjUser = new User();
saveObjUser.setId(dto.getUserId());
saveObjUser.setUserName(dto.getUserName());
saveObjUser.setNickName(dto.getNickName());
saveObjUser.setHeadImage(dto.getAvatar());
saveObjUser.setHeadImageThumb(dto.getAvatar());
if (ObjectUtil.isEmpty(user)) {
// 创建用户
saveObjUser.setPassword(passwordEncoder.encode(dto.generatePassword()));
baseMapper.insert(saveObjUser);
} else {
// 更新用户
baseMapper.updateById(saveObjUser);
if (user != null) {
updateProfileIfChanged(user, dto);
return;
}
User newUser = buildUserProfile(dto);
// 昵称是可选参数,但数据库不允许为空。
if (newUser.getNickName() == null) {
newUser.setNickName(dto.getUserName());
}
newUser.setPassword(passwordEncoder.encode(dto.generatePassword()));
try {
baseMapper.insert(newUser);
} catch (DuplicateKeyException e) {
// 多个后端实例可能同时初始化同一个公共账号。
User concurrentUser = baseMapper.selectById(dto.getUserId());
if (concurrentUser == null) {
throw new GlobalException(ResultCode.USERNAME_ALREADY_REGISTER);
}
updateProfileIfChanged(concurrentUser, dto);
}
}
private void updateProfileIfChanged(User user, ThirdCreateOrUpdateUserDTO dto) {
boolean unchanged = Objects.equals(user.getUserName(), dto.getUserName())
&& (dto.getNickName() == null || Objects.equals(user.getNickName(), dto.getNickName()))
&& (dto.getAvatar() == null || (Objects.equals(user.getHeadImage(), dto.getAvatar())
&& Objects.equals(user.getHeadImageThumb(), dto.getAvatar())));
if (unchanged) {
return;
}
try {
// 只同步资料,保留已有密码、封禁状态等账号信息。
baseMapper.updateById(buildUserProfile(dto));
} catch (DuplicateKeyException e) {
throw new GlobalException(ResultCode.USERNAME_ALREADY_REGISTER);
}
}
private User buildUserProfile(ThirdCreateOrUpdateUserDTO dto) {
User user = new User();
user.setId(dto.getUserId());
user.setUserName(dto.getUserName());
user.setNickName(dto.getNickName());
user.setHeadImage(dto.getAvatar());
user.setHeadImageThumb(dto.getAvatar());
return user;
}
/**
@@ -0,0 +1,54 @@
package com.bx.implatform.util;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.bx.implatform.contant.Constant;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.enums.MessageType;
import com.bx.implatform.exception.GlobalException;
/** The existing mobile kick protocol is a live session command, never chat history. */
public final class PrivateMessageSessionControl {
public static final String KICK_PREFIX = "__APP_KICK_SESSION__:";
// Literal, case-sensitive prefix comparison avoids treating underscores as LIKE wildcards.
// COALESCE keeps unrelated legacy records with NULL fields visible.
private static final String HISTORY_FILTER = "NOT (COALESCE(send_id, 0) = {0}"
+ " AND COALESCE(type, -1) = {1}"
+ " AND BINARY LEFT(COALESCE(content, ''), {2}) = BINARY {3})";
private PrivateMessageSessionControl() {
}
public static boolean hasKickPrefix(String content) {
return content != null && content.startsWith(KICK_PREFIX);
}
public static boolean isKickMessage(PrivateMessage message) {
return message != null && Constant.COM_MESSAGE_USER_ID.equals(message.getSendId())
&& MessageType.TEXT.code().equals(message.getType()) && hasKickPrefix(message.getContent());
}
public static void rejectReservedKickContent(String content) {
if (hasKickPrefix(content)) {
throw new GlobalException("消息内容使用了保留的会话控制前缀");
}
}
public static void validateKickMessage(Long sendId, Integer type, String content) {
if (hasKickPrefix(content) && (!Constant.COM_MESSAGE_USER_ID.equals(sendId)
|| !MessageType.TEXT.code().equals(type))) {
throw new GlobalException("会话控制消息只能由消息通知账号发送文本指令");
}
}
public static void applyHistoryFilter(LambdaQueryWrapper<PrivateMessage> wrapper) {
wrapper.apply(HISTORY_FILTER, Constant.COM_MESSAGE_USER_ID, MessageType.TEXT.code(),
KICK_PREFIX.length(), KICK_PREFIX);
}
public static void applyHistoryFilter(QueryWrapper<PrivateMessage> wrapper) {
wrapper.apply(HISTORY_FILTER, Constant.COM_MESSAGE_USER_ID, MessageType.TEXT.code(),
KICK_PREFIX.length(), KICK_PREFIX);
}
}
@@ -0,0 +1,28 @@
package com.bx.implatform.vo.third.privatemsg;
import com.fasterxml.jackson.annotation.JsonIgnore;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
@Data
@Schema(description = "第三方私聊会话,双方按照用户ID从小到大排列")
public class ThirdPrivateConversationVO {
// Unique parent row identity for MyBatis nested result mappings.
@JsonIgnore
@Schema(hidden = true)
private Long latestMessageId;
private Participant userA;
private Participant userB;
private Long messageCount;
private ThirdPrivateMessageVO lastMessage;
@Data
public static class Participant {
private Long id;
private String nickname;
private String username;
private String avatar;
}
}
@@ -0,0 +1,13 @@
package com.bx.implatform.vo.third.privatemsg;
import lombok.Data;
@Data
public class ThirdPrivateMessageModerateVO {
private Long messageId;
private Integer status;
private Long recallMessageId;
private Boolean alreadyHidden;
/** Whether both online notification attempts were submitted, not a device acknowledgement. */
private Boolean notificationSent;
}
@@ -3,6 +3,131 @@
<mapper namespace="com.bx.implatform.mapper.PrivateMessageMapper">
<resultMap id="conversationResult" type="com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO">
<id property="latestMessageId" column="id"/>
<result property="messageCount" column="message_count"/>
<association property="userA" javaType="com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO$Participant">
<id property="id" column="user_a_id"/>
<result property="nickname" column="user_a_nickname"/>
<result property="username" column="user_a_username"/>
<result property="avatar" column="user_a_avatar"/>
</association>
<association property="userB" javaType="com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO$Participant">
<id property="id" column="user_b_id"/>
<result property="nickname" column="user_b_nickname"/>
<result property="username" column="user_b_username"/>
<result property="avatar" column="user_b_avatar"/>
</association>
<association property="lastMessage" javaType="com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageVO">
<id property="id" column="id"/>
<result property="sendId" column="send_id"/>
<result property="recvId" column="recv_id"/>
<result property="content" column="content"/>
<result property="type" column="type"/>
<result property="status" column="status"/>
<result property="sendTime" column="send_time"/>
</association>
</resultMap>
<sql id="senderMatchesUser">
(CAST(m.send_id AS CHAR) = #{dto.userKeyword}
OR su.nick_name LIKE CONCAT('%', #{dto.userKeyword}, '%')
OR su.user_name LIKE CONCAT('%', #{dto.userKeyword}, '%'))
</sql>
<sql id="receiverMatchesUser">
(CAST(m.recv_id AS CHAR) = #{dto.userKeyword}
OR ru.nick_name LIKE CONCAT('%', #{dto.userKeyword}, '%')
OR ru.user_name LIKE CONCAT('%', #{dto.userKeyword}, '%'))
</sql>
<sql id="senderMatchesPeer">
(CAST(m.send_id AS CHAR) = #{dto.peerKeyword}
OR su.nick_name LIKE CONCAT('%', #{dto.peerKeyword}, '%')
OR su.user_name LIKE CONCAT('%', #{dto.peerKeyword}, '%'))
</sql>
<sql id="receiverMatchesPeer">
(CAST(m.recv_id AS CHAR) = #{dto.peerKeyword}
OR ru.nick_name LIKE CONCAT('%', #{dto.peerKeyword}, '%')
OR ru.user_name LIKE CONCAT('%', #{dto.peerKeyword}, '%'))
</sql>
<!-- Derived-table aggregation works on MySQL 5.7 and 8. Filter participants before
grouping, but apply the last-chat date afterwards so counts cover the whole pair. -->
<sql id="groupedConversations">
SELECT LEAST(m.send_id, m.recv_id) AS user_a_id,
GREATEST(m.send_id, m.recv_id) AS user_b_id,
COUNT(*) AS message_count,
MAX(m.send_time) AS send_time
FROM im_private_message m
<if test="(dto.userKeyword != null and dto.userKeyword != '') or (dto.peerKeyword != null and dto.peerKeyword != '')">
LEFT JOIN im_user su ON su.id = m.send_id
LEFT JOIN im_user ru ON ru.id = m.recv_id
</if>
<where>
<choose>
<when test="dto.userKeyword != null and dto.userKeyword != '' and dto.peerKeyword != null and dto.peerKeyword != ''">
((<include refid="senderMatchesUser"/> AND <include refid="receiverMatchesPeer"/>)
OR (<include refid="receiverMatchesUser"/> AND <include refid="senderMatchesPeer"/>))
</when>
<when test="dto.userKeyword != null and dto.userKeyword != ''">
(<include refid="senderMatchesUser"/> OR <include refid="receiverMatchesUser"/>)
</when>
<when test="dto.peerKeyword != null and dto.peerKeyword != ''">
(<include refid="senderMatchesPeer"/> OR <include refid="receiverMatchesPeer"/>)
</when>
</choose>
</where>
GROUP BY LEAST(m.send_id, m.recv_id), GREATEST(m.send_id, m.recv_id)
</sql>
<sql id="latestConversationWhere">
<where>
<if test="dto.lastSendTimeList != null and dto.lastSendTimeList.size() == 2">
r.send_time BETWEEN #{dto.lastSendTimeList[0]} AND #{dto.lastSendTimeList[1]}
</if>
</where>
</sql>
<select id="countConversations" resultType="long">
SELECT COUNT(*) FROM (
<include refid="groupedConversations"/>
) r
<include refid="latestConversationWhere"/>
</select>
<select id="selectConversations" resultMap="conversationResult">
SELECT r.user_a_id, r.user_b_id, r.message_count,
ua.nick_name AS user_a_nickname, ua.user_name AS user_a_username,
COALESCE(NULLIF(ua.head_image_thumb, ''), ua.head_image) AS user_a_avatar,
ub.nick_name AS user_b_nickname, ub.user_name AS user_b_username,
COALESCE(NULLIF(ub.head_image_thumb, ''), ub.head_image) AS user_b_avatar,
m.id, m.send_id, m.recv_id, m.content, m.type, m.status, m.send_time
FROM (
<include refid="groupedConversations"/>
) r
<!-- Probe each direction separately so the pair/time index supports both lookups.
NULL-safe equality keeps all-NULL timestamps. Each group has a latest message;
the signed BIGINT minimum only fills an absent direction, without assuming IDs > 0. -->
JOIN im_private_message m ON m.id = GREATEST(
COALESCE((
SELECT MAX(latest_forward.id)
FROM im_private_message latest_forward
WHERE latest_forward.send_id = r.user_a_id AND latest_forward.recv_id = r.user_b_id
AND latest_forward.send_time &lt;=&gt; r.send_time
), -9223372036854775808),
COALESCE((
SELECT MAX(latest_reverse.id)
FROM im_private_message latest_reverse
WHERE latest_reverse.send_id = r.user_b_id AND latest_reverse.recv_id = r.user_a_id
AND latest_reverse.send_time &lt;=&gt; r.send_time
), -9223372036854775808)
)
LEFT JOIN im_user ua ON ua.id = r.user_a_id
LEFT JOIN im_user ub ON ub.id = r.user_b_id
<include refid="latestConversationWhere"/>
ORDER BY r.send_time DESC, m.id DESC
LIMIT #{dto.pageSize} OFFSET #{offset}
</select>
<sql id="whereA">
<if test="dto.sendId != null">
and a.send_id = #{dto.sendId}
@@ -0,0 +1,134 @@
package com.bx.implatform.controller.third;
import com.bx.implatform.config.props.JwtProperties;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateConversationDTO;
import com.bx.implatform.exception.GlobalExceptionHandler;
import com.bx.implatform.filter.CacheFilter;
import com.bx.implatform.interceptor.AuthInterceptor;
import com.bx.implatform.service.third.ThirdAuthService;
import com.bx.implatform.service.third.ThirdPrivateMsgService;
import com.bx.implatform.vo.third.ThirdTableVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageVO;
import jakarta.servlet.http.HttpServletRequest;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.http.MediaType;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import java.util.Collections;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
@ExtendWith(MockitoExtension.class)
class ThirdPrivateConversationControllerTest {
private static final String URL = "/third/privateMsg/conversation/page";
@Mock
private ThirdPrivateMsgService service;
@Mock
private ThirdAuthService authService;
private MockMvc mvc;
private boolean validSignature = true;
@BeforeEach
void setUp() {
when(authService.checkSign(any(HttpServletRequest.class))).thenAnswer(ignored -> validSignature);
mvc = MockMvcBuilders.standaloneSetup(new ThirdPrivateMsgController(service))
.setControllerAdvice(new GlobalExceptionHandler())
.addInterceptors(new AuthInterceptor(new JwtProperties(), authService))
.addFilters(new CacheFilter()).build();
}
@Test
void responseKeepsBothParticipantsAndLastMessageInPagedContract() throws Exception {
ThirdPrivateConversationVO row = new ThirdPrivateConversationVO();
ThirdPrivateConversationVO.Participant a = new ThirdPrivateConversationVO.Participant();
a.setId(1L);
a.setNickname("Alice");
a.setUsername("alice-account");
a.setAvatar("https://example.test/alice.png");
ThirdPrivateConversationVO.Participant b = new ThirdPrivateConversationVO.Participant();
b.setId(2L);
b.setNickname("Bob");
row.setUserA(a);
row.setUserB(b);
row.setMessageCount(4L);
ThirdPrivateMessageVO last = new ThirdPrivateMessageVO();
last.setId(102L);
last.setSendId(2L);
last.setRecvId(1L);
last.setContent("last message");
last.setType(0L);
last.setStatus(3L);
row.setLastMessage(last);
when(service.conversationPage(any())).thenReturn(new ThirdTableVO<>(List.of(row), 1));
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content("{}"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.data.total").value(1))
.andExpect(jsonPath("$.data.rows[0].userA.id").value(1))
.andExpect(jsonPath("$.data.rows[0].userA.nickname").value("Alice"))
.andExpect(jsonPath("$.data.rows[0].userB.id").value(2))
.andExpect(jsonPath("$.data.rows[0].messageCount").value(4))
.andExpect(jsonPath("$.data.rows[0].lastMessage.sendId").value(2));
}
@Test
void acceptsEpochMillisAndKeepsEmptyPageAsArray() throws Exception {
when(service.conversationPage(any())).thenReturn(new ThirdTableVO<>(Collections.emptyList(), 0));
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON)
.content("{\"lastSendTimeList\":[1793462400000,1793548799999]}"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.data.rows").isArray())
.andExpect(jsonPath("$.data.rows").isEmpty())
.andExpect(jsonPath("$.data.total").value(0));
ArgumentCaptor<ThirdPrivateConversationDTO> captor = ArgumentCaptor.forClass(ThirdPrivateConversationDTO.class);
verify(service).conversationPage(captor.capture());
assertEquals(1793462400000L, captor.getValue().getLastSendTimeList().get(0).getTime());
assertEquals(1793548799999L, captor.getValue().getLastSendTimeList().get(1).getTime());
assertEquals(1, captor.getValue().getPageNum());
assertEquals(10, captor.getValue().getPageSize());
}
@ParameterizedTest
@ValueSource(strings = {"{\"pageNum\":0}", "{\"pageNum\":-1}", "{\"pageSize\":0}", "{\"pageSize\":101}"})
void rejectsInvalidPageBoundsBeforeCallingService(String body) throws Exception {
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(body))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(500));
verifyNoInteractions(service);
}
@Test
void invalidSignatureCannotReadConversations() throws Exception {
validSignature = false;
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content("{}"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(401));
verifyNoInteractions(service);
}
@Test
void databaseFailureIsAnErrorRatherThanSuccessfulEmptyPage() throws Exception {
when(service.conversationPage(any())).thenThrow(new IllegalStateException("internal database failure"));
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content("{}"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(500))
.andExpect(jsonPath("$.message").value("系统繁忙,请稍后再试"));
}
}
@@ -0,0 +1,109 @@
package com.bx.implatform.controller.third;
import com.bx.implatform.config.props.JwtProperties;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageModerateDTO;
import com.bx.implatform.exception.GlobalExceptionHandler;
import com.bx.implatform.filter.CacheFilter;
import com.bx.implatform.interceptor.AuthInterceptor;
import com.bx.implatform.service.PrivateMessageModerationService;
import com.bx.implatform.service.third.ThirdAuthService;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageModerateVO;
import jakarta.servlet.http.HttpServletRequest;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.http.MediaType;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
@ExtendWith(MockitoExtension.class)
class ThirdPrivateMessageModerationControllerTest {
private static final String URL = "/third/privateMsg/moderate";
private static final String VALID_BODY = "{\"messageId\":101,\"operatorId\":42,\"reason\":\"违规内容\"}";
@Mock
private PrivateMessageModerationService service;
@Mock
private ThirdAuthService authService;
private MockMvc mvc;
private boolean validSignature = true;
@BeforeEach
void setUp() {
when(authService.checkSign(any(HttpServletRequest.class))).thenAnswer(ignored -> validSignature);
mvc = MockMvcBuilders.standaloneSetup(new ThirdPrivateMessageModerationController(service))
.setControllerAdvice(new GlobalExceptionHandler())
.addInterceptors(new AuthInterceptor(new JwtProperties(), authService))
.addFilters(new CacheFilter()).build();
}
@Test
void returnsModerationReceiptAndKeepsOriginalMessageId() throws Exception {
ThirdPrivateMessageModerateVO response = new ThirdPrivateMessageModerateVO();
response.setMessageId(101L);
response.setStatus(2);
response.setRecallMessageId(202L);
response.setAlreadyHidden(false);
response.setNotificationSent(true);
when(service.moderate(any())).thenReturn(response);
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(VALID_BODY))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.data.messageId").value(101))
.andExpect(jsonPath("$.data.status").value(2))
.andExpect(jsonPath("$.data.recallMessageId").value(202))
.andExpect(jsonPath("$.data.alreadyHidden").value(false))
.andExpect(jsonPath("$.data.notificationSent").value(true));
ArgumentCaptor<ThirdPrivateMessageModerateDTO> captor = ArgumentCaptor.forClass(ThirdPrivateMessageModerateDTO.class);
verify(service).moderate(captor.capture());
assertEquals(42L, captor.getValue().getOperatorId());
assertEquals("违规内容", captor.getValue().getReason());
}
@ParameterizedTest
@ValueSource(strings = {"{}", "{\"messageId\":101}", "{\"operatorId\":42}",
"{\"messageId\":0,\"operatorId\":42}", "{\"messageId\":101,\"operatorId\":0}"})
void rejectsMissingOrInvalidIdsBeforeMutating(String body) throws Exception {
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(body))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(500));
verifyNoInteractions(service);
}
@Test
void rejectsReasonLongerThanTwoHundredCharacters() throws Exception {
String body = "{\"messageId\":101,\"operatorId\":42,\"reason\":\"" + "x".repeat(201) + "\"}";
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(body))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(500));
verifyNoInteractions(service);
}
@Test
void invalidSignatureCannotModerateAnyMessage() throws Exception {
validSignature = false;
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(VALID_BODY))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(401));
verifyNoInteractions(service);
}
@Test
void databaseFailureDoesNotReturnSuccessfulModerationReceipt() throws Exception {
when(service.moderate(any())).thenThrow(new IllegalStateException("database error"));
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(VALID_BODY))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(500))
.andExpect(jsonPath("$.message").value("系统繁忙,请稍后再试"));
}
}
@@ -0,0 +1,151 @@
package com.bx.implatform.controller.third;
import com.bx.implatform.config.props.JwtProperties;
import com.bx.implatform.dto.third.user.ThirdCreateOrUpdateUserDTO;
import com.bx.implatform.enums.ResultCode;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.exception.GlobalExceptionHandler;
import com.bx.implatform.filter.CacheFilter;
import com.bx.implatform.interceptor.AuthInterceptor;
import com.bx.implatform.service.third.ThirdAuthService;
import com.bx.implatform.service.third.ThirdUserService;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.servlet.http.HttpServletRequest;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.http.MediaType;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import java.util.Map;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
@ExtendWith(MockitoExtension.class)
class ThirdUserControllerTest {
private static final String URL = "/third/user/createOrUpdate";
private static final String VALID_BODY = """
{"userId":-100,"userName":"message-notification","phone":"15555555555","nickName":"消息通知"}
""";
@Mock
private ThirdUserService service;
@Mock
private ThirdAuthService authService;
private MockMvc mvc;
private boolean validSignature = true;
@BeforeEach
void setUp() {
when(authService.checkSign(any(HttpServletRequest.class))).thenAnswer(ignored -> validSignature);
mvc = MockMvcBuilders.standaloneSetup(new ThirdUserController(service))
.setControllerAdvice(new GlobalExceptionHandler())
.addInterceptors(new AuthInterceptor(new JwtProperties(), authService))
.addFilters(new CacheFilter())
.build();
}
@Test
void successUsesExistingJsonResponseContract() throws Exception {
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(VALID_BODY))
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.message").value("成功"))
.andExpect(jsonPath("$.data").isEmpty());
verify(service).createOrUpdate(any(ThirdCreateOrUpdateUserDTO.class));
}
@ParameterizedTest
@ValueSource(strings = {"", " ", "\t", "\n"})
void blankUsernameReturnsJsonValidationError(String username) throws Exception {
String body = new ObjectMapper().writeValueAsString(
Map.of("userId", -100, "userName", username, "phone", "15555555555"));
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(body))
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andExpect(jsonPath("$.code").value(500))
.andExpect(jsonPath("$.message").value("用户名不可为空"));
verifyNoInteractions(service);
}
@Test
void missingUserIdReturnsJsonValidationError() throws Exception {
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON)
.content("{\"userName\":\"message\",\"phone\":\"15555555555\"}"))
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andExpect(jsonPath("$.message").value("用户id不可为空"));
verifyNoInteractions(service);
}
@Test
void blankPhoneReturnsJsonValidationError() throws Exception {
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON)
.content(VALID_BODY.replace("15555555555", " ")))
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andExpect(jsonPath("$.message").value("用户手机号不可为空"));
verifyNoInteractions(service);
}
@Test
void malformedRequestReturnsJsonError() throws Exception {
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content("{invalid"))
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andExpect(jsonPath("$.code").value(500));
verifyNoInteractions(service);
}
@Test
void unexpectedServiceFailureReturnsJsonWithoutExposingInternalError() throws Exception {
doThrow(new IllegalStateException("internal database failure"))
.when(service).createOrUpdate(any(ThirdCreateOrUpdateUserDTO.class));
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(VALID_BODY))
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andExpect(jsonPath("$.code").value(500))
.andExpect(jsonPath("$.message").value(ResultCode.PROGRAM_ERROR.getMsg()));
}
@Test
void usernameConflictReturnsJsonBusinessError() throws Exception {
doThrow(new GlobalException(ResultCode.USERNAME_ALREADY_REGISTER))
.when(service).createOrUpdate(any(ThirdCreateOrUpdateUserDTO.class));
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(VALID_BODY))
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andExpect(jsonPath("$.code").value(10003));
}
@Test
void invalidSignatureReturnsJsonAndDoesNotWriteUser() throws Exception {
validSignature = false;
mvc.perform(post(URL).contentType(MediaType.APPLICATION_JSON).content(VALID_BODY))
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andExpect(jsonPath("$.code").value(401));
verifyNoInteractions(service);
}
}
@@ -0,0 +1,391 @@
package com.bx.implatform.mapper;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder;
import com.bx.implatform.config.MybatisPlusConfig;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateConversationDTO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO;
import org.apache.ibatis.builder.xml.XMLMapperBuilder;
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
import org.apache.ibatis.io.Resources;
import org.apache.ibatis.mapping.Environment;
import org.apache.ibatis.mapping.BoundSql;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.scripting.defaults.DefaultParameterHandler;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
/**
* Opt-in real MySQL 5.7/8 verification. Supply IM_TEST_MYSQL_URL, IM_TEST_MYSQL_USER and
* IM_TEST_MYSQL_PASSWORD pointing at a disposable localhost instance. Each case creates a
* uniquely named test database and drops only that database; existing tables are never modified.
*/
class PrivateConversationMapperMySqlTest {
private SqlSession session;
private PrivateMessageMapper mapper;
private String testDatabase;
@BeforeEach
void setUp() throws Exception {
String url = System.getenv("IM_TEST_MYSQL_URL");
assumeTrue(url != null && !url.isBlank(), "Set IM_TEST_MYSQL_URL to opt in to isolated MySQL tests");
assertTrue(url.matches("jdbc:mysql://(localhost|127\\.0\\.0\\.1):[0-9]+/[a-zA-Z0-9_]+(?:\\?.*)?"),
"MySQL test connections must use an explicit localhost port and database");
UnpooledDataSource dataSource = new UnpooledDataSource("com.mysql.cj.jdbc.Driver", url,
System.getenv().getOrDefault("IM_TEST_MYSQL_USER", "root"),
System.getenv().getOrDefault("IM_TEST_MYSQL_PASSWORD", ""));
MybatisConfiguration configuration = new MybatisConfiguration();
configuration.setMapUnderscoreToCamelCase(true);
configuration.addInterceptor(new MybatisPlusConfig().mybatisPlusInterceptor());
configuration.setEnvironment(new Environment("conversation-test", new JdbcTransactionFactory(), dataSource));
try (InputStream xml = Resources.getResourceAsStream("mapper/PrivateMessageMapper.xml")) {
new XMLMapperBuilder(xml, configuration, "mapper/PrivateMessageMapper.xml", configuration.getSqlFragments()).parse();
}
session = new MybatisSqlSessionFactoryBuilder().build(configuration).openSession(true);
DatabaseMetaData metadata = session.getConnection().getMetaData();
assertTrue(metadata.getDatabaseMajorVersion() > 5
|| (metadata.getDatabaseMajorVersion() == 5 && metadata.getDatabaseMinorVersion() >= 7),
"Conversation SQL supports MySQL 5.7 and newer");
try (Statement statement = session.getConnection().createStatement()) {
String newDatabase = "codex_im_conversation_" + UUID.randomUUID().toString().replace("-", "");
statement.execute("CREATE DATABASE " + newDatabase);
testDatabase = newDatabase;
}
session.getConnection().setCatalog(testDatabase);
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("CREATE TABLE im_user (id BIGINT PRIMARY KEY, user_name VARCHAR(255), "
+ "nick_name VARCHAR(255), head_image VARCHAR(255), head_image_thumb VARCHAR(255))");
statement.execute("CREATE TABLE im_private_message (id BIGINT PRIMARY KEY, "
+ "send_id BIGINT NOT NULL, recv_id BIGINT NOT NULL, content TEXT, type INT, status INT, "
+ "send_time DATETIME(3), INDEX idx_pair(send_id,recv_id,send_time,id))");
statement.execute("INSERT INTO im_user VALUES "
+ "(1,'alice-account','Alice','alice.png','alice-thumb.png'),"
+ "(2,'bob-account','Bob','bob.png',''),(3,'carol-account','Carol','carol.png',NULL)");
statement.execute("INSERT INTO im_private_message VALUES "
+ "(100,1,2,'old',0,3,'2026-11-01 09:00:00'),"
+ "(101,2,1,'same time older id',0,3,'2026-11-03 11:00:00'),"
+ "(102,1,2,'latest by time and id',1,2,'2026-11-03 11:00:00'),"
+ "(500,2,1,'higher id but older time',0,3,'2026-11-02 11:00:00'),"
+ "(200,1,3,'carol outbound',0,3,'2026-11-02 10:00:00'),"
+ "(201,3,1,'carol inbound',0,3,'2026-11-02 11:00:00'),"
+ "(300,2,9,'missing user at day end',0,3,'2026-11-01 23:59:59.999')");
}
mapper = session.getMapper(PrivateMessageMapper.class);
}
@AfterEach
void tearDown() throws Exception {
if (session != null) {
try {
if (testDatabase != null) {
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("DROP DATABASE " + testDatabase);
}
}
} finally {
session.close();
}
}
}
@Test
void mergesDirectionsAndSelectsNewestByTimeThenIdRatherThanLargestId() {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
assertEquals(3, mapper.countConversations(dto));
List<ThirdPrivateConversationVO> rows = mapper.selectConversations(dto, 0L);
assertEquals(3, rows.size());
ThirdPrivateConversationVO first = rows.get(0);
assertEquals(1L, first.getUserA().getId());
assertEquals(2L, first.getUserB().getId());
assertEquals(4L, first.getMessageCount());
assertEquals(102L, first.getLastMessage().getId());
assertEquals("latest by time and id", first.getLastMessage().getContent());
assertEquals(1L, first.getLastMessage().getType());
assertEquals(2L, first.getLastMessage().getStatus());
assertEquals("Alice", first.getUserA().getNickname());
assertEquals("alice-account", first.getUserA().getUsername());
}
@Test
void pagesConversationsWithoutDroppingOrDuplicatingPairs() {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setPageSize(1);
assertEquals(3, mapper.countConversations(dto));
assertEquals(102L, mapper.selectConversations(dto, 0L).get(0).getLastMessage().getId());
assertEquals(201L, mapper.selectConversations(dto, 1L).get(0).getLastMessage().getId());
assertEquals(300L, mapper.selectConversations(dto, 2L).get(0).getLastMessage().getId());
assertTrue(mapper.selectConversations(dto, 3L).isEmpty());
}
@Test
void equalLatestTimesUseMessageIdForStablePaginationAcrossPairs() throws Exception {
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("INSERT INTO im_private_message VALUES "
+ "(501,3,4,'same time lower id',0,3,'2026-11-03 11:00:00'),"
+ "(502,4,3,'same time higher id',0,3,'2026-11-03 11:00:00'),"
+ "(503,5,6,'same time newest pair',0,3,'2026-11-03 11:00:00')");
}
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setPageSize(1);
assertEquals(5, mapper.countConversations(dto));
long[] expectedIds = {503L, 502L, 102L, 201L, 300L};
for (int offset = 0; offset < expectedIds.length; offset++) {
List<ThirdPrivateConversationVO> rows = mapper.selectConversations(dto, offset);
assertEquals(1, rows.size());
assertEquals(expectedIds[offset], rows.get(0).getLastMessage().getId());
}
assertTrue(mapper.selectConversations(dto, expectedIds.length).isEmpty());
}
@Test
void allNullTimesRetainPairAndUseLargestIdAfterDatedConversations() throws Exception {
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("INSERT INTO im_private_message VALUES "
+ "(400,4,5,'null time lower id',0,3,NULL),"
+ "(401,5,4,'null time higher id',0,3,NULL)");
}
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
assertEquals(4, mapper.countConversations(dto));
List<ThirdPrivateConversationVO> rows = mapper.selectConversations(dto, 0L);
assertEquals(4, rows.size());
ThirdPrivateConversationVO nullTimePair = rows.get(3);
assertEquals(4L, nullTimePair.getUserA().getId());
assertEquals(5L, nullTimePair.getUserB().getId());
assertEquals(2L, nullTimePair.getMessageCount());
assertEquals(401L, nullTimePair.getLastMessage().getId());
assertNull(nullTimePair.getLastMessage().getSendTime());
dto.setUserKeyword("4");
dto.setPeerKeyword("5");
assertEquals(1, mapper.countConversations(dto));
assertEquals(1, mapper.selectConversations(dto, 0L).size());
dto.setLastSendTimeList(day("2026-11-01"));
assertEquals(0, mapper.countConversations(dto));
assertTrue(mapper.selectConversations(dto, 0L).isEmpty());
}
@Test
void datedMessageWinsOverNullTimesEvenWithLargerNullMessageId() throws Exception {
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("INSERT INTO im_private_message VALUES "
+ "(600,1,4,'null time lower id',0,3,NULL),"
+ "(601,4,1,'dated latest message',0,3,'2026-11-01 12:00:00'),"
+ "(602,1,4,'null time higher id',0,3,NULL)");
}
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setUserKeyword("4");
dto.setPeerKeyword("1");
assertEquals(1, mapper.countConversations(dto));
ThirdPrivateConversationVO row = mapper.selectConversations(dto, 0L).get(0);
assertEquals(3L, row.getMessageCount());
assertEquals(601L, row.getLastMessage().getId());
assertEquals(Timestamp.valueOf("2026-11-01 12:00:00").getTime(),
row.getLastMessage().getSendTime().getTime());
dto.setLastSendTimeList(day("2026-11-01"));
assertEquals(1, mapper.countConversations(dto));
assertEquals(3L, mapper.selectConversations(dto, 0L).get(0).getMessageCount());
}
@Test
void missingDirectionRetainsMinimumNegativeAndZeroMessageIds() throws Exception {
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("INSERT INTO im_private_message VALUES "
+ "(-9223372036854775808,4,5,'minimum signed id',0,3,'2026-11-03 12:00:00'),"
+ "(-11,7,6,'older negative id',0,3,'2026-11-03 11:00:00'),"
+ "(-10,7,6,'latest negative id',0,3,'2026-11-03 12:00:00'),"
+ "(0,9,8,'zero id',0,3,'2026-11-03 12:00:00')");
}
long[][] expectedPairs = {{4L, 5L, Long.MIN_VALUE, 1L}, {6L, 7L, -10L, 2L}, {8L, 9L, 0L, 1L}};
for (long[] expected : expectedPairs) {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setUserKeyword(Long.toString(expected[0]));
dto.setPeerKeyword(Long.toString(expected[1]));
assertEquals(1, mapper.countConversations(dto));
List<ThirdPrivateConversationVO> rows = mapper.selectConversations(dto, 0L);
assertEquals(1, rows.size());
assertEquals(expected[2], rows.get(0).getLastMessage().getId());
assertEquals(expected[3], rows.get(0).getMessageCount());
}
}
@Test
void participantFiltersAreSymmetricAndKeepBothDirectionsInCount() {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setUserKeyword("Alice");
dto.setPeerKeyword("bob-account");
assertEquals(1, mapper.countConversations(dto));
assertEquals(4L, mapper.selectConversations(dto, 0L).get(0).getMessageCount());
dto.setUserKeyword("bob-account");
dto.setPeerKeyword("Alice");
assertEquals(1, mapper.countConversations(dto));
assertEquals(4L, mapper.selectConversations(dto, 0L).get(0).getMessageCount());
dto.setUserKeyword("Alice");
dto.setPeerKeyword("Alice");
assertEquals(0, mapper.countConversations(dto));
}
@Test
void unmatchedParticipantReturnsEmptyAndNumericIdMatchesExactly() {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setUserKeyword("no-such-person");
assertEquals(0, mapper.countConversations(dto));
assertTrue(mapper.selectConversations(dto, 0L).isEmpty());
dto.setUserKeyword("2");
assertEquals(2, mapper.countConversations(dto));
dto.setUserKeyword("20");
assertEquals(0, mapper.countConversations(dto));
}
@Test
void recentDateFiltersAfterAggregationInsteadOfSelectingOlderMessageInRange() {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setLastSendTimeList(day("2026-11-02"));
assertEquals(1, mapper.countConversations(dto));
List<ThirdPrivateConversationVO> rows = mapper.selectConversations(dto, 0L);
assertEquals(1, rows.size());
assertEquals(3L, rows.get(0).getUserB().getId());
assertEquals(2L, rows.get(0).getMessageCount());
}
@Test
void retainsMissingProfileAndIncludesEntireSelectedDay() {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setUserKeyword("9");
dto.setLastSendTimeList(day("2026-11-01"));
assertEquals(1, mapper.countConversations(dto));
ThirdPrivateConversationVO row = mapper.selectConversations(dto, 0L).get(0);
assertEquals(2L, row.getUserA().getId());
assertEquals(9L, row.getUserB().getId());
assertEquals(300L, row.getLastMessage().getId());
}
@Test
void equalMessageCountsDoNotMergeDifferentConversationsInResultMapping() throws Exception {
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("INSERT INTO im_private_message VALUES "
+ "(400,1,8,'first one-message pair',0,3,'2026-11-01 20:00:00'),"
+ "(401,2,8,'second one-message pair',0,3,'2026-11-01 21:00:00')");
}
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
assertEquals(5, mapper.countConversations(dto));
List<ThirdPrivateConversationVO> rows = mapper.selectConversations(dto, 0L);
assertEquals(5, rows.size());
assertEquals(5, rows.stream().map(row -> row.getLastMessage().getId()).distinct().count());
assertEquals(3, rows.stream().filter(row -> row.getMessageCount() == 1L).count());
}
@Test
void indexMigrationCanRunTwiceWithoutChangingMessageRows() throws Exception {
executeIndexMigration();
executeIndexMigration();
try (Statement statement = session.getConnection().createStatement();
ResultSet result = statement.executeQuery("SELECT GROUP_CONCAT(column_name ORDER BY seq_in_index) "
+ "FROM information_schema.statistics WHERE table_schema=DATABASE() "
+ "AND table_name='im_private_message' AND index_name='idx_private_pair_time'")) {
assertTrue(result.next());
assertEquals("send_id,recv_id,send_time,id", result.getString(1));
}
try (Statement statement = session.getConnection().createStatement();
ResultSet result = statement.executeQuery("SELECT COUNT(*) FROM im_private_message")) {
assertTrue(result.next());
assertEquals(7, result.getLong(1));
}
}
@Test
void recordsRepresentativeExplainForTenThousandMessages() throws Exception {
executeIndexMigration();
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("ALTER TABLE im_private_message DROP INDEX idx_pair");
String digits = "(SELECT 0 n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 "
+ "UNION ALL SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 "
+ "UNION ALL SELECT 8 UNION ALL SELECT 9)";
statement.execute("INSERT INTO im_private_message (id,send_id,recv_id,content,type,status,send_time) "
+ "SELECT 10000+n,1000+MOD(n,400),2000+MOD(n,400),REPEAT('fixture',20),0,3,"
+ "DATE_ADD('2026-11-02 00:00:00',INTERVAL n SECOND) FROM "
+ "(SELECT a.n+b.n*10+c.n*100+d.n*1000 n FROM " + digits + " a CROSS JOIN "
+ digits + " b CROSS JOIN " + digits + " c CROSS JOIN " + digits + " d) numbers");
statement.execute("ANALYZE TABLE im_private_message");
}
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
assertEquals(403, mapper.countConversations(dto));
assertEquals(dto.getPageSize(), mapper.selectConversations(dto, 0L).size());
MappedStatement mapped = session.getConfiguration().getMappedStatement(
PrivateMessageMapper.class.getName() + ".selectConversations");
Map<String, Object> parameters = Map.of("dto", dto, "offset", 0L);
BoundSql boundSql = mapped.getBoundSql(parameters);
try (PreparedStatement statement = session.getConnection().prepareStatement("EXPLAIN " + boundSql.getSql())) {
new DefaultParameterHandler(mapped, parameters, boundSql).setParameters(statement);
try (ResultSet result = statement.executeQuery()) {
StringBuilder plan = new StringBuilder();
int indexedLatestLookups = 0;
while (result.next()) {
String table = result.getString("table");
plan.append(result.getString("select_type")).append(" table=")
.append(table).append(" type=")
.append(result.getString("type")).append(" key=")
.append(result.getString("key")).append(" rows=")
.append(result.getString("rows")).append(" extra=")
.append(result.getString("Extra")).append('\n');
if ("latest_forward".equals(table) || "latest_reverse".equals(table)) {
assertEquals("idx_private_pair_time", result.getString("key"),
table + " must use the pair/time index");
assertNotEquals("ALL", result.getString("type"),
table + " must not scan all messages for each conversation");
indexedLatestLookups++;
}
}
assertFalse(plan.toString().isBlank());
assertEquals(2, indexedLatestLookups, "Both directions must have indexed latest-message lookups");
System.out.println("Conversation EXPLAIN (10,007 synthetic messages, not a production benchmark):\n" + plan);
}
}
try (Statement statement = session.getConnection().createStatement();
ResultSet result = statement.executeQuery("EXPLAIN SELECT * FROM im_private_message "
+ "WHERE (send_id=1 AND recv_id=2) OR (send_id=2 AND recv_id=1) "
+ "ORDER BY send_time DESC,id DESC LIMIT 10")) {
assertTrue(result.next());
assertEquals("idx_private_pair_time", result.getString("key"));
System.out.println("Bilateral detail EXPLAIN: key=" + result.getString("key")
+ " type=" + result.getString("type") + " extra=" + result.getString("Extra"));
}
}
private void executeIndexMigration() throws Exception {
Path script = Path.of("db", "20261103_private_conversation_index.sql");
if (!Files.exists(script)) {
script = Path.of("..", "db", "20261103_private_conversation_index.sql");
}
String sql = Files.readString(script).replaceAll("(?m)^--.*$", "");
try (Statement statement = session.getConnection().createStatement()) {
for (String command : sql.split(";")) {
if (!command.isBlank()) {
statement.execute(command);
}
}
}
}
private List<Date> day(String day) {
return List.of(Timestamp.valueOf(day + " 00:00:00"), Timestamp.valueOf(day + " 23:59:59.999"));
}
}
@@ -0,0 +1,172 @@
package com.bx.implatform.mapper;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.bx.implatform.config.MybatisPlusConfig;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateConversationDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.util.PrivateMessageSessionControl;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO;
import org.apache.ibatis.builder.xml.XMLMapperBuilder;
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
import org.apache.ibatis.io.Resources;
import org.apache.ibatis.mapping.Environment;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.InputStream;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.Statement;
import java.util.List;
import java.util.UUID;
import java.util.stream.LongStream;
import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
/** Real MySQL 5.7/8 predicates, using only a fresh disposable localhost schema per test. */
class PrivateMessageSessionControlMySqlTest {
private SqlSession session;
private PrivateMessageMapper mapper;
private String testDatabase;
@BeforeEach
void setUp() throws Exception {
String url = System.getenv("IM_TEST_MYSQL_URL");
assumeTrue(url != null && !url.isBlank(), "Set IM_TEST_MYSQL_URL for isolated MySQL verification");
assertTrue(url.matches("jdbc:mysql://(localhost|127\\.0\\.0\\.1):[0-9]+/[a-zA-Z0-9_]+(?:\\?.*)?"),
"MySQL tests require an explicit localhost port and database");
UnpooledDataSource dataSource = new UnpooledDataSource("com.mysql.cj.jdbc.Driver", url,
System.getenv().getOrDefault("IM_TEST_MYSQL_USER", "root"),
System.getenv().getOrDefault("IM_TEST_MYSQL_PASSWORD", ""));
MybatisConfiguration configuration = new MybatisConfiguration();
configuration.setMapUnderscoreToCamelCase(true);
configuration.addInterceptor(new MybatisPlusConfig().mybatisPlusInterceptor());
configuration.setEnvironment(new Environment("session-control-test", new JdbcTransactionFactory(), dataSource));
try (InputStream xml = Resources.getResourceAsStream("mapper/PrivateMessageMapper.xml")) {
new XMLMapperBuilder(xml, configuration, "mapper/PrivateMessageMapper.xml", configuration.getSqlFragments()).parse();
}
session = new MybatisSqlSessionFactoryBuilder().build(configuration).openSession(true);
DatabaseMetaData metadata = session.getConnection().getMetaData();
assertTrue(metadata.getDatabaseMajorVersion() > 5
|| (metadata.getDatabaseMajorVersion() == 5 && metadata.getDatabaseMinorVersion() >= 7),
"Session-control history predicates support MySQL 5.7 and newer");
try (Statement statement = session.getConnection().createStatement()) {
String newDatabase = "codex_im_session_control_" + UUID.randomUUID().toString().replace("-", "");
statement.execute("CREATE DATABASE " + newDatabase + " CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci");
testDatabase = newDatabase;
}
session.getConnection().setCatalog(testDatabase);
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("CREATE TABLE im_user (id BIGINT PRIMARY KEY, user_name VARCHAR(255), "
+ "nick_name VARCHAR(255), head_image VARCHAR(255), head_image_thumb VARCHAR(255))");
// Nullable legacy values exercise SQL three-valued logic instead of losing unrelated messages.
statement.execute("CREATE TABLE im_private_message (id BIGINT PRIMARY KEY, tmp_id VARCHAR(32), "
+ "send_id BIGINT, recv_id BIGINT, content TEXT, type INT, status INT, send_time DATETIME(3))");
}
List<Fixture> fixtures = List.of(
new Fixture(1, -100L, 0, "ordinary message"),
new Fixture(2, -100L, 0, null),
new Fixture(3, -100L, 0, "xxAPP_KICK_SESSION__:underscore wildcard lookalike"),
new Fixture(4, -100L, 0, "__APPxKICK_SESSION__:underscore wildcard lookalike"),
new Fixture(5, -100L, 0, "__APP_KICKxSESSION__:underscore wildcard lookalike"),
new Fixture(6, -100L, 0, "__APP_KICK_SESSIONxx:underscore wildcard lookalike"),
new Fixture(7, -100L, 0, "__app_kick_session__:different case"),
new Fixture(8, -100L, 0, " __APP_KICK_SESSION__:leading space"),
new Fixture(9, -100L, 0, "__APP_KICK_SESSION__"),
new Fixture(10, 7L, 0, "__APP_KICK_SESSION__:ordinary sender"),
new Fixture(11, -100L, 1, "__APP_KICK_SESSION__:different message type"),
new Fixture(12, null, 0, "__APP_KICK_SESSION__:unknown sender"),
new Fixture(13, -100L, null, "__APP_KICK_SESSION__:unknown type"),
new Fixture(14, null, null, null),
new Fixture(15, -101L, 0, "__APP_KICK_SESSION__:other system sender"),
new Fixture(101, -100L, 0, "__APP_KICK_SESSION__:banned"),
new Fixture(102, -100L, 0, "__APP_KICK_SESSION__:"),
new Fixture(103, -100L, 0, "__APP_KICK_SESSION__:line one\nline two"));
try (PreparedStatement statement = session.getConnection().prepareStatement(
"INSERT INTO im_private_message (id,send_id,recv_id,content,type,status,send_time) "
+ "VALUES (?,?,42,?,?,3,'2026-11-03 12:00:00')")) {
for (Fixture fixture : fixtures) {
statement.setLong(1, fixture.id());
statement.setObject(2, fixture.sendId());
statement.setString(3, fixture.content());
statement.setObject(4, fixture.type());
statement.addBatch();
}
statement.executeBatch();
}
mapper = session.getMapper(PrivateMessageMapper.class);
}
@AfterEach
void tearDown() throws Exception {
if (session != null) {
try {
if (testDatabase != null) {
try (Statement statement = session.getConnection().createStatement()) {
statement.execute("DROP DATABASE " + testDatabase);
}
}
} finally {
session.close();
}
}
}
@Test
void lambdaHistoryExcludesControlsBeforeApplyingLimit() {
LambdaQueryWrapper<PrivateMessage> query = new LambdaQueryWrapper<PrivateMessage>()
.eq(PrivateMessage::getRecvId, 42L).orderByDesc(PrivateMessage::getId).last("LIMIT 3");
PrivateMessageSessionControl.applyHistoryFilter(query);
assertEquals(List.of(15L, 14L, 13L), ids(mapper.selectList(query)));
}
@Test
void stringColumnHistoryExcludesControlsBeforeApplyingLimit() {
QueryWrapper<PrivateMessage> query = new QueryWrapper<PrivateMessage>()
.eq("recv_id", 42L).orderByDesc("id").last("LIMIT 3");
PrivateMessageSessionControl.applyHistoryFilter(query);
assertEquals(List.of(15L, 14L, 13L), ids(mapper.selectList(query)));
}
@Test
void bothPredicatesRetainNullsOrdinaryMessagesAndLiteralPrefixLookalikes() {
List<Long> expected = LongStream.rangeClosed(1, 15).boxed().toList();
LambdaQueryWrapper<PrivateMessage> lambda = new LambdaQueryWrapper<PrivateMessage>()
.eq(PrivateMessage::getRecvId, 42L).orderByAsc(PrivateMessage::getId);
PrivateMessageSessionControl.applyHistoryFilter(lambda);
assertEquals(expected, ids(mapper.selectList(lambda)));
QueryWrapper<PrivateMessage> columns = new QueryWrapper<PrivateMessage>()
.eq("recv_id", 42L).orderByAsc("id");
PrivateMessageSessionControl.applyHistoryFilter(columns);
assertEquals(expected, ids(mapper.selectList(columns)));
}
@Test
void adminQueriesKeepControlContentAndFullConversationCount() {
assertEquals(18, mapper.selectList(new QueryWrapper<PrivateMessage>().eq("recv_id", 42L)).size());
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setUserKeyword("-100");
dto.setPeerKeyword("42");
assertEquals(1, mapper.countConversations(dto));
List<ThirdPrivateConversationVO> rows = mapper.selectConversations(dto, 0L);
assertEquals(1, rows.size());
assertEquals(14L, rows.get(0).getMessageCount());
assertEquals(103L, rows.get(0).getLastMessage().getId());
assertEquals("__APP_KICK_SESSION__:line one\nline two", rows.get(0).getLastMessage().getContent());
}
private List<Long> ids(List<PrivateMessage> messages) {
return messages.stream().map(PrivateMessage::getId).toList();
}
private record Fixture(long id, Long sendId, Integer type, String content) {
}
}
@@ -0,0 +1,202 @@
package com.bx.implatform.service;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.config.GlobalConfig;
import com.baomidou.mybatisplus.core.toolkit.GlobalConfigUtils;
import com.bx.imclient.IMClient;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageModerateDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.service.impl.PrivateMessageServiceImpl;
import com.bx.implatform.util.SensitiveFilterUtil;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageModerateVO;
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
import org.apache.ibatis.mapping.Environment;
import org.apache.ibatis.session.SqlSessionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.transaction.SpringManagedTransactionFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.transaction.support.TransactionTemplate;
import java.sql.Connection;
import java.sql.Statement;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
/** Real row locks and transaction rollback, in a fresh localhost-only schema per test. */
class PrivateMessageModerationMySqlTest {
private UnpooledDataSource adminDataSource;
private String testDatabase;
private JdbcTemplate jdbc;
private PrivateMessageMapper mapper;
private PrivateMessageModerationService service;
private IMClient client;
@BeforeEach
void setUp() throws Exception {
String url = System.getenv("IM_TEST_MYSQL_URL");
assumeTrue(url != null && !url.isBlank(), "Set IM_TEST_MYSQL_URL for isolated MySQL verification");
assertTrue(url.matches("jdbc:mysql://(localhost|127\\.0\\.0\\.1):[0-9]+/[a-zA-Z0-9_]+(?:\\?.*)?"));
String user = System.getenv().getOrDefault("IM_TEST_MYSQL_USER", "root");
String password = System.getenv().getOrDefault("IM_TEST_MYSQL_PASSWORD", "");
adminDataSource = new UnpooledDataSource("com.mysql.cj.jdbc.Driver", url, user, password);
String newDatabase = "codex_im_moderation_" + UUID.randomUUID().toString().replace("-", "");
try (Connection connection = adminDataSource.getConnection(); Statement statement = connection.createStatement()) {
assertTrue(connection.getMetaData().getDatabaseMajorVersion() >= 8);
statement.execute("CREATE DATABASE " + newDatabase);
testDatabase = newDatabase;
}
String testUrl = url.replaceFirst("/[a-zA-Z0-9_]+(?=\\?|$)", "/" + testDatabase);
UnpooledDataSource dataSource = new UnpooledDataSource("com.mysql.cj.jdbc.Driver", testUrl, user, password);
jdbc = new JdbcTemplate(dataSource);
jdbc.execute("CREATE TABLE im_private_message (id BIGINT AUTO_INCREMENT PRIMARY KEY, tmp_id VARCHAR(32), "
+ "send_id BIGINT NOT NULL, recv_id BIGINT NOT NULL, content TEXT, type INT NOT NULL, status INT NOT NULL, "
+ "send_time DATETIME(3), INDEX idx_private_pair_time(send_id,recv_id,send_time,id)) ENGINE=InnoDB");
jdbc.update("INSERT INTO im_private_message VALUES (101,'ordinary-client-id',7,42,'original content',0,1,NOW(3))");
MybatisConfiguration configuration = new MybatisConfiguration();
configuration.setMapUnderscoreToCamelCase(true);
configuration.setEnvironment(new Environment("moderation-test", new SpringManagedTransactionFactory(), dataSource));
GlobalConfig globalConfig = new GlobalConfig();
globalConfig.setDbConfig(new GlobalConfig.DbConfig().setIdType(IdType.AUTO));
GlobalConfigUtils.setGlobalConfig(configuration, globalConfig);
configuration.addMapper(PrivateMessageMapper.class);
SqlSessionFactory factory = new MybatisSqlSessionFactoryBuilder().build(configuration);
mapper = new SqlSessionTemplate(factory).getMapper(PrivateMessageMapper.class);
client = mock(IMClient.class);
service = new PrivateMessageModerationService(mapper, client,
new TransactionTemplate(new DataSourceTransactionManager(dataSource)));
}
@AfterEach
void tearDown() throws Exception {
if (testDatabase != null) {
try (Connection connection = adminDataSource.getConnection(); Statement statement = connection.createStatement()) {
statement.execute("DROP DATABASE " + testDatabase);
}
}
}
@Test
void concurrentRequestsLockSameSourceAndCreateOnlyOneCommittedRecall() throws Exception {
doAnswer(invocation -> {
// JdbcTemplate outside the completed service transaction observes committed rows.
assertEquals(2, jdbc.queryForObject("SELECT status FROM im_private_message WHERE id=101", Integer.class));
assertEquals(1, jdbc.queryForObject("SELECT COUNT(*) FROM im_private_message WHERE tmp_id='admin-m:101'", Integer.class));
return null;
}).when(client).sendPrivateMessage(any());
ExecutorService executor = Executors.newFixedThreadPool(2);
CountDownLatch ready = new CountDownLatch(2);
CountDownLatch start = new CountDownLatch(1);
try {
java.util.concurrent.Callable<ThirdPrivateMessageModerateVO> task = () -> {
ready.countDown();
assertTrue(start.await(5, TimeUnit.SECONDS));
return service.moderate(request());
};
Future<ThirdPrivateMessageModerateVO> first = executor.submit(task);
Future<ThirdPrivateMessageModerateVO> second = executor.submit(task);
assertTrue(ready.await(5, TimeUnit.SECONDS));
start.countDown();
ThirdPrivateMessageModerateVO a = first.get(15, TimeUnit.SECONDS);
ThirdPrivateMessageModerateVO b = second.get(15, TimeUnit.SECONDS);
assertEquals(a.getRecallMessageId(), b.getRecallMessageId());
assertNotEquals(a.getAlreadyHidden(), b.getAlreadyHidden());
assertTrue(a.getNotificationSent());
assertTrue(b.getNotificationSent());
assertEquals(2, jdbc.queryForObject("SELECT COUNT(*) FROM im_private_message", Integer.class));
assertEquals("original content", jdbc.queryForObject("SELECT content FROM im_private_message WHERE id=101", String.class));
verify(client, times(4)).sendPrivateMessage(any());
} finally {
executor.shutdownNow();
assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
}
}
@Test
void databaseFailureRollsBackSourceStatusAndCreatesNoRecall() {
jdbc.execute("ALTER TABLE im_private_message ADD CONSTRAINT no_recall_fixture CHECK(type <> 10)");
assertThrows(RuntimeException.class, () -> service.moderate(request()));
assertEquals(1, jdbc.queryForObject("SELECT status FROM im_private_message WHERE id=101", Integer.class));
assertEquals(1, jdbc.queryForObject("SELECT COUNT(*) FROM im_private_message", Integer.class));
verifyNoInteractions(client);
}
@Test
void notificationFailureRetainsDurableTombstoneAndCanBeRetried() {
doThrow(new IllegalStateException("transport unavailable")).when(client).sendPrivateMessage(any());
ThirdPrivateMessageModerateVO first = service.moderate(request());
assertFalse(first.getNotificationSent());
assertEquals(2, jdbc.queryForObject("SELECT status FROM im_private_message WHERE id=101", Integer.class));
reset(client);
ThirdPrivateMessageModerateVO retry = service.moderate(request());
assertTrue(retry.getAlreadyHidden());
assertTrue(retry.getNotificationSent());
assertEquals(first.getRecallMessageId(), retry.getRecallMessageId());
assertEquals(1, jdbc.queryForObject("SELECT COUNT(*) FROM im_private_message WHERE type=10", Integer.class));
}
@Test
void offlineQueryReplaysOldAdminRecallBeyondCursorWithoutExposingOriginalOrOtherUsers() {
ThirdPrivateMessageModerateVO result = service.moderate(request());
jdbc.update("UPDATE im_private_message SET send_time='2020-01-01 00:00:00' WHERE id=?", result.getRecallMessageId());
jdbc.update("INSERT INTO im_private_message (tmp_id,send_id,recv_id,content,type,status,send_time) VALUES "
+ "('admin-m:500',8,9,'500',10,3,'2020-01-01'),"
+ "('ADMIN-M:501',7,42,'501',10,3,'2020-01-01'),"
+ "('normal-recall',7,42,'502',10,3,'2020-01-01')");
PrivateMessageServiceImpl history = new PrivateMessageServiceImpl(mock(FriendService.class), client,
mock(SensitiveFilterUtil.class), service);
LambdaQueryWrapper<PrivateMessage> query = ReflectionTestUtils.invokeMethod(history, "buildOfflineQuery", 42L, 999999L);
List<PrivateMessage> messages = mapper.selectList(query);
assertEquals(1, messages.size());
assertEquals(result.getRecallMessageId(), messages.get(0).getId());
assertEquals(101L, service.toMobileVO(messages.get(0)).getId());
assertEquals("101", service.toMobileVO(messages.get(0)).getContent());
assertEquals(3, service.toMobileVO(messages.get(0)).getStatus());
}
@Test
void staleOfflineSnapshotIsRecheckedAfterModeration() {
List<PrivateMessage> stale = mapper.selectList(null);
assertEquals("original content", stale.get(0).getContent());
service.moderate(request());
assertTrue(service.refreshVisibleMessages(stale).isEmpty());
}
@Test
void caseInsensitiveMarkerLookalikeCannotBlockOrReplaceAdministratorRecall() {
jdbc.update("INSERT INTO im_private_message (tmp_id,send_id,recv_id,content,type,status,send_time) "
+ "VALUES ('ADMIN-M:101',7,42,'101',10,3,NOW(3))");
ThirdPrivateMessageModerateVO receipt = service.moderate(request());
assertTrue(receipt.getNotificationSent());
assertEquals("admin-m:101", jdbc.queryForObject("SELECT tmp_id FROM im_private_message WHERE id=?",
String.class, receipt.getRecallMessageId()));
assertEquals(2, jdbc.queryForObject("SELECT COUNT(*) FROM im_private_message WHERE type=10", Integer.class));
}
private ThirdPrivateMessageModerateDTO request() {
ThirdPrivateMessageModerateDTO dto = new ThirdPrivateMessageModerateDTO();
dto.setMessageId(101L);
dto.setOperatorId(42L);
dto.setReason("integration fixture");
return dto;
}
}
@@ -0,0 +1,131 @@
package com.bx.implatform.service;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.bx.imclient.IMClient;
import com.bx.imcommon.enums.IMSendCode;
import com.bx.imcommon.model.IMSendResult;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.listener.PrivateMessageListener;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.service.impl.PrivateMessageServiceImpl;
import com.bx.implatform.session.UserSession;
import com.bx.implatform.util.SensitiveFilterUtil;
import com.bx.implatform.vo.PrivateMessageVO;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.springframework.mock.web.MockHttpServletRequest;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
class PrivateMessageModerationReadPathTest {
private PrivateMessageMapper mapper;
private PrivateMessageModerationService moderation;
private PrivateMessageServiceImpl service;
@BeforeEach
void setUp() {
mapper = mock(PrivateMessageMapper.class);
moderation = mock(PrivateMessageModerationService.class);
service = new PrivateMessageServiceImpl(mock(FriendService.class), mock(IMClient.class),
mock(SensitiveFilterUtil.class), moderation);
ReflectionTestUtils.setField(service, "baseMapper", mapper);
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "moderation-read-test"), PrivateMessage.class);
UserSession session = new UserSession();
session.setUserId(42L);
session.setTerminal(1);
MockHttpServletRequest request = new MockHttpServletRequest();
request.setAttribute("session", session);
RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
}
@AfterEach
void tearDown() {
RequestContextHolder.resetRequestAttributes();
}
@Test
@SuppressWarnings({"unchecked", "rawtypes"})
void pendingOfflineAcknowledgementCannotOverwriteConcurrentlyHiddenStatus() {
PrivateMessage snapshot = new PrivateMessage();
snapshot.setId(101L);
snapshot.setRecvId(42L);
snapshot.setStatus(0);
when(mapper.selectList(any(Wrapper.class))).thenReturn(List.of(snapshot));
// The re-read observes that moderation has hidden the source since the initial query.
when(moderation.refreshVisibleMessages(anyList())).thenReturn(Collections.emptyList());
assertTrue(service.loadOfflineMessage(0L).isEmpty());
ArgumentCaptor<Wrapper<PrivateMessage>> update = ArgumentCaptor.forClass(Wrapper.class);
verify(mapper).update(isNull(), update.capture());
String predicate = update.getValue().getSqlSegment();
assertTrue(predicate.contains("status ="), predicate);
assertTrue(((com.baomidou.mybatisplus.core.conditions.AbstractWrapper) update.getValue())
.getParamNameValuePairs().containsValue(0));
InOrder order = inOrder(mapper, moderation);
order.verify(mapper).selectList(any(Wrapper.class));
order.verify(mapper).update(isNull(), any(Wrapper.class));
order.verify(moderation).refreshVisibleMessages(anyList());
}
@Test
void regularUserCannotRecallAdministratorCompensationRecord() {
PrivateMessage event = new PrivateMessage();
event.setId(202L);
event.setType(10);
event.setContent("101");
event.setTmpId("admin-m:101");
when(mapper.selectById(202L)).thenReturn(event);
assertThrows(GlobalException.class, () -> service.recallMessage(202L));
verify(mapper, never()).updateById(any(PrivateMessage.class));
verify(mapper, never()).insert(any(PrivateMessage.class));
}
@Test
@SuppressWarnings("unchecked")
void delayedSourceAcknowledgementsReassertModerationOnceAndIgnoreRecallCallbacks() {
PrivateMessageService privateService = mock(PrivateMessageService.class);
PrivateMessageListener listener = new PrivateMessageListener();
ReflectionTestUtils.setField(listener, "privateMessageService", privateService);
ReflectionTestUtils.setField(listener, "moderationService", moderation);
IMSendResult<PrivateMessageVO> source = delivered(101L, 0);
listener.process(Arrays.asList(source, source, delivered(101L, 10), delivered(-1L, 0), null));
InOrder order = inOrder(privateService, moderation);
order.verify(privateService).update(any(Wrapper.class));
order.verify(moderation).replayIfModerated(Set.of(101L));
verifyNoMoreInteractions(moderation);
}
private IMSendResult<PrivateMessageVO> delivered(Long id, int type) {
PrivateMessageVO message = new PrivateMessageVO();
message.setId(id);
message.setType(type);
message.setSendId(7L);
message.setRecvId(42L);
IMSendResult<PrivateMessageVO> result = new IMSendResult<>();
result.setCode(IMSendCode.SUCCESS.code());
result.setSender(new IMUserInfo(7L, 1));
result.setReceiver(new IMUserInfo(42L, 1));
result.setData(message);
return result;
}
}
@@ -0,0 +1,257 @@
package com.bx.implatform.service;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.bx.imclient.IMClient;
import com.bx.imcommon.enums.IMTerminalType;
import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageModerateDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.vo.PrivateMessageVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessageModerateVO;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.SimpleTransactionStatus;
import org.springframework.transaction.support.TransactionTemplate;
import java.util.Date;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
@ExtendWith(MockitoExtension.class)
class PrivateMessageModerationServiceTest {
@Mock
private PrivateMessageMapper mapper;
@Mock
private IMClient client;
@Mock
private PlatformTransactionManager transactionManager;
private PrivateMessageModerationService service;
@BeforeEach
void setUp() {
lenient().when(transactionManager.getTransaction(any())).thenReturn(new SimpleTransactionStatus());
service = new PrivateMessageModerationService(mapper, client, new TransactionTemplate(transactionManager));
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "moderation-test"), PrivateMessage.class);
}
@Test
@SuppressWarnings({"rawtypes", "unchecked"})
void commitsHiddenStatusAndDurableEventBeforeNotifyingBothParticipants() {
when(mapper.selectForUpdate(101L)).thenReturn(source(0, 1));
when(mapper.updateById(any(PrivateMessage.class))).thenReturn(1);
when(mapper.insert(any(PrivateMessage.class))).thenAnswer(invocation -> {
((PrivateMessage) invocation.getArgument(0)).setId(202L);
return 1;
});
ThirdPrivateMessageModerateVO result = service.moderate(request());
assertEquals(101L, result.getMessageId());
assertEquals(2, result.getStatus());
assertEquals(202L, result.getRecallMessageId());
assertFalse(result.getAlreadyHidden());
assertTrue(result.getNotificationSent());
InOrder order = inOrder(mapper, transactionManager, client);
order.verify(transactionManager).getTransaction(any());
order.verify(mapper).selectForUpdate(101L);
order.verify(mapper).updateById(any(PrivateMessage.class));
order.verify(mapper).selectModerationEvent("admin-m:101", 7L, 42L);
order.verify(mapper).insert(any(PrivateMessage.class));
order.verify(transactionManager).commit(any());
order.verify(client, times(2)).sendPrivateMessage(any());
ArgumentCaptor<PrivateMessage> update = ArgumentCaptor.forClass(PrivateMessage.class);
verify(mapper).updateById(update.capture());
assertEquals(2, update.getValue().getStatus());
assertNull(update.getValue().getContent(), "Moderation must retain the original content in storage");
ArgumentCaptor<PrivateMessage> inserted = ArgumentCaptor.forClass(PrivateMessage.class);
verify(mapper).insert(inserted.capture());
assertEquals("admin-m:101", inserted.getValue().getTmpId());
assertEquals("101", inserted.getValue().getContent());
assertEquals(10, inserted.getValue().getType());
assertEquals(3, inserted.getValue().getStatus());
ArgumentCaptor<TransactionDefinition> transaction = ArgumentCaptor.forClass(TransactionDefinition.class);
verify(transactionManager).getTransaction(transaction.capture());
assertEquals(TransactionDefinition.PROPAGATION_REQUIRES_NEW, transaction.getValue().getPropagationBehavior());
ArgumentCaptor<IMPrivateMessage<PrivateMessageVO>> pushes = ArgumentCaptor.forClass(IMPrivateMessage.class);
verify(client, times(2)).sendPrivateMessage(pushes.capture());
assertEquals(List.of(7L, 42L), pushes.getAllValues().stream().map(IMPrivateMessage::getRecvId).toList());
for (IMPrivateMessage<PrivateMessageVO> push : pushes.getAllValues()) {
assertEquals(IMTerminalType.codes(), push.getRecvTerminals());
assertFalse(push.getSendToSelf());
assertFalse(push.getSendResult());
assertEquals(101L, push.getData().getId());
assertEquals("101", push.getData().getContent());
assertEquals(3, push.getData().getStatus());
assertEquals(7L, push.getData().getSendId());
assertEquals(42L, push.getData().getRecvId());
}
}
@Test
void repeatModerationReusesStoredEventAndRetriesNotification() {
when(mapper.selectForUpdate(101L)).thenReturn(source(0, 2));
when(mapper.selectModerationEvent("admin-m:101", 7L, 42L)).thenReturn(event());
ThirdPrivateMessageModerateVO result = service.moderate(request());
assertTrue(result.getAlreadyHidden());
assertTrue(result.getNotificationSent());
assertEquals(202L, result.getRecallMessageId());
verify(mapper, never()).updateById(any(PrivateMessage.class));
verify(mapper, never()).insert(any(PrivateMessage.class));
verify(client, times(2)).sendPrivateMessage(any());
}
@Test
void notificationFailureDoesNotUndoCommittedModerationOrSkipSecondParticipant() {
when(mapper.selectForUpdate(101L)).thenReturn(source(0, 2));
when(mapper.selectModerationEvent("admin-m:101", 7L, 42L)).thenReturn(event());
doThrow(new IllegalStateException("push unavailable")).doNothing().when(client).sendPrivateMessage(any());
ThirdPrivateMessageModerateVO result = service.moderate(request());
assertFalse(result.getNotificationSent());
assertEquals(2, result.getStatus());
verify(transactionManager).commit(any());
verify(transactionManager, never()).rollback(any());
verify(client, times(2)).sendPrivateMessage(any());
}
@Test
void failedEventInsertRollsBackAndNeverNotifies() {
when(mapper.selectForUpdate(101L)).thenReturn(source(0, 1));
when(mapper.updateById(any(PrivateMessage.class))).thenReturn(1);
when(mapper.insert(any(PrivateMessage.class))).thenReturn(0);
assertThrows(GlobalException.class, () -> service.moderate(request()));
verify(transactionManager).rollback(any());
verify(transactionManager, never()).commit(any());
verifyNoInteractions(client);
}
@Test
void failedCommitNeverPublishesAnUncommittedRecall() {
when(mapper.selectForUpdate(101L)).thenReturn(source(0, 2));
when(mapper.selectModerationEvent("admin-m:101", 7L, 42L)).thenReturn(event());
doThrow(new IllegalStateException("commit unavailable")).when(transactionManager).commit(any());
assertThrows(IllegalStateException.class, () -> service.moderate(request()));
verifyNoInteractions(client);
}
@ParameterizedTest
@ValueSource(ints = {-1, 10, 11, 12, 21, 30, 50})
void refusesControlMessagesWithoutWritingOrNotifying(int type) {
when(mapper.selectForUpdate(101L)).thenReturn(source(type, 1));
assertThrows(GlobalException.class, () -> service.moderate(request()));
verify(mapper, never()).updateById(any(PrivateMessage.class));
verify(mapper, never()).insert(any(PrivateMessage.class));
verifyNoInteractions(client);
}
@Test
void missingSourceCannotCreateAnOrphanRecall() {
assertThrows(GlobalException.class, () -> service.moderate(request()));
verify(mapper, never()).insert(any(PrivateMessage.class));
verifyNoInteractions(client);
}
@Test
void wireProjectionUsesSourceIdWithoutMutatingStoredEvent() {
PrivateMessage event = event();
PrivateMessageVO wire = service.toMobileVO(event);
assertEquals(101L, wire.getId());
assertEquals("101", wire.getContent());
assertEquals(3, wire.getStatus());
assertEquals(202L, event.getId());
assertEquals("admin-m:101", event.getTmpId());
}
@Test
void delayedDeliveryReplaysOnlyExistingAdminModerationEvents() {
when(mapper.selectList(any(Wrapper.class))).thenReturn(List.of(source(0, 2)));
when(mapper.selectModerationEvent("admin-m:101", 7L, 42L)).thenReturn(event());
service.replayIfModerated(101L);
verify(client, times(2)).sendPrivateMessage(any());
verify(mapper, never()).insert(any(PrivateMessage.class));
}
@Test
@SuppressWarnings({"rawtypes", "unchecked"})
void ordinaryDeliveryBatchChecksHiddenStatusOnceWithoutFetchingBodiesOrEvents() {
when(mapper.selectList(any(Wrapper.class))).thenReturn(List.of());
service.replayIfModerated(List.of(101L, 102L));
ArgumentCaptor<Wrapper<PrivateMessage>> query = ArgumentCaptor.forClass(Wrapper.class);
verify(mapper).selectList(query.capture());
String selected = query.getValue().getSqlSelect().toLowerCase();
assertTrue(selected.contains("id"));
assertTrue(selected.contains("status"));
assertFalse(selected.contains("content"), selected);
assertFalse(selected.contains("tmp_id"), selected);
assertTrue(query.getValue().getSqlSegment().toLowerCase().contains("status"));
verify(mapper, never()).selectById(anyLong());
verify(mapper, never()).selectModerationEvent(anyString(), anyLong(), anyLong());
verifyNoInteractions(client);
}
@Test
void refreshedOfflineSnapshotDropsHiddenAndDeletedRowsButKeepsRecallEvent() {
PrivateMessage stale = source(0, 1);
PrivateMessage deleted = source(0, 1);
deleted.setId(102L);
PrivateMessage current = source(0, 2);
when(mapper.selectBatchIds(anyCollection())).thenReturn(List.of(current, event()));
List<PrivateMessage> visible = service.refreshVisibleMessages(List.of(stale, deleted, event()));
assertEquals(1, visible.size());
assertEquals(202L, visible.get(0).getId());
assertEquals(10, visible.get(0).getType());
}
@Test
void reservedMarkerCannotBeUsedByOrdinarySenders() {
assertThrows(GlobalException.class, () -> PrivateMessageModerationService.rejectReservedTmpId("admin-m:101"));
assertDoesNotThrow(() -> PrivateMessageModerationService.rejectReservedTmpId("normal-client-id"));
assertDoesNotThrow(() -> PrivateMessageModerationService.rejectReservedTmpId(null));
}
private ThirdPrivateMessageModerateDTO request() {
ThirdPrivateMessageModerateDTO dto = new ThirdPrivateMessageModerateDTO();
dto.setMessageId(101L);
dto.setOperatorId(42L);
dto.setReason("违规内容");
return dto;
}
private PrivateMessage source(int type, int status) {
PrivateMessage message = new PrivateMessage();
message.setId(101L);
message.setSendId(7L);
message.setRecvId(42L);
message.setType(type);
message.setStatus(status);
message.setContent("original private content");
message.setSendTime(new Date());
return message;
}
private PrivateMessage event() {
PrivateMessage event = source(10, 3);
event.setId(202L);
event.setTmpId("admin-m:101");
event.setContent("101");
return event;
}
}
@@ -0,0 +1,177 @@
package com.bx.implatform.service;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.conditions.AbstractWrapper;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.bx.imclient.IMClient;
import com.bx.imcommon.enums.IMTerminalType;
import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.implatform.dto.PrivateMessageDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.service.impl.PrivateMessageServiceImpl;
import com.bx.implatform.session.UserSession;
import com.bx.implatform.util.BeanUtils;
import com.bx.implatform.util.PrivateMessageSessionControl;
import com.bx.implatform.util.SensitiveFilterUtil;
import com.bx.implatform.vo.PrivateMessageVO;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.mock.web.MockHttpServletRequest;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
class PrivateMessageSessionControlReadPathTest {
private PrivateMessageMapper mapper;
private PrivateMessageModerationService moderation;
private FriendService friends;
private IMClient imClient;
private PrivateMessageServiceImpl service;
@BeforeEach
void setUp() {
mapper = mock(PrivateMessageMapper.class);
moderation = mock(PrivateMessageModerationService.class);
friends = mock(FriendService.class);
imClient = mock(IMClient.class);
service = new PrivateMessageServiceImpl(friends, imClient, mock(SensitiveFilterUtil.class), moderation);
ReflectionTestUtils.setField(service, "baseMapper", mapper);
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "session-control-read-test"),
PrivateMessage.class);
UserSession session = new UserSession();
session.setUserId(42L);
session.setTerminal(IMTerminalType.WEB.code());
MockHttpServletRequest request = new MockHttpServletRequest();
request.setAttribute("session", session);
RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
when(moderation.refreshVisibleMessages(anyList())).thenAnswer(call -> call.getArgument(0));
when(moderation.toMobileVO(any(PrivateMessage.class))).thenAnswer(call ->
BeanUtils.copyProperties(call.getArgument(0), PrivateMessageVO.class));
}
@AfterEach
void tearDown() {
RequestContextHolder.resetRequestAttributes();
}
@Test
void ordinaryUsersCannotInjectKickProtocolEvenWithNonTextType() {
PrivateMessageDTO dto = new PrivateMessageDTO();
dto.setRecvId(7L);
dto.setType(1);
dto.setContent(PrivateMessageSessionControl.KICK_PREFIX + "伪造踢登录");
assertThrows(GlobalException.class, () -> service.sendMessage(dto));
verifyNoInteractions(friends, mapper, imClient);
}
@Test
@SuppressWarnings({"unchecked", "rawtypes"})
void offlineRestDoesNotReplayLegacyDisableKickAfterReenable() {
PrivateMessage kick = legacyKick();
PrivateMessage normal = message(102L, 7L, "普通消息");
PrivateMessage nullContent = message(103L, -100L, null);
when(mapper.selectList(any(Wrapper.class))).thenReturn(List.of(kick, normal, nullContent));
List<PrivateMessageVO> result = service.loadOfflineMessage(0L);
assertEquals(List.of(102L, 103L), result.stream().map(PrivateMessageVO::getId).toList());
verify(moderation, never()).toMobileVO(kick);
ArgumentCaptor<Wrapper<PrivateMessage>> query = ArgumentCaptor.forClass(Wrapper.class);
verify(mapper).selectList(query.capture());
assertLegacyKickSqlFilter((AbstractWrapper) query.getValue());
}
@Test
@SuppressWarnings({"unchecked", "rawtypes"})
void historyFiltersBeforePaginationAndRechecksRefreshedRows() {
PrivateMessage normal = message(102L, -100L, "普通消息");
PrivateMessage kick = legacyKick();
when(mapper.selectList(any(Wrapper.class))).thenReturn(List.of(normal));
// A fresh-state read must not reintroduce a control record excluded by the initial query.
when(moderation.refreshVisibleMessages(anyList())).thenReturn(List.of(normal, kick));
List<PrivateMessageVO> result = service.findHistoryMessage(-100L, 1L, 10L);
assertEquals(List.of(102L), result.stream().map(PrivateMessageVO::getId).toList());
verify(moderation, never()).toMobileVO(kick);
ArgumentCaptor<Wrapper<PrivateMessage>> query = ArgumentCaptor.forClass(Wrapper.class);
verify(mapper).selectList(query.capture());
AbstractWrapper wrapper = (AbstractWrapper) query.getValue();
assertLegacyKickSqlFilter(wrapper);
String sql = wrapper.getSqlSegment();
assertTrue(sql.contains("limit 0,10"), sql);
assertTrue(sql.indexOf("NOT (") < sql.indexOf("limit 0,10"), sql);
}
@Test
@SuppressWarnings("unchecked")
void offlineWebSocketDoesNotPushHistoricalKickButCompletesLoading() throws Exception {
PrivateMessage kick = legacyKick();
PrivateMessage normal = message(102L, 7L, "普通消息");
when(mapper.selectList(any(Wrapper.class))).thenReturn(List.of(normal));
when(moderation.refreshVisibleMessages(anyList())).thenReturn(List.of(kick, normal));
when(imClient.isOnline(eq(42L), any(IMTerminalType.class))).thenReturn(true);
List<PrivateMessageVO> pushed = Collections.synchronizedList(new ArrayList<>());
CountDownLatch finished = new CountDownLatch(1);
doAnswer(call -> {
IMPrivateMessage<PrivateMessageVO> envelope = call.getArgument(0);
PrivateMessageVO data = envelope.getData();
pushed.add(data);
if (Integer.valueOf(30).equals(data.getType()) && "false".equals(data.getContent())) {
finished.countDown();
}
return null;
}).when(imClient).sendPrivateMessage(any(IMPrivateMessage.class));
service.pullOfflineMessage(0L);
assertTrue(finished.await(5, TimeUnit.SECONDS), "Offline push must finish loading");
assertEquals(3, pushed.size());
assertEquals(List.of(102L), pushed.stream().map(PrivateMessageVO::getId)
.filter(java.util.Objects::nonNull).toList());
assertTrue(pushed.stream().noneMatch(m -> PrivateMessageSessionControl.hasKickPrefix(m.getContent())));
verify(moderation, never()).toMobileVO(kick);
}
private void assertLegacyKickSqlFilter(AbstractWrapper<?, ?, ?> wrapper) {
String sql = wrapper.getSqlSegment();
assertTrue(sql.contains("NOT (COALESCE(send_id"), sql);
assertTrue(sql.contains("BINARY LEFT(COALESCE(content"), sql);
assertFalse(sql.contains("LIKE"), sql);
assertTrue(wrapper.getParamNameValuePairs().containsValue(PrivateMessageSessionControl.KICK_PREFIX));
assertTrue(wrapper.getParamNameValuePairs().containsValue(-100L));
}
private PrivateMessage legacyKick() {
return message(101L, -100L, PrivateMessageSessionControl.KICK_PREFIX + "账号已被禁用");
}
private PrivateMessage message(Long id, Long senderId, String content) {
PrivateMessage message = new PrivateMessage();
message.setId(id);
message.setSendId(senderId);
message.setRecvId(42L);
message.setType(0);
message.setStatus(3);
message.setContent(content);
return message;
}
}
@@ -0,0 +1,129 @@
package com.bx.implatform.service.third.impl;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.bx.imclient.IMClient;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateConversationDTO;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.mapper.UserMapper;
import com.bx.implatform.service.FriendService;
import com.bx.implatform.util.SensitiveFilterUtil;
import com.bx.implatform.vo.third.ThirdTableVO;
import com.bx.implatform.vo.third.privatemsg.ThirdPrivateConversationVO;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.test.util.ReflectionTestUtils;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
@ExtendWith(MockitoExtension.class)
class ThirdPrivateConversationServiceTest {
@Mock
private PrivateMessageMapper mapper;
@Mock
private UserMapper userMapper;
private ThirdPrivateMsgServiceImpl service;
@BeforeEach
void setUp() {
service = new ThirdPrivateMsgServiceImpl(userMapper, mapper, mock(IMClient.class),
mock(FriendService.class), mock(SensitiveFilterUtil.class));
ReflectionTestUtils.setField(service, "baseMapper", mapper);
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "test"), PrivateMessage.class);
}
@Test
void noMatchingConversationsReturnsEmptyRowsAndSkipsPageQuery() {
ThirdTableVO<ThirdPrivateConversationVO> page = service.conversationPage(new ThirdPrivateConversationDTO());
assertEquals(0, page.getTotal());
assertNotNull(page.getRows());
assertTrue(page.getRows().isEmpty());
verify(mapper, never()).selectConversations(any(), anyLong());
}
@Test
void pageBeyondEndRetainsTotalAndReturnsEmptyRows() {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setPageNum(3);
dto.setPageSize(10);
when(mapper.countConversations(any())).thenReturn(12L);
ThirdTableVO<ThirdPrivateConversationVO> page = service.conversationPage(dto);
assertEquals(12, page.getTotal());
assertTrue(page.getRows().isEmpty());
verify(mapper, never()).selectConversations(any(), anyLong());
}
@Test
void trimsParticipantFiltersAndUsesRequestedOffset() {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setPageNum(2);
dto.setPageSize(10);
dto.setUserKeyword(" Alice ");
dto.setPeerKeyword(" 2 ");
when(mapper.countConversations(any())).thenReturn(12L);
when(mapper.selectConversations(any(), eq(10L))).thenReturn(Collections.emptyList());
service.conversationPage(dto);
ArgumentCaptor<ThirdPrivateConversationDTO> captor = ArgumentCaptor.forClass(ThirdPrivateConversationDTO.class);
verify(mapper).selectConversations(captor.capture(), eq(10L));
assertEquals("Alice", captor.getValue().getUserKeyword());
assertEquals("2", captor.getValue().getPeerKeyword());
}
@Test
void rejectsMalformedOrReversedConversationDatesBeforeReadingDatabase() {
for (List<Date> dates : List.of(List.of(new Date(10)),
List.of(new Date(20), new Date(10)), Arrays.asList(new Date(10), null))) {
ThirdPrivateConversationDTO dto = new ThirdPrivateConversationDTO();
dto.setLastSendTimeList(dates);
assertThrows(GlobalException.class, () -> service.conversationPage(dto));
}
verifyNoInteractions(mapper);
}
@Test
void malformedDetailDateRangeDoesNotReachMapper() {
ThirdPrivateMessageDTO dto = new ThirdPrivateMessageDTO();
dto.setPageNum(1);
dto.setPageSize(10);
dto.setSendTimeList(List.of(new Date(10)));
assertThrows(GlobalException.class, () -> service.page(dto));
verifyNoInteractions(mapper);
}
@Test
@SuppressWarnings({"rawtypes", "unchecked"})
void bilateralDetailUsesTimeThenIdOrderingAndAlwaysReturnsArray() {
ThirdPrivateMessageDTO dto = new ThirdPrivateMessageDTO();
dto.setPageNum(1);
dto.setPageSize(10);
dto.setSendId(1L);
dto.setRecvId(2L);
dto.setSendTimeList(Collections.emptyList());
when(mapper.selectPage(any(Page.class), any())).thenReturn(new Page<PrivateMessage>(1, 10));
ThirdTableVO<?> page = service.page(dto);
assertNotNull(page.getRows());
assertTrue(page.getRows().isEmpty());
ArgumentCaptor<Wrapper<PrivateMessage>> captor = ArgumentCaptor.forClass(Wrapper.class);
verify(mapper).selectPage(any(Page.class), captor.capture());
String sql = captor.getValue().getSqlSegment().toLowerCase();
assertTrue(sql.contains("order by send_time desc,id desc"), sql);
assertTrue(sql.contains(" or "), sql);
}
}
@@ -0,0 +1,150 @@
package com.bx.implatform.service.third.impl;
import com.bx.imclient.IMClient;
import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.implatform.contant.Constant;
import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageSendDTO;
import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.enums.MessageStatus;
import com.bx.implatform.enums.MessageType;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.PrivateMessageMapper;
import com.bx.implatform.mapper.UserMapper;
import com.bx.implatform.service.FriendService;
import com.bx.implatform.util.PrivateMessageSessionControl;
import com.bx.implatform.util.SensitiveFilterUtil;
import com.bx.implatform.vo.PrivateMessageVO;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.test.util.ReflectionTestUtils;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
class ThirdPrivateSessionControlServiceTest {
private PrivateMessageMapper mapper;
private UserMapper userMapper;
private IMClient imClient;
private FriendService friends;
private SensitiveFilterUtil sensitiveFilter;
private ThirdPrivateMsgServiceImpl service;
@BeforeEach
void setUp() {
mapper = mock(PrivateMessageMapper.class);
userMapper = mock(UserMapper.class);
imClient = mock(IMClient.class);
friends = mock(FriendService.class);
sensitiveFilter = mock(SensitiveFilterUtil.class);
service = new ThirdPrivateMsgServiceImpl(userMapper, mapper, imClient, friends, sensitiveFilter);
ReflectionTestUtils.setField(service, "baseMapper", mapper);
}
@Test
@SuppressWarnings({"rawtypes", "unchecked"})
void kickUsesExistingPrivateEnvelopeWithoutPersistingOrRequestingDeliveryCallback() {
ThirdPrivateMessageSendDTO dto = kick();
PrivateMessageVO result = service.sendMessage(dto);
assertNotNull(result);
assertEquals(0L, result.getId());
assertEquals(dto.getSendId(), result.getSendId());
assertEquals(dto.getRecvId(), result.getRecvId());
assertEquals(dto.getContent(), result.getContent());
assertEquals(dto.getTmpId(), result.getTmpId());
assertEquals(MessageType.TEXT.code(), result.getType());
assertEquals(MessageStatus.READED.code(), result.getStatus());
assertNotNull(result.getSendTime());
ArgumentCaptor<IMPrivateMessage> push = ArgumentCaptor.forClass(IMPrivateMessage.class);
verify(imClient).sendPrivateMessage(push.capture());
assertSame(result, push.getValue().getData());
assertEquals(dto.getSendId(), push.getValue().getSender().getId());
assertEquals(dto.getTerminal(), push.getValue().getSender().getTerminal());
assertEquals(dto.getRecvId(), push.getValue().getRecvId());
assertFalse(push.getValue().getSendToSelf());
assertFalse(push.getValue().getSendResult());
verifyNoInteractions(mapper, userMapper, friends, sensitiveFilter);
}
@Test
void forgedSenderOrNonTextKickContentIsRejectedBeforeAnySideEffect() {
ThirdPrivateMessageSendDTO ordinarySender = kick();
ordinarySender.setSendId(7L);
assertThrows(GlobalException.class, () -> service.sendMessage(ordinarySender));
ThirdPrivateMessageSendDTO nonText = kick();
nonText.setType(MessageType.IMAGE.code());
assertThrows(GlobalException.class, () -> service.sendMessage(nonText));
ThirdPrivateMessageSendDTO noType = kick();
noType.setType(null);
assertThrows(GlobalException.class, () -> service.sendMessage(noType));
verifyNoInteractions(mapper, userMapper, friends, sensitiveFilter, imClient);
}
@Test
void failedRealtimePushDoesNotLeaveAnOfflineKickToReplay() {
doThrow(new IllegalStateException("temporary transport failure")).when(imClient).sendPrivateMessage(any());
assertThrows(IllegalStateException.class, () -> service.sendMessage(kick()));
verifyNoInteractions(mapper, userMapper, friends, sensitiveFilter);
}
@Test
@SuppressWarnings({"rawtypes", "unchecked"})
void ordinaryTextKeepsPersistedDeliveryAndFriendshipChecks() {
ThirdPrivateMessageSendDTO dto = kick();
dto.setSendId(7L);
dto.setContent("ordinary chat text");
when(friends.isFriend(7L, dto.getRecvId())).thenReturn(true);
when(sensitiveFilter.filter(dto.getContent())).thenReturn("filtered chat text");
when(mapper.insert(any(PrivateMessage.class))).thenAnswer(invocation -> {
invocation.<PrivateMessage>getArgument(0).setId(101L);
return 1;
});
PrivateMessageVO result = service.sendMessage(dto);
assertEquals(101L, result.getId());
assertEquals("filtered chat text", result.getContent());
assertEquals(MessageStatus.PENDING.code(), result.getStatus());
verify(friends).isFriend(7L, dto.getRecvId());
verify(mapper).insert(any(PrivateMessage.class));
ArgumentCaptor<IMPrivateMessage> push = ArgumentCaptor.forClass(IMPrivateMessage.class);
verify(imClient).sendPrivateMessage(push.capture());
assertTrue(push.getValue().getSendToSelf());
assertTrue(push.getValue().getSendResult());
}
@Test
void recognitionUsesExactPrefixAndTrustedSenderAndType() {
PrivateMessage legacy = new PrivateMessage();
legacy.setSendId(Constant.COM_MESSAGE_USER_ID);
legacy.setType(MessageType.TEXT.code());
legacy.setContent(PrivateMessageSessionControl.KICK_PREFIX + "账号已被禁用");
assertTrue(PrivateMessageSessionControl.isKickMessage(legacy));
legacy.setSendId(7L);
assertFalse(PrivateMessageSessionControl.isKickMessage(legacy));
legacy.setSendId(Constant.COM_MESSAGE_USER_ID);
legacy.setType(null);
assertFalse(PrivateMessageSessionControl.isKickMessage(legacy));
legacy.setType(MessageType.TEXT.code());
legacy.setContent(null);
assertFalse(PrivateMessageSessionControl.isKickMessage(legacy));
assertFalse(PrivateMessageSessionControl.isKickMessage(null));
assertFalse(PrivateMessageSessionControl.hasKickPrefix("__app_kick_session__:text"));
assertFalse(PrivateMessageSessionControl.hasKickPrefix("xxAPPxKICKxSESSIONxx:text"));
assertFalse(PrivateMessageSessionControl.hasKickPrefix("prefix " + PrivateMessageSessionControl.KICK_PREFIX));
assertFalse(PrivateMessageSessionControl.hasKickPrefix(null));
}
private ThirdPrivateMessageSendDTO kick() {
ThirdPrivateMessageSendDTO dto = new ThirdPrivateMessageSendDTO();
dto.setSendId(Constant.COM_MESSAGE_USER_ID);
dto.setRecvId(42L);
dto.setTerminal(1);
dto.setTmpId("kick-42");
dto.setType(MessageType.TEXT.code());
dto.setContent(PrivateMessageSessionControl.KICK_PREFIX + "账号已被禁用");
return dto;
}
}
@@ -0,0 +1,217 @@
package com.bx.implatform.service.third.impl;
import com.bx.imclient.IMClient;
import com.bx.implatform.config.props.JwtProperties;
import com.bx.implatform.dto.third.user.ThirdCreateOrUpdateUserDTO;
import com.bx.implatform.entity.User;
import com.bx.implatform.enums.ResultCode;
import com.bx.implatform.exception.GlobalException;
import com.bx.implatform.mapper.UserMapper;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.security.crypto.password.PasswordEncoder;
import org.springframework.test.util.ReflectionTestUtils;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
@ExtendWith(MockitoExtension.class)
class ThirdUserServiceImplTest {
@Mock
private UserMapper mapper;
@Mock
private PasswordEncoder passwordEncoder;
private ThirdUserServiceImpl service;
@BeforeEach
void setUp() {
service = new ThirdUserServiceImpl(passwordEncoder, new JwtProperties(), mock(IMClient.class));
ReflectionTestUtils.setField(service, "baseMapper", mapper);
}
@Test
void createsUserWithRequestedIdAndInitialPassword() {
ThirdCreateOrUpdateUserDTO dto = request();
when(passwordEncoder.encode(dto.generatePassword())).thenReturn("initial-password-hash");
service.createOrUpdate(dto);
ArgumentCaptor<User> captor = ArgumentCaptor.forClass(User.class);
verify(mapper).insert(captor.capture());
User inserted = captor.getValue();
assertEquals(-100L, inserted.getId());
assertEquals(dto.getUserName(), inserted.getUserName());
assertEquals(dto.getNickName(), inserted.getNickName());
assertEquals(dto.getAvatar(), inserted.getHeadImage());
assertEquals(dto.getAvatar(), inserted.getHeadImageThumb());
assertEquals("initial-password-hash", inserted.getPassword());
}
@Test
void missingNicknameDefaultsToUsernameOnCreation() {
ThirdCreateOrUpdateUserDTO dto = request();
dto.setNickName(null);
when(passwordEncoder.encode(dto.generatePassword())).thenReturn("initial-password-hash");
service.createOrUpdate(dto);
ArgumentCaptor<User> captor = ArgumentCaptor.forClass(User.class);
verify(mapper).insert(captor.capture());
assertEquals(dto.getUserName(), captor.getValue().getNickName());
}
@Test
void repeatedInitializationSkipsWritesAndPasswordEncoding() {
when(mapper.selectById(-100L)).thenReturn(existingUser());
service.createOrUpdate(request());
verify(mapper, never()).insert(any(User.class));
verify(mapper, never()).updateById(any(User.class));
verifyNoInteractions(passwordEncoder);
}
@Test
void updatesProfileWithoutOverwritingPasswordOrBanState() {
User existing = existingUser();
existing.setPassword("existing-password");
existing.setIsBanned(true);
existing.setReason("existing-ban-reason");
when(mapper.selectById(-100L)).thenReturn(existing);
ThirdCreateOrUpdateUserDTO dto = request();
dto.setNickName("new nickname");
service.createOrUpdate(dto);
ArgumentCaptor<User> captor = ArgumentCaptor.forClass(User.class);
verify(mapper).updateById(captor.capture());
User update = captor.getValue();
assertEquals("new nickname", update.getNickName());
assertNull(update.getPassword());
assertNull(update.getIsBanned());
assertNull(update.getReason());
assertEquals("existing-password", existing.getPassword());
assertTrue(existing.getIsBanned());
verifyNoInteractions(passwordEncoder);
}
@Test
void omittedOptionalFieldsDoNotTriggerAnUpdate() {
when(mapper.selectById(-100L)).thenReturn(existingUser());
ThirdCreateOrUpdateUserDTO dto = request();
dto.setNickName(null);
dto.setAvatar(null);
service.createOrUpdate(dto);
verify(mapper, never()).updateById(any(User.class));
verifyNoInteractions(passwordEncoder);
}
@Test
void outdatedThumbnailIsUpdatedEvenWhenOriginalAvatarMatches() {
User existing = existingUser();
existing.setHeadImageThumb("old-thumbnail");
when(mapper.selectById(-100L)).thenReturn(existing);
service.createOrUpdate(request());
ArgumentCaptor<User> captor = ArgumentCaptor.forClass(User.class);
verify(mapper).updateById(captor.capture());
assertEquals(request().getAvatar(), captor.getValue().getHeadImageThumb());
}
@Test
void concurrentCreationOfSameUserAndProfileSucceedsWithoutUpdatingPassword() {
when(mapper.selectById(-100L)).thenReturn(null, existingUser());
when(passwordEncoder.encode(any())).thenReturn("unused-password-hash");
when(mapper.insert(any(User.class))).thenThrow(new DuplicateKeyException("PRIMARY"));
assertDoesNotThrow(() -> service.createOrUpdate(request()));
verify(mapper, never()).updateById(any(User.class));
}
@Test
void concurrentCreationUpdatesRequestedProfileButKeepsExistingCredentials() {
User concurrentUser = existingUser();
concurrentUser.setNickName("old nickname");
when(mapper.selectById(-100L)).thenReturn(null, concurrentUser);
when(passwordEncoder.encode(any())).thenReturn("unused-password-hash");
when(mapper.insert(any(User.class))).thenThrow(new DuplicateKeyException("PRIMARY"));
service.createOrUpdate(request());
ArgumentCaptor<User> captor = ArgumentCaptor.forClass(User.class);
verify(mapper).updateById(captor.capture());
assertEquals(request().getNickName(), captor.getValue().getNickName());
assertNull(captor.getValue().getPassword());
}
@Test
void usernameConflictOnCreationReturnsBusinessErrorWithoutUpdatingAnotherUser() {
when(passwordEncoder.encode(any())).thenReturn("initial-password-hash");
when(mapper.insert(any(User.class))).thenThrow(new DuplicateKeyException("idx_user_name"));
GlobalException exception = assertThrows(GlobalException.class,
() -> service.createOrUpdate(request()));
assertEquals(ResultCode.USERNAME_ALREADY_REGISTER.getCode(), exception.getCode());
verify(mapper, never()).updateById(any(User.class));
}
@Test
void usernameConflictOnUpdateReturnsBusinessError() {
User existing = existingUser();
existing.setUserName("old-username");
when(mapper.selectById(-100L)).thenReturn(existing);
when(mapper.updateById(any(User.class))).thenThrow(new DuplicateKeyException("idx_user_name"));
GlobalException exception = assertThrows(GlobalException.class,
() -> service.createOrUpdate(request()));
assertEquals(ResultCode.USERNAME_ALREADY_REGISTER.getCode(), exception.getCode());
}
@Test
void databaseFailureIsNotReportedAsSuccessfulSynchronization() {
User existing = existingUser();
existing.setNickName("old nickname");
when(mapper.selectById(-100L)).thenReturn(existing);
DataAccessResourceFailureException failure = new DataAccessResourceFailureException("database unavailable");
when(mapper.updateById(any(User.class))).thenThrow(failure);
assertSame(failure, assertThrows(DataAccessResourceFailureException.class,
() -> service.createOrUpdate(request())));
}
private ThirdCreateOrUpdateUserDTO request() {
ThirdCreateOrUpdateUserDTO dto = new ThirdCreateOrUpdateUserDTO();
dto.setUserId(-100L);
dto.setUserName("message-notification");
dto.setPhone("15555555555");
dto.setNickName("消息通知");
dto.setAvatar("https://example.com/avatar.png");
return dto;
}
private User existingUser() {
ThirdCreateOrUpdateUserDTO dto = request();
User user = new User();
user.setId(dto.getUserId());
user.setUserName(dto.getUserName());
user.setNickName(dto.getNickName());
user.setHeadImage(dto.getAvatar());
user.setHeadImageThumb(dto.getAvatar());
return user;
}
}
+140
View File
@@ -0,0 +1,140 @@
#!/usr/bin/env node
'use strict'
// Exercises the installed mobile client's real store code without modifying or starting the app.
// Run: node --test scripts/test-moderation-client.cjs [CLIENT_REPO may be provided in the environment].
const assert = require('node:assert/strict')
const fs = require('node:fs')
const path = require('node:path')
const vm = require('node:vm')
const { test } = require('node:test')
const clientRoot = process.env.CLIENT_REPO || path.resolve(__dirname, '../../../Friendship-and-chat-uniapp')
const storePath = path.join(clientRoot, 'src/chat-store/chatStore.js')
const enumsPath = path.join(clientRoot, 'src/common/enums.js')
const clone = (value) => value === undefined ? undefined : JSON.parse(JSON.stringify(value))
const chatInfo = { type: 'PRIVATE', targetId: 7 }
function harness() {
const memory = new Map()
const context = vm.createContext({
defineStore: (_name, definition) => definition,
useUserStore: () => ({ userInfo: { id: 42 } }),
useFriendStore: () => ({ findFriend: () => null }),
useGroupStore: () => ({ findGroup: () => null }),
console: { log() {} },
uni: {
setStorageSync: (key, value) => memory.set(key, clone(value)),
getStorageSync: (key) => clone(memory.get(key)),
removeStorageSync: (key) => memory.delete(key)
}
})
const enumSource = fs.readFileSync(enumsPath, 'utf8').replace(/export\s*\{[\s\S]*?\}\s*;?\s*$/, '')
vm.runInContext(enumSource, context, { filename: enumsPath })
const storeSource = fs.readFileSync(storePath, 'utf8')
.replace(/^import\s+.*?;?\s*$/gm, '')
.replace('export default defineStore', 'globalThis.definition = defineStore')
vm.runInContext(storeSource, context, { filename: storePath })
const definition = context.definition
const store = definition.state()
for (const [key, action] of Object.entries(definition.actions)) store[key] = action.bind(store)
for (const [key, getter] of Object.entries(definition.getters)) {
Object.defineProperty(store, key, { get: () => getter(store) })
}
return { store, memory }
}
function message(id, content = 'original private content') {
return { id, type: 0, status: 3, content, sendId: 7, recvId: 42, sendTime: 1000 + id, selfSend: false }
}
function quote(id) {
return { ...message(id, 'reply text'), quoteMessage: message(101) }
}
function conversation(messages, hotMinIdx = 0) {
return {
...chatInfo, showName: 'Sender', messages, hotMinIdx, unreadCount: 2,
readedMessageIdx: 0, lastContent: 'original private content', lastSendTime: 1101,
stored: false, hidden: false
}
}
function recall(selfSend = false) {
// The persisted event has its own ID; its legacy-client wire projection must use the source ID.
return { id: 101, content: '101', type: 10, status: 3, sendId: 7, recvId: 42, sendTime: 2000, selfSend }
}
function assertHidden(value, expectedText = '对方撤回了一条消息') {
assert.equal(value.type, 21)
assert.equal(value.status, 2)
assert.equal(value.content, expectedText)
}
for (const { name, split, messages } of [
{ name: 'hot source and hot quote', split: 0, messages: [message(101), quote(102)] },
{ name: 'cold source and hot quote', split: 1, messages: [message(101), quote(102)] },
{ name: 'cold source and cold quote', split: 2, messages: [message(101), quote(102), message(103, 'later message')] }
]) {
test(`legacy RECALL hides ${name} and persists both affected partitions`, () => {
const { store, memory } = harness()
const chat = conversation(clone(messages), split)
store.chats = [chat]
store.saveToStorage(true)
store.recallMessage(recall(), chatInfo)
assertHidden(chat.messages[0])
assertHidden(chat.messages[1].quoteMessage, '引用内容已撤回')
const cold = memory.get('chats-app-42-PRIVATE-7')?.messages || []
const hot = memory.get('chats-app-42-PRIVATE-7-hot')?.messages || []
const persisted = [...cold, ...hot]
assertHidden(persisted.find((item) => item.id === 101))
assertHidden(persisted.find((item) => item.id === 102).quoteMessage, '引用内容已撤回')
assert.equal(chat.unreadCount, 2)
assert.equal(chat.lastContent, '对方撤回了一条消息')
})
}
test('repeat notifications with status 3 do not add unread counts or duplicate rows', () => {
const { store, memory } = harness()
const chat = conversation([message(101), quote(102)], 1)
store.chats = [chat]
store.saveToStorage(true)
store.recallMessage(recall(), chatInfo)
store.recallMessage(recall(), chatInfo)
assert.equal(chat.unreadCount, 2)
assert.equal(chat.messages.length, 2)
assert.equal(memory.get('chats-app-42-PRIVATE-7-hot').unreadCount, 2)
assertHidden(chat.messages[0])
})
test('sender receives the same compatible projection and sees their own recall tip', () => {
const { store } = harness()
const chat = conversation([message(101)])
store.chats = [chat]
store.recallMessage(recall(true), chatInfo)
assertHidden(chat.messages[0], '你撤回了一条消息')
assert.equal(chat.unreadCount, 2)
})
test('offline buffer processing persists hidden source and quotes after refresh', () => {
const { store, memory } = harness()
store.initChats({ chats: [conversation([message(101), quote(102)], 2)], privateMsgMaxId: 102 })
store.setLoading(true)
store.recallMessage(recall(), chatInfo)
assert.equal(memory.size, 0)
store.setLoading(false)
store.refreshChats()
assertHidden(store.chats[0].messages[0])
assertHidden(store.chats[0].messages[1].quoteMessage, '引用内容已撤回')
const cold = memory.get('chats-app-42-PRIVATE-7').messages
assertHidden(cold[0])
assertHidden(cold[1].quoteMessage, '引用内容已撤回')
assert.equal(store.chats[0].unreadCount, 2)
})
test('unknown conversation is ignored without creating empty chats or unread counts', () => {
const { store, memory } = harness()
assert.doesNotThrow(() => store.recallMessage(recall(), chatInfo))
assert.equal(store.chats.length, 0)
assert.equal(memory.size, 0)
})