From d371ca59dbb1f4db993841414b008780cb997a01 Mon Sep 17 00:00:00 2001 From: xiangning Date: Fri, 24 Jul 2026 09:22:17 +0800 Subject: [PATCH] perf: batch monitoring record upserts --- .../Application/Sync/SyncProcessor.cs | 248 +++++++++++++----- .../Sync/CountingUpsertInterceptor.cs | 125 +++++++++ .../Sync/MonitoringUpsertTests.cs | 184 +++++++++++++ .../Sync/PostgresMonitoringCollection.cs | 13 + .../Sync/PostgresMonitoringFixture.cs | 118 +++++++++ .../Sync/RecordingLogger.cs | 25 ++ .../Sync/StaticExternalConnector.cs | 14 + 7 files changed, 660 insertions(+), 67 deletions(-) create mode 100644 tests/BuildFlow.Monitoring.Tests/Sync/CountingUpsertInterceptor.cs create mode 100644 tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs create mode 100644 tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringCollection.cs create mode 100644 tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringFixture.cs create mode 100644 tests/BuildFlow.Monitoring.Tests/Sync/RecordingLogger.cs create mode 100644 tests/BuildFlow.Monitoring.Tests/Sync/StaticExternalConnector.cs diff --git a/src/modules/monitoring/Application/Sync/SyncProcessor.cs b/src/modules/monitoring/Application/Sync/SyncProcessor.cs index 97edfe8..f0404b5 100644 --- a/src/modules/monitoring/Application/Sync/SyncProcessor.cs +++ b/src/modules/monitoring/Application/Sync/SyncProcessor.cs @@ -1,3 +1,6 @@ +using System.Data; +using System.Data.Common; +using System.Text; using System.Text.Json; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; @@ -27,6 +30,8 @@ public class SyncProcessor( IServiceProvider services, ILogger logger) { + private const int RecordsPerUpsert = 500; + /// 对单个外部系统执行一次同步。返回本次 SyncJobEntity 记录（已落库）。 public async Task ExecuteAsync(ExternalSystemEntity system, string triggerSource, CancellationToken ct) { @@ -55,31 +60,41 @@ public class SyncProcessor( try { var result = await connector.FetchAsync(system, ct); + var records = NormalizeRecords(system.Id, job.Id, result.Records); - // 4. 幂等落库 SyncedRecordEntity - await PersistRecordsAsync(system.Id, job.Id, result.Records, ct); - - // 5. 更新 SyncJobEntity 为 Succeeded（汇总指标写入 json_data） - job.Status = SyncJobStatusEnum.Succeeded; - job.FinishedAt = DateTime.UtcNow; - job.DurationMs = (long)(job.FinishedAt.Value - job.StartedAt!.Value).TotalMilliseconds; - job.RecordCount = result.RecordCount; - job.Data = new JobData + await using (var transaction = await db.Database.BeginTransactionAsync(ct)) { - TriggerSource = triggerSource, - SummaryMetrics = result.SummaryMetrics, - ErrorMessage = null, - }; - await db.SaveChangesAsync(ct); + try + { + await PersistRecordsAsync(records, ct); - // 6. 更新 ExternalSystemEntity 的最近同步信息（写入 json_data） - var sysData = system.Data; - sysData.LastSyncedAt = new DateTimeOffset(job.FinishedAt.Value).ToUnixTimeMilliseconds(); - sysData.LastSyncJobId = job.Id; - system.Data = sysData; // 触发序列化回 JsonData - await db.SaveChangesAsync(ct); + job.Status = SyncJobStatusEnum.Succeeded; + job.FinishedAt = DateTime.UtcNow; + job.DurationMs = (long)(job.FinishedAt.Value - job.StartedAt!.Value).TotalMilliseconds; + job.RecordCount = result.RecordCount; + job.Data = new JobData + { + TriggerSource = triggerSource, + SummaryMetrics = result.SummaryMetrics, + ErrorMessage = null, + }; - // 7. 触发告警评估（失败不影响主流程） + var sysData = system.Data; + sysData.LastSyncedAt = new DateTimeOffset(job.FinishedAt.Value).ToUnixTimeMilliseconds(); + sysData.LastSyncJobId = job.Id; + system.Data = sysData; + + await db.SaveChangesAsync(ct); + await transaction.CommitAsync(ct); + } + catch + { + await transaction.RollbackAsync(CancellationToken.None); + throw; + } + } + + // 成功数据提交后再评估告警，避免告警读取到未提交的同步结果。 try { await EvaluateAlertsAsync(system.Id, result.SummaryMetrics, ct); @@ -95,63 +110,135 @@ public class SyncProcessor( } catch (Exception ex) { + await ReloadAfterFailureAsync(system, job, ct); return await RecordFailureAsync(system, triggerSource, ex.Message, job, ct); } } - /// UPSERT 同步记录：同一 (ExternalSystemId, ExternalId, RecordType) 只保留最新。 + /// 按固定上限执行参数化 UPSERT，同一业务唯一键只保留最新快照。 private async Task PersistRecordsAsync( - Guid systemId, Guid jobId, IReadOnlyList records, CancellationToken ct) + IReadOnlyList records, + CancellationToken ct) { - if (records.Count == 0) return; - - foreach (var data in records) + foreach (var batch in records.Chunk(RecordsPerUpsert)) { - var payload = ParsePayload(data.RawJson); - var existing = await db.SyncedRecords.FirstOrDefaultAsync( - r => r.ExternalSystemId == systemId - && r.ExternalId == data.ExternalId - && r.RecordType == data.RecordType, ct); - - if (existing is not null) - { - existing.SyncJobId = jobId; - existing.OccurredAt = data.OccurredAt; - existing.Data = new RecordData - { - ExternalId = data.ExternalId, - RecordType = data.RecordType, - Payload = payload, - }; - existing.UpdatedAt = DateTime.UtcNow; - } - else - { - db.SyncedRecords.Add(new SyncedRecordEntity - { - ExternalSystemId = systemId, - SyncJobId = jobId, - ExternalId = data.ExternalId, - RecordType = data.RecordType, - OccurredAt = data.OccurredAt, - Data = new RecordData - { - ExternalId = data.ExternalId, - RecordType = data.RecordType, - Payload = payload, - }, - }); - } + await ExecuteUpsertBatchAsync(batch, ct); } - await db.SaveChangesAsync(ct); } - /// 把原始 JSON 解析为 payload 字典（异常返回原始字符串包裹的单键字典）。 - private static Dictionary ParsePayload(string rawJson) + private IReadOnlyList NormalizeRecords( + Guid systemId, + Guid jobId, + IReadOnlyList records) + { + if (records.Count == 0) + { + return []; + } + + var normalized = new List(records.Count); + var now = DateTime.UtcNow; + for (var index = 0; index < records.Count; index++) + { + var data = records[index]; + normalized.Add(new SyncedRecordEntity + { + ExternalSystemId = systemId, + SyncJobId = jobId, + ExternalId = data.ExternalId, + RecordType = data.RecordType, + OccurredAt = data.OccurredAt, + Data = new RecordData + { + ExternalId = data.ExternalId, + RecordType = data.RecordType, + Payload = ParsePayload(systemId, data), + }, + CreatedBy = "system", + CreatedAt = now, + UpdatedBy = "system", + UpdatedAt = now, + IsDeleted = false, + OperatorIP = "system", + }); + } + + return normalized; + } + + private async Task ExecuteUpsertBatchAsync( + IReadOnlyList records, + CancellationToken ct) + { + await using var command = db.Database.GetDbConnection().CreateCommand(); + var sql = new StringBuilder( + """ + INSERT INTO b_synced_records ( + id, external_system_id, sync_job_id, external_id, record_type, + occurred_at, json_data, created_by, created_at, updated_by, + updated_at, is_deleted, operator_ip) + VALUES + """); + + for (var index = 0; index < records.Count; index++) + { + if (index > 0) + { + sql.Append(','); + } + + sql.AppendLine(); + sql.Append($"(@id_{index}, @system_{index}, @job_{index}, @external_{index}, @type_{index}, "); + sql.Append($"@occurred_{index}, CAST(@json_{index} AS jsonb), @created_by_{index}, "); + sql.Append($"@created_at_{index}, @updated_by_{index}, @updated_at_{index}, "); + sql.Append($"@deleted_{index}, @operator_ip_{index})"); + + var record = records[index]; + AddParameter(command, $"id_{index}", record.Id, DbType.Guid); + AddParameter(command, $"system_{index}", record.ExternalSystemId, DbType.Guid); + AddParameter(command, $"job_{index}", record.SyncJobId, DbType.Guid); + AddParameter(command, $"external_{index}", record.ExternalId, DbType.String); + AddParameter(command, $"type_{index}", record.RecordType, DbType.String); + AddParameter(command, $"occurred_{index}", record.OccurredAt, DbType.DateTime); + AddParameter(command, $"json_{index}", record.JsonData, DbType.String); + AddParameter(command, $"created_by_{index}", record.CreatedBy, DbType.String); + AddParameter(command, $"created_at_{index}", record.CreatedAt, DbType.DateTime); + AddParameter(command, $"updated_by_{index}", record.UpdatedBy, DbType.String); + AddParameter(command, $"updated_at_{index}", record.UpdatedAt, DbType.DateTime); + AddParameter(command, $"deleted_{index}", record.IsDeleted, DbType.Boolean); + AddParameter(command, $"operator_ip_{index}", record.OperatorIP, DbType.String); + } + + sql.AppendLine(); + sql.Append( + """ + ON CONFLICT (external_system_id, external_id, record_type) + DO UPDATE SET + sync_job_id = EXCLUDED.sync_job_id, + occurred_at = EXCLUDED.occurred_at, + json_data = EXCLUDED.json_data, + updated_by = EXCLUDED.updated_by, + updated_at = EXCLUDED.updated_at, + operator_ip = EXCLUDED.operator_ip + """); + command.CommandText = sql.ToString(); + await db.Database.ExecuteSqlRawAsync( + command.CommandText, + command.Parameters.Cast(), + ct); + } + + /// 把原始 JSON 解析为 payload 字典；非法内容保留原文但日志只记录标识符。 + private Dictionary ParsePayload(Guid systemId, SyncedRecordData data) { try { - var doc = JsonDocument.Parse(rawJson); + using var doc = JsonDocument.Parse(data.RawJson); + if (doc.RootElement.ValueKind != JsonValueKind.Object) + { + throw new JsonException("同步记录 JSON 根节点必须是对象"); + } + var dict = new Dictionary(); foreach (var prop in doc.RootElement.EnumerateObject()) { @@ -167,12 +254,39 @@ public class SyncProcessor( } return dict; } - catch + catch (JsonException) { - return new Dictionary { ["_raw"] = rawJson }; + logger.LogWarning( + "同步记录 JSON 无效，保留原始内容。系统 {SystemId}，外部记录 {ExternalId}，类型 {RecordType}", + systemId, + data.ExternalId, + data.RecordType); + return new Dictionary { ["_raw"] = data.RawJson }; } } + private static void AddParameter( + DbCommand command, + string name, + object? value, + DbType dbType) + { + var parameter = command.CreateParameter(); + parameter.ParameterName = name; + parameter.DbType = dbType; + parameter.Value = value ?? DBNull.Value; + command.Parameters.Add(parameter); + } + + private async Task ReloadAfterFailureAsync( + ExternalSystemEntity system, + SyncJobEntity job, + CancellationToken ct) + { + await db.Entry(job).ReloadAsync(ct); + await db.Entry(system).ReloadAsync(ct); + } + /// 记录失败任务（job 已存在则更新，否则新建）。 private async Task RecordFailureAsync( ExternalSystemEntity system, string triggerSource, string error, SyncJobEntity? job, CancellationToken ct) diff --git a/tests/BuildFlow.Monitoring.Tests/Sync/CountingUpsertInterceptor.cs b/tests/BuildFlow.Monitoring.Tests/Sync/CountingUpsertInterceptor.cs new file mode 100644 index 0000000..56b4e95 --- /dev/null +++ b/tests/BuildFlow.Monitoring.Tests/Sync/CountingUpsertInterceptor.cs @@ -0,0 +1,125 @@ +using System.Data.Common; +using Microsoft.EntityFrameworkCore.Diagnostics; + +namespace BuildFlow.Monitoring.Tests.Sync; + +/// +/// 统计同步记录查询和 UPSERT 命令，并可让两个旧式先查后写查询在同一点汇合以稳定复现竞争。 +/// +public sealed class CountingUpsertInterceptor(int concurrentSelectParticipants = 0) : DbCommandInterceptor +{ + private readonly TaskCompletionSource _selectBarrier = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _recordSelectCount; + private int _upsertCommandCount; + private int _parameterlessUpsertCount; + private int _selectBarrierArrivals; + + /// 访问同步记录表的 SELECT 命令数。 + public int RecordSelectCount => Volatile.Read(ref _recordSelectCount); + + /// 包含 ON CONFLICT 的同步记录 INSERT 命令数。 + public int UpsertCommandCount => Volatile.Read(ref _upsertCommandCount); + + /// 所有 UPSERT 是否都通过数据库参数传值。 + public bool AllUpsertsParameterized => Volatile.Read(ref _parameterlessUpsertCount) == 0; + + public override InterceptionResult ReaderExecuting( + DbCommand command, + CommandEventData eventData, + InterceptionResult result) + { + Count(command); + return result; + } + + public override async ValueTask> ReaderExecutingAsync( + DbCommand command, + CommandEventData eventData, + InterceptionResult result, + CancellationToken cancellationToken = default) + { + Count(command); + if (concurrentSelectParticipants > 0 && IsRecordSelect(command)) + { + var arrivals = Interlocked.Increment(ref _selectBarrierArrivals); + if (arrivals == concurrentSelectParticipants) + { + _selectBarrier.TrySetResult(); + } + + if (arrivals <= concurrentSelectParticipants) + { + await _selectBarrier.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + } + } + + return result; + } + + public override InterceptionResult NonQueryExecuting( + DbCommand command, + CommandEventData eventData, + InterceptionResult result) + { + Count(command); + return result; + } + + public override ValueTask> NonQueryExecutingAsync( + DbCommand command, + CommandEventData eventData, + InterceptionResult result, + CancellationToken cancellationToken = default) + { + Count(command); + return ValueTask.FromResult(result); + } + + public override InterceptionResult ScalarExecuting( + DbCommand command, + CommandEventData eventData, + InterceptionResult result) + { + Count(command); + return result; + } + + public override ValueTask> ScalarExecutingAsync( + DbCommand command, + CommandEventData eventData, + InterceptionResult result, + CancellationToken cancellationToken = default) + { + Count(command); + return ValueTask.FromResult(result); + } + + private void Count(DbCommand command) + { + if (IsRecordSelect(command)) + { + Interlocked.Increment(ref _recordSelectCount); + } + + if (!IsUpsert(command)) + { + return; + } + + Interlocked.Increment(ref _upsertCommandCount); + if (command.Parameters.Count == 0) + { + Interlocked.Increment(ref _parameterlessUpsertCount); + } + } + + private static bool IsRecordSelect(DbCommand command) + => command.CommandText.TrimStart().StartsWith("SELECT", StringComparison.OrdinalIgnoreCase) + && command.CommandText.Contains("b_synced_records", StringComparison.OrdinalIgnoreCase); + + private static bool IsUpsert(DbCommand command) + => command.CommandText.TrimStart().StartsWith("INSERT", StringComparison.OrdinalIgnoreCase) + && command.CommandText.Contains("b_synced_records", StringComparison.OrdinalIgnoreCase) + && command.CommandText.Contains("ON CONFLICT", StringComparison.OrdinalIgnoreCase); +} diff --git a/tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs b/tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs new file mode 100644 index 0000000..4f75cac --- /dev/null +++ b/tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs @@ -0,0 +1,184 @@ +using FluentAssertions; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Monitoring.Application.Sync; +using Monitoring.Domain.Entities.ExternalSystem; +using Monitoring.Domain.Enums; +using Monitoring.Domain.Services; +using Monitoring.Domain.Values; + +namespace BuildFlow.Monitoring.Tests.Sync; + +/// Monitoring 同步记录批量 UPSERT、并发幂等和日志安全集成测试。 +[Collection(PostgresMonitoringCollection.Name)] +public sealed class MonitoringUpsertTests(PostgresMonitoringFixture fixture) +{ + [Fact] + public async Task Persists1001RecordsInThreeParameterizedUpsertsWithoutRecordSelects() + { + var interceptor = new CountingUpsertInterceptor(); + await using var db = fixture.CreateDbContext(interceptor); + var system = CreateSystem("batch-upsert"); + db.ExternalSystems.Add(system); + await db.SaveChangesAsync(); + + var records = Enumerable.Range(0, 1001) + .Select(index => new SyncedRecordData( + $"order-{index}", + "order", + $$"""{"value":{{index}}}""", + DateTime.UtcNow.AddMinutes(index))) + .ToArray(); + using var services = CreateServices(new StaticExternalConnector(CreateResult(records))); + var processor = new SyncProcessor(db, services, new RecordingLogger()); + + var job = await processor.ExecuteAsync(system, "test", CancellationToken.None); + var upsertCount = interceptor.UpsertCommandCount; + var selectCount = interceptor.RecordSelectCount; + var allParameterized = interceptor.AllUpsertsParameterized; + db.ChangeTracker.Clear(); + var persistedCount = await db.SyncedRecords.CountAsync(x => x.ExternalSystemId == system.Id); + + job.Status.Should().Be(SyncJobStatusEnum.Succeeded); + persistedCount.Should().Be(1001); + upsertCount.Should().Be(3); + selectCount.Should().Be(0); + allParameterized.Should().BeTrue(); + } + + [Fact] + public async Task ConcurrentSynchronizationsUpsertOneCoherentLatestRecord() + { + var systemId = Guid.NewGuid(); + await using (var seedDb = fixture.CreateDbContext()) + { + seedDb.ExternalSystems.Add(CreateSystem("concurrent", systemId)); + await seedDb.SaveChangesAsync(); + } + + var occurredAtA = new DateTime(2026, 7, 23, 1, 0, 0, DateTimeKind.Utc); + var occurredAtB = occurredAtA.AddMinutes(1); + var interceptor = new CountingUpsertInterceptor(concurrentSelectParticipants: 2); + await using var dbA = fixture.CreateDbContext(interceptor); + await using var dbB = fixture.CreateDbContext(interceptor); + var systemA = await dbA.ExternalSystems.SingleAsync(x => x.Id == systemId); + var systemB = await dbB.ExternalSystems.SingleAsync(x => x.Id == systemId); + using var servicesA = CreateServices(new StaticExternalConnector(CreateResult( + [new SyncedRecordData("shared-order", "order", """{"source":"A"}""", occurredAtA)]))); + using var servicesB = CreateServices(new StaticExternalConnector(CreateResult( + [new SyncedRecordData("shared-order", "order", """{"source":"B"}""", occurredAtB)]))); + var processorA = new SyncProcessor(dbA, servicesA, new RecordingLogger()); + var processorB = new SyncProcessor(dbB, servicesB, new RecordingLogger()); + + var jobs = await Task.WhenAll( + processorA.ExecuteAsync(systemA, "concurrent-a", CancellationToken.None), + processorB.ExecuteAsync(systemB, "concurrent-b", CancellationToken.None)); + + jobs.Should().OnlyContain(job => job.Status == SyncJobStatusEnum.Succeeded); + await using var verifyDb = fixture.CreateDbContext(); + var records = await verifyDb.SyncedRecords + .Where(x => x.ExternalSystemId == systemId) + .ToListAsync(); + records.Should().ContainSingle(); + var record = records.Single(); + new[] { jobs[0].Id, jobs[1].Id }.Should().Contain(record.SyncJobId); + if (record.SyncJobId == jobs[0].Id) + { + record.OccurredAt.Should().Be(occurredAtA); + record.Data.Payload["source"]?.ToString().Should().Be("A"); + } + else + { + record.OccurredAt.Should().Be(occurredAtB); + record.Data.Payload["source"]?.ToString().Should().Be("B"); + } + } + + [Fact] + public async Task SequentialSynchronizationReplacesJobTimeAndPayload() + { + await using var db = fixture.CreateDbContext(); + var system = CreateSystem("sequential"); + db.ExternalSystems.Add(system); + await db.SaveChangesAsync(); + + var firstTime = new DateTime(2026, 7, 23, 2, 0, 0, DateTimeKind.Utc); + var secondTime = firstTime.AddHours(1); + using var firstServices = CreateServices(new StaticExternalConnector(CreateResult( + [new SyncedRecordData("updated-order", "order", """{"version":"first"}""", firstTime)]))); + var firstJob = await new SyncProcessor(db, firstServices, new RecordingLogger()) + .ExecuteAsync(system, "first", CancellationToken.None); + using var secondServices = CreateServices(new StaticExternalConnector(CreateResult( + [new SyncedRecordData("updated-order", "order", """{"version":"second"}""", secondTime)]))); + + var secondJob = await new SyncProcessor(db, secondServices, new RecordingLogger()) + .ExecuteAsync(system, "second", CancellationToken.None); + + db.ChangeTracker.Clear(); + var record = await db.SyncedRecords.SingleAsync(x => + x.ExternalSystemId == system.Id && x.ExternalId == "updated-order"); + firstJob.Status.Should().Be(SyncJobStatusEnum.Succeeded); + secondJob.Status.Should().Be(SyncJobStatusEnum.Succeeded); + record.SyncJobId.Should().Be(secondJob.Id); + record.OccurredAt.Should().Be(secondTime); + record.Data.Payload["version"]?.ToString().Should().Be("second"); + } + + [Fact] + public async Task InvalidJsonPreservesRawPayloadAndLogsIdentifiersOnly() + { + const string rawPayload = "invalid-json-sensitive-payload"; + const string externalId = "invalid-order"; + const string recordType = "order"; + await using var db = fixture.CreateDbContext(); + var system = CreateSystem("invalid-json"); + db.ExternalSystems.Add(system); + await db.SaveChangesAsync(); + using var services = CreateServices(new StaticExternalConnector(CreateResult( + [new SyncedRecordData(externalId, recordType, rawPayload, DateTime.UtcNow)]))); + var logger = new RecordingLogger(); + var processor = new SyncProcessor(db, services, logger); + + var job = await processor.ExecuteAsync(system, "invalid-json", CancellationToken.None); + + db.ChangeTracker.Clear(); + var record = await db.SyncedRecords.SingleAsync(x => x.ExternalSystemId == system.Id); + var warnings = logger.Entries.Where(entry => entry.Level == LogLevel.Warning).ToArray(); + job.Status.Should().Be(SyncJobStatusEnum.Succeeded); + record.Data.Payload["_raw"]?.ToString().Should().Be(rawPayload); + warnings.Should().ContainSingle(); + warnings[0].Message.Should().Contain(system.Id.ToString()); + warnings[0].Message.Should().Contain(externalId); + warnings[0].Message.Should().Contain(recordType); + warnings[0].Message.Should().NotContain(rawPayload); + } + + private static ExternalSystemEntity CreateSystem(string name, Guid? id = null) + { + var system = new ExternalSystemEntity + { + Data = new SystemData + { + Name = name, + SystemType = ExternalSystemTypeEnum.GenericHttp, + BaseUrl = "https://example.test", + }, + }; + if (id.HasValue) + { + system.Id = id.Value; + } + + return system; + } + + private static FetchResult CreateResult(IReadOnlyList records) + => new(records.Count, new Dictionary(), records); + + private static ServiceProvider CreateServices(IExternalConnector connector) + => new ServiceCollection() + .AddSingleton(connector) + .AddSingleton(connector) + .BuildServiceProvider(); +} diff --git a/tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringCollection.cs b/tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringCollection.cs new file mode 100644 index 0000000..27ad2ee --- /dev/null +++ b/tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringCollection.cs @@ -0,0 +1,13 @@ +using Xunit; + +namespace BuildFlow.Monitoring.Tests.Sync; + +/// +/// 串行执行 Monitoring PostgreSQL 集成测试，避免数据库创建和并发用例互相干扰。 +/// +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class PostgresMonitoringCollection : ICollectionFixture +{ + /// 集合名称。 + public const string Name = "monitoring-postgres"; +} diff --git a/tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringFixture.cs b/tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringFixture.cs new file mode 100644 index 0000000..de96b3f --- /dev/null +++ b/tests/BuildFlow.Monitoring.Tests/Sync/PostgresMonitoringFixture.cs @@ -0,0 +1,118 @@ +using BuildFlow.Shared.Domain.Common; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Monitoring.Infrastructure.Persistence; +using Npgsql; +using Xunit; + +namespace BuildFlow.Monitoring.Tests.Sync; + +/// +/// 为 Monitoring 集成测试创建并销毁独立 PostgreSQL 数据库。 +/// +public sealed class PostgresMonitoringFixture : IAsyncLifetime +{ + private const string DefaultConnection = + "Host=127.0.0.1;Port=55432;Username=buildflow;Password=buildflow;Database=postgres"; + + private string _databaseName = default!; + private string _adminConnectionString = default!; + + /// 当前测试数据库连接串。 + public string ConnectionString { get; private set; } = default!; + + public async Task InitializeAsync() + { + var configured = Environment.GetEnvironmentVariable("BUILDFLOW_MONITORING_TEST_CONNECTION") + ?? DefaultConnection; + var adminBuilder = new NpgsqlConnectionStringBuilder(configured) + { + Database = "postgres", + Pooling = false, + }; + _adminConnectionString = adminBuilder.ConnectionString; + _databaseName = $"monitoring_upsert_{Guid.NewGuid():N}"; + + await using (var connection = new NpgsqlConnection(_adminConnectionString)) + { + await connection.OpenAsync(); + await using var command = connection.CreateCommand(); + command.CommandText = $"CREATE DATABASE \"{_databaseName}\""; + await command.ExecuteNonQueryAsync(); + } + + var databaseBuilder = new NpgsqlConnectionStringBuilder(configured) + { + Database = _databaseName, + Pooling = true, + }; + ConnectionString = databaseBuilder.ConnectionString; + + await using var db = CreateDbContext(); + await db.Database.EnsureCreatedAsync(); + } + + /// 创建指向独立测试数据库的 MonitoringDbContext。 + public MonitoringDbContext CreateDbContext(params IInterceptor[] interceptors) + { + var optionsBuilder = new DbContextOptionsBuilder() + .UseNpgsql(ConnectionString) + .UseSnakeCaseNamingConvention(); + if (interceptors.Length > 0) + { + optionsBuilder.AddInterceptors(interceptors); + } + + var db = new MonitoringDbContext(optionsBuilder.Options); + db.SavingChanges += FillAuditFields; + return db; + } + + public async Task DisposeAsync() + { + NpgsqlConnection.ClearAllPools(); + await using var connection = new NpgsqlConnection(_adminConnectionString); + await connection.OpenAsync(); + + await using (var terminate = connection.CreateCommand()) + { + terminate.CommandText = + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = @database AND pid <> pg_backend_pid()"; + terminate.Parameters.AddWithValue("database", _databaseName); + await terminate.ExecuteNonQueryAsync(); + } + + await using var drop = connection.CreateCommand(); + drop.CommandText = $"DROP DATABASE IF EXISTS \"{_databaseName}\""; + await drop.ExecuteNonQueryAsync(); + } + + private static void FillAuditFields(object? sender, EventArgs _) + { + if (sender is not DbContext db) + { + return; + } + + var now = DateTime.UtcNow; + foreach (var entry in db.ChangeTracker.Entries()) + { + if (entry.Entity is IAuditable auditable) + { + auditable.CreatedBy ??= "test"; + auditable.UpdatedBy ??= "test"; + if (auditable.CreatedAt == default) + { + auditable.CreatedAt = now; + } + + auditable.UpdatedAt = now; + } + + if (entry.Entity is IHasOperatorIP operatorEntity) + { + operatorEntity.OperatorIP ??= "127.0.0.1"; + } + } + } +} diff --git a/tests/BuildFlow.Monitoring.Tests/Sync/RecordingLogger.cs b/tests/BuildFlow.Monitoring.Tests/Sync/RecordingLogger.cs new file mode 100644 index 0000000..fac11f6 --- /dev/null +++ b/tests/BuildFlow.Monitoring.Tests/Sync/RecordingLogger.cs @@ -0,0 +1,25 @@ +using System.Collections.Concurrent; +using Microsoft.Extensions.Logging; + +namespace BuildFlow.Monitoring.Tests.Sync; + +/// 记录结构化日志格式化后的级别和文本，供安全断言使用。 +public sealed class RecordingLogger : ILogger +{ + private readonly ConcurrentQueue<(LogLevel Level, string Message)> _entries = new(); + + /// 当前已记录的所有日志。 + public IReadOnlyCollection<(LogLevel Level, string Message)> Entries => _entries.ToArray(); + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + => _entries.Enqueue((logLevel, formatter(state, exception))); +} diff --git a/tests/BuildFlow.Monitoring.Tests/Sync/StaticExternalConnector.cs b/tests/BuildFlow.Monitoring.Tests/Sync/StaticExternalConnector.cs new file mode 100644 index 0000000..0b2f05b --- /dev/null +++ b/tests/BuildFlow.Monitoring.Tests/Sync/StaticExternalConnector.cs @@ -0,0 +1,14 @@ +using Monitoring.Domain.Entities.ExternalSystem; +using Monitoring.Domain.Enums; +using Monitoring.Domain.Services; + +namespace BuildFlow.Monitoring.Tests.Sync; + +/// 返回固定拉取结果的 Monitoring 集成测试连接器。 +public sealed class StaticExternalConnector(FetchResult result) : IExternalConnector +{ + public ExternalSystemTypeEnum SystemType => ExternalSystemTypeEnum.GenericHttp; + + public Task FetchAsync(ExternalSystemEntity system, CancellationToken ct) + => Task.FromResult(result); +}