From 9daa786cfbf5c5ea6659164fa231535800e047fe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A5=D0=B0=D0=BB=D0=B8=D0=BC=D0=BE=D0=B2=20=D0=A0=D1=83?= =?UTF-8?q?=D1=81=D1=82=D0=B0=D0=BC?= Date: Thu, 7 May 2026 23:27:37 +0300 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D0=BE=D1=82=D0=BE=D0=B2?= =?UTF-8?q?=D0=BA=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/src/Host/Program.cs | 4 +- .../Send/SendMessageCommandHandler.cs | 38 ++- .../Conversations/DependencyInjection.cs | 17 +- .../Domain/IIdempotencyKeyRepository.cs | 26 ++ .../Domain/IdempotencyKeyRecord.cs | 57 ++++ .../Domain/IdempotencyKeyRepository.cs | 112 +++++++ .../Endpoints/MessagesEndpoints.cs | 8 +- .../Middleware/IdempotencyMiddleware.cs | 69 ++++ client-mobile/app/build.gradle.kts | 11 + .../com/knot/messenger/MainApplication.kt | 14 +- client-mobile/chats/data/local/dao/ChatDao.kt | 45 +++ .../chats/data/local/dao/MessageDao.kt | 70 ++++ .../chats/data/local/dao/UserProfileDao.kt | 36 ++ .../chats/data/local/database/AppDatabase.kt | 33 ++ .../chats/data/local/database/ChatEntity.kt | 37 +++ .../data/local/database/MessageEntity.kt | 66 ++++ .../local/database/MessageStatusConverter.kt | 17 + .../data/local/database/UserProfileEntity.kt | 27 ++ .../chats/data/local/mappers/ChatMappers.kt | 34 ++ .../data/local/mappers/MessageMappers.kt | 117 +++++++ .../chats/data/remote/api/ChatApi.kt | 6 +- .../data/repository/ChatRepositoryImpl.kt | 307 +++++++++++++----- .../chats/data/workers/ChatSyncWorker.kt | 184 +++++++++++ .../chats/data/workers/SendMessageWorker.kt | 150 +++++++++ client-mobile/chats/di/ChatModule.kt | 9 +- client-mobile/chats/di/LocalDatabaseModule.kt | 50 +++ client-mobile/chats/domain/model/Message.kt | 5 +- .../chats/domain/model/MessageStatus.kt | 30 ++ .../chats/domain/repository/ChatRepository.kt | 23 +- .../chat_detail/ChatDetailViewModel.kt | 128 ++++---- .../chat_list/ChatListViewModel.kt | 57 +++- .../presentation/components/MessageBubble.kt | 45 ++- client-mobile/core/security/TokenManager.kt | 8 + 33 files changed, 1662 insertions(+), 178 deletions(-) create mode 100644 backend/src/Modules/Conversations/Domain/IIdempotencyKeyRepository.cs create mode 100644 backend/src/Modules/Conversations/Domain/IdempotencyKeyRecord.cs create mode 100644 backend/src/Modules/Conversations/Domain/IdempotencyKeyRepository.cs create mode 100644 backend/src/Modules/Conversations/Presentation/Middleware/IdempotencyMiddleware.cs create mode 100644 client-mobile/chats/data/local/dao/ChatDao.kt create mode 100644 client-mobile/chats/data/local/dao/MessageDao.kt create mode 100644 client-mobile/chats/data/local/dao/UserProfileDao.kt create mode 100644 client-mobile/chats/data/local/database/AppDatabase.kt create mode 100644 client-mobile/chats/data/local/database/ChatEntity.kt create mode 100644 client-mobile/chats/data/local/database/MessageEntity.kt create mode 100644 client-mobile/chats/data/local/database/MessageStatusConverter.kt create mode 100644 client-mobile/chats/data/local/database/UserProfileEntity.kt create mode 100644 client-mobile/chats/data/local/mappers/ChatMappers.kt create mode 100644 client-mobile/chats/data/local/mappers/MessageMappers.kt create mode 100644 client-mobile/chats/data/workers/ChatSyncWorker.kt create mode 100644 client-mobile/chats/data/workers/SendMessageWorker.kt create mode 100644 client-mobile/chats/di/LocalDatabaseModule.kt create mode 100644 client-mobile/chats/domain/model/MessageStatus.kt diff --git a/backend/src/Host/Program.cs b/backend/src/Host/Program.cs index 6f1384e..e54cf69 100644 --- a/backend/src/Host/Program.cs +++ b/backend/src/Host/Program.cs @@ -12,6 +12,7 @@ using Knot.Modules.Conversations; using Knot.Modules.Conversations.Infrastructure.Persistence; using Knot.Modules.Conversations.Infrastructure.SignalR; using Knot.Modules.Conversations.Presentation.Endpoints; +using Knot.Modules.Conversations.Presentation.Middleware; using Knot.Modules.Federation; using Knot.Modules.Federation.Presentation.Endpoints; using Knot.Modules.Klipy; @@ -32,11 +33,11 @@ using Knot.Modules.TelegramImport.Presentation.Endpoints; using Knot.Modules.WebRtc; using Knot.Modules.WebRtc.Presentation.Endpoints; using Knot.Shared.Infrastructure; +using MediatR; using Microsoft.AspNetCore.Authentication.JwtBearer; using Microsoft.AspNetCore.SignalR; using Microsoft.EntityFrameworkCore; using Microsoft.IdentityModel.Tokens; -using MediatR; @@ -242,6 +243,7 @@ if (app.Environment.IsDevelopment()) // Не раздаем статические файлы, так как теперь используем MinIO app.UseAuthentication(); +app.UseIdempotencyMiddleware(); 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 ca1c007..6fa462f 100644 --- a/backend/src/Modules/Conversations/Application/Messages/Send/SendMessageCommandHandler.cs +++ b/backend/src/Modules/Conversations/Application/Messages/Send/SendMessageCommandHandler.cs @@ -1,9 +1,11 @@ +using Knot.Contracts.Conversations.Application.Abstractions; +using Knot.Contracts.Conversations.Domain; using Knot.Contracts.Messaging.Application.Abstractions; using Knot.Contracts.Messaging.Domain; using Knot.Contracts.Settings.Application.Abstractions; -using Knot.Contracts.Conversations.Application.Abstractions; -using Knot.Contracts.Conversations.Domain; +using Knot.Modules.Conversations.Domain; using Knot.Shared.Kernel; +using Microsoft.Extensions.Logging; namespace Knot.Modules.Conversations.Application.Messages.Send; @@ -30,7 +32,8 @@ public sealed record SendMessageCommand( DateTime? PollExpiresAt = null, string? CallType = null, string? CallStatus = null, - int? Duration = null) : ICommand; + int? Duration = null, + string? IdempotencyKey = null) : ICommand; public sealed class SendMessageCommandHandler : ICommandHandler { @@ -39,24 +42,41 @@ public sealed class SendMessageCommandHandler : ICommandHandler _logger; public SendMessageCommandHandler( IChatRepository chatRepository, IMessageRepository messageRepository, IChatsUnitOfWork unitOfWork, MediatR.IMediator mediator, - IMessagesSettings messagesSettings) + IMessagesSettings messagesSettings, + IIdempotencyKeyRepository idempotencyRepository, + ILogger logger) { _chatRepository = chatRepository; _messageRepository = messageRepository; _unitOfWork = unitOfWork; _mediator = mediator; _messagesSettings = messagesSettings; + _idempotencyRepository = idempotencyRepository; + _logger = logger; } public async Task> Handle(SendMessageCommand request, CancellationToken cancellationToken) { - // 1. ��������� ������������� ���� + // 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) { @@ -193,10 +213,16 @@ public sealed class SendMessageCommandHandler : ICommandHandler(); services.AddScoped(); + // Idempotency support + services.AddScoped(); + services.AddMediatR(config => config.RegisterServicesFromAssembly(typeof(DependencyInjection).Assembly)); @@ -42,7 +53,7 @@ public static class DependencyInjection services.AddScoped(); services.AddScoped(); services.AddScoped(); - + return services; } } diff --git a/backend/src/Modules/Conversations/Domain/IIdempotencyKeyRepository.cs b/backend/src/Modules/Conversations/Domain/IIdempotencyKeyRepository.cs new file mode 100644 index 0000000..b2e2a58 --- /dev/null +++ b/backend/src/Modules/Conversations/Domain/IIdempotencyKeyRepository.cs @@ -0,0 +1,26 @@ +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 new file mode 100644 index 0000000..c551848 --- /dev/null +++ b/backend/src/Modules/Conversations/Domain/IdempotencyKeyRecord.cs @@ -0,0 +1,57 @@ +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 new file mode 100644 index 0000000..f71c6cf --- /dev/null +++ b/backend/src/Modules/Conversations/Domain/IdempotencyKeyRepository.cs @@ -0,0 +1,112 @@ +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/Presentation/Endpoints/MessagesEndpoints.cs b/backend/src/Modules/Conversations/Presentation/Endpoints/MessagesEndpoints.cs index 6e89b10..29135d8 100644 --- a/backend/src/Modules/Conversations/Presentation/Endpoints/MessagesEndpoints.cs +++ b/backend/src/Modules/Conversations/Presentation/Endpoints/MessagesEndpoints.cs @@ -55,11 +55,14 @@ public static class MessagesEndpoints return result.IsSuccess ? Results.Ok(result.Value) : Results.BadRequest(result.Error.Description); }); - group.MapPost("chat/{chatId:guid}", async ([FromRoute] Guid chatId, [FromBody] SendMessageRequest request, ISender sender, IUserContext userContext, CancellationToken ct) => + group.MapPost("chat/{chatId:guid}", async ([FromRoute] Guid chatId, [FromBody] SendMessageRequest request, ISender sender, IUserContext userContext, HttpRequest httpRequest, CancellationToken ct) => { var attachments = request.Attachments?.Select(a => new AttachmentRequest(a.Type, a.Url, a.FileName, a.FileSize)).ToList(); + // Получаем idempotency ключ из заголовка + httpRequest.Headers.TryGetValue("X-Idempotency-Key", out var idempotencyKey); + var command = new SendMessageCommand( chatId, userContext.UserId, @@ -68,7 +71,8 @@ public static class MessagesEndpoints attachments, request.ReplyToId, request.Quote, - request.ForwardedFromId); + request.ForwardedFromId, + IdempotencyKey: idempotencyKey.ToString()); var result = await sender.Send(command, ct); return result.IsSuccess ? Results.Ok(result.Value) : Results.BadRequest(result.Error.Description); diff --git a/backend/src/Modules/Conversations/Presentation/Middleware/IdempotencyMiddleware.cs b/backend/src/Modules/Conversations/Presentation/Middleware/IdempotencyMiddleware.cs new file mode 100644 index 0000000..b926c82 --- /dev/null +++ b/backend/src/Modules/Conversations/Presentation/Middleware/IdempotencyMiddleware.cs @@ -0,0 +1,69 @@ +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 3885839..d936692 100644 --- a/client-mobile/app/build.gradle.kts +++ b/client-mobile/app/build.gradle.kts @@ -90,6 +90,8 @@ dependencies { implementation("com.google.dagger:hilt-android:2.48") kapt("com.google.dagger:hilt-android-compiler:2.48") implementation("androidx.hilt:hilt-navigation-compose:1.1.0") + implementation("androidx.hilt:hilt-work:1.1.0") + kapt("androidx.hilt:hilt-compiler:1.1.0") // Network & SignalR implementation("com.squareup.retrofit2:retrofit:2.9.0") @@ -128,6 +130,15 @@ dependencies { implementation("androidx.room:room-ktx:$room_version") kapt("androidx.room:room-compiler:$room_version") + // WorkManager + val work_version = "2.9.0" + implementation("androidx.work:work-runtime-ktx:$work_version") + + // Paging 3 + val paging_version = "3.2.1" + implementation("androidx.paging:paging-runtime-ktx:$paging_version") + implementation("androidx.paging:paging-compose:$paging_version") + // Testing testImplementation("junit:junit:4.13.2") androidTestImplementation("androidx.test.ext:junit:1.1.5") diff --git a/client-mobile/app/src/main/kotlin/com/knot/messenger/MainApplication.kt b/client-mobile/app/src/main/kotlin/com/knot/messenger/MainApplication.kt index 1698408..d1415f3 100644 --- a/client-mobile/app/src/main/kotlin/com/knot/messenger/MainApplication.kt +++ b/client-mobile/app/src/main/kotlin/com/knot/messenger/MainApplication.kt @@ -1,13 +1,25 @@ package com.knot.messenger import android.app.Application +import androidx.hilt.work.HiltWorkerFactory +import androidx.work.Configuration import coil.ImageLoader import coil.ImageLoaderFactory import coil.decode.VideoFrameDecoder import dagger.hilt.android.HiltAndroidApp +import javax.inject.Inject @HiltAndroidApp -class MainApplication : Application(), ImageLoaderFactory { +class MainApplication : Application(), ImageLoaderFactory, Configuration.Provider { + + @Inject + lateinit var workerFactory: HiltWorkerFactory + + override val workManagerConfiguration: Configuration + get() = Configuration.Builder() + .setWorkerFactory(workerFactory) + .build() + override fun newImageLoader(): ImageLoader { return ImageLoader.Builder(this) .components { diff --git a/client-mobile/chats/data/local/dao/ChatDao.kt b/client-mobile/chats/data/local/dao/ChatDao.kt new file mode 100644 index 0000000..809b970 --- /dev/null +++ b/client-mobile/chats/data/local/dao/ChatDao.kt @@ -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> + + @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) + + @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() +} diff --git a/client-mobile/chats/data/local/dao/MessageDao.kt b/client-mobile/chats/data/local/dao/MessageDao.kt new file mode 100644 index 0000000..e53e6e0 --- /dev/null +++ b/client-mobile/chats/data/local/dao/MessageDao.kt @@ -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> + + @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 + + @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 + + @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): List + + @Query("SELECT * FROM messages WHERE status = :status") + fun getMessagesByStatusFlow(status: MessageStatus): Flow> + + @Query("SELECT * FROM messages WHERE status IN (:statuses)") + suspend fun getPendingMessages(statuses: List): List + + @Insert(onConflict = OnConflictStrategy.REPLACE) + suspend fun insertMessage(message: MessageEntity) + + @Insert(onConflict = OnConflictStrategy.REPLACE) + suspend fun insertMessages(messages: List) + + @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): Int + + @Query("SELECT * FROM messages WHERE chatId = :chatId AND status = :status LIMIT 1") + suspend fun getFirstMessageByStatus(chatId: String, status: MessageStatus): MessageEntity? +} diff --git a/client-mobile/chats/data/local/dao/UserProfileDao.kt b/client-mobile/chats/data/local/dao/UserProfileDao.kt new file mode 100644 index 0000000..41ec972 --- /dev/null +++ b/client-mobile/chats/data/local/dao/UserProfileDao.kt @@ -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 + + @Query("SELECT * FROM user_profile") + fun getAllUsers(): Flow> + + @Insert(onConflict = OnConflictStrategy.REPLACE) + suspend fun insertUser(user: UserProfileEntity) + + @Insert(onConflict = OnConflictStrategy.REPLACE) + suspend fun insertUsers(users: List) + + @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) +} diff --git a/client-mobile/chats/data/local/database/AppDatabase.kt b/client-mobile/chats/data/local/database/AppDatabase.kt new file mode 100644 index 0000000..e06a88d --- /dev/null +++ b/client-mobile/chats/data/local/database/AppDatabase.kt @@ -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" + } +} diff --git a/client-mobile/chats/data/local/database/ChatEntity.kt b/client-mobile/chats/data/local/database/ChatEntity.kt new file mode 100644 index 0000000..12291f0 --- /dev/null +++ b/client-mobile/chats/data/local/database/ChatEntity.kt @@ -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() +) diff --git a/client-mobile/chats/data/local/database/MessageEntity.kt b/client-mobile/chats/data/local/database/MessageEntity.kt new file mode 100644 index 0000000..f458253 --- /dev/null +++ b/client-mobile/chats/data/local/database/MessageEntity.kt @@ -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 +) diff --git a/client-mobile/chats/data/local/database/MessageStatusConverter.kt b/client-mobile/chats/data/local/database/MessageStatusConverter.kt new file mode 100644 index 0000000..1977145 --- /dev/null +++ b/client-mobile/chats/data/local/database/MessageStatusConverter.kt @@ -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) +} diff --git a/client-mobile/chats/data/local/database/UserProfileEntity.kt b/client-mobile/chats/data/local/database/UserProfileEntity.kt new file mode 100644 index 0000000..74ac13b --- /dev/null +++ b/client-mobile/chats/data/local/database/UserProfileEntity.kt @@ -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() +) diff --git a/client-mobile/chats/data/local/mappers/ChatMappers.kt b/client-mobile/chats/data/local/mappers/ChatMappers.kt new file mode 100644 index 0000000..839f348 --- /dev/null +++ b/client-mobile/chats/data/local/mappers/ChatMappers.kt @@ -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 загружается отдельно + ) +} diff --git a/client-mobile/chats/data/local/mappers/MessageMappers.kt b/client-mobile/chats/data/local/mappers/MessageMappers.kt new file mode 100644 index 0000000..328c954 --- /dev/null +++ b/client-mobile/chats/data/local/mappers/MessageMappers.kt @@ -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>() {}.type + val reactionsMapType = object : TypeToken>() {}.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>(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>(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() + } +} diff --git a/client-mobile/chats/data/remote/api/ChatApi.kt b/client-mobile/chats/data/remote/api/ChatApi.kt index 0451675..393caaf 100644 --- a/client-mobile/chats/data/remote/api/ChatApi.kt +++ b/client-mobile/chats/data/remote/api/ChatApi.kt @@ -33,7 +33,11 @@ interface ChatApi { ): List @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") diff --git a/client-mobile/chats/data/repository/ChatRepositoryImpl.kt b/client-mobile/chats/data/repository/ChatRepositoryImpl.kt index 6859d4d..0dd0618 100644 --- a/client-mobile/chats/data/repository/ChatRepositoryImpl.kt +++ b/client-mobile/chats/data/repository/ChatRepositoryImpl.kt @@ -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> { + // Single Source of Truth - данные из Room + return chatDao.getAllChats().map { entities -> + entities.map { it.toDomain() } + } + } + override suspend fun getChats(): List { 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> { - // Кэш отключён - всегда возвращаем пустой поток - 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> { + // 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 { 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?, 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, 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() + ), + 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, diff --git a/client-mobile/chats/data/workers/ChatSyncWorker.kt b/client-mobile/chats/data/workers/ChatSyncWorker.kt new file mode 100644 index 0000000..6ed9866 --- /dev/null +++ b/client-mobile/chats/data/workers/ChatSyncWorker.kt @@ -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( + 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() + .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() + + 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, + 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() + ), + status = if (dto.senderId == currentUserId) MessageStatus.SENT else MessageStatus.DELIVERED, + createdAtMillis = System.currentTimeMillis(), + updatedAtMillis = System.currentTimeMillis() + ) + messageDao.insertMessage(entity) + } + } + } +} diff --git a/client-mobile/chats/data/workers/SendMessageWorker.kt b/client-mobile/chats/data/workers/SendMessageWorker.kt new file mode 100644 index 0000000..02bedbc --- /dev/null +++ b/client-mobile/chats/data/workers/SendMessageWorker.kt @@ -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() + .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")) + ) + } + } +} diff --git a/client-mobile/chats/di/ChatModule.kt b/client-mobile/chats/di/ChatModule.kt index bc1a293..302bfe3 100644 --- a/client-mobile/chats/di/ChatModule.kt +++ b/client-mobile/chats/di/ChatModule.kt @@ -26,13 +26,8 @@ object ChatModule { @Provides @Singleton fun provideChatRepository( - api: ChatApi, - tokenManager: TokenManager, - serverConfig: ServerConfig, - hubClient: chats.data.remote.signalr.ChatHubClient - ): ChatRepository { - return ChatRepositoryImpl(api, tokenManager, serverConfig, hubClient) - } + impl: ChatRepositoryImpl + ): ChatRepository = impl @Provides @Singleton diff --git a/client-mobile/chats/di/LocalDatabaseModule.kt b/client-mobile/chats/di/LocalDatabaseModule.kt new file mode 100644 index 0000000..830aa0c --- /dev/null +++ b/client-mobile/chats/di/LocalDatabaseModule.kt @@ -0,0 +1,50 @@ +package chats.di + +import android.content.Context +import androidx.room.Room +import androidx.work.WorkManager +import chats.data.local.dao.ChatDao +import chats.data.local.dao.MessageDao +import chats.data.local.dao.UserProfileDao +import chats.data.local.database.AppDatabase +import dagger.Module +import dagger.Provides +import dagger.hilt.InstallIn +import dagger.hilt.android.qualifiers.ApplicationContext +import dagger.hilt.components.SingletonComponent +import javax.inject.Singleton + +/** + * Hilt модуль для предоставления зависимостей локальной базы данных и WorkManager + */ +@Module +@InstallIn(SingletonComponent::class) +object LocalDatabaseModule { + + @Provides + @Singleton + fun provideAppDatabase(@ApplicationContext context: Context): AppDatabase { + return Room.databaseBuilder( + context, + AppDatabase::class.java, + AppDatabase.DATABASE_NAME + ) + .fallbackToDestructiveMigration() + .build() + } + + @Provides + fun provideMessageDao(database: AppDatabase): MessageDao = database.messageDao() + + @Provides + fun provideChatDao(database: AppDatabase): ChatDao = database.chatDao() + + @Provides + fun provideUserProfileDao(database: AppDatabase): UserProfileDao = database.userProfileDao() + + @Provides + @Singleton + fun provideWorkManager(@ApplicationContext context: Context): WorkManager { + return WorkManager.getInstance(context) + } +} diff --git a/client-mobile/chats/domain/model/Message.kt b/client-mobile/chats/domain/model/Message.kt index d8e6c2d..9fc2e3d 100644 --- a/client-mobile/chats/domain/model/Message.kt +++ b/client-mobile/chats/domain/model/Message.kt @@ -2,6 +2,8 @@ package chats.domain.model import androidx.compose.runtime.Immutable +import chats.domain.model.MessageStatus + @Immutable data class Message( val id: String, @@ -19,7 +21,8 @@ data class Message( val isPinned: Boolean = false, val isForwarded: Boolean = false, val forwardedFromName: String? = null, - val replyTo: Message? = null + val replyTo: Message? = null, + val status: MessageStatus = MessageStatus.SENT // Новое поле для офлайн-статуса ) @Immutable diff --git a/client-mobile/chats/domain/model/MessageStatus.kt b/client-mobile/chats/domain/model/MessageStatus.kt new file mode 100644 index 0000000..f446433 --- /dev/null +++ b/client-mobile/chats/domain/model/MessageStatus.kt @@ -0,0 +1,30 @@ +package chats.domain.model + +/** + * Статусы сообщения для отображения в UI и синхронизации + */ +enum class MessageStatus { + /** Сообщение создано локально, ожидает отправки */ + PENDING, + + /** Начата отправка на сервер */ + SENDING, + + /** Сообщение успешно отправлено на сервер */ + SENT, + + /** Сообщение доставлено получателю */ + DELIVERED, + + /** Сообщение прочитано получателем */ + READ, + + /** Ошибка отправки */ + FAILED, + + /** Удалено локально */ + DELETED, + + /** Неизвестный статус */ + UNKNOWN +} diff --git a/client-mobile/chats/domain/repository/ChatRepository.kt b/client-mobile/chats/domain/repository/ChatRepository.kt index c5b620a..9e91b3d 100644 --- a/client-mobile/chats/domain/repository/ChatRepository.kt +++ b/client-mobile/chats/domain/repository/ChatRepository.kt @@ -3,10 +3,19 @@ package chats.domain.repository import chats.domain.model.Chat import chats.domain.model.Message +import kotlinx.coroutines.flow.Flow + interface ChatRepository { + // Чаты + fun getChatsFlow(): Flow> suspend fun getChats(): List - fun getMessagesFlow(chatId: String): kotlinx.coroutines.flow.Flow> + suspend fun syncChats() + + // Сообщения - Single Source of Truth через Flow + fun getMessagesFlow(chatId: String): Flow> suspend fun getMessages(chatId: String, cursor: String? = null, pivot: Long? = null, limit: Int? = null): List + + // Отправка сообщений с поддержкой офлайн suspend fun sendMessage( chatId: String, content: String?, @@ -15,15 +24,27 @@ interface ChatRepository { replyToId: String? = null, forwardedFromId: String? = null ): Message + + suspend fun retryFailedMessage(localId: String) + suspend fun syncMessagesForChat(chatId: String) + suspend fun schedulePeriodicSync() + + // Реакции suspend fun addReaction(messageId: String, emoji: String) suspend fun sendTypingStatus(chatId: String) suspend fun markMessagesAsRead(chatId: String, lastMessageId: String, lastReadSequenceId: Int) + + // Локальное хранение suspend fun saveMessage(message: Message) suspend fun deleteLocalMessage(messageId: String) + + // Медиа suspend fun uploadMedia(file: java.io.File): String suspend fun getTrendingGifs(page: Int = 0): List suspend fun searchGifs(query: String, page: Int = 0): List suspend fun getGifCategories(): List + + // Управление чатами suspend fun createPersonalChat(userId: String): Chat suspend fun deleteMessage(messageId: String, forEveryone: Boolean) suspend fun editMessage(messageId: String, content: String): Message diff --git a/client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt b/client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt index 8930cdb..8148e04 100644 --- a/client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt +++ b/client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt @@ -2,6 +2,7 @@ package chats.presentation.chat_detail import androidx.lifecycle.ViewModel import androidx.lifecycle.viewModelScope +import android.util.Log import chats.data.remote.signalr.ChatEvent import chats.data.remote.signalr.ChatHubClient import chats.data.repository.toDomain @@ -20,6 +21,7 @@ import core.utils.copyUriToFile import core.utils.ImageUtils import chats.data.remote.api.KlipyGifDto +import chats.domain.model.MessageStatus data class ChatDetailState( val messages: List = emptyList(), @@ -47,7 +49,8 @@ data class ChatDetailState( val forwardingMessages: List = emptyList(), val availableChatsToForward: List = emptyList(), val selectedMessageIds: Set = emptySet(), - val pinnedMessages: List = emptyList() + val pinnedMessages: List = emptyList(), + val isOffline: Boolean = false ) @HiltViewModel @@ -93,6 +96,8 @@ class ChatDetailViewModel @Inject constructor( return tokenManager.getUserId() ?: "" } + private var messagesFlowJob: Job? = null + fun setChatId(chatId: String) { if (currentChatId == chatId) return currentChatId = chatId @@ -115,30 +120,43 @@ class ChatDetailViewModel @Inject constructor( loadChatInfo(chatId) observeSignalREvents(chatId) - // Initial sync from network - refreshMessages(chatId) + // Подписываемся на Flow из Room (Single Source of Truth) + observeMessagesFromLocal(chatId) + + // Запускаем синхронизацию с сервером + viewModelScope.launch { + repository.syncMessagesForChat(chatId) + } } - private fun updateMessages(messages: List) { - val sortedMessages = messages.sortedByDescending { it.sequenceId } - _state.update { it.copy( - messages = sortedMessages, - isLoading = false, - initialScrollIndex = 0 // In reverse layout, 0 is the bottom - ) } + /** + * Подписка на локальные сообщения из Room (Single Source of Truth) + */ + private fun observeMessagesFromLocal(chatId: String) { + messagesFlowJob?.cancel() + messagesFlowJob = viewModelScope.launch { + repository.getMessagesFlow(chatId) + .catch { e -> + Log.e("ChatDetailVM", "Error observing messages", e) + _state.update { it.copy(isLoading = false, error = e.message) } + } + .collect { messages -> + val sortedMessages = messages.sortedByDescending { it.sequenceId } + _state.update { currentState -> + currentState.copy( + messages = sortedMessages, + isLoading = false, + initialScrollIndex = if (currentState.initialScrollIndex == null) 0 else currentState.initialScrollIndex + ) + } + } + } } fun onInitialScrollDone() { _state.update { it.copy(initialScrollIndex = -1) } } - fun refreshMessages(chatId: String) { - viewModelScope.launch { - val messages = repository.getMessages(chatId) - updateMessages(messages) - } - } - fun loadMoreMessages() { val chatId = currentChatId ?: return if (_state.value.isLoading || _state.value.isLoadingMore) return @@ -488,45 +506,24 @@ class ChatDetailViewModel @Inject constructor( return } - // Clear input immediately to avoid double clicks and ensure UI experience - _state.update { it.copy(inputText = "", replyingMessage = null) } + // Clear input immediately для отзывчивого UI + _state.update { it.copy(inputText = "", replyingMessage = null, pendingAttachments = emptyList()) } - val tempId = "temp_${System.currentTimeMillis()}" - val userId = getCurrentUserId() - - // Determine mediaType based on attachments + // Определяем тип медиа val mediaType = when { - pending.isEmpty() -> chats.domain.model.MediaType.TEXT - pending.any { it.extension.lowercase() in listOf("jpg", "jpeg", "png", "webp", "gif") } -> chats.domain.model.MediaType.IMAGE - pending.any { it.extension.lowercase() in listOf("mp4", "mov", "webm") } -> chats.domain.model.MediaType.VIDEO - else -> chats.domain.model.MediaType.TEXT - } - - val tempMessage = Message( - id = tempId, - chatId = chatId, - senderId = userId, - content = if (text.isBlank()) null else text, - createdAt = java.util.Date().toString(), - mediaType = mediaType, - media = emptyList(), - senderName = "Вы", - senderAvatar = null, - reactions = emptyMap(), - isRead = false, - sequenceId = 0 - ) - - viewModelScope.launch { - repository.saveMessage(tempMessage) + pending.isEmpty() -> "text" + pending.any { it.extension.lowercase() in listOf("jpg", "jpeg", "png", "webp", "gif") } -> "image" + pending.any { it.extension.lowercase() in listOf("mp4", "mov", "webm") } -> "video" + pending.any { it.extension.lowercase() in listOf("mp3", "m4a", "wav") } -> "audio" + else -> "text" } viewModelScope.launch { try { - // Upload attachments if any - val attachmentRequests = if (_state.value.pendingAttachments.isNotEmpty()) { + // Upload attachments если есть + val attachmentRequests = if (pending.isNotEmpty()) { _state.update { it.copy(isUploading = true) } - val requests = _state.value.pendingAttachments.map { file -> + val requests = pending.map { file -> val url = repository.uploadMedia(file) chats.data.remote.api.AttachmentRequest( type = when { @@ -540,31 +537,46 @@ class ChatDetailViewModel @Inject constructor( fileSize = file.length() ) } - _state.update { it.copy(isUploading = false, pendingAttachments = emptyList()) } + _state.update { it.copy(isUploading = false) } requests } else { null } - val sentMessage = repository.sendMessage( + // Отправляем сообщение - теперь оно создается локально и ставится в очередь + repository.sendMessage( chatId = chatId, content = if (text.isBlank()) null else text, - type = if (attachmentRequests != null) "media" else "text", + type = mediaType, attachments = attachmentRequests, replyToId = replyToId ) - repository.deleteLocalMessage(tempId) - repository.saveMessage(sentMessage) - // Clear attachments on success - _state.update { it.copy(pendingAttachments = emptyList()) } + // Сообщение автоматически появится в UI через Flow из Room } catch (e: Exception) { - repository.deleteLocalMessage(tempId) _state.update { it.copy(error = e.localizedMessage, isUploading = false) } onFail(text) } } } + /** + * Повторная отправка неудачного сообщения + */ + fun retryMessage(messageId: String) { + viewModelScope.launch { + repository.retryFailedMessage(messageId) + } + } + + /** + * Удаление локального сообщения (отмена отправки) + */ + fun cancelMessage(messageId: String) { + viewModelScope.launch { + repository.deleteLocalMessage(messageId) + } + } + fun addReaction(messageId: String, emoji: String) { val chatId = currentChatId ?: return @@ -581,7 +593,7 @@ 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() diff --git a/client-mobile/chats/presentation/chat_list/ChatListViewModel.kt b/client-mobile/chats/presentation/chat_list/ChatListViewModel.kt index 44fae4c..25dbbd6 100644 --- a/client-mobile/chats/presentation/chat_list/ChatListViewModel.kt +++ b/client-mobile/chats/presentation/chat_list/ChatListViewModel.kt @@ -51,10 +51,51 @@ class ChatListViewModel @Inject constructor( signalrClient.connect(baseUrl.removeSuffix("/api/"), token) } - loadChats() + // Single Source of Truth - подписываемся на Flow из Room + observeChatsFromLocal() + + // Запускаем периодическую синхронизацию + viewModelScope.launch { + repository.schedulePeriodicSync() + } + observeSignalREvents() } + /** + * Подписка на локальные чаты из Room (Single Source of Truth) + */ + private fun observeChatsFromLocal() { + repository.getChatsFlow() + .onEach { chats -> + _state.update { currentState -> + currentState.copy( + chats = sortChats(chats), + isLoading = false + ) + } + } + .catch { e -> + android.util.Log.e("ChatListVM", "Error observing chats", e) + _state.update { it.copy(isLoading = false, error = e.message) } + } + .launchIn(viewModelScope) + } + + /** + * Принудительная синхронизация чатов с сервером + */ + fun loadChats() { + viewModelScope.launch { + _state.update { it.copy(isLoading = true) } + try { + repository.syncChats() + } catch (e: Exception) { + _state.update { it.copy(isLoading = false, error = e.message) } + } + } + } + private fun updatePushToken() { com.google.firebase.messaging.FirebaseMessaging.getInstance().token.addOnCompleteListener { task -> if (task.isSuccessful) { @@ -70,21 +111,8 @@ class ChatListViewModel @Inject constructor( } } - private fun getCurrentUserId(): String = tokenManager.getUserId() ?: "" - fun loadChats() { - viewModelScope.launch { - _state.update { it.copy(isLoading = true) } - try { - val chats = repository.getChats() - _state.update { it.copy(chats = sortChats(chats), isLoading = false) } - } catch (e: Exception) { - _state.update { it.copy(isLoading = false, error = e.message) } - } - } - } - private fun sortChats(chats: List): List { return chats.sortedWith(compareByDescending { it.name.equals("Избранное", ignoreCase = true) || it.name.equals("Saved Messages", ignoreCase = true) @@ -121,7 +149,6 @@ class ChatListViewModel @Inject constructor( .launchIn(viewModelScope) } - private fun updateChatsWithNewMessage(event: ChatEvent.NewMessage) { val baseUrl = serverConfig.getBaseUrl().removeSuffix("/api/") val currentUserId = getCurrentUserId() diff --git a/client-mobile/chats/presentation/components/MessageBubble.kt b/client-mobile/chats/presentation/components/MessageBubble.kt index 8aa7d46..2665646 100644 --- a/client-mobile/chats/presentation/components/MessageBubble.kt +++ b/client-mobile/chats/presentation/components/MessageBubble.kt @@ -29,6 +29,7 @@ import androidx.compose.ui.window.Popup import chats.domain.model.Message import chats.domain.model.MediaType import chats.domain.model.Media +import chats.domain.model.MessageStatus import coil.compose.AsyncImage import core.presentation.components.AppVideoPlayer import core.presentation.components.AppAudioPlayer @@ -483,12 +484,44 @@ fun MessageBubble( } if (isCurrentUser) { Spacer(modifier = Modifier.width(2.dp)) - Icon( - imageVector = if (message.isRead) Icons.Default.DoneAll else Icons.Default.Done, - contentDescription = null, - modifier = Modifier.size(12.dp), - tint = contentColor.copy(alpha = 0.6f) - ) + // Отображение статуса отправки сообщения + when (message.status) { + MessageStatus.PENDING -> { + // Часы - ожидает отправки + Icon( + imageVector = Icons.Default.Schedule, + contentDescription = "Pending", + modifier = Modifier.size(12.dp), + tint = contentColor.copy(alpha = 0.6f) + ) + } + MessageStatus.SENDING -> { + // Круговой индикатор загрузки + CircularProgressIndicator( + modifier = Modifier.size(10.dp), + strokeWidth = 1.5.dp, + color = contentColor.copy(alpha = 0.6f) + ) + } + MessageStatus.FAILED -> { + // Красный крестик - ошибка отправки + Icon( + imageVector = Icons.Default.Error, + contentDescription = "Failed", + modifier = Modifier.size(12.dp), + tint = Color(0xFFFF5252) + ) + } + else -> { + // Одна галочка для SENT, две для DELIVERED/READ + Icon( + imageVector = if (message.status == MessageStatus.READ) Icons.Default.DoneAll else Icons.Default.Done, + contentDescription = null, + modifier = Modifier.size(12.dp), + tint = if (message.status == MessageStatus.READ) Color(0xFF4CAF50) else contentColor.copy(alpha = 0.6f) + ) + } + } } } } // End of bubble Column diff --git a/client-mobile/core/security/TokenManager.kt b/client-mobile/core/security/TokenManager.kt index 112ab94..ff34652 100644 --- a/client-mobile/core/security/TokenManager.kt +++ b/client-mobile/core/security/TokenManager.kt @@ -43,6 +43,14 @@ class TokenManager @Inject constructor(context: Context) { return prefs.getString("user_id", null) } + fun getUsername(): String? { + return prefs.getString("username", null) + } + + fun saveUsername(username: String) { + prefs.edit().putString("username", username).apply() + } + fun deleteToken() { prefs.edit().remove("jwt_token").remove("user_id").remove("refresh_token").apply() }