asinglesprinkle · Apr 23, 2026 · Apr 22, 2026
diff --git a/ChatClient/src/lib/api.ts b/ChatClient/src/lib/api.ts
@@ -77,6 +77,12 @@ export async function getGroups(): Promise<GroupResponse[]> {
   return res.json()
 }
 
+export async function getGroup(groupId: string): Promise<GroupResponse> {
+  const res = await authedFetch(`${API_BASE}/groups/${groupId}`)
+  if (!res.ok) throw new Error('Failed to fetch group')
+  return res.json()
+}
+
 export async function createGroup(name: string): Promise<GroupResponse> {
   const res = await authedFetch(`${API_BASE}/groups`, {
     method: 'POST',

diff --git a/ChatClient/src/lib/wsManager.ts b/ChatClient/src/lib/wsManager.ts
@@ -1,6 +1,8 @@
 import { WS_URL, WS_RECONNECT_DELAY_MS } from './constants'
+import { getGroup, getGroups } from './api'
 import { useMessageStore } from '../store/messageStore'
 import { usePresenceStore } from '../store/presenceStore'
+import { useGroupStore } from '../store/groupStore'
 import type { WsIncomingMessage, InboundWsSend } from '../types'
 
 class WsManager {
@@ -24,6 +26,12 @@ class WsManager {
 
     this.ws = new WebSocket(`${WS_URL}?token=${this.token}`)
 
+    this.ws.onopen = () => {
+      getGroups().then((groups) => {
+        useGroupStore.getState().setGroups(groups)
+      }).catch(() => {})
+    }
+
     this.ws.onmessage = (event) => {
       try {
         const msg: WsIncomingMessage = JSON.parse(event.data)
@@ -36,6 +44,10 @@ class WsManager {
           })
         } else if (msg.type === 'presence') {
           usePresenceStore.getState().setUserPresence(msg.userId, msg.online)
+        } else if (msg.type === 'group_added') {
+          getGroup(msg.conversationId).then((group) => {
+            useGroupStore.getState().addGroup(group)
+          }).catch((e) => console.error('Failed to handle group_added event:', e))
         }
       } catch (e) {
         console.error('WS message error:', e)

diff --git a/ChatClient/src/store/groupStore.ts b/ChatClient/src/store/groupStore.ts
@@ -13,6 +13,8 @@ export const useGroupStore = create<GroupState>((set) => ({
   groups: [],
   activeGroupId: null,
   setGroups: (groups) => set({ groups }),
-  addGroup: (group) => set((s) => ({ groups: [...s.groups, group] })),
+  addGroup: (group) => set((s) => ({
+    groups: s.groups.some((g) => g.id === group.id) ? s.groups : [...s.groups, group],
+  })),
   setActiveGroup: (id) => set({ activeGroupId: id }),
 }))
diff --git a/ChatClient/src/types/index.ts b/ChatClient/src/types/index.ts
@@ -38,7 +38,12 @@ export interface WsPresenceEvent {
   online: boolean
 }
 
-export type WsIncomingMessage = OutboundWsMessage | WsPresenceEvent
+export interface WsGroupAddedEvent {
+  type: 'group_added'
+  conversationId: string
+}
+
+export type WsIncomingMessage = OutboundWsMessage | WsPresenceEvent | WsGroupAddedEvent
 
 export interface InboundWsSend {
   conversationId: string

diff --git a/ChatServer/src/main/kotlin/di/AppModule.kt b/ChatServer/src/main/kotlin/di/AppModule.kt
@@ -48,7 +48,7 @@ val appModule =
         single { PendingMessageRepository(get()) }
         single { ConversationRegistry(get()) }
         single { MessageRepository(get()) }
-        single { SqsConsumer(get(), get(), get(), get()) }
+        single { SqsConsumer(get(), get(), get(), get(), get()) }
         single { MessageRouter(get(), get(), get(), get(), get(), get()) }
         single { SessionStore() }
         single {

diff --git a/ChatServer/src/main/kotlin/models/GroupAddedEvent.kt b/ChatServer/src/main/kotlin/models/GroupAddedEvent.kt
@@ -0,0 +1,12 @@
+package org.chatserver.models
+
+import kotlinx.serialization.EncodeDefault
+import kotlinx.serialization.ExperimentalSerializationApi
+import kotlinx.serialization.Serializable
+
+@OptIn(ExperimentalSerializationApi::class)
+@Serializable
+data class GroupAddedEvent(
+    @EncodeDefault val type: String = "group_added",
+    val conversationId: String,
+)
diff --git a/ChatServer/src/main/kotlin/models/GroupAddedPayload.kt b/ChatServer/src/main/kotlin/models/GroupAddedPayload.kt
@@ -0,0 +1,9 @@
+package org.chatserver.models
+
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class GroupAddedPayload(
+    val conversationId: String,
+    val userId: String,
+)
diff --git a/ChatServer/src/main/kotlin/models/SqsEnvelope.kt b/ChatServer/src/main/kotlin/models/SqsEnvelope.kt
@@ -10,5 +10,6 @@ data class SqsEnvelope(
     companion object {
         const val CHAT = "CHAT"
         const val PRESENCE = "PRESENCE"
+        const val GROUP_ADDED = "GROUP_ADDED"
     }
 }
diff --git a/ChatServer/src/main/kotlin/plugins/ConversationRoutes.kt b/ChatServer/src/main/kotlin/plugins/ConversationRoutes.kt
@@ -8,14 +8,25 @@ import io.ktor.server.response.respond
 import io.ktor.server.routing.delete
 import io.ktor.server.routing.post
 import io.ktor.server.routing.routing
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
 import org.chatserver.data.registry.ConversationRegistry
+import org.chatserver.data.registry.ServerRegistry
 import org.chatserver.models.CreateConversationRequest
+import org.chatserver.models.GroupAddedPayload
+import org.chatserver.models.SqsEnvelope
 import org.koin.ktor.ext.inject
+import org.slf4j.LoggerFactory
+import software.amazon.awssdk.services.sqs.SqsClient
+
+private val logger = LoggerFactory.getLogger("ConversationRoutes")
 
 fun Application.configureConversationRoutes(
     internalApiKey: String = System.getenv("INTERNAL_API_KEY") ?: error("INTERNAL_API_KEY env var not set"),
 ) {
     val conversationRegistry by inject<ConversationRegistry>()
+    val serverRegistry by inject<ServerRegistry>()
+    val sqsClient by inject<SqsClient>()
 
     routing {
         post("/conversations") {
@@ -34,6 +45,7 @@ fun Application.configureConversationRoutes(
             val conversationId = call.parameters["conversationId"] ?: return@post call.respond(HttpStatusCode.BadRequest)
             val userId = call.parameters["userId"] ?: return@post call.respond(HttpStatusCode.BadRequest)
             conversationRegistry.addMember(conversationId, userId)
+            broadcastGroupAdded(conversationId, userId, serverRegistry, sqsClient)
             call.respond(HttpStatusCode.Created)
         }
 
@@ -48,3 +60,28 @@ fun Application.configureConversationRoutes(
         }
     }
 }
+
+private fun broadcastGroupAdded(
+    conversationId: String,
+    userId: String,
+    serverRegistry: ServerRegistry,
+    sqsClient: SqsClient,
+) {
+    val envelope =
+        Json.encodeToString(
+            SqsEnvelope(
+                type = SqsEnvelope.GROUP_ADDED,
+                payload = Json.encodeToString(GroupAddedPayload(conversationId = conversationId, userId = userId)),
+            ),
+        )
+    for (queueUrl in serverRegistry.getAllQueueUrls()) {
+        try {
+            sqsClient.sendMessage {
+                it.queueUrl(queueUrl)
+                it.messageBody(envelope)
+            }
+        } catch (e: Exception) {
+            logger.warn("Failed to broadcast GROUP_ADDED to $queueUrl: ${e.message}")
+        }
+    }
+}
diff --git a/ChatServer/src/main/kotlin/services/sqs/SqsConsumer.kt b/ChatServer/src/main/kotlin/services/sqs/SqsConsumer.kt
@@ -4,8 +4,11 @@ import io.ktor.websocket.Frame
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 import org.chatserver.data.registry.ConversationRegistry
+import org.chatserver.data.registry.UserRegistry
 import org.chatserver.data.repository.PendingMessageRepository
 import org.chatserver.models.ChatMessage
+import org.chatserver.models.GroupAddedEvent
+import org.chatserver.models.GroupAddedPayload
 import org.chatserver.models.PresenceEvent
 import org.chatserver.models.SqsEnvelope
 import org.chatserver.session.SessionStore
@@ -17,6 +20,7 @@ class SqsConsumer(
     private val sessionStore: SessionStore,
     private val pendingMessageRepository: PendingMessageRepository,
     private val conversationRegistry: ConversationRegistry,
+    private val userRegistry: UserRegistry,
 ) {
     companion object {
         private const val MAX_MESSAGES = 10
@@ -36,11 +40,16 @@ class SqsConsumer(
                 }
 
             for (message in response.messages()) {
-                val envelope = Json.decodeFromString<SqsEnvelope>(message.body())
-                when (envelope.type) {
-                    SqsEnvelope.CHAT -> handleChat(Json.decodeFromString(envelope.payload))
-                    SqsEnvelope.PRESENCE -> handlePresence(Json.decodeFromString(envelope.payload))
-                    else -> logger.warn("Unknown SQS message type: ${envelope.type}")
+                try {
+                    val envelope = Json.decodeFromString<SqsEnvelope>(message.body())
+                    when (envelope.type) {
+                        SqsEnvelope.CHAT -> handleChat(Json.decodeFromString(envelope.payload))
+                        SqsEnvelope.PRESENCE -> handlePresence(Json.decodeFromString(envelope.payload))
+                        SqsEnvelope.GROUP_ADDED -> handleGroupAdded(Json.decodeFromString(envelope.payload))
+                        else -> logger.warn("Unknown SQS message type: ${envelope.type}")
+                    }
+                } catch (e: Exception) {
+                    logger.error("Failed to process SQS message: ${e.message}", e)
                 }
 
                 sqsClient.deleteMessage {
@@ -71,4 +80,32 @@ class SqsConsumer(
             "Presence for ${event.userId} online=${event.online}: ${groupmates.size} groupmate(s), ${recipients.size} local recipient(s)",
         )
     }
+
+    private suspend fun handleGroupAdded(payload: GroupAddedPayload) {
+        val members = conversationRegistry.getMembers(payload.conversationId)
+        val addedUserId = payload.userId
+        val existingMembers = members.toSet().minus(addedUserId)
+
+        // Notify the added user if connected to this server
+        val addedSession = sessionStore.get(addedUserId)
+        if (addedSession != null) {
+            addedSession.send(Frame.Text(Json.encodeToString(GroupAddedEvent(conversationId = payload.conversationId))))
+            val onlineMembers = userRegistry.getConnectedUsersFrom(existingMembers)
+            for (uid in onlineMembers) {
+                addedSession.send(Frame.Text(Json.encodeToString(PresenceEvent(userId = uid, online = true))))
+            }
+            logger.info("GROUP_ADDED: notified $addedUserId of ${payload.conversationId}, ${onlineMembers.size} online member(s) sent")
+        }
+
+        // Notify locally-connected existing members if the added user is online
+        val addedOnline = addedSession != null || userRegistry.getConnectedUsersFrom(setOf(addedUserId)).isNotEmpty()
+        if (addedOnline) {
+            val presenceFrame = Frame.Text(Json.encodeToString(PresenceEvent(userId = addedUserId, online = true)))
+            val localExisting = sessionStore.getAll().filter { it in existingMembers }
+            localExisting.forEach { uid -> sessionStore.get(uid)?.send(presenceFrame) }
+            if (localExisting.isNotEmpty()) {
+                logger.info("GROUP_ADDED: broadcast $addedUserId online to ${localExisting.size} local member(s)")
+            }
+        }
+    }
 }
diff --git a/ChatServer/src/test/kotlin/plugins/ConversationRoutesTest.kt b/ChatServer/src/test/kotlin/plugins/ConversationRoutesTest.kt
@@ -10,11 +10,14 @@ import io.ktor.http.contentType
 import io.ktor.server.application.install
 import io.ktor.server.testing.ApplicationTestBuilder
 import io.ktor.server.testing.testApplication
+import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
 import org.chatserver.data.registry.ConversationRegistry
+import org.chatserver.data.registry.ServerRegistry
 import org.koin.dsl.module
 import org.koin.ktor.plugin.Koin
+import software.amazon.awssdk.services.sqs.SqsClient
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
@@ -24,6 +27,8 @@ class ConversationRoutesTest {
     }
 
     private val registry = mockk<ConversationRegistry>(relaxed = true)
+    private val serverRegistry = mockk<ServerRegistry>(relaxed = true)
+    private val sqsClient = mockk<SqsClient>(relaxed = true)
 
     private fun withConversationApp(block: suspend ApplicationTestBuilder.() -> Unit) {
         testApplication {
@@ -32,6 +37,8 @@ class ConversationRoutesTest {
                     modules(
                         module {
                             single { registry }
+                            single { serverRegistry }
+                            single { sqsClient }
                         },
                     )
                 }
@@ -87,6 +94,8 @@ class ConversationRoutesTest {
 
     @Test
     fun `POST conversations members userId returns 201 and calls addMember`() {
+        every { serverRegistry.getAllQueueUrls() } returns emptyList()
+
         withConversationApp {
             val response =
                 client.post("/conversations/conv-1/members/alice") {
@@ -98,6 +107,21 @@ class ConversationRoutesTest {
         verify { registry.addMember("conv-1", "alice") }
     }
 
+    @Test
+    fun `POST conversations members userId broadcasts GROUP_ADDED via SQS`() {
+        every { serverRegistry.getAllQueueUrls() } returns listOf("https://sqs/queue-1")
+
+        withConversationApp {
+            client.post("/conversations/conv-1/members/alice") {
+                header("x-internal-key", TEST_INTERNAL_KEY)
+            }
+        }
+
+        verify(exactly = 1) {
+            sqsClient.sendMessage(any<java.util.function.Consumer<software.amazon.awssdk.services.sqs.model.SendMessageRequest.Builder>>())
+        }
+    }
+
     @Test
     fun `DELETE conversations members userId returns 204 and calls removeMember`() {
         withConversationApp {

diff --git a/ChatServer/src/test/kotlin/services/sqs/SqsConsumerTest.kt b/ChatServer/src/test/kotlin/services/sqs/SqsConsumerTest.kt
@@ -14,8 +14,10 @@ import kotlinx.coroutines.test.runTest
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 import org.chatserver.data.registry.ConversationRegistry
+import org.chatserver.data.registry.UserRegistry
 import org.chatserver.data.repository.PendingMessageRepository
 import org.chatserver.models.ChatMessage
+import org.chatserver.models.GroupAddedPayload
 import org.chatserver.models.PresenceEvent
 import org.chatserver.models.SqsEnvelope
 import org.chatserver.session.SessionStore
@@ -32,7 +34,8 @@ class SqsConsumerTest {
     private val sessionStore = mockk<SessionStore>()
     private val pendingMessageRepository = mockk<PendingMessageRepository>(relaxed = true)
     private val conversationRegistry = mockk<ConversationRegistry>()
-    private val consumer = SqsConsumer(sqsClient, sessionStore, pendingMessageRepository, conversationRegistry)
+    private val userRegistry = mockk<UserRegistry>()
+    private val consumer = SqsConsumer(sqsClient, sessionStore, pendingMessageRepository, conversationRegistry, userRegistry)
 
     private val chatMessage = ChatMessage(fromUserId = "alice", toUserId = "bob", conversationId = "conv-1", content = "hello")
     private val chatMessageJson = Json.encodeToString(chatMessage)
@@ -67,7 +70,7 @@ class SqsConsumerTest {
             val job = launch { consumer.start("https://sqs/queue") }
             job.join()
 
-            coVerify { session.send(match { it is Frame.Text && (it as Frame.Text).readText() == chatMessageJson }) }
+            coVerify { session.send(match { it is Frame.Text && it.readText() == chatMessageJson }) }
             verify(exactly = 0) { pendingMessageRepository.save(any()) }
         }
 
@@ -192,4 +195,51 @@ class SqsConsumerTest {
             coVerify(exactly = 0) { nonGroupmateSession.send(any<Frame>()) }
             verify(exactly = 0) { pendingMessageRepository.save(any()) }
         }
+
+    @Test
+    fun `GROUP_ADDED sends group event and presence snapshot to locally connected added user`() =
+        runTest {
+            val bobSession = mockk<DefaultWebSocketServerSession>()
+            coEvery { bobSession.send(any<Frame>()) } returns Unit
+
+            val payload = GroupAddedPayload(conversationId = "conv-1", userId = "bob")
+            val envelopeJson = Json.encodeToString(SqsEnvelope(SqsEnvelope.GROUP_ADDED, Json.encodeToString(payload)))
+
+            stubReceiveOnce(ReceiveMessageResponse.builder().messages(buildSqsMessage(envelopeJson)).build())
+
+            every { conversationRegistry.getMembers("conv-1") } returns listOf("alice", "bob")
+            every { sessionStore.get("bob") } returns bobSession
+            every { sessionStore.getAll() } returns setOf("bob")
+            every { userRegistry.getConnectedUsersFrom(setOf("alice")) } returns setOf("alice")
+
+            val job = launch { consumer.start("https://sqs/queue") }
+            job.join()
+
+            // bob receives GroupAddedEvent + PresenceEvent for alice
+            coVerify(exactly = 2) { bobSession.send(any<Frame>()) }
+        }
+
+    @Test
+    fun `GROUP_ADDED notifies locally connected existing members when added user is online elsewhere`() =
+        runTest {
+            val aliceSession = mockk<DefaultWebSocketServerSession>()
+            coEvery { aliceSession.send(any<Frame>()) } returns Unit
+
+            val payload = GroupAddedPayload(conversationId = "conv-1", userId = "bob")
+            val envelopeJson = Json.encodeToString(SqsEnvelope(SqsEnvelope.GROUP_ADDED, Json.encodeToString(payload)))
+
+            stubReceiveOnce(ReceiveMessageResponse.builder().messages(buildSqsMessage(envelopeJson)).build())
+
+            every { conversationRegistry.getMembers("conv-1") } returns listOf("alice", "bob")
+            every { sessionStore.get("bob") } returns null
+            every { sessionStore.get("alice") } returns aliceSession
+            every { sessionStore.getAll() } returns setOf("alice")
+            every { userRegistry.getConnectedUsersFrom(setOf("bob")) } returns setOf("bob")
+
+            val job = launch { consumer.start("https://sqs/queue") }
+            job.join()
+
+            // alice receives bob's presence event
+            coVerify(exactly = 1) { aliceSession.send(any<Frame>()) }
+        }
 }