forked from xiongyuxing/tiku-backend.net
refactor: consolidate backend into modular monolith
This commit is contained in:
@@ -56,6 +56,7 @@ public sealed class ApiTestFactory(
|
||||
{
|
||||
var values = new Dictionary<string, string?>
|
||||
{
|
||||
["BackgroundProcessing:Enabled"] = "false",
|
||||
["Security:Jwt:KeyId"] = TestJwtKeys.KeyId,
|
||||
["Security:Jwt:PrivateKeyPem"] = TestJwtKeys.PrivateKeyPem,
|
||||
["Tenancy:Resolution:TenantCodePathPrefixes:0"] = "/api"
|
||||
|
||||
@@ -7,17 +7,15 @@ using Microsoft.AspNetCore.Mvc.Routing;
|
||||
using Microsoft.AspNetCore.Mvc.Controllers;
|
||||
using Microsoft.AspNetCore.Routing;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using MassTransit;
|
||||
using Tiku.Api.Security;
|
||||
using Tiku.Application.Security;
|
||||
using Tiku.Infrastructure.Messaging;
|
||||
|
||||
namespace Tiku.IntegrationTests.Api;
|
||||
|
||||
public sealed class AuthorizationManifestTests
|
||||
{
|
||||
private const int ExpectedActionCount = 397;
|
||||
private const string ExpectedSha256 = "fe0636f609e86c8c7540d84914f8254106d20194ee5bc715d616c5a4f84c7a94";
|
||||
private const int ExpectedActionCount = 399;
|
||||
private const string ExpectedSha256 = "e4460d18dbd88cb8a4293c650688f03ddaa67e69a423e70d6675c635e237c316";
|
||||
|
||||
[Fact]
|
||||
public void Controller_authorization_surface_matches_reviewed_manifest()
|
||||
@@ -71,30 +69,6 @@ public sealed class AuthorizationManifestTests
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Message_consumers_have_reviewed_authorization_and_audit_metadata()
|
||||
{
|
||||
var consumers = typeof(MessagingOptions).Assembly.GetTypes()
|
||||
.Where(type => !type.IsAbstract && type.GetInterfaces().Any(candidate =>
|
||||
candidate.IsGenericType && candidate.GetGenericTypeDefinition() == typeof(IConsumer<>)))
|
||||
.ToArray();
|
||||
|
||||
Assert.Equal(2, consumers.Length);
|
||||
foreach (var consumer in consumers)
|
||||
{
|
||||
var metadata = consumer.GetCustomAttribute<ConsumerAuthorizationMetadataAttribute>();
|
||||
Assert.NotNull(metadata);
|
||||
Assert.Contains(metadata.Realm, new[] { "tenant", "platform", "system" });
|
||||
Assert.False(string.IsNullOrWhiteSpace(metadata.Module));
|
||||
Assert.False(string.IsNullOrWhiteSpace(metadata.AuditAction));
|
||||
if (consumer.Name == "BackgroundJobRequestedConsumer")
|
||||
{
|
||||
Assert.Equal(CapabilityOperation.Write, metadata.Operation);
|
||||
Assert.True(metadata.RequiresSystemScope);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("questions", SaasFeatureCatalog.PrivateQuestionBank)]
|
||||
[InlineData("vocabulary", SaasFeatureCatalog.Vocabulary)]
|
||||
|
||||
192
Tiku.IntegrationTests/Api/MonolithBackgroundProcessingTests.cs
Normal file
192
Tiku.IntegrationTests/Api/MonolithBackgroundProcessingTests.cs
Normal file
@@ -0,0 +1,192 @@
|
||||
using System.Text.Json;
|
||||
using System.Net;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Tiku.Api.BackgroundProcessing;
|
||||
using Tiku.Application.Jobs;
|
||||
using Tiku.Domain.Operations;
|
||||
using Tiku.Domain.Tenancy;
|
||||
using Tiku.Infrastructure.Persistence;
|
||||
|
||||
namespace Tiku.IntegrationTests.Api;
|
||||
|
||||
public sealed class MonolithBackgroundProcessingTests
|
||||
{
|
||||
private static readonly IReadOnlyDictionary<string, string?> EnabledConfiguration =
|
||||
new Dictionary<string, string?>
|
||||
{
|
||||
["BackgroundProcessing:Enabled"] = "true",
|
||||
["BackgroundProcessing:JobPollSeconds"] = "1",
|
||||
["BackgroundProcessing:JobParallelism"] = "2",
|
||||
["BackgroundProcessing:JobBatchSize"] = "2",
|
||||
["TenantDomains:Enabled"] = "false",
|
||||
["SaasSubscriptions:Enabled"] = "false",
|
||||
["FeatureUsageReconciliation:Enabled"] = "false"
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public async Task Readiness_reports_only_postgres_and_redis_dependencies()
|
||||
{
|
||||
await using var factory = new ApiTestFactory();
|
||||
using var client = factory.CreateClient();
|
||||
|
||||
var response = await client.GetAsync("/api/health/ready");
|
||||
using var document = await JsonDocument.ParseAsync(await response.Content.ReadAsStreamAsync());
|
||||
|
||||
Assert.Equal(HttpStatusCode.OK, response.StatusCode);
|
||||
Assert.True(document.RootElement.TryGetProperty("database", out _));
|
||||
Assert.True(document.RootElement.TryGetProperty("redis", out _));
|
||||
Assert.False(document.RootElement.TryGetProperty("rabbitMq", out _));
|
||||
Assert.False(document.RootElement.TryGetProperty("outbox", out _));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Background_processing_registration_obeys_master_switch()
|
||||
{
|
||||
await using var disabledFactory = new ApiTestFactory();
|
||||
using var disabledClient = disabledFactory.CreateClient();
|
||||
var disabledNames = disabledFactory.Services.GetServices<IHostedService>()
|
||||
.Select(service => service.GetType().Name)
|
||||
.ToArray();
|
||||
|
||||
Assert.False(disabledFactory.Services.GetRequiredService<IOptions<BackgroundProcessingOptions>>().Value.Enabled);
|
||||
Assert.Contains("TenantDomainBackgroundService", disabledNames);
|
||||
Assert.Contains("SaasSubscriptionBackgroundService", disabledNames);
|
||||
Assert.Contains("FeatureUsageBackgroundService", disabledNames);
|
||||
Assert.Contains("BackgroundJobsBackgroundService", disabledNames);
|
||||
|
||||
await using var enabledFactory = new ApiTestFactory(configurationOverrides: EnabledConfiguration);
|
||||
using var enabledClient = enabledFactory.CreateClient();
|
||||
var enabledNames = enabledFactory.Services.GetServices<IHostedService>()
|
||||
.Select(service => service.GetType().Name)
|
||||
.ToArray();
|
||||
|
||||
Assert.True(enabledFactory.Services.GetRequiredService<IOptions<BackgroundProcessingOptions>>().Value.Enabled);
|
||||
Assert.Contains("TenantDomainBackgroundService", enabledNames);
|
||||
Assert.Contains("SaasSubscriptionBackgroundService", enabledNames);
|
||||
Assert.Contains("FeatureUsageBackgroundService", enabledNames);
|
||||
Assert.Contains("BackgroundJobsBackgroundService", enabledNames);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Api_host_processes_immediate_and_due_postgres_jobs()
|
||||
{
|
||||
await using var factory = new ApiTestFactory(configurationOverrides: EnabledConfiguration);
|
||||
using var client = factory.CreateClient();
|
||||
var tenantId = Guid.NewGuid();
|
||||
await factory.SeedAsync(new Tenant
|
||||
{
|
||||
Id = tenantId,
|
||||
Slug = tenantId.ToString("N"),
|
||||
Name = "Monolith Background Processing"
|
||||
});
|
||||
|
||||
BackgroundJobItem immediate;
|
||||
BackgroundJobItem delayed;
|
||||
using (var scope = factory.CreateSystemScope("Queue monolith background jobs"))
|
||||
{
|
||||
var jobs = scope.ServiceProvider.GetRequiredService<IBackgroundJobService>();
|
||||
immediate = await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
|
||||
tenantId,
|
||||
"statistics_aggregation",
|
||||
JsonSerializer.SerializeToElement(new { scope = "tenant" })));
|
||||
delayed = await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
|
||||
tenantId,
|
||||
"statistics_aggregation",
|
||||
JsonSerializer.SerializeToElement(new { scope = "tenant" }),
|
||||
DateTimeOffset.UtcNow.AddMinutes(5)));
|
||||
}
|
||||
|
||||
Assert.Equal(BackgroundJobStatus.Succeeded, await WaitForStatusAsync(factory, immediate.Id));
|
||||
Assert.Equal(BackgroundJobStatus.Pending, await ReadStatusAsync(factory, delayed.Id));
|
||||
|
||||
using (var scope = factory.CreateSystemScope("Make delayed monolith job due"))
|
||||
{
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
await dbContext.BackgroundJobs
|
||||
.Where(job => job.Id == delayed.Id)
|
||||
.ExecuteUpdateAsync(setters => setters.SetProperty(job => job.RunAfter, DateTimeOffset.UtcNow.AddSeconds(-1)));
|
||||
}
|
||||
|
||||
Assert.Equal(BackgroundJobStatus.Succeeded, await WaitForStatusAsync(factory, delayed.Id));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Postgres_leases_recover_expired_work_without_duplicate_claims()
|
||||
{
|
||||
await using var factory = new ApiTestFactory();
|
||||
var tenantId = Guid.NewGuid();
|
||||
await factory.SeedAsync(new Tenant
|
||||
{
|
||||
Id = tenantId,
|
||||
Slug = tenantId.ToString("N"),
|
||||
Name = "Monolith Lease Recovery"
|
||||
});
|
||||
|
||||
Guid expiredJobId;
|
||||
Guid pendingJobId;
|
||||
using (var scope = factory.CreateSystemScope("Seed lease recovery jobs"))
|
||||
{
|
||||
var jobs = scope.ServiceProvider.GetRequiredService<IBackgroundJobService>();
|
||||
expiredJobId = (await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
|
||||
tenantId,
|
||||
"statistics_aggregation",
|
||||
JsonSerializer.SerializeToElement(new { scope = "tenant" })))).Id;
|
||||
pendingJobId = (await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
|
||||
tenantId,
|
||||
"statistics_aggregation",
|
||||
JsonSerializer.SerializeToElement(new { scope = "tenant" })))).Id;
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
await dbContext.BackgroundJobs
|
||||
.Where(job => job.Id == expiredJobId)
|
||||
.ExecuteUpdateAsync(setters => setters
|
||||
.SetProperty(job => job.Status, BackgroundJobStatus.Processing)
|
||||
.SetProperty(job => job.LockedBy, "stopped-worker")
|
||||
.SetProperty(job => job.LockExpiresAt, DateTimeOffset.UtcNow.AddSeconds(-1)));
|
||||
}
|
||||
|
||||
var processed = await Task.WhenAll(
|
||||
ProcessPendingAsync(factory, "monolith-lease-a"),
|
||||
ProcessPendingAsync(factory, "monolith-lease-b"));
|
||||
|
||||
Assert.Equal(2, processed.Sum());
|
||||
Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, expiredJobId));
|
||||
Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, pendingJobId));
|
||||
}
|
||||
|
||||
private static async Task<int> ProcessPendingAsync(ApiTestFactory factory, string workerId)
|
||||
{
|
||||
using var scope = factory.CreateSystemScope($"Process jobs with {workerId}");
|
||||
return await scope.ServiceProvider.GetRequiredService<IBackgroundJobService>()
|
||||
.ProcessPendingAsync(workerId, 1);
|
||||
}
|
||||
|
||||
private static async Task<BackgroundJobStatus> WaitForStatusAsync(ApiTestFactory factory, Guid jobId)
|
||||
{
|
||||
var timeout = DateTimeOffset.UtcNow.AddSeconds(10);
|
||||
while (DateTimeOffset.UtcNow < timeout)
|
||||
{
|
||||
var status = await ReadStatusAsync(factory, jobId);
|
||||
if (status is BackgroundJobStatus.Succeeded or BackgroundJobStatus.Failed)
|
||||
{
|
||||
return status;
|
||||
}
|
||||
|
||||
await Task.Delay(100);
|
||||
}
|
||||
|
||||
return await ReadStatusAsync(factory, jobId);
|
||||
}
|
||||
|
||||
private static async Task<BackgroundJobStatus> ReadStatusAsync(ApiTestFactory factory, Guid jobId)
|
||||
{
|
||||
using var scope = factory.CreateSystemScope("Read monolith background job status");
|
||||
return await scope.ServiceProvider.GetRequiredService<TikuDbContext>()
|
||||
.BackgroundJobs.AsNoTracking()
|
||||
.Where(job => job.Id == jobId)
|
||||
.Select(job => job.Status)
|
||||
.SingleAsync();
|
||||
}
|
||||
}
|
||||
@@ -8,7 +8,7 @@ public sealed class PlatformAdminFrontendSeparationTests
|
||||
[InlineData("/platform-admin")]
|
||||
[InlineData("/platform-admin/")]
|
||||
[InlineData("/platform-admin/app.js")]
|
||||
public async Task WebApi_does_not_host_platform_admin_frontend(string path)
|
||||
public async Task Unauthenticated_platform_admin_paths_use_the_api_fallback_policy(string path)
|
||||
{
|
||||
await using var factory = new ApiTestFactory();
|
||||
using var client = factory.CreateClient(new()
|
||||
@@ -18,6 +18,6 @@ public sealed class PlatformAdminFrontendSeparationTests
|
||||
|
||||
using var response = await client.GetAsync(path);
|
||||
|
||||
Assert.Equal(HttpStatusCode.NotFound, response.StatusCode);
|
||||
Assert.Equal(HttpStatusCode.Unauthorized, response.StatusCode);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -196,7 +196,7 @@ public sealed class TenantCommerceEndpointTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Worker_processes_content_export_and_statistics_aggregation()
|
||||
public async Task Background_service_processes_content_export_and_statistics_aggregation()
|
||||
{
|
||||
await using var factory = new ApiTestFactory(paymentProviderGateway: new FakePaymentGateway());
|
||||
var tenantId = Guid.NewGuid();
|
||||
|
||||
@@ -74,7 +74,7 @@ public sealed class ArchitectureBoundaryTests
|
||||
public void Auth_sessions_are_accessed_only_through_the_session_store()
|
||||
{
|
||||
var root = FindRepositoryRoot();
|
||||
var sourceRoots = new[] { "Tiku.Api", "Tiku.Application", "Tiku.Infrastructure", "Tiku.Worker" };
|
||||
var sourceRoots = new[] { "Tiku.Api", "Tiku.Application", "Tiku.Infrastructure" };
|
||||
var allowedFiles = new[]
|
||||
{
|
||||
"TikuDbContext.cs",
|
||||
@@ -228,7 +228,7 @@ public sealed class ArchitectureBoundaryTests
|
||||
public void Production_code_does_not_reference_removed_wechat_payment_sdk()
|
||||
{
|
||||
var root = FindRepositoryRoot();
|
||||
var sourceRoots = new[] { "Tiku.Api", "Tiku.Application", "Tiku.Domain", "Tiku.Infrastructure", "Tiku.Worker" };
|
||||
var sourceRoots = new[] { "Tiku.Api", "Tiku.Application", "Tiku.Domain", "Tiku.Infrastructure" };
|
||||
var forbidden = new[]
|
||||
{
|
||||
"SKIT.FlurlHttpClient.Wechat"
|
||||
|
||||
@@ -1,419 +0,0 @@
|
||||
using MassTransit;
|
||||
using MassTransit.EntityFrameworkCoreIntegration;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using StackExchange.Redis;
|
||||
using System.Diagnostics;
|
||||
using System.Net;
|
||||
using System.Net.Http.Headers;
|
||||
using System.Net.Sockets;
|
||||
using System.Text.Json;
|
||||
using Tiku.Application;
|
||||
using Tiku.Application.Jobs;
|
||||
using Tiku.Contracts;
|
||||
using Tiku.Domain.Operations;
|
||||
using Tiku.Domain.Tenancy;
|
||||
using Tiku.Infrastructure;
|
||||
using Tiku.Infrastructure.Messaging;
|
||||
using Tiku.Infrastructure.Persistence;
|
||||
|
||||
namespace Tiku.IntegrationTests;
|
||||
|
||||
public sealed class MassTransitOutboxTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Bus_outbox_drains_after_real_broker_restart()
|
||||
{
|
||||
var rabbitMqHost = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ");
|
||||
var containerName = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_CONTAINER");
|
||||
if (string.IsNullOrWhiteSpace(rabbitMqHost) ||
|
||||
Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_RESTART") != "1" ||
|
||||
string.IsNullOrWhiteSpace(containerName))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await using var factory = CreateRabbitFactory(rabbitMqHost);
|
||||
using var client = factory.CreateClient();
|
||||
Assert.True(await WaitForReadyAsync(client), "API dependencies did not become ready before restart drill.");
|
||||
|
||||
await RunDockerAsync("stop", containerName);
|
||||
try
|
||||
{
|
||||
Assert.True(await WaitForBrokerPortClosedAsync(new Uri(rabbitMqHost)),
|
||||
"RabbitMQ AMQP port remained reachable after stopping the test container.");
|
||||
|
||||
using (var scope = factory.CreateSystemScope("Commit outbox while RabbitMQ is stopped"))
|
||||
{
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
var publisher = scope.ServiceProvider.GetRequiredService<ISecurityEventPublisher>();
|
||||
await using var transaction = await dbContext.Database.BeginTransactionAsync();
|
||||
await publisher.AuthorizationChangedAsync(
|
||||
null, null, "broker_restart_test", 3, Guid.NewGuid().ToString("N"));
|
||||
await dbContext.SaveChangesAsync();
|
||||
await transaction.CommitAsync();
|
||||
}
|
||||
|
||||
using var verification = factory.CreateSystemScope("Verify restart outbox backlog");
|
||||
var verificationDbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
Assert.NotEmpty(await verificationDbContext.Set<OutboxMessage>().ToArrayAsync());
|
||||
}
|
||||
finally
|
||||
{
|
||||
await RunDockerAsync("start", containerName);
|
||||
}
|
||||
|
||||
Assert.True(await WaitForReadyAsync(client, 240),
|
||||
"RabbitMQ did not become ready within 60 seconds after restart.");
|
||||
var drained = false;
|
||||
for (var attempt = 0; attempt < 240; attempt++)
|
||||
{
|
||||
using var verification = factory.CreateSystemScope("Wait for post-restart outbox drain");
|
||||
var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
if (!await dbContext.Set<OutboxMessage>().AnyAsync())
|
||||
{
|
||||
drained = true;
|
||||
break;
|
||||
}
|
||||
await Task.Delay(250);
|
||||
}
|
||||
Assert.True(drained, "Bus outbox did not drain within 60 seconds after RabbitMQ restart.");
|
||||
}
|
||||
|
||||
private static async Task RunDockerAsync(string operation, string containerName)
|
||||
{
|
||||
using var process = Process.Start(new ProcessStartInfo
|
||||
{
|
||||
FileName = "docker",
|
||||
ArgumentList = { operation, containerName },
|
||||
RedirectStandardOutput = true,
|
||||
RedirectStandardError = true,
|
||||
UseShellExecute = false
|
||||
}) ?? throw new InvalidOperationException("Failed to start Docker CLI for RabbitMQ restart drill.");
|
||||
var standardOutput = await process.StandardOutput.ReadToEndAsync();
|
||||
var standardError = await process.StandardError.ReadToEndAsync();
|
||||
await process.WaitForExitAsync();
|
||||
if (process.ExitCode != 0)
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"docker {operation} failed for the RabbitMQ test container: {standardError}{standardOutput}");
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task<bool> WaitForBrokerPortClosedAsync(Uri broker)
|
||||
{
|
||||
var port = broker.IsDefaultPort ? 5672 : broker.Port;
|
||||
for (var attempt = 0; attempt < 40; attempt++)
|
||||
{
|
||||
using var tcpClient = new TcpClient();
|
||||
try
|
||||
{
|
||||
await tcpClient.ConnectAsync(broker.Host, port).WaitAsync(TimeSpan.FromMilliseconds(250));
|
||||
}
|
||||
catch (Exception exception) when (exception is SocketException or TimeoutException)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
await Task.Delay(250);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Background_job_request_is_transactional_consumed_once_and_keeps_database_status_view()
|
||||
{
|
||||
var rabbitMqHost = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ");
|
||||
if (string.IsNullOrWhiteSpace(rabbitMqHost))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await using var factory = CreateRabbitFactory(rabbitMqHost);
|
||||
using var client = factory.CreateClient();
|
||||
Assert.True(await WaitForReadyAsync(client), "API dependencies did not become ready within 10 seconds.");
|
||||
|
||||
var tenantId = Guid.NewGuid();
|
||||
await factory.SeedAsync(new Tenant
|
||||
{
|
||||
Id = tenantId,
|
||||
Slug = tenantId.ToString("N"),
|
||||
Name = "RabbitMQ Background Job Tenant"
|
||||
});
|
||||
|
||||
var workerBuilder = Host.CreateApplicationBuilder();
|
||||
workerBuilder.Services.AddApplication();
|
||||
workerBuilder.Services.AddInfrastructure(factory.DatabaseConnectionString);
|
||||
workerBuilder.Services.AddReliableMessaging(CreateRabbitOptions(rabbitMqHost, configureConsumers: true));
|
||||
using var worker = workerBuilder.Build();
|
||||
await worker.StartAsync();
|
||||
|
||||
Guid rolledBackJobId;
|
||||
using (var scope = factory.CreateSystemScope("Roll back background job request"))
|
||||
{
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
var jobs = scope.ServiceProvider.GetRequiredService<IBackgroundJobService>();
|
||||
await using var transaction = await dbContext.Database.BeginTransactionAsync();
|
||||
var job = await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
|
||||
tenantId, "tenant_domain_recheck", JsonSerializer.SerializeToElement(new { })));
|
||||
rolledBackJobId = job.Id;
|
||||
await transaction.RollbackAsync();
|
||||
}
|
||||
|
||||
using (var verification = factory.CreateSystemScope("Verify rolled back background job request"))
|
||||
{
|
||||
var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
Assert.False(await dbContext.BackgroundJobs.AnyAsync(item => item.Id == rolledBackJobId));
|
||||
Assert.False(await dbContext.Set<OutboxMessage>().AnyAsync(
|
||||
item => item.MessageId == rolledBackJobId));
|
||||
}
|
||||
|
||||
BackgroundJobItem committedJob;
|
||||
using (var scope = factory.CreateSystemScope("Commit background job request"))
|
||||
{
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
var jobs = scope.ServiceProvider.GetRequiredService<IBackgroundJobService>();
|
||||
await using var transaction = await dbContext.Database.BeginTransactionAsync();
|
||||
committedJob = await jobs.EnqueueAsync(new CreateBackgroundJobCommand(
|
||||
tenantId, "tenant_domain_recheck", JsonSerializer.SerializeToElement(new { })));
|
||||
await transaction.CommitAsync();
|
||||
}
|
||||
|
||||
BackgroundJobStatus? status = null;
|
||||
for (var attempt = 0; attempt < 60; attempt++)
|
||||
{
|
||||
using var verification = factory.CreateSystemScope("Wait for RabbitMQ background job consumer");
|
||||
var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
status = await dbContext.BackgroundJobs
|
||||
.Where(item => item.Id == committedJob.Id)
|
||||
.Select(item => (BackgroundJobStatus?)item.Status)
|
||||
.SingleAsync();
|
||||
if (status == BackgroundJobStatus.Succeeded) break;
|
||||
await Task.Delay(250);
|
||||
}
|
||||
Assert.Equal(BackgroundJobStatus.Succeeded, status);
|
||||
|
||||
using (var verification = factory.CreateSystemScope("Verify background job inbox and status view"))
|
||||
{
|
||||
var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
Assert.Single(await dbContext.Set<InboxState>()
|
||||
.Where(item => item.MessageId == committedJob.Id)
|
||||
.ToArrayAsync());
|
||||
var jobs = verification.ServiceProvider.GetRequiredService<IBackgroundJobService>();
|
||||
var statusView = await jobs.ListAsync(tenantId, "tenant_domain_recheck");
|
||||
Assert.Contains(statusView, item =>
|
||||
item.Id == committedJob.Id && item.Status == BackgroundJobStatus.Succeeded);
|
||||
}
|
||||
|
||||
var managementEndpoint = ResolveRabbitManagementEndpoint(rabbitMqHost);
|
||||
if (managementEndpoint is not null)
|
||||
{
|
||||
Assert.Equal(0, await GetQueueMessageCountAsync(
|
||||
managementEndpoint, "background-job-requested_error"));
|
||||
}
|
||||
|
||||
await worker.StopAsync();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Worker_consumer_uses_inbox_and_updates_non_authoritative_redis_version()
|
||||
{
|
||||
var rabbitMqHost = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ");
|
||||
var redisConnection = Environment.GetEnvironmentVariable("TIKU_TEST_REDIS");
|
||||
if (string.IsNullOrWhiteSpace(rabbitMqHost) || string.IsNullOrWhiteSpace(redisConnection))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await using var factory = CreateRabbitFactory(rabbitMqHost);
|
||||
using var client = factory.CreateClient();
|
||||
Assert.True(await WaitForReadyAsync(client), "API dependencies did not become ready within 10 seconds.");
|
||||
|
||||
var redisEnvironment = $"consumer-{Guid.NewGuid():N}";
|
||||
var workerBuilder = Host.CreateApplicationBuilder();
|
||||
workerBuilder.Services.AddApplication();
|
||||
workerBuilder.Services.AddInfrastructure(factory.DatabaseConnectionString);
|
||||
workerBuilder.Services.AddRedisSecurity(redisConnection, redisEnvironment);
|
||||
workerBuilder.Services.AddReliableMessaging(CreateRabbitOptions(rabbitMqHost, configureConsumers: true));
|
||||
using var worker = workerBuilder.Build();
|
||||
await worker.StartAsync();
|
||||
|
||||
var tenantId = Guid.NewGuid();
|
||||
var userId = Guid.NewGuid();
|
||||
var messageId = Guid.NewGuid();
|
||||
const long version = 123456789;
|
||||
using (var scope = factory.CreateSystemScope("Publish duplicate inbox test message"))
|
||||
{
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
var publishEndpoint = scope.ServiceProvider.GetRequiredService<IPublishEndpoint>();
|
||||
await using var transaction = await dbContext.Database.BeginTransactionAsync();
|
||||
var message = new AuthorizationStateChangedV1(
|
||||
Guid.NewGuid(), tenantId, userId, "consumer_test", version,
|
||||
DateTimeOffset.UtcNow, Guid.NewGuid().ToString("N"));
|
||||
await publishEndpoint.Publish(message, context => context.MessageId = messageId);
|
||||
await publishEndpoint.Publish(message, context => context.MessageId = messageId);
|
||||
await dbContext.SaveChangesAsync();
|
||||
await transaction.CommitAsync();
|
||||
}
|
||||
|
||||
var redisKey = $"tiku:{redisEnvironment}:auth-inv:authorization:{tenantId:N}:{userId:N}";
|
||||
var consumed = false;
|
||||
for (var attempt = 0; attempt < 60; attempt++)
|
||||
{
|
||||
var multiplexer = worker.Services.GetRequiredService<IConnectionMultiplexer>();
|
||||
if (await multiplexer.GetDatabase().StringGetAsync(redisKey) == version)
|
||||
{
|
||||
consumed = true;
|
||||
break;
|
||||
}
|
||||
await Task.Delay(250);
|
||||
}
|
||||
Assert.True(consumed, "Worker did not consume the security event within 15 seconds.");
|
||||
|
||||
using (var verification = factory.CreateSystemScope("Verify duplicate consumer inbox"))
|
||||
{
|
||||
var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
Assert.Single(await dbContext.Set<InboxState>()
|
||||
.Where(item => item.MessageId == messageId)
|
||||
.ToArrayAsync());
|
||||
}
|
||||
|
||||
var managementEndpoint = ResolveRabbitManagementEndpoint(rabbitMqHost);
|
||||
if (managementEndpoint is not null)
|
||||
{
|
||||
Assert.Equal(0, await GetQueueMessageCountAsync(
|
||||
managementEndpoint, "security-state-changed_error"));
|
||||
}
|
||||
|
||||
var redis = worker.Services.GetRequiredService<IConnectionMultiplexer>();
|
||||
await redis.GetDatabase().KeyDeleteAsync(redisKey);
|
||||
await worker.StopAsync();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RabbitMq_health_and_bus_outbox_follow_database_transaction()
|
||||
{
|
||||
var rabbitMqHost = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ");
|
||||
if (string.IsNullOrWhiteSpace(rabbitMqHost))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await using var factory = CreateRabbitFactory(rabbitMqHost);
|
||||
using var client = factory.CreateClient();
|
||||
Assert.True(await WaitForReadyAsync(client), "API dependencies did not become ready within 10 seconds.");
|
||||
var ready = await client.GetAsync("/api/health/ready");
|
||||
using var readyJson = JsonDocument.Parse(await ready.Content.ReadAsStringAsync());
|
||||
|
||||
Assert.Equal(HttpStatusCode.OK, ready.StatusCode);
|
||||
Assert.True(readyJson.RootElement.GetProperty("rabbitMq").GetProperty("configured").GetBoolean());
|
||||
Assert.True(readyJson.RootElement.GetProperty("rabbitMq").GetProperty("ready").GetBoolean());
|
||||
|
||||
using (var rollbackScope = factory.CreateSystemScope("Verify rolled back bus outbox"))
|
||||
{
|
||||
var dbContext = rollbackScope.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
var publisher = rollbackScope.ServiceProvider.GetRequiredService<ISecurityEventPublisher>();
|
||||
await using var transaction = await dbContext.Database.BeginTransactionAsync();
|
||||
await publisher.AuthorizationChangedAsync(
|
||||
null, null, "rollback_test", 1, Guid.NewGuid().ToString("N"));
|
||||
await dbContext.SaveChangesAsync();
|
||||
await transaction.RollbackAsync();
|
||||
}
|
||||
|
||||
using (var verification = factory.CreateSystemScope("Verify rolled back outbox is empty"))
|
||||
{
|
||||
var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
Assert.Empty(await dbContext.Set<OutboxMessage>().ToArrayAsync());
|
||||
}
|
||||
|
||||
using (var commitScope = factory.CreateSystemScope("Verify committed bus outbox"))
|
||||
{
|
||||
var dbContext = commitScope.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
var publisher = commitScope.ServiceProvider.GetRequiredService<ISecurityEventPublisher>();
|
||||
await using var transaction = await dbContext.Database.BeginTransactionAsync();
|
||||
await publisher.AuthorizationChangedAsync(
|
||||
null, null, "commit_test", 2, Guid.NewGuid().ToString("N"));
|
||||
await dbContext.SaveChangesAsync();
|
||||
await transaction.CommitAsync();
|
||||
}
|
||||
|
||||
var drained = false;
|
||||
for (var attempt = 0; attempt < 40; attempt++)
|
||||
{
|
||||
using var verification = factory.CreateSystemScope("Wait for committed outbox delivery");
|
||||
var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
if (!await dbContext.Set<OutboxMessage>().AnyAsync())
|
||||
{
|
||||
drained = true;
|
||||
break;
|
||||
}
|
||||
await Task.Delay(250);
|
||||
}
|
||||
if (!drained)
|
||||
{
|
||||
using var diagnostics = factory.CreateSystemScope("Inspect undelivered outbox");
|
||||
var dbContext = diagnostics.ServiceProvider.GetRequiredService<TikuDbContext>();
|
||||
var messages = await dbContext.Set<OutboxMessage>().CountAsync();
|
||||
var states = await dbContext.Set<OutboxState>().CountAsync();
|
||||
Assert.Fail($"Committed MassTransit outbox was not delivered within 10 seconds. messages={messages}, states={states}");
|
||||
}
|
||||
}
|
||||
|
||||
private static Api.ApiTestFactory CreateRabbitFactory(string rabbitMqHost) =>
|
||||
new(configurationOverrides: new Dictionary<string, string?>
|
||||
{
|
||||
["RabbitMq:Host"] = rabbitMqHost,
|
||||
["RabbitMq:Username"] = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_USERNAME") ?? "guest",
|
||||
["RabbitMq:Password"] = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_PASSWORD") ?? "guest"
|
||||
});
|
||||
|
||||
private static MessagingOptions CreateRabbitOptions(string rabbitMqHost, bool configureConsumers) => new()
|
||||
{
|
||||
Host = rabbitMqHost,
|
||||
Username = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_USERNAME") ?? "guest",
|
||||
Password = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_PASSWORD") ?? "guest",
|
||||
ConfigureConsumers = configureConsumers
|
||||
};
|
||||
|
||||
private static async Task<bool> WaitForReadyAsync(HttpClient client, int attempts = 40)
|
||||
{
|
||||
for (var attempt = 0; attempt < attempts; attempt++)
|
||||
{
|
||||
if ((await client.GetAsync("/api/health/ready")).StatusCode == HttpStatusCode.OK)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
await Task.Delay(250);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private static Uri? ResolveRabbitManagementEndpoint(string rabbitMqHost)
|
||||
{
|
||||
var configured = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_MANAGEMENT");
|
||||
if (!string.IsNullOrWhiteSpace(configured))
|
||||
{
|
||||
return new Uri(configured);
|
||||
}
|
||||
|
||||
var broker = new Uri(rabbitMqHost);
|
||||
return broker.IsLoopback ? new Uri($"http://{broker.Host}:15672") : null;
|
||||
}
|
||||
|
||||
private static async Task<int> GetQueueMessageCountAsync(Uri managementEndpoint, string queueName)
|
||||
{
|
||||
using var client = new HttpClient { BaseAddress = managementEndpoint };
|
||||
var username = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_USERNAME") ?? "guest";
|
||||
var password = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_PASSWORD") ?? "guest";
|
||||
var credentials = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes($"{username}:{password}"));
|
||||
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", credentials);
|
||||
using var response = await client.GetAsync($"/api/queues/%2F/{Uri.EscapeDataString(queueName)}");
|
||||
if (response.StatusCode == HttpStatusCode.NotFound)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
response.EnsureSuccessStatusCode();
|
||||
using var document = JsonDocument.Parse(await response.Content.ReadAsStringAsync());
|
||||
return document.RootElement.GetProperty("messages").GetInt32();
|
||||
}
|
||||
}
|
||||
@@ -13,7 +13,6 @@
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Tiku.Api\Tiku.Api.csproj" />
|
||||
<ProjectReference Include="..\Tiku.Worker\Tiku.Worker.csproj" />
|
||||
<ProjectReference Include="..\Tiku.Infrastructure\Tiku.Infrastructure.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
|
||||
Reference in New Issue
Block a user