add sse for user app

Author: xcodeassociated

http/update.http

@@ -0,0 +1,4 @@
+###
+GET {{host}}/update/user
+Content-Type: text/event-stream
+#Authorization: Bearer {{oauthToken}}

src/main/kotlin/com/softeno/template/app/config/security/SecurityConfig.kt

@@ -100,7 +100,9 @@ class SecurityConfig {
                     "/webjars/**",
                     "/swagger-resources/**",
                     "/swagger-ui/**",
-                    "/v3/api-docs/**"
+                    "/v3/api-docs/**",
+                    // todo: make this endpoint secure
+                    "/update/**"
                 )
                     .permitAll()
                     .pathMatchers("/reactive/**", "/coroutine/**", "/ws/**", "/graphql/**", "/external/**").hasAuthority("ROLE_ADMIN")

src/main/kotlin/com/softeno/template/app/event/AppEvent.kt

@@ -1,6 +1,7 @@
 package com.softeno.template.app.event
 
 import com.softeno.template.app.kafka.dto.KafkaMessage
+import com.softeno.template.app.user.service.UserUpdateEmitter
 import com.softeno.template.sample.http.internal.serverevents.Event
 import com.softeno.template.sample.http.internal.serverevents.UserNotificationService
 import com.softeno.template.sample.kafka.ReactiveKafkaSampleProducer
@@ -20,6 +21,7 @@ data class AppEvent(val source: String, val traceId: String? = null, val spanId:
 @Component
 class SampleApplicationEventPublisher(
     private val reactiveMessageService: ReactiveMessageService,
+    private val userUpdateEmitter: UserUpdateEmitter,
     private val reactiveKafkaProducer: ReactiveKafkaSampleProducer,
     private val userNotificationService: UserNotificationService,
 ) : ApplicationListener<AppEvent> {
@@ -35,6 +37,7 @@ class SampleApplicationEventPublisher(
 
         log.info("[event handler]: Received event: $event")
         reactiveMessageService.broadcast(event.toMessage())
+        userUpdateEmitter.broadcast(event.toMessage())
 
         reactiveKafkaProducer.send(event.toKafkaMessage())
 

src/main/kotlin/com/softeno/template/app/user/api/Controller.kt

@@ -6,18 +6,23 @@ import com.softeno.template.app.permission.api.PermissionDto
 import com.softeno.template.app.user.UserModifyCommand
 import com.softeno.template.app.user.mapper.toDto
 import com.softeno.template.app.user.service.UserService
+import com.softeno.template.app.user.service.UserUpdateEmitter
 import io.micrometer.tracing.Tracer
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
 import org.apache.commons.logging.LogFactory
 import org.slf4j.MDC
+import org.springframework.http.MediaType
+import org.springframework.http.codec.ServerSentEvent
+import org.springframework.http.server.reactive.ServerHttpResponse
 import org.springframework.security.core.annotation.AuthenticationPrincipal
 import org.springframework.stereotype.Component
 import org.springframework.validation.annotation.Validated
 import org.springframework.web.bind.annotation.*
 import org.springframework.web.server.ServerWebExchange
 import org.springframework.web.server.WebFilter
 import org.springframework.web.server.WebFilterChain
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import java.security.Principal
 import java.time.LocalDateTime
@@ -88,7 +93,23 @@ class CoroutineUserController(
 
     @GetMapping("/usersCount")
     suspend fun getUserSize(): Long = userService.size()
+}
 
+@RestController
+@RequestMapping("/update/")
+@Validated
+class UpdateController(
+    val updateEmitter: UserUpdateEmitter
+) {
+    @GetMapping("/user", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
+    fun getUserUpdate(response: ServerHttpResponse): Flux<ServerSentEvent<String>> {
+        response.headers.apply {
+            set("Cache-Control", "no-cache, no-store, must-revalidate")
+            set("Connection", "keep-alive")
+            set("X-Accel-Buffering", "no")
+        }
+        return updateEmitter.getSink()
+    }
 }
 
 data class UserDto(

src/main/kotlin/com/softeno/template/app/user/service/Service.kt

@@ -1,5 +1,6 @@
 package com.softeno.template.app.user.service
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import com.softeno.template.app.common.ErrorFactory
 import com.softeno.template.app.common.PrincipalHandler
 import com.softeno.template.app.common.getPageRequest
@@ -16,6 +17,8 @@ import com.softeno.template.app.user.db.UserCoroutineRepository
 import com.softeno.template.app.user.db.UserDocument
 import com.softeno.template.app.user.mapper.toDocument
 import com.softeno.template.app.user.mapper.toDomain
+import com.softeno.template.sample.websocket.Message
+import com.softeno.template.sample.websocket.toJson
 import io.micrometer.tracing.Tracer
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.asFlow
@@ -26,9 +29,15 @@ import kotlinx.coroutines.withContext
 import org.apache.commons.logging.LogFactory
 import org.slf4j.MDC
 import org.springframework.context.ApplicationEventPublisher
+import org.springframework.http.codec.ServerSentEvent
+import org.springframework.stereotype.Component
 import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.core.publisher.Sinks
+import reactor.util.concurrent.Queues
 import java.security.Principal
+import java.time.Duration
 
 
 @Service
@@ -149,3 +158,71 @@ class UserDocumentService(
         return@withContext userCoroutineRepository.deleteById(id)
     }
 }
+
+@Component
+class UserUpdateEmitter(
+    private val objectMapper: ObjectMapper,
+) {
+    private val sink: Sinks.Many<ServerSentEvent<String>> = Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false)
+    private val log = LogFactory.getLog(javaClass)
+
+    fun getSink(): Flux<ServerSentEvent<String>> {
+        val heartbeatFlux = Flux.interval(Duration.ofSeconds(10))
+            .map {
+                ServerSentEvent.builder<String>()
+                    .event("heartbeat")
+                    .data(Message(from = "SYSTEM", to = "ALL", content = "ping").toJson(objectMapper))
+                    .build()
+            }.doOnError { error -> log.error("Heartbeat error", error) }
+
+
+        val events = sink.asFlux()
+            .doOnSubscribe { log.info("New SSE client subscribed") }
+            .doOnCancel { log.info("SSE client disconnected") }
+            .doOnTerminate { log.info("SSE client terminated") }
+            .doOnError { error -> log.error("Event stream error", error) }
+
+
+        return Flux.merge(heartbeatFlux, events)
+            .doOnCancel { log.info("Canceling SSE stream") }
+            .doOnTerminate { log.info("Terminating SSE stream") }
+            .onErrorResume { error ->
+                log.error("SSE stream error, sending error event", error)
+                Mono.just(
+                    ServerSentEvent.builder<String>()
+                        .event("error")
+                        .data(Message(from = "SYSTEM", to = "ALL", content = "Connection error: ${error.message}").toJson(objectMapper))
+                        .build()
+                )
+            }
+    }
+
+    fun broadcast(message: Message): Boolean =
+        try {
+            val payload = message.toJson(objectMapper)
+            val sse = ServerSentEvent.builder(payload).event("update").build()
+            val result = sink.tryEmitNext(sse)
+
+            when (result) {
+                Sinks.EmitResult.OK -> {
+                    log.debug("Message broadcasted successfully: ${message.content}")
+                    true
+                }
+                Sinks.EmitResult.FAIL_CANCELLED -> {
+                    log.warn("Failed to broadcast message - emitter cancelled")
+                    false
+                }
+                Sinks.EmitResult.FAIL_OVERFLOW -> {
+                    log.warn("Failed to broadcast message - buffer overflow")
+                    false
+                }
+                else -> {
+                    log.error("Failed to broadcast message: $result")
+                    false
+                }
+            }
+        } catch (e: Exception) {
+            log.error("Error broadcasting message", e)
+            false
+        }
+}