Reorganize root folder structure: Remove apps layer layer
This commit is contained in:
Халимов Рустам
@@ -0,0 +1,207 @@
|
||||
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 AddReadReceiptsAsync(Guid userId, List<Guid> messageIds, CancellationToken cancellationToken)
|
||||
{
|
||||
var filter = Builders<Message>.Filter.In(m => m.Id, messageIds);
|
||||
|
||||
var messages = await _messages.Find(filter).ToListAsync(cancellationToken);
|
||||
|
||||
var writes = new List<WriteModel<Message>>();
|
||||
foreach (var msg in messages)
|
||||
{
|
||||
if (!msg.ReadBy.Any(r => r.UserId == userId))
|
||||
{
|
||||
var receipt = new ReadReceipt(msg.Id, userId);
|
||||
var pushUpdate = Builders<Message>.Update.Push("ReadBy", receipt);
|
||||
var updateModel = new UpdateOneModel<Message>(Builders<Message>.Filter.Eq(m => m.Id, msg.Id), pushUpdate);
|
||||
writes.Add(updateModel);
|
||||
}
|
||||
}
|
||||
|
||||
if (writes.Any())
|
||||
{
|
||||
await _messages.BulkWriteAsync(writes, cancellationToken: cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<bool> AddReactionAsync(Guid messageId, Guid userId, string emoji, CancellationToken cancellationToken)
|
||||
{
|
||||
var filter = Builders<Message>.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<Message>.Update.Push("Reactions", reaction);
|
||||
await _messages.UpdateOneAsync(filter, update, cancellationToken: cancellationToken);
|
||||
return true;
|
||||
}
|
||||
|
||||
public async Task<bool> RemoveReactionAsync(Guid messageId, Guid userId, string emoji, CancellationToken cancellationToken)
|
||||
{
|
||||
var filter = Builders<Message>.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<Message>.Update.PullFilter("Reactions",
|
||||
Builders<BsonDocument>.Filter.And(
|
||||
Builders<BsonDocument>.Filter.Eq("UserId", userId),
|
||||
Builders<BsonDocument>.Filter.Eq("Emoji", emoji)
|
||||
));
|
||||
|
||||
await _messages.UpdateOneAsync(filter, update, cancellationToken: cancellationToken);
|
||||
return true;
|
||||
}
|
||||
|
||||
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<int> GetUnreadCountAsync(Guid chatId, Guid userId, CancellationToken cancellationToken)
|
||||
{
|
||||
var notReadFilter = Builders<Message>.Filter.Not(
|
||||
Builders<Message>.Filter.ElemMatch("ReadBy",
|
||||
Builders<BsonDocument>.Filter.Eq("UserId", userId))
|
||||
);
|
||||
|
||||
var finalFilter = Builders<Message>.Filter.And(
|
||||
Builders<Message>.Filter.Eq(m => m.ChatId, chatId),
|
||||
Builders<Message>.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<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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user