diff --git a/src/BuildFlow.Host/Configuration/DatabaseInitializer.cs b/src/BuildFlow.Host/Configuration/DatabaseInitializer.cs index 5041352..ebc678b 100644 --- a/src/BuildFlow.Host/Configuration/DatabaseInitializer.cs +++ b/src/BuildFlow.Host/Configuration/DatabaseInitializer.cs @@ -7,6 +7,7 @@ using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using MongoDB.Entities; using Workflow.Infrastructure.Persistence; using RagSeedData = BuildFlow.Infrastructure.Persistence.SeedData; using PlatformSeedData = BuildFlow.Platform.Infrastructure.Persistence.SeedData; @@ -37,13 +38,19 @@ public static class DatabaseInitializer await MigrateAsync(sp, logger); await MigrateAsync(sp, logger, suppressPendingChanges: true); - // 2. IM MongoDB 建索引(可选,Mongo 不可用时跳过) + // 2. IM MongoDB 建索引(MongoDB.Entities:用 DB.Index 创建两个复合索引) + // 两个复合索引支撑两种查询路径:按时间倒序拉取最新消息、按 ObjectId 游标分页上拉更早消息。 await SafeRunAsync( async () => { - var mongoContext = sp.GetService(); - if (mongoContext is not null) - await mongoContext.CreateIndexesAsync(); + await DB.Default.Index() + .Key(m => m.ConversationId, KeyType.Ascending) + .Key(m => m.CreatedAt, KeyType.Descending) + .CreateAsync(); + await DB.Default.Index() + .Key(m => m.ConversationId, KeyType.Ascending) + .Key(m => m.ID, KeyType.Descending) + .CreateAsync(); }, logger, "IM MongoDB 建索引"); diff --git a/src/modules/im/Application/Chat/Handlers/GetMessageHistoryQueryHandler.cs b/src/modules/im/Application/Chat/Handlers/GetMessageHistoryQueryHandler.cs index 0b01da6..83b9643 100644 --- a/src/modules/im/Application/Chat/Handlers/GetMessageHistoryQueryHandler.cs +++ b/src/modules/im/Application/Chat/Handlers/GetMessageHistoryQueryHandler.cs @@ -5,7 +5,7 @@ using IM.Domain.Common; using IM.Infrastructure.Mongo; using IM.Infrastructure.Persistence; using MediatR; -using MongoDB.Driver; +using MongoDB.Entities; namespace IM.Application.Chat.Handlers; @@ -14,7 +14,6 @@ namespace IM.Application.Chat.Handlers; /// public class GetMessageHistoryQueryHandler( ImDbContext db, - MongoDbContext mongoDb, IUserInfoService userInfoService) : IRequestHandler { public async Task Handle(GetMessageHistoryQuery request, CancellationToken ct) @@ -26,21 +25,20 @@ public class GetMessageHistoryQueryHandler( if (!isParticipant) throw new UnauthorizedException("无权查看该会话的消息"); - var filter = Builders.Filter.Eq(m => m.ConversationId, request.ConversationId); + var query = DB.Default.Find() + .Match(m => m.ConversationId == request.ConversationId); - // 游标条件:ObjectId 大致按时间单调递增,故 Lt(Id, BeforeId) 等价于"取该游标之前的消息" + // 游标条件:ObjectId 大致按时间单调递增,故 ID < BeforeId 等价于"取该游标之前的消息" if (!string.IsNullOrEmpty(request.BeforeId)) { - var beforeFilter = Builders.Filter.Lt(m => m.Id, request.BeforeId); - filter = Builders.Filter.And(filter, beforeFilter); + query = query.Match(m => m.ID.CompareTo(request.BeforeId) < 0); } // Limit+1 技巧:多取一条用于判断是否还有更早的消息(HasMore),无需额外 COUNT 查询 - var messages = await mongoDb.Messages - .Find(filter) - .SortByDescending(m => m.CreatedAt) + var messages = await query + .Sort(m => m.CreatedAt, Order.Descending) .Limit(request.Limit + 1) - .ToListAsync(ct); + .ExecuteAsync(ct); var hasMore = messages.Count > request.Limit; if (hasMore) @@ -66,7 +64,7 @@ public class GetMessageHistoryQueryHandler( } return new MessageDto( - m.Id, + m.ID, m.ConversationId, m.SenderId, m.SenderName, diff --git a/src/modules/im/Application/Chat/Handlers/RecallMessageCommandHandler.cs b/src/modules/im/Application/Chat/Handlers/RecallMessageCommandHandler.cs index 5a5f642..f89d39a 100644 --- a/src/modules/im/Application/Chat/Handlers/RecallMessageCommandHandler.cs +++ b/src/modules/im/Application/Chat/Handlers/RecallMessageCommandHandler.cs @@ -3,7 +3,7 @@ using IM.Application.Chat.Commands; using IM.Infrastructure.Mongo; using IM.Infrastructure.Persistence; using MediatR; -using MongoDB.Driver; +using MongoDB.Entities; namespace IM.Application.Chat.Handlers; @@ -19,7 +19,7 @@ namespace IM.Application.Chat.Handlers; /// /// 副作用:更新 Mongo 文档 IsRecalled=true;若撤回的是会话最后一条,更新会话摘要为"[消息已撤回]"。 /// -public class RecallMessageCommandHandler(ImDbContext db, MongoDbContext mongoDb) : IRequestHandler +public class RecallMessageCommandHandler(ImDbContext db) : IRequestHandler { /// 发送者本人撤回的时间窗(分钟)。超时后本人不可撤回,仅管理员可。 public const int RecallTimeWindowMinutes = 2; @@ -27,10 +27,10 @@ public class RecallMessageCommandHandler(ImDbContext db, MongoDbContext mongoDb) public async Task Handle(RecallMessageCommand request, CancellationToken ct) { // 查消息原文(用于校验发送者、创建时间) - var message = await mongoDb.Messages - .Find(m => m.Id == request.MessageId && m.ConversationId == request.ConversationId) - .FirstOrDefaultAsync(ct) - ?? throw new NotFoundException("消息不存在"); + var messages = await DB.Default.Find() + .Match(m => m.ID == request.MessageId && m.ConversationId == request.ConversationId) + .ExecuteAsync(ct); + var message = messages.FirstOrDefault() ?? throw new NotFoundException("消息不存在"); if (message.IsRecalled) throw new BusinessException("该消息已撤回"); @@ -48,14 +48,13 @@ public class RecallMessageCommandHandler(ImDbContext db, MongoDbContext mongoDb) throw new BusinessException($"消息发送超过 {RecallTimeWindowMinutes} 分钟,无法撤回"); } - // 更新 Mongo:标记撤回 - var update = Builders.Update - .Set(m => m.IsRecalled, true) - .Set(m => m.RecalledAt, DateTime.UtcNow) - .Set(m => m.RecalledBy, request.OperatorId); - await mongoDb.Messages.UpdateOneAsync( - Builders.Filter.Eq(m => m.Id, request.MessageId), - update, cancellationToken: ct); + // 更新 Mongo:标记撤回(按主键定位,逐字段 Modify,无需构造 Builders) + await DB.Default.Update() + .MatchID(request.MessageId) + .Modify(m => m.IsRecalled, true) + .Modify(m => m.RecalledAt, DateTime.UtcNow) + .Modify(m => m.RecalledBy, request.OperatorId) + .ExecuteAsync(ct); // 若撤回的是会话最后一条消息,更新摘要为撤回提示,避免列表显示已撤回内容 var conversation = await db.Conversations.FirstOrDefaultAsync(c => c.Id == request.ConversationId, ct); diff --git a/src/modules/im/Application/Chat/Handlers/SendMessageCommandHandler.cs b/src/modules/im/Application/Chat/Handlers/SendMessageCommandHandler.cs index e3ab448..8fc0045 100644 --- a/src/modules/im/Application/Chat/Handlers/SendMessageCommandHandler.cs +++ b/src/modules/im/Application/Chat/Handlers/SendMessageCommandHandler.cs @@ -7,6 +7,7 @@ using IM.Domain.Enums; using IM.Infrastructure.Mongo; using IM.Infrastructure.Persistence; using MediatR; +using MongoDB.Entities; namespace IM.Application.Chat.Handlers; @@ -15,7 +16,6 @@ namespace IM.Application.Chat.Handlers; /// public class SendMessageCommandHandler( ImDbContext db, - MongoDbContext mongoDb, IUserInfoService userInfoService, IUnreadCounter unreadCounter, IOnlinePresenceService onlinePresence, @@ -76,12 +76,12 @@ public class SendMessageCommandHandler( CreatedAt = DateTime.UtcNow }; - // ① 写 MongoDB:消息正文持久化,InsertOne 后自动生成 ObjectId 作为消息主键 - await mongoDb.Messages.InsertOneAsync(message, cancellationToken: ct); + // ① 写 MongoDB:消息正文持久化,SaveAsync 后自动生成 ObjectId 作为消息主键(ID) + await DB.Default.SaveAsync(message, ct); // ② 写 PostgreSQL:仅同步会话列表摘要与会话级"最后一条消息"指针(读扩散热数据)。 // 注意:未读数累加已移出此事务(见下文 ③),避免大群发消息时的 PG 行锁竞争与写放大。 - conversation.LastMessageId = message.Id; + conversation.LastMessageId = message.ID; conversation.LastMessageContent = BuildLastMessageContent(request.ContentType, request.Content, request.Metadata); conversation.LastMessageAt = DateTime.UtcNow; @@ -125,7 +125,7 @@ public class SendMessageCommandHandler( await TryTriggerAiBotAsync(request, conversation, ct); return new MessageDto( - message.Id, + message.ID, message.ConversationId, message.SenderId, request.SenderName, @@ -184,10 +184,10 @@ public class SendMessageCommandHandler( ReplyToId = null, CreatedAt = DateTime.UtcNow }; - await mongoDb.Messages.InsertOneAsync(aiMessage, cancellationToken: ct); + await DB.Default.SaveAsync(aiMessage, ct); // 更新会话摘要 - conversation.LastMessageId = aiMessage.Id; + conversation.LastMessageId = aiMessage.ID; conversation.LastMessageContent = $"[AI] {Truncate(answer, MaxLastMessageContentLength - 5)}"; conversation.LastMessageAt = DateTime.UtcNow; await db.SaveChangesAsync(ct); diff --git a/src/modules/im/Infrastructure/IMInfrastructureModule.cs b/src/modules/im/Infrastructure/IMInfrastructureModule.cs index ba9f195..55b4233 100644 --- a/src/modules/im/Infrastructure/IMInfrastructureModule.cs +++ b/src/modules/im/Infrastructure/IMInfrastructureModule.cs @@ -7,6 +7,7 @@ using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using MongoDB.Driver; +using MongoDB.Entities; using Volo.Abp.Modularity; namespace IM.Infrastructure; @@ -33,15 +34,12 @@ public class IMInfrastructureModule : AbpModule .AddInterceptors(sp.GetRequiredService())); services.AddScoped(); - // MongoDB:消息正文。IMongoClient 单例复用连接池,MongoDbContext Scoped 保证每次请求独立上下文 - services.AddSingleton(_ => - new MongoClient(config.GetConnectionString("MongoDb"))); - services.AddScoped(sp => - { - var client = sp.GetRequiredService(); - var dbName = MongoUrl.Create(config.GetConnectionString("MongoDb")!).DatabaseName; - return new Mongo.MongoDbContext(client, dbName); - }); + // MongoDB(MongoDB.Entities):消息正文。连接串由 ConnectionStrings:MongoDb 提供, + // 库名从连接串路径解析(与原 MongoUrl 逻辑一致)。 + // 用静态 DB.InitAsync 一次性初始化,之后全局通过静态 DB/实体实例方法访问,无需 DI 注入。 + var mongoConn = config.GetConnectionString("MongoDb")!; + var mongoDbName = MongoUrl.Create(mongoConn).DatabaseName!; + DB.InitAsync(mongoDbName, MongoClientSettings.FromConnectionString(mongoConn)).GetAwaiter().GetResult(); services.AddImRedis(config); diff --git a/src/modules/im/Infrastructure/Mongo/MessageDocument.cs b/src/modules/im/Infrastructure/Mongo/MessageDocument.cs index 8246fb8..751d844 100644 --- a/src/modules/im/Infrastructure/Mongo/MessageDocument.cs +++ b/src/modules/im/Infrastructure/Mongo/MessageDocument.cs @@ -1,20 +1,18 @@ using IM.Domain.Enums; using MongoDB.Bson; using MongoDB.Bson.Serialization.Attributes; +using MongoDB.Entities; namespace IM.Infrastructure.Mongo; /// /// 消息 MongoDB 文档模型。IM 消息量级大、写频繁,用 Mongo 承载消息正文与元数据,支撑历史消息检索。 +/// 继承 :主键 ID(string/ObjectId)由基类提供并自动生成, +/// 替代原 [BsonId] Id。[BsonRepresentation] 特性全部保留,存量数据格式兼容。 /// -public class MessageDocument +public class MessageDocument : Entity { - // ObjectId 作为主键,BsonRepresentation 使其在 C# 端表现为 string,便于游标透传 - [BsonId] - [BsonRepresentation(BsonType.ObjectId)] - public string Id { get; set; } = default!; - // Guid 以 String 形式存入 Mongo,保证与 PostgreSQL 中的 ConversationId/SenderId 一致可比 [BsonRepresentation(BsonType.String)] public Guid ConversationId { get; set; } diff --git a/src/modules/im/Infrastructure/Mongo/MongoDbContext.cs b/src/modules/im/Infrastructure/Mongo/MongoDbContext.cs deleted file mode 100644 index 9ceca4d..0000000 --- a/src/modules/im/Infrastructure/Mongo/MongoDbContext.cs +++ /dev/null @@ -1,44 +0,0 @@ -using MongoDB.Driver; - -namespace IM.Infrastructure.Mongo; - -/// -/// MongoDB 上下文,负责承载消息集合(高频追加写场景)。 -/// 与 PostgreSQL 的 ImDbContext 分工:消息正文落 MongoDB,会话/群组元数据落 PostgreSQL。 -/// -public class MongoDbContext -{ - private readonly IMongoDatabase _database; - - public MongoDbContext(IMongoClient mongoClient, string databaseName) - { - _database = mongoClient.GetDatabase(databaseName); - } - - /// 消息集合。每条文档对应一条聊天消息,主键为 MongoDB ObjectId。 - public IMongoCollection Messages => - _database.GetCollection("messages"); - - /// - /// 创建查询所需索引。Admin Host 启动时调用一次(幂等)。 - /// - /// - /// 两个复合索引分别支撑两种查询路径: - /// - /// {ConversationId ASC, CreatedAt DESC} —— 按时间倒序拉取会话最新消息(默认进入页面)。 - /// {ConversationId ASC, Id DESC} —— 游标分页按 ObjectId 倒序,支撑"上拉加载更早消息"。 - /// - /// - public async Task CreateIndexesAsync() - { - var indexKeys = Builders.IndexKeys - .Ascending(m => m.ConversationId) - .Descending(m => m.CreatedAt); - await Messages.Indexes.CreateOneAsync(new CreateIndexModel(indexKeys)); - - var cursorIndexKeys = Builders.IndexKeys - .Ascending(m => m.ConversationId) - .Descending(m => m.Id); - await Messages.Indexes.CreateOneAsync(new CreateIndexModel(cursorIndexKeys)); - } -} diff --git a/src/modules/im/Infrastructure/Persistence/ImDbContext.cs b/src/modules/im/Infrastructure/Persistence/ImDbContext.cs index 930b36e..9af3e29 100644 --- a/src/modules/im/Infrastructure/Persistence/ImDbContext.cs +++ b/src/modules/im/Infrastructure/Persistence/ImDbContext.cs @@ -7,7 +7,7 @@ namespace IM.Infrastructure.Persistence; /// /// IM 服务 PostgreSQL 上下文。管理会话、参与者、群组、群成员四张元数据表。 -/// 消息正文不在此处,存于 MongoDB(见 )。 +/// 消息正文不在此处,存于 MongoDB(见 ,经 MongoDB.Entities 访问)。 /// public class ImDbContext : DbContext {