User online status

.gitignore

@@ -14,7 +14,6 @@ target/
 .env
 
 ### IntelliJ IDEA ###
-.idea
 *.iws
 *.iml
 *.ipr

common/src/main/java/com/ligg/common/constants/Constants.java

@@ -124,6 +124,28 @@ public class Constants {
      */
     public static final String AUTH_USER_SESSIONS_KEY = "animeflow:auth:user:sessions";
 
+    /** 在线状态单设备记录，完整键为 animeflow:presence:{presenceId}。 */
+    public static final String PRESENCE_KEY = "animeflow:presence";
+
+    /** 在线设备索引，成员为 presenceId，score 为最近心跳时间戳。 */
+    public static final String PRESENCE_DEVICES_KEY = "animeflow:presence:devices";
+
+    /** 在线用户去重索引，成员为 user:{id} 或 visitor:{visitorId}。 */
+    public static final String PRESENCE_USERS_KEY = "animeflow:presence:users";
+
+    /** 番剧在线设备索引，完整键为 animeflow:presence:subject:{subjectId}:devices。 */
+    public static final String PRESENCE_SUBJECT_KEY = "animeflow:presence:subject";
+
+    /** 正在播放番剧索引，成员为 subjectId，score 为最近观看时间戳。 */
+    public static final String PRESENCE_WATCHING_SUBJECTS_KEY = "animeflow:presence:watching-subjects";
+
+    /** 正在播放番剧排名索引，成员为 subjectId，score 为去重后的在线用户数。 */
+    public static final String PRESENCE_WATCHING_SUBJECTS_RANK_KEY =
+            "animeflow:presence:watching-subjects:rank";
+
+    /** Presence 状态变化通知频道。 */
+    public static final String PRESENCE_CHANGED_CHANNEL = "animeflow:presence:changed";
+
     /**
      * user_oauth.platform 字段：Bangumi 第三方登录
      */

common/src/main/java/com/ligg/common/handler/GlobalExceptionHandler.java

@@ -28,6 +28,7 @@
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.RestControllerAdvice;
+import org.springframework.web.context.request.async.AsyncRequestTimeoutException;
 import org.springframework.web.method.annotation.MethodArgumentTypeMismatchException;
 
 /**
@@ -224,6 +225,14 @@ public Result<Void> handleVerificationCode(VerificationCodeException e) {
         return Result.error(ResponseCode.PARAM_ERROR, e.getMessage());
     }
 
+    /**
+     * SSE 连接超时或被容器关闭。
+     */
+    @ExceptionHandler(AsyncRequestTimeoutException.class)
+    public void handleAsyncRequestTimeout(AsyncRequestTimeoutException e) {
+        log.debug("SSE connection timed out or was closed");
+    }
+
     /**
      * 兜底异常处理
      */

db/anime_flow.sql

@@ -416,7 +416,8 @@ CREATE TABLE `user_bgm_collection_sync_task` (
   UNIQUE KEY `uk_collection_sync_active_user` (`active_user_id`),
   PRIMARY KEY (`id`), UNIQUE KEY `uk_collection_sync_request` (`user_id`,`request_id`),
   KEY `idx_collection_sync_active` (`user_id`,`status`,`id`),
-  KEY `idx_collection_sync_cleanup` (`status`,`finished_at`,`id`)
+  KEY `idx_collection_sync_cleanup` (`status`,`finished_at`,`id`),
+  KEY `idx_sync_task_recovery` (`status`,`heartbeat_at`,`created_at`,`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户 Bangumi 收藏双向同步任务';
 
 CREATE TABLE `user_bgm_collection_sync_item` (

flow-bootstrap/src/main/java/com/ligg/flow_bootstrap/config/RedisConfig.java

@@ -12,6 +12,7 @@
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.core.script.DefaultRedisScript;
 import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
@@ -48,6 +49,24 @@ public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factor
         return template;
     }
 
+    /**
+     * Redis 字符串模板统一使用 UTF-8 字符串序列化。
+     *
+     * <p>限流、Lua 脚本、队列和 Presence 索引都使用该模板；不要让这些
+     * Redis 数据被对象 JSON 序列化器包装，否则会影响脚本参数和跨模块读取。</p>
+     */
+    @Bean
+    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
+        StringRedisTemplate template = new StringRedisTemplate(factory);
+        StringRedisSerializer serializer = new StringRedisSerializer();
+        template.setKeySerializer(serializer);
+        template.setValueSerializer(serializer);
+        template.setHashKeySerializer(serializer);
+        template.setHashValueSerializer(serializer);
+        template.afterPropertiesSet();
+        return template;
+    }
+
     @Bean
     public DefaultRedisScript<Long> ipRateLimitRedisScript() {
         DefaultRedisScript<Long> script = new DefaultRedisScript<>();

flow-bootstrap/src/main/resources/application.yaml

@@ -107,7 +107,7 @@ anime-flow:
   ## 接口限流
   rate-limit:
     enabled: true
-    max-requests: 60   #单个时间窗口内允许的最大请求次数。
+    max-requests: 100   #单个时间窗口内允许的最大请求次数。
     window-seconds: 60  #时间窗口长度（秒）。
     path-patterns: #拦截路径
       - /api/v1/**

flow-client/src/main/java/com/ligg/flowclient/config/PresenceSseConfiguration.java

@@ -0,0 +1,25 @@
+package com.ligg.flowclient.config;
+
+import com.ligg.flowclient.service.PresenceSseService;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+
+/** Presence SSE 的 Redis Pub/Sub 监听配置。 */
+@Configuration
+public class PresenceSseConfiguration {
+
+    @Bean
+    public RedisMessageListenerContainer presenceSseListener(
+            RedisConnectionFactory factory,
+            PresenceSseService presenceSseService) {
+        var container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(factory);
+        container.addMessageListener(
+                presenceSseService,
+                new ChannelTopic(PresenceSseService.CHANNEL));
+        return container;
+    }
+}

flow-client/src/main/java/com/ligg/flowclient/config/WebMvcConfig.java

@@ -56,7 +56,10 @@ public void addInterceptors(InterceptorRegistry registry) {
                 .order(2);
 
         registry.addInterceptor(optionalAuthorizationInterceptor)
-                .addPathPatterns("/api/v1/bangumi/subjects/**", "/api/v1/bangumi/search/subjects")
+                .addPathPatterns(
+                        "/api/v1/bangumi/subjects/**",
+                        "/api/v1/bangumi/search/subjects",
+                        "/api/v1/presence/**")
                 .order(3);
     }
 

flow-client/src/main/java/com/ligg/flowclient/controller/CollectionController.java

@@ -94,7 +94,6 @@ public Result<UserBgmCollectionSyncStatusVo> syncCollections(
     /**
      * 查询 Bangumi 收藏同步任务状态。
      */
-    @Deprecated
     @GetMapping("/sync")
     public Result<UserBgmCollectionSyncStatusVo> getCollectionSyncStatus(
             @RequestAttribute(AuthorizationInterceptor.ACCESS_TOKEN_REQUEST_ATTRIBUTE) String accessToken) {