FCM 메시지를 받을 유저 필터링 로직 구현

src/backend/state_server/build.gradle

@@ -24,12 +24,10 @@ repositories {
 }
 
 dependencies {
-    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
     implementation 'org.springframework.boot:spring-boot-starter-data-redis'
     implementation 'org.springframework.boot:spring-boot-starter-web'
     implementation 'org.springframework.kafka:spring-kafka'
     implementation group: 'com.google.code.gson', name: 'gson', version: '2.8.6'
-    runtimeOnly 'org.postgresql:postgresql:42.7.4'
     compileOnly 'org.projectlombok:lombok'
     annotationProcessor 'org.projectlombok:lombok'
     annotationProcessor "org.springframework.boot:spring-boot-configuration-processor"

src/backend/state_server/src/main/java/com/jootalkpia/state_server/StateServerApplication.java

@@ -2,7 +2,9 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
 
+@ConfigurationPropertiesScan
 @SpringBootApplication
 public class StateServerApplication {
 

src/backend/state_server/src/main/java/com/jootalkpia/state_server/config/RedisConfig.java

@@ -0,0 +1,46 @@
+package com.jootalkpia.state_server.config;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+@RequiredArgsConstructor
+@Configuration
+public class RedisConfig {
+    private final RedisProperties redisProperties;
+
+    @Bean
+    public RedisConnectionFactory redisConnectionFactory() {
+        return new LettuceConnectionFactory(redisProperties.host(), redisProperties.port());
+    }
+    @Bean
+    public RedisTemplate<String, String> stringOperRedisTemplate(RedisConnectionFactory connectionFactory) {
+        RedisTemplate<String, String> template = new RedisTemplate<>();
+        template.setConnectionFactory(connectionFactory);
+
+        template.setKeySerializer(new StringRedisSerializer());
+        template.setValueSerializer(new StringRedisSerializer());
+
+        return template;
+    }
+
+    @Bean
+    public RedisTemplate<String, Object> objectOperRedisTemplate(RedisConnectionFactory connectionFactory) {
+        RedisTemplate<String, Object> template = new RedisTemplate<>();
+        template.setConnectionFactory(connectionFactory);
+
+        template.setKeySerializer(new StringRedisSerializer());
+        template.setHashKeySerializer(new StringRedisSerializer());
+
+        Jackson2JsonRedisSerializer<Object> jsonSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
+        template.setValueSerializer(jsonSerializer);
+        template.setHashValueSerializer(jsonSerializer);
+
+        return template;
+    }
+}

src/backend/state_server/src/main/java/com/jootalkpia/state_server/config/RedisProperties.java

@@ -0,0 +1,11 @@
+package com.jootalkpia.state_server.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "spring.data.redis")
+public record RedisProperties(
+        String host,
+        int port
+) {
+}
+

src/backend/state_server/src/main/java/com/jootalkpia/state_server/entity/RedisKeys.java

@@ -0,0 +1,21 @@
+package com.jootalkpia.state_server.entity;
+
+public class RedisKeys {
+    private static final String SESSION_PREFIX = "sessions";
+    private static final String USER_PREFIX = "user";
+    private static final String CHANNEL_PREFIX = "channel";
+    private static final String SUBSCRIBERS_PREFIX = "subscribers";
+    private static final String ACTIVE_PREFIX = "active";
+
+    public static String userSessions(String userId) {
+        return USER_PREFIX + ":" + userId + ":" + SESSION_PREFIX;
+    }
+
+    public static String channelActive(String channelId) {
+        return CHANNEL_PREFIX + ":" + channelId + ":" + ACTIVE_PREFIX;
+    }
+
+    public static String channelSubscribers(String channelId) {
+        return CHANNEL_PREFIX + ":" + channelId + ":" + SUBSCRIBERS_PREFIX;
+    }
+}

src/backend/state_server/src/main/java/com/jootalkpia/state_server/service/KafkaConsumer.java

@@ -1,14 +1,17 @@
 package com.jootalkpia.state_server.service;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.jootalkpia.state_server.ChatMessageToKafka;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Service;
 
 @Service
 @Slf4j
+@RequiredArgsConstructor
 public class KafkaConsumer {
+    private final StateService stateService;
 
     @KafkaListener(
             topics = "${topic.chat}",
@@ -20,11 +23,12 @@ public void processState(String kafkaMessage) {
         ObjectMapper mapper = new ObjectMapper();
 
         try {
-            ChatMessageToKafka chatMessageToKafka = mapper.readValue(kafkaMessage, ChatMessageToKafka.class);
+            JsonNode rootNode = mapper.readTree(kafkaMessage);
+            JsonNode commonNode = rootNode.get("common");
+            JsonNode messagesNode = rootNode.get("message");
 
-            // 유저 상태 검증 로직
+            stateService.findNotificationTargets(commonNode.get("channelId").asText());
 
-            log.info("dto ===> " + chatMessageToKafka.toString());
         } catch (Exception ex) {
             log.error(ex.getMessage(), ex); // 추후에 GlobalException 처리
         }

src/backend/state_server/src/main/java/com/jootalkpia/state_server/service/StateService.java

@@ -0,0 +1,80 @@
+package com.jootalkpia.state_server.service;
+
+import com.jootalkpia.state_server.entity.RedisKeys;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class StateService {
+    private final RedisTemplate<String, String> stringOperRedisTemplate;
+    private final RedisTemplate<String, Object> objectOperRedisTemplate;
+
+    public Set<String> findNotificationTargets(String channelId) {
+        Set<String> subscriber = findSubscribers(channelId);
+        Set<String> onlineSessions = findOnlineSessions(subscriber);
+        Set<String> activeSessions = findActiveSessions(channelId, subscriber);
+
+        log.info("[Notification Targets] Channel: {}", channelId);
+        log.info("├── Online Sessions: {}", onlineSessions);
+        log.info("├── Active Sessions: {}", activeSessions);
+
+        onlineSessions.removeAll(activeSessions);
+        log.info("└── Target Sessions: {}", onlineSessions);
+
+        return onlineSessions;
+    }
+
+    private Set<String> findSubscribers(String channelId) {
+        return stringOperRedisTemplate.opsForSet().members(RedisKeys.channelSubscribers(channelId));
+    }
+
+    private Set<String> findOnlineSessions(Set<String> subscriber) {
+        Set<String> onlineSessions = new HashSet<>();
+
+        for (String userId : subscriber) {
+            Set<String> userSessions = stringOperRedisTemplate.opsForSet()
+                    .members(RedisKeys.userSessions(userId));
+            if (userSessions != null) {
+                onlineSessions.addAll(userSessions);
+            }
+        }
+
+        return onlineSessions;
+    }
+
+    private Set<String> findActiveSessions(String channelId, Set<String> subscriber) {
+        Set<String> activeSessions = new HashSet<>();
+
+        for (String userId : subscriber) {
+            Object userActiveSessions = objectOperRedisTemplate.opsForHash()
+                    .get(RedisKeys.channelActive(channelId), userId);
+
+            if (userActiveSessions != null) {
+                Set<String> sessions = convertToSet(userActiveSessions);
+                sessions.forEach(session ->
+                        activeSessions.add("\"" + session + "\""));
+            }
+        }
+
+        return activeSessions;
+    }
+
+    @SuppressWarnings("unchecked")
+    private Set<String> convertToSet(Object storedValue) {
+        if (storedValue instanceof Set) {
+            return (Set<String>) storedValue;
+        } else if (storedValue instanceof List) {
+            return new HashSet<>((List<String>) storedValue);
+        } else {
+            return new HashSet<>();
+        }
+    }
+}

src/backend/state_server/src/main/resources/application.yml

@@ -8,12 +8,15 @@ spring:
       acks: all # 멱등성 프로듀서를 위해 all로 설정, 0 또는 1이면 enable.idempotence=true 불가
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.apache.kafka.common.serialization.StringSerializer
+  data:
+    redis:
+      host: ${REDIS_HOST}
+      port: ${REDIS_PORT}
 
 topic:
-  chat: jootalkpia.chat.prd.notification
+  chat: jootalkpia.chat.prd.message
 group:
   status: user-status-handle-consumer-group
 
-
 server:
   port: ${STATE_PORT}