553 lines
27 KiB
Kotlin
553 lines
27 KiB
Kotlin
package chats.data.remote.signalr
|
|
|
|
import android.content.Context
|
|
import core.network.NetworkManager
|
|
import core.network.ServerConfig
|
|
import core.notifications.data.ActiveChatTracker
|
|
import core.notifications.data.NotificationHelper
|
|
import core.security.TokenManager
|
|
import chats.data.sync.MessageSyncWorker
|
|
import dagger.hilt.android.qualifiers.ApplicationContext
|
|
import kotlinx.coroutines.CoroutineScope
|
|
import kotlinx.coroutines.Dispatchers
|
|
import kotlinx.coroutines.SupervisorJob
|
|
import kotlinx.coroutines.launch
|
|
import kotlinx.coroutines.flow.filterIsInstance
|
|
import kotlinx.coroutines.flow.launchIn
|
|
import kotlinx.coroutines.flow.onEach
|
|
import kotlinx.coroutines.flow.filter
|
|
import kotlinx.coroutines.flow.distinctUntilChanged
|
|
import kotlinx.coroutines.flow.collect
|
|
import javax.inject.Inject
|
|
import javax.inject.Singleton
|
|
import chats.data.remote.signalr.ConnectionStatus
|
|
import okhttp3.OkHttpClient
|
|
import okhttp3.Request
|
|
import java.util.concurrent.TimeUnit
|
|
|
|
@Singleton
|
|
class SignalRNotificationObserver @Inject constructor(
|
|
private val signalrClient: ChatHubClient,
|
|
private val activeChatTracker: ActiveChatTracker,
|
|
private val tokenManager: TokenManager,
|
|
private val chatRepository: chats.domain.repository.ChatRepository,
|
|
private val serverConfig: ServerConfig,
|
|
private val networkManager: NetworkManager,
|
|
private val messageDao: core.database.data.MessageDao,
|
|
private val api: chats.data.remote.api.ChatApi,
|
|
@ApplicationContext private val context: Context
|
|
) {
|
|
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Main)
|
|
private var isStarted = false
|
|
private val processedMessageIds = mutableSetOf<String>()
|
|
|
|
// OkHttpClient для ping запроса
|
|
private val pingClient = OkHttpClient.Builder()
|
|
.connectTimeout(5, TimeUnit.SECONDS)
|
|
.readTimeout(5, TimeUnit.SECONDS)
|
|
.build()
|
|
|
|
fun start() {
|
|
if (isStarted) return
|
|
isStarted = true
|
|
|
|
// Запускаем мониторинг сети
|
|
networkManager.startMonitoring()
|
|
|
|
// Принудительно обновляем состояние сети при старте
|
|
networkManager.refreshNetworkState()
|
|
|
|
// Подключаемся к SignalR при старте приложения
|
|
connectSignalR()
|
|
|
|
// Initial count load
|
|
refresh()
|
|
|
|
// Запускаем периодическую проверку подключения SignalR
|
|
startConnectionHealthCheck()
|
|
|
|
// Слушаем восстановление сети и переподключаем SignalR
|
|
networkManager.isOnline
|
|
.filter { it } // Только переход в онлайн
|
|
.distinctUntilChanged()
|
|
.onEach {
|
|
android.util.Log.d("SignalRNtfObserver", "Network restored! Reconnecting SignalR and syncing...")
|
|
// Небольшая задержка чтобы сеть стабилизировалась
|
|
kotlinx.coroutines.delay(1500)
|
|
reconnectOnNetworkRestored()
|
|
}
|
|
.launchIn(scope)
|
|
|
|
// Также отслеживаем состояние SignalR для переподключения
|
|
signalrClient.status
|
|
.filter { it == ConnectionStatus.DISCONNECTED }
|
|
.onEach {
|
|
android.util.Log.d("SignalRNtfObserver", "SignalR disconnected, checking network...")
|
|
// Если мы offline и SignalR отключен - не делаем ничего
|
|
// Подключимся когда сеть восстановится
|
|
}
|
|
.launchIn(scope)
|
|
|
|
// При восстановлении соединения SignalR обновляем список чатов и вступаем в них
|
|
signalrClient.status
|
|
.filter { it == ConnectionStatus.CONNECTED }
|
|
.onEach {
|
|
android.util.Log.d("SignalRNtfObserver", "SignalR connected, refreshing chats and joining rooms")
|
|
onSignalRConnected()
|
|
}
|
|
.launchIn(scope)
|
|
|
|
// Слушаем события SignalR для уведомлений и обновления списка чатов
|
|
android.util.Log.d("SignalRNtfObserver", "Starting to listen to signalrClient.events")
|
|
signalrClient.events
|
|
.onEach { event ->
|
|
android.util.Log.d("SignalRNtfObserver", ">>> Received event: ${event::class.simpleName}")
|
|
when (event) {
|
|
is ChatEvent.NewMessage -> {
|
|
val currentUserId = tokenManager.getUserId()
|
|
val message = event.message
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "New message: ${message.id} from ${message.senderId}, current: $currentUserId")
|
|
|
|
// Don't show if it's our message or already processed
|
|
if (message.senderId == currentUserId) {
|
|
android.util.Log.d("SignalRNtfObserver", "Skipping - own message")
|
|
return@onEach
|
|
}
|
|
if (processedMessageIds.contains(message.id)) {
|
|
android.util.Log.d("SignalRNtfObserver", "Skipping - already processed")
|
|
return@onEach
|
|
}
|
|
|
|
// Mark as processed
|
|
processedMessageIds.add(message.id)
|
|
if (processedMessageIds.size > 200) {
|
|
processedMessageIds.remove(processedMessageIds.first())
|
|
}
|
|
|
|
// Increment immediately for UI feedback
|
|
activeChatTracker.incrementUnreadCount()
|
|
|
|
// Refresh total count from source of truth in background
|
|
refresh()
|
|
|
|
// Don't show if this chat is currently open
|
|
if (activeChatTracker.currentChatId.value == message.chatId) {
|
|
android.util.Log.d("SignalRNtfObserver", "Skipping - chat is open: ${message.chatId}")
|
|
return@onEach
|
|
}
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "Showing notification for chat: ${message.chatId}")
|
|
NotificationHelper.showNotification(
|
|
context = context,
|
|
title = message.sender?.displayName ?: "Новое сообщение",
|
|
body = message.content ?: "Вам прислали вложение",
|
|
type = "chat",
|
|
chatId = message.chatId,
|
|
notificationId = message.id.hashCode(),
|
|
totalCount = activeChatTracker.totalUnreadCount.value
|
|
)
|
|
}
|
|
is ChatEvent.MessagesRead -> {
|
|
android.util.Log.d("SignalRNtfObserver", "Messages read event")
|
|
// If anyone read messages, sync our total count
|
|
refresh()
|
|
}
|
|
else -> {
|
|
android.util.Log.d("SignalRNtfObserver", "Unhandled event: ${event::class.simpleName}")
|
|
Unit
|
|
}
|
|
}
|
|
}
|
|
.launchIn(scope)
|
|
}
|
|
|
|
private fun startConnectionHealthCheck() {
|
|
// Каждые 10 секунд проверяем подключение и при необходимости переподключаемся
|
|
scope.launch {
|
|
var consecutiveFailures = 0
|
|
|
|
while (true) {
|
|
kotlinx.coroutines.delay(10000)
|
|
val currentStatus = signalrClient.status.value
|
|
val isOnline = networkManager.isOnline.value
|
|
|
|
if (isOnline && currentStatus == ConnectionStatus.DISCONNECTED) {
|
|
consecutiveFailures++
|
|
android.util.Log.d("SignalRNtfObserver", "Health check: Network online but SignalR disconnected (failures: $consecutiveFailures), reconnecting...")
|
|
signalrClient.reconnect()
|
|
} else if (isOnline && currentStatus == ConnectionStatus.CONNECTED) {
|
|
// Сбрасываем счетчик ошибок при успешном подключении
|
|
consecutiveFailures = 0
|
|
android.util.Log.d("SignalRNtfObserver", "Health check: Connection healthy")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private fun onSignalRConnected() {
|
|
android.util.Log.d("SignalRNtfObserver", "SignalR connected - syncing missed messages...")
|
|
refresh()
|
|
// Вступаем во все чаты для получения событий
|
|
joinAllChats()
|
|
// Синхронизируем пропущенные сообщения
|
|
syncMissedMessages()
|
|
}
|
|
|
|
/**
|
|
* Синхронизирует пропущенные сообщения после восстановления соединения
|
|
* Запрашивает только НОВЫЕ сообщения с последнего известного sequenceId
|
|
* Показывает уведомления для непрочитанных сообщений
|
|
*/
|
|
private fun syncMissedMessages() {
|
|
scope.launch {
|
|
try {
|
|
android.util.Log.d("SignalRNtfObserver", "Starting missed messages sync...")
|
|
val chats = chatRepository.getChats()
|
|
android.util.Log.d("SignalRNtfObserver", "Syncing ${chats.size} chats for missed messages")
|
|
|
|
// Запрашиваем только новые сообщения для каждого чата
|
|
chats.forEach { chat ->
|
|
try {
|
|
// Получаем последний известный sequenceId из локальной базы
|
|
val lastSequenceId = chatRepository.getLastKnownSequenceId(chat.id)
|
|
|
|
if (lastSequenceId != null) {
|
|
// Запрашиваем сообщения ПОСЛЕ lastSequenceId (только новые)
|
|
android.util.Log.d("SignalRNtfObserver", "Chat ${chat.id}: last known seqId=$lastSequenceId, fetching newer...")
|
|
val newMessages = chatRepository.getMessages(
|
|
chatId = chat.id,
|
|
afterSequenceId = lastSequenceId.toLong(),
|
|
limit = 100
|
|
)
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "Chat ${chat.id}: fetched ${newMessages.size} messages from API")
|
|
|
|
// Показываем уведомление если есть новые сообщения и чат не открыт
|
|
if (newMessages.isNotEmpty() && activeChatTracker.currentChatId.value != chat.id) {
|
|
android.util.Log.d("SignalRNtfObserver", "Chat ${chat.id}: ${newMessages.size} new messages, showing notification")
|
|
showMissedMessagesNotification(chat, newMessages)
|
|
} else {
|
|
android.util.Log.d("SignalRNtfObserver", "Chat ${chat.id}: skipping notification (messages=${newMessages.size}, chatOpen=${activeChatTracker.currentChatId.value == chat.id})")
|
|
}
|
|
} else {
|
|
// Нет локальных сообщений - загружаем последние 50
|
|
android.util.Log.d("SignalRNtfObserver", "Chat ${chat.id}: no local messages, fetching last 50")
|
|
chatRepository.getMessages(chatId = chat.id, limit = 50)
|
|
}
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "Synced chat ${chat.id}")
|
|
// Небольшая пауза между чатами чтобы не перегружать сервер
|
|
kotlinx.coroutines.delay(100)
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "Failed to sync chat ${chat.id}", e)
|
|
}
|
|
}
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "Missed messages sync completed")
|
|
|
|
// Обновляем счетчик непрочитанных после синхронизации
|
|
refresh()
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "Sync failed", e)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Показывает уведомление о пропущенных сообщениях
|
|
*/
|
|
private fun showMissedMessagesNotification(chat: chats.domain.model.Chat, newMessages: List<chats.domain.model.Message>) {
|
|
scope.launch {
|
|
try {
|
|
android.util.Log.d("SignalRNtfObserver", "showMissedMessagesNotification called for chat ${chat.id} with ${newMessages.size} messages")
|
|
|
|
// Фильтруем сообщения не от текущего пользователя
|
|
val currentUserId = tokenManager.getUserId()
|
|
android.util.Log.d("SignalRNtfObserver", "Current user ID: $currentUserId")
|
|
|
|
val messagesFromOthers = newMessages.filter { it.senderId != currentUserId }
|
|
android.util.Log.d("SignalRNtfObserver", "Messages from others: ${messagesFromOthers.size}")
|
|
|
|
if (messagesFromOthers.isEmpty()) {
|
|
android.util.Log.d("SignalRNtfObserver", "No new messages from others in chat ${chat.id} - all messages are from current user")
|
|
return@launch
|
|
}
|
|
|
|
// Группируем сообщения по отправителям
|
|
val messagesBySender = messagesFromOthers.groupBy { it.senderId }
|
|
android.util.Log.d("SignalRNtfObserver", "Messages grouped by ${messagesBySender.size} sender(s)")
|
|
|
|
// Для каждого отправителя показываем уведомление
|
|
messagesBySender.forEach { (senderId, messages) ->
|
|
val senderName = messages.firstOrNull()?.senderName ?: "Контакт"
|
|
val messageCount = messages.size
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "Processing sender $senderName with $messageCount messages")
|
|
|
|
// Формируем текст уведомления
|
|
val notificationText = when {
|
|
messageCount == 1 -> {
|
|
messages.firstOrNull()?.content ?: "Новое сообщение"
|
|
}
|
|
messageCount <= 3 -> {
|
|
messages.take(3).mapNotNull { it.content }.joinToString(", ")
|
|
}
|
|
else -> {
|
|
"$messageCount новых сообщений"
|
|
}
|
|
}
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "Showing notification for chat ${chat.id}: $senderName - $notificationText")
|
|
|
|
// Небольшая задержка перед показом уведомления
|
|
kotlinx.coroutines.delay(500)
|
|
|
|
// Показываем уведомление
|
|
NotificationHelper.showNotification(
|
|
context = context,
|
|
title = senderName,
|
|
body = notificationText,
|
|
type = "chat",
|
|
chatId = chat.id,
|
|
notificationId = chat.id.hashCode(),
|
|
totalCount = activeChatTracker.totalUnreadCount.value
|
|
)
|
|
}
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "Failed to show missed messages notification", e)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Получает последний известный sequenceId для чата из локальной базы
|
|
*/
|
|
private suspend fun getLastKnownSequenceId(chatId: String): Int? {
|
|
return try {
|
|
chatRepository.getLastKnownSequenceId(chatId)
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "Failed to get last sequenceId for $chatId", e)
|
|
null
|
|
}
|
|
}
|
|
|
|
private fun reconnectOnNetworkRestored() {
|
|
android.util.Log.d("SignalRNtfObserver", "Network restored, starting reconnection sequence...")
|
|
|
|
scope.launch {
|
|
// 1. Сначала делаем HTTP ping запрос чтобы "разбудить" сетевой стек
|
|
android.util.Log.d("SignalRNtfObserver", "Sending HTTP ping to wake up network...")
|
|
val pingSuccess = sendHttpPing()
|
|
android.util.Log.d("SignalRNtfObserver", "HTTP ping result: $pingSuccess")
|
|
|
|
// 2. Небольшая пауза для стабилизации
|
|
kotlinx.coroutines.delay(1000)
|
|
|
|
// 3. Принудительное переподключение SignalR
|
|
android.util.Log.d("SignalRNtfObserver", "Forcing SignalR reconnect...")
|
|
signalrClient.reconnect()
|
|
|
|
// 4. Ждем пока SignalR подключится (максимум 10 секунд)
|
|
var waitCount = 0
|
|
while (signalrClient.status.value != ConnectionStatus.CONNECTED && waitCount < 20) {
|
|
kotlinx.coroutines.delay(500)
|
|
waitCount++
|
|
}
|
|
|
|
if (signalrClient.status.value == ConnectionStatus.CONNECTED) {
|
|
android.util.Log.d("SignalRNtfObserver", "SignalR reconnected, syncing missed messages...")
|
|
// 5. Синхронизируем пропущенные сообщения
|
|
syncMissedMessages()
|
|
|
|
// 6. Пробуем отправить отложенные сообщения НЕМЕДЛЕННО (не через WorkManager)
|
|
sendPendingMessagesImmediately()
|
|
|
|
// 7. Также планируем WorkManager на всякий случай
|
|
MessageSyncWorker.scheduleSync(context)
|
|
android.util.Log.d("SignalRNtfObserver", "Outgoing sync worker scheduled")
|
|
} else {
|
|
android.util.Log.w("SignalRNtfObserver", "SignalR failed to reconnect within timeout")
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Немедленно отправляет отложенные сообщения без ожидания WorkManager
|
|
* Вызывается сразу после восстановления соединения
|
|
*/
|
|
private fun sendPendingMessagesImmediately() {
|
|
scope.launch {
|
|
try {
|
|
android.util.Log.d("SignalRNtfObserver", "Checking for pending messages to send...")
|
|
|
|
// Получаем все сообщения со статусом SYNCING
|
|
val pendingMessages = messageDao.getPendingSyncMessages()
|
|
android.util.Log.d("SignalRNtfObserver", "Found ${pendingMessages.size} pending messages")
|
|
|
|
if (pendingMessages.isEmpty()) {
|
|
android.util.Log.d("SignalRNtfObserver", "No pending messages to send")
|
|
return@launch
|
|
}
|
|
|
|
var successCount = 0
|
|
var failureCount = 0
|
|
|
|
for (message in pendingMessages) {
|
|
try {
|
|
android.util.Log.d("SignalRNtfObserver", "Sending pending message: ${message.id}")
|
|
|
|
val attachments = parseAttachments(message.mediaJson)
|
|
val request = chats.data.remote.api.SendMessageRequest(
|
|
content = message.content,
|
|
type = message.mediaType.lowercase(),
|
|
attachments = attachments,
|
|
replyToId = message.replyToId
|
|
)
|
|
|
|
val response = api.sendMessage(message.chatId, request)
|
|
|
|
// Обновляем сообщение в базе
|
|
val syncedMessage = message.copy(
|
|
id = response.id,
|
|
sequenceId = response.sequenceId ?: message.sequenceId,
|
|
createdAt = response.createdAt ?: message.createdAt,
|
|
syncStatus = core.database.data.SyncStatus.SYNCED,
|
|
isDeletedLocally = false,
|
|
isEditedLocally = false,
|
|
editedContent = null,
|
|
lastUpdated = System.currentTimeMillis()
|
|
)
|
|
|
|
messageDao.insertMessage(syncedMessage)
|
|
android.util.Log.d("SignalRNtfObserver", "Message sent successfully: ${response.id}")
|
|
successCount++
|
|
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "Failed to send message ${message.id}", e)
|
|
failureCount++
|
|
}
|
|
}
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "Pending messages sync completed. Success: $successCount, Failed: $failureCount")
|
|
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "sendPendingMessagesImmediately failed", e)
|
|
}
|
|
}
|
|
}
|
|
|
|
private fun parseAttachments(mediaJson: String): List<chats.data.remote.api.AttachmentRequest>? {
|
|
return try {
|
|
val gson = com.google.gson.Gson()
|
|
val mediaList = gson.fromJson(mediaJson, Array::class.java)
|
|
?.map { elem ->
|
|
val map = elem as Map<*, *>
|
|
chats.data.remote.api.AttachmentRequest(
|
|
type = map["type"] as? String ?: "file",
|
|
url = map["url"] as? String ?: "",
|
|
fileName = map["filename"] as? String ?: "file",
|
|
fileSize = (map["size"] as? Number)?.toLong() ?: 0L
|
|
)
|
|
}
|
|
mediaList?.takeIf { it.isNotEmpty() }
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "Failed to parse attachments", e)
|
|
null
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Отправляет HTTP ping запрос для активации сетевого соединения
|
|
*/
|
|
private suspend fun sendHttpPing(): Boolean {
|
|
return try {
|
|
val baseUrl = serverConfig.getBaseUrl()
|
|
val token = tokenManager.getToken()
|
|
|
|
if (baseUrl.isBlank() || token == null) {
|
|
android.util.Log.w("SignalRNtfObserver", "Cannot ping: missing baseUrl or token")
|
|
return false
|
|
}
|
|
|
|
val url = "${baseUrl.removeSuffix("/api/")}/api/auth/refresh"
|
|
val request = Request.Builder()
|
|
.url(url)
|
|
.post(okhttp3.RequestBody.create(null, "{}"))
|
|
.addHeader("Authorization", "Bearer $token")
|
|
.build()
|
|
|
|
val response = pingClient.newCall(request).execute()
|
|
val success = response.isSuccessful || response.code == 401 // 401 OK для refresh
|
|
android.util.Log.d("SignalRNtfObserver", "HTTP ping to $url: ${response.code}")
|
|
response.close()
|
|
success
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "HTTP ping failed", e)
|
|
false
|
|
}
|
|
}
|
|
|
|
private fun connectSignalR() {
|
|
val token = tokenManager.getToken()
|
|
val baseUrl = serverConfig.getBaseUrl()
|
|
|
|
if (token == null || baseUrl.isBlank()) {
|
|
android.util.Log.w("SignalRNtfObserver", "Cannot connect SignalR: token=${token != null}, baseUrl=$baseUrl")
|
|
return
|
|
}
|
|
|
|
val isOnline = networkManager.isOnline.value
|
|
if (!isOnline) {
|
|
android.util.Log.w("SignalRNtfObserver", "Cannot connect SignalR: network is offline")
|
|
return
|
|
}
|
|
|
|
// Проверяем текущее состояние SignalR
|
|
val currentStatus = signalrClient.status.value
|
|
if (currentStatus == ConnectionStatus.CONNECTED) {
|
|
android.util.Log.d("SignalRNtfObserver", "SignalR already connected, skipping")
|
|
return
|
|
}
|
|
|
|
android.util.Log.d("SignalRNtfObserver", "Connecting SignalR with token: ${token.take(10)}..., baseUrl: $baseUrl")
|
|
signalrClient.connect(baseUrl.removeSuffix("/api/"), token)
|
|
|
|
// Если подключение не удалось в течение 5 секунд - пробуем снова
|
|
scope.launch {
|
|
kotlinx.coroutines.delay(5000)
|
|
if (signalrClient.status.value == ConnectionStatus.DISCONNECTED) {
|
|
android.util.Log.w("SignalRNtfObserver", "SignalR connection timeout, retrying with reconnect()...")
|
|
signalrClient.reconnect()
|
|
}
|
|
}
|
|
}
|
|
|
|
private fun joinAllChats() {
|
|
scope.launch {
|
|
try {
|
|
val chats = chatRepository.getChats()
|
|
android.util.Log.d("SignalRNtfObserver", "Joining ${chats.size} chat rooms")
|
|
chats.forEach { chat ->
|
|
signalrClient.joinChat(chat.id)
|
|
}
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "Failed to join chats", e)
|
|
}
|
|
}
|
|
}
|
|
|
|
fun refresh() {
|
|
scope.launch {
|
|
try {
|
|
val chats = chatRepository.getChats()
|
|
val total = chats.sumOf { it.unreadCount }
|
|
activeChatTracker.setTotalUnreadCount(total)
|
|
android.util.Log.d("SignalRNtfObserver", "Refreshed chats: ${chats.size}, total unread: $total")
|
|
} catch (e: Exception) {
|
|
android.util.Log.e("SignalRNtfObserver", "Failed to refresh", e)
|
|
}
|
|
}
|
|
}
|
|
}
|