Fix Outbox dispatched query

src/Paramore.Brighter.Outbox.MsSql/MsSqlQueries.cs

@@ -2,7 +2,7 @@
 {
     public class MsSqlQueries : IRelationDatabaseOutboxQueries
     {
-        public string PagedDispatchedCommand { get; } = "SELECT * FROM (SELECT ROW_NUMBER() OVER(ORDER BY Timestamp DESC) AS NUMBER, * FROM {0}) AS TBL WHERE DISPATCHED IS NOT NULL AND DISPATCHED < DATEADD(millisecond, -@OutStandingSince, getutcdate()) AND NUMBER BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) ORDER BY Timestamp DESC";
+        public string PagedDispatchedCommand { get; } = "SELECT * FROM (SELECT ROW_NUMBER() OVER(ORDER BY Timestamp DESC) AS NUMBER, * FROM {0}) AS TBL WHERE DISPATCHED IS NOT NULL AND DISPATCHED < DATEADD(millisecond, @OutStandingSince, getutcdate()) AND NUMBER BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) ORDER BY Timestamp DESC";
         public string PagedReadCommand { get; } = "SELECT * FROM (SELECT ROW_NUMBER() OVER(ORDER BY Timestamp DESC) AS NUMBER, * FROM {0}) AS TBL WHERE NUMBER BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) ORDER BY Timestamp DESC";
         public string PagedOutstandingCommand { get; } = "SELECT * FROM (SELECT ROW_NUMBER() OVER(ORDER BY Timestamp ASC) AS NUMBER, * FROM {0} WHERE DISPATCHED IS NULL) AS TBL WHERE TIMESTAMP < DATEADD(millisecond, -@OutStandingSince, getutcdate()) AND NUMBER BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) ORDER BY Timestamp ASC";
         public string AddCommand { get; } = "INSERT INTO {0} (MessageId, MessageType, Topic, Timestamp, CorrelationId, ReplyTo, ContentType, HeaderBag, Body) VALUES (@MessageId, @MessageType, @Topic, @Timestamp, @CorrelationId, @ReplyTo, @ContentType, @HeaderBag, @Body)";
@@ -13,6 +13,6 @@ public class MsSqlQueries : IRelationDatabaseOutboxQueries
         public string GetMessagesCommand { get; } = "SELECT * FROM {0} WHERE MessageId IN ( {1} )";
         public string DeleteMessagesCommand { get; } = "DELETE FROM {0} WHERE MessageId IN ( {1} )";
 
-        public string DispatchedCommand { get; } = "Select top(@PageSize) * FROM {0} WHERE Dispatched is not NULL and Dispatched < DATEADD(hour, (@DispatchedSince * -1), getutcdate()) Order BY Dispatched";
+        public string DispatchedCommand { get; } = "Select top(@PageSize) * FROM {0} WHERE Dispatched is not NULL and Dispatched < DATEADD(hour, @DispatchedSince, getutcdate()) Order BY Dispatched";
     }
 }

src/Paramore.Brighter.Outbox.MySql/MySqlQueries.cs

@@ -2,9 +2,9 @@
 {
     public class MySqlQueries : IRelationDatabaseOutboxQueries
     {
-        public string PagedDispatchedCommand { get; } = "SELECT * FROM {0} AS TBL WHERE `CreatedID` BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) AND DISPATCHED IS NOT NULL AND DISPATCHED < DATE_ADD(UTC_TIMESTAMP(), INTERVAL -@OutstandingSince MICROSECOND) AND NUMBER BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) ORDER BY Timestamp DESC";
+        public string PagedDispatchedCommand { get; } = "SELECT * FROM {0} AS TBL WHERE `CreatedID` BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) AND DISPATCHED IS NOT NULL AND DISPATCHED < DATE_ADD(UTC_TIMESTAMP(), INTERVAL @OutstandingSince MICROSECOND) AND NUMBER BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) ORDER BY Timestamp DESC";
         public string PagedReadCommand { get; } = "SELECT * FROM {0} AS TBL WHERE `CreatedID` BETWEEN ((@PageNumber-1)*@PageSize+1) AND (@PageNumber*@PageSize) ORDER BY Timestamp DESC";
-        public string PagedOutstandingCommand { get; } = "SELECT * FROM {0} WHERE DISPATCHED IS NULL AND Timestamp < DATE_ADD(UTC_TIMESTAMP(), INTERVAL -@OutStandingSince SECOND) ORDER BY Timestamp DESC LIMIT @PageSize OFFSET @OffsetValue";
+        public string PagedOutstandingCommand { get; } = "SELECT * FROM {0} WHERE DISPATCHED IS NULL AND Timestamp < DATE_ADD(UTC_TIMESTAMP(), INTERVAL @OutStandingSince SECOND) ORDER BY Timestamp DESC LIMIT @PageSize OFFSET @OffsetValue";
         public string AddCommand { get; } = "INSERT INTO {0} (MessageId, MessageType, Topic, Timestamp, CorrelationId, ReplyTo, ContentType, HeaderBag, Body) VALUES (@MessageId, @MessageType, @Topic, @Timestamp, @CorrelationId, @ReplyTo, @ContentType, @HeaderBag, @Body)";
         public string BulkAddCommand { get; } = "INSERT INTO {0} (MessageId, MessageType, Topic, Timestamp, CorrelationId, ReplyTo, ContentType, HeaderBag, Body) VALUES {1}";
         public string MarkDispatchedCommand { get; } = "UPDATE {0} SET Dispatched = @DispatchedAt WHERE MessageId = @MessageId";
@@ -13,6 +13,6 @@ public class MySqlQueries : IRelationDatabaseOutboxQueries
         public string GetMessagesCommand { get; } = "SELECT * FROM {0} WHERE `MessageID` IN ( {1} )";
         public string DeleteMessagesCommand { get; } = "DELETE FROM {0} WHERE MessageId IN ( {1} )";
 
-        public string DispatchedCommand { get; } = "Select * FROM {0} WHERE Dispatched is not NULL and Dispatched < DATEADD(hour, -@DispatchedSince, getutcdate()) LIMIT @PageSize Order BY Dispatched";
+        public string DispatchedCommand { get; } = "Select * FROM {0} WHERE Dispatched is not NULL and Dispatched < DATEADD(hour, @DispatchedSince, getutcdate()) LIMIT @PageSize Order BY Dispatched";
     }
 }

src/Paramore.Brighter.Outbox.Sqlite/SqliteQueries.cs

@@ -13,6 +13,6 @@ public class SqliteQueries : IRelationDatabaseOutboxQueries
         public string GetMessagesCommand { get; } = "SELECT * FROM {0} WHERE MessageId IN ( {1} )";
         public string DeleteMessagesCommand { get; } = "DELETE FROM {0} WHERE MessageId IN ( {1} )";
 
-        public string DispatchedCommand { get; } = "Select top(@PageSize) * FROM {0} WHERE Dispatched is not NULL and Dispatched < DATEADD(hour, -@DispatchedSince, getutcdate()) Order BY Dispatched";
+        public string DispatchedCommand { get; } = "Select top(@PageSize) * FROM {0} WHERE Dispatched is not NULL and Dispatched < DATEADD(hour, @DispatchedSince, getutcdate()) Order BY Dispatched";
     }
 }

src/Paramore.Brighter/OutboxArchiver.cs

@@ -79,7 +79,7 @@ public async Task ArchiveAsync(int minimumAge, CancellationToken cancellationTok
             try
             {
                 var messages = await _outboxAsync.DispatchedMessagesAsync(minimumAge, _batchSize,
-                    cancellationToken: cancellationToken);
+                    cancellationToken);
 
                 if (!messages.Any()) return;
 

tests/Paramore.Brighter.MSSQL.Tests/Outbox/When_Retrieving_Messages_To_Archive.cs

@@ -0,0 +1,55 @@
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Paramore.Brighter.Outbox.MsSql;
+using Xunit;
+
+namespace Paramore.Brighter.MSSQL.Tests.Outbox;
+
+[Trait("Category", "MSSQL")]
+public class MsSqlArchiveFetchTests : IDisposable
+{
+
+    private readonly MsSqlTestHelper _msSqlTestHelper;
+    private readonly Message _messageEarliest;
+    private readonly Message _messageDispatched;
+    private readonly Message _messageUnDispatched;
+    private readonly MsSqlOutbox _sqlOutbox;
+
+    public MsSqlArchiveFetchTests()
+    {
+        _msSqlTestHelper = new MsSqlTestHelper();
+        _msSqlTestHelper.SetupMessageDb();
+
+        _sqlOutbox = new MsSqlOutbox(_msSqlTestHelper.OutboxConfiguration);
+        _messageEarliest = new Message(new MessageHeader(Guid.NewGuid(), "test_topic", MessageType.MT_DOCUMENT),
+            new MessageBody("message body"));
+        _messageDispatched = new Message(new MessageHeader(Guid.NewGuid(), "test_topic", MessageType.MT_DOCUMENT),
+            new MessageBody("message body"));
+        _messageUnDispatched = new Message(new MessageHeader(Guid.NewGuid(), "test_topic", MessageType.MT_DOCUMENT),
+            new MessageBody("message body"));
+    }
+
+    [Fact]
+    public async Task When_Retrieving_Messages_To_Archive_Async()
+    {
+        await _sqlOutbox.AddAsync(new []{_messageEarliest, _messageDispatched, _messageUnDispatched});
+        await _sqlOutbox.MarkDispatchedAsync(_messageEarliest.Id, DateTime.UtcNow.AddHours(-3));
+        await _sqlOutbox.MarkDispatchedAsync(_messageDispatched.Id);
+
+        var allDispatched = await _sqlOutbox.DispatchedMessagesAsync(0, 100, CancellationToken.None);
+        var messagesOverAnHour = await _sqlOutbox.DispatchedMessagesAsync(1, 100, CancellationToken.None);
+        var messagesOver4Hours = await _sqlOutbox.DispatchedMessagesAsync(4, 100, CancellationToken.None);
+
+        //Assert
+        Assert.Equal(2, allDispatched.Count());
+        Assert.Single(messagesOverAnHour);
+        Assert.Empty(messagesOver4Hours);
+    }
+
+    public void Dispose()
+    {
+        _msSqlTestHelper.CleanUpDb();
+    }
+}