feat: Implement Redis-based WebSocket broadcasting for multi-pod support

commons/pom.xml

@@ -75,6 +75,12 @@
             <artifactId>spring-boot-starter-data-redis</artifactId>
         </dependency>
 
+        <!-- Micrometer for Redis metrics -->
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-core</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.mapstruct</groupId>
             <artifactId>mapstruct</artifactId>

commons/src/main/java/com/tomassirio/wanderer/commons/config/RedisConfig.java

@@ -1,10 +1,13 @@
 package com.tomassirio.wanderer.commons.config;
 
+import io.lettuce.core.resource.ClientResources;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 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.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.StringRedisTemplate;
 
@@ -18,11 +21,22 @@ public class RedisConfig {
     @Value("${spring.data.redis.port:6379}")
     private int redisPort;
 
+    @Autowired(required = false)
+    private ClientResources clientResources;
+
     @Bean
     public RedisConnectionFactory redisConnectionFactory() {
         RedisStandaloneConfiguration config =
                 new RedisStandaloneConfiguration(redisHost, redisPort);
-        return new LettuceConnectionFactory(config);
+
+        LettuceClientConfiguration.LettuceClientConfigurationBuilder clientConfigBuilder =
+                LettuceClientConfiguration.builder();
+
+        if (clientResources != null) {
+            clientConfigBuilder.clientResources(clientResources);
+        }
+
+        return new LettuceConnectionFactory(config, clientConfigBuilder.build());
     }
 
     @Bean

commons/src/main/java/com/tomassirio/wanderer/commons/config/RedisMetricsConfig.java

@@ -0,0 +1,27 @@
+package com.tomassirio.wanderer.commons.config;
+
+import io.lettuce.core.metrics.MicrometerCommandLatencyRecorder;
+import io.lettuce.core.metrics.MicrometerOptions;
+import io.lettuce.core.resource.ClientResources;
+import io.micrometer.core.instrument.MeterRegistry;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/** Redis metrics configuration for Prometheus. */
+@Configuration
+@ConditionalOnProperty(
+        value = "management.metrics.redis.enabled",
+        havingValue = "true",
+        matchIfMissing = true)
+public class RedisMetricsConfig {
+
+    @Bean(destroyMethod = "shutdown")
+    public ClientResources clientResources(MeterRegistry meterRegistry) {
+        return ClientResources.builder()
+                .commandLatencyRecorder(
+                        new MicrometerCommandLatencyRecorder(
+                                meterRegistry, MicrometerOptions.create()))
+                .build();
+    }
+}

wanderer-auth/src/main/resources/application.properties

@@ -46,6 +46,7 @@ management.endpoint.health.show-details=always
 management.metrics.export.prometheus.enabled=true
 management.metrics.enable.cache=true
 management.metrics.cache.instrument-cache=true
+management.metrics.redis.enabled=true
 
 # Bootstrap Admin Configuration
 # Set via environment variables per GitHub environment

wanderer-command/src/main/java/com/tomassirio/wanderer/command/config/RedisWebSocketConfig.java

@@ -0,0 +1,66 @@
+package com.tomassirio.wanderer.command.config;
+
+import com.tomassirio.wanderer.command.websocket.RedisWebSocketMessageListener;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.PatternTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+/**
+ * Configuration for Redis-based WebSocket message broadcasting across multiple pods.
+ *
+ * <p>This configuration sets up: - Redis Pub/Sub listener for receiving WebSocket broadcast
+ * messages - Pattern subscription to all websocket:* channels
+ */
+@Configuration
+@RequiredArgsConstructor
+public class RedisWebSocketConfig {
+
+    private final RedisWebSocketMessageListener redisWebSocketMessageListener;
+
+    /** Creates a Redis message listener container that listens for WebSocket broadcast messages. */
+    @Bean
+    public RedisMessageListenerContainer redisMessageListenerContainer(
+            RedisConnectionFactory connectionFactory) {
+
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(connectionFactory);
+
+        // Subscribe to all websocket:* channels using PatternTopic for wildcard matching
+        MessageListenerAdapter adapter = new MessageListenerAdapter(redisWebSocketMessageListener);
+        container.addMessageListener(adapter, new PatternTopic("websocket:*"));
+
+        return container;
+    }
+
+    /**
+     * Redis template for publishing WebSocket messages. Uses String serializer for both keys and
+     * values since we're sending JSON. Marked as @Primary to be the default RedisTemplate<String,
+     * String> bean.
+     */
+    @Bean
+    @Primary
+    public RedisTemplate<String, String> webSocketRedisTemplate(
+            RedisConnectionFactory connectionFactory) {
+
+        RedisTemplate<String, String> template = new RedisTemplate<>();
+        template.setConnectionFactory(connectionFactory);
+
+        // Use String serializers for both keys and values
+        StringRedisSerializer stringSerializer = new StringRedisSerializer();
+        template.setKeySerializer(stringSerializer);
+        template.setValueSerializer(stringSerializer);
+        template.setHashKeySerializer(stringSerializer);
+        template.setHashValueSerializer(stringSerializer);
+
+        template.afterPropertiesSet();
+        return template;
+    }
+}

wanderer-command/src/main/java/com/tomassirio/wanderer/command/websocket/RedisWebSocketBroadcaster.java

@@ -0,0 +1,48 @@
+package com.tomassirio.wanderer.command.websocket;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * Broadcasts WebSocket messages across multiple pods using Redis Pub/Sub.
+ *
+ * <p>When a message needs to be broadcast to a topic, this component publishes it to Redis. All
+ * pods (including the sender) subscribe to Redis and deliver messages to their local WebSocket
+ * sessions.
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class RedisWebSocketBroadcaster {
+
+    private final RedisTemplate<String, String> redisTemplate;
+
+    /**
+     * Publishes a WebSocket message to Redis for multi-pod broadcasting.
+     *
+     * @param topic the WebSocket topic (e.g., "/topic/trips/123")
+     * @param message the JSON message to broadcast
+     */
+    public void publishToRedis(String topic, String message) {
+        try {
+            String redisChannel = getRedisChannel(topic);
+            redisTemplate.convertAndSend(redisChannel, message);
+            log.debug("Published message to Redis channel: {}", redisChannel);
+        } catch (Exception e) {
+            log.error("Error publishing message to Redis for topic: {}", topic, e);
+        }
+    }
+
+    /**
+     * Converts a WebSocket topic to a Redis channel name.
+     *
+     * @param topic the WebSocket topic
+     * @return the Redis channel name
+     */
+    public static String getRedisChannel(String topic) {
+        // Use a prefix to avoid conflicts with other Redis keys
+        return "websocket:" + topic;
+    }
+}

wanderer-command/src/main/java/com/tomassirio/wanderer/command/websocket/RedisWebSocketMessageListener.java

@@ -0,0 +1,57 @@
+package com.tomassirio.wanderer.command.websocket;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * Listens for WebSocket messages published to Redis and broadcasts them to local WebSocket
+ * sessions.
+ *
+ * <p>This listener receives messages from all pods (including the same pod that published). When a
+ * message arrives, it extracts the WebSocket topic and broadcasts to local sessions subscribed to
+ * that topic.
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class RedisWebSocketMessageListener implements MessageListener {
+
+    private final WebSocketSessionManager sessionManager;
+
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+        try {
+            String channel = new String(message.getChannel());
+            String messageBody = new String(message.getBody());
+
+            // Extract the WebSocket topic from the Redis channel name
+            // Channel format: "websocket:/topic/trips/123"
+            String topic = extractTopicFromChannel(channel);
+
+            log.debug("Received Redis message for topic: {} (channel: {})", topic, channel);
+
+            // Broadcast to local WebSocket sessions subscribed to this topic
+            sessionManager.broadcastToLocalSessions(topic, messageBody);
+
+        } catch (Exception e) {
+            log.error("Error processing Redis WebSocket message", e);
+        }
+    }
+
+    /**
+     * Extracts the WebSocket topic from the Redis channel name.
+     *
+     * @param channel the Redis channel (e.g., "websocket:/topic/trips/123")
+     * @return the WebSocket topic (e.g., "/topic/trips/123")
+     */
+    private String extractTopicFromChannel(String channel) {
+        // Remove the "websocket:" prefix
+        if (channel.startsWith("websocket:")) {
+            return channel.substring("websocket:".length());
+        }
+        return channel;
+    }
+}

wanderer-command/src/main/java/com/tomassirio/wanderer/command/websocket/WebSocketEventService.java

@@ -9,21 +9,24 @@
 import org.springframework.stereotype.Service;
 
 /**
- * Service for broadcasting WebSocket events to subscribers.
+ * Service for broadcasting WebSocket events to subscribers across all pods.
  *
  * <p>This service provides a simple interface for broadcasting events that implement {@link
  * Broadcastable}. The event itself knows its topic, event type, and payload.
+ *
+ * <p>Messages are broadcast to all pods via Redis Pub/Sub, ensuring that all connected clients
+ * receive updates regardless of which pod they're connected to.
  */
 @Slf4j
 @Service
 @RequiredArgsConstructor
 public class WebSocketEventService {
 
-    private final WebSocketSessionManager sessionManager;
+    private final RedisWebSocketBroadcaster redisBroadcaster;
     private final ObjectMapper objectMapper;
 
     /**
-     * Broadcasts a Broadcastable event to its designated topic.
+     * Broadcasts a Broadcastable event to its designated topic across all pods.
      *
      * @param event the event to broadcast (must implement {@link Broadcastable})
      */
@@ -35,12 +38,12 @@ public void broadcast(Broadcastable event) {
 
         try {
             String message = objectMapper.writeValueAsString(wsEvent);
-            sessionManager.broadcast(topic, message);
-            log.info(
-                    "Broadcast {} event to {} ({} subscribers)",
-                    event.getEventType(),
-                    topic,
-                    sessionManager.getSubscribersCount(topic));
+
+            // Publish to Redis - all pods (including this one) will receive and broadcast to their
+            // local sessions
+            redisBroadcaster.publishToRedis(topic, message);
+
+            log.info("Published {} event to Redis for topic {}", event.getEventType(), topic);
         } catch (JsonProcessingException e) {
             log.error("Error serializing WebSocket event: {}", event.getEventType(), e);
         }

wanderer-command/src/main/java/com/tomassirio/wanderer/command/websocket/WebSocketSessionManager.java

@@ -20,7 +20,7 @@ public class WebSocketSessionManager {
 
     // sessionId -> userId (null not allowed in ConcurrentHashMap, so we use Optional)
     private final Map<String, UUID> sessionUsers = new ConcurrentHashMap<>();
-    
+
     // sessionId -> anonymous flag (since ConcurrentHashMap doesn't support null values)
     private final Map<String, Boolean> sessionAnonymousFlags = new ConcurrentHashMap<>();
 
@@ -36,15 +36,15 @@ public class WebSocketSessionManager {
 
     public void registerSession(WebSocketSession session, UUID userId) {
         sessions.put(session.getId(), session);
-        
+
         // ConcurrentHashMap doesn't allow null values, so track anonymous status separately
         boolean isAnonymous = (userId == null);
         if (!isAnonymous) {
             sessionUsers.put(session.getId(), userId);
         }
         sessionAnonymousFlags.put(session.getId(), isAnonymous);
         sessionSubscriptionCounts.put(session.getId(), 0);
-        
+
         log.info(
                 "Registered session: {} for user: {} (anonymous: {})",
                 session.getId(),
@@ -86,10 +86,7 @@ public void subscribe(WebSocketSession session, String topic) {
 
         topicSubscriptions.computeIfAbsent(topic, k -> new CopyOnWriteArraySet<>()).add(sessionId);
         log.debug(
-                "Session {} subscribed to topic {} (anonymous: {})",
-                sessionId,
-                topic,
-                isAnonymous);
+                "Session {} subscribed to topic {} (anonymous: {})", sessionId, topic, isAnonymous);
     }
 
     public void unsubscribe(WebSocketSession session, String topic) {
@@ -110,23 +107,30 @@ public void unsubscribe(WebSocketSession session, String topic) {
         }
     }
 
-    public void broadcast(String topic, String message) {
+    /**
+     * Broadcasts a message to LOCAL sessions subscribed to the topic. This method is called by the
+     * Redis listener when a message is received from Redis.
+     *
+     * @param topic the WebSocket topic
+     * @param message the message to broadcast
+     */
+    public void broadcastToLocalSessions(String topic, String message) {
         Set<String> subscribers = topicSubscriptions.get(topic);
         if (subscribers == null || subscribers.isEmpty()) {
-            log.debug("No subscribers for topic: {}", topic);
+            log.debug("No local subscribers for topic: {}", topic);
             return;
         }
 
-        log.debug("Broadcasting to {} subscribers on topic: {}", subscribers.size(), topic);
+        log.debug("Broadcasting to {} local subscribers on topic: {}", subscribers.size(), topic);
 
         for (String sessionId : subscribers) {
             WebSocketSession session = sessions.get(sessionId);
             if (session != null && session.isOpen()) {
                 try {
                     session.sendMessage(new TextMessage(message));
-                    log.debug("Sent message to session: {}", sessionId);
+                    log.debug("Sent message to local session: {}", sessionId);
                 } catch (IOException e) {
-                    log.error("Error sending message to session: {}", sessionId, e);
+                    log.error("Error sending message to local session: {}", sessionId, e);
                 }
             } else {
                 log.warn("Session {} is not open, removing from subscriptions", sessionId);
@@ -135,6 +139,15 @@ public void broadcast(String topic, String message) {
         }
     }
 
+    /**
+     * @deprecated Use broadcastToLocalSessions instead. This method is kept for backward
+     *     compatibility but will be removed once all callers are updated to use Redis broadcasting.
+     */
+    @Deprecated
+    public void broadcast(String topic, String message) {
+        broadcastToLocalSessions(topic, message);
+    }
+
     public UUID getUserId(WebSocketSession session) {
         return sessionUsers.get(session.getId());
     }