111

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';

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;
+}

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;
 }

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;
 }

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<Long> 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<Long> 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<List> 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<IMAppForegroundDevice> 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<String> keys(Long userId) {
+        return List.of(SESSION_PREFIX + userId, FOREGROUND_PREFIX + userId);
+    }
+}

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);
+    }
+}

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<ThirdPrivateMessagePushContextVO> context(@Valid @RequestBody ThirdPrivateMessagePushContextDTO request) {
+        return ResultUtils.success(service.getContext(request.getMessageId()));
+    }
+}

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;
+}

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;
+}

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<PrivateMessagePushEvent> {
+    @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<PrivateMessagePushEvent> 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);
+}

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");
+            }
+        }
+    }
+}

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<User> 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<Friend>()
+                .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);
+    }
+}

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;
+    }
+}

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<PrivateMessageMapper,
     private final IMClient imClient;
     private final SensitiveFilterUtil sensitiveFilterUtil;
     private final PrivateMessageModerationService moderationService;
+    private final PrivateMessagePushOutboxService pushOutboxService;
     private static final ScheduledThreadPoolExecutor EXECUTOR = ThreadPoolExecutorFactory.getThreadPoolExecutor();
 
     @Override
+    @Transactional(rollbackFor = Exception.class)
     public PrivateMessageVO sendMessage(PrivateMessageDTO dto) {
         PrivateMessageModerationService.rejectReservedTmpId(dto.getTmpId());
         PrivateMessageSessionControl.rejectReservedKickContent(dto.getContent());
@@ -70,6 +75,7 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
             msg.setContent(sensitiveFilterUtil.filter(dto.getContent()));
         }
         this.save(msg);
+        pushOutboxService.enqueue(msg);
         // 推送消息
         PrivateMessageVO msgInfo = BeanUtils.copyProperties(msg, PrivateMessageVO.class);
         IMPrivateMessage<PrivateMessageVO> sendMessage = new IMPrivateMessage<>();
@@ -78,11 +84,29 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
         sendMessage.setSendToSelf(true);
         sendMessage.setData(msgInfo);
         sendMessage.setSendResult(true);
-        imClient.sendPrivateMessage(sendMessage);
+        if (TransactionSynchronizationManager.isSynchronizationActive()) {
+            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+                @Override
+                public void afterCommit() {
+                    publishNewMessage(sendMessage);
+                }
+            });
+        } else {
+            publishNewMessage(sendMessage);
+        }
         log.info("发送私聊消息，发送id:{},接收id:{}，内容:{}", session.getUserId(), dto.getRecvId(), dto.getContent());
         return msgInfo;
     }
 
+    private void publishNewMessage(IMPrivateMessage<PrivateMessageVO> 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) {

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<PrivateMessageMapper
     private final IMClient imClient;
     private final FriendService friendService;
     private final SensitiveFilterUtil sensitiveFilterUtil;
+    private final PrivateMessagePushOutboxService pushOutboxService;
 
     @Override
     @Transactional(readOnly = true)
@@ -118,6 +120,7 @@ public class ThirdPrivateMsgServiceImpl extends ServiceImpl<PrivateMessageMapper
             }
         }
         this.save(msg);
+        pushOutboxService.enqueue(msg);
         // 推送消息
         PrivateMessageVO msgInfo = BeanUtils.copyProperties(msg, PrivateMessageVO.class);
         IMPrivateMessage<PrivateMessageVO> sendMessage = new IMPrivateMessage<>();
@@ -131,16 +134,25 @@ public class ThirdPrivateMsgServiceImpl extends ServiceImpl<PrivateMessageMapper
                     new org.springframework.transaction.support.TransactionSynchronization() {
                         @Override
                         public void afterCommit() {
-                            imClient.sendPrivateMessage(sendMessage);
+                            publishNewMessage(sendMessage);
                         }
                     });
         } else {
-            imClient.sendPrivateMessage(sendMessage);
+            publishNewMessage(sendMessage);
         }
         log.info("发送私聊消息，发送id:{},接收id:{}，内容:{}", dto.getSendId(), dto.getRecvId(), dto.getContent());
         return msgInfo;
     }
 
+    private void publishNewMessage(IMPrivateMessage<PrivateMessageVO> 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.

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);
+    }
+}

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<IMAppForegroundDevice> foregroundDevices = List.of();
+}

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:}

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);
+    }
+}

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<IMAppForegroundDevice> devices = store.getForegroundDevices(id);
+        assertEquals(1, devices.size());
+        return devices.get(0);
+    }
+}

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<String> body = new AtomicReference<>();
+    private final AtomicReference<String> signature = new AtomicReference<>();
+    private final AtomicReference<String> 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("{}"));
+    }
+}

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<PrivateMessage> query = ReflectionTestUtils.invokeMethod(history, "buildOfflineQuery", 42L, 999999L);
         List<PrivateMessage> messages = mapper.selectList(query);
         assertEquals(1, messages.size());

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();

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<Wrapper<Friend>> 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;
+    }
+}

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<PrivateMessagePushEvent> 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;
+    }
+}

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<String> 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<String> 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;
+    }
+}

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> 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;
+        }
+    }
+}

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);

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')");

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);
     }

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);
     }
 

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;
+    }
+}

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() {
     }

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<IMSendInfo> {
         if (userId == null || terminal == null) {
             return;
         }
+        String pushToken = ctx.channel().attr(AttributeKey.<String>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())) {

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<IMHeartbeatInfo
 
     private final RedisMQTemplate redisMQTemplate;
     private final com.bx.imcommon.util.MemberAccessStore memberAccessStore;
+    private final AppPushSessionStore pushSessionStore;
 
     @Override
     public void process(ChannelHandlerContext ctx, IMHeartbeatInfo beatInfo) {
@@ -50,6 +52,23 @@ public class HeartbeatProcessor extends AbstractMessageProcessor<IMHeartbeatInfo
         AttributeKey<Long> 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.<String>valueOf(ChannelAttrKey.PUSH_SESSION_TOKEN)).get();
+        if (pushToken != null) {
+            AttributeKey<String> registrationAttr = AttributeKey.valueOf(ChannelAttrKey.PUSH_REGISTRATION_ID);
+            AttributeKey<Boolean> 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<Long> userIdAttr = AttributeKey.valueOf(ChannelAttrKey.USER_ID);

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<IMLoginInfo> {
 
     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<IMLoginInfo> {
         // 初始化心跳次数
         AttributeKey<Long> 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.<String>valueOf(ChannelAttrKey.PUSH_SESSION_TOKEN)).set(token);
+            ctx.channel().attr(AttributeKey.<String>valueOf(ChannelAttrKey.PUSH_REGISTRATION_ID)).set(registrationId);
+            ctx.channel().attr(AttributeKey.<Boolean>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<IMLoginInfo> {
         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();
+    }
 }

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.<Long>valueOf(ChannelAttrKey.USER_ID)).set(42L);
+        channel.attr(AttributeKey.<Long>valueOf(ChannelAttrKey.AUTH_VERSION)).set(0L);
+        channel.attr(AttributeKey.<Long>valueOf(ChannelAttrKey.HEARTBEAT_TIMES)).set(0L);
+        when(access.accepts(42L, 0L)).thenReturn(true);
+    }
+
+    @AfterEach
+    void tearDown() {
+        channel.finishAndReleaseAll();
+    }
+
+    @Test
+    void backgroundTransitionClearsForegroundWithoutClosingSocket() {
+        channel.attr(AttributeKey.<String>valueOf(ChannelAttrKey.PUSH_SESSION_TOKEN)).set("server:channel");
+        channel.attr(AttributeKey.<String>valueOf(ChannelAttrKey.PUSH_REGISTRATION_ID)).set("reg-a");
+        channel.attr(AttributeKey.<Boolean>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.<String>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);
+    }
+}

im-server/src/test/java/com/bx/imserver/netty/processor/MemberAccessConnectionTest.java

@@ -35,7 +35,7 @@ class MemberAccessConnectionTest {
     void heartbeatRejectsRevokedSessionAndClosesServerConnection() {
         channel.attr(AttributeKey.<Long>valueOf(ChannelAttrKey.USER_ID)).set(131L);
         channel.attr(AttributeKey.<Long>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.<Long>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());
     }