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

Вторая часть по кэшу

Browse Source
This commit is contained in:
Халимов Рустам
2026-05-08 22:48:56 +03:00
parent d4c41b333a
commit a0b50b57b0
21 changed files with 237 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
.ps1 Normal file
View File

1
backend/src/Host/Host.csproj
View File

@@ -24,6 +24,7 @@
<PackageReference Include="Swashbuckle.AspNetCore" Version="10.1.5" />
<PackageReference Include="System.IdentityModel.Tokens.Jwt" Version="8.16.0" />
<PackageReference Include="Microsoft.IdentityModel.Tokens" Version="8.16.0" />
<PackageReference Include="DKNet.AspCore.Idempotency" Version="1.0.0" />
</ItemGroup>
<ItemGroup>

3
backend/src/Host/Program.cs
View File

@@ -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();
// Регистрация эндпоинтов

21
backend/src/Modules/Conversations/Application/Messages/Send/SendMessageCommandHandler.cs
View File

@@ -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<SendMessageComma
private readonly IChatsUnitOfWork _unitOfWork;
private readonly MediatR.IMediator _mediator;
private readonly IMessagesSettings _messagesSettings;
private readonly IIdempotencyKeyRepository _idempotencyRepository;
private readonly ILogger<SendMessageCommandHandler> _logger;
public SendMessageCommandHandler(
@@ -51,7 +49,6 @@ public sealed class SendMessageCommandHandler : ICommandHandler<SendMessageComma
IChatsUnitOfWork unitOfWork,
MediatR.IMediator mediator,
IMessagesSettings messagesSettings,
IIdempotencyKeyRepository idempotencyRepository,
ILogger<SendMessageCommandHandler> logger)
{
_chatRepository = chatRepository;
@@ -59,23 +56,11 @@ public sealed class SendMessageCommandHandler : ICommandHandler<SendMessageComma
_unitOfWork = unitOfWork;
_mediator = mediator;
_messagesSettings = messagesSettings;
_idempotencyRepository = idempotencyRepository;
_logger = logger;
}
public async Task<Result<Guid>> 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<SendMessageComma
_messageRepository.Add(message);
await _unitOfWork.SaveChangesAsync(cancellationToken);
// 6. Сохранение idempotency ключа после успешного создания сообщения
if (!string.IsNullOrWhiteSpace(request.IdempotencyKey))
{
await _idempotencyRepository.SaveKeyAsync(request.IdempotencyKey, message.Id, request.ChatId, cancellationToken);
}
await _mediator.Publish(new MessageSentDomainEvent(
message.Id,

12
backend/src/Modules/Conversations/DependencyInjection.cs
View File

@@ -2,20 +2,11 @@ using Knot.Contracts.Conversations.Application.Abstractions;
using Knot.Contracts.Conversations.Domain;
using Knot.Contracts.Conversations.Infrastructure.Persistence;
using Knot.Contracts.Messaging.Application.Abstractions;
using Knot.Modules.Conversations.Domain;
using Knot.Modules.Conversations.Infrastructure.Persistence;
using Knot.Modules.Conversations.Infrastructure.Persistence;
using Knot.Modules.Conversations.Infrastructure.Persistence.Mongo;
using Knot.Modules.Conversations.Infrastructure.Persistence.Mongo;
using Knot.Modules.Conversations.Infrastructure.Services;
using Knot.Modules.Conversations.Infrastructure.Services;
using Knot.Shared.Kernel;
using Knot.Shared.Kernel;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection;
namespace Knot.Modules.Conversations;
@@ -42,9 +33,6 @@ public static class DependencyInjection
services.AddScoped<IUserChatSettingsRepository, UserChatSettingsRepository>();
services.AddScoped<Knot.Contracts.Conversations.Domain.IUserFolderSettingsRepository, UserFolderSettingsRepository>();
// Idempotency support
services.AddScoped<IIdempotencyKeyRepository, IdempotencyKeyRepository>();
services.AddMediatR(config =>
config.RegisterServicesFromAssembly(typeof(DependencyInjection).Assembly));

26
backend/src/Modules/Conversations/Domain/IIdempotencyKeyRepository.cs
View File

@@ -1,26 +0,0 @@
using System.Threading;
using System.Threading.Tasks;
namespace Knot.Modules.Conversations.Domain;
/// <summary>
/// Репозиторий для управления idempotency ключами
/// </summary>
public interface IIdempotencyKeyRepository
{
/// <summary>
/// Проверяет наличие ключа и возвращает сообщение если оно уже было обработано
/// </summary>
Task<Guid?> GetProcessedMessageIdAsync(string key, CancellationToken cancellationToken = default);
/// <summary>
/// Сохраняет idempotency ключ и связывает его с сообщением
/// </summary>
Task SaveKeyAsync(string key, Guid messageId, Guid chatId, CancellationToken cancellationToken = default);
/// <summary>
/// Проверяет и сохраняет атомарно (избегаем race condition)
/// Возвращает: MessageId если ключ уже был, null если ключ был сохранен успешно
/// </summary>
Task<Guid?> GetOrSaveKeyAsync(string key, Guid messageId, Guid chatId, CancellationToken cancellationToken = default);
}

57
backend/src/Modules/Conversations/Domain/IdempotencyKeyRecord.cs
View File

@@ -1,57 +0,0 @@
using System;
using MongoDB.Bson.Serialization.Attributes;
namespace Knot.Modules.Conversations.Domain;
/// <summary>
/// Запись об обработанном idempotency ключе для предотвращения дубликатов сообщений.
/// Используется для обеспечения идемпотентности при повторной отправке сообщений офлайн.
/// </summary>
public class IdempotencyKeyRecord
{
[BsonId]
[BsonRepresentation(MongoDB.Bson.BsonType.String)]
public string Id { get; set; } = string.Empty;
/// <summary>
/// Idempotency ключ из заголовка запроса
/// </summary>
[BsonElement("key")]
public string Key { get; set; } = string.Empty;
/// <summary>
/// ID созданного сообщения
/// </summary>
[BsonElement("messageId")]
public Guid MessageId { get; set; }
/// <summary>
/// ID чата
/// </summary>
[BsonElement("chatId")]
public Guid ChatId { get; set; }
/// <summary>
/// Время создания записи
/// </summary>
[BsonElement("createdAt")]
public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
/// <summary>
/// Время истечения записи (через 24 часа для очистки устаревших ключей)
/// </summary>
[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);
}
}

112
backend/src/Modules/Conversations/Domain/IdempotencyKeyRepository.cs
View File

@@ -1,112 +0,0 @@
using Microsoft.Extensions.Logging;
using MongoDB.Driver;
using MongoDB.Bson;
namespace Knot.Modules.Conversations.Domain;
/// <summary>
/// Реализация репозитория idempotency ключей на основе MongoDB
/// </summary>
public class IdempotencyKeyRepository : IIdempotencyKeyRepository
{
private readonly IMongoCollection<IdempotencyKeyRecord> _collection;
private readonly ILogger<IdempotencyKeyRepository> _logger;
public IdempotencyKeyRepository(IMongoClient mongoClient, ILogger<IdempotencyKeyRepository> logger)
{
var database = mongoClient.GetDatabase("KnotDb");
_collection = database.GetCollection<IdempotencyKeyRecord>("idempotency_keys");
_logger = logger;
// Создаем индекс по ключу для быстрого поиска
CreateIndexes();
}
private void CreateIndexes()
{
var keyIndexModel = new CreateIndexModel<IdempotencyKeyRecord>(
Builders<IdempotencyKeyRecord>.IndexKeys.Ascending(x => x.Key),
new CreateIndexOptions { Unique = true }
);
var expireIndexModel = new CreateIndexModel<IdempotencyKeyRecord>(
Builders<IdempotencyKeyRecord>.IndexKeys.Ascending(x => x.ExpiresAt),
new CreateIndexOptions { ExpireAfter = TimeSpan.Zero } // TTL индекс
);
_collection.Indexes.CreateOne(keyIndexModel);
_collection.Indexes.CreateOne(expireIndexModel);
}
public async Task<Guid?> 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<Guid?> 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;
}
}
}

1
backend/src/Modules/Conversations/Knot.Modules.Conversations.csproj
View File

@@ -29,6 +29,7 @@
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="10.0.1" />
<PackageReference Include="MongoDB.Driver" Version="3.2.0" />
<PackageReference Include="SixLabors.ImageSharp" Version="3.1.12" />
<PackageReference Include="DKNet.AspCore.Idempotency" Version="1.0.0" />
</ItemGroup>
<ItemGroup>

69
backend/src/Modules/Conversations/Presentation/Middleware/IdempotencyMiddleware.cs
View File

@@ -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;
/// <summary>
/// Middleware для проверки идемпотентности POST-запросов к сообщениям
/// </summary>
public class IdempotencyMiddleware
{
private readonly RequestDelegate _next;
private readonly ILogger<IdempotencyMiddleware> _logger;
public IdempotencyMiddleware(RequestDelegate next, ILogger<IdempotencyMiddleware> 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<IdempotencyMiddleware>();
}
}

2
client-mobile/app/build.gradle.kts
View File

@@ -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")

4
client-mobile/chats/data/local/dao/ChatDao.kt
View File

@@ -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<List<ChatEntity>>
@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)

13
client-mobile/chats/data/local/dao/MessageDao.kt
View File

@@ -14,6 +14,19 @@ interface MessageDao {
@Query("SELECT * FROM messages WHERE chatId = :chatId ORDER BY sequenceId ASC")
fun getMessagesByChatId(chatId: String): Flow<List<MessageEntity>>
/**
* Пагинированная загрузка сообщений для 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<MessageEntity>
@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<MessageEntity>

2
client-mobile/chats/data/local/database/ChatEntity.kt
View File

@@ -31,7 +31,7 @@ data class ChatEntity(
val lastMessageText: String? = null,
val lastMessageTimestamp: String? = null,
val lastMessageTimestamp: Long? = null,
val updatedAtMillis: Long = System.currentTimeMillis()
)

14
client-mobile/chats/data/local/mappers/ChatMappers.kt
View File

@@ -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
}
}

43
client-mobile/chats/data/local/paging/MessagePagingSource.kt Normal file
View File

@@ -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<Int, MessageEntity>() {
companion object {
private const val PAGE_SIZE = 30
}
override suspend fun load(params: LoadParams<Int>): LoadResult<Int, MessageEntity> {
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, MessageEntity>): Int? {
return state.anchorPosition
}
}

101
client-mobile/chats/data/local/paging/MessageRemoteMediator.kt Normal file
View File

@@ -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<Int, MessageEntity>() {
companion object {
private const val PAGE_SIZE = 30
}
override suspend fun load(
loadType: LoadType,
state: PagingState<Int, MessageEntity>
): 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<String, Int>()
),
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
}

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

@@ -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<PagingData<Message>> {
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<Message> {
val baseUrl = serverConfig.getBaseUrl().removeSuffix("/api/")
val currentUserId = tokenManager.getUserId() ?: ""

12
client-mobile/chats/data/workers/ChatSyncWorker.kt
View File

@@ -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)
}

3
client-mobile/chats/domain/repository/ChatRepository.kt
View File

@@ -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<List<Message>>
fun getMessagesPagingSource(chatId: String): Flow<PagingData<Message>>
suspend fun getMessages(chatId: String, cursor: String? = null, pivot: Long? = null, limit: Int? = null): List<Message>
// Отправка сообщений с поддержкой офлайн

82
client-mobile/chats/presentation/chat_detail/ChatDetailViewModel.kt
View File

@@ -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) }
}
}