KArtSell.Aegis/src/KArtSell.BuildingBlocks/Reliability/DapperOutboxMessageReader.cs

using Dapper;
using KArtSell.BuildingBlocks.Data;

namespace KArtSell.BuildingBlocks.Reliability;

public sealed class DapperOutboxMessageReader(IDbConnectionFactory connectionFactory)
{
    private const string SelectUnpublishedSql = """
        select
            message_id as MessageId,
            event_type as EventType,
            schema_version as SchemaVersion,
            payload_json::text as PayloadJson,
            correlation_id as CorrelationId,
            occurred_at as OccurredAt,
            payload_hash as PayloadHash,
            attempt as Attempt
        from building_blocks.outbox_message
        where published_at is null
        order by occurred_at asc
        limit @BatchSize
        """;

    private const string MarkPublishedSql = """
        update building_blocks.outbox_message
        set published_at = @PublishedAt, attempt = attempt + 1
        where message_id = @MessageId
        """;

    private const string InsertInboxSql = """
        insert into building_blocks.inbox_message
        (consumer, message_id, received_at, payload_hash)
        values (@Consumer, @MessageId, @ReceivedAt, @PayloadHash)
        on conflict (consumer, message_id) do nothing
        """;

    public async Task<IReadOnlyList<OutboxMessageRow>> GetUnpublishedAsync(
        int batchSize,
        CancellationToken cancellationToken)
    {
        await using var connection = await connectionFactory.OpenAsync(cancellationToken);
        var messages = await connection.QueryAsync<OutboxMessageRow>(
            new CommandDefinition(
                SelectUnpublishedSql,
                new { BatchSize = batchSize },
                cancellationToken: cancellationToken));
        return messages.ToList();
    }

    public async Task MarkPublishedAsync(
        Guid messageId,
        DateTimeOffset publishedAt,
        CancellationToken cancellationToken)
    {
        await using var connection = await connectionFactory.OpenAsync(cancellationToken);
        await connection.ExecuteAsync(
            new CommandDefinition(
                MarkPublishedSql,
                new { MessageId = messageId, PublishedAt = publishedAt },
                cancellationToken: cancellationToken));
    }

    /// <summary>
    /// Insert inbox record to mark message as published/delivery-ready.
    /// Consumer parameter identifies the delivery mechanism (e.g., 'outbox-poller' = ready marker).
    /// Actual downstream consumers poll inbox_message to retrieve and deliver to end recipients.
    /// </summary>
    public async Task InsertInboxAsync(
        string consumer,
        Guid messageId,
        DateTimeOffset receivedAt,
        string payloadHash,
        CancellationToken cancellationToken)
    {
        await using var connection = await connectionFactory.OpenAsync(cancellationToken);
        await connection.ExecuteAsync(
            new CommandDefinition(
                InsertInboxSql,
                new { Consumer = consumer, MessageId = messageId, ReceivedAt = receivedAt, PayloadHash = payloadHash },
                cancellationToken: cancellationToken));
    }

    public sealed record OutboxMessageRow(
        Guid MessageId,
        string EventType,
        int SchemaVersion,
        string PayloadJson,
        string CorrelationId,
        DateTime OccurredAt,
        string PayloadHash,
        int Attempt);
}