diff --git a/src/BuildFlow.Host/Endpoints/Rag/Chat/EditMessageStreamEndpoint.cs b/src/BuildFlow.Host/Endpoints/Rag/Chat/EditMessageStreamEndpoint.cs index 72fa5cc..d288220 100644 --- a/src/BuildFlow.Host/Endpoints/Rag/Chat/EditMessageStreamEndpoint.cs +++ b/src/BuildFlow.Host/Endpoints/Rag/Chat/EditMessageStreamEndpoint.cs @@ -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; /// -/// 编辑已发送的用户消息并重新流式生成 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 响应。 /// -public class EditMessageStreamEndpoint(RagDbContext db, IAIChatAgent chatAgent, IChatMessageCache cache) +public class EditMessageStreamEndpoint(IMediator mediator) : Endpoint { public override void Configure() @@ -37,88 +20,29 @@ public class EditMessageStreamEndpoint(RagDbContext db, IAIChatAgent chatAgent, { var conversationId = Route("ConversationId"); var messageId = Route("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); } } diff --git a/src/BuildFlow.Host/Endpoints/Rag/Chat/StreamMessageEndpoint.cs b/src/BuildFlow.Host/Endpoints/Rag/Chat/StreamMessageEndpoint.cs index 061f901..8a805f8 100644 --- a/src/BuildFlow.Host/Endpoints/Rag/Chat/StreamMessageEndpoint.cs +++ b/src/BuildFlow.Host/Endpoints/Rag/Chat/StreamMessageEndpoint.cs @@ -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) +/// +/// 将聊天流式请求转换为应用层命令并编码为 SSE 响应。 +/// +public class StreamMessageEndpoint(IMediator mediator) : Endpoint { 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("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); } } diff --git a/src/modules/rag/Application/Chat/Commands/EditMessageStreamCommand.cs b/src/modules/rag/Application/Chat/Commands/EditMessageStreamCommand.cs new file mode 100644 index 0000000..f6cbe06 --- /dev/null +++ b/src/modules/rag/Application/Chat/Commands/EditMessageStreamCommand.cs @@ -0,0 +1,12 @@ +using BuildFlow.Application.Chat.DTOs; +using MediatR; + +namespace BuildFlow.Application.Chat.Commands; + +/// +/// 编辑当前用户会话中的提问并重新流式生成回复。 +/// +public sealed record EditMessageStreamCommand( + Guid ConversationId, + Guid MessageId, + string Content) : IStreamRequest; diff --git a/src/modules/rag/Application/Chat/Commands/StreamMessageCommand.cs b/src/modules/rag/Application/Chat/Commands/StreamMessageCommand.cs new file mode 100644 index 0000000..6337e74 --- /dev/null +++ b/src/modules/rag/Application/Chat/Commands/StreamMessageCommand.cs @@ -0,0 +1,10 @@ +using BuildFlow.Application.Chat.DTOs; +using MediatR; + +namespace BuildFlow.Application.Chat.Commands; + +/// +/// 向当前用户拥有的会话发送流式消息。 +/// +public sealed record StreamMessageCommand(Guid ConversationId, string Content) + : IStreamRequest; diff --git a/src/modules/rag/Application/Chat/Dtos/ChatStreamEvent.cs b/src/modules/rag/Application/Chat/Dtos/ChatStreamEvent.cs new file mode 100644 index 0000000..c1f4fa1 --- /dev/null +++ b/src/modules/rag/Application/Chat/Dtos/ChatStreamEvent.cs @@ -0,0 +1,11 @@ +namespace BuildFlow.Application.Chat.DTOs; + +/// +/// 聊天流的内容片段或完成事件。 +/// +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); +} diff --git a/src/modules/rag/Application/Chat/Handlers/EditMessageStreamCommandHandler.cs b/src/modules/rag/Application/Chat/Handlers/EditMessageStreamCommandHandler.cs new file mode 100644 index 0000000..cbb5632 --- /dev/null +++ b/src/modules/rag/Application/Chat/Handlers/EditMessageStreamCommandHandler.cs @@ -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; + +/// +/// 编辑当前用户的提问，截断后续消息并重新流式生成回复。 +/// +public sealed class EditMessageStreamCommandHandler( + RagDbContext db, + IAIChatAgent chatAgent, + IChatMessageCache cache, + ICurrentUserContext userContext) + : IStreamRequestHandler +{ + public async IAsyncEnumerable 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); + } +} diff --git a/src/modules/rag/Application/Chat/Handlers/StreamMessageCommandHandler.cs b/src/modules/rag/Application/Chat/Handlers/StreamMessageCommandHandler.cs new file mode 100644 index 0000000..f8d8cb8 --- /dev/null +++ b/src/modules/rag/Application/Chat/Handlers/StreamMessageCommandHandler.cs @@ -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; + +/// +/// 向当前用户会话发送消息并产出流式回复事件。 +/// +public sealed class StreamMessageCommandHandler( + RagDbContext db, + IAIChatAgent chatAgent, + IChatMessageCache cache, + ICurrentUserContext userContext) + : IStreamRequestHandler +{ + public async IAsyncEnumerable 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); + } +} diff --git a/tests/BuildFlow.Architecture.Tests/RagSecurityArchitectureTests.cs b/tests/BuildFlow.Architecture.Tests/RagSecurityArchitectureTests.cs index 235aeae..2f7761e 100644 --- a/tests/BuildFlow.Architecture.Tests/RagSecurityArchitectureTests.cs +++ b/tests/BuildFlow.Architecture.Tests/RagSecurityArchitectureTests.cs @@ -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( diff --git a/tests/BuildFlow.Rag.Tests/ChatStreamOwnershipTests.cs b/tests/BuildFlow.Rag.Tests/ChatStreamOwnershipTests.cs new file mode 100644 index 0000000..39829e9 --- /dev/null +++ b/tests/BuildFlow.Rag.Tests/ChatStreamOwnershipTests.cs @@ -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; + +/// +/// RAG 流式发送和编辑重生成的对象所有权与事件契约测试。 +/// +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() + .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() + .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() + .UseInMemoryDatabase($"chat-stream-{Guid.NewGuid():N}") + .AddInterceptors(new AuditInterceptor(userContext)) + .Options; + return new RagDbContext(options); + } + + private static async Task 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> CollectAsync(IAsyncEnumerable source) + { + var result = new List(); + await foreach (var item in source) result.Add(item); + return result; + } +}