package chats.data.remote.signalr import android.util.Log import io.reactivex.rxjava3.core.Single import com.microsoft.signalr.HubConnection import com.microsoft.signalr.HubConnectionBuilder import com.microsoft.signalr.HubConnectionState import chats.data.remote.dto.ChatDto import chats.data.remote.dto.MessageDto import kotlinx.coroutines.flow.* import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.launch import javax.inject.Inject import javax.inject.Singleton import kotlinx.coroutines.delay data class ReadMessagesRequest( val chatId: String, val lastReadMessageId: String, val lastReadSequenceId: Int ) data class MessagesReadEvent( @com.google.gson.annotations.SerializedName("chatId", alternate = ["ChatId"]) val chatId: String? = null, @com.google.gson.annotations.SerializedName("userId", alternate = ["UserId"]) val userId: String? = null, @com.google.gson.annotations.SerializedName("lastReadSequenceId", alternate = ["LastReadSequenceId"]) val lastReadSequenceId: Int? = null ) { val effectiveChatId: String get() = chatId ?: "" val effectiveUserId: String get() = userId ?: "" val effectiveLastReadSequenceId: Int get() = lastReadSequenceId ?: 0 } data class ReactionEvent( @com.google.gson.annotations.SerializedName("messageId", alternate = ["MessageId"]) val messageId: String? = null, @com.google.gson.annotations.SerializedName("chatId", alternate = ["ChatId"]) val chatId: String? = null, @com.google.gson.annotations.SerializedName("userId", alternate = ["UserId"]) val userId: String? = null, @com.google.gson.annotations.SerializedName("username", alternate = ["Username", "UserName"]) val username: String? = null, @com.google.gson.annotations.SerializedName("emoji", alternate = ["Emoji"]) val emoji: String? = null ) enum class ConnectionStatus { CONNECTED, CONNECTING, DISCONNECTED } @Singleton class ChatHubClient @Inject constructor() { private var hubConnection: HubConnection? = null private val _events = MutableSharedFlow(extraBufferCapacity = 1024) val events: SharedFlow = _events.asSharedFlow() private val _status = MutableStateFlow(ConnectionStatus.DISCONNECTED) val status: StateFlow = _status.asStateFlow() private val scope = CoroutineScope(Dispatchers.IO) private var lastBaseUrl: String? = null private var lastToken: String? = null fun connect(baseUrl: String, accessToken: String) { if (hubConnection?.connectionState == HubConnectionState.CONNECTED) return lastBaseUrl = baseUrl lastToken = accessToken _status.value = ConnectionStatus.CONNECTING hubConnection = HubConnectionBuilder.create("${baseUrl}/hubs/chat") .withAccessTokenProvider(Single.just(accessToken)) .build() setupHandlers() hubConnection?.onClosed { exception -> Log.e("ChatHubClient", "Connection closed. Reconnecting...", exception) _status.value = ConnectionStatus.DISCONNECTED scope.launch { delay(5000) connect(baseUrl, accessToken) } } scope.launch { try { hubConnection?.start()?.blockingAwait() _status.value = ConnectionStatus.CONNECTED Log.d("ChatHubClient", "SignalR Connected") } catch (e: Exception) { Log.e("ChatHubClient", "SignalR Connection Error", e) _status.value = ConnectionStatus.DISCONNECTED } } } private fun setupHandlers() { hubConnection?.let { conn -> conn.on("new_message", { message: MessageDto -> _events.tryEmit(ChatEvent.NewMessage(message)) }, MessageDto::class.java) conn.on("messages_read", { data: MessagesReadEvent -> _events.tryEmit(ChatEvent.MessagesRead( data.effectiveChatId, data.effectiveUserId, data.effectiveLastReadSequenceId )) }, MessagesReadEvent::class.java) conn.on("user_typing", { data: ReactionEvent -> _events.tryEmit(ChatEvent.UserTyping(data.chatId ?: "", data.userId ?: "")) }, ReactionEvent::class.java) conn.on("user_stopped_typing", { data: ReactionEvent -> _events.tryEmit(ChatEvent.UserStoppedTyping(data.chatId ?: "", data.userId ?: "")) }, ReactionEvent::class.java) conn.on("user_online", { userId: String -> _events.tryEmit(ChatEvent.UserOnline(userId)) }, String::class.java) conn.on("new_chat", { chat: ChatDto -> _events.tryEmit(ChatEvent.NewChat(chat)) }, ChatDto::class.java) conn.on("reaction_added", { data: ReactionEvent -> _events.tryEmit(ChatEvent.ReactionUpdated( data.messageId ?: "", data.chatId ?: "", data.userId ?: "", data.emoji ?: "" )) }, ReactionEvent::class.java) conn.on("reaction_removed", { data: ReactionEvent -> _events.tryEmit(ChatEvent.ReactionUpdated( data.messageId ?: "", data.chatId ?: "", data.userId ?: "", "" // empty emoji signals removal )) }, ReactionEvent::class.java) // WebRTC Signaling Handlers conn.on("call_incoming", { chatId: String, from: String, offer: String, callType: String -> _events.tryEmit(ChatEvent.CallIncoming(chatId, from, offer, callType)) }, String::class.java, String::class.java, String::class.java, String::class.java) conn.on("call_answered", { chatId: String, answer: String -> _events.tryEmit(ChatEvent.CallAnswered(chatId, answer)) }, String::class.java, String::class.java) conn.on("ice_candidate", { chatId: String, candidate: String -> _events.tryEmit(ChatEvent.IceCandidateReceived(chatId, candidate)) }, String::class.java, String::class.java) conn.on("call_ended", { chatId: String -> _events.tryEmit(ChatEvent.CallEnded(chatId)) }, String::class.java) } } fun disconnect() { hubConnection?.stop() _status.value = ConnectionStatus.DISCONNECTED } fun readMessages(request: ReadMessagesRequest) { if (hubConnection?.connectionState == HubConnectionState.CONNECTED) { hubConnection?.send("read_messages", request) Log.d("ChatHubClient", "Sent read_messages for chat: ${request.chatId}") } } fun joinChat(chatId: String) { scope.launch { // Wait for connection to be established if it's currently connecting var attempts = 0 while (hubConnection?.connectionState != HubConnectionState.CONNECTED && attempts < 10) { delay(500) attempts++ } if (hubConnection?.connectionState == HubConnectionState.CONNECTED) { hubConnection?.send("join_chat", chatId) Log.d("ChatHubClient", "Joined chat room: $chatId") } else { Log.e("ChatHubClient", "Failed to join chat room $chatId: Not connected") } } } fun sendTypingIndicator(chatId: String) { if (hubConnection?.connectionState == HubConnectionState.CONNECTED) { hubConnection?.send("typing_start", chatId) Log.d("ChatHubClient", "Sent typing indicator for chat: $chatId") } } fun sendUserStoppedTyping(chatId: String) { if (hubConnection?.connectionState == HubConnectionState.CONNECTED) { hubConnection?.send("typing_stop", chatId) Log.d("ChatHubClient", "Sent user stopped typing for chat: $chatId") } } }