diff --git a/.ps1 b/.ps1 new file mode 100644 index 0000000..e69de29 diff --git a/backend/src/Host/Host.csproj b/backend/src/Host/Host.csproj index 746b68c..a6740d5 100644 --- a/backend/src/Host/Host.csproj +++ b/backend/src/Host/Host.csproj @@ -24,6 +24,7 @@ + diff --git a/backend/src/Host/Program.cs b/backend/src/Host/Program.cs index e54cf69..434d696 100644 --- a/backend/src/Host/Program.cs +++ b/backend/src/Host/Program.cs @@ -141,6 +141,7 @@ builder.Services.AddRouting(options => builder.Services.AddMemoryCache(); builder.Services.AddHttpClient(); +builder.Services.AddIdempotency(); // DKNet.AspCore.Idempotency DI registration builder.Services.ConfigureHttpJsonOptions(options => { @@ -243,7 +244,7 @@ if (app.Environment.IsDevelopment()) // Не раздаем статические файлы, так как теперь используем MinIO app.UseAuthentication(); -app.UseIdempotencyMiddleware(); +app.UseIdempotency(); // DKNet.AspCore.Idempotency middleware app.UseAuthorization(); // Регистрация эндпоинтов diff --git a/backend/src/Modules/Conversations/Application/Messages/Send/SendMessageCommandHandler.cs b/backend/src/Modules/Conversations/Application/Messages/Send/SendMessageCommandHandler.cs index 6fa462f..2026c44 100644 --- a/backend/src/Modules/Conversations/Application/Messages/Send/SendMessageCommandHandler.cs +++ b/backend/src/Modules/Conversations/Application/Messages/Send/SendMessageCommandHandler.cs @@ -3,7 +3,6 @@ using Knot.Contracts.Conversations.Domain; using Knot.Contracts.Messaging.Application.Abstractions; using Knot.Contracts.Messaging.Domain; using Knot.Contracts.Settings.Application.Abstractions; -using Knot.Modules.Conversations.Domain; using Knot.Shared.Kernel; using Microsoft.Extensions.Logging; @@ -42,7 +41,6 @@ public sealed class SendMessageCommandHandler : ICommandHandler _logger; public SendMessageCommandHandler( @@ -51,7 +49,6 @@ public sealed class SendMessageCommandHandler : ICommandHandler logger) { _chatRepository = chatRepository; @@ -59,23 +56,11 @@ public sealed class SendMessageCommandHandler : ICommandHandler> Handle(SendMessageCommand request, CancellationToken cancellationToken) { - // 0. Проверка идемпотентности - if (!string.IsNullOrWhiteSpace(request.IdempotencyKey)) - { - var existingMessageId = await _idempotencyRepository.GetProcessedMessageIdAsync(request.IdempotencyKey, cancellationToken); - if (existingMessageId.HasValue) - { - _logger.LogInformation("Idempotency key already processed: {Key}, returning existing message: {MessageId}", request.IdempotencyKey, existingMessageId.Value); - return Result.Success(existingMessageId.Value); - } - } - // 1. Проверка существования чата var chat = await _chatRepository.GetByIdAsync(request.ChatId, cancellationToken); if (chat is null) @@ -217,12 +202,6 @@ public sealed class SendMessageCommandHandler : ICommandHandler(); services.AddScoped(); - // Idempotency support - services.AddScoped(); - services.AddMediatR(config => config.RegisterServicesFromAssembly(typeof(DependencyInjection).Assembly)); diff --git a/backend/src/Modules/Conversations/Domain/IIdempotencyKeyRepository.cs b/backend/src/Modules/Conversations/Domain/IIdempotencyKeyRepository.cs deleted file mode 100644 index b2e2a58..0000000 --- a/backend/src/Modules/Conversations/Domain/IIdempotencyKeyRepository.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System.Threading; -using System.Threading.Tasks; - -namespace Knot.Modules.Conversations.Domain; - -/// -/// Репозиторий для управления idempotency ключами -/// -public interface IIdempotencyKeyRepository -{ - /// - /// Проверяет наличие ключа и возвращает сообщение если оно уже было обработано - /// - Task GetProcessedMessageIdAsync(string key, CancellationToken cancellationToken = default); - - /// - /// Сохраняет idempotency ключ и связывает его с сообщением - /// - Task SaveKeyAsync(string key, Guid messageId, Guid chatId, CancellationToken cancellationToken = default); - - /// - /// Проверяет и сохраняет атомарно (избегаем race condition) - /// Возвращает: MessageId если ключ уже был, null если ключ был сохранен успешно - /// - Task GetOrSaveKeyAsync(string key, Guid messageId, Guid chatId, CancellationToken cancellationToken = default); -} diff --git a/backend/src/Modules/Conversations/Domain/IdempotencyKeyRecord.cs b/backend/src/Modules/Conversations/Domain/IdempotencyKeyRecord.cs deleted file mode 100644 index c551848..0000000 --- a/backend/src/Modules/Conversations/Domain/IdempotencyKeyRecord.cs +++ /dev/null @@ -1,57 +0,0 @@ -using System; -using MongoDB.Bson.Serialization.Attributes; - -namespace Knot.Modules.Conversations.Domain; - -/// -/// Запись об обработанном idempotency ключе для предотвращения дубликатов сообщений. -/// Используется для обеспечения идемпотентности при повторной отправке сообщений офлайн. -/// -public class IdempotencyKeyRecord -{ - [BsonId] - [BsonRepresentation(MongoDB.Bson.BsonType.String)] - public string Id { get; set; } = string.Empty; - - /// - /// Idempotency ключ из заголовка запроса - /// - [BsonElement("key")] - public string Key { get; set; } = string.Empty; - - /// - /// ID созданного сообщения - /// - [BsonElement("messageId")] - public Guid MessageId { get; set; } - - /// - /// ID чата - /// - [BsonElement("chatId")] - public Guid ChatId { get; set; } - - /// - /// Время создания записи - /// - [BsonElement("createdAt")] - public DateTime CreatedAt { get; set; } = DateTime.UtcNow; - - /// - /// Время истечения записи (через 24 часа для очистки устаревших ключей) - /// - [BsonElement("expiresAt")] - public DateTime ExpiresAt { get; set; } - - public IdempotencyKeyRecord() { } - - public IdempotencyKeyRecord(string key, Guid messageId, Guid chatId) - { - Id = Guid.NewGuid().ToString(); - Key = key; - MessageId = messageId; - ChatId = chatId; - CreatedAt = DateTime.UtcNow; - ExpiresAt = DateTime.UtcNow.AddHours(24); - } -} diff --git a/backend/src/Modules/Conversations/Domain/IdempotencyKeyRepository.cs b/backend/src/Modules/Conversations/Domain/IdempotencyKeyRepository.cs deleted file mode 100644 index f71c6cf..0000000 --- a/backend/src/Modules/Conversations/Domain/IdempotencyKeyRepository.cs +++ /dev/null @@ -1,112 +0,0 @@ -using Microsoft.Extensions.Logging; -using MongoDB.Driver; -using MongoDB.Bson; - -namespace Knot.Modules.Conversations.Domain; - -/// -/// Реализация репозитория idempotency ключей на основе MongoDB -/// -public class IdempotencyKeyRepository : IIdempotencyKeyRepository -{ - private readonly IMongoCollection _collection; - private readonly ILogger _logger; - - public IdempotencyKeyRepository(IMongoClient mongoClient, ILogger logger) - { - var database = mongoClient.GetDatabase("KnotDb"); - _collection = database.GetCollection("idempotency_keys"); - _logger = logger; - - // Создаем индекс по ключу для быстрого поиска - CreateIndexes(); - } - - private void CreateIndexes() - { - var keyIndexModel = new CreateIndexModel( - Builders.IndexKeys.Ascending(x => x.Key), - new CreateIndexOptions { Unique = true } - ); - - var expireIndexModel = new CreateIndexModel( - Builders.IndexKeys.Ascending(x => x.ExpiresAt), - new CreateIndexOptions { ExpireAfter = TimeSpan.Zero } // TTL индекс - ); - - _collection.Indexes.CreateOne(keyIndexModel); - _collection.Indexes.CreateOne(expireIndexModel); - } - - public async Task GetProcessedMessageIdAsync(string key, CancellationToken cancellationToken = default) - { - try - { - var record = await _collection - .Find(x => x.Key == key) - .FirstOrDefaultAsync(cancellationToken); - - return record?.MessageId; - } - catch (Exception ex) - { - _logger.LogError(ex, "Error getting processed message id for key: {Key}", key); - return null; - } - } - - public async Task SaveKeyAsync(string key, Guid messageId, Guid chatId, CancellationToken cancellationToken = default) - { - try - { - var record = new IdempotencyKeyRecord(key, messageId, chatId); - await _collection.InsertOneAsync(record, cancellationToken: cancellationToken); - _logger.LogDebug("Saved idempotency key: {Key} for message: {MessageId}", key, messageId); - } - catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey) - { - // Ключ уже существует - это нормально, игнорируем - _logger.LogDebug("Idempotency key already exists: {Key}", key); - } - catch (Exception ex) - { - _logger.LogError(ex, "Error saving idempotency key: {Key}", key); - throw; - } - } - - public async Task GetOrSaveKeyAsync(string key, Guid messageId, Guid chatId, CancellationToken cancellationToken = default) - { - try - { - // Пробуем найти существующий ключ - var existingRecord = await _collection - .Find(x => x.Key == key) - .FirstOrDefaultAsync(cancellationToken); - - if (existingRecord != null) - { - _logger.LogDebug("Idempotency key already processed: {Key}, returning existing message: {MessageId}", key, existingRecord.MessageId); - return existingRecord.MessageId; - } - - // Пробуем вставить новую запись - var record = new IdempotencyKeyRecord(key, messageId, chatId); - await _collection.InsertOneAsync(record, cancellationToken: cancellationToken); - _logger.LogDebug("Saved new idempotency key: {Key} for message: {MessageId}", key, messageId); - return null; - } - catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey) - { - // Race condition: другой запрос успел сохранить ключ - // Повторяем поиск - _logger.LogDebug("Race condition on idempotency key: {Key}, retrying", key); - return await GetProcessedMessageIdAsync(key, cancellationToken); - } - catch (Exception ex) - { - _logger.LogError(ex, "Error in GetOrSaveKey for: {Key}", key); - throw; - } - } -} diff --git a/backend/src/Modules/Conversations/Knot.Modules.Conversations.csproj b/backend/src/Modules/Conversations/Knot.Modules.Conversations.csproj index e2be4c1..6c6891e 100644 --- a/backend/src/Modules/Conversations/Knot.Modules.Conversations.csproj +++ b/backend/src/Modules/Conversations/Knot.Modules.Conversations.csproj @@ -29,6 +29,7 @@ + diff --git a/backend/src/Modules/Conversations/Presentation/Middleware/IdempotencyMiddleware.cs b/backend/src/Modules/Conversations/Presentation/Middleware/IdempotencyMiddleware.cs deleted file mode 100644 index b926c82..0000000 --- a/backend/src/Modules/Conversations/Presentation/Middleware/IdempotencyMiddleware.cs +++ /dev/null @@ -1,69 +0,0 @@ -using System.Net; -using System.Text.Json; -using Microsoft.AspNetCore.Builder; -using Microsoft.AspNetCore.Http; -using Microsoft.Extensions.Logging; -using Knot.Modules.Conversations.Domain; -using Knot.Shared.Kernel; - -namespace Knot.Modules.Conversations.Presentation.Middleware; - -/// -/// Middleware для проверки идемпотентности POST-запросов к сообщениям -/// -public class IdempotencyMiddleware -{ - private readonly RequestDelegate _next; - private readonly ILogger _logger; - - public IdempotencyMiddleware(RequestDelegate next, ILogger logger) - { - _next = next; - _logger = logger; - } - - public async Task InvokeAsync(HttpContext context, IIdempotencyKeyRepository idempotencyRepository) - { - // Обрабатываем только POST запросы к /api/messages/chat/ - if (context.Request.Method == HttpMethods.Post && - context.Request.Path.StartsWithSegments("/api/messages/chat/")) - { - if (context.Request.Headers.TryGetValue("X-Idempotency-Key", out var idempotencyKey)) - { - var key = idempotencyKey.ToString().Trim(); - - if (!string.IsNullOrWhiteSpace(key)) - { - _logger.LogDebug("Processing idempotency key: {Key}", key); - - // Проверяем, был ли уже обработан этот ключ - var existingMessageId = await idempotencyRepository.GetProcessedMessageIdAsync(key); - - if (existingMessageId.HasValue) - { - _logger.LogInformation("Returning cached response for idempotency key: {Key}, MessageId: {MessageId}", key, existingMessageId.Value); - - // Возвращаем успешный ответ с ID существующего сообщения - context.Response.StatusCode = (int)HttpStatusCode.OK; - context.Response.ContentType = "application/json"; - - var response = JsonSerializer.Serialize(new { id = existingMessageId.Value.ToString() }); - await context.Response.WriteAsync(response); - return; - } - } - } - } - - await _next(context); - } -} - -// Extension method для упрощения использования -public static class IdempotencyMiddlewareExtensions -{ - public static IApplicationBuilder UseIdempotencyMiddleware(this IApplicationBuilder builder) - { - return builder.UseMiddleware(); - } -} diff --git a/client-mobile/app/build.gradle.kts b/client-mobile/app/build.gradle.kts index d936692..49279eb 100644 --- a/client-mobile/app/build.gradle.kts +++ b/client-mobile/app/build.gradle.kts @@ -135,7 +135,7 @@ dependencies { implementation("androidx.work:work-runtime-ktx:$work_version") // Paging 3 - val paging_version = "3.2.1" + val paging_version = "3.3.0" implementation("androidx.paging:paging-runtime-ktx:$paging_version") implementation("androidx.paging:paging-compose:$paging_version") diff --git a/client-mobile/chats/data/local/dao/ChatDao.kt b/client-mobile/chats/data/local/dao/ChatDao.kt index 809b970..b28b825 100644 --- a/client-mobile/chats/data/local/dao/ChatDao.kt +++ b/client-mobile/chats/data/local/dao/ChatDao.kt @@ -10,7 +10,7 @@ import kotlinx.coroutines.flow.Flow @Dao interface ChatDao { - @Query("SELECT * FROM chats ORDER BY updatedAtMillis DESC") + @Query("SELECT * FROM chats ORDER BY lastMessageTimestamp DESC") fun getAllChats(): Flow> @Query("SELECT * FROM chats WHERE remoteId = :remoteId LIMIT 1") @@ -29,7 +29,7 @@ interface ChatDao { 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?) + suspend fun updateLastMessage(chatId: String, lastMessageText: String?, timestamp: Long?) @Query("UPDATE chats SET unreadCount = :count WHERE remoteId = :chatId") suspend fun updateUnreadCount(chatId: String, count: Int) diff --git a/client-mobile/chats/data/local/dao/MessageDao.kt b/client-mobile/chats/data/local/dao/MessageDao.kt index e53e6e0..91beb19 100644 --- a/client-mobile/chats/data/local/dao/MessageDao.kt +++ b/client-mobile/chats/data/local/dao/MessageDao.kt @@ -14,6 +14,19 @@ interface MessageDao { @Query("SELECT * FROM messages WHERE chatId = :chatId ORDER BY sequenceId ASC") fun getMessagesByChatId(chatId: String): Flow> + /** + * Пагинированная загрузка сообщений для Paging 3 + * sequenceId увеличивается от старых к новым, поэтому ORDER BY DESC для отображения newest внизу + */ + @Query("SELECT * FROM messages WHERE chatId = :chatId ORDER BY sequenceId DESC LIMIT :limit OFFSET :offset") + suspend fun getMessagesPaged(chatId: String, offset: Int, limit: Int): List + + @Query("SELECT * FROM messages WHERE chatId = :chatId ORDER BY sequenceId DESC LIMIT 1") + suspend fun getLastMessage(chatId: String): MessageEntity? + + @Query("SELECT * FROM messages WHERE chatId = :chatId ORDER BY sequenceId ASC LIMIT 1") + suspend fun getFirstMessage(chatId: String): 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 diff --git a/client-mobile/chats/data/local/database/ChatEntity.kt b/client-mobile/chats/data/local/database/ChatEntity.kt index 12291f0..e459c67 100644 --- a/client-mobile/chats/data/local/database/ChatEntity.kt +++ b/client-mobile/chats/data/local/database/ChatEntity.kt @@ -31,7 +31,7 @@ data class ChatEntity( val lastMessageText: String? = null, - val lastMessageTimestamp: String? = null, + val lastMessageTimestamp: Long? = null, val updatedAtMillis: Long = System.currentTimeMillis() ) diff --git a/client-mobile/chats/data/local/mappers/ChatMappers.kt b/client-mobile/chats/data/local/mappers/ChatMappers.kt index 839f348..28720ad 100644 --- a/client-mobile/chats/data/local/mappers/ChatMappers.kt +++ b/client-mobile/chats/data/local/mappers/ChatMappers.kt @@ -7,6 +7,7 @@ import chats.domain.model.Chat * Преобразует Domain Chat в Entity */ fun Chat.toEntity(): ChatEntity { + val timestamp = this.lastMessage?.createdAt?.let { parseTimestamp(it) } return ChatEntity( localId = this.id.takeIf { it.isNotBlank() } ?: java.util.UUID.randomUUID().toString(), remoteId = this.id, @@ -15,7 +16,7 @@ fun Chat.toEntity(): ChatEntity { avatar = this.avatar, unreadCount = this.unreadCount, lastMessageText = this.lastMessage?.content, - lastMessageTimestamp = this.lastMessage?.createdAt + lastMessageTimestamp = timestamp ) } @@ -32,3 +33,14 @@ fun ChatEntity.toDomain(): Chat { lastMessage = null // lastMessage загружается отдельно ) } + +/** + * Парсит ISO-8601 timestamp в Long (millis) + */ +private fun parseTimestamp(isoTimestamp: String): Long? { + return try { + java.time.ZonedDateTime.parse(isoTimestamp).toInstant().toEpochMilli() + } catch (e: Exception) { + null + } +} diff --git a/client-mobile/chats/data/local/paging/MessagePagingSource.kt b/client-mobile/chats/data/local/paging/MessagePagingSource.kt new file mode 100644 index 0000000..c39fcb4 --- /dev/null +++ b/client-mobile/chats/data/local/paging/MessagePagingSource.kt @@ -0,0 +1,43 @@ +package chats.data.local.paging + +import androidx.paging.PagingSource +import androidx.paging.PagingState +import chats.data.local.dao.MessageDao +import chats.data.local.database.MessageEntity + +/** + * PagingSource для загрузки сообщений из Room Database + */ +class MessagePagingSource( + private val chatId: String, + private val messageDao: MessageDao +) : PagingSource() { + + companion object { + private const val PAGE_SIZE = 30 + } + + override suspend fun load(params: LoadParams): LoadResult { + return try { + val position = params.key ?: 0 // Начинаем с 0 + + val messages = messageDao.getMessagesPaged( + chatId = chatId, + offset = position, + limit = PAGE_SIZE + ) + + LoadResult.Page( + data = messages, + prevKey = if (position > 0) position - PAGE_SIZE else null, + nextKey = if (messages.isEmpty()) null else position + PAGE_SIZE + ) + } catch (e: Exception) { + LoadResult.Error(e) + } + } + + override fun getRefreshKey(state: PagingState): Int? { + return state.anchorPosition + } +} diff --git a/client-mobile/chats/data/local/paging/MessageRemoteMediator.kt b/client-mobile/chats/data/local/paging/MessageRemoteMediator.kt new file mode 100644 index 0000000..a2820ef --- /dev/null +++ b/client-mobile/chats/data/local/paging/MessageRemoteMediator.kt @@ -0,0 +1,101 @@ +package chats.data.local.paging + +import androidx.paging.ExperimentalPagingApi +import androidx.paging.LoadType +import androidx.paging.PagingState +import androidx.paging.RemoteMediator +import androidx.room.withTransaction +import chats.data.local.dao.ChatDao +import chats.data.local.dao.MessageDao +import chats.data.local.database.AppDatabase +import chats.data.local.database.MessageEntity +import chats.data.local.mappers.toEntity +import chats.data.repository.toDomain as dtoToDomain +import chats.data.remote.api.ChatApi +import chats.domain.model.MessageStatus +import core.security.TokenManager + +/** + * RemoteMediator для синхронизации сообщений с сервером. + * Работает с Room через withTransaction для атомарности. + */ +@OptIn(ExperimentalPagingApi::class) +class MessageRemoteMediator( + private val chatId: String, + private val messageDao: MessageDao, + private val chatDao: ChatDao, + private val chatApi: ChatApi, + private val tokenManager: TokenManager, + private val appDatabase: AppDatabase +) : RemoteMediator() { + + companion object { + private const val PAGE_SIZE = 30 + } + + override suspend fun load( + loadType: LoadType, + state: PagingState + ): RemoteMediator.MediatorResult { + return try { + val currentUserId = tokenManager.getUserId() ?: "" + val baseUrl = "https://your-server.com" + + val cursor = when (loadType) { + LoadType.REFRESH -> { + // При refresh загружаем новые сообщения после последнего локального + messageDao.getLastMessage(chatId)?.serverId + } + LoadType.PREPEND -> { + // Загружаем более старые сообщения перед первым локальным + messageDao.getFirstMessage(chatId)?.serverId + } + LoadType.APPEND -> { + // Загружаем новые сообщения после последнего локального + messageDao.getLastMessage(chatId)?.serverId + } + } + + val remoteMessages = chatApi.getMessages(chatId, cursor = cursor, limit = PAGE_SIZE) + val endOfPaginationReached = remoteMessages.size < PAGE_SIZE + + appDatabase.withTransaction { + if (loadType == LoadType.REFRESH) { + // При полном обновлении можно очистить старые SENT/DELIVERED/READ, + // но оставить локальные PENDING/FAILED + // messageDao.deleteMessagesByChatId(chatId) // Опционально + } + + val entities = remoteMessages.map { dto -> + val existing = messageDao.getMessageByServerId(dto.id) + if (existing != null) { + // Обновляем существующее, сохраняя localId + existing.copy( + content = dto.content, + reactionsJson = com.google.gson.Gson().toJson( + dto.reactions?.associate { it.emoji to it.count } ?: emptyMap() + ), + status = if (dto.senderId == currentUserId) MessageStatus.SENT else MessageStatus.DELIVERED, + updatedAtMillis = System.currentTimeMillis() + ) + } else { + // Создаём новое + dto.dtoToDomain(currentUserId, baseUrl).toEntity(MessageStatus.SENT).copy( + localId = java.util.UUID.randomUUID().toString(), + serverId = dto.id, + idempotencyKey = dto.id + ) + } + } + + messageDao.insertMessages(entities) + } + + RemoteMediator.MediatorResult.Success(endOfPaginationReached = endOfPaginationReached) + } catch (e: Exception) { + RemoteMediator.MediatorResult.Error(e) + } + } + + override suspend fun initialize(): InitializeAction = InitializeAction.SKIP_INITIAL_REFRESH +} diff --git a/client-mobile/chats/data/repository/ChatRepositoryImpl.kt b/client-mobile/chats/data/repository/ChatRepositoryImpl.kt index 0dd0618..39557fe 100644 --- a/client-mobile/chats/data/repository/ChatRepositoryImpl.kt +++ b/client-mobile/chats/data/repository/ChatRepositoryImpl.kt @@ -1,6 +1,11 @@ package chats.data.repository import android.util.Log +import androidx.paging.ExperimentalPagingApi +import androidx.paging.Pager +import androidx.paging.PagingConfig +import androidx.paging.PagingData +import androidx.paging.map import androidx.work.ExistingPeriodicWorkPolicy import androidx.work.ExistingWorkPolicy import androidx.work.WorkManager @@ -10,6 +15,8 @@ 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.local.paging.MessagePagingSource +import chats.data.local.paging.MessageRemoteMediator import chats.data.remote.api.ChatApi import chats.data.remote.api.SendMessageRequest import chats.data.remote.dto.MessageDto @@ -39,6 +46,7 @@ class ChatRepositoryImpl @Inject constructor( private val messageDao: MessageDao, private val chatDao: ChatDao, private val workManager: WorkManager, + private val appDatabase: chats.data.local.database.AppDatabase, private val hubClient: chats.data.remote.signalr.ChatHubClient ) : ChatRepository { @@ -101,6 +109,34 @@ class ChatRepositoryImpl @Inject constructor( } } + /** + * Получает сообщения с помощью Paging 3 и RemoteMediator + */ + @OptIn(ExperimentalPagingApi::class) + override fun getMessagesPagingSource(chatId: String): Flow> { + return Pager( + config = PagingConfig( + pageSize = 30, + prefetchDistance = 10, + enablePlaceholders = false, + initialLoadSize = 60 + ), + pagingSourceFactory = { + MessagePagingSource(chatId, messageDao) + }, + remoteMediator = MessageRemoteMediator( + chatId = chatId, + messageDao = messageDao, + chatDao = chatDao, + chatApi = api, + tokenManager = tokenManager, + appDatabase = appDatabase + ) + ).flow.map { pagingData -> + pagingData.map { entity -> entity.toDomain() } + } + } + override suspend fun getMessages(chatId: String, cursor: String?, pivot: Long?, limit: Int?): List { val baseUrl = serverConfig.getBaseUrl().removeSuffix("/api/") val currentUserId = tokenManager.getUserId() ?: "" diff --git a/client-mobile/chats/data/workers/ChatSyncWorker.kt b/client-mobile/chats/data/workers/ChatSyncWorker.kt index 6ed9866..27161da 100644 --- a/client-mobile/chats/data/workers/ChatSyncWorker.kt +++ b/client-mobile/chats/data/workers/ChatSyncWorker.kt @@ -26,7 +26,8 @@ class ChatSyncWorker @AssistedInject constructor( @Assisted params: WorkerParameters, private val chatDao: ChatDao, private val messageDao: MessageDao, - private val chatApi: chats.data.remote.api.ChatApi + private val chatApi: chats.data.remote.api.ChatApi, + private val tokenManager: TokenManager ) : CoroutineWorker(context, params) { companion object { @@ -106,6 +107,11 @@ class ChatSyncWorker @AssistedInject constructor( remoteChats.forEach { dto -> val existingChat = chatDao.getChatByRemoteId(dto.id) + val lastMessage = dto.messages.firstOrNull() + val timestamp = lastMessage?.createdAt?.let { + try { java.time.ZonedDateTime.parse(it).toInstant().toEpochMilli() } + catch (e: Exception) { null } + } val chatEntity = chats.data.local.database.ChatEntity( localId = existingChat?.localId ?: java.util.UUID.randomUUID().toString(), remoteId = dto.id, @@ -113,8 +119,8 @@ class ChatSyncWorker @AssistedInject constructor( name = dto.name ?: "", avatar = dto.avatar, unreadCount = dto.unreadCount, - lastMessageText = dto.messages.firstOrNull()?.content, - lastMessageTimestamp = dto.messages.firstOrNull()?.createdAt + lastMessageText = lastMessage?.content, + lastMessageTimestamp = timestamp ) localChats.add(chatEntity) } diff --git a/client-mobile/chats/domain/repository/ChatRepository.kt b/client-mobile/chats/domain/repository/ChatRepository.kt index 9e91b3d..43715f0 100644 --- a/client-mobile/chats/domain/repository/ChatRepository.kt +++ b/client-mobile/chats/domain/repository/ChatRepository.kt @@ -1,8 +1,8 @@ package chats.domain.repository +import androidx.paging.PagingData import chats.domain.model.Chat import chats.domain.model.Message - import kotlinx.coroutines.flow.Flow interface ChatRepository { @@ -13,6 +13,7 @@ interface ChatRepository { // Сообщения - Single Source of Truth через Flow fun getMessagesFlow(chatId: String): Flow> + fun getMessagesPagingSource(chatId: String): Flow> suspend fun getMessages(chatId: String, cursor: String? = null, pivot: Long? = null, limit: Int? = null): List // Отправка сообщений с поддержкой офлайн diff --git a/client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt b/client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt index 8148e04..068344c 100644 --- a/client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt +++ b/client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt @@ -594,35 +594,12 @@ class ChatDetailViewModel @Inject constructor( val replyToId = _state.value.replyingMessage?.id _state.update { it.copy(replyingMessage = null) } - val tempId = "temp_voice_${System.currentTimeMillis()}" - val userId = getCurrentUserId() - - val tempMessage = Message( - id = tempId, - chatId = chatId, - senderId = userId, - content = null, - createdAt = java.util.Date().toString(), - mediaType = chats.domain.model.MediaType.AUDIO, - media = emptyList(), - senderName = "Вы", - senderAvatar = null, - reactions = emptyMap(), - isRead = false, - sequenceId = 0 - ) - - // Добавляем временное сообщение в стейт для индикации отправки - _state.update { it.copy(messages = listOf(tempMessage) + it.messages) } - viewModelScope.launch { try { android.util.Log.d("ChatDetailVM", "Starting voice upload...") - // 1. Upload the audio file val url = repository.uploadMedia(file) android.util.Log.d("ChatDetailVM", "Voice upload successful, url: $url") - // 2. Send the message with the attachment val attachment = chats.data.remote.api.AttachmentRequest( type = "voice", url = url, @@ -631,34 +608,18 @@ class ChatDetailViewModel @Inject constructor( ) android.util.Log.d("ChatDetailVM", "Sending message with voice attachment...") - val sentMessage = repository.sendMessage( + repository.sendMessage( chatId = chatId, content = null, type = "media", attachments = listOf(attachment), replyToId = replyToId ) - android.util.Log.d("ChatDetailVM", "Voice message sent successfully: ${sentMessage.id}") - - // Заменяем временное сообщение на настоящее - _state.update { currentState -> - val filtered = currentState.messages.filter { it.id != tempId } - // Избегаем дубликатов, если SignalR уже добавил сообщение - if (filtered.any { it.id == sentMessage.id }) { - currentState.copy(messages = filtered) - } else { - currentState.copy(messages = listOf(sentMessage) + filtered) - } - } + // Сообщение автоматически появится в UI через Flow из Room + android.util.Log.d("ChatDetailVM", "Voice message queued successfully") } catch (e: Exception) { android.util.Log.e("ChatDetailVM", "Error sending voice message", e) - // Удаляем временное сообщение и показываем ошибку - _state.update { currentState -> - currentState.copy( - messages = currentState.messages.filter { it.id != tempId }, - error = "Ошибка отправки голосового: ${e.localizedMessage}" - ) - } + _state.update { it.copy(error = "Ошибка отправки голосового: ${e.localizedMessage}") } } } } @@ -776,28 +737,6 @@ class ChatDetailViewModel @Inject constructor( val replyToId = _state.value.replyingMessage?.id _state.update { it.copy(replyingMessage = null) } - val tempId = "temp_gif_${System.currentTimeMillis()}" - val userId = getCurrentUserId() - - val tempMessage = Message( - id = tempId, - chatId = chatId, - senderId = userId, - content = url, - createdAt = java.util.Date().toString(), - mediaType = chats.domain.model.MediaType.IMAGE, // We map GIF to IMAGE for rendering - media = listOf(chats.domain.model.Media(url = url, type = "image", id = "temp_media")), - senderName = "Вы", - senderAvatar = null, - reactions = emptyMap(), - isRead = false, - sequenceId = 0 - ) - - viewModelScope.launch { - repository.saveMessage(tempMessage) - } - viewModelScope.launch { try { val attachment = chats.data.remote.api.AttachmentRequest( @@ -806,10 +745,14 @@ class ChatDetailViewModel @Inject constructor( fileName = "gif.gif", fileSize = 0 ) - val sentMessage = repository.sendMessage(chatId, null, "image", listOf(attachment), replyToId) - - repository.deleteLocalMessage(tempId) - repository.saveMessage(sentMessage) + repository.sendMessage( + chatId = chatId, + content = null, + type = "image", + attachments = listOf(attachment), + replyToId = replyToId + ) + // Сообщение автоматически появится в UI через Flow из Room // Add to recent val allGifs = _state.value.trendingGifs + _state.value.searchedGifs + _state.value.recentGifs @@ -828,7 +771,6 @@ class ChatDetailViewModel @Inject constructor( _state.update { it.copy(recentGifs = newList) } } } catch (e: Exception) { - repository.deleteLocalMessage(tempId) _state.update { it.copy(error = e.localizedMessage) } } }