KArtSell.Aegis/tests/KArtSell.Integration.Tests/OutboxPollerJobTests.cs

using System.Data;
using System.Data.Common;
using Dapper;
using KArtSell.BuildingBlocks.Data;
using KArtSell.BuildingBlocks.Hashing;
using KArtSell.BuildingBlocks.Reliability;
using KArtSell.BuildingBlocks.Time;
using KArtSell.Host.Jobs;
using Microsoft.Extensions.Logging;
using Npgsql;
using Xunit;

namespace KArtSell.Integration.Tests;

public sealed class OutboxPollerJobTests : IAsyncLifetime
{
    private readonly string _connectionString;
    private readonly NpgsqlDataSource _dataSource;
    private readonly IDbConnectionFactory _connectionFactory;
    private readonly IClock _clock = new SystemClock();
    private readonly ILogger<OutboxPollerJob> _logger = new NoOpLogger();

    public OutboxPollerJobTests()
    {
        _connectionString = Environment.GetEnvironmentVariable("KARTSELL_POSTGRES")
            ?? "Host=localhost;Port=5432;Database=kartselldb;Username=kartsell;Password=kartsell4321@!";
        _dataSource = new NpgsqlDataSourceBuilder(_connectionString).Build();
        _connectionFactory = new NpgsqlConnectionFactory(_dataSource);
    }

    public async Task InitializeAsync()
    {
        // Verify database is accessible
        await using var connection = await _dataSource.OpenConnectionAsync();
        await using var cmd = connection.CreateCommand();
        cmd.CommandText = "SELECT 1";
        await cmd.ExecuteScalarAsync();
    }

    public async Task DisposeAsync()
    {
        await _dataSource.DisposeAsync();
    }

    [Fact]
    public async Task ExecuteAsync_ProcessesUnpublishedMessages_MarksAsPublished()
    {
        // Arrange: Insert unpublished message
        var messageId = Guid.NewGuid();
        var now = _clock.UtcNow;
        const string payload = """{"decision_id":"test"}""";
        var hash = ContentHasher.Sha256(payload);

        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            await connection.ExecuteAsync("""
                insert into building_blocks.outbox_message
                (message_id, event_type, schema_version, payload_json, correlation_id, occurred_at, payload_hash)
                values (@MessageId, @EventType, 1, cast(@PayloadJson as jsonb), @CorrelationId, @OccurredAt, @Hash)
                """,
                new
                {
                    MessageId = messageId,
                    EventType = "TestEvent",
                    PayloadJson = payload,
                    CorrelationId = Guid.NewGuid().ToString(),
                    OccurredAt = now.AddMinutes(-2),
                    Hash = hash
                });
        }

        var reader = new DapperOutboxMessageReader(_connectionFactory);
        var job = new OutboxPollerJob(reader, _clock, _logger);

        // Act
        await job.ExecuteAsync(CancellationToken.None);

        // Assert: Verify published_at is set AND inbox_message is created
        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            var published = await connection.QuerySingleOrDefaultAsync<DateTime?>(
                "select published_at from building_blocks.outbox_message where message_id = @MessageId",
                new { MessageId = messageId });

            Assert.NotNull(published);
            Assert.True(published > now.AddSeconds(-5).DateTime, "published_at should be recent");

            // Verify inbox_message is created
            var inboxMsg = await connection.QuerySingleOrDefaultAsync<(string Consumer, Guid MessageId)?>(
                "select consumer, message_id from building_blocks.inbox_message where message_id = @MessageId",
                new { MessageId = messageId });

            Assert.NotNull(inboxMsg);
            Assert.Equal("outbox-poller", inboxMsg!.Value.Consumer);
            Assert.Equal(messageId, inboxMsg.Value.MessageId);
        }

        // Cleanup
        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            await connection.ExecuteAsync(
                "delete from building_blocks.outbox_message where message_id = @MessageId",
                new { MessageId = messageId });
        }
    }

    [Fact]
    public async Task ExecuteAsync_SkipsMessagesBeyondCutoff_OnlyProcessesRecentMessages()
    {
        // Arrange: Insert two messages - one old, one recent
        var oldMessageId = Guid.NewGuid();
        var recentMessageId = Guid.NewGuid();
        var now = _clock.UtcNow;
        const string payload = """{"data":"test"}""";
        var hash = ContentHasher.Sha256(payload);

        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            await connection.ExecuteAsync("""
                insert into building_blocks.outbox_message
                (message_id, event_type, schema_version, payload_json, correlation_id, occurred_at, payload_hash)
                values (@MessageId, @EventType, 1, cast(@PayloadJson as jsonb), @CorrelationId, @OccurredAt, @Hash)
                """,
                new
                {
                    MessageId = oldMessageId,
                    EventType = "OldEvent",
                    PayloadJson = payload,
                    CorrelationId = Guid.NewGuid().ToString(),
                    OccurredAt = now.AddMinutes(-10),
                    Hash = hash
                });

            await connection.ExecuteAsync("""
                insert into building_blocks.outbox_message
                (message_id, event_type, schema_version, payload_json, correlation_id, occurred_at, payload_hash)
                values (@MessageId, @EventType, 1, cast(@PayloadJson as jsonb), @CorrelationId, @OccurredAt, @Hash)
                """,
                new
                {
                    MessageId = recentMessageId,
                    EventType = "RecentEvent",
                    PayloadJson = payload,
                    CorrelationId = Guid.NewGuid().ToString(),
                    OccurredAt = now.AddMinutes(-2),
                    Hash = hash
                });
        }

        var reader = new DapperOutboxMessageReader(_connectionFactory);
        var job = new OutboxPollerJob(reader, _clock, _logger);

        // Act
        await job.ExecuteAsync(CancellationToken.None);

        // Assert: Only recent message should be published (and have inbox entry)
        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            var oldPublished = await connection.QuerySingleOrDefaultAsync<DateTime?>(
                "select published_at from building_blocks.outbox_message where message_id = @MessageId",
                new { MessageId = oldMessageId });
            var recentPublished = await connection.QuerySingleOrDefaultAsync<DateTime?>(
                "select published_at from building_blocks.outbox_message where message_id = @MessageId",
                new { MessageId = recentMessageId });

            Assert.Null(oldPublished);
            Assert.NotNull(recentPublished);

            // Verify inbox entry exists ONLY for recent message
            var oldInbox = await connection.QuerySingleOrDefaultAsync<Guid?>(
                "select message_id from building_blocks.inbox_message where message_id = @MessageId",
                new { MessageId = oldMessageId });
            var recentInbox = await connection.QuerySingleOrDefaultAsync<Guid?>(
                "select message_id from building_blocks.inbox_message where message_id = @MessageId",
                new { MessageId = recentMessageId });

            Assert.Null(oldInbox);
            Assert.Equal(recentMessageId, recentInbox);
        }

        // Cleanup
        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            await connection.ExecuteAsync(
                "delete from building_blocks.outbox_message where message_id in (@OldId, @RecentId)",
                new { OldId = oldMessageId, RecentId = recentMessageId });
        }
    }

    [Fact]
    public async Task ExecuteAsync_SkipsMessagesExceedingMaxAttempts_LogsAsDeadLetter()
    {
        // Arrange: Insert message with high attempt count
        var messageId = Guid.NewGuid();
        var now = _clock.UtcNow;
        const string payload = """{"error":"too many retries"}""";
        var hash = ContentHasher.Sha256(payload);

        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            await connection.ExecuteAsync("""
                insert into building_blocks.outbox_message
                (message_id, event_type, schema_version, payload_json, correlation_id, occurred_at, payload_hash, attempt)
                values (@MessageId, @EventType, 1, cast(@PayloadJson as jsonb), @CorrelationId, @OccurredAt, @Hash, @Attempt)
                """,
                new
                {
                    MessageId = messageId,
                    EventType = "FailedEvent",
                    PayloadJson = payload,
                    CorrelationId = Guid.NewGuid().ToString(),
                    OccurredAt = now.AddMinutes(-2),
                    Hash = hash,
                    Attempt = 3
                });
        }

        var reader = new DapperOutboxMessageReader(_connectionFactory);
        var job = new OutboxPollerJob(reader, _clock, _logger);

        // Act
        await job.ExecuteAsync(CancellationToken.None);

        // Assert: Message should NOT be published (remains unpublished) and NO inbox entry
        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            var published = await connection.QuerySingleOrDefaultAsync<DateTimeOffset?>(
                "select published_at from building_blocks.outbox_message where message_id = @MessageId",
                new { MessageId = messageId });

            Assert.Null(published);

            // Verify no inbox entry for dead-lettered message
            var inboxMsg = await connection.QuerySingleOrDefaultAsync<Guid?>(
                "select message_id from building_blocks.inbox_message where message_id = @MessageId",
                new { MessageId = messageId });

            Assert.Null(inboxMsg);
        }

        // Cleanup
        await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
        {
            await connection.ExecuteAsync(
                "delete from building_blocks.outbox_message where message_id = @MessageId",
                new { MessageId = messageId });
        }
    }

    private sealed class NoOpLogger : ILogger<OutboxPollerJob>
    {
        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
        public bool IsEnabled(LogLevel logLevel) => false;
        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter) { }
    }
}

// Database factory for tests
public sealed class NpgsqlConnectionFactory : IDbConnectionFactory
{
    private readonly NpgsqlDataSource _dataSource;

    public NpgsqlConnectionFactory(NpgsqlDataSource dataSource) => _dataSource = dataSource;

    public async ValueTask<DbConnection> OpenAsync(CancellationToken cancellationToken)
        => await _dataSource.OpenConnectionAsync(cancellationToken);
}