1CREATE TABLE dbo.OrderOutbox (
2    Id BIGINT IDENTITY(1,1) PRIMARY KEY,
3    EventType NVARCHAR(100) NOT NULL,
4    RoutingKey NVARCHAR(200) NOT NULL,
5    Payload NVARCHAR(MAX) NOT NULL,
6    CreatedAt DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(),
7    PublishedAt DATETIME2 NULL,
8    RetryCount INT NOT NULL DEFAULT 0,
9    LastError NVARCHAR(1000) NULL
10);
11GO
12
13CREATE TABLE dbo.Orders (
14    OrderId INT IDENTITY(1,1) PRIMARY KEY,
15    CustomerId INT NOT NULL,
16    Amount DECIMAL(12,2) NOT NULL,
17    CreatedAt DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME()
18);
19GO
20
21CREATE OR ALTER PROCEDURE dbo.CreateOrder
22    @CustomerId INT,
23    @Amount DECIMAL(12,2)
24AS
25BEGIN
26    SET NOCOUNT ON;
27
28    BEGIN TRAN;
29
30    INSERT INTO dbo.Orders (CustomerId, Amount)
31    VALUES (@CustomerId, @Amount);
32
33    DECLARE @OrderId INT = SCOPE_IDENTITY();
34
35    INSERT INTO dbo.OrderOutbox (EventType, RoutingKey, Payload)
36    VALUES (
37        N'OrderCreated',
38        N'orders.created',
39        CONCAT(
40            N'{"orderId":', @OrderId,
41            N',"customerId":', @CustomerId,
42            N',"amount":', CONVERT(NVARCHAR(50), @Amount),
43            N'}'
44        )
45    );
46
47    COMMIT TRAN;
48END;
49GO

1using System.Text;
2using Microsoft.Data.SqlClient;
3using RabbitMQ.Client;
4
5var sql = "Server=localhost;Database=SalesDb;Trusted_Connection=True;TrustServerCertificate=True";
6var factory = new ConnectionFactory { HostName = "localhost" };
7
8using var rabbitConn = factory.CreateConnection();
9using var channel = rabbitConn.CreateModel();
10channel.ExchangeDeclare("app.events", ExchangeType.Topic, durable: true);
11
12while (true)
13{
14    using var conn = new SqlConnection(sql);
15    await conn.OpenAsync();
16
17    var selectCmd = new SqlCommand(@"
18        SELECT TOP (50) Id, RoutingKey, Payload
19        FROM dbo.OrderOutbox WITH (READPAST)
20        WHERE PublishedAt IS NULL
21        ORDER BY Id", conn);
22
23    using var reader = await selectCmd.ExecuteReaderAsync();
24    var rows = new List<(long Id, string Key, string Body)>();
25
26    while (await reader.ReadAsync())
27    {
28        rows.Add((reader.GetInt64(0), reader.GetString(1), reader.GetString(2)));
29    }
30
31    foreach (var row in rows)
32    {
33        try
34        {
35            var body = Encoding.UTF8.GetBytes(row.Body);
36            var props = channel.CreateBasicProperties();
37            props.Persistent = true;
38
39            channel.BasicPublish("app.events", row.Key, props, body);
40
41            var markCmd = new SqlCommand(
42                "UPDATE dbo.OrderOutbox SET PublishedAt = SYSUTCDATETIME() WHERE Id = @id", conn);
43            markCmd.Parameters.AddWithValue("@id", row.Id);
44            await markCmd.ExecuteNonQueryAsync();
45        }
46        catch (Exception ex)
47        {
48            var retryCmd = new SqlCommand(@"
49                UPDATE dbo.OrderOutbox
50                SET RetryCount = RetryCount + 1,
51                    LastError = @err
52                WHERE Id = @id", conn);
53            retryCmd.Parameters.AddWithValue("@id", row.Id);
54            retryCmd.Parameters.AddWithValue("@err", ex.Message);
55            await retryCmd.ExecuteNonQueryAsync();
56        }
57    }
58
59    await Task.Delay(TimeSpan.FromSeconds(2));
60}