ByPassingKafka to Update Pings when deployed into a single node

Author: luismr

docs/adrs/adr-001-websocket-scalability.md

@@ -6,24 +6,33 @@ The `flight-tracker-event-server` currently sends flight updates directly to cli
 ## Decision
 Refactor the application so that the `PingEventPublisher` publishes flight events to a Kafka topic. A new dedicated component will consume these events and manage active WebSocket sessions for message delivery.
 
+Additionally, we will implement feature flags to control the WebSocket delivery mechanism, allowing the system to run in different deployment configurations:
+- **Monolithic Mode**: All components run in the same service with in-memory WebSocket sessions
+- **Decoupled Mode**: WebSocket delivery runs as a separate component consuming from Kafka
+
 This decision enables separation of responsibilities and prepares the system for a future migration to a STOMP-based architecture (e.g., RabbitMQ) if scale demands increase.
 
 ## Justification
 - **Time-to-market**: quick delivery without frontend changes
 - **Decoupling**: separates event generation from delivery logic
 - **Reuses existing infrastructure**: Kafka is already in place
 - **Low impact**: avoids protocol changes or client modifications for now
+- **Deployment Flexibility**: allows gradual transition between deployment models
+- **Reduced Complexity**: initial implementation can stay within the same service
 
 ## Alternatives Considered
 - **STOMP Broker with RabbitMQ**: powerful but requires client refactor and more setup
 - **Redis Streams/PubSub**: simple, fast, but with delivery and clustering limitations
 - **Optimizing the current implementation**: fast but not future-proof
+- **Immediate Service Split**: would require more upfront work and coordination
 
 ## Consequences
 - The system becomes modular and more scalable
 - Kafka enables better backpressure and failover handling
 - Prepares for a transition to STOMP when scale justifies it
 - A new WebSocket delivery component must be monitored closely
+- Feature flags add complexity but provide deployment flexibility
+- Allows for gradual migration of WebSocket handling to a separate service
 
 ## Links
 - [Technical Analysis – WebSocket Scalability](../analysis/technical-analysis-websocket-flight-tracker.md)

src/main/java/dev/luismachadoreis/flighttracker/server/ping/infrastructure/pubsub/PingEventPublisherByPassingKafka.java

@@ -0,0 +1,40 @@
+package dev.luismachadoreis.flighttracker.server.ping.infrastructure.pubsub;
+
+import org.springframework.stereotype.Component;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.util.StringUtils;
+import org.springframework.transaction.event.TransactionalEventListener;
+import org.springframework.transaction.event.TransactionPhase;
+
+import dev.luismachadoreis.flighttracker.server.ping.infrastructure.websocket.MapUpdatesHandler;
+import dev.luismachadoreis.flighttracker.server.ping.application.dto.PingMapper;
+import dev.luismachadoreis.flighttracker.server.ping.application.dto.PingDTO;
+import dev.luismachadoreis.flighttracker.server.common.utils.JsonUtils;        
+import dev.luismachadoreis.flighttracker.server.ping.domain.PingCreatedEvent;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;   
+
+
+@Slf4j
+@Component
+@AllArgsConstructor
+@ConditionalOnExpression("${app.ping.publisher.enabled:false} == false and ${app.ping.websocket.enabled:false} == true")
+public class PingEventPublisherByPassingKafka {
+
+    private final MapUpdatesHandler mapUpdatesHandler;
+    private final PingMapper pingMapper;
+    private final JsonUtils jsonUtils;
+
+    /**
+     * Handles PingCreated events.
+     * @param event the PingCreated event
+     */ 
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)   
+    public void handlePingCreated(PingCreatedEvent event) {
+        PingDTO pingDTO = pingMapper.toDTO(event.ping());
+        String message = jsonUtils.toJson(pingDTO);
+        mapUpdatesHandler.sendMessage(message);
+    }
+
+} 

src/main/resources/application.yml

@@ -101,8 +101,7 @@ app:
     subscriber:
       enabled: true
     publisher:
-      enabled: true
+      enabled: false
     websocket:
       enabled: true
 
-  # ... rest of the file remains unchanged ... 

src/test/java/dev/luismachadoreis/flighttracker/server/ping/infrastructure/pubsub/PingEventPublisherByPassingKafkaTest.java

@@ -0,0 +1,92 @@
+package dev.luismachadoreis.flighttracker.server.ping.infrastructure.pubsub;
+
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import dev.luismachadoreis.flighttracker.server.ping.application.dto.PingDTO;
+import dev.luismachadoreis.flighttracker.server.ping.application.dto.PingMapper;
+import dev.luismachadoreis.flighttracker.server.ping.domain.Ping;
+import dev.luismachadoreis.flighttracker.server.ping.domain.PingCreatedEvent;
+import dev.luismachadoreis.flighttracker.server.ping.infrastructure.websocket.MapUpdatesHandler;
+import dev.luismachadoreis.flighttracker.server.common.utils.JsonUtils;
+
+@ExtendWith(MockitoExtension.class)
+class PingEventPublisherByPassingKafkaTest {
+
+    @Mock
+    private MapUpdatesHandler mapUpdatesHandler;
+
+    @Mock
+    private PingMapper pingMapper;
+
+    @Mock
+    private JsonUtils jsonUtils;
+
+    private PingEventPublisherByPassingKafka publisher;
+
+    @BeforeEach
+    void setUp() {
+        publisher = new PingEventPublisherByPassingKafka(mapUpdatesHandler, pingMapper, jsonUtils);
+    }
+
+    @Test
+    void handlePingCreated_ShouldSendMessageToWebSocket() {
+        // Arrange
+        Instant now = Instant.now();
+        var aircraft = new Ping.Aircraft("ABC123", "FL123", "US", now, "7700", true, new Integer[]{1, 2});
+        var vector = new Ping.Vector(500.0, 180.0, 0.0);
+        var position = new Ping.Position(10.0, 20.0, 30000.0, 29000.0, false, 1, now);
+        var ping = new Ping(aircraft, vector, position);
+        var event = new PingCreatedEvent(ping, now);
+
+        var pingDTO = new PingDTO(
+            ping.getId(),
+            new PingDTO.Aircraft(
+                aircraft.icao24(),
+                aircraft.callsign(),
+                aircraft.originCountry(),
+                aircraft.lastContact(),
+                aircraft.squawk(),
+                aircraft.spi(),
+                aircraft.sensors()
+            ),
+            new PingDTO.Vector(
+                vector.velocity(),
+                vector.trueTrack(),
+                vector.verticalRate()
+            ),
+            new PingDTO.Position(
+                position.longitude(),
+                position.latitude(),
+                position.geoAltitude(),
+                position.baroAltitude(),
+                position.onGround(),
+                position.source(),
+                position.time()
+            ),
+            ping.getLastUpdate()
+        );
+
+        String expectedJson = "{\"id\":\"" + ping.getId() + "\",\"aircraft\":{\"icao24\":\"ABC123\",\"callsign\":\"FL123\",\"origin_country\":\"US\",\"last_contact\":\"" + now + "\",\"squawk\":\"7700\",\"spi\":true,\"sensors\":[1,2]},\"vector\":{\"velocity\":500.0,\"true_track\":180.0,\"vertical_rate\":0.0},\"position\":{\"longitude\":10.0,\"latitude\":20.0,\"geo_altitude\":30000.0,\"baro_altitude\":29000.0,\"on_ground\":false,\"source\":1,\"time\":\"" + now + "\"},\"last_update\":\"" + now + "\"}";
+
+        when(pingMapper.toDTO(ping)).thenReturn(pingDTO);
+        when(jsonUtils.toJson(pingDTO)).thenReturn(expectedJson);
+
+        // Act
+        publisher.handlePingCreated(event);
+
+        // Assert
+        verify(pingMapper).toDTO(ping);
+        verify(jsonUtils).toJson(pingDTO);
+        verify(mapUpdatesHandler).sendMessage(expectedJson);
+    }
+}