From 184c9868af905e16742b24145841b0d0216e3106 Mon Sep 17 00:00:00 2001 From: Cole Michaels Date: Wed, 22 Apr 2026 20:28:02 +0900 Subject: [PATCH] fixed bug with presence and made it so if a user is invite to a group the group appears in the UI automatically. --- ChatClient/src/lib/api.ts | 6 +++ ChatClient/src/lib/wsManager.ts | 12 +++++ ChatClient/src/store/groupStore.ts | 4 +- ChatClient/src/types/index.ts | 7 ++- ChatServer/src/main/kotlin/di/AppModule.kt | 2 +- .../src/main/kotlin/models/GroupAddedEvent.kt | 12 +++++ .../main/kotlin/models/GroupAddedPayload.kt | 9 ++++ .../src/main/kotlin/models/SqsEnvelope.kt | 1 + .../main/kotlin/plugins/ConversationRoutes.kt | 37 +++++++++++++ .../main/kotlin/services/sqs/SqsConsumer.kt | 47 ++++++++++++++-- .../kotlin/plugins/ConversationRoutesTest.kt | 24 +++++++++ .../kotlin/services/sqs/SqsConsumerTest.kt | 54 ++++++++++++++++++- 12 files changed, 205 insertions(+), 10 deletions(-) create mode 100644 ChatServer/src/main/kotlin/models/GroupAddedEvent.kt create mode 100644 ChatServer/src/main/kotlin/models/GroupAddedPayload.kt diff --git a/ChatClient/src/lib/api.ts b/ChatClient/src/lib/api.ts index 506d3d1..e6c8306 100644 --- a/ChatClient/src/lib/api.ts +++ b/ChatClient/src/lib/api.ts @@ -77,6 +77,12 @@ export async function getGroups(): Promise { return res.json() } +export async function getGroup(groupId: string): Promise { + 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 { const res = await authedFetch(`${API_BASE}/groups`, { method: 'POST', diff --git a/ChatClient/src/lib/wsManager.ts b/ChatClient/src/lib/wsManager.ts index 8da56a9..1b527f9 100644 --- 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 index a0bd760..b61df6f 100644 --- a/ChatClient/src/store/groupStore.ts +++ b/ChatClient/src/store/groupStore.ts @@ -13,6 +13,8 @@ export const useGroupStore = create((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 index 6327e37..e556126 100644 --- 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 index e9f4fe0..427293d 100644 --- 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 new file mode 100644 index 0000000..6857f02 --- /dev/null +++ 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 new file mode 100644 index 0000000..88db14c --- /dev/null +++ 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 index ff41002..874e94c 100644 --- 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 index c2fc4a9..07ee7b3 100644 --- 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() + val serverRegistry by inject() + val sqsClient by inject() 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 index c1a641c..2a80131 100644 --- 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(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(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 index d19db78..09ea728 100644 --- 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(relaxed = true) + private val serverRegistry = mockk(relaxed = true) + private val sqsClient = mockk(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>()) + } + } + @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 index 69abd5f..931861f 100644 --- 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() private val pendingMessageRepository = mockk(relaxed = true) private val conversationRegistry = mockk() - private val consumer = SqsConsumer(sqsClient, sessionStore, pendingMessageRepository, conversationRegistry) + private val userRegistry = mockk() + 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()) } 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() + coEvery { bobSession.send(any()) } 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()) } + } + + @Test + fun `GROUP_ADDED notifies locally connected existing members when added user is online elsewhere`() = + runTest { + val aliceSession = mockk() + coEvery { aliceSession.send(any()) } 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()) } + } }