From 6fe6b3a12b4c78844650aa708da3ff6f17f0ee84 Mon Sep 17 00:00:00 2001 From: dzc <2388224167@qq.com> Date: Wed, 11 Nov 2026 19:39:21 +0800 Subject: [PATCH] 111 --- db/20261111_private_message_push_event.sql | 15 ++ .../imcommon/model/IMAppForegroundDevice.java | 11 ++ .../bx/imcommon/model/IMHeartbeatInfo.java | 2 + .../com/bx/imcommon/model/IMLoginInfo.java | 3 + .../bx/imcommon/util/AppPushSessionStore.java | 93 ++++++++++ .../props/ImPushCallbackProperties.java | 21 +++ .../ThirdPrivateMessagePushController.java | 27 +++ .../ThirdPrivateMessagePushContextDTO.java | 12 ++ .../entity/PrivateMessagePushEvent.java | 23 +++ .../mapper/PrivateMessagePushEventMapper.java | 28 +++ .../service/ImPushCallbackClient.java | 36 ++++ .../PrivateMessagePushContextService.java | 75 ++++++++ .../PrivateMessagePushOutboxService.java | 44 +++++ .../impl/PrivateMessageServiceImpl.java | 26 ++- .../impl/ThirdPrivateMsgServiceImpl.java | 16 +- .../PrivateMessagePushDeliveryTask.java | 69 ++++++++ .../ThirdPrivateMessagePushContextVO.java | 20 +++ .../src/main/resources/application.yml | 4 + ...ThirdPrivateMessagePushControllerTest.java | 68 ++++++++ .../service/AppPushSessionStoreRedisTest.java | 100 +++++++++++ .../service/ImPushCallbackClientTest.java | 69 ++++++++ .../PrivateMessageModerationMySqlTest.java | 2 +- .../PrivateMessageModerationReadPathTest.java | 2 +- .../PrivateMessagePushContextServiceTest.java | 124 +++++++++++++ .../PrivateMessagePushOutboxMySqlTest.java | 159 +++++++++++++++++ .../PrivateMessagePushOutboxServiceTest.java | 88 ++++++++++ .../PrivateMessagePushTransactionTest.java | 163 ++++++++++++++++++ ...vateMessageSessionControlReadPathTest.java | 2 +- .../service/UserProfileSyncMySqlTest.java | 2 +- .../ThirdPrivateConversationServiceTest.java | 4 +- ...ThirdPrivateSessionControlServiceTest.java | 4 +- .../PrivateMessagePushDeliveryTaskTest.java | 88 ++++++++++ .../bx/imserver/constant/ChannelAttrKey.java | 3 + .../bx/imserver/netty/IMChannelHandler.java | 9 + .../netty/processor/HeartbeatProcessor.java | 19 ++ .../netty/processor/LoginProcessor.java | 23 +++ .../netty/processor/AppPushHeartbeatTest.java | 69 ++++++++ .../processor/MemberAccessConnectionTest.java | 4 +- 38 files changed, 1516 insertions(+), 11 deletions(-) create mode 100644 db/20261111_private_message_push_event.sql create mode 100644 im-common/src/main/java/com/bx/imcommon/model/IMAppForegroundDevice.java create mode 100644 im-common/src/main/java/com/bx/imcommon/util/AppPushSessionStore.java create mode 100644 im-platform/src/main/java/com/bx/implatform/config/props/ImPushCallbackProperties.java create mode 100644 im-platform/src/main/java/com/bx/implatform/controller/third/ThirdPrivateMessagePushController.java create mode 100644 im-platform/src/main/java/com/bx/implatform/dto/third/privatemsg/ThirdPrivateMessagePushContextDTO.java create mode 100644 im-platform/src/main/java/com/bx/implatform/entity/PrivateMessagePushEvent.java create mode 100644 im-platform/src/main/java/com/bx/implatform/mapper/PrivateMessagePushEventMapper.java create mode 100644 im-platform/src/main/java/com/bx/implatform/service/ImPushCallbackClient.java create mode 100644 im-platform/src/main/java/com/bx/implatform/service/PrivateMessagePushContextService.java create mode 100644 im-platform/src/main/java/com/bx/implatform/service/PrivateMessagePushOutboxService.java create mode 100644 im-platform/src/main/java/com/bx/implatform/task/schedule/PrivateMessagePushDeliveryTask.java create mode 100644 im-platform/src/main/java/com/bx/implatform/vo/third/privatemsg/ThirdPrivateMessagePushContextVO.java create mode 100644 im-platform/src/test/java/com/bx/implatform/controller/third/ThirdPrivateMessagePushControllerTest.java create mode 100644 im-platform/src/test/java/com/bx/implatform/service/AppPushSessionStoreRedisTest.java create mode 100644 im-platform/src/test/java/com/bx/implatform/service/ImPushCallbackClientTest.java create mode 100644 im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushContextServiceTest.java create mode 100644 im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushOutboxMySqlTest.java create mode 100644 im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushOutboxServiceTest.java create mode 100644 im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushTransactionTest.java create mode 100644 im-platform/src/test/java/com/bx/implatform/task/schedule/PrivateMessagePushDeliveryTaskTest.java create mode 100644 im-server/src/test/java/com/bx/imserver/netty/processor/AppPushHeartbeatTest.java diff --git a/db/20261111_private_message_push_event.sql b/db/20261111_private_message_push_event.sql new file mode 100644 index 0000000..3c904c7 --- /dev/null +++ b/db/20261111_private_message_push_event.sql @@ -0,0 +1,15 @@ +-- Apply before enabling IM_PUSH_CALLBACK_URL and IM_PUSH_CALLBACK_SECRET. +CREATE TABLE IF NOT EXISTS `im_private_message_push_event` ( + `id` bigint NOT NULL AUTO_INCREMENT, + `message_id` bigint NOT NULL COMMENT 'IM private message id', + `payload` text NOT NULL COMMENT 'Signed callback JSON snapshot', + `status` tinyint NOT NULL DEFAULT 0 COMMENT '0 pending, 1 accepted, 2 expired', + `attempt_count` int NOT NULL DEFAULT 0, + `next_retry_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + `last_error` varchar(200) DEFAULT NULL, + `create_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + `update_time` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (`id`), + UNIQUE KEY `uk_private_message_push_message` (`message_id`), + KEY `idx_private_message_push_due` (`status`, `next_retry_time`, `id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Durable private-message push delivery'; diff --git a/im-common/src/main/java/com/bx/imcommon/model/IMAppForegroundDevice.java b/im-common/src/main/java/com/bx/imcommon/model/IMAppForegroundDevice.java new file mode 100644 index 0000000..84c3617 --- /dev/null +++ b/im-common/src/main/java/com/bx/imcommon/model/IMAppForegroundDevice.java @@ -0,0 +1,11 @@ +package com.bx.imcommon.model; + +import lombok.Data; + +@Data +public class IMAppForegroundDevice { + private String registrationId; + private String sessionToken; + private Long heartbeatVersion; + private Long leaseExpiresAt; +} diff --git a/im-common/src/main/java/com/bx/imcommon/model/IMHeartbeatInfo.java b/im-common/src/main/java/com/bx/imcommon/model/IMHeartbeatInfo.java index 4b50271..5eda588 100644 --- a/im-common/src/main/java/com/bx/imcommon/model/IMHeartbeatInfo.java +++ b/im-common/src/main/java/com/bx/imcommon/model/IMHeartbeatInfo.java @@ -4,4 +4,6 @@ import lombok.Data; @Data public class IMHeartbeatInfo { + private String registrationId; + private Boolean foreground; } diff --git a/im-common/src/main/java/com/bx/imcommon/model/IMLoginInfo.java b/im-common/src/main/java/com/bx/imcommon/model/IMLoginInfo.java index afccfca..91e4ad7 100644 --- a/im-common/src/main/java/com/bx/imcommon/model/IMLoginInfo.java +++ b/im-common/src/main/java/com/bx/imcommon/model/IMLoginInfo.java @@ -6,4 +6,7 @@ import lombok.Data; public class IMLoginInfo { private String accessToken; + + private String registrationId; + private Boolean foreground; } diff --git a/im-common/src/main/java/com/bx/imcommon/util/AppPushSessionStore.java b/im-common/src/main/java/com/bx/imcommon/util/AppPushSessionStore.java new file mode 100644 index 0000000..22757e6 --- /dev/null +++ b/im-common/src/main/java/com/bx/imcommon/util/AppPushSessionStore.java @@ -0,0 +1,93 @@ +package com.bx.imcommon.util; + +import com.bx.imcommon.model.IMAppForegroundDevice; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.stereotype.Component; + +import java.util.List; + +/** Keeps short-lived foreground presence independent of IM connection routing. */ +@Component +public class AppPushSessionStore { + private static final String SESSION_PREFIX = "im:push:app_session:"; + private static final String FOREGROUND_PREFIX = "im:push:app_foreground:"; + private static final DefaultRedisScript UPDATE = new DefaultRedisScript<>(""" + if ARGV[1] == 'login' then + redis.call('SET', KEYS[1], ARGV[2], 'EX', 600) + redis.call('DEL', KEYS[2]) + elseif redis.call('GET', KEYS[1]) ~= ARGV[2] then + return 0 + else + redis.call('EXPIRE', KEYS[1], 600) + end + if ARGV[3] ~= '' and ARGV[4] == 'true' then + local now = redis.call('TIME') + local expires = now[1] * 1000 + math.floor(now[2] / 1000) + 45000 + redis.call('HSET', KEYS[2], 'registrationId', ARGV[3], 'sessionToken', ARGV[2], + 'heartbeatVersion', ARGV[5], 'leaseExpiresAt', tostring(expires)) + redis.call('PEXPIRE', KEYS[2], 45000) + else + redis.call('DEL', KEYS[2]) + end + return 1 + """, Long.class); + private static final DefaultRedisScript REMOVE = new DefaultRedisScript<>(""" + if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end + redis.call('DEL', KEYS[1]) + if redis.call('HGET', KEYS[2], 'sessionToken') == ARGV[1] then + redis.call('DEL', KEYS[2]) + end + return 1 + """, Long.class); + private static final DefaultRedisScript READ = new DefaultRedisScript<>(""" + local owner = redis.call('GET', KEYS[1]) + if not owner or redis.call('HGET', KEYS[2], 'sessionToken') ~= owner then return {} end + return redis.call('HMGET', KEYS[2], 'registrationId', 'sessionToken', + 'heartbeatVersion', 'leaseExpiresAt') + """, List.class); + + private final StringRedisTemplate redis; + + public AppPushSessionStore(RedisConnectionFactory factory) { + redis = new StringRedisTemplate(factory); + } + + public void login(Long userId, String token, String registrationId, boolean foreground) { + update("login", userId, token, registrationId, foreground, 0); + } + + public void heartbeat(Long userId, String token, String registrationId, boolean foreground, long version) { + update("heartbeat", userId, token, registrationId, foreground, version); + } + + private void update(String action, Long userId, String token, String registrationId, + boolean foreground, long version) { + redis.execute(UPDATE, keys(userId), action, token, + registrationId == null ? "" : registrationId, String.valueOf(foreground), String.valueOf(version)); + } + + public void logout(Long userId, String token) { + if (token != null) { + redis.execute(REMOVE, keys(userId), token); + } + } + + public List getForegroundDevices(Long userId) { + List values = redis.execute(READ, keys(userId)); + if (values == null || values.size() != 4 || values.contains(null)) { + return List.of(); + } + IMAppForegroundDevice device = new IMAppForegroundDevice(); + device.setRegistrationId(values.get(0).toString()); + device.setSessionToken(values.get(1).toString()); + device.setHeartbeatVersion(Long.parseLong(values.get(2).toString())); + device.setLeaseExpiresAt(Long.parseLong(values.get(3).toString())); + return List.of(device); + } + + private List keys(Long userId) { + return List.of(SESSION_PREFIX + userId, FOREGROUND_PREFIX + userId); + } +} diff --git a/im-platform/src/main/java/com/bx/implatform/config/props/ImPushCallbackProperties.java b/im-platform/src/main/java/com/bx/implatform/config/props/ImPushCallbackProperties.java new file mode 100644 index 0000000..bbd5721 --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/config/props/ImPushCallbackProperties.java @@ -0,0 +1,21 @@ +package com.bx.implatform.config.props; + +import lombok.Data; +import lombok.ToString; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +@Data +@Component +@ConfigurationProperties("im.push-callback") +public class ImPushCallbackProperties { + private boolean enabled = true; + private String url; + @ToString.Exclude + private String secret; + + public boolean isConfigured() { + return enabled && StringUtils.hasText(url) && StringUtils.hasText(secret); + } +} diff --git a/im-platform/src/main/java/com/bx/implatform/controller/third/ThirdPrivateMessagePushController.java b/im-platform/src/main/java/com/bx/implatform/controller/third/ThirdPrivateMessagePushController.java new file mode 100644 index 0000000..4bfd02b --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/controller/third/ThirdPrivateMessagePushController.java @@ -0,0 +1,27 @@ +package com.bx.implatform.controller.third; + +import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessagePushContextDTO; +import com.bx.implatform.result.Result; +import com.bx.implatform.result.ResultUtils; +import com.bx.implatform.service.PrivateMessagePushContextService; +import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessagePushContextVO; +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 ThirdPrivateMessagePushController { + private final PrivateMessagePushContextService service; + + @PostMapping("/pushContext") + @Operation(summary = "查询私聊推送实时上下文") + public Result context(@Valid @RequestBody ThirdPrivateMessagePushContextDTO request) { + return ResultUtils.success(service.getContext(request.getMessageId())); + } +} diff --git a/im-platform/src/main/java/com/bx/implatform/dto/third/privatemsg/ThirdPrivateMessagePushContextDTO.java b/im-platform/src/main/java/com/bx/implatform/dto/third/privatemsg/ThirdPrivateMessagePushContextDTO.java new file mode 100644 index 0000000..858155b --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/dto/third/privatemsg/ThirdPrivateMessagePushContextDTO.java @@ -0,0 +1,12 @@ +package com.bx.implatform.dto.third.privatemsg; + +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +@Data +public class ThirdPrivateMessagePushContextDTO { + @NotNull + @Min(1) + private Long messageId; +} diff --git a/im-platform/src/main/java/com/bx/implatform/entity/PrivateMessagePushEvent.java b/im-platform/src/main/java/com/bx/implatform/entity/PrivateMessagePushEvent.java new file mode 100644 index 0000000..fe29829 --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/entity/PrivateMessagePushEvent.java @@ -0,0 +1,23 @@ +package com.bx.implatform.entity; + +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; + +import java.util.Date; + +@Data +@TableName("im_private_message_push_event") +public class PrivateMessagePushEvent { + @TableId + private Long id; + private Long messageId; + private String payload; + /** 0 pending, 1 accepted by the business API, 2 expired. */ + private Integer status; + private Integer attemptCount; + private Date nextRetryTime; + private String lastError; + private Date createTime; + private Date updateTime; +} diff --git a/im-platform/src/main/java/com/bx/implatform/mapper/PrivateMessagePushEventMapper.java b/im-platform/src/main/java/com/bx/implatform/mapper/PrivateMessagePushEventMapper.java new file mode 100644 index 0000000..76a3390 --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/mapper/PrivateMessagePushEventMapper.java @@ -0,0 +1,28 @@ +package com.bx.implatform.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.bx.implatform.entity.PrivateMessagePushEvent; +import org.apache.ibatis.annotations.Insert; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; +import org.apache.ibatis.annotations.Update; + +import java.util.Date; +import java.util.List; + +public interface PrivateMessagePushEventMapper extends BaseMapper { + @Insert("INSERT INTO im_private_message_push_event " + + "(message_id,payload,status,attempt_count,next_retry_time,create_time,update_time) " + + "VALUES (#{messageId},#{payload},0,0,#{now},#{now},#{now}) " + + "ON DUPLICATE KEY UPDATE message_id=VALUES(message_id)") + int insertPending(@Param("messageId") Long messageId, @Param("payload") String payload, @Param("now") Date now); + + @Select("SELECT * FROM im_private_message_push_event WHERE status=0 AND next_retry_time<=#{now} " + + "ORDER BY next_retry_time,id LIMIT 100") + List selectDue(@Param("now") Date now); + + @Update("UPDATE im_private_message_push_event SET status=#{status},attempt_count=#{attemptCount}," + + "next_retry_time=#{nextRetryTime},last_error=#{lastError},update_time=CURRENT_TIMESTAMP(3) " + + "WHERE id=#{id} AND status=0") + int updatePending(PrivateMessagePushEvent event); +} diff --git a/im-platform/src/main/java/com/bx/implatform/service/ImPushCallbackClient.java b/im-platform/src/main/java/com/bx/implatform/service/ImPushCallbackClient.java new file mode 100644 index 0000000..585f8a3 --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/service/ImPushCallbackClient.java @@ -0,0 +1,36 @@ +package com.bx.implatform.service; + +import cn.hutool.http.HttpRequest; +import cn.hutool.http.HttpResponse; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import com.bx.implatform.config.props.ImPushCallbackProperties; +import lombok.RequiredArgsConstructor; +import org.apache.commons.codec.digest.HmacUtils; +import org.springframework.stereotype.Component; + +import java.time.Instant; + +@Component +@RequiredArgsConstructor +public class ImPushCallbackClient { + private final ImPushCallbackProperties properties; + + public void deliver(String payload) { + String timestamp = String.valueOf(Instant.now().getEpochSecond()); + String signature = new HmacUtils("HmacSHA256", properties.getSecret()).hmacHex(timestamp + "\n" + payload); + try (HttpResponse response = HttpRequest.post(properties.getUrl()) + .header("Content-Type", "application/json; charset=UTF-8") + .header("X-IM-TIMESTAMP", timestamp) + .header("X-IM-SIGNATURE", signature) + .body(payload).timeout(5000).execute()) { + if (!response.isOk()) { + throw new IllegalStateException("HTTP_" + response.getStatus()); + } + JSONObject result = JSON.parseObject(response.body()); + if (result == null || !Integer.valueOf(0).equals(result.getInteger("code"))) { + throw new IllegalStateException("BUSINESS_REJECTED"); + } + } + } +} diff --git a/im-platform/src/main/java/com/bx/implatform/service/PrivateMessagePushContextService.java b/im-platform/src/main/java/com/bx/implatform/service/PrivateMessagePushContextService.java new file mode 100644 index 0000000..e7e4785 --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/service/PrivateMessagePushContextService.java @@ -0,0 +1,75 @@ +package com.bx.implatform.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.bx.imcommon.model.MemberAccessState; +import com.bx.imcommon.util.AppPushSessionStore; +import com.bx.implatform.entity.Friend; +import com.bx.implatform.entity.PrivateMessage; +import com.bx.implatform.entity.User; +import com.bx.implatform.enums.MessageStatus; +import com.bx.implatform.mapper.FriendMapper; +import com.bx.implatform.mapper.PrivateMessageMapper; +import com.bx.implatform.mapper.UserMapper; +import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessagePushContextVO; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +import java.util.List; +import java.util.Objects; + +@Service +@RequiredArgsConstructor +public class PrivateMessagePushContextService { + private final PrivateMessageMapper messageMapper; + private final FriendMapper friendMapper; + private final UserMapper userMapper; + private final AppPushSessionStore sessionStore; + + public ThirdPrivateMessagePushContextVO getContext(Long messageId) { + ThirdPrivateMessagePushContextVO result = new ThirdPrivateMessagePushContextVO(); + result.setMessageId(messageId); + PrivateMessage message = messageMapper.selectById(messageId); + if (message == null) { + result.setSkipReason("MESSAGE_NOT_FOUND"); + return result; + } + result.setSendId(message.getSendId()); + result.setRecvId(message.getRecvId()); + result.setType(message.getType()); + result.setContent(message.getContent()); + result.setSendTime(message.getSendTime() == null ? null : message.getSendTime().getTime()); + result.setStatus(message.getStatus()); + if (!PrivateMessagePushOutboxService.isMemberMessage(message)) { + result.setSkipReason("INVALID_MESSAGE"); + return result; + } + if (Objects.equals(message.getStatus(), MessageStatus.RECALL.code())) { + result.setSkipReason("MESSAGE_RECALLED"); + return result; + } + if (Objects.equals(message.getStatus(), MessageStatus.READED.code())) { + result.setSkipReason("MESSAGE_READ"); + return result; + } + List users = userMapper.selectBatchIds(List.of(message.getSendId(), message.getRecvId())); + if (users.size() != 2 || users.stream().anyMatch(this::isUnavailable)) { + result.setSkipReason("MEMBER_UNAVAILABLE"); + return result; + } + Friend friend = friendMapper.selectOne(new LambdaQueryWrapper() + .eq(Friend::getUserId, message.getRecvId()).eq(Friend::getFriendId, message.getSendId()) + .last("LIMIT 1")); + if (friend != null && Boolean.TRUE.equals(friend.getIsDnd())) { + result.setSkipReason("DND"); + return result; + } + result.setForegroundDevices(sessionStore.getForegroundDevices(message.getRecvId())); + result.setEligible(true); + return result; + } + + private boolean isUnavailable(User user) { + return Boolean.TRUE.equals(user.getIsBanned()) || Boolean.TRUE.equals(user.getAccessDisabled()) + || !Objects.equals(user.getMemberState(), MemberAccessState.ACTIVE); + } +} diff --git a/im-platform/src/main/java/com/bx/implatform/service/PrivateMessagePushOutboxService.java b/im-platform/src/main/java/com/bx/implatform/service/PrivateMessagePushOutboxService.java new file mode 100644 index 0000000..a63329f --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/service/PrivateMessagePushOutboxService.java @@ -0,0 +1,44 @@ +package com.bx.implatform.service; + +import com.alibaba.fastjson.JSONObject; +import com.bx.implatform.config.props.ImPushCallbackProperties; +import com.bx.implatform.entity.PrivateMessage; +import com.bx.implatform.mapper.PrivateMessagePushEventMapper; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; + +import java.util.Date; + +@Service +@RequiredArgsConstructor +public class PrivateMessagePushOutboxService { + private final ImPushCallbackProperties properties; + private final PrivateMessagePushEventMapper mapper; + + @Transactional(propagation = Propagation.MANDATORY) + public void enqueue(PrivateMessage message) { + if (!properties.isConfigured() || !isMemberMessage(message)) { + return; + } + JSONObject payload = new JSONObject(true); + payload.put("eventId", "private:" + message.getId()); + payload.put("messageId", message.getId()); + payload.put("sendId", message.getSendId()); + payload.put("recvId", message.getRecvId()); + payload.put("type", message.getType()); + payload.put("content", message.getContent() == null ? "" : message.getContent()); + payload.put("sendTime", message.getSendTime().getTime()); + mapper.insertPending(message.getId(), payload.toJSONString(), new Date()); + } + + public static boolean isMemberMessage(PrivateMessage message) { + return message != null && message.getId() != null && message.getId() > 0 + && message.getSendId() != null && message.getSendId() > 0 + && message.getRecvId() != null && message.getRecvId() > 0 + && !message.getSendId().equals(message.getRecvId()) + && message.getType() != null && message.getType() >= 0 && message.getType() <= 4 + && message.getSendTime() != null; + } +} diff --git a/im-platform/src/main/java/com/bx/implatform/service/impl/PrivateMessageServiceImpl.java b/im-platform/src/main/java/com/bx/implatform/service/impl/PrivateMessageServiceImpl.java index e964525..a310c88 100644 --- a/im-platform/src/main/java/com/bx/implatform/service/impl/PrivateMessageServiceImpl.java +++ b/im-platform/src/main/java/com/bx/implatform/service/impl/PrivateMessageServiceImpl.java @@ -20,6 +20,7 @@ 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.service.PrivateMessagePushOutboxService; import com.bx.implatform.session.SessionContext; import com.bx.implatform.session.UserSession; import com.bx.implatform.util.BeanUtils; @@ -31,6 +32,8 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.time.DateUtils; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; import java.util.Date; import java.util.List; @@ -49,9 +52,11 @@ public class PrivateMessageServiceImpl extends ServiceImpl sendMessage = new IMPrivateMessage<>(); @@ -78,11 +84,29 @@ public class PrivateMessageServiceImpl extends ServiceImpl message) { + try { + imClient.sendPrivateMessage(message); + } catch (RuntimeException e) { + log.warn("私聊已保存，实时投递失败,messageId:{},exceptionType:{}", + message.getData().getId(), e.getClass().getSimpleName()); + } + } + @Transactional @Override public PrivateMessageVO recallMessage(Long id) { diff --git a/im-platform/src/main/java/com/bx/implatform/service/third/impl/ThirdPrivateMsgServiceImpl.java b/im-platform/src/main/java/com/bx/implatform/service/third/impl/ThirdPrivateMsgServiceImpl.java index 1667911..647e81e 100644 --- a/im-platform/src/main/java/com/bx/implatform/service/third/impl/ThirdPrivateMsgServiceImpl.java +++ b/im-platform/src/main/java/com/bx/implatform/service/third/impl/ThirdPrivateMsgServiceImpl.java @@ -29,6 +29,7 @@ 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.PrivateMessagePushOutboxService; import com.bx.implatform.service.third.ThirdPrivateMsgService; import com.bx.implatform.session.SessionContext; import com.bx.implatform.session.UserSession; @@ -61,6 +62,7 @@ public class ThirdPrivateMsgServiceImpl extends ServiceImpl sendMessage = new IMPrivateMessage<>(); @@ -131,16 +134,25 @@ public class ThirdPrivateMsgServiceImpl extends ServiceImpl message) { + try { + imClient.sendPrivateMessage(message); + } catch (RuntimeException e) { + log.warn("私聊已保存，实时投递失败,messageId:{},exceptionType:{}", + message.getData().getId(), e.getClass().getSimpleName()); + } + } + private PrivateMessageVO sendSessionControl(ThirdPrivateMessageSendDTO dto) { PrivateMessageVO command = BeanUtils.copyProperties(dto, PrivateMessageVO.class); // Keep the existing response/envelope shape, without creating a replayable message row. diff --git a/im-platform/src/main/java/com/bx/implatform/task/schedule/PrivateMessagePushDeliveryTask.java b/im-platform/src/main/java/com/bx/implatform/task/schedule/PrivateMessagePushDeliveryTask.java new file mode 100644 index 0000000..09433c6 --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/task/schedule/PrivateMessagePushDeliveryTask.java @@ -0,0 +1,69 @@ +package com.bx.implatform.task.schedule; + +import com.bx.implatform.config.props.ImPushCallbackProperties; +import com.bx.implatform.entity.PrivateMessagePushEvent; +import com.bx.implatform.mapper.PrivateMessagePushEventMapper; +import com.bx.implatform.service.ImPushCallbackClient; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.redisson.api.RLock; +import org.redisson.api.RedissonClient; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.util.Date; + +@Slf4j +@Component +@RequiredArgsConstructor +public class PrivateMessagePushDeliveryTask { + private static final long MAX_AGE_MILLIS = Duration.ofHours(24).toMillis(); + private static final long[] RETRY_SECONDS = {5, 30, 120, 600}; + private final ImPushCallbackProperties properties; + private final PrivateMessagePushEventMapper mapper; + private final ImPushCallbackClient callbackClient; + private final RedissonClient redissonClient; + + @Scheduled(fixedDelay = 1000, initialDelay = 1000) + public void run() { + if (!properties.isConfigured()) { + return; + } + RLock lock = redissonClient.getLock("im:lock:task:private-message-push"); + // With no explicit lease Redisson renews ownership while a batch is running. + if (!lock.tryLock()) { + return; + } + try { + for (PrivateMessagePushEvent event : mapper.selectDue(new Date())) { + deliver(event); + } + } finally { + lock.unlock(); + } + } + + void deliver(PrivateMessagePushEvent event) { + long now = System.currentTimeMillis(); + if (now - event.getCreateTime().getTime() >= MAX_AGE_MILLIS) { + event.setStatus(2); + event.setLastError("EXPIRED"); + mapper.updatePending(event); + return; + } + event.setAttemptCount(event.getAttemptCount() + 1); + try { + callbackClient.deliver(event.getPayload()); + event.setStatus(1); + event.setLastError(null); + } catch (RuntimeException e) { + int index = Math.min(event.getAttemptCount() - 1, RETRY_SECONDS.length - 1); + event.setNextRetryTime(new Date(System.currentTimeMillis() + RETRY_SECONDS[index] * 1000)); + event.setLastError(e.getClass().getSimpleName()); + log.warn("交付私聊推送事件失败,messageId:{},attempt:{},exceptionType:{}", + event.getMessageId(), event.getAttemptCount(), e.getClass().getSimpleName()); + } + mapper.updatePending(event); + } +} diff --git a/im-platform/src/main/java/com/bx/implatform/vo/third/privatemsg/ThirdPrivateMessagePushContextVO.java b/im-platform/src/main/java/com/bx/implatform/vo/third/privatemsg/ThirdPrivateMessagePushContextVO.java new file mode 100644 index 0000000..96fb9f9 --- /dev/null +++ b/im-platform/src/main/java/com/bx/implatform/vo/third/privatemsg/ThirdPrivateMessagePushContextVO.java @@ -0,0 +1,20 @@ +package com.bx.implatform.vo.third.privatemsg; + +import com.bx.imcommon.model.IMAppForegroundDevice; +import lombok.Data; + +import java.util.List; + +@Data +public class ThirdPrivateMessagePushContextVO { + private Long messageId; + private Long sendId; + private Long recvId; + private Integer type; + private String content; + private Long sendTime; + private Integer status; + private boolean eligible; + private String skipReason; + private List foregroundDevices = List.of(); +} diff --git a/im-platform/src/main/resources/application.yml b/im-platform/src/main/resources/application.yml index c1fc721..4ff5f8e 100644 --- a/im-platform/src/main/resources/application.yml +++ b/im-platform/src/main/resources/application.yml @@ -34,3 +34,7 @@ jwt: im: auth: business-only: true + push-callback: + enabled: ${IM_PUSH_CALLBACK_ENABLED:true} + url: ${IM_PUSH_CALLBACK_URL:} + secret: ${IM_PUSH_CALLBACK_SECRET:} diff --git a/im-platform/src/test/java/com/bx/implatform/controller/third/ThirdPrivateMessagePushControllerTest.java b/im-platform/src/test/java/com/bx/implatform/controller/third/ThirdPrivateMessagePushControllerTest.java new file mode 100644 index 0000000..4e87e6f --- /dev/null +++ b/im-platform/src/test/java/com/bx/implatform/controller/third/ThirdPrivateMessagePushControllerTest.java @@ -0,0 +1,68 @@ +package com.bx.implatform.controller.third; + +import com.bx.implatform.config.props.JwtProperties; +import com.bx.implatform.exception.GlobalExceptionHandler; +import com.bx.implatform.filter.CacheFilter; +import com.bx.implatform.interceptor.AuthInterceptor; +import com.bx.implatform.service.MemberAccountService; +import com.bx.implatform.service.PrivateMessagePushContextService; +import com.bx.implatform.service.third.ThirdAuthService; +import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessagePushContextVO; +import jakarta.servlet.http.HttpServletRequest; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.http.MediaType; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +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.*; + +class ThirdPrivateMessagePushControllerTest { + private final PrivateMessagePushContextService service = mock(PrivateMessagePushContextService.class); + private final ThirdAuthService auth = mock(ThirdAuthService.class); + private MockMvc mvc; + + @BeforeEach + void setUp() { + when(auth.checkSign(any(HttpServletRequest.class))).thenReturn(true); + mvc = MockMvcBuilders.standaloneSetup(new ThirdPrivateMessagePushController(service)) + .setControllerAdvice(new GlobalExceptionHandler()) + .addInterceptors(new AuthInterceptor(new JwtProperties(), auth, mock(MemberAccountService.class))) + .addFilters(new CacheFilter()).build(); + } + + @Test + void exposesExpectedWireShapeThroughExistingThirdPartyAuthentication() throws Exception { + ThirdPrivateMessagePushContextVO context = new ThirdPrivateMessagePushContextVO(); + context.setMessageId(101L); + context.setRecvId(42L); + context.setSendId(7L); + context.setEligible(true); + when(service.getContext(101L)).thenReturn(context); + mvc.perform(post("/third/privateMsg/pushContext").contentType(MediaType.APPLICATION_JSON) + .content("{\"messageId\":101}")) + .andExpect(status().isOk()).andExpect(jsonPath("$.code").value(200)) + .andExpect(jsonPath("$.data.eligible").value(true)) + .andExpect(jsonPath("$.data.foregroundDevices").isArray()); + } + + @Test + void invalidSignatureCannotReadPushContext() throws Exception { + when(auth.checkSign(any(HttpServletRequest.class))).thenReturn(false); + mvc.perform(post("/third/privateMsg/pushContext").contentType(MediaType.APPLICATION_JSON) + .content("{\"messageId\":101}")) + .andExpect(jsonPath("$.code").value(401)); + verifyNoInteractions(service); + } + + @Test + void validatesMessageIdentifierBeforeQuerying() throws Exception { + mvc.perform(post("/third/privateMsg/pushContext").contentType(MediaType.APPLICATION_JSON) + .content("{\"messageId\":0}")) + .andExpect(jsonPath("$.code").value(500)); + verifyNoInteractions(service); + } +} diff --git a/im-platform/src/test/java/com/bx/implatform/service/AppPushSessionStoreRedisTest.java b/im-platform/src/test/java/com/bx/implatform/service/AppPushSessionStoreRedisTest.java new file mode 100644 index 0000000..dcff9e8 --- /dev/null +++ b/im-platform/src/test/java/com/bx/implatform/service/AppPushSessionStoreRedisTest.java @@ -0,0 +1,100 @@ +package com.bx.implatform.service; + +import com.bx.imcommon.model.IMAppForegroundDevice; +import com.bx.imcommon.util.AppPushSessionStore; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; +import org.redisson.spring.data.connection.RedissonConnectionFactory; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +class AppPushSessionStoreRedisTest { + private final Long id = 9223372036854774999L; + private final String ownerKey = "im:push:app_session:" + id; + private final String foregroundKey = "im:push:app_foreground:" + id; + private RedissonClient client; + private StringRedisTemplate redis; + private AppPushSessionStore store; + + @BeforeEach + void setUp() { + String port = System.getenv("IM_TEST_REDIS_PORT"); + assumeTrue(port != null, "Set IM_TEST_REDIS_PORT for isolated Redis tests"); + Config config = new Config(); + config.setThreads(2).setNettyThreads(2); + config.useSingleServer().setAddress("redis://127.0.0.1:" + Integer.parseInt(port)) + .setConnectionMinimumIdleSize(1).setConnectionPoolSize(2); + client = Redisson.create(config); + RedissonConnectionFactory factory = new RedissonConnectionFactory(client); + redis = new StringRedisTemplate(factory); + store = new AppPushSessionStore(factory); + redis.delete(List.of(ownerKey, foregroundKey)); + } + + @AfterEach + void tearDown() { + if (client != null) { + redis.delete(List.of(ownerKey, foregroundKey)); + client.shutdown(); + } + } + + @Test + void foregroundLeaseUsesRegistrationAndAdvancingHeartbeatVersion() { + store.login(id, "server:one", "reg-a", true); + assertEquals(0L, device().getHeartbeatVersion()); + store.heartbeat(id, "server:one", "reg-a", true, 4); + assertEquals("reg-a", device().getRegistrationId()); + assertEquals(4L, device().getHeartbeatVersion()); + assertTrue(redis.getExpire(foregroundKey, TimeUnit.MILLISECONDS) <= 45000); + assertTrue(redis.getExpire(foregroundKey, TimeUnit.MILLISECONDS) > 43000); + assertTrue(device().getLeaseExpiresAt() > System.currentTimeMillis()); + } + + @Test + void backgroundAndDisconnectImmediatelyRemoveForegroundPresence() { + store.login(id, "server:one", "reg-a", true); + store.heartbeat(id, "server:one", "reg-a", false, 1); + assertTrue(store.getForegroundDevices(id).isEmpty()); + store.heartbeat(id, "server:one", "reg-a", true, 2); + store.logout(id, "server:one"); + assertTrue(store.getForegroundDevices(id).isEmpty()); + } + + @Test + void oldConnectionCannotRenewOrDeleteReplacementSession() { + store.login(id, "server:one", "reg-a", true); + store.login(id, "server:two", "reg-b", true); + store.heartbeat(id, "server:one", "reg-a", true, 999); + store.logout(id, "server:one"); + assertEquals("reg-b", device().getRegistrationId()); + assertEquals("server:two", device().getSessionToken()); + assertEquals(0L, device().getHeartbeatVersion()); + } + + @Test + void vanishedForegroundLeaseFallsBackToPushAndLegacyLoginCreatesNoLease() throws Exception { + store.login(id, "server:one", "reg-a", true); + redis.expire(foregroundKey, Duration.ofMillis(1)); + Thread.sleep(10); + assertTrue(store.getForegroundDevices(id).isEmpty()); + store.login(id, "server:two", null, false); + assertTrue(store.getForegroundDevices(id).isEmpty()); + } + + private IMAppForegroundDevice device() { + List devices = store.getForegroundDevices(id); + assertEquals(1, devices.size()); + return devices.get(0); + } +} diff --git a/im-platform/src/test/java/com/bx/implatform/service/ImPushCallbackClientTest.java b/im-platform/src/test/java/com/bx/implatform/service/ImPushCallbackClientTest.java new file mode 100644 index 0000000..f540131 --- /dev/null +++ b/im-platform/src/test/java/com/bx/implatform/service/ImPushCallbackClientTest.java @@ -0,0 +1,69 @@ +package com.bx.implatform.service; + +import com.bx.implatform.config.props.ImPushCallbackProperties; +import com.sun.net.httpserver.HttpServer; +import org.apache.commons.codec.digest.HmacUtils; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +class ImPushCallbackClientTest { + private HttpServer server; + private ImPushCallbackClient client; + private final AtomicReference body = new AtomicReference<>(); + private final AtomicReference signature = new AtomicReference<>(); + private final AtomicReference timestamp = new AtomicReference<>(); + private String responseBody = "{\"code\":0,\"data\":true}"; + private int responseCode = 200; + + @BeforeEach + void setUp() throws Exception { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/callback", exchange -> { + body.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + timestamp.set(exchange.getRequestHeaders().getFirst("X-IM-TIMESTAMP")); + signature.set(exchange.getRequestHeaders().getFirst("X-IM-SIGNATURE")); + byte[] bytes = responseBody.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(responseCode, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + }); + server.start(); + ImPushCallbackProperties properties = new ImPushCallbackProperties(); + properties.setUrl("http://127.0.0.1:" + server.getAddress().getPort() + "/callback"); + properties.setSecret("test-secret"); + client = new ImPushCallbackClient(properties); + } + + @AfterEach + void tearDown() { + server.stop(0); + } + + @Test + void signsExactUtf8BodyAndAcceptsIdempotentBusinessAcknowledgement() { + String payload = "{\"eventId\":\"private:101\",\"content\":\"中文与\\n换行\"}"; + client.deliver(payload); + assertEquals(payload, body.get()); + assertEquals(new HmacUtils("HmacSHA256", "test-secret").hmacHex(timestamp.get() + "\n" + payload), signature.get()); + assertTrue(Math.abs(System.currentTimeMillis() / 1000 - Long.parseLong(timestamp.get())) <= 2); + } + + @Test + void rejectsHttpSuccessWithoutBusinessAcceptance() { + responseBody = "{\"code\":500,\"msg\":\"failed\"}"; + assertThrows(IllegalStateException.class, () -> client.deliver("{}")); + } + + @Test + void rejectsNonSuccessfulHttpStatusEvenWithCodeZero() { + responseCode = 503; + assertThrows(IllegalStateException.class, () -> client.deliver("{}")); + } +} diff --git a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageModerationMySqlTest.java b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageModerationMySqlTest.java index 39f21f7..74dfe90 100644 --- a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageModerationMySqlTest.java +++ b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageModerationMySqlTest.java @@ -163,7 +163,7 @@ class PrivateMessageModerationMySqlTest { + "('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); + mock(SensitiveFilterUtil.class), service, mock(PrivateMessagePushOutboxService.class)); LambdaQueryWrapper query = ReflectionTestUtils.invokeMethod(history, "buildOfflineQuery", 42L, 999999L); List messages = mapper.selectList(query); assertEquals(1, messages.size()); diff --git a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageModerationReadPathTest.java b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageModerationReadPathTest.java index 1baa573..1ed8ed0 100644 --- a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageModerationReadPathTest.java +++ b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageModerationReadPathTest.java @@ -45,7 +45,7 @@ class PrivateMessageModerationReadPathTest { mapper = mock(PrivateMessageMapper.class); moderation = mock(PrivateMessageModerationService.class); service = new PrivateMessageServiceImpl(mock(FriendService.class), mock(IMClient.class), - mock(SensitiveFilterUtil.class), moderation); + mock(SensitiveFilterUtil.class), moderation, mock(PrivateMessagePushOutboxService.class)); ReflectionTestUtils.setField(service, "baseMapper", mapper); TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "moderation-read-test"), PrivateMessage.class); UserSession session = new UserSession(); diff --git a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushContextServiceTest.java b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushContextServiceTest.java new file mode 100644 index 0000000..bc7057c --- /dev/null +++ b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushContextServiceTest.java @@ -0,0 +1,124 @@ +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.imcommon.model.IMAppForegroundDevice; +import com.bx.imcommon.util.AppPushSessionStore; +import com.bx.implatform.entity.Friend; +import com.bx.implatform.entity.PrivateMessage; +import com.bx.implatform.entity.User; +import com.bx.implatform.mapper.FriendMapper; +import com.bx.implatform.mapper.PrivateMessageMapper; +import com.bx.implatform.mapper.UserMapper; +import com.bx.implatform.vo.third.privatemsg.ThirdPrivateMessagePushContextVO; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +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.*; + +class PrivateMessagePushContextServiceTest { + private final PrivateMessageMapper messages = mock(PrivateMessageMapper.class); + private final FriendMapper friends = mock(FriendMapper.class); + private final UserMapper users = mock(UserMapper.class); + private final AppPushSessionStore sessions = mock(AppPushSessionStore.class); + private final PrivateMessagePushContextService service = new PrivateMessagePushContextService(messages, friends, users, sessions); + private PrivateMessage source; + + @BeforeEach + void setUp() { + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "push-context"), Friend.class); + source = new PrivateMessage(); + source.setId(101L); + source.setSendId(7L); + source.setRecvId(42L); + source.setType(0); + source.setStatus(0); + source.setContent("过滤后的正文"); + source.setSendTime(new Date()); + when(messages.selectById(101L)).thenReturn(source); + when(users.selectBatchIds(anyCollection())).thenReturn(List.of(activeUser(7L), activeUser(42L))); + } + + @Test + void returnsCurrentPerDeviceForegroundLeaseAndAllowsMissingRecipientFriendRow() { + IMAppForegroundDevice device = new IMAppForegroundDevice(); + device.setRegistrationId("device-a"); + device.setSessionToken("server:channel"); + device.setHeartbeatVersion(3L); + device.setLeaseExpiresAt(System.currentTimeMillis() + 45000); + when(sessions.getForegroundDevices(42L)).thenReturn(List.of(device)); + ThirdPrivateMessagePushContextVO context = service.getContext(101L); + assertTrue(context.isEligible()); + assertEquals("过滤后的正文", context.getContent()); + assertEquals(List.of(device), context.getForegroundDevices()); + ArgumentCaptor> query = ArgumentCaptor.forClass(Wrapper.class); + verify(friends).selectOne(query.capture()); + assertTrue(query.getValue().getSqlSegment().contains("user_id")); + assertTrue(query.getValue().getSqlSegment().contains("friend_id")); + } + + @Test + void recipientDndSuppressesNotificationRegardlessOfImDeliveryStatus() { + Friend friend = new Friend(); + friend.setIsDnd(true); + when(friends.selectOne(any(Wrapper.class))).thenReturn(friend); + source.setStatus(1); + assertEquals("DND", service.getContext(101L).getSkipReason()); + verifyNoInteractions(sessions); + } + + @Test + void recalledOrModeratedMessageCannotBePushed() { + source.setStatus(2); + ThirdPrivateMessagePushContextVO context = service.getContext(101L); + assertFalse(context.isEligible()); + assertEquals("MESSAGE_RECALLED", context.getSkipReason()); + verifyNoInteractions(sessions); + } + + @Test + void readMessageStopsPendingPushAfterTheAppGoesToBackground() { + source.setStatus(3); + ThirdPrivateMessagePushContextVO context = service.getContext(101L); + assertFalse(context.isEligible()); + assertEquals("MESSAGE_READ", context.getSkipReason()); + verifyNoInteractions(sessions); + } + + @Test + void socketDeliveryAloneDoesNotSuppressBackgroundNotification() { + source.setStatus(1); + assertTrue(service.getContext(101L).isEligible()); + } + + @Test + void missingAndDisabledMembersCannotBePushed() { + when(users.selectBatchIds(anyCollection())).thenReturn(List.of(activeUser(7L))); + assertEquals("MEMBER_UNAVAILABLE", service.getContext(101L).getSkipReason()); + User disabled = activeUser(42L); + disabled.setAccessDisabled(true); + when(users.selectBatchIds(anyCollection())).thenReturn(List.of(activeUser(7L), disabled)); + assertEquals("MEMBER_UNAVAILABLE", service.getContext(101L).getSkipReason()); + } + + @Test + void vanishedSourceIsAnExplicitTerminalDecision() { + when(messages.selectById(101L)).thenReturn(null); + assertEquals("MESSAGE_NOT_FOUND", service.getContext(101L).getSkipReason()); + } + + private User activeUser(Long id) { + User user = new User(); + user.setId(id); + user.setMemberState(0); + return user; + } +} diff --git a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushOutboxMySqlTest.java b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushOutboxMySqlTest.java new file mode 100644 index 0000000..6b0c211 --- /dev/null +++ b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushOutboxMySqlTest.java @@ -0,0 +1,159 @@ +package com.bx.implatform.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder; +import com.bx.imcommon.util.AppPushSessionStore; +import com.bx.implatform.config.props.ImPushCallbackProperties; +import com.bx.implatform.entity.PrivateMessage; +import com.bx.implatform.entity.PrivateMessagePushEvent; +import com.bx.implatform.mapper.FriendMapper; +import com.bx.implatform.mapper.PrivateMessageMapper; +import com.bx.implatform.mapper.PrivateMessagePushEventMapper; +import com.bx.implatform.mapper.UserMapper; +import org.apache.ibatis.datasource.unpooled.UnpooledDataSource; +import org.apache.ibatis.mapping.Environment; +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.transaction.support.TransactionTemplate; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Date; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assumptions.assumeTrue; +import static org.mockito.Mockito.*; + +class PrivateMessagePushOutboxMySqlTest { + private JdbcTemplate admin; + private JdbcTemplate jdbc; + private String database; + private TransactionTemplate transactions; + private PrivateMessageMapper messages; + private PrivateMessagePushEventMapper events; + private PrivateMessagePushOutboxService outbox; + private PrivateMessagePushContextService context; + + @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 database tests"); + assertTrue(url.matches("jdbc:mysql://(localhost|127\\.0\\.0\\.1):[0-9]+/[a-zA-Z0-9_]+(?:\\?.*)?")); + String password = System.getenv().getOrDefault("IM_TEST_MYSQL_PASSWORD", ""); + admin = new JdbcTemplate(new UnpooledDataSource("com.mysql.cj.jdbc.Driver", url, "root", password)); + database = "codex_im_push_" + UUID.randomUUID().toString().replace("-", ""); + admin.execute("CREATE DATABASE " + database + " CHARACTER SET utf8mb4"); + int queryIndex = url.indexOf('?'); + String base = queryIndex < 0 ? url : url.substring(0, queryIndex); + String localUrl = base.substring(0, base.lastIndexOf('/') + 1) + database + + (queryIndex < 0 ? "" : url.substring(queryIndex)); + UnpooledDataSource source = new UnpooledDataSource("com.mysql.cj.jdbc.Driver", localUrl, "root", password); + jdbc = new JdbcTemplate(source); + executeScript("../db/im-platform.sql"); + executeScript("../db/20261111_private_message_push_event.sql"); + transactions = new TransactionTemplate(new DataSourceTransactionManager(source)); + MybatisConfiguration configuration = new MybatisConfiguration(); + configuration.setMapUnderscoreToCamelCase(true); + configuration.setEnvironment(new Environment("im-push-test", new SpringManagedTransactionFactory(), source)); + configuration.addMapper(PrivateMessageMapper.class); + configuration.addMapper(PrivateMessagePushEventMapper.class); + configuration.addMapper(UserMapper.class); + configuration.addMapper(FriendMapper.class); + SqlSessionTemplate sessions = new SqlSessionTemplate(new MybatisSqlSessionFactoryBuilder().build(configuration)); + messages = sessions.getMapper(PrivateMessageMapper.class); + events = sessions.getMapper(PrivateMessagePushEventMapper.class); + ImPushCallbackProperties properties = new ImPushCallbackProperties(); + properties.setUrl("http://127.0.0.1/callback"); + properties.setSecret("test-secret"); + outbox = new PrivateMessagePushOutboxService(properties, events); + AppPushSessionStore foreground = mock(AppPushSessionStore.class); + when(foreground.getForegroundDevices(42L)).thenReturn(List.of()); + context = new PrivateMessagePushContextService(messages, sessions.getMapper(FriendMapper.class), + sessions.getMapper(UserMapper.class), foreground); + jdbc.update("INSERT INTO im_user (id,user_name,nick_name,password) VALUES (7,'u7','Sender','test'),(42,'u42','Receiver','test')"); + } + + @AfterEach + void tearDown() { + if (admin != null && database != null) { + admin.execute("DROP DATABASE " + database); + } + } + + @Test + void committedMessageAndEventAreQueryableAndRepeatedEnqueueIsIdempotent() { + PrivateMessage message = source(); + transactions.executeWithoutResult(status -> { + messages.insert(message); + outbox.enqueue(message); + outbox.enqueue(message); + }); + assertEquals(1, jdbc.queryForObject("SELECT COUNT(*) FROM im_private_message_push_event", Integer.class)); + List pending = events.selectDue(new Date()); + assertEquals(1, pending.size()); + assertEquals(101L, pending.get(0).getMessageId()); + assertEquals(0, pending.get(0).getStatus()); + assertNotNull(pending.get(0).getCreateTime()); + assertTrue(Math.abs(System.currentTimeMillis() - pending.get(0).getCreateTime().getTime()) < 5000); + pending.get(0).setStatus(1); + events.updatePending(pending.get(0)); + assertTrue(events.selectDue(new Date()).isEmpty()); + assertTrue(context.getContext(101L).isEligible()); + } + + @Test + void failedTransactionLeavesNeitherMessageNorEvent() { + assertThrows(IllegalStateException.class, () -> transactions.executeWithoutResult(status -> { + PrivateMessage message = source(); + messages.insert(message); + outbox.enqueue(message); + throw new IllegalStateException("force rollback"); + })); + assertNull(messages.selectById(101L)); + assertEquals(0, jdbc.queryForObject("SELECT COUNT(*) FROM im_private_message_push_event", Integer.class)); + } + + @Test + void contextSeesDndAndRecallChangesMadeAfterEventCommit() { + transactions.executeWithoutResult(status -> { + PrivateMessage message = source(); + messages.insert(message); + outbox.enqueue(message); + }); + assertTrue(context.getContext(101L).isEligible()); + jdbc.update("INSERT INTO im_friend (user_id,friend_id,friend_nick_name,is_dnd) VALUES (42,7,'Sender',1)"); + assertEquals("DND", context.getContext(101L).getSkipReason()); + jdbc.update("UPDATE im_friend SET is_dnd=0 WHERE user_id=42 AND friend_id=7"); + jdbc.update("UPDATE im_private_message SET status=3 WHERE id=101"); + assertEquals("MESSAGE_READ", context.getContext(101L).getSkipReason()); + jdbc.update("UPDATE im_private_message SET status=2 WHERE id=101"); + assertEquals("MESSAGE_RECALLED", context.getContext(101L).getSkipReason()); + } + + private void executeScript(String path) throws Exception { + for (String sql : Files.readString(Path.of(path)).split(";")) { + if (!sql.isBlank()) { + jdbc.execute(sql); + } + } + } + + private PrivateMessage source() { + PrivateMessage message = new PrivateMessage(); + message.setId(101L); + message.setSendId(7L); + message.setRecvId(42L); + message.setType(0); + message.setStatus(0); + message.setContent("过滤后的正文"); + message.setSendTime(new Date()); + return message; + } +} diff --git a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushOutboxServiceTest.java b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushOutboxServiceTest.java new file mode 100644 index 0000000..af0d372 --- /dev/null +++ b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushOutboxServiceTest.java @@ -0,0 +1,88 @@ +package com.bx.implatform.service; + +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import com.bx.implatform.config.props.ImPushCallbackProperties; +import com.bx.implatform.entity.PrivateMessage; +import com.bx.implatform.mapper.PrivateMessagePushEventMapper; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.util.Date; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +class PrivateMessagePushOutboxServiceTest { + private final ImPushCallbackProperties properties = new ImPushCallbackProperties(); + private final PrivateMessagePushEventMapper mapper = mock(PrivateMessagePushEventMapper.class); + private final PrivateMessagePushOutboxService service = new PrivateMessagePushOutboxService(properties, mapper); + + @Test + void unconfiguredCallbackDoesNotRequireTheEventTable() { + service.enqueue(message()); + verifyNoInteractions(mapper); + } + + @Test + void createsStableEventFromFilteredPersistedMessageAndPreservesMilliseconds() { + configure(); + service.enqueue(message()); + ArgumentCaptor json = ArgumentCaptor.forClass(String.class); + verify(mapper).insertPending(eq(101L), json.capture(), any(Date.class)); + JSONObject event = JSON.parseObject(json.getValue()); + assertEquals("private:101", event.getString("eventId")); + assertEquals(101L, event.getLong("messageId")); + assertEquals(7L, event.getLong("sendId")); + assertEquals(42L, event.getLong("recvId")); + assertEquals(0, event.getInteger("type")); + assertEquals("过滤后的中文内容", event.getString("content")); + assertEquals(1730000000123L, event.getLong("sendTime")); + } + + @Test + void ignoresSystemMessagesControlSignalsAndDisabledConfiguration() { + configure(); + PrivateMessage source = message(); + source.setSendId(-200L); + service.enqueue(source); + source = message(); + source.setType(10); + service.enqueue(source); + source = message(); + source.setRecvId(source.getSendId()); + service.enqueue(source); + properties.setEnabled(false); + service.enqueue(message()); + verifyNoInteractions(mapper); + } + + @Test + void whitespaceTextWhoseFilterReturnedNullStillUsesAValidCallbackShape() { + configure(); + PrivateMessage source = message(); + source.setContent(null); + service.enqueue(source); + ArgumentCaptor json = ArgumentCaptor.forClass(String.class); + verify(mapper).insertPending(eq(101L), json.capture(), any(Date.class)); + assertEquals("", JSON.parseObject(json.getValue()).getString("content")); + } + + private void configure() { + properties.setUrl("http://127.0.0.1/callback"); + properties.setSecret("test-secret"); + } + + private PrivateMessage message() { + PrivateMessage message = new PrivateMessage(); + message.setId(101L); + message.setSendId(7L); + message.setRecvId(42L); + message.setType(0); + message.setContent("过滤后的中文内容"); + message.setSendTime(new Date(1730000000123L)); + return message; + } +} diff --git a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushTransactionTest.java b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushTransactionTest.java new file mode 100644 index 0000000..c4522d9 --- /dev/null +++ b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessagePushTransactionTest.java @@ -0,0 +1,163 @@ +package com.bx.implatform.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.bx.imclient.IMClient; +import com.bx.implatform.dto.PrivateMessageDTO; +import com.bx.implatform.dto.third.privatemsg.ThirdPrivateMessageSendDTO; +import com.bx.implatform.entity.PrivateMessage; +import com.bx.implatform.mapper.PrivateMessageMapper; +import com.bx.implatform.mapper.UserMapper; +import com.bx.implatform.service.impl.PrivateMessageServiceImpl; +import com.bx.implatform.service.third.impl.ThirdPrivateMsgServiceImpl; +import com.bx.implatform.session.UserSession; +import com.bx.implatform.util.SensitiveFilterUtil; +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.springframework.aop.framework.ProxyFactory; +import org.springframework.mock.web.MockHttpServletRequest; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.transaction.support.AbstractPlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionStatus; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.web.context.request.RequestContextHolder; +import org.springframework.web.context.request.ServletRequestAttributes; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +class PrivateMessagePushTransactionTest { + private final PrivateMessageMapper mapper = mock(PrivateMessageMapper.class); + private final FriendService friends = mock(FriendService.class); + private final IMClient client = mock(IMClient.class); + private final SensitiveFilterUtil filter = mock(SensitiveFilterUtil.class); + private final PrivateMessagePushOutboxService outbox = mock(PrivateMessagePushOutboxService.class); + private final TestTransactionManager transactions = new TestTransactionManager(); + + @BeforeEach + void setUp() { + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "push-transactions"), PrivateMessage.class); + when(friends.isFriend(7L, 42L)).thenReturn(true); + when(filter.filter("original")).thenReturn("filtered"); + when(mapper.insert(any(PrivateMessage.class))).thenAnswer(invocation -> { + assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); + ((PrivateMessage) invocation.getArgument(0)).setId(101L); + return 1; + }); + doAnswer(invocation -> { + assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); + assertFalse(transactions.committed); + assertEquals("filtered", ((PrivateMessage) invocation.getArgument(0)).getContent()); + return null; + }).when(outbox).enqueue(any()); + doAnswer(invocation -> { + assertTrue(transactions.committed, "IM publication must happen after the database commit"); + return null; + }).when(client).sendPrivateMessage(any()); + UserSession session = new UserSession(); + session.setUserId(7L); + session.setTerminal(1); + MockHttpServletRequest request = new MockHttpServletRequest(); + request.setAttribute("session", session); + RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request)); + } + + @AfterEach + void tearDown() { + RequestContextHolder.resetRequestAttributes(); + } + + @Test + void ordinaryEntryEnqueuesInsideTransactionAndPublishesAfterCommit() { + PrivateMessageServiceImpl service = ordinary(); + assertEquals(101L, service.sendMessage(request()).getId()); + verify(outbox).enqueue(any()); + verify(client).sendPrivateMessage(any()); + assertTrue(transactions.committed); + } + + @Test + void thirdPartyEntryUsesTheSameAtomicOutboxAndAfterCommitPublication() { + ThirdPrivateMsgServiceImpl target = new ThirdPrivateMsgServiceImpl(mock(UserMapper.class), mapper, client, friends, filter, outbox); + ReflectionTestUtils.setField(target, "baseMapper", mapper); + ThirdPrivateMessageSendDTO request = new ThirdPrivateMessageSendDTO(); + request.setSendId(7L); + request.setRecvId(42L); + request.setTerminal(1); + request.setType(0); + request.setContent("original"); + assertEquals(101L, transactional(target).sendMessage(request).getId()); + verify(outbox).enqueue(any()); + verify(client).sendPrivateMessage(any()); + } + + @Test + void outboxFailureRollsBackMessageWithoutPublishingIt() { + doThrow(new IllegalStateException("insert failed")).when(outbox).enqueue(any()); + assertThrows(IllegalStateException.class, () -> ordinary().sendMessage(request())); + assertTrue(transactions.rolledBack); + assertFalse(transactions.committed); + verifyNoInteractions(client); + } + + @Test + void realtimeTransportFailureAfterCommitDoesNotTurnSuccessfulSendIntoARetry() { + doThrow(new IllegalStateException("connection down")).when(client).sendPrivateMessage(any()); + assertEquals(101L, ordinary().sendMessage(request()).getId()); + assertTrue(transactions.committed); + assertFalse(transactions.rolledBack); + } + + private PrivateMessageServiceImpl ordinary() { + PrivateMessageServiceImpl target = new PrivateMessageServiceImpl(friends, client, filter, + mock(PrivateMessageModerationService.class), outbox); + ReflectionTestUtils.setField(target, "baseMapper", mapper); + return transactional(target); + } + + @SuppressWarnings("unchecked") + private T transactional(T target) { + ProxyFactory factory = new ProxyFactory(target); + factory.setProxyTargetClass(true); + factory.addAdvice(new TransactionInterceptor(transactions, new AnnotationTransactionAttributeSource())); + return (T) factory.getProxy(); + } + + private PrivateMessageDTO request() { + PrivateMessageDTO request = new PrivateMessageDTO(); + request.setRecvId(42L); + request.setType(0); + request.setContent("original"); + return request; + } + + private static class TestTransactionManager extends AbstractPlatformTransactionManager { + private boolean committed; + private boolean rolledBack; + + @Override + protected Object doGetTransaction() { + return new Object(); + } + + @Override + protected void doBegin(Object transaction, TransactionDefinition definition) { + } + + @Override + protected void doCommit(DefaultTransactionStatus status) { + committed = true; + } + + @Override + protected void doRollback(DefaultTransactionStatus status) { + rolledBack = true; + } + } +} diff --git a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageSessionControlReadPathTest.java b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageSessionControlReadPathTest.java index 0f80120..c19b960 100644 --- a/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageSessionControlReadPathTest.java +++ b/im-platform/src/test/java/com/bx/implatform/service/PrivateMessageSessionControlReadPathTest.java @@ -50,7 +50,7 @@ class PrivateMessageSessionControlReadPathTest { moderation = mock(PrivateMessageModerationService.class); friends = mock(FriendService.class); imClient = mock(IMClient.class); - service = new PrivateMessageServiceImpl(friends, imClient, mock(SensitiveFilterUtil.class), moderation); + service = new PrivateMessageServiceImpl(friends, imClient, mock(SensitiveFilterUtil.class), moderation, mock(PrivateMessagePushOutboxService.class)); ReflectionTestUtils.setField(service, "baseMapper", mapper); TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "session-control-read-test"), PrivateMessage.class); diff --git a/im-platform/src/test/java/com/bx/implatform/service/UserProfileSyncMySqlTest.java b/im-platform/src/test/java/com/bx/implatform/service/UserProfileSyncMySqlTest.java index 68df57f..1693215 100644 --- a/im-platform/src/test/java/com/bx/implatform/service/UserProfileSyncMySqlTest.java +++ b/im-platform/src/test/java/com/bx/implatform/service/UserProfileSyncMySqlTest.java @@ -94,7 +94,7 @@ class UserProfileSyncMySqlTest { client = mock(IMClient.class); accounts = new MemberAccountService(users, new TransactionTemplate(manager), store, client); ThirdPrivateMsgServiceImpl messageTarget = new ThirdPrivateMsgServiceImpl(users, - sessions.getMapper(PrivateMessageMapper.class), client, friendService, mock(SensitiveFilterUtil.class)); + sessions.getMapper(PrivateMessageMapper.class), client, friendService, mock(SensitiveFilterUtil.class), mock(PrivateMessagePushOutboxService.class)); ReflectionTestUtils.setField(messageTarget, "baseMapper", sessions.getMapper(PrivateMessageMapper.class)); messages = transactional(messageTarget, manager); jdbc.update("INSERT INTO im_user (id,user_name,nick_name,password) VALUES (100,'老芋头','老芋头','retained'),(131,'芋头','芋头','retained'),(-200,'客服','客服','retained')"); diff --git a/im-platform/src/test/java/com/bx/implatform/service/third/impl/ThirdPrivateConversationServiceTest.java b/im-platform/src/test/java/com/bx/implatform/service/third/impl/ThirdPrivateConversationServiceTest.java index 03ff47b..36fdfd8 100644 --- a/im-platform/src/test/java/com/bx/implatform/service/third/impl/ThirdPrivateConversationServiceTest.java +++ b/im-platform/src/test/java/com/bx/implatform/service/third/impl/ThirdPrivateConversationServiceTest.java @@ -1,5 +1,7 @@ package com.bx.implatform.service.third.impl; +import com.bx.implatform.service.PrivateMessagePushOutboxService; + import com.baomidou.mybatisplus.core.MybatisConfiguration; import com.baomidou.mybatisplus.core.conditions.Wrapper; import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; @@ -44,7 +46,7 @@ class ThirdPrivateConversationServiceTest { @BeforeEach void setUp() { service = new ThirdPrivateMsgServiceImpl(userMapper, mapper, mock(IMClient.class), - mock(FriendService.class), mock(SensitiveFilterUtil.class)); + mock(FriendService.class), mock(SensitiveFilterUtil.class), mock(PrivateMessagePushOutboxService.class)); ReflectionTestUtils.setField(service, "baseMapper", mapper); TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "test"), PrivateMessage.class); } diff --git a/im-platform/src/test/java/com/bx/implatform/service/third/impl/ThirdPrivateSessionControlServiceTest.java b/im-platform/src/test/java/com/bx/implatform/service/third/impl/ThirdPrivateSessionControlServiceTest.java index 8620d2e..3b420fd 100644 --- a/im-platform/src/test/java/com/bx/implatform/service/third/impl/ThirdPrivateSessionControlServiceTest.java +++ b/im-platform/src/test/java/com/bx/implatform/service/third/impl/ThirdPrivateSessionControlServiceTest.java @@ -1,5 +1,7 @@ package com.bx.implatform.service.third.impl; +import com.bx.implatform.service.PrivateMessagePushOutboxService; + import com.bx.imclient.IMClient; import com.bx.imcommon.model.IMPrivateMessage; import com.bx.implatform.contant.Constant; @@ -38,7 +40,7 @@ class ThirdPrivateSessionControlServiceTest { imClient = mock(IMClient.class); friends = mock(FriendService.class); sensitiveFilter = mock(SensitiveFilterUtil.class); - service = new ThirdPrivateMsgServiceImpl(userMapper, mapper, imClient, friends, sensitiveFilter); + service = new ThirdPrivateMsgServiceImpl(userMapper, mapper, imClient, friends, sensitiveFilter, mock(PrivateMessagePushOutboxService.class)); ReflectionTestUtils.setField(service, "baseMapper", mapper); } diff --git a/im-platform/src/test/java/com/bx/implatform/task/schedule/PrivateMessagePushDeliveryTaskTest.java b/im-platform/src/test/java/com/bx/implatform/task/schedule/PrivateMessagePushDeliveryTaskTest.java new file mode 100644 index 0000000..72df216 --- /dev/null +++ b/im-platform/src/test/java/com/bx/implatform/task/schedule/PrivateMessagePushDeliveryTaskTest.java @@ -0,0 +1,88 @@ +package com.bx.implatform.task.schedule; + +import com.bx.implatform.config.props.ImPushCallbackProperties; +import com.bx.implatform.entity.PrivateMessagePushEvent; +import com.bx.implatform.mapper.PrivateMessagePushEventMapper; +import com.bx.implatform.service.ImPushCallbackClient; +import org.junit.jupiter.api.Test; +import org.redisson.api.RLock; +import org.redisson.api.RedissonClient; + +import java.time.Duration; +import java.util.Date; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +class PrivateMessagePushDeliveryTaskTest { + private final ImPushCallbackProperties properties = new ImPushCallbackProperties(); + private final PrivateMessagePushEventMapper mapper = mock(PrivateMessagePushEventMapper.class); + private final ImPushCallbackClient callback = mock(ImPushCallbackClient.class); + private final RedissonClient redisson = mock(RedissonClient.class); + private final PrivateMessagePushDeliveryTask task = new PrivateMessagePushDeliveryTask(properties, mapper, callback, redisson); + + @Test + void acknowledgesOnlyAfterBusinessApiAcceptance() { + PrivateMessagePushEvent event = event(); + task.deliver(event); + assertEquals(1, event.getStatus()); + assertEquals(1, event.getAttemptCount()); + verify(callback).deliver("{}"); + verify(mapper).updatePending(event); + } + + @Test + void failedDeliveriesUseBoundedBackoffAndRemainPending() { + doThrow(new IllegalStateException()).when(callback).deliver(anyString()); + PrivateMessagePushEvent event = event(); + for (long expectedSeconds : new long[]{5, 30, 120, 600, 600}) { + long before = System.currentTimeMillis(); + task.deliver(event); + assertEquals(0, event.getStatus()); + long delay = event.getNextRetryTime().getTime() - before; + assertTrue(delay >= expectedSeconds * 1000 && delay < expectedSeconds * 1000 + 1000); + } + } + + @Test + void expiredEventStopsWithoutAnHttpRequest() { + PrivateMessagePushEvent event = event(); + event.setCreateTime(new Date(System.currentTimeMillis() - Duration.ofHours(24).toMillis())); + task.deliver(event); + assertEquals(2, event.getStatus()); + assertEquals("EXPIRED", event.getLastError()); + verifyNoInteractions(callback); + } + + @Test + void disabledTaskDoesNotTouchRedisOrTheUnmigratedDatabase() { + task.run(); + verifyNoInteractions(redisson, mapper, callback); + } + + @Test + void workerUsesWatchdogLockAndReleasesItAfterDatabaseFailure() { + properties.setUrl("http://127.0.0.1/callback"); + properties.setSecret("test-secret"); + RLock lock = mock(RLock.class); + when(redisson.getLock(anyString())).thenReturn(lock); + when(lock.tryLock()).thenReturn(true); + when(mapper.selectDue(any())).thenThrow(new IllegalStateException()); + assertThrows(IllegalStateException.class, task::run); + verify(lock).tryLock(); + verify(lock).unlock(); + } + + private PrivateMessagePushEvent event() { + PrivateMessagePushEvent event = new PrivateMessagePushEvent(); + event.setId(1L); + event.setMessageId(101L); + event.setPayload("{}"); + event.setStatus(0); + event.setAttemptCount(0); + event.setCreateTime(new Date()); + event.setNextRetryTime(new Date()); + return event; + } +} diff --git a/im-server/src/main/java/com/bx/imserver/constant/ChannelAttrKey.java b/im-server/src/main/java/com/bx/imserver/constant/ChannelAttrKey.java index 93d2d42..222c66d 100644 --- a/im-server/src/main/java/com/bx/imserver/constant/ChannelAttrKey.java +++ b/im-server/src/main/java/com/bx/imserver/constant/ChannelAttrKey.java @@ -3,6 +3,9 @@ package com.bx.imserver.constant; public final class ChannelAttrKey { public static final String AUTH_VERSION = "AUTH_VERSION"; + public static final String PUSH_SESSION_TOKEN = "PUSH_SESSION_TOKEN"; + public static final String PUSH_REGISTRATION_ID = "PUSH_REGISTRATION_ID"; + public static final String PUSH_FOREGROUND = "PUSH_FOREGROUND"; private ChannelAttrKey() { } diff --git a/im-server/src/main/java/com/bx/imserver/netty/IMChannelHandler.java b/im-server/src/main/java/com/bx/imserver/netty/IMChannelHandler.java index 00209a8..ea26751 100644 --- a/im-server/src/main/java/com/bx/imserver/netty/IMChannelHandler.java +++ b/im-server/src/main/java/com/bx/imserver/netty/IMChannelHandler.java @@ -4,6 +4,7 @@ import com.bx.imcommon.contant.IMRedisKey; import com.bx.imcommon.enums.IMCmdType; import com.bx.imcommon.model.IMSendInfo; import com.bx.imcommon.mq.RedisMQTemplate; +import com.bx.imcommon.util.AppPushSessionStore; import com.bx.imserver.constant.ChannelAttrKey; import com.bx.imserver.netty.processor.AbstractMessageProcessor; import com.bx.imserver.netty.processor.ProcessorFactory; @@ -68,6 +69,14 @@ public class IMChannelHandler extends SimpleChannelInboundHandler { if (userId == null || terminal == null) { return; } + String pushToken = ctx.channel().attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_SESSION_TOKEN)).get(); + if (pushToken != null) { + try { + SpringContextHolder.getBean(AppPushSessionStore.class).logout(userId, pushToken); + } catch (RuntimeException e) { + log.warn("清理App前台状态失败,userId:{},exceptionType:{}", userId, e.getClass().getSimpleName()); + } + } ChannelHandlerContext context = UserChannelCtxMap.getChannelCtx(userId, terminal); // 判断一下，避免异地登录导致的误删 if (context != null && ctx.channel().id().equals(context.channel().id())) { diff --git a/im-server/src/main/java/com/bx/imserver/netty/processor/HeartbeatProcessor.java b/im-server/src/main/java/com/bx/imserver/netty/processor/HeartbeatProcessor.java index 41783f5..cd71f35 100644 --- a/im-server/src/main/java/com/bx/imserver/netty/processor/HeartbeatProcessor.java +++ b/im-server/src/main/java/com/bx/imserver/netty/processor/HeartbeatProcessor.java @@ -6,6 +6,7 @@ import com.bx.imcommon.contant.IMRedisKey; import com.bx.imcommon.enums.IMCmdType; import com.bx.imcommon.model.IMHeartbeatInfo; import com.bx.imcommon.model.IMSendInfo; +import com.bx.imcommon.util.AppPushSessionStore; import com.bx.imcommon.mq.RedisMQTemplate; import com.bx.imserver.constant.ChannelAttrKey; import io.netty.channel.ChannelHandlerContext; @@ -24,6 +25,7 @@ public class HeartbeatProcessor extends AbstractMessageProcessor heartBeatAttr = AttributeKey.valueOf(ChannelAttrKey.HEARTBEAT_TIMES); Long heartbeatTimes = ctx.channel().attr(heartBeatAttr).get(); ctx.channel().attr(heartBeatAttr).set(++heartbeatTimes); + String pushToken = ctx.channel().attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_SESSION_TOKEN)).get(); + if (pushToken != null) { + AttributeKey registrationAttr = AttributeKey.valueOf(ChannelAttrKey.PUSH_REGISTRATION_ID); + AttributeKey foregroundAttr = AttributeKey.valueOf(ChannelAttrKey.PUSH_FOREGROUND); + if (beatInfo.getRegistrationId() != null) { + ctx.channel().attr(registrationAttr).set(LoginProcessor.validRegistrationId(beatInfo.getRegistrationId())); + } + if (beatInfo.getForeground() != null) { + ctx.channel().attr(foregroundAttr).set(beatInfo.getForeground()); + } + try { + pushSessionStore.heartbeat(currentUserId, pushToken, ctx.channel().attr(registrationAttr).get(), + Boolean.TRUE.equals(ctx.channel().attr(foregroundAttr).get()), heartbeatTimes); + } catch (RuntimeException e) { + log.warn("更新App前台状态失败,userId:{},exceptionType:{}", currentUserId, e.getClass().getSimpleName()); + } + } if (heartbeatTimes % 10 == 0) { // 每心跳10次，用户在线状态续一次命 AttributeKey userIdAttr = AttributeKey.valueOf(ChannelAttrKey.USER_ID); diff --git a/im-server/src/main/java/com/bx/imserver/netty/processor/LoginProcessor.java b/im-server/src/main/java/com/bx/imserver/netty/processor/LoginProcessor.java index dae9746..e9da8bc 100644 --- a/im-server/src/main/java/com/bx/imserver/netty/processor/LoginProcessor.java +++ b/im-server/src/main/java/com/bx/imserver/netty/processor/LoginProcessor.java @@ -5,6 +5,8 @@ import com.alibaba.fastjson.JSON; import com.bx.imcommon.contant.IMConstant; import com.bx.imcommon.contant.IMRedisKey; import com.bx.imcommon.enums.IMCmdType; +import com.bx.imcommon.enums.IMTerminalType; +import com.bx.imcommon.util.AppPushSessionStore; import com.bx.imcommon.model.IMLoginInfo; import com.bx.imcommon.model.IMSendInfo; import com.bx.imcommon.model.IMSessionInfo; @@ -30,6 +32,7 @@ public class LoginProcessor extends AbstractMessageProcessor { private final RedisMQTemplate redisMQTemplate; private final com.bx.imcommon.util.MemberAccessStore memberAccessStore; + private final AppPushSessionStore pushSessionStore; @Value("${jwt.accessToken.secret}") private String accessTokenSecret; @@ -81,6 +84,19 @@ public class LoginProcessor extends AbstractMessageProcessor { // 初始化心跳次数 AttributeKey heartBeatAttr = AttributeKey.valueOf(ChannelAttrKey.HEARTBEAT_TIMES); ctx.channel().attr(heartBeatAttr).set(0L); + if (IMTerminalType.APP.code().equals(terminal) && userId > 0) { + String token = IMServerGroup.serverId + ":" + ctx.channel().id().asLongText(); + String registrationId = validRegistrationId(loginInfo.getRegistrationId()); + ctx.channel().attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_SESSION_TOKEN)).set(token); + ctx.channel().attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_REGISTRATION_ID)).set(registrationId); + ctx.channel().attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_FOREGROUND)) + .set(Boolean.TRUE.equals(loginInfo.getForeground())); + try { + pushSessionStore.login(userId, token, registrationId, Boolean.TRUE.equals(loginInfo.getForeground())); + } catch (RuntimeException e) { + log.warn("记录App前台状态失败,userId:{},exceptionType:{}", userId, e.getClass().getSimpleName()); + } + } // 在redis上记录每个user的channelId，15秒没有心跳，则自动过期 String key = String.join(":", IMRedisKey.IM_USER_SERVER_ID, userId.toString(), terminal.toString()); redisMQTemplate.opsForValue().set(key, IMServerGroup.serverId, IMConstant.ONLINE_TIMEOUT_SECOND, TimeUnit.SECONDS); @@ -101,4 +117,11 @@ public class LoginProcessor extends AbstractMessageProcessor { HashMap map = (HashMap) o; return BeanUtil.fillBeanWithMap(map, new IMLoginInfo(), false); } + + static String validRegistrationId(String value) { + if (value == null || value.isBlank() || value.length() > 64) { + return null; + } + return value.trim(); + } } diff --git a/im-server/src/test/java/com/bx/imserver/netty/processor/AppPushHeartbeatTest.java b/im-server/src/test/java/com/bx/imserver/netty/processor/AppPushHeartbeatTest.java new file mode 100644 index 0000000..3fb1c26 --- /dev/null +++ b/im-server/src/test/java/com/bx/imserver/netty/processor/AppPushHeartbeatTest.java @@ -0,0 +1,69 @@ +package com.bx.imserver.netty.processor; + +import com.bx.imcommon.model.IMHeartbeatInfo; +import com.bx.imcommon.mq.RedisMQTemplate; +import com.bx.imcommon.util.AppPushSessionStore; +import com.bx.imcommon.util.MemberAccessStore; +import com.bx.imserver.constant.ChannelAttrKey; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.AttributeKey; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +class AppPushHeartbeatTest { + private final RedisMQTemplate redis = mock(RedisMQTemplate.class); + private final MemberAccessStore access = mock(MemberAccessStore.class); + private final AppPushSessionStore sessions = mock(AppPushSessionStore.class); + private final EmbeddedChannel channel = new EmbeddedChannel(new ChannelInboundHandlerAdapter()); + private final HeartbeatProcessor processor = new HeartbeatProcessor(redis, access, sessions); + + @BeforeEach + void setUp() { + channel.attr(AttributeKey.valueOf(ChannelAttrKey.USER_ID)).set(42L); + channel.attr(AttributeKey.valueOf(ChannelAttrKey.AUTH_VERSION)).set(0L); + channel.attr(AttributeKey.valueOf(ChannelAttrKey.HEARTBEAT_TIMES)).set(0L); + when(access.accepts(42L, 0L)).thenReturn(true); + } + + @AfterEach + void tearDown() { + channel.finishAndReleaseAll(); + } + + @Test + void backgroundTransitionClearsForegroundWithoutClosingSocket() { + channel.attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_SESSION_TOKEN)).set("server:channel"); + channel.attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_REGISTRATION_ID)).set("reg-a"); + channel.attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_FOREGROUND)).set(true); + IMHeartbeatInfo heartbeat = new IMHeartbeatInfo(); + heartbeat.setForeground(false); + processor.process(channel.pipeline().firstContext(), heartbeat); + verify(sessions).heartbeat(42L, "server:channel", "reg-a", false, 1); + assertTrue(channel.isActive()); + } + + @Test + void registrationAvailableAfterLoginCanBeAttachedToCurrentSession() { + channel.attr(AttributeKey.valueOf(ChannelAttrKey.PUSH_SESSION_TOKEN)).set("server:channel"); + IMHeartbeatInfo heartbeat = new IMHeartbeatInfo(); + heartbeat.setForeground(true); + heartbeat.setRegistrationId("reg-new"); + processor.process(channel.pipeline().firstContext(), heartbeat); + verify(sessions).heartbeat(42L, "server:channel", "reg-new", true, 1); + processor.process(channel.pipeline().firstContext(), new IMHeartbeatInfo()); + verify(sessions).heartbeat(42L, "server:channel", "reg-new", true, 2); + } + + @Test + void legacyHeartbeatKeepsWorkingWithoutPushPresence() { + processor.process(channel.pipeline().firstContext(), new IMHeartbeatInfo()); + assertTrue(channel.isActive()); + assertNotNull(channel.readOutbound()); + verifyNoInteractions(sessions); + } +} diff --git a/im-server/src/test/java/com/bx/imserver/netty/processor/MemberAccessConnectionTest.java b/im-server/src/test/java/com/bx/imserver/netty/processor/MemberAccessConnectionTest.java index 681dbfe..5e8af56 100644 --- a/im-server/src/test/java/com/bx/imserver/netty/processor/MemberAccessConnectionTest.java +++ b/im-server/src/test/java/com/bx/imserver/netty/processor/MemberAccessConnectionTest.java @@ -35,7 +35,7 @@ class MemberAccessConnectionTest { void heartbeatRejectsRevokedSessionAndClosesServerConnection() { channel.attr(AttributeKey.valueOf(ChannelAttrKey.USER_ID)).set(131L); channel.attr(AttributeKey.valueOf(ChannelAttrKey.AUTH_VERSION)).set(0L); - new HeartbeatProcessor(redis, access).process(channel.pipeline().firstContext(), new IMHeartbeatInfo()); + new HeartbeatProcessor(redis, access, mock(com.bx.imcommon.util.AppPushSessionStore.class)).process(channel.pipeline().firstContext(), new IMHeartbeatInfo()); assertFalse(channel.isActive()); IMSendInfo reply = channel.readOutbound(); assertEquals(IMCmdType.FORCE_LOGUT.code(), reply.getCmd()); @@ -63,7 +63,7 @@ class MemberAccessConnectionTest { void redisFailureStopsHeartbeatEligibility() { channel.attr(AttributeKey.valueOf(ChannelAttrKey.USER_ID)).set(131L); when(access.accepts(131L, null)).thenThrow(new IllegalStateException("redis down")); - new HeartbeatProcessor(redis, access).process(channel.pipeline().firstContext(), new IMHeartbeatInfo()); + new HeartbeatProcessor(redis, access, mock(com.bx.imcommon.util.AppPushSessionStore.class)).process(channel.pipeline().firstContext(), new IMHeartbeatInfo()); assertFalse(channel.isActive()); }