From aa4c54041a8a990838523d2b68241fe56658a702 Mon Sep 17 00:00:00 2001
From: savorboard <yangxiaodong1214@126.com>
Date: Sun, 19 Nov 2023 22:10:16 +0800
Subject: [PATCH] Before thinking of a good solution to fixes issue #1429.

---
 src/DotNetCore.CAP.PostgreSql/IDataStorage.PostgreSql.cs | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/src/DotNetCore.CAP.PostgreSql/IDataStorage.PostgreSql.cs b/src/DotNetCore.CAP.PostgreSql/IDataStorage.PostgreSql.cs
index 7c6fd691d..c2e4398e5 100644
--- a/src/DotNetCore.CAP.PostgreSql/IDataStorage.PostgreSql.cs
+++ b/src/DotNetCore.CAP.PostgreSql/IDataStorage.PostgreSql.cs
@@ -231,11 +231,11 @@ public async Task ScheduleMessagesOfDelayedAsync(Func<object, IEnumerable<Medium
             $"SELECT \"Id\",\"Content\",\"Retries\",\"Added\",\"ExpiresAt\" FROM {_pubName} WHERE \"Version\"=@Version " +
             $"AND ((\"ExpiresAt\"< @TwoMinutesLater AND \"StatusName\" = '{StatusName.Delayed}') OR (\"ExpiresAt\"< @OneMinutesAgo AND \"StatusName\" = '{StatusName.Queued}')) FOR UPDATE SKIP LOCKED;";
 
-        object[] sqlParams =
+        var sqlParams = new object[]
         {
             new NpgsqlParameter("@Version", _capOptions.Value.Version),
             new NpgsqlParameter("@TwoMinutesLater", DateTime.Now.AddMinutes(2)),
-            new NpgsqlParameter("@OneMinutesAgo", DateTime.Now.AddMinutes(-1))
+            new NpgsqlParameter("@OneMinutesAgo", QueuedMessageFetchTime())
         };
 
         await using var connection = new NpgsqlConnection(_options.Value.ConnectionString);
@@ -269,6 +269,11 @@ public IMonitoringApi GetMonitoringApi()
         return new PostgreSqlMonitoringApi(_options, _initializer, _serializer);
     }
 
+    protected virtual DateTime QueuedMessageFetchTime()
+    {
+        return DateTime.Now.AddMinutes(-1);
+    }
+
     private async Task ChangeMessageStateAsync(string tableName, MediumMessage message, StatusName state,
         object? transaction = null)
     {