diff --git a/src/KArtSell.Host/Jobs/DownstreamConsumerJob.cs b/src/KArtSell.Host/Jobs/DownstreamConsumerJob.cs index 0da4c921..c8819150 100644 --- a/src/KArtSell.Host/Jobs/DownstreamConsumerJob.cs +++ b/src/KArtSell.Host/Jobs/DownstreamConsumerJob.cs @@ -3,6 +3,7 @@ using Hangfire; using KArtSell.BuildingBlocks.Data; using KArtSell.BuildingBlocks.Time; using KArtSell.Host.Consumers; +using KArtSell.Modules.IdentityAccess.ManageIdentityAndRoles.Events; using KArtSell.Modules.ModelOperations.ShadowRun.Events; using Microsoft.Extensions.Logging; using System.Text.Json; @@ -19,6 +20,9 @@ public sealed class DownstreamConsumerJob( ShadowRunCompletedConsumer shadowRunConsumer, ApprovalQueueConsumer approvalQueueConsumer, AuditLogConsumer auditLogConsumer, + IdentityCreatedConsumer identityCreatedConsumer, + IdentityAuditConsumer identityAuditConsumer, + MfaReminderJob mfaReminderJob, IClock clock, ILogger logger) { @@ -107,6 +111,23 @@ public sealed class DownstreamConsumerJob( processedCount++; break; + case "IdentityCreated": + var identityEvent = JsonSerializer.Deserialize(payloadJson) + ?? throw new InvalidOperationException($"Failed to deserialize {eventType} payload for {messageId}"); + + // Route to identity consumers and MFA reminder job + await identityCreatedConsumer.HandleAsync(identityEvent, cancellationToken); + await identityAuditConsumer.HandleAsync(identityEvent, cancellationToken); + + // Schedule MFA reminder for 24 hours later (via Hangfire) + BackgroundJob.Schedule( + () => mfaReminderJob.ExecuteAsync(identityEvent, cancellationToken), + TimeSpan.FromHours(24)); + + LogMessageProcessed(logger, messageId, eventType, null); + processedCount++; + break; + case "TestEvent": case "OldEvent": case "RecentEvent": diff --git a/tests/KArtSell.IdentityAccess.IntegrationTests/Features/RegisterIdentityE2ETests.cs b/tests/KArtSell.IdentityAccess.IntegrationTests/Features/RegisterIdentityE2ETests.cs new file mode 100644 index 00000000..bddd21db --- /dev/null +++ b/tests/KArtSell.IdentityAccess.IntegrationTests/Features/RegisterIdentityE2ETests.cs @@ -0,0 +1,367 @@ +using Xunit; +using Dapper; +using Npgsql; +using KArtSell.Modules.IdentityAccess.ManageIdentityAndRoles.Features.RegisterIdentity; +using KArtSell.BuildingBlocks.Reliability; +using KArtSell.BuildingBlocks.Data; +using System.Data; +using System.Text.Json; + +namespace KArtSell.IdentityAccess.IntegrationTests.Features; + +[Collection("Database")] +public class RegisterIdentityE2ETests : IAsyncLifetime +{ + private readonly string _connectionString; + private NpgsqlDataSource _dataSource = null!; + private RegisterIdentitySql _sql = null!; + private IOutboxWriter _outboxWriter = null!; + private IDbConnectionFactory _connectionFactory = null!; + + public RegisterIdentityE2ETests() + { + _connectionString = Environment.GetEnvironmentVariable("KARTSELL_POSTGRES") + ?? "Host=127.0.0.1;Port=5432;Database=kartselldb;Username=postgres;Password=postgres"; + } + + public async Task InitializeAsync() + { + _dataSource = new NpgsqlDataSourceBuilder(_connectionString).Build(); + _sql = new RegisterIdentitySql(async () => await _dataSource.OpenConnectionAsync()); + _outboxWriter = new DapperOutboxWriter(); + _connectionFactory = new NpgsqlConnectionFactory(_dataSource); + + await CleanupAsync(); + } + + public async Task DisposeAsync() + { + await CleanupAsync(); + await _dataSource.DisposeAsync(); + } + + private async Task CleanupAsync() + { + using var conn = await _dataSource.OpenConnectionAsync(); + await conn.ExecuteAsync("DELETE FROM public.identity WHERE email LIKE 'test-e2e-%'"); + await conn.ExecuteAsync("DELETE FROM building_blocks.outbox_message WHERE event_type = 'IdentityCreated'"); + await conn.ExecuteAsync("DELETE FROM building_blocks.inbox_message WHERE event_type = 'IdentityCreated'"); + await conn.ExecuteAsync("DELETE FROM public.identity_audit_log WHERE email LIKE 'test-e2e-%'"); + await conn.ExecuteAsync("DELETE FROM public.identity_mfa_reminder"); + } + + [Fact] + public async Task RegisterIdentity_E2E_CreatesIdentityWritesOutboxAndTriggersConsumers() + { + // Arrange + var email = "test-e2e-001@example.com"; + var displayName = "Test E2E 001"; + var correlationId = Guid.NewGuid().ToString(); + var identityId = Guid.NewGuid(); + + // Act: Create identity with outbox write + var conn = await _dataSource.OpenConnectionAsync() as NpgsqlConnection + ?? throw new InvalidOperationException("Failed to open connection"); + + await using (conn) + { + await using var transaction = await conn.BeginTransactionAsync(IsolationLevel.ReadCommitted); + + var createdId = await _sql.CreateIdentityAsync(conn, transaction, identityId, email, displayName, correlationId, CancellationToken.None); + + var identityCreatedEvent = new KArtSell.Modules.IdentityAccess.ManageIdentityAndRoles.Events.IdentityCreated + { + IdentityId = createdId, + Email = email, + DisplayName = displayName, + CorrelationId = correlationId, + OccurredAt = DateTime.UtcNow + }; + + var payloadJson = JsonSerializer.Serialize(identityCreatedEvent); + var outboxMessage = new OutboxMessage( + MessageId: Guid.NewGuid(), + EventType: "IdentityCreated", + SchemaVersion: 1, + PayloadJson: payloadJson, + CorrelationId: correlationId, + OccurredAt: DateTimeOffset.UtcNow, + PayloadHash: ComputeSha256(payloadJson)); + + await _outboxWriter.AddAsync(conn, transaction, outboxMessage, CancellationToken.None); + await transaction.CommitAsync(); + + // Assert: Verify identity was created + var (returnedId, returnedEmail, _, returnedState) = await _sql.GetIdentityAsync(createdId, CancellationToken.None); + Assert.Equal(createdId, returnedId); + Assert.Equal(email, returnedEmail); + Assert.Equal("ACTIVE", returnedState); + + // Assert: Verify outbox message was written + var outboxCount = await conn.QuerySingleAsync( + "SELECT COUNT(1) FROM building_blocks.outbox_message WHERE correlation_id = @correlationId", + new { correlationId }); + Assert.Equal(1, outboxCount); + } + } + + [Fact] + public async Task RegisterIdentity_E2E_OutboxPollerMarksInboxAndTriggersConsumers() + { + // Arrange: Create identity with outbox + var email = "test-e2e-002@example.com"; + var displayName = "Test E2E 002"; + var correlationId = Guid.NewGuid().ToString(); + var identityId = Guid.NewGuid(); + var messageId = Guid.NewGuid(); + + var conn = await _dataSource.OpenConnectionAsync() as NpgsqlConnection + ?? throw new InvalidOperationException("Failed to open connection"); + + await using (conn) + { + await using var transaction = await conn.BeginTransactionAsync(IsolationLevel.ReadCommitted); + + var createdId = await _sql.CreateIdentityAsync(conn, transaction, identityId, email, displayName, correlationId, CancellationToken.None); + + var identityCreatedEvent = new KArtSell.Modules.IdentityAccess.ManageIdentityAndRoles.Events.IdentityCreated + { + IdentityId = createdId, + Email = email, + DisplayName = displayName, + CorrelationId = correlationId, + OccurredAt = DateTime.UtcNow + }; + + var payloadJson = JsonSerializer.Serialize(identityCreatedEvent); + var outboxMessage = new OutboxMessage( + MessageId: messageId, + EventType: "IdentityCreated", + SchemaVersion: 1, + PayloadJson: payloadJson, + CorrelationId: correlationId, + OccurredAt: DateTimeOffset.UtcNow, + PayloadHash: ComputeSha256(payloadJson)); + + await _outboxWriter.AddAsync(conn, transaction, outboxMessage, CancellationToken.None); + await transaction.CommitAsync(); + + // Act: Manually insert inbox record (simulating OutboxPollerJob) + const string inboxSql = """ + INSERT INTO building_blocks.inbox_message (message_id, event_type, payload_json, received_at, consumer) + VALUES (@MessageId, 'IdentityCreated', @PayloadJson, NOW(), 'outbox-poller') + """; + + await conn.ExecuteAsync(inboxSql, new { messageId, payloadJson }); + + // Assert: Verify inbox message was created + var inboxCount = await conn.QuerySingleAsync( + "SELECT COUNT(1) FROM building_blocks.inbox_message WHERE message_id = @messageId", + new { messageId }); + Assert.Equal(1, inboxCount); + + // Assert: Verify we can read the inbox message + var inboxMessage = await conn.QuerySingleOrDefaultAsync<(Guid MessageId, string PayloadJson)>( + """ + SELECT message_id, payload_json + FROM building_blocks.inbox_message + WHERE message_id = @MessageId + """, + new { messageId }); + + Assert.NotEqual(default, inboxMessage); + Assert.Equal(payloadJson, inboxMessage.PayloadJson); + } + } + + [Fact] + public async Task RegisterIdentity_E2E_FullFlowCreatesAuditAndMfaRecords() + { + // Arrange: Create identity with outbox in transaction + var email = "test-e2e-003@example.com"; + var displayName = "Test E2E 003"; + var correlationId = Guid.NewGuid().ToString(); + var identityId = Guid.NewGuid(); + + var conn = await _dataSource.OpenConnectionAsync() as NpgsqlConnection + ?? throw new InvalidOperationException("Failed to open connection"); + + await using (conn) + { + // Step 1: Create identity + write to outbox (transactional) + await using var transaction = await conn.BeginTransactionAsync(IsolationLevel.ReadCommitted); + + var createdId = await _sql.CreateIdentityAsync(conn, transaction, identityId, email, displayName, correlationId, CancellationToken.None); + + var identityCreatedEvent = new KArtSell.Modules.IdentityAccess.ManageIdentityAndRoles.Events.IdentityCreated + { + IdentityId = createdId, + Email = email, + DisplayName = displayName, + CorrelationId = correlationId, + OccurredAt = DateTime.UtcNow + }; + + var payloadJson = JsonSerializer.Serialize(identityCreatedEvent); + var outboxMessage = new OutboxMessage( + MessageId: Guid.NewGuid(), + EventType: "IdentityCreated", + SchemaVersion: 1, + PayloadJson: payloadJson, + CorrelationId: correlationId, + OccurredAt: DateTimeOffset.UtcNow, + PayloadHash: ComputeSha256(payloadJson)); + + await _outboxWriter.AddAsync(conn, transaction, outboxMessage, CancellationToken.None); + await transaction.CommitAsync(); + + // Step 2: Simulate DownstreamConsumerJob reading outbox → inbox + const string inboxSql = """ + INSERT INTO building_blocks.inbox_message (message_id, event_type, payload_json, received_at, consumer) + SELECT message_id, event_type, payload_json, NOW(), 'outbox-poller' + FROM building_blocks.outbox_message + WHERE event_type = 'IdentityCreated' AND correlation_id = @correlationId + """; + + await conn.ExecuteAsync(inboxSql, new { correlationId }); + + // Step 3: Simulate DownstreamConsumerJob calling consumers + // Write audit log (simulating IdentityAuditConsumer) + const string auditSql = """ + INSERT INTO public.identity_audit_log (identity_id, action, email, display_name, correlation_id, occurred_at) + VALUES (@identityId, 'CREATED', @email, @displayName, @correlationId, @occurredAt) + """; + + await conn.ExecuteAsync(auditSql, new + { + identityId = createdId, + email, + displayName, + correlationId, + occurredAt = identityCreatedEvent.OccurredAt + }); + + // Write MFA reminder tracking (simulating MfaReminderJob) + const string mfaSql = """ + INSERT INTO public.identity_mfa_reminder (identity_id, sent_at) + VALUES (@identityId, NOW()) + ON CONFLICT (identity_id) DO NOTHING + """; + + await conn.ExecuteAsync(mfaSql, new { identityId = createdId }); + + // Assert: Verify complete E2E flow + var identity = await _sql.GetIdentityAsync(createdId, CancellationToken.None); + Assert.Equal(email, identity.Email); + + var outboxCount = await conn.QuerySingleAsync( + "SELECT COUNT(1) FROM building_blocks.outbox_message WHERE correlation_id = @correlationId", + new { correlationId }); + Assert.Equal(1, outboxCount); + + var inboxCount = await conn.QuerySingleAsync( + "SELECT COUNT(1) FROM building_blocks.inbox_message WHERE event_type = 'IdentityCreated'"); + Assert.True(inboxCount > 0); + + var auditCount = await conn.QuerySingleAsync( + "SELECT COUNT(1) FROM public.identity_audit_log WHERE identity_id = @identityId AND action = 'CREATED'", + new { identityId = createdId }); + Assert.Equal(1, auditCount); + + var mfaCount = await conn.QuerySingleAsync( + "SELECT COUNT(1) FROM public.identity_mfa_reminder WHERE identity_id = @identityId", + new { identityId = createdId }); + Assert.Equal(1, mfaCount); + } + } + + [Fact] + public async Task RegisterIdentity_E2E_MfaReminderIsIdempotent() + { + // Arrange + var email = "test-e2e-004@example.com"; + var displayName = "Test E2E 004"; + var correlationId = Guid.NewGuid().ToString(); + var identityId = Guid.NewGuid(); + + var conn = await _dataSource.OpenConnectionAsync() as NpgsqlConnection + ?? throw new InvalidOperationException("Failed to open connection"); + + await using (conn) + { + // Create identity first + await using var transaction = await conn.BeginTransactionAsync(IsolationLevel.ReadCommitted); + var createdId = await _sql.CreateIdentityAsync(conn, transaction, identityId, email, displayName, correlationId, CancellationToken.None); + await transaction.CommitAsync(); + + // Act: Record MFA reminder twice (should be idempotent) + const string mfaSql = """ + INSERT INTO public.identity_mfa_reminder (identity_id, sent_at) + VALUES (@identityId, NOW()) + ON CONFLICT (identity_id) DO NOTHING + """; + + await conn.ExecuteAsync(mfaSql, new { identityId = createdId }); + await conn.ExecuteAsync(mfaSql, new { identityId = createdId }); + + // Assert: Only one record exists + var mfaCount = await conn.QuerySingleAsync( + "SELECT COUNT(1) FROM public.identity_mfa_reminder WHERE identity_id = @identityId", + new { identityId = createdId }); + Assert.Equal(1, mfaCount); + } + } + + [Fact] + public async Task RegisterIdentity_E2E_AuditLogIsImmutable() + { + // Arrange + var email = "test-e2e-005@example.com"; + var displayName = "Test E2E 005"; + var correlationId = Guid.NewGuid().ToString(); + var identityId = Guid.NewGuid(); + + var conn = await _dataSource.OpenConnectionAsync() as NpgsqlConnection + ?? throw new InvalidOperationException("Failed to open connection"); + + await using (conn) + { + // Create identity + await using var transaction = await conn.BeginTransactionAsync(IsolationLevel.ReadCommitted); + var createdId = await _sql.CreateIdentityAsync(conn, transaction, identityId, email, displayName, correlationId, CancellationToken.None); + await transaction.CommitAsync(); + + // Insert audit record + const string auditSql = """ + INSERT INTO public.identity_audit_log (identity_id, action, email, display_name, correlation_id, occurred_at) + VALUES (@identityId, 'CREATED', @email, @displayName, @correlationId, NOW()) + """; + + await conn.ExecuteAsync(auditSql, new + { + identityId = createdId, + email, + displayName, + correlationId + }); + + // Act: Try to update audit record (should fail due to trigger) + const string updateAuditSql = """ + UPDATE public.identity_audit_log SET action = 'MODIFIED' WHERE identity_id = @identityId + """; + + var ex = await Assert.ThrowsAsync(async () => + await conn.ExecuteAsync(updateAuditSql, new { identityId = createdId })); + + // Assert: Exception should mention immutability + Assert.Contains("immutable", ex.Message, StringComparison.OrdinalIgnoreCase); + } + } + + private static string ComputeSha256(string input) + { + using var hasher = System.Security.Cryptography.SHA256.Create(); + var hash = hasher.ComputeHash(System.Text.Encoding.UTF8.GetBytes(input)); + return Convert.ToHexString(hash); + } +}