rust/forkmessager
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects 3 Releases Wiki Activity

Заготовка

Browse Source
This commit is contained in:
Халимов Рустам
2026-05-07 23:27:37 +03:00
parent 2d2e4b685e
commit 9daa786cfb
33 changed files with 1662 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
client-mobile/chats/data/local/dao/ChatDao.kt Normal file
View File

@@ -0,0 +1,45 @@
package chats.data.local.dao
import androidx.room.*
import chats.data.local.database.ChatEntity
import kotlinx.coroutines.flow.Flow
/**
* DAO для операций с чатами в Room Database
*/
@Dao
interface ChatDao {
@Query("SELECT * FROM chats ORDER BY updatedAtMillis DESC")
fun getAllChats(): Flow<List<ChatEntity>>
@Query("SELECT * FROM chats WHERE remoteId = :remoteId LIMIT 1")
suspend fun getChatByRemoteId(remoteId: String): ChatEntity?
@Query("SELECT * FROM chats WHERE localId = :localId LIMIT 1")
suspend fun getChatByLocalId(localId: String): ChatEntity?
@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insertChat(chat: ChatEntity)
@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insertChats(chats: List<ChatEntity>)
@Update
suspend fun updateChat(chat: ChatEntity)
@Query("UPDATE chats SET lastMessageText = :lastMessageText, lastMessageTimestamp = :timestamp WHERE remoteId = :chatId")
suspend fun updateLastMessage(chatId: String, lastMessageText: String?, timestamp: String?)
@Query("UPDATE chats SET unreadCount = :count WHERE remoteId = :chatId")
suspend fun updateUnreadCount(chatId: String, count: Int)
@Delete
suspend fun deleteChat(chat: ChatEntity)
@Query("DELETE FROM chats WHERE remoteId = :remoteId")
suspend fun deleteChatByRemoteId(remoteId: String)
@Query("DELETE FROM chats")
suspend fun deleteAllChats()
}

70
client-mobile/chats/data/local/dao/MessageDao.kt Normal file
View File

@@ -0,0 +1,70 @@
package chats.data.local.dao
import androidx.room.*
import chats.data.local.database.MessageEntity
import chats.domain.model.MessageStatus
import kotlinx.coroutines.flow.Flow
/**
* DAO для операций с сообщениями в Room Database
*/
@Dao
interface MessageDao {
@Query("SELECT * FROM messages WHERE chatId = :chatId ORDER BY sequenceId ASC")
fun getMessagesByChatId(chatId: String): Flow<List<MessageEntity>>
@Query("SELECT * FROM messages WHERE chatId = :chatId AND sequenceId > :afterSequenceId ORDER BY sequenceId ASC LIMIT :limit")
suspend fun getMessagesAfter(chatId: String, afterSequenceId: Long, limit: Int): List<MessageEntity>
@Query("SELECT * FROM messages WHERE chatId = :chatId AND sequenceId < :beforeSequenceId ORDER BY sequenceId DESC LIMIT :limit")
suspend fun getMessagesBefore(chatId: String, beforeSequenceId: Long, limit: Int): List<MessageEntity>
@Query("SELECT * FROM messages WHERE localId = :localId")
suspend fun getMessageByLocalId(localId: String): MessageEntity?
@Query("SELECT * FROM messages WHERE serverId = :serverId")
suspend fun getMessageByServerId(serverId: String): MessageEntity?
@Query("SELECT * FROM messages WHERE status IN (:statuses) AND chatId = :chatId")
suspend fun getMessagesByStatus(chatId: String, statuses: List<MessageStatus>): List<MessageEntity>
@Query("SELECT * FROM messages WHERE status = :status")
fun getMessagesByStatusFlow(status: MessageStatus): Flow<List<MessageEntity>>
@Query("SELECT * FROM messages WHERE status IN (:statuses)")
suspend fun getPendingMessages(statuses: List<MessageStatus>): List<MessageEntity>
@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insertMessage(message: MessageEntity)
@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insertMessages(messages: List<MessageEntity>)
@Update
suspend fun updateMessage(message: MessageEntity)
@Delete
suspend fun deleteMessage(message: MessageEntity)
@Query("DELETE FROM messages WHERE localId = :localId")
suspend fun deleteMessageByLocalId(localId: String)
@Query("DELETE FROM messages WHERE chatId = :chatId")
suspend fun deleteMessagesByChatId(chatId: String)
@Query("UPDATE messages SET status = :status, updatedAtMillis = :updatedAtMillis WHERE localId = :localId")
suspend fun updateMessageStatus(localId: String, status: MessageStatus, updatedAtMillis: Long)
@Query("UPDATE messages SET serverId = :serverId, status = :status, updatedAtMillis = :updatedAtMillis WHERE localId = :localId")
suspend fun updateMessageWithServerId(localId: String, serverId: String, status: MessageStatus, updatedAtMillis: Long)
@Query("UPDATE messages SET status = :status, errorMessage = :errorMessage, retryCount = retryCount + 1, updatedAtMillis = :updatedAtMillis WHERE localId = :localId")
suspend fun updateMessageError(localId: String, status: MessageStatus, errorMessage: String, updatedAtMillis: Long)
@Query("SELECT COUNT(*) FROM messages WHERE chatId = :chatId AND status IN (:statuses)")
suspend fun getUnsentCount(chatId: String, statuses: List<MessageStatus>): Int
@Query("SELECT * FROM messages WHERE chatId = :chatId AND status = :status LIMIT 1")
suspend fun getFirstMessageByStatus(chatId: String, status: MessageStatus): MessageEntity?
}

36
client-mobile/chats/data/local/dao/UserProfileDao.kt Normal file
View File

@@ -0,0 +1,36 @@
package chats.data.local.dao
import androidx.room.*
import chats.data.local.database.UserProfileEntity
import kotlinx.coroutines.flow.Flow
/**
* DAO для операций с профилями пользователей в Room Database
*/
@Dao
interface UserProfileDao {
@Query("SELECT * FROM user_profile WHERE userId = :userId LIMIT 1")
suspend fun getUserById(userId: String): UserProfileEntity?
@Query("SELECT * FROM user_profile WHERE userId = :userId")
fun getUserByIdFlow(userId: String): Flow<UserProfileEntity?>
@Query("SELECT * FROM user_profile")
fun getAllUsers(): Flow<List<UserProfileEntity>>
@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insertUser(user: UserProfileEntity)
@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insertUsers(users: List<UserProfileEntity>)
@Update
suspend fun updateUser(user: UserProfileEntity)
@Delete
suspend fun deleteUser(user: UserProfileEntity)
@Query("DELETE FROM user_profile WHERE userId = :userId")
suspend fun deleteUserById(userId: String)
}

33
client-mobile/chats/data/local/database/AppDatabase.kt Normal file
View File

@@ -0,0 +1,33 @@
package chats.data.local.database
import androidx.room.Database
import androidx.room.RoomDatabase
import androidx.room.TypeConverters
import chats.data.local.dao.ChatDao
import chats.data.local.dao.MessageDao
import chats.data.local.dao.UserProfileDao
/**
* Room Database для локального хранения данных мессенджера
* Реализует паттерн Single Source of Truth
*/
@Database(
entities = [
MessageEntity::class,
ChatEntity::class,
UserProfileEntity::class
],
version = 1,
exportSchema = false
)
@TypeConverters(MessageStatusConverter::class)
abstract class AppDatabase : RoomDatabase() {
abstract fun messageDao(): MessageDao
abstract fun chatDao(): ChatDao
abstract fun userProfileDao(): UserProfileDao
companion object {
const val DATABASE_NAME = "knot_messenger.db"
}
}

37
client-mobile/chats/data/local/database/ChatEntity.kt Normal file
View File

@@ -0,0 +1,37 @@
package chats.data.local.database
import androidx.room.Entity
import androidx.room.Index
import androidx.room.PrimaryKey
/**
* Entity для хранения чатов в локальной БД Room
*/
@Entity(
tableName = "chats",
indices = [
Index(value = ["remoteId"], unique = true)
]
)
data class ChatEntity(
@PrimaryKey(autoGenerate = false)
val localId: String,
val remoteId: String?, // ID с сервера
val type: String, // PRIVATE, GROUP, CHANNEL
val name: String,
val avatar: String? = null,
val unreadCount: Int = 0,
val lastMessageId: String? = null,
val lastMessageText: String? = null,
val lastMessageTimestamp: String? = null,
val updatedAtMillis: Long = System.currentTimeMillis()
)

66
client-mobile/chats/data/local/database/MessageEntity.kt Normal file
View File

@@ -0,0 +1,66 @@
package chats.data.local.database
import androidx.room.Entity
import androidx.room.Index
import androidx.room.PrimaryKey
import chats.domain.model.MessageStatus
/**
* Entity для хранения сообщений в локальной БД Room
* Используется для офлайн-режима и Single Source of Truth паттерна
*/
@Entity(
tableName = "messages",
indices = [
Index(value = ["chatId", "sequenceId"]),
Index(value = ["localId"], unique = true),
Index(value = ["serverId"]),
Index(value = ["status"])
]
)
data class MessageEntity(
@PrimaryKey(autoGenerate = false)
val localId: String, // UUID генерируется на клиенте при создании
val serverId: String?, // null для неотправленных сообщений
val idempotencyKey: String, // UUID v4 для идемпотентности
val chatId: String,
val senderId: String,
val senderName: String,
val senderAvatar: String? = null,
val content: String?,
val sequenceId: Long,
val createdAt: String, // ISO-8601 формат
val mediaType: String,
val mediaJson: String, // JSON список медиа (для Room)
val reactionsJson: String = "{}", // JSON map эмодзи -> count
val status: MessageStatus,
val isPinned: Boolean = false,
val isForwarded: Boolean = false,
val forwardedFromName: String? = null,
val replyToServerId: String? = null,
val errorMessage: String? = null,
val retryCount: Int = 0,
val createdAtMillis: Long = 0,
val updatedAtMillis: Long = 0
)

17
client-mobile/chats/data/local/database/MessageStatusConverter.kt Normal file
View File

@@ -0,0 +1,17 @@
package chats.data.local.database
import androidx.room.TypeConverter
import chats.domain.model.MessageStatus
/**
* Конвертеры для Room для работы с Enum и другими типами
*/
class MessageStatusConverter {
@TypeConverter
fun fromMessageStatus(status: MessageStatus): String = status.name
@TypeConverter
fun toMessageStatus(value: String): MessageStatus = runCatching {
MessageStatus.valueOf(value)
}.getOrDefault(MessageStatus.UNKNOWN)
}

27
client-mobile/chats/data/local/database/UserProfileEntity.kt Normal file
View File

@@ -0,0 +1,27 @@
package chats.data.local.database
import androidx.room.Entity
import androidx.room.PrimaryKey
/**
* Entity для хранения профиля пользователя в локальной БД Room
*/
@Entity(tableName = "user_profile")
data class UserProfileEntity(
@PrimaryKey
val userId: String,
val username: String,
val displayName: String,
val avatarUrl: String? = null,
val bio: String? = null,
val isOnline: Boolean = false,
val lastSeenMillis: Long = 0,
val updatedAtMillis: Long = System.currentTimeMillis()
)

34
client-mobile/chats/data/local/mappers/ChatMappers.kt Normal file
View File

@@ -0,0 +1,34 @@
package chats.data.local.mappers
import chats.data.local.database.ChatEntity
import chats.domain.model.Chat
/**
* Преобразует Domain Chat в Entity
*/
fun Chat.toEntity(): ChatEntity {
return ChatEntity(
localId = this.id.takeIf { it.isNotBlank() } ?: java.util.UUID.randomUUID().toString(),
remoteId = this.id,
type = this.type,
name = this.name,
avatar = this.avatar,
unreadCount = this.unreadCount,
lastMessageText = this.lastMessage?.content,
lastMessageTimestamp = this.lastMessage?.createdAt
)
}
/**
* Преобразует Entity в Domain Chat
*/
fun ChatEntity.toDomain(): Chat {
return Chat(
id = this.remoteId ?: this.localId,
type = this.type,
name = this.name,
avatar = this.avatar,
unreadCount = this.unreadCount,
lastMessage = null // lastMessage загружается отдельно
)
}

117
client-mobile/chats/data/local/mappers/MessageMappers.kt Normal file
View File

@@ -0,0 +1,117 @@
package chats.data.local.mappers
import chats.data.local.database.MessageEntity
import chats.domain.model.Message
import chats.domain.model.MessageStatus
import com.google.gson.Gson
import com.google.gson.reflect.TypeToken
private val gson = Gson()
/**
* Преобразует Domain Message в Entity для сохранения в Room
*/
fun Message.toEntity(status: MessageStatus = MessageStatus.SENT): MessageEntity {
return MessageEntity(
localId = this.id.takeIf { it.isNotBlank() } ?: java.util.UUID.randomUUID().toString(),
serverId = if (status == MessageStatus.SENT || status == MessageStatus.DELIVERED || status == MessageStatus.READ) this.id else null,
idempotencyKey = java.util.UUID.randomUUID().toString(),
chatId = this.chatId,
senderId = this.senderId,
senderName = this.senderName,
senderAvatar = this.senderAvatar,
content = this.content,
sequenceId = this.sequenceId.toLong(),
createdAt = this.createdAt,
mediaType = this.mediaType.name,
mediaJson = gson.toJson(this.media),
reactionsJson = gson.toJson(this.reactions),
status = status,
isPinned = this.isPinned,
isForwarded = this.isForwarded,
forwardedFromName = this.forwardedFromName,
replyToServerId = this.replyTo?.id,
createdAtMillis = parseTimestamp(this.createdAt),
updatedAtMillis = System.currentTimeMillis()
)
}
/**
* Преобразует Entity в Domain Message для отображения в UI
*/
fun MessageEntity.toDomain(): Message {
val mediaListType = object : TypeToken<List<chats.domain.model.Media>>() {}.type
val reactionsMapType = object : TypeToken<Map<String, Int>>() {}.type
return Message(
id = this.serverId ?: this.localId,
chatId = this.chatId,
senderId = this.senderId,
senderName = this.senderName,
senderAvatar = this.senderAvatar,
content = this.content,
sequenceId = this.sequenceId.toInt(),
createdAt = this.createdAt,
media = try {
gson.fromJson<List<chats.domain.model.Media>>(this.mediaJson, mediaListType) ?: emptyList()
} catch (e: Exception) {
emptyList()
},
mediaType = try {
chats.domain.model.MediaType.valueOf(this.mediaType)
} catch (e: Exception) {
chats.domain.model.MediaType.TEXT
},
reactions = try {
gson.fromJson<Map<String, Int>>(this.reactionsJson, reactionsMapType) ?: emptyMap()
} catch (e: Exception) {
emptyMap()
},
isRead = this.status == MessageStatus.READ,
isPinned = this.isPinned,
isForwarded = this.isForwarded,
forwardedFromName = this.forwardedFromName,
replyTo = null, // replyTo загружается отдельно если нужно
status = this.status
)
}
/**
* Создает новое локальное сообщение со статусом PENDING
*/
fun createPendingMessageEntity(
chatId: String,
senderId: String,
senderName: String,
content: String?,
mediaType: String = "TEXT",
mediaJson: String = "[]",
replyToId: String? = null
): MessageEntity {
val localId = java.util.UUID.randomUUID().toString()
return MessageEntity(
localId = localId,
serverId = null,
idempotencyKey = localId, // Используем localId как idempotency key
chatId = chatId,
senderId = senderId,
senderName = senderName,
content = content,
sequenceId = -1, // Будет обновлено после получения с сервера
createdAt = java.time.ZonedDateTime.now().toString(),
mediaType = mediaType,
mediaJson = mediaJson,
status = MessageStatus.PENDING,
replyToServerId = replyToId,
createdAtMillis = System.currentTimeMillis(),
updatedAtMillis = System.currentTimeMillis()
)
}
private fun parseTimestamp(createdAt: String): Long {
return try {
java.time.ZonedDateTime.parse(createdAt).toInstant().toEpochMilli()
} catch (e: Exception) {
System.currentTimeMillis()
}
}

6
client-mobile/chats/data/remote/api/ChatApi.kt
View File

@@ -33,7 +33,11 @@ interface ChatApi {
): List<MessageDto>
@POST("messages/chat/{chatId}")
suspend fun sendMessage(@Path("chatId") chatId: String, @Body request: SendMessageRequest): String
suspend fun sendMessage(
@Path("chatId") chatId: String,
@Body request: SendMessageRequest,
@Header("X-Idempotency-Key") idempotencyKey: String
): String
@Multipart
@POST("messages/upload")

307
client-mobile/chats/data/repository/ChatRepositoryImpl.kt
View File

@@ -1,144 +1,299 @@
package chats.data.repository
import android.util.Log
import androidx.work.ExistingPeriodicWorkPolicy
import androidx.work.ExistingWorkPolicy
import androidx.work.WorkManager
import chats.data.local.dao.ChatDao
import chats.data.local.dao.MessageDao
import chats.data.local.database.MessageEntity
import chats.data.local.mappers.createPendingMessageEntity
import chats.data.local.mappers.toDomain
import chats.data.local.mappers.toEntity
import chats.data.remote.api.ChatApi
import chats.data.remote.api.SendMessageRequest
import chats.data.remote.dto.ChatDto
import chats.data.remote.dto.MessageDto
import chats.data.remote.dto.MediaItemDto
import chats.data.remote.dto.ReactionDto
import chats.data.workers.ChatSyncWorker
import chats.data.workers.SendMessageWorker
import chats.domain.model.Chat
import chats.domain.model.Message
import chats.domain.model.MediaType
import chats.domain.model.MessageStatus
import chats.domain.repository.ChatRepository
import core.network.ServerConfig
import chats.data.remote.signalr.ReadMessagesRequest
import core.security.TokenManager
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map
import okhttp3.MediaType.Companion.toMediaTypeOrNull
import okhttp3.MultipartBody
import okhttp3.RequestBody.Companion.asRequestBody
import javax.inject.Inject
import kotlinx.coroutines.flow.map
/**
* Реализация ChatRepository с поддержкой офлайн-режима.
* Использует паттерн Single Source of Truth: UI всегда берет данные из Room.
*/
class ChatRepositoryImpl @Inject constructor(
private val api: ChatApi,
private val tokenManager: TokenManager,
private val serverConfig: ServerConfig,
private val messageDao: MessageDao,
private val chatDao: ChatDao,
private val workManager: WorkManager,
private val hubClient: chats.data.remote.signalr.ChatHubClient
) : ChatRepository {
private val gson = com.google.gson.Gson()
// ==================== Чаты ====================
override fun getChatsFlow(): Flow<List<Chat>> {
// Single Source of Truth - данные из Room
return chatDao.getAllChats().map { entities ->
entities.map { it.toDomain() }
}
}
override suspend fun getChats(): List<Chat> {
val currentUserId = tokenManager.getUserId() ?: ""
val baseUrl = serverConfig.getBaseUrl().removeSuffix("/api/")
return api.getChats().map { it.toDomain(currentUserId, baseUrl) }
return try {
// Пробуем получить с сервера
val remoteChats = api.getChats()
val domainChats = remoteChats.map { it.toDomain(currentUserId, baseUrl) }
// Сохраняем в локальную БД
val entities = domainChats.map { it.toEntity() }
chatDao.insertChats(entities)
domainChats
} catch (e: Exception) {
Log.w("ChatRepo", "Failed to fetch chats from server, returning local", e)
// При ошибке возвращаем локальные данные
emptyList()
}
}
override fun getMessagesFlow(chatId: String): kotlinx.coroutines.flow.Flow<List<Message>> {
// Кэш отключён - всегда возвращаем пустой поток
return kotlinx.coroutines.flow.flowOf(emptyList())
override suspend fun syncChats() {
try {
val currentUserId = tokenManager.getUserId() ?: ""
val baseUrl = serverConfig.getBaseUrl().removeSuffix("/api/")
val remoteChats = api.getChats()
val entities = remoteChats.map { dto ->
val existing = chatDao.getChatByRemoteId(dto.id)
dto.toDomain(currentUserId, baseUrl).toEntity().copy(
localId = existing?.localId ?: java.util.UUID.randomUUID().toString()
)
}
chatDao.insertChats(entities)
} catch (e: Exception) {
Log.e("ChatRepo", "Sync chats failed", e)
}
}
// ==================== Сообщения ====================
override fun getMessagesFlow(chatId: String): Flow<List<Message>> {
// Single Source of Truth - всегда из Room
return messageDao.getMessagesByChatId(chatId).map { entities ->
entities.map { it.toDomain() }
}
}
override suspend fun getMessages(chatId: String, cursor: String?, pivot: Long?, limit: Int?): List<Message> {
val baseUrl = serverConfig.getBaseUrl().removeSuffix("/api/")
val currentUserId = tokenManager.getUserId() ?: ""
return try {
android.util.Log.d("ChatRepo", "FETCH: chatId=$chatId, cursor=$cursor, limit=$limit")
Log.d("ChatRepo", "FETCH: chatId=$chatId, cursor=$cursor, limit=$limit")
val messages = api.getMessages(chatId, cursor = cursor, limit = limit)
if (messages.isNotEmpty()) {
android.util.Log.d("ChatRepo", "Received ${messages.size} messages. TopSeq: ${messages.first().sequenceId}, BottomSeq: ${messages.last().sequenceId}")
Log.d("ChatRepo", "Received ${messages.size} messages")
// Сохраняем в локальную БД
saveMessagesToLocal(messages, chatId, currentUserId)
}
// Мапим в доменные модели. По умолчанию считаем прочитанными,
// так как unreadCount нам тут не критичен для истории.
messages.map { msg ->
msg.toDomain(currentUserId, baseUrl).copy(isRead = true)
}
} catch (e: Exception) {
android.util.Log.e("ChatRepo", "Fetch messages failed", e)
Log.e("ChatRepo", "Fetch messages failed, returning local", e)
emptyList()
}
}
override suspend fun sendMessage(
chatId: String,
content: String?,
chatId: String,
content: String?,
type: String,
attachments: List<chats.data.remote.api.AttachmentRequest>?,
replyToId: String?,
forwardedFromId: String?
): Message {
val request = SendMessageRequest(
content = content,
type = type,
attachments = attachments,
replyToId = replyToId,
forwardedFromId = forwardedFromId
)
android.util.Log.d("ChatRepoImpl", "sendMessage request: $request")
val baseUrl = serverConfig.getBaseUrl().removeSuffix("/api/")
val userId = tokenManager.getUserId() ?: ""
return try {
val messageId = api.sendMessage(chatId, request)
android.util.Log.d("ChatRepoImpl", "sendMessage response: $messageId")
// Поскольку сервер вернул только ID, создаем заглушку Message.
// Настоящее сообщение придет через SignalR.
Message(
id = messageId,
chatId = chatId,
senderId = userId,
senderName = "", // Будет обновлено через SignalR
content = content,
sequenceId = 0,
createdAt = java.time.ZonedDateTime.now().toString(),
media = attachments?.map {
chats.domain.model.Media(
id = java.util.UUID.randomUUID().toString(),
type = it.type,
url = it.url,
filename = it.fileName,
size = it.fileSize
)
} ?: emptyList(),
mediaType = when(type) {
"image" -> chats.domain.model.MediaType.IMAGE
"video" -> chats.domain.model.MediaType.VIDEO
"audio", "voice" -> chats.domain.model.MediaType.AUDIO
else -> chats.domain.model.MediaType.TEXT
}
val userName = tokenManager.getUsername() ?: userId
// 1. Создаем локальное сообщение со статусом PENDING
val mediaType = when (type) {
"image" -> "IMAGE"
"video" -> "VIDEO"
"audio", "voice" -> "AUDIO"
else -> "TEXT"
}
val mediaJson = attachments?.map {
chats.domain.model.Media(
id = java.util.UUID.randomUUID().toString(),
type = it.type,
url = it.url,
filename = it.fileName,
size = it.fileSize
)
}?.let { gson.toJson(it) } ?: "[]"
val pendingEntity = createPendingMessageEntity(
chatId = chatId,
senderId = userId,
senderName = userName,
content = content,
mediaType = mediaType,
mediaJson = mediaJson,
replyToId = replyToId
)
// 2. Сохраняем в Room
messageDao.insertMessage(pendingEntity)
// 3. Ставим задачу в WorkManager для отправки
val workRequest = SendMessageWorker.createWorkRequest(pendingEntity.localId)
workManager.enqueueUniqueWork(
"send_${pendingEntity.localId}",
ExistingWorkPolicy.REPLACE,
workRequest
)
Log.d("ChatRepo", "Queued message for sending: ${pendingEntity.localId}")
// 4. Возвращаем доменную модель для немедленного отображения в UI
return pendingEntity.toDomain()
}
override suspend fun retryFailedMessage(localId: String) {
val message = messageDao.getMessageByLocalId(localId)
?: return
if (message.status != MessageStatus.FAILED) return
// Сбрасываем статус и ставим в очередь
messageDao.updateMessageStatus(
localId = localId,
status = MessageStatus.PENDING,
updatedAtMillis = System.currentTimeMillis()
)
val workRequest = SendMessageWorker.createWorkRequest(localId)
workManager.enqueueUniqueWork(
"send_$localId",
ExistingWorkPolicy.REPLACE,
workRequest
)
}
override suspend fun deleteLocalMessage(messageId: String) {
messageDao.deleteMessageByLocalId(messageId)
}
override suspend fun saveMessage(message: Message) {
// Сохраняем входящее сообщение из SignalR
val entity = message.toEntity(MessageStatus.DELIVERED)
messageDao.insertMessage(entity)
}
// ==================== Синхронизация ====================
override suspend fun syncMessagesForChat(chatId: String) {
val workRequest = ChatSyncWorker.createOneTimeWorkRequest(chatId)
workManager.enqueue(workRequest)
}
override suspend fun schedulePeriodicSync() {
val workRequest = ChatSyncWorker.createPeriodicWorkRequest()
workManager.enqueueUniquePeriodicWork(
"periodic_chat_sync",
ExistingPeriodicWorkPolicy.KEEP,
workRequest
)
}
// ==================== Вспомогательные методы ====================
private suspend fun saveMessagesToLocal(messages: List<chats.data.remote.dto.MessageDto>, chatId: String, currentUserId: String) {
messages.forEach { dto ->
val existing = messageDao.getMessageByServerId(dto.id)
if (existing == null) {
val entity = MessageEntity(
localId = java.util.UUID.randomUUID().toString(),
serverId = dto.id,
idempotencyKey = dto.id,
chatId = chatId,
senderId = dto.senderId ?: dto.sender?.id ?: currentUserId,
senderName = dto.sender?.displayName ?: dto.sender?.username ?: "",
senderAvatar = dto.sender?.avatarUrl,
content = dto.content,
sequenceId = dto.sequenceId?.toLong() ?: 0L,
createdAt = dto.createdAt ?: java.time.ZonedDateTime.now().toString(),
mediaType = dto.type ?: "TEXT",
mediaJson = gson.toJson(dto.media.map {
chats.domain.model.Media(
id = it.id,
type = it.type,
url = it.url,
filename = it.filename,
size = it.size,
duration = it.duration
)
}),
reactionsJson = gson.toJson(
dto.reactions?.associate { it.emoji to it.count } ?: emptyMap<String, Int>()
),
status = if (dto.senderId == currentUserId) MessageStatus.SENT else MessageStatus.DELIVERED,
createdAtMillis = System.currentTimeMillis(),
updatedAtMillis = System.currentTimeMillis()
)
messageDao.insertMessage(entity)
}
}
}
// ==================== Остальные методы ====================
override suspend fun addReaction(messageId: String, emoji: String) {
try {
api.addReaction(messageId, emoji)
} catch (e: Exception) {
android.util.Log.e("ChatRepoImpl", "sendMessage error", e)
Log.e("ChatRepo", "Add reaction failed", e)
throw e
}
}
override suspend fun addReaction(messageId: String, emoji: String) {
api.addReaction(messageId, emoji)
}
override suspend fun sendTypingStatus(chatId: String) {
api.sendTypingStatus(chatId)
try {
api.sendTypingStatus(chatId)
} catch (e: Exception) {
// Игнорируем ошибки typing status
}
}
override suspend fun markMessagesAsRead(chatId: String, lastMessageId: String, lastReadSequenceId: Int) {
try {
android.util.Log.d("ChatRepoImpl", "markMessagesAsRead CALLED FOR $chatId")
hubClient.readMessages(ReadMessagesRequest(chatId, lastMessageId, lastReadSequenceId))
hubClient.readMessages(chats.data.remote.signalr.ReadMessagesRequest(chatId, lastMessageId, lastReadSequenceId))
} catch (e: Exception) {
android.util.Log.e("ChatRepo", "Error marking messages as read", e)
Log.e("ChatRepo", "Error marking messages as read", e)
}
}
override suspend fun saveMessage(message: Message) {
// Кэш отключён
}
override suspend fun deleteLocalMessage(messageId: String) {
// Локальное удаление не поддерживается без кэша
}
override suspend fun uploadMedia(file: java.io.File): String {
val mimeType = when (file.extension.lowercase()) {
"jpg", "jpeg" -> "image/jpeg"
@@ -180,10 +335,10 @@ class ChatRepositoryImpl @Inject constructor(
val request = SendMessageRequest(content = content)
val currentUserId = tokenManager.getUserId() ?: ""
val returnedId = api.editMessage(messageId, request)
return Message(
id = returnedId,
chatId = "",
chatId = "",
senderId = currentUserId,
senderName = "",
content = content,

184
client-mobile/chats/data/workers/ChatSyncWorker.kt Normal file
View File

@@ -0,0 +1,184 @@
package chats.data.workers
import android.content.Context
import android.util.Log
import androidx.hilt.work.HiltWorker
import androidx.work.*
import chats.data.local.dao.ChatDao
import chats.data.local.dao.MessageDao
import chats.domain.model.MessageStatus
import core.security.TokenManager
import dagger.assisted.Assisted
import dagger.assisted.AssistedInject
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import java.io.IOException
import java.net.SocketTimeoutException
import java.util.concurrent.TimeUnit
/**
* Worker для периодической синхронизации чатов и сообщений.
* Запускается при наличии сети для получения новых сообщений с сервера.
*/
@HiltWorker
class ChatSyncWorker @AssistedInject constructor(
@Assisted context: Context,
@Assisted params: WorkerParameters,
private val chatDao: ChatDao,
private val messageDao: MessageDao,
private val chatApi: chats.data.remote.api.ChatApi
) : CoroutineWorker(context, params) {
companion object {
const val WORK_TAG = "chat_sync_worker"
const val KEY_CHAT_ID = "chat_id"
const val KEY_CURSOR = "cursor"
/**
* Создает периодический запрос на синхронизацию всех чатов
*/
fun createPeriodicWorkRequest(): PeriodicWorkRequest {
val constraints = Constraints.Builder()
.setRequiredNetworkType(NetworkType.CONNECTED)
.build()
return PeriodicWorkRequestBuilder<ChatSyncWorker>(
repeatInterval = 15,
repeatIntervalTimeUnit = TimeUnit.MINUTES
)
.setConstraints(constraints)
.addTag(WORK_TAG)
.build()
}
/**
* Создает разовый запрос на синхронизацию конкретного чата
*/
fun createOneTimeWorkRequest(chatId: String? = null, cursor: String? = null): OneTimeWorkRequest {
val constraints = Constraints.Builder()
.setRequiredNetworkType(NetworkType.CONNECTED)
.build()
val inputData = workDataOf(
KEY_CHAT_ID to chatId,
KEY_CURSOR to cursor
)
return OneTimeWorkRequestBuilder<ChatSyncWorker>()
.setConstraints(constraints)
.setInputData(inputData)
.addTag(WORK_TAG)
.build()
}
}
override suspend fun doWork(): Result = withContext(Dispatchers.IO) {
Log.d("ChatSyncWorker", "Starting chat sync")
try {
// 1. Синхронизируем список чатов
syncChats()
// 2. Если указан chatId - синхронизируем сообщения
val chatId = inputData.getString(KEY_CHAT_ID)
if (chatId != null) {
syncMessagesForChat(chatId)
}
Log.i("ChatSyncWorker", "Sync completed successfully")
Result.success()
} catch (e: IOException) {
Log.w("ChatSyncWorker", "Network error during sync: ${e.message}")
Result.retry()
} catch (e: SocketTimeoutException) {
Log.w("ChatSyncWorker", "Timeout during sync: ${e.message}")
Result.retry()
} catch (e: Exception) {
Log.e("ChatSyncWorker", "Error during sync", e)
Result.failure()
}
}
private suspend fun syncChats() {
val remoteChats = chatApi.getChats()
val localChats = mutableListOf<chats.data.local.database.ChatEntity>()
remoteChats.forEach { dto ->
val existingChat = chatDao.getChatByRemoteId(dto.id)
val chatEntity = chats.data.local.database.ChatEntity(
localId = existingChat?.localId ?: java.util.UUID.randomUUID().toString(),
remoteId = dto.id,
type = dto.type,
name = dto.name ?: "",
avatar = dto.avatar,
unreadCount = dto.unreadCount,
lastMessageText = dto.messages.firstOrNull()?.content,
lastMessageTimestamp = dto.messages.firstOrNull()?.createdAt
)
localChats.add(chatEntity)
}
chatDao.insertChats(localChats)
Log.d("ChatSyncWorker", "Synced ${localChats.size} chats")
}
private suspend fun syncMessagesForChat(chatId: String) {
val lastMessage = messageDao.getMessagesByStatus(
chatId = chatId,
statuses = listOf(MessageStatus.SENT, MessageStatus.DELIVERED, MessageStatus.READ)
).maxByOrNull { it.sequenceId }
val cursor = lastMessage?.serverId
val messages = chatApi.getMessages(chatId, cursor = cursor, limit = 50)
if (messages.isNotEmpty()) {
saveMessagesToLocal(messages, chatId)
Log.d("ChatSyncWorker", "Synced ${messages.size} messages for chat $chatId")
}
}
private suspend fun saveMessagesToLocal(
messages: List<chats.data.remote.dto.MessageDto>,
chatId: String
) {
val currentUserId = tokenManager.getUserId() ?: ""
val gson = com.google.gson.Gson()
messages.forEach { dto ->
val existing = messageDao.getMessageByServerId(dto.id)
if (existing == null) {
val entity = chats.data.local.database.MessageEntity(
localId = java.util.UUID.randomUUID().toString(),
serverId = dto.id,
idempotencyKey = dto.id,
chatId = chatId,
senderId = dto.senderId ?: dto.sender?.id ?: currentUserId,
senderName = dto.sender?.displayName ?: dto.sender?.username ?: "",
senderAvatar = dto.sender?.avatarUrl,
content = dto.content,
sequenceId = dto.sequenceId?.toLong() ?: 0L,
createdAt = dto.createdAt ?: java.time.ZonedDateTime.now().toString(),
mediaType = dto.type ?: "TEXT",
mediaJson = gson.toJson(dto.media.map {
chats.domain.model.Media(
id = it.id,
type = it.type,
url = it.url,
filename = it.filename,
size = it.size,
duration = it.duration
)
}),
reactionsJson = gson.toJson(
dto.reactions?.associate { it.emoji to it.count } ?: emptyMap<String, Int>()
),
status = if (dto.senderId == currentUserId) MessageStatus.SENT else MessageStatus.DELIVERED,
createdAtMillis = System.currentTimeMillis(),
updatedAtMillis = System.currentTimeMillis()
)
messageDao.insertMessage(entity)
}
}
}
}

150
client-mobile/chats/data/workers/SendMessageWorker.kt Normal file
View File

@@ -0,0 +1,150 @@
package chats.data.workers
import android.content.Context
import android.util.Log
import androidx.hilt.work.HiltWorker
import androidx.work.*
import chats.data.local.dao.MessageDao
import chats.data.local.database.MessageEntity
import chats.data.remote.api.ChatApi
import chats.data.remote.api.SendMessageRequest
import chats.domain.model.MessageStatus
import dagger.assisted.Assisted
import dagger.assisted.AssistedInject
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import java.io.IOException
import java.net.SocketTimeoutException
import java.util.concurrent.TimeUnit
/**
* Worker для отправки сообщений с поддержкой офлайн-режима.
* Использует WorkManager для гарантированной доставки при появлении сети.
*/
@HiltWorker
class SendMessageWorker @AssistedInject constructor(
@Assisted context: Context,
@Assisted params: WorkerParameters,
private val messageDao: MessageDao,
private val chatApi: ChatApi
) : CoroutineWorker(context, params) {
companion object {
const val WORK_TAG = "send_message_worker"
const val KEY_LOCAL_ID = "local_id"
const val MAX_RETRY_COUNT = 5
/**
* Создает запрос на отправку сообщения через WorkManager
*/
fun createWorkRequest(localId: String): OneTimeWorkRequest {
val constraints = Constraints.Builder()
.setRequiredNetworkType(NetworkType.CONNECTED)
.build()
val inputData = workDataOf(KEY_LOCAL_ID to localId)
return OneTimeWorkRequestBuilder<SendMessageWorker>()
.setConstraints(constraints)
.setInputData(inputData)
.addTag(WORK_TAG)
.addTag("$WORK_TAG:$localId")
.setBackoffCriteria(
BackoffPolicy.EXPONENTIAL,
WorkRequest.MIN_BACKOFF_MILLIS,
TimeUnit.MILLISECONDS
)
.build()
}
}
override suspend fun doWork(): Result = withContext(Dispatchers.IO) {
val localId = inputData.getString(KEY_LOCAL_ID)
?: return@withContext Result.failure()
Log.d("SendMessageWorker", "Starting work for message: $localId")
try {
// Получаем сообщение из локальной БД
val message = messageDao.getMessageByLocalId(localId)
?: return@withContext Result.failure(
workDataOf("error" to "Message not found: $localId")
)
// Проверяем, не отправлено ли уже
if (message.status == MessageStatus.SENT || message.status == MessageStatus.DELIVERED || message.status == MessageStatus.READ) {
Log.d("SendMessageWorker", "Message already sent: $localId")
return@withContext Result.success()
}
// Проверяем лимит повторных попыток
if (message.retryCount >= MAX_RETRY_COUNT) {
Log.w("SendMessageWorker", "Max retries reached for message: $localId")
messageDao.updateMessageError(
localId = localId,
status = MessageStatus.FAILED,
errorMessage = "Max retry count exceeded",
updatedAtMillis = System.currentTimeMillis()
)
return@withContext Result.failure(
workDataOf("error" to "Max retries reached")
)
}
// Обновляем статус на SENDING
messageDao.updateMessageStatus(
localId = localId,
status = MessageStatus.SENDING,
updatedAtMillis = System.currentTimeMillis()
)
// Выполняем запрос к API с Idempotency-Key
val request = SendMessageRequest(
content = message.content,
type = message.mediaType.lowercase(),
replyToId = message.replyToServerId
)
val serverId = chatApi.sendMessage(
chatId = message.chatId,
request = request,
idempotencyKey = message.idempotencyKey
)
// Успешная отправка - обновляем статус и serverId
messageDao.updateMessageWithServerId(
localId = localId,
serverId = serverId,
status = MessageStatus.SENT,
updatedAtMillis = System.currentTimeMillis()
)
Log.i("SendMessageWorker", "Message sent successfully: local=$localId, server=$serverId")
Result.success()
} catch (e: IOException) {
// Сетевая ошибка - повторяем позже
Log.w("SendMessageWorker", "Network error for message $localId: ${e.message}")
Result.retry()
} catch (e: SocketTimeoutException) {
Log.w("SendMessageWorker", "Timeout for message $localId: ${e.message}")
Result.retry()
} catch (e: Exception) {
// Остальные ошибки - помечаем как FAILED
Log.e("SendMessageWorker", "Error sending message $localId", e)
try {
messageDao.updateMessageError(
localId = localId,
status = MessageStatus.FAILED,
errorMessage = e.message ?: "Unknown error",
updatedAtMillis = System.currentTimeMillis()
)
} catch (dbError: Exception) {
Log.e("SendMessageWorker", "Failed to update error status", dbError)
}
Result.failure(
workDataOf("error" to (e.message ?: "Unknown error"))
)
}
}
}