using Microsoft.Extensions.Configuration; using MongoDB.Driver; using MongoDB.Driver.Linq; using Knot.Modules.Chats.Domain; using Knot.Shared.Kernel; using System.Text.RegularExpressions; using MongoDB.Bson; namespace Knot.Modules.Chats.Infrastructure.Persistence; public sealed class MessageRepository : IMessageRepository { private readonly IMongoCollection _messages; private readonly ChatsDbContext _dbContext; private readonly MediatR.IMediator _mediator; public MessageRepository(IMongoDatabase mongoDatabase, ChatsDbContext dbContext, MediatR.IMediator mediator) { _messages = mongoDatabase.GetCollection("messages"); _dbContext = dbContext; _mediator = mediator; } public void Add(Message message) { _messages.InsertOne(message); // Publish domain events manualy for mongo entities var events = message.GetDomainEvents().ToList(); message.ClearDomainEvents(); // This runs synchronously or without waiting, better to run async but Add is void // In this implementation setting, fire and forget or wrap sync foreach (var domainEvent in events) { _mediator.Publish(domainEvent).GetAwaiter().GetResult(); } } public async Task GetByIdAsync(Guid id, CancellationToken cancellationToken) { var filter = Builders.Filter.Eq(m => m.Id, id); return await _messages.Find(filter).FirstOrDefaultAsync(cancellationToken); } public async Task> GetChatMessagesAsync(Guid chatId, int limit, int offset, CancellationToken cancellationToken) { var filter = Builders.Filter.Eq(m => m.ChatId, chatId); return await _messages.Find(filter) .SortByDescending(m => m.CreatedAt) .Skip(offset) .Limit(limit) .ToListAsync(cancellationToken); } public async Task GetLatestChatMessageAsync(Guid chatId, CancellationToken cancellationToken) { var filter = Builders.Filter.Eq(m => m.ChatId, chatId); return await _messages.Find(filter) .SortByDescending(m => m.CreatedAt) .FirstOrDefaultAsync(cancellationToken); } public async Task> GetChatMessagesCursorAsync(Guid chatId, DateTime? cursor, int limit, CancellationToken cancellationToken) { var builder = Builders.Filter; var filter = builder.Eq(m => m.ChatId, chatId); if (cursor.HasValue) { filter &= builder.Lt(m => m.CreatedAt, cursor.Value); } return await _messages.Find(filter) .SortByDescending(m => m.CreatedAt) .Limit(limit) .ToListAsync(cancellationToken); } public async Task> SearchMessagesAsync(string query, Guid? chatId, Guid requestingUserId, CancellationToken cancellationToken) { // Not ideal for SQL/Mongo combination but keeping the signature var validChatIdsQuery = _dbContext.Chats .Where(c => c.Members.Any(m => m.UserId == requestingUserId)) .Select(c => c.Id) .ToList(); var builder = Builders.Filter; var filter = builder.In(m => m.ChatId, validChatIdsQuery); if (chatId.HasValue) { filter &= builder.Eq(m => m.ChatId, chatId.Value); } var textFilter = Builders.Filter.Regex("Content", new BsonRegularExpression(Regex.Escape(query), "i")); filter &= textFilter; return await _messages.Find(filter) .SortByDescending(m => m.CreatedAt) .Limit(ChatConstants.SearchMessagesLimit) .ToListAsync(cancellationToken); } public async Task AddReadReceiptsAsync(Guid userId, List messageIds, CancellationToken cancellationToken) { var filter = Builders.Filter.In(m => m.Id, messageIds); var messages = await _messages.Find(filter).ToListAsync(cancellationToken); var writes = new List>(); foreach (var msg in messages) { if (!msg.ReadBy.Any(r => r.UserId == userId)) { var receipt = new ReadReceipt(msg.Id, userId); var pushUpdate = Builders.Update.Push("ReadBy", receipt); var updateModel = new UpdateOneModel(Builders.Filter.Eq(m => m.Id, msg.Id), pushUpdate); writes.Add(updateModel); } } if (writes.Any()) { await _messages.BulkWriteAsync(writes, cancellationToken: cancellationToken); } } public async Task AddReactionAsync(Guid messageId, Guid userId, string emoji, CancellationToken cancellationToken) { var filter = Builders.Filter.Eq(m => m.Id, messageId); var msg = await _messages.Find(filter).FirstOrDefaultAsync(cancellationToken); if (msg == null) return false; if (msg.Reactions.Any(r => r.UserId == userId && r.Emoji == emoji)) { return true; } var reaction = new Reaction(messageId, userId, emoji); var update = Builders.Update.Push("Reactions", reaction); await _messages.UpdateOneAsync(filter, update, cancellationToken: cancellationToken); return true; } public async Task RemoveReactionAsync(Guid messageId, Guid userId, string emoji, CancellationToken cancellationToken) { var filter = Builders.Filter.Eq(m => m.Id, messageId); var msg = await _messages.Find(filter).FirstOrDefaultAsync(cancellationToken); if (msg == null) return false; var reaction = msg.Reactions.FirstOrDefault(r => r.UserId == userId && r.Emoji == emoji); if (reaction == null) return false; var update = Builders.Update.PullFilter("Reactions", Builders.Filter.And( Builders.Filter.Eq("UserId", userId), Builders.Filter.Eq("Emoji", emoji) )); await _messages.UpdateOneAsync(filter, update, cancellationToken: cancellationToken); return true; } public async Task GetLastStoryMessageAsync(Guid chatId, Guid storyId, CancellationToken cancellationToken) { var filter = Builders.Filter.And( Builders.Filter.Eq(m => m.ChatId, chatId), Builders.Filter.Eq("_t", "StoryMessage"), Builders.Filter.Eq("StoryId", storyId) ); return await _messages.Find(filter) .SortByDescending(m => m.CreatedAt) .FirstOrDefaultAsync(cancellationToken); } public async Task GetUnreadCountAsync(Guid chatId, Guid userId, CancellationToken cancellationToken) { var notReadFilter = Builders.Filter.Not( Builders.Filter.ElemMatch("ReadBy", Builders.Filter.Eq("UserId", userId)) ); var finalFilter = Builders.Filter.And( Builders.Filter.Eq(m => m.ChatId, chatId), Builders.Filter.Ne(m => m.SenderId, userId), notReadFilter ); return (int)await _messages.CountDocumentsAsync(finalFilter, cancellationToken: cancellationToken); } public async Task UpdateAsync(Message message, CancellationToken cancellationToken) { var filter = Builders.Filter.Eq(m => m.Id, message.Id); await _messages.ReplaceOneAsync(filter, message, new ReplaceOptions { IsUpsert = true }, cancellationToken); // Publish domain events var events = message.GetDomainEvents().ToList(); message.ClearDomainEvents(); foreach (var domainEvent in events) { await _mediator.Publish(domainEvent, cancellationToken); } } }