Feat(Chat): 채팅 메시지 발송 Websocket, Stomp API 구현

application/src/main/kotlin/com/threedays/application/chat/port/outbound/SessionClient.kt

@@ -0,0 +1,13 @@
+package com.threedays.application.chat.port.outbound
+
+import com.threedays.domain.chat.entity.Message
+import com.threedays.domain.chat.entity.Session
+
+interface SessionClient {
+
+    suspend fun sendMessage(
+        session: Session,
+        message: Message,
+    )
+
+}

application/src/main/kotlin/com/threedays/application/chat/service/ReceiveMessageService.kt

@@ -1,6 +1,11 @@
 package com.threedays.application.chat.service
 
 import com.threedays.application.chat.port.inbound.ReceiveMessage
+import com.threedays.application.chat.port.outbound.SessionClient
+import com.threedays.domain.chat.entity.Channel
+import com.threedays.domain.chat.entity.Session
+import com.threedays.domain.chat.repository.ChannelRepository
+import com.threedays.domain.chat.repository.SessionRepository
 import io.github.oshai.kotlinlogging.KotlinLogging
 import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
@@ -9,24 +14,31 @@ import kotlinx.coroutines.launch
 import org.springframework.stereotype.Service
 
 @Service
-class ReceiveMessageService() : ReceiveMessage {
+class ReceiveMessageService(
+    private val sessionClient: SessionClient,
+    private val channelRepository: ChannelRepository,
+    private val sessionRepository: SessionRepository,
+) : ReceiveMessage {
 
     companion object {
 
         private val logger = KotlinLogging.logger { }
     }
 
-    private val scope = CoroutineScope(Dispatchers.IO)
-
-    override fun invoke(command: ReceiveMessage.Command) {
-        scope.launch(exceptionHandler) {
-            logger.info { "Received message: $command" }
-            // TODO Implement message processing
-        }
-    }
-
     private val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
         logger.error(throwable) { "An error occurred while processing the message" }
     }
 
+    private val scope = CoroutineScope(Dispatchers.IO + exceptionHandler)
+
+    override fun invoke(command: ReceiveMessage.Command) {
+        val channel: Channel = channelRepository.findById(command.message.channelId) ?: return
+        val sessions: List<Session> = channel.getMemberSessions(sessionRepository)
+
+        sessions.forEach { session ->
+            scope.launch {
+                sessionClient.sendMessage(session, command.message)
+            }
+        }
+    }
 }

bootstrap/api/build.gradle.kts

@@ -19,6 +19,7 @@ dependencies {
     implementation(project(":infrastructure:aws"))
 
     implementation(libs.spring.boot.starter.web)
+    implementation(libs.spring.boot.starter.websocket)
     implementation(libs.springdoc.openapi.webmvc.ui)
     implementation(libs.swagger.annotations)
     implementation(libs.spring.boot.starter.validation)

bootstrap/api/src/main/kotlin/com/threedays/bootstrap/api/ThreeDaysApplication.kt

@@ -8,11 +8,12 @@ import com.threedays.rest.support.config.RestConfig
 import com.threedays.sms.support.SmsConfig
 import jakarta.annotation.PostConstruct
 import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.context.properties.ConfigurationPropertiesScan
+import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.boot.runApplication
 import org.springframework.context.annotation.Import
 import java.util.*
 
-@SpringBootApplication
 @Import(
     value = [
         RestConfig::class,
@@ -23,6 +24,9 @@ import java.util.*
         ApplicationConfig::class,
     ]
 )
+@SpringBootApplication
+@EnableConfigurationProperties
+@ConfigurationPropertiesScan
 class ThreeDaysApplication {
 
     companion object {

bootstrap/api/src/main/kotlin/com/threedays/bootstrap/api/chat/WebSocketSessionClient.kt

@@ -0,0 +1,35 @@
+package com.threedays.bootstrap.api.chat
+
+import com.threedays.application.chat.port.outbound.SessionClient
+import com.threedays.domain.chat.entity.Message
+import com.threedays.domain.chat.entity.Session
+import io.github.oshai.kotlinlogging.KotlinLogging
+import org.springframework.stereotype.Component
+import org.springframework.web.socket.TextMessage
+import org.springframework.web.socket.WebSocketSession
+
+@Component
+class WebSocketSessionClient(
+    private val webSocketSessionManager: WebSocketSessionManager,
+) : SessionClient {
+
+    companion object {
+
+        private val logger = KotlinLogging.logger {}
+    }
+
+    override suspend fun sendMessage(
+        session: Session,
+        message: Message
+    ) {
+        val webSocketSession: WebSocketSession =
+            webSocketSessionManager.findById(session.id) ?: run {
+                logger.warn { "WebSocketSession not found: sessionId=${session.id}" }
+                return
+            }
+
+        val textMessage = TextMessage(message.toString())
+        webSocketSession.sendMessage(textMessage)
+    }
+
+}

bootstrap/api/src/main/kotlin/com/threedays/bootstrap/api/chat/WebSocketSessionHandlerDecorator.kt

@@ -0,0 +1,38 @@
+package com.threedays.bootstrap.api.chat
+
+import io.github.oshai.kotlinlogging.KotlinLogging
+import org.springframework.stereotype.Component
+import org.springframework.web.socket.WebSocketHandler
+import org.springframework.web.socket.WebSocketSession
+import org.springframework.web.socket.handler.WebSocketHandlerDecorator
+import org.springframework.web.socket.handler.WebSocketHandlerDecoratorFactory
+
+@Component
+class WebSocketSessionHandlerDecorator(
+    private val webSocketSessionManager: WebSocketSessionManager,
+) : WebSocketHandlerDecoratorFactory {
+
+    companion object {
+
+        private val logger = KotlinLogging.logger {}
+    }
+
+    override fun decorate(handler: WebSocketHandler) = object : WebSocketHandlerDecorator(handler) {
+        override fun afterConnectionEstablished(session: WebSocketSession) {
+            super.afterConnectionEstablished(session)
+            webSocketSessionManager.addSession(session)
+            logger.info { "WebSocketSession established: sessionId=${session.id}" }
+        }
+
+        override fun afterConnectionClosed(
+            session: WebSocketSession,
+            closeStatus: org.springframework.web.socket.CloseStatus
+        ) {
+            super.afterConnectionClosed(session, closeStatus)
+            webSocketSessionManager.removeSession(session)
+            logger.info { "WebSocketSession closed: sessionId=${session.id}" }
+        }
+    }
+
+
+}

bootstrap/api/src/main/kotlin/com/threedays/bootstrap/api/chat/WebSocketSessionManager.kt

@@ -0,0 +1,34 @@
+package com.threedays.bootstrap.api.chat
+
+import com.threedays.bootstrap.api.support.config.WebSocketProperties
+import com.threedays.domain.chat.entity.Session
+import org.springframework.stereotype.Component
+import org.springframework.web.socket.WebSocketSession
+import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator
+import java.util.concurrent.ConcurrentHashMap
+
+@Component
+class WebSocketSessionManager(
+    private val webSocketProperties: WebSocketProperties,
+) {
+
+    private val sessions = ConcurrentHashMap<String, WebSocketSession>()
+
+    fun addSession(session: WebSocketSession) {
+        val concurrentSession = ConcurrentWebSocketSessionDecorator(
+            /* delegate = */ session,
+            /* sendTimeLimit = */ webSocketProperties.session.sendTimeLimit,
+            /* bufferSizeLimit = */ webSocketProperties.session.sendBufferSizeLimit
+        )
+        sessions[session.id] = concurrentSession
+    }
+
+    fun removeSession(session: WebSocketSession) {
+        sessions.remove(session.id)
+    }
+
+    fun findById(sessionId: Session.Id): WebSocketSession? {
+        return sessions[sessionId.value]
+    }
+
+}

bootstrap/api/src/main/kotlin/com/threedays/bootstrap/api/support/config/WebSocketConfig.kt

@@ -0,0 +1,32 @@
+package com.threedays.bootstrap.api.support.config
+
+import com.threedays.bootstrap.api.chat.WebSocketSessionHandlerDecorator
+import org.springframework.context.annotation.Configuration
+import org.springframework.messaging.simp.config.MessageBrokerRegistry
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer
+import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration
+
+@Configuration
+@EnableWebSocketMessageBroker
+class WebSocketConfig(
+    private val properties: WebSocketProperties,
+    private val webSocketSessionHandlerDecorator: WebSocketSessionHandlerDecorator,
+) : WebSocketMessageBrokerConfigurer {
+
+    override fun configureMessageBroker(registry: MessageBrokerRegistry) {
+        registry.enableSimpleBroker(*properties.broker.simpleBroker.toTypedArray())
+        registry.setApplicationDestinationPrefixes(*properties.broker.applicationDestinationPrefixes.toTypedArray())
+    }
+
+    override fun registerStompEndpoints(registry: StompEndpointRegistry) {
+        registry
+            .addEndpoint(properties.endpoint)
+            .setAllowedOrigins(*properties.allowedOrigins.toTypedArray())
+    }
+
+    override fun configureWebSocketTransport(registry: WebSocketTransportRegistration) {
+        registry.addDecoratorFactory(webSocketSessionHandlerDecorator)
+    }
+}

bootstrap/api/src/main/kotlin/com/threedays/bootstrap/api/support/config/WebSocketProperties.kt

@@ -0,0 +1,39 @@
+package com.threedays.bootstrap.api.support.config
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+
+/**
+ * @param endpoint the endpoint to connect to
+ * @param allowedOrigins list of allowed origins
+ * @param sockJs whether to use SockJS
+ * @param broker configuration for the broker
+ * @param session configuration for the session
+ */
+@ConfigurationProperties(prefix = "websocket")
+data class WebSocketProperties(
+    val endpoint: String,
+    val allowedOrigins: List<String>,
+    val sockJs: Boolean,
+    val broker: Broker,
+    val session: Session,
+) {
+
+    /**
+     * @param simpleBroker list of broker destinations
+     * @param applicationDestinationPrefixes list of application destinations
+     */
+    data class Broker(
+        val simpleBroker: List<String>,
+        val applicationDestinationPrefixes: List<String>,
+    )
+
+    /**
+     * @param sendTimeLimit in milliseconds
+     * @param sendBufferSizeLimit in bytes
+     */
+    data class Session(
+        val sendTimeLimit: Int,
+        val sendBufferSizeLimit: Int,
+    )
+
+}

bootstrap/api/src/main/kotlin/com/threedays/bootstrap/api/support/security/interceptor/StompAuthInterceptor.kt

@@ -0,0 +1,66 @@
+package com.threedays.bootstrap.api.support.security.interceptor
+
+import com.threedays.application.auth.config.AuthProperties
+import com.threedays.domain.auth.entity.AccessToken
+import com.threedays.domain.chat.entity.Session
+import com.threedays.domain.chat.repository.SessionRepository
+import org.springframework.messaging.Message
+import org.springframework.messaging.MessageChannel
+import org.springframework.messaging.simp.stomp.StompCommand
+import org.springframework.messaging.simp.stomp.StompHeaderAccessor
+import org.springframework.messaging.support.ChannelInterceptor
+import org.springframework.stereotype.Component
+
+@Component
+class StompAuthInterceptor(
+    private val authProperties: AuthProperties,
+    private val sessionRepository: SessionRepository,
+) : ChannelInterceptor {
+
+    companion object {
+
+        private const val AUTHORIZATION_HEADER = "Authorization"
+        private const val BEARER_PREFIX = "Bearer "
+    }
+
+    override fun preSend(
+        message: Message<*>,
+        channel: MessageChannel,
+    ): Message<*>? {
+        val accessor = StompHeaderAccessor.wrap(message)
+        val command = accessor.command
+        val sessionId = accessor.sessionId
+            ?.let { Session.Id(it) }
+            ?: return message
+
+        when (command) {
+            StompCommand.CONNECT -> handleConnect(sessionId, message)
+            else -> return message
+        }
+
+        return message
+    }
+
+    private fun handleConnect(
+        sessionId: Session.Id,
+        message: Message<*>,
+    ) {
+        extractToken(message)?.let { token ->
+            AccessToken.verify(
+                value = token,
+                secret = authProperties.tokenSecret,
+            )
+                .let { Session.create(sessionId, it.userId) }
+                .also { sessionRepository.save(it) }
+        }
+    }
+
+    private fun extractToken(message: Message<*>): String? {
+        val accessor = StompHeaderAccessor.wrap(message)
+        val bearerToken: String? = accessor.getFirstNativeHeader(AUTHORIZATION_HEADER)
+        return if (bearerToken != null && bearerToken.startsWith(BEARER_PREFIX)) {
+            bearerToken.substring(BEARER_PREFIX.length)
+        } else null
+    }
+
+}

bootstrap/api/src/main/resources/application.yaml

@@ -14,6 +14,19 @@ spring:
   docker:
     compose:
       enabled: false
+websocket:
+  endpoint: /ws
+  allowed-origins:
+    - "*"
+  sock-js: true
+  broker:
+    application-destination-prefixes:
+      - /app
+    simple-broker:
+      - /channel
+  session:
+    send-time-limit: 10000
+    send-buffer-size-limit: 256
 ---
 spring:
   config:

domain/src/main/kotlin/com/threedays/domain/chat/entity/Channel.kt

@@ -1,5 +1,6 @@
 package com.threedays.domain.chat.entity
 
+import com.threedays.domain.chat.repository.SessionRepository
 import com.threedays.domain.connection.entity.Connection
 import com.threedays.support.common.base.domain.AggregateRoot
 import com.threedays.support.common.base.domain.TypeId
@@ -27,4 +28,8 @@ data class Channel(
 
     }
 
+    fun getMemberSessions(sessionRepository: SessionRepository): List<Session> =
+        members.mapNotNull { member -> sessionRepository.findByUserId(member.id) }
+
+
 }