IM_NEW/MessageService.Infrastructure/Reposities/MessageReposity.cs

using MessageService.Domain.Entities;
using MessageService.Domain.IReposities;
using Microsoft.EntityFrameworkCore;

namespace MessageService.Infrastructure.Reposities
{
    public class MessageReposity : IMessageReposity
    {
        private readonly MessageDbContext db;

        public MessageReposity(MessageDbContext db)
        {
            this.db = db;
        }

        public void Create(Message message)
        {
            db.Messages.Add(message);
        }

        public async Task<Message?> FindByIdAsync(Guid id, CancellationToken cancellationToken = default)
        {
            return await db.Messages.FirstOrDefaultAsync(x => x.Id == id, cancellationToken);
        }

        public async Task<(IEnumerable<Message> messages, bool hasMore)> GetAsync(string streamKey, long? cusor, int direction, int limit, CancellationToken cancellationToken = default)
        {
            var query = db.Messages.Where(x => x.StreamKey == streamKey);
            List<Message> fetched;
            if (direction == 0) // Before: 找比锚点小的，按倒序排
            {
                if (cusor.HasValue)
                    query = query.Where(m => m.SequenceId < cusor.Value);

                fetched = await query
                    .OrderByDescending(m => m.SequenceId) // 最新消息在最前
                    .Take(limit + 1)
                    .ToListAsync(cancellationToken);
            }
            else
            {
                if (cusor is null)
                    return (Array.Empty<Message>(), false);

                fetched = await query
                    .Where(m => m.SequenceId > cusor.Value)
                    .OrderBy(o => o.SequenceId)
                    .Take(limit + 1)
                    .ToListAsync(cancellationToken);
            }

            var hasMore = fetched.Count > limit;
            var messages = fetched.Take(limit).OrderBy(s => s.SequenceId).ToList();
            return (messages, hasMore);
        }

        public async Task<(IEnumerable<Message> messages, bool hasMore)> SearchAsync(
            string streamKey,
            string keyword,
            long? cursor,
            int limit,
            CancellationToken cancellationToken = default)
        {
            var query = db.Messages
                .AsNoTracking()
                .Where(message =>
                    message.StreamKey == streamKey &&
                    message.MsgType == Domain.Enums.MessageType.Text &&
                    message.State == Domain.Enums.MessageState.Sent &&
                    message.Content.Fallback.Contains(keyword));

            if (cursor.HasValue)
            {
                query = query.Where(message => message.SequenceId < cursor.Value);
            }

            var fetched = await query
                .OrderByDescending(message => message.SequenceId)
                .Take(limit + 1)
                .ToListAsync(cancellationToken);

            return (fetched.Take(limit).ToList(), fetched.Count > limit);
        }
    }
}