build-flow/build-flow.api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: move RAG chat streams behind MediatR

Browse Source
This commit is contained in:
xiangning 2026-07-23 18:14:18 +08:00
parent 0d1fd55ef0
commit 8988c880bc
9 changed files with 463 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
src/BuildFlow.Host/Endpoints/Rag/Chat/EditMessageStreamEndpoint.cs
View File

@ -1,30 +1,13 @@
using System.Text;
using System.Text.Json;
using BuildFlow.Domain.Entities.ChatMessage;
using BuildFlow.Domain.Enums;
using BuildFlow.Domain.Interfaces;
using BuildFlow.Infrastructure.Persistence;
using BuildFlow.Shared.Domain.Errors;
using AiChatMessage = Microsoft.Extensions.AI.ChatMessage;
using AiChatRole = Microsoft.Extensions.AI.ChatRole;
using BuildFlow.Application.Chat.Commands;
using MediatR;
namespace BuildFlow.Host.Endpoints.Rag.Chat;
/// <summary>
/// 编辑已发送的用户消息并重新流式生成 AI 回复。
/// 路由：POST /chat/conversations/{ConversationId}/messages/{MessageId}/edit-stream
///
/// 处理流程（与 StreamMessageEndpoint 保持一致的 endpoint 风格，逻辑在端点内完成）：
/// 1. 校验会话存在 + 目标消息存在且 Role == User（只允许编辑自己的提问）。
/// 2. 更新该 User 消息的 Content 与审计字段（DateTime.UtcNow）。
/// 3. 物理删除该消息"之后"的所有消息（含它的旧 AI 回复及后续所有轮次）——
/// ChatMessageEntity 未实现 ISoftDelete，直接 Remove 符合现状。
/// 4. 清理该会话的全部消息缓存，按 DB 现状重建缓存。
/// 5. 基于更新后的完整历史重新流式生成 AI 回复，流式结束后持久化 Assistant 消息。
///
/// SSE 协议与 StreamMessageEndpoint 一致：data:{content} / event:done。
/// 将聊天编辑流式请求转换为应用层命令并编码为 SSE 响应。
/// </summary>
public class EditMessageStreamEndpoint(RagDbContext db, IAIChatAgent chatAgent, IChatMessageCache cache)
public class EditMessageStreamEndpoint(IMediator mediator)
: Endpoint<EditMessageStreamRequest>
{
public override void Configure()
@ -37,88 +20,29 @@ public class EditMessageStreamEndpoint(RagDbContext db, IAIChatAgent chatAgent,
{
var conversationId = Route<Guid>("ConversationId");
var messageId = Route<Guid>("MessageId");
var conversation = await db.Conversations.FindAsync([conversationId], ct);
if (conversation is null)
{
throw new NotFoundException(ErrorCode.NotFound, "会话不存在");
}
var targetMessage = await db.ChatMessages.FindAsync([messageId], ct);
// 只允许编辑自己的提问（Role == User）。Assistant 消息不允许编辑。
if (targetMessage is null || targetMessage.ConversationId != conversationId)
{
throw new NotFoundException(ErrorCode.NotFound, "消息不存在");
}
if (targetMessage.Role != ChatRoleEnum.User)
{
throw new BusinessException(ErrorCode.Business, "仅支持编辑用户提问消息");
}
var editedAt = DateTime.UtcNow;
// 1. 更新目标消息内容与审计字段
targetMessage.Content = req.Content;
targetMessage.UpdatedAt = editedAt;
targetMessage.UpdatedBy = "system"; // 编辑流无独立操作者，统一以 system 标记；端点已要求 chat:send 权限
// 2. 物理删除该消息"之后"的所有消息（CreatedAt 严格大于该消息）
// 用 editedAt 备份原 CreatedAt 之前的值不必要：删除判定基于消息原始 CreatedAt。
var messagesToRemove = await db.ChatMessages
.Where(m => m.ConversationId == conversationId && m.CreatedAt > targetMessage.CreatedAt)
.ToListAsync(ct);
db.ChatMessages.RemoveRange(messagesToRemove);
await db.SaveChangesAsync(ct);
// 3. 重建该会话的消息缓存：删除旧缓存，按 DB 现状全量回填
await cache.RemoveAsync(conversationId, ct);
var remainingMessages = await db.ChatMessages
.Where(m => m.ConversationId == conversationId)
.OrderBy(m => m.CreatedAt)
.ToListAsync(ct);
await cache.SetMessagesAsync(
conversationId,
remainingMessages.Select(m => new CachedChatMessage(
m.Id, m.Role.ToString(), m.Content, m.TokenUsage, m.CreatedAt)).ToList(),
ct);
// 4. 构造 LLM 上下文（更新后的完整历史）
var chatMessages = remainingMessages.Select(m => new AiChatMessage(
m.Role == ChatRoleEnum.User ? AiChatRole.User : AiChatRole.Assistant,
m.Content)).ToList();
// 5. 流式生成 + SSE 推送
HttpContext.Response.ContentType = "text/event-stream";
HttpContext.Response.Headers.CacheControl = "no-cache";
HttpContext.Response.Headers.Connection = "keep-alive";
var fullReply = new StringBuilder();
await foreach (var chunk in chatAgent.RunStreamingAsync(chatMessages, ct))
await foreach (var streamEvent in mediator.CreateStream(
new EditMessageStreamCommand(
conversationId,
messageId,
req.Content),
ct))
{
fullReply.Append(chunk);
var sseData = JsonSerializer.Serialize(new { content = chunk });
await HttpContext.Response.WriteAsync($"data: {sseData}\n\n", ct);
if (streamEvent.MessageId is Guid completedMessageId)
{
var doneData = JsonSerializer.Serialize(new { messageId = completedMessageId });
await HttpContext.Response.WriteAsync($"event: done\ndata: {doneData}\n\n", ct);
}
else
{
var data = JsonSerializer.Serialize(new { content = streamEvent.Content });
await HttpContext.Response.WriteAsync($"data: {data}\n\n", ct);
}
await HttpContext.Response.Body.FlushAsync(ct);
}
var replyText = fullReply.ToString();
var assistantMessage = new ChatMessageEntity
{
ConversationId = conversationId,
Role = ChatRoleEnum.Assistant,
Content = replyText
};
await db.ChatMessages.AddAsync(assistantMessage, ct);
await db.SaveChangesAsync(ct);
await cache.AppendMessageAsync(conversationId,
new CachedChatMessage(assistantMessage.Id, ChatRoleEnum.Assistant.ToString(),
assistantMessage.Content, null, assistantMessage.CreatedAt), ct);
var doneData = JsonSerializer.Serialize(new { messageId = assistantMessage.Id });
await HttpContext.Response.WriteAsync($"event: done\ndata: {doneData}\n\n", ct);
await HttpContext.Response.Body.FlushAsync(ct);
}
}

79
src/BuildFlow.Host/Endpoints/Rag/Chat/StreamMessageEndpoint.cs
View File

@ -1,15 +1,13 @@
using System.Text;
using System.Text.Json;
using BuildFlow.Domain.Entities.ChatMessage;
using BuildFlow.Domain.Enums;
using BuildFlow.Domain.Interfaces;
using BuildFlow.Infrastructure.Persistence;
using AiChatMessage = Microsoft.Extensions.AI.ChatMessage;
using AiChatRole = Microsoft.Extensions.AI.ChatRole;
using BuildFlow.Application.Chat.Commands;
using MediatR;
namespace BuildFlow.Host.Endpoints.Rag.Chat;
public class StreamMessageEndpoint(RagDbContext db, IAIChatAgent chatAgent, IChatMessageCache cache)
/// <summary>
/// 将聊天流式请求转换为应用层命令并编码为 SSE 响应。
/// </summary>
public class StreamMessageEndpoint(IMediator mediator)
: Endpoint<StreamMessageRequest>
{
public override void Configure()
@ -21,63 +19,26 @@ public class StreamMessageEndpoint(RagDbContext db, IAIChatAgent chatAgent, ICha
public override async Task HandleAsync(StreamMessageRequest req, CancellationToken ct)
{
var conversationId = Route<Guid>("ConversationId");
var conversation = await db.Conversations.FindAsync([conversationId], ct);
if (conversation is null)
{
HttpContext.Response.StatusCode = 404;
return;
}
var userMessage = new ChatMessageEntity
{
ConversationId = conversationId,
Role = ChatRoleEnum.User,
Content = req.Content
};
await db.ChatMessages.AddAsync(userMessage, ct);
await db.SaveChangesAsync(ct);
await cache.AppendMessageAsync(conversationId,
new CachedChatMessage(userMessage.Id, ChatRoleEnum.User.ToString(), userMessage.Content, null, userMessage.CreatedAt), ct);
var dbMessages = await db.ChatMessages
.Where(m => m.ConversationId == conversationId)
.OrderBy(m => m.CreatedAt)
.ToListAsync(ct);
var chatMessages = dbMessages.Select(m => new AiChatMessage(
m.Role == ChatRoleEnum.User ? AiChatRole.User : AiChatRole.Assistant,
m.Content)).ToList();
HttpContext.Response.ContentType = "text/event-stream";
HttpContext.Response.Headers.CacheControl = "no-cache";
HttpContext.Response.Headers.Connection = "keep-alive";
var fullReply = new StringBuilder();
await foreach (var chunk in chatAgent.RunStreamingAsync(chatMessages, ct))
await foreach (var streamEvent in mediator.CreateStream(
new StreamMessageCommand(conversationId, req.Content),
ct))
{
fullReply.Append(chunk);
var sseData = JsonSerializer.Serialize(new { content = chunk });
await HttpContext.Response.WriteAsync($"data: {sseData}\n\n", ct);
if (streamEvent.MessageId is Guid messageId)
{
var doneData = JsonSerializer.Serialize(new { messageId });
await HttpContext.Response.WriteAsync($"event: done\ndata: {doneData}\n\n", ct);
}
else
{
var data = JsonSerializer.Serialize(new { content = streamEvent.Content });
await HttpContext.Response.WriteAsync($"data: {data}\n\n", ct);
}
await HttpContext.Response.Body.FlushAsync(ct);
}
var replyText = fullReply.ToString();
var assistantMessage = new ChatMessageEntity
{
ConversationId = conversationId,
Role = ChatRoleEnum.Assistant,
Content = replyText
};
await db.ChatMessages.AddAsync(assistantMessage, ct);
await db.SaveChangesAsync(ct);
await cache.AppendMessageAsync(conversationId,
new CachedChatMessage(assistantMessage.Id, ChatRoleEnum.Assistant.ToString(), assistantMessage.Content, null, assistantMessage.CreatedAt), ct);
var doneData = JsonSerializer.Serialize(new { messageId = assistantMessage.Id });
await HttpContext.Response.WriteAsync($"event: done\ndata: {doneData}\n\n", ct);
await HttpContext.Response.Body.FlushAsync(ct);
}
}

12
src/modules/rag/Application/Chat/Commands/EditMessageStreamCommand.cs Normal file
View File

@ -0,0 +1,12 @@
using BuildFlow.Application.Chat.DTOs;
using MediatR;
namespace BuildFlow.Application.Chat.Commands;
/// <summary>
/// 编辑当前用户会话中的提问并重新流式生成回复。
/// </summary>
public sealed record EditMessageStreamCommand(
Guid ConversationId,
Guid MessageId,
string Content) : IStreamRequest<ChatStreamEvent>;

10
src/modules/rag/Application/Chat/Commands/StreamMessageCommand.cs Normal file
View File

@ -0,0 +1,10 @@
using BuildFlow.Application.Chat.DTOs;
using MediatR;
namespace BuildFlow.Application.Chat.Commands;
/// <summary>
/// 向当前用户拥有的会话发送流式消息。
/// </summary>
public sealed record StreamMessageCommand(Guid ConversationId, string Content)
: IStreamRequest<ChatStreamEvent>;

11
src/modules/rag/Application/Chat/Dtos/ChatStreamEvent.cs Normal file
View File

@ -0,0 +1,11 @@
namespace BuildFlow.Application.Chat.DTOs;
/// <summary>
/// 聊天流的内容片段或完成事件。
/// </summary>
public sealed record ChatStreamEvent(string? Content, Guid? MessageId)
{
public static ChatStreamEvent Chunk(string content) => new(content, null);
public static ChatStreamEvent Done(Guid messageId) => new(null, messageId);
}

101
src/modules/rag/Application/Chat/Handlers/EditMessageStreamCommandHandler.cs Normal file
View File

@ -0,0 +1,101 @@
using System.Runtime.CompilerServices;
using System.Text;
using BuildFlow.Application.Chat.Commands;
using BuildFlow.Application.Chat.DTOs;
using BuildFlow.Domain.Entities.ChatMessage;
using BuildFlow.Domain.Enums;
using BuildFlow.Domain.Interfaces;
using BuildFlow.Infrastructure.Persistence;
using BuildFlow.Shared.Domain.Common;
using BuildFlow.Shared.Domain.Exceptions;
using MediatR;
using AiChatMessage = Microsoft.Extensions.AI.ChatMessage;
using AiChatRole = Microsoft.Extensions.AI.ChatRole;
namespace BuildFlow.Application.Chat.Handlers;
/// <summary>
/// 编辑当前用户的提问，截断后续消息并重新流式生成回复。
/// </summary>
public sealed class EditMessageStreamCommandHandler(
RagDbContext db,
IAIChatAgent chatAgent,
IChatMessageCache cache,
ICurrentUserContext userContext)
: IStreamRequestHandler<EditMessageStreamCommand, ChatStreamEvent>
{
public async IAsyncEnumerable<ChatStreamEvent> Handle(
EditMessageStreamCommand request,
[EnumeratorCancellation] CancellationToken ct)
{
var userIdText = userContext.GetRequiredUserId();
var userId = Guid.Parse(userIdText);
var conversation = await db.Conversations.FirstOrDefaultAsync(
x => x.Id == request.ConversationId && x.UserId == userId,
ct)
?? throw new NotFoundException(ErrorCode.NotFound, "会话不存在");
var targetMessage = await db.ChatMessages.FirstOrDefaultAsync(
x => x.Id == request.MessageId && x.ConversationId == conversation.Id,
ct)
?? throw new NotFoundException(ErrorCode.NotFound, "消息不存在");
if (targetMessage.Role != ChatRoleEnum.User)
throw new BusinessException(ErrorCode.Business, "仅支持编辑用户提问消息");
var targetCreatedAt = targetMessage.CreatedAt;
targetMessage.Content = request.Content;
targetMessage.UpdatedAt = DateTime.UtcNow;
targetMessage.UpdatedBy = userIdText;
var messagesToRemove = await db.ChatMessages
.Where(x => x.ConversationId == conversation.Id && x.CreatedAt > targetCreatedAt)
.ToListAsync(ct);
db.ChatMessages.RemoveRange(messagesToRemove);
await db.SaveChangesAsync(ct);
await cache.RemoveAsync(conversation.Id, ct);
var remainingMessages = await db.ChatMessages
.Where(x => x.ConversationId == conversation.Id)
.OrderBy(x => x.CreatedAt)
.ToListAsync(ct);
await cache.SetMessagesAsync(
conversation.Id,
remainingMessages.Select(x => new CachedChatMessage(
x.Id,
x.Role.ToString(),
x.Content,
x.TokenUsage,
x.CreatedAt)).ToList(),
ct);
var chatMessages = remainingMessages.Select(x => new AiChatMessage(
x.Role == ChatRoleEnum.User ? AiChatRole.User : AiChatRole.Assistant,
x.Content)).ToList();
var fullReply = new StringBuilder();
await foreach (var chunk in chatAgent.RunStreamingAsync(chatMessages, ct))
{
fullReply.Append(chunk);
yield return ChatStreamEvent.Chunk(chunk);
}
var assistantMessage = new ChatMessageEntity
{
ConversationId = conversation.Id,
Role = ChatRoleEnum.Assistant,
Content = fullReply.ToString(),
};
await db.ChatMessages.AddAsync(assistantMessage, ct);
await db.SaveChangesAsync(ct);
await cache.AppendMessageAsync(
conversation.Id,
new CachedChatMessage(
assistantMessage.Id,
ChatRoleEnum.Assistant.ToString(),
assistantMessage.Content,
null,
assistantMessage.CreatedAt),
ct);
yield return ChatStreamEvent.Done(assistantMessage.Id);
}
}

90
src/modules/rag/Application/Chat/Handlers/StreamMessageCommandHandler.cs Normal file
View File

@ -0,0 +1,90 @@
using System.Runtime.CompilerServices;
using System.Text;
using BuildFlow.Application.Chat.Commands;
using BuildFlow.Application.Chat.DTOs;
using BuildFlow.Domain.Entities.ChatMessage;
using BuildFlow.Domain.Enums;
using BuildFlow.Domain.Interfaces;
using BuildFlow.Infrastructure.Persistence;
using BuildFlow.Shared.Domain.Common;
using BuildFlow.Shared.Domain.Exceptions;
using MediatR;
using AiChatMessage = Microsoft.Extensions.AI.ChatMessage;
using AiChatRole = Microsoft.Extensions.AI.ChatRole;
namespace BuildFlow.Application.Chat.Handlers;
/// <summary>
/// 向当前用户会话发送消息并产出流式回复事件。
/// </summary>
public sealed class StreamMessageCommandHandler(
RagDbContext db,
IAIChatAgent chatAgent,
IChatMessageCache cache,
ICurrentUserContext userContext)
: IStreamRequestHandler<StreamMessageCommand, ChatStreamEvent>
{
public async IAsyncEnumerable<ChatStreamEvent> Handle(
StreamMessageCommand request,
[EnumeratorCancellation] CancellationToken ct)
{
var userId = Guid.Parse(userContext.GetRequiredUserId());
var conversation = await db.Conversations.FirstOrDefaultAsync(
x => x.Id == request.ConversationId && x.UserId == userId,
ct)
?? throw new NotFoundException(ErrorCode.NotFound, "会话不存在");
var userMessage = new ChatMessageEntity
{
ConversationId = conversation.Id,
Role = ChatRoleEnum.User,
Content = request.Content,
};
await db.ChatMessages.AddAsync(userMessage, ct);
await db.SaveChangesAsync(ct);
await cache.AppendMessageAsync(
conversation.Id,
new CachedChatMessage(
userMessage.Id,
ChatRoleEnum.User.ToString(),
userMessage.Content,
null,
userMessage.CreatedAt),
ct);
var history = await db.ChatMessages
.Where(x => x.ConversationId == conversation.Id)
.OrderBy(x => x.CreatedAt)
.ToListAsync(ct);
var chatMessages = history.Select(x => new AiChatMessage(
x.Role == ChatRoleEnum.User ? AiChatRole.User : AiChatRole.Assistant,
x.Content)).ToList();
var fullReply = new StringBuilder();
await foreach (var chunk in chatAgent.RunStreamingAsync(chatMessages, ct))
{
fullReply.Append(chunk);
yield return ChatStreamEvent.Chunk(chunk);
}
var assistantMessage = new ChatMessageEntity
{
ConversationId = conversation.Id,
Role = ChatRoleEnum.Assistant,
Content = fullReply.ToString(),
};
await db.ChatMessages.AddAsync(assistantMessage, ct);
await db.SaveChangesAsync(ct);
await cache.AppendMessageAsync(
conversation.Id,
new CachedChatMessage(
assistantMessage.Id,
ChatRoleEnum.Assistant.ToString(),
assistantMessage.Content,
null,
assistantMessage.CreatedAt),
ct);
yield return ChatStreamEvent.Done(assistantMessage.Id);
}
}

15
tests/BuildFlow.Architecture.Tests/RagSecurityArchitectureTests.cs
View File

@ -17,6 +17,21 @@ public class RagSecurityArchitectureTests
Assert.DoesNotContain("AllowAnonymous()", source);
}
[Theory]
[InlineData("src/BuildFlow.Host/Endpoints/Rag/Chat/StreamMessageEndpoint.cs")]
[InlineData("src/BuildFlow.Host/Endpoints/Rag/Chat/EditMessageStreamEndpoint.cs")]
public void ChatStreamEndpointsShouldUseMediatorWithoutDbContext(string relativePath)
{
var source = ReadSource(relativePath);
Assert.Contains("IMediator mediator", source);
Assert.Contains("mediator.CreateStream", source);
Assert.Contains("event: done", source);
Assert.DoesNotContain("RagDbContext", source);
Assert.DoesNotContain("IAIChatAgent", source);
Assert.DoesNotContain("IChatMessageCache", source);
}
private static string ReadSource(string relativePath)
{
return File.ReadAllText(Path.Combine(

183
tests/BuildFlow.Rag.Tests/ChatStreamOwnershipTests.cs Normal file
View File

@ -0,0 +1,183 @@
using BuildFlow.Application.Chat.Commands;
using BuildFlow.Application.Chat.DTOs;
using BuildFlow.Application.Chat.Handlers;
using BuildFlow.Domain.Entities.ChatMessage;
using BuildFlow.Domain.Entities.Conversation;
using BuildFlow.Domain.Enums;
using BuildFlow.Infrastructure.Persistence;
using BuildFlow.Rag.Tests.Stub;
using BuildFlow.Shared.Domain.Errors;
using BuildFlow.Shared.Domain.Exceptions;
using BuildFlow.Shared.Infrastructure;
using FluentAssertions;
using Microsoft.EntityFrameworkCore;
using Xunit;
namespace BuildFlow.Rag.Tests;
/// <summary>
/// RAG 流式发送和编辑重生成的对象所有权与事件契约测试。
/// </summary>
public class ChatStreamOwnershipTests
{
[Fact]
public async Task StreamMessageForOwnerShouldYieldChunksThenDoneAndPersistMessages()
{
var userContext = new TestCurrentUserContext();
await using var db = CreateDb(userContext);
var conversationId = await SeedConversationAsync(
db,
Guid.Parse(userContext.UserId));
var cache = new StubChatMessageCache();
var handler = new StreamMessageCommandHandler(
db,
new StubAIChatAgent { StreamTokens = ["你", "好", "！"] },
cache,
userContext);
var events = await CollectAsync(handler.Handle(
new StreamMessageCommand(conversationId, "你好"),
CancellationToken.None));
events.Take(3).Select(x => x.Content).Should().Equal("你", "好", "！");
events.Should().HaveCount(4);
events[^1].MessageId.Should().NotBeNull();
events[^1].MessageId!.Value.Should().NotBeEmpty();
db.ChatMessages.Count(x => x.Role == ChatRoleEnum.User).Should().Be(1);
db.ChatMessages.Count(x => x.Role == ChatRoleEnum.Assistant).Should().Be(1);
cache.Count(conversationId).Should().Be(2);
}
[Fact]
public async Task StreamMessageForNonOwnerShouldThrowNotFoundBeforeWriting()
{
var userContext = new TestCurrentUserContext();
await using var db = CreateDb(userContext);
var conversationId = await SeedConversationAsync(db, Guid.NewGuid());
var cache = new StubChatMessageCache();
var handler = new StreamMessageCommandHandler(
db,
new StubAIChatAgent(),
cache,
userContext);
var act = () => CollectAsync(handler.Handle(
new StreamMessageCommand(conversationId, "越权消息"),
CancellationToken.None));
await act.Should().ThrowAsync<NotFoundException>()
.Where(x => x.Code == ErrorCode.NotFound && x.Message == "会话不存在");
db.ChatMessages.Should().BeEmpty();
cache.Count(conversationId).Should().Be(0);
}
[Fact]
public async Task EditStreamForNonOwnerShouldThrowNotFoundBeforeEditing()
{
var userContext = new TestCurrentUserContext();
await using var db = CreateDb(userContext);
var conversationId = await SeedConversationAsync(db, Guid.NewGuid());
var target = new ChatMessageEntity
{
ConversationId = conversationId,
Role = ChatRoleEnum.User,
Content = "原始问题",
};
var later = new ChatMessageEntity
{
ConversationId = conversationId,
Role = ChatRoleEnum.Assistant,
Content = "原始回答",
};
db.ChatMessages.AddRange(target, later);
await db.SaveChangesAsync();
var handler = new EditMessageStreamCommandHandler(
db,
new StubAIChatAgent(),
new StubChatMessageCache(),
userContext);
var act = () => CollectAsync(handler.Handle(
new EditMessageStreamCommand(conversationId, target.Id, "越权编辑"),
CancellationToken.None));
await act.Should().ThrowAsync<NotFoundException>()
.Where(x => x.Code == ErrorCode.NotFound && x.Message == "会话不存在");
target.Content.Should().Be("原始问题");
db.ChatMessages.Should().HaveCount(2);
}
[Fact]
public async Task EditStreamForOwnerShouldPreserveChunkAndDoneEventContract()
{
var userContext = new TestCurrentUserContext();
await using var db = CreateDb(userContext);
var conversationId = await SeedConversationAsync(
db,
Guid.Parse(userContext.UserId));
var target = new ChatMessageEntity
{
ConversationId = conversationId,
Role = ChatRoleEnum.User,
Content = "原始问题",
};
db.ChatMessages.Add(target);
await db.SaveChangesAsync();
await Task.Delay(5);
db.ChatMessages.Add(new ChatMessageEntity
{
ConversationId = conversationId,
Role = ChatRoleEnum.Assistant,
Content = "旧回答",
});
await db.SaveChangesAsync();
var cache = new StubChatMessageCache();
var handler = new EditMessageStreamCommandHandler(
db,
new StubAIChatAgent { StreamTokens = ["新", "回", "答"] },
cache,
userContext);
var events = await CollectAsync(handler.Handle(
new EditMessageStreamCommand(conversationId, target.Id, "新问题"),
CancellationToken.None));
events.Take(3).Select(x => x.Content).Should().Equal("新", "回", "答");
events.Should().HaveCount(4);
events[^1].MessageId.Should().NotBeNull();
events[^1].MessageId!.Value.Should().NotBeEmpty();
target.Content.Should().Be("新问题");
db.ChatMessages.Should().HaveCount(2);
db.ChatMessages.Single(x => x.Role == ChatRoleEnum.Assistant)
.Content.Should().Be("新回答");
cache.Count(conversationId).Should().Be(2);
}
private static RagDbContext CreateDb(TestCurrentUserContext userContext)
{
var options = new DbContextOptionsBuilder<RagDbContext>()
.UseInMemoryDatabase($"chat-stream-{Guid.NewGuid():N}")
.AddInterceptors(new AuditInterceptor(userContext))
.Options;
return new RagDbContext(options);
}
private static async Task<Guid> SeedConversationAsync(RagDbContext db, Guid userId)
{
var conversation = new ConversationEntity
{
Title = "测试会话",
UserId = userId,
};
db.Conversations.Add(conversation);
await db.SaveChangesAsync();
return conversation.Id;
}
private static async Task<List<T>> CollectAsync<T>(IAsyncEnumerable<T> source)
{
var result = new List<T>();
await foreach (var item in source) result.Add(item);
return result;
}
}