forkmessager/backend/src/Modules/Chats/Infrastructure/Persistence/MessageRepository.cs

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<Message> _messages;
    private readonly ChatsDbContext _dbContext;
    private readonly MediatR.IMediator _mediator;

    public MessageRepository(IMongoDatabase mongoDatabase, ChatsDbContext dbContext, MediatR.IMediator mediator)
    {
        _messages = mongoDatabase.GetCollection<Message>("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<Message?> GetByIdAsync(Guid id, CancellationToken cancellationToken)
    {
        var filter = Builders<Message>.Filter.Eq(m => m.Id, id);
        return await _messages.Find(filter).FirstOrDefaultAsync(cancellationToken);
    }

    public async Task<List<Message>> GetChatMessagesAsync(Guid chatId, int limit, int offset, CancellationToken cancellationToken)
    {
        var filter = Builders<Message>.Filter.Eq(m => m.ChatId, chatId);
        return await _messages.Find(filter)
            .SortByDescending(m => m.CreatedAt)
            .Skip(offset)
            .Limit(limit)
            .ToListAsync(cancellationToken);
    }

    public async Task<Message?> GetLatestChatMessageAsync(Guid chatId, CancellationToken cancellationToken)
    {
        var filter = Builders<Message>.Filter.Eq(m => m.ChatId, chatId);
        return await _messages.Find(filter)
            .SortByDescending(m => m.CreatedAt)
            .FirstOrDefaultAsync(cancellationToken);
    }

    public async Task<List<Message>> GetChatMessagesCursorAsync(Guid chatId, DateTime? cursor, int limit, CancellationToken cancellationToken)
    {
        var builder = Builders<Message>.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<List<Message>> 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<Message>.Filter;
        var filter = builder.In(m => m.ChatId, validChatIdsQuery);

        if (chatId.HasValue)
        {
            filter &= builder.Eq(m => m.ChatId, chatId.Value);
        }

        var textFilter = Builders<Message>.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<Message?> GetLastStoryMessageAsync(Guid chatId, Guid storyId, CancellationToken cancellationToken)
    {
        var filter = Builders<Message>.Filter.And(
            Builders<Message>.Filter.Eq(m => m.ChatId, chatId),
            Builders<Message>.Filter.Eq("_t", "StoryMessage"),
            Builders<Message>.Filter.Eq("StoryId", storyId)
        );

        return await _messages.Find(filter)
            .SortByDescending(m => m.CreatedAt)
            .FirstOrDefaultAsync(cancellationToken);
    }



    public async Task UpdateAsync(Message message, CancellationToken cancellationToken)
    {
        var filter = Builders<Message>.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);
        }
    }
}