diff --git a/src/modules/rag/Infrastructure/Cache/ChatMessageCache.cs b/src/modules/rag/Infrastructure/Cache/ChatMessageCache.cs index af3e270..1b22e01 100644 --- a/src/modules/rag/Infrastructure/Cache/ChatMessageCache.cs +++ b/src/modules/rag/Infrastructure/Cache/ChatMessageCache.cs @@ -1,39 +1,95 @@ using System.Text.Json; using BuildFlow.Domain.Interfaces; +using Microsoft.Extensions.Options; using StackExchange.Redis; namespace BuildFlow.Infrastructure.Cache; /// -/// 聊天消息 Redis 缓存。以整个会话的消息列表为粒度序列化为单个 JSON 串存储(Key=chat:messages:{convId}), -/// TTL 1 小时,超时自动失效后回源 DB。用作流式接口与最近消息的热读加速层,DB 才是真源。 +/// 聊天消息 Redis 缓存。每条消息是 List 的一个 JSON 元素,DB 是权威数据源。 +/// 追加和 TTL 刷新由同一个 Lua 脚本原子执行,避免并发读改写丢消息。 /// -public class ChatMessageCache(IConnectionMultiplexer redis) : IChatMessageCache +public class ChatMessageCache( + IConnectionMultiplexer redis, + IOptions options) : IChatMessageCache { + private const string AppendScript = """ + redis.call('RPUSH', KEYS[1], ARGV[1]) + redis.call('PEXPIRE', KEYS[1], ARGV[2]) + return redis.call('LLEN', KEYS[1]) + """; + + private const string ReplaceScript = """ + redis.call('DEL', KEYS[1]) + for index = 1, #ARGV - 1 do + redis.call('RPUSH', KEYS[1], ARGV[index]) + end + redis.call('PEXPIRE', KEYS[1], ARGV[#ARGV]) + return #ARGV - 1 + """; + private readonly IDatabase _db = redis.GetDatabase(); + private readonly long _ttlMilliseconds = + checked((long)TimeSpan.FromMinutes(options.Value.CacheTtlMinutes).TotalMilliseconds); private static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; private static string Key(Guid conversationId) => $"chat:messages:{conversationId}"; public async Task?> GetMessagesAsync(Guid conversationId, CancellationToken ct) { - var value = await _db.StringGetAsync(Key(conversationId)); - if (value.IsNullOrEmpty) return null; - return JsonSerializer.Deserialize>((string)value!, JsonOptions); + ct.ThrowIfCancellationRequested(); + var values = await _db.ListRangeAsync(Key(conversationId), 0, -1); + ct.ThrowIfCancellationRequested(); + if (values.Length == 0) + { + return null; + } + + var messages = new List(values.Length); + foreach (var value in values) + { + var message = JsonSerializer.Deserialize((string)value!, JsonOptions); + if (message is not null) + { + messages.Add(message); + } + } + + return messages; } public async Task SetMessagesAsync(Guid conversationId, List messages, CancellationToken ct) { - var json = JsonSerializer.Serialize(messages, JsonOptions); - await _db.StringSetAsync(Key(conversationId), json, TimeSpan.FromHours(1)); + ct.ThrowIfCancellationRequested(); + if (messages.Count == 0) + { + await _db.KeyDeleteAsync(Key(conversationId)); + ct.ThrowIfCancellationRequested(); + return; + } + + var values = new RedisValue[messages.Count + 1]; + for (var index = 0; index < messages.Count; index++) + { + values[index] = JsonSerializer.Serialize(messages[index], JsonOptions); + } + values[^1] = _ttlMilliseconds; + + await _db.ScriptEvaluateAsync( + ReplaceScript, + [Key(conversationId)], + values); + ct.ThrowIfCancellationRequested(); } - // 追加采用读-改-写:本场景无并发写入同一会话(单用户顺序发送),未引入锁以保持简单 public async Task AppendMessageAsync(Guid conversationId, CachedChatMessage message, CancellationToken ct) { - var messages = await GetMessagesAsync(conversationId, ct) ?? []; - messages.Add(message); - await SetMessagesAsync(conversationId, messages, ct); + ct.ThrowIfCancellationRequested(); + await _db.ScriptEvaluateAsync( + AppendScript, + [Key(conversationId)], + [JsonSerializer.Serialize(message, JsonOptions), _ttlMilliseconds]); + ct.ThrowIfCancellationRequested(); } public async Task RemoveAsync(Guid conversationId, CancellationToken ct) diff --git a/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheCollection.cs b/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheCollection.cs new file mode 100644 index 0000000..c0c6d09 --- /dev/null +++ b/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheCollection.cs @@ -0,0 +1,11 @@ +using Xunit; + +namespace BuildFlow.Rag.Tests.Chat; + +/// 串行执行真实 Redis 聊天缓存测试,避免共享实例上的时序干扰。 +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class RedisChatMessageCacheCollection : ICollectionFixture +{ + /// 集合名称。 + public const string Name = "rag-chat-redis"; +} diff --git a/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheFixture.cs b/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheFixture.cs new file mode 100644 index 0000000..7ce5a89 --- /dev/null +++ b/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheFixture.cs @@ -0,0 +1,33 @@ +using StackExchange.Redis; +using Xunit; + +namespace BuildFlow.Rag.Tests.Chat; + +/// 管理连接到本地独立基础设施 Redis 的测试连接。 +public sealed class RedisChatMessageCacheFixture : IAsyncLifetime +{ + private const string DefaultConnection = "127.0.0.1:56379,abortConnect=false"; + + /// 测试复用的 Redis 连接。 + public IConnectionMultiplexer Redis { get; private set; } = default!; + + public async Task InitializeAsync() + { + var connection = Environment.GetEnvironmentVariable("BUILDFLOW_RAG_REDIS_TEST_CONNECTION") + ?? DefaultConnection; + Redis = await ConnectionMultiplexer.ConnectAsync(connection); + } + + public Task DisposeAsync() + { + Redis.Dispose(); + return Task.CompletedTask; + } + + /// 删除单个测试会话的固定缓存 key。 + public Task DeleteAsync(Guid conversationId) + => Redis.GetDatabase().KeyDeleteAsync(Key(conversationId)); + + /// 返回生产缓存使用的固定 key。 + public static string Key(Guid conversationId) => $"chat:messages:{conversationId}"; +} diff --git a/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheTests.cs b/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheTests.cs new file mode 100644 index 0000000..ee09407 --- /dev/null +++ b/tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheTests.cs @@ -0,0 +1,101 @@ +using BuildFlow.Domain.Interfaces; +using BuildFlow.Infrastructure.Cache; +using FluentAssertions; +using Microsoft.Extensions.Options; +using StackExchange.Redis; +using Xunit; + +namespace BuildFlow.Rag.Tests.Chat; + +/// Redis List 聊天缓存的并发追加、传输粒度、顺序和 TTL 集成测试。 +[Collection(RedisChatMessageCacheCollection.Name)] +public sealed class RedisChatMessageCacheTests(RedisChatMessageCacheFixture fixture) +{ + [Fact] + public async Task ConcurrentAppendsPreserveEveryMessage() + { + var conversationId = Guid.NewGuid(); + await fixture.DeleteAsync(conversationId); + var cache = CreateCache(); + var messages = Enumerable.Range(0, 100) + .Select(index => CreateMessage(index)) + .ToArray(); + + await Task.WhenAll(messages.Select(message => + cache.AppendMessageAsync(conversationId, message, CancellationToken.None))); + + var stored = await cache.GetMessagesAsync(conversationId, CancellationToken.None); + var database = fixture.Redis.GetDatabase(); + stored.Should().NotBeNull(); + var storedMessages = stored!; + storedMessages.Should().HaveCount(100); + storedMessages.Select(x => x.Id).Should().OnlyHaveUniqueItems(); + (await database.KeyTypeAsync(RedisChatMessageCacheFixture.Key(conversationId))) + .Should().Be(RedisType.List); + (await database.ListLengthAsync(RedisChatMessageCacheFixture.Key(conversationId))) + .Should().Be(100); + } + + [Fact] + public async Task AppendAddsOneListElementAndRefreshesConfiguredTtl() + { + var conversationId = Guid.NewGuid(); + await fixture.DeleteAsync(conversationId); + var cache = CreateCache(); + var existing = Enumerable.Range(0, 500) + .Select(CreateMessage) + .ToList(); + await cache.SetMessagesAsync(conversationId, existing, CancellationToken.None); + var database = fixture.Redis.GetDatabase(); + var key = RedisChatMessageCacheFixture.Key(conversationId); + + await cache.AppendMessageAsync( + conversationId, + CreateMessage(500), + CancellationToken.None); + + (await database.KeyTypeAsync(key)).Should().Be(RedisType.List); + (await database.ListLengthAsync(key)).Should().Be(501); + var ttl = await database.KeyTimeToLiveAsync(key); + ttl.Should().NotBeNull(); + ttl!.Value.Should().BeGreaterThan(TimeSpan.FromMinutes(4)); + ttl.Value.Should().BeLessThanOrEqualTo(TimeSpan.FromMinutes(5)); + } + + [Fact] + public async Task SetAndGetRoundTripCompleteHistoryInOrder() + { + var conversationId = Guid.NewGuid(); + await fixture.DeleteAsync(conversationId); + var cache = CreateCache(); + var messages = Enumerable.Range(0, 20) + .Select(CreateMessage) + .ToList(); + + await cache.SetMessagesAsync(conversationId, messages, CancellationToken.None); + var result = await cache.GetMessagesAsync(conversationId, CancellationToken.None); + + result.Should().NotBeNull(); + var storedMessages = result!; + storedMessages.Select(x => x.Id).Should().Equal(messages.Select(x => x.Id)); + storedMessages.Select(x => x.Content).Should().Equal(messages.Select(x => x.Content)); + } + + private ChatMessageCache CreateCache() + => new( + fixture.Redis, + Options.Create(new ChatOptions + { + MaxContextMessages = 50, + MaxContextTokens = 8000, + CacheTtlMinutes = 5, + })); + + private static CachedChatMessage CreateMessage(int index) + => new( + Guid.NewGuid(), + index % 2 == 0 ? "User" : "Assistant", + $"message-{index}", + null, + new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).AddMinutes(index)); +}