refactor(im): 将 MongoDB 访问方式从原生驱动改为 MongoDB.Entities
- 移除 MongoDbContext 和 IMongoClient 依赖注入配置 - 引入 MongoDB.Entities 库替代原生 MongoDB 驱动 - 使用 DB.Default 静态方法进行数据库操作 - 更新消息文档模型继承 Entity 基类 - 修改索引创建方式为 MongoDB.Entities 格式 - 调整消息历史查询和撤回功能适配新 API - 更新消息发送和 AI 消息处理的数据访问逻辑
This commit is contained in:
xiangning
parent
69593b2b5f
commit
47c45bd15f
@ -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<ImDbContext>(sp, logger);
|
||||
await MigrateAsync<WorkflowDbContext>(sp, logger, suppressPendingChanges: true);
|
||||
|
||||
// 2. IM MongoDB 建索引（可选，Mongo 不可用时跳过）
|
||||
// 2. IM MongoDB 建索引（MongoDB.Entities：用 DB.Index<MessageDocument> 创建两个复合索引）
|
||||
// 两个复合索引支撑两种查询路径：按时间倒序拉取最新消息、按 ObjectId 游标分页上拉更早消息。
|
||||
await SafeRunAsync(
|
||||
async () =>
|
||||
{
|
||||
var mongoContext = sp.GetService<MongoDbContext>();
|
||||
if (mongoContext is not null)
|
||||
await mongoContext.CreateIndexesAsync();
|
||||
await DB.Default.Index<MessageDocument>()
|
||||
.Key(m => m.ConversationId, KeyType.Ascending)
|
||||
.Key(m => m.CreatedAt, KeyType.Descending)
|
||||
.CreateAsync();
|
||||
await DB.Default.Index<MessageDocument>()
|
||||
.Key(m => m.ConversationId, KeyType.Ascending)
|
||||
.Key(m => m.ID, KeyType.Descending)
|
||||
.CreateAsync();
|
||||
},
|
||||
logger,
|
||||
"IM MongoDB 建索引");
|
||||
|
||||
@ -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;
|
||||
/// </summary>
|
||||
public class GetMessageHistoryQueryHandler(
|
||||
ImDbContext db,
|
||||
MongoDbContext mongoDb,
|
||||
IUserInfoService userInfoService) : IRequestHandler<GetMessageHistoryQuery, MessageHistoryDto>
|
||||
{
|
||||
public async Task<MessageHistoryDto> Handle(GetMessageHistoryQuery request, CancellationToken ct)
|
||||
@ -26,21 +25,20 @@ public class GetMessageHistoryQueryHandler(
|
||||
if (!isParticipant)
|
||||
throw new UnauthorizedException("无权查看该会话的消息");
|
||||
|
||||
var filter = Builders<MessageDocument>.Filter.Eq(m => m.ConversationId, request.ConversationId);
|
||||
var query = DB.Default.Find<MessageDocument>()
|
||||
.Match(m => m.ConversationId == request.ConversationId);
|
||||
|
||||
// 游标条件：ObjectId 大致按时间单调递增，故 Lt(Id, BeforeId) 等价于"取该游标之前的消息"
|
||||
// 游标条件：ObjectId 大致按时间单调递增，故 ID < BeforeId 等价于"取该游标之前的消息"
|
||||
if (!string.IsNullOrEmpty(request.BeforeId))
|
||||
{
|
||||
var beforeFilter = Builders<MessageDocument>.Filter.Lt(m => m.Id, request.BeforeId);
|
||||
filter = Builders<MessageDocument>.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,
|
||||
|
||||
@ -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;
|
||||
/// </list>
|
||||
/// 副作用：更新 Mongo 文档 IsRecalled=true；若撤回的是会话最后一条，更新会话摘要为"[消息已撤回]"。
|
||||
/// </remarks>
|
||||
public class RecallMessageCommandHandler(ImDbContext db, MongoDbContext mongoDb) : IRequestHandler<RecallMessageCommand, Unit>
|
||||
public class RecallMessageCommandHandler(ImDbContext db) : IRequestHandler<RecallMessageCommand, Unit>
|
||||
{
|
||||
/// <summary>发送者本人撤回的时间窗（分钟）。超时后本人不可撤回，仅管理员可。</summary>
|
||||
public const int RecallTimeWindowMinutes = 2;
|
||||
@ -27,10 +27,10 @@ public class RecallMessageCommandHandler(ImDbContext db, MongoDbContext mongoDb)
|
||||
public async Task<Unit> 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<MessageDocument>()
|
||||
.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<MessageDocument>.Update
|
||||
.Set(m => m.IsRecalled, true)
|
||||
.Set(m => m.RecalledAt, DateTime.UtcNow)
|
||||
.Set(m => m.RecalledBy, request.OperatorId);
|
||||
await mongoDb.Messages.UpdateOneAsync(
|
||||
Builders<MessageDocument>.Filter.Eq(m => m.Id, request.MessageId),
|
||||
update, cancellationToken: ct);
|
||||
// 更新 Mongo：标记撤回（按主键定位，逐字段 Modify，无需构造 Builders）
|
||||
await DB.Default.Update<MessageDocument>()
|
||||
.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);
|
||||
|
||||
@ -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;
|
||||
/// </summary>
|
||||
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);
|
||||
|
||||
@ -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<AuditInterceptor>()));
|
||||
services.AddScoped<AuditInterceptor>();
|
||||
|
||||
// MongoDB：消息正文。IMongoClient 单例复用连接池，MongoDbContext Scoped 保证每次请求独立上下文
|
||||
services.AddSingleton<IMongoClient>(_ =>
|
||||
new MongoClient(config.GetConnectionString("MongoDb")));
|
||||
services.AddScoped(sp =>
|
||||
{
|
||||
var client = sp.GetRequiredService<IMongoClient>();
|
||||
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);
|
||||
|
||||
|
||||
@ -1,20 +1,18 @@
|
||||
using IM.Domain.Enums;
|
||||
using MongoDB.Bson;
|
||||
using MongoDB.Bson.Serialization.Attributes;
|
||||
using MongoDB.Entities;
|
||||
|
||||
namespace IM.Infrastructure.Mongo;
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// 消息 MongoDB 文档模型。IM 消息量级大、写频繁，用 Mongo 承载消息正文与元数据，支撑历史消息检索。
|
||||
/// 继承 <see cref="Entity"/>：主键 <c>ID</c>（string/ObjectId）由基类提供并自动生成，
|
||||
/// 替代原 <c>[BsonId] Id</c>。<c>[BsonRepresentation]</c> 特性全部保留，存量数据格式兼容。
|
||||
/// </summary>
|
||||
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; }
|
||||
|
||||
@ -1,44 +0,0 @@
|
||||
using MongoDB.Driver;
|
||||
|
||||
namespace IM.Infrastructure.Mongo;
|
||||
|
||||
/// <summary>
|
||||
/// MongoDB 上下文，负责承载消息集合（高频追加写场景）。
|
||||
/// 与 PostgreSQL 的 <c>ImDbContext</c> 分工：消息正文落 MongoDB，会话/群组元数据落 PostgreSQL。
|
||||
/// </summary>
|
||||
public class MongoDbContext
|
||||
{
|
||||
private readonly IMongoDatabase _database;
|
||||
|
||||
public MongoDbContext(IMongoClient mongoClient, string databaseName)
|
||||
{
|
||||
_database = mongoClient.GetDatabase(databaseName);
|
||||
}
|
||||
|
||||
/// <summary>消息集合。每条文档对应一条聊天消息，主键为 MongoDB ObjectId。</summary>
|
||||
public IMongoCollection<MessageDocument> Messages =>
|
||||
_database.GetCollection<MessageDocument>("messages");
|
||||
|
||||
/// <summary>
|
||||
/// 创建查询所需索引。Admin Host 启动时调用一次（幂等）。
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 两个复合索引分别支撑两种查询路径：
|
||||
/// <list type="bullet">
|
||||
/// <item>{ConversationId ASC, CreatedAt DESC} —— 按时间倒序拉取会话最新消息（默认进入页面）。</item>
|
||||
/// <item>{ConversationId ASC, Id DESC} —— 游标分页按 ObjectId 倒序，支撑"上拉加载更早消息"。</item>
|
||||
/// </list>
|
||||
/// </remarks>
|
||||
public async Task CreateIndexesAsync()
|
||||
{
|
||||
var indexKeys = Builders<MessageDocument>.IndexKeys
|
||||
.Ascending(m => m.ConversationId)
|
||||
.Descending(m => m.CreatedAt);
|
||||
await Messages.Indexes.CreateOneAsync(new CreateIndexModel<MessageDocument>(indexKeys));
|
||||
|
||||
var cursorIndexKeys = Builders<MessageDocument>.IndexKeys
|
||||
.Ascending(m => m.ConversationId)
|
||||
.Descending(m => m.Id);
|
||||
await Messages.Indexes.CreateOneAsync(new CreateIndexModel<MessageDocument>(cursorIndexKeys));
|
||||
}
|
||||
}
|
||||
@ -7,7 +7,7 @@ namespace IM.Infrastructure.Persistence;
|
||||
|
||||
/// <summary>
|
||||
/// IM 服务 PostgreSQL 上下文。管理会话、参与者、群组、群成员四张元数据表。
|
||||
/// 消息正文不在此处，存于 MongoDB（见 <see cref="Mongo.MongoDbContext"/>）。
|
||||
/// 消息正文不在此处，存于 MongoDB（见 <see cref="Mongo.MessageDocument"/>，经 MongoDB.Entities 访问）。
|
||||
/// </summary>
|
||||
public class ImDbContext : DbContext
|
||||
{
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user