From be66b04b1a4e683c7bc1226fab5c37507a1d2157 Mon Sep 17 00:00:00 2001 From: argxh Date: Mon, 2 Feb 2026 13:19:21 +0100 Subject: [PATCH 1/3] feature inserted: - add support for prefix tables - add unit test fix: - fix RemoveFromList: removed limit 1 --- src/Hangfire.PostgreSql/CountersAggregator.cs | 17 +- src/Hangfire.PostgreSql/ExpirationManager.cs | 14 +- .../Hangfire.PostgreSql.csproj | 8 +- .../PostgreSqlConnection.cs | 69 +- .../PostgreSqlDistributedLock.cs | 28 +- .../PostgreSqlFetchedJob.cs | 14 +- src/Hangfire.PostgreSql/PostgreSqlJobQueue.cs | 16 +- .../PostgreSqlJobQueueMonitoringApi.cs | 16 +- .../PostgreSqlMonitoringApi.cs | 50 +- .../PostgreSqlObjectsInstaller.cs | 77 +- src/Hangfire.PostgreSql/PostgreSqlStorage.cs | 2 +- .../PostgreSqlStorageOptions.cs | 4 + .../PostgreSqlWriteOnlyTransaction.cs | 58 +- src/Hangfire.PostgreSql/Utils/TryExecute.cs | 6 + .../PostgreSqlInstallerFacts.cs | 20 + .../PostgreSqlTablePrefixFacts.cs | 1027 +++++++++++++++++ .../Hangfire.PostgreSql.Tests/Utils/Helper.cs | 26 +- 17 files changed, 1332 insertions(+), 120 deletions(-) create mode 100644 tests/Hangfire.PostgreSql.Tests/PostgreSqlTablePrefixFacts.cs diff --git a/src/Hangfire.PostgreSql/CountersAggregator.cs b/src/Hangfire.PostgreSql/CountersAggregator.cs index 31f4844b..2c8cb7cf 100644 --- a/src/Hangfire.PostgreSql/CountersAggregator.cs +++ b/src/Hangfire.PostgreSql/CountersAggregator.cs @@ -25,6 +25,7 @@ using Hangfire.Common; using Hangfire.Logging; using Hangfire.Server; +using Utility = Hangfire.PostgreSql.Utils.Utils; namespace Hangfire.PostgreSql { @@ -81,27 +82,33 @@ public void Execute(CancellationToken cancellationToken) private string GetAggregationQuery() { string schemaName = _storage.Options.SchemaName; + return $""" BEGIN; - INSERT INTO "{schemaName}"."aggregatedcounter" ("key", "value", "expireat") + INSERT INTO "{schemaName}"."{TableNameHandler("aggregatedcounter")}" ("key", "value", "expireat") SELECT "key", SUM("value"), MAX("expireat") - FROM "{schemaName}"."counter" + FROM "{schemaName}"."{TableNameHandler("counter")}" GROUP BY "key" ON CONFLICT("key") DO UPDATE - SET "value" = "aggregatedcounter"."value" + EXCLUDED."value", "expireat" = EXCLUDED."expireat"; + SET "value" = "{schemaName}"."{TableNameHandler("aggregatedcounter")}"."value" + EXCLUDED."value", "expireat" = EXCLUDED."expireat"; - DELETE FROM "{schemaName}"."counter" + DELETE FROM "{schemaName}"."{TableNameHandler("counter")}" WHERE "key" IN ( - SELECT "key" FROM "{schemaName}"."aggregatedcounter" + SELECT "key" FROM "{schemaName}"."{TableNameHandler("aggregatedcounter")}" ); COMMIT; """; } + + private string TableNameHandler(string baseName) + { + return Utility.GetTableName(baseName, _storage.Options.UseTablePrefix, _storage.Options.TablePrefixName); + } } } diff --git a/src/Hangfire.PostgreSql/ExpirationManager.cs b/src/Hangfire.PostgreSql/ExpirationManager.cs index 541d1320..0aa3095d 100644 --- a/src/Hangfire.PostgreSql/ExpirationManager.cs +++ b/src/Hangfire.PostgreSql/ExpirationManager.cs @@ -27,6 +27,7 @@ using Hangfire.Logging; using Hangfire.Server; using Hangfire.Storage; +using Utility = Hangfire.PostgreSql.Utils.Utils; namespace Hangfire.PostgreSql { @@ -82,10 +83,10 @@ public void Execute(CancellationToken cancellationToken) { using IDbTransaction transaction = connection.BeginTransaction(); removedCount = connection.Execute($@" - DELETE FROM ""{_storage.Options.SchemaName}"".""{table}"" + DELETE FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler(table)}"" WHERE ""id"" IN ( SELECT ""id"" - FROM ""{_storage.Options.SchemaName}"".""{table}"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler(table)}"" WHERE ""expireat"" < NOW() LIMIT {_storage.Options.DeleteExpiredBatchSize.ToString(CultureInfo.InvariantCulture)} )", transaction: transaction); @@ -129,7 +130,7 @@ private void AggregateCounter(string counterName) using IDbTransaction transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted); string aggregateQuery = $@" WITH ""counters"" AS ( - DELETE FROM ""{_storage.Options.SchemaName}"".""counter"" + DELETE FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("counter")}"" WHERE ""key"" = @Key AND ""expireat"" IS NULL RETURNING * @@ -143,7 +144,7 @@ DELETE FROM ""{_storage.Options.SchemaName}"".""counter"" if (aggregatedValue > 0) { - string insertQuery = $@"INSERT INTO ""{_storage.Options.SchemaName}"".""counter""(""key"", ""value"") VALUES (@Key, @Value);"; + string insertQuery = $@"INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("counter")}""(""key"", ""value"") VALUES (@Key, @Value);"; connection.Execute(insertQuery, new { Key = counterName, Value = aggregatedValue }); } }); @@ -176,5 +177,10 @@ private void UseConnectionDistributedLock(PostgreSqlStorage storage, ActionPostgreSql storage implementation for Hangfire (background job system for ASP.NET and aspnet core applications). Copyright © 2014-2024 Frank Hommers and others Hangfire PostgreSql Storage - 1.9.4 + 1.9.5-tablePrefix Frank Hommers, Vytautas Kasparavičius, Žygimantas Arūna netstandard2.0 Hangfire.PostgreSql @@ -14,9 +14,9 @@ https://github.com/frankhommers/Hangfire.PostgreSql/releases http://hmm.rs/Hangfire.PostgreSql - 1.9.4.0 - 1.9.4.0 - 1.9.4.0 + 1.9.5-tablePrefix + 1.9.5-tablePrefix + 1.9.5-tablePrefix True LICENSE.md https://github.com/frankhommers/Hangfire.PostgreSql diff --git a/src/Hangfire.PostgreSql/PostgreSqlConnection.cs b/src/Hangfire.PostgreSql/PostgreSqlConnection.cs index 3c0114d0..83927fbf 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlConnection.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlConnection.cs @@ -34,6 +34,8 @@ using Hangfire.Storage; using Npgsql; using IsolationLevel = System.Transactions.IsolationLevel; +using Utility = Hangfire.PostgreSql.Utils.Utils; + namespace Hangfire.PostgreSql { @@ -114,7 +116,7 @@ public override string CreateExpiredJob( } string createJobSql = $@" - INSERT INTO ""{_options.SchemaName}"".""job"" (""invocationdata"", ""arguments"", ""createdat"", ""expireat"") + INSERT INTO ""{_options.SchemaName}"".""{TableNameHandler("job")}"" (""invocationdata"", ""arguments"", ""createdat"", ""expireat"") VALUES (@InvocationData, @Arguments, @CreatedAt, @ExpireAt) RETURNING ""id""; "; @@ -144,7 +146,7 @@ public override string CreateExpiredJob( } string insertParameterSql = $@" - INSERT INTO ""{_options.SchemaName}"".""jobparameter"" (""jobid"", ""name"", ""value"") + INSERT INTO ""{_options.SchemaName}"".""{TableNameHandler("jobparameter")}"" (""jobid"", ""name"", ""value"") VALUES (@JobId, @Name, @Value); "; @@ -164,7 +166,7 @@ public override JobData GetJobData(string id) string sql = $@" SELECT ""invocationdata"" ""invocationData"", ""statename"" ""stateName"", ""arguments"", ""createdat"" ""createdAt"" - FROM ""{_options.SchemaName}"".""job"" + FROM ""{_options.SchemaName}"".""{TableNameHandler("job")}"" WHERE ""id"" = @Id; "; @@ -211,8 +213,8 @@ public override StateData GetStateData(string jobId) string sql = $@" SELECT s.""name"" ""Name"", s.""reason"" ""Reason"", s.""data"" ""Data"" - FROM ""{_options.SchemaName}"".""state"" s - INNER JOIN ""{_options.SchemaName}"".""job"" j on j.""stateid"" = s.""id"" + FROM ""{_options.SchemaName}"".""{TableNameHandler("state")}"" s + INNER JOIN ""{_options.SchemaName}"".""{TableNameHandler("job")}"" j on j.""stateid"" = s.""id"" WHERE j.""id"" = @JobId; "; @@ -245,14 +247,14 @@ public override void SetJobParameter(string id, string name, string value) WITH ""inputvalues"" AS ( SELECT @JobId ""jobid"", @Name ""name"", @Value ""value"" ), ""updatedrows"" AS ( - UPDATE ""{_options.SchemaName}"".""jobparameter"" ""updatetarget"" + UPDATE ""{_options.SchemaName}"".""{TableNameHandler("jobparameter")}"" ""updatetarget"" SET ""value"" = ""inputvalues"".""value"" FROM ""inputvalues"" WHERE ""updatetarget"".""jobid"" = ""inputvalues"".""jobid"" AND ""updatetarget"".""name"" = ""inputvalues"".""name"" RETURNING ""updatetarget"".""jobid"", ""updatetarget"".""name"" ) - INSERT INTO ""{_options.SchemaName}"".""jobparameter""(""jobid"", ""name"", ""value"") + INSERT INTO ""{_options.SchemaName}"".""{TableNameHandler("jobparameter")}""(""jobid"", ""name"", ""value"") SELECT ""jobid"", ""name"", ""value"" FROM ""inputvalues"" ""insertvalues"" WHERE NOT EXISTS ( @@ -279,7 +281,7 @@ public override string GetJobParameter(string id, string name) throw new ArgumentNullException(nameof(name)); } - string query = $@"SELECT ""value"" FROM ""{_options.SchemaName}"".""jobparameter"" WHERE ""jobid"" = @Id AND ""name"" = @Name;"; + string query = $@"SELECT ""value"" FROM ""{_options.SchemaName}"".""{TableNameHandler("jobparameter")}"" WHERE ""jobid"" = @Id AND ""name"" = @Name;"; return _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingleOrDefault(query, new { Id = Convert.ToInt64(id, CultureInfo.InvariantCulture), Name = name })); @@ -292,7 +294,7 @@ public override HashSet GetAllItemsFromSet(string key) throw new ArgumentNullException(nameof(key)); } - string query = $@"SELECT ""value"" FROM ""{_options.SchemaName}"".""set"" WHERE ""key"" = @Key;"; + string query = $@"SELECT ""value"" FROM ""{_options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = @Key;"; return _storage.UseConnection(_dedicatedConnection, connection => { IEnumerable result = connection.Query(query, new { Key = key }); @@ -316,7 +318,7 @@ public override string GetFirstByLowestScoreFromSet(string key, double fromScore return _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingleOrDefault($@" SELECT ""value"" - FROM ""{_options.SchemaName}"".""set"" + FROM ""{_options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = @Key AND ""score"" BETWEEN @FromScore AND @ToScore ORDER BY ""score"" LIMIT 1; @@ -344,7 +346,7 @@ public override List GetFirstByLowestScoreFromSet(string key, double fro return _storage.UseConnection(_dedicatedConnection, connection => connection .Query($@" SELECT ""value"" - FROM ""{_options.SchemaName}"".""set"" + FROM ""{_options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = @Key AND ""score"" BETWEEN @FromScore AND @ToScore ORDER BY ""score"" LIMIT @Limit; @@ -369,14 +371,14 @@ public override void SetRangeInHash(string key, IEnumerable GetAllEntriesFromHash(string key) Dictionary result = _storage.UseConnection(_dedicatedConnection, connection => connection .Query($@" SELECT ""field"" ""Field"", ""value"" ""Value"" - FROM ""{_options.SchemaName}"".""hash"" + FROM ""{_options.SchemaName}"".""{TableNameHandler("hash")}"" WHERE ""key"" = @Key;", new { Key = key }) .ToDictionary(x => x.Field, x => x.Value)); @@ -455,13 +457,13 @@ public override void AnnounceServer(string serverId, ServerContext context) WITH ""inputvalues"" AS ( SELECT @Id ""id"", @Data ""data"", NOW() ""lastheartbeat"" ), ""updatedrows"" AS ( - UPDATE ""{_options.SchemaName}"".""server"" ""updatetarget"" + UPDATE ""{_options.SchemaName}"".""{TableNameHandler("server")}"" ""updatetarget"" SET ""data"" = ""inputvalues"".""data"", ""lastheartbeat"" = ""inputvalues"".""lastheartbeat"" FROM ""inputvalues"" WHERE ""updatetarget"".""id"" = ""inputvalues"".""id"" RETURNING ""updatetarget"".""id"" ) - INSERT INTO ""{_options.SchemaName}"".""server""(""id"", ""data"", ""lastheartbeat"") + INSERT INTO ""{_options.SchemaName}"".""{TableNameHandler("server")}""(""id"", ""data"", ""lastheartbeat"") SELECT ""id"", ""data"", ""lastheartbeat"" FROM ""inputvalues"" ""insertvalues"" WHERE NOT EXISTS ( @@ -483,7 +485,7 @@ public override void RemoveServer(string serverId) } _storage.UseConnection(_dedicatedConnection, connection => connection - .Execute($@"DELETE FROM ""{_options.SchemaName}"".""server"" WHERE ""id"" = @Id;", new { Id = serverId })); + .Execute($@"DELETE FROM ""{_options.SchemaName}"".""{TableNameHandler("server")}"" WHERE ""id"" = @Id;", new { Id = serverId })); } public override void Heartbeat(string serverId) @@ -494,7 +496,7 @@ public override void Heartbeat(string serverId) } string query = $@" - UPDATE ""{_options.SchemaName}"".""server"" + UPDATE ""{_options.SchemaName}"".""{TableNameHandler("server")}"" SET ""lastheartbeat"" = NOW() WHERE ""id"" = @Id; "; @@ -516,7 +518,7 @@ public override int RemoveTimedOutServers(TimeSpan timeOut) } string query = $@" - DELETE FROM ""{_options.SchemaName}"".""server"" + DELETE FROM ""{_options.SchemaName}"".""{TableNameHandler("server")}"" WHERE ""lastheartbeat"" < (NOW() - INTERVAL '{((long)timeOut.TotalMilliseconds).ToString(CultureInfo.InvariantCulture)} MILLISECONDS');"; return _storage.UseConnection(_dedicatedConnection, connection => connection.Execute(query)); @@ -529,7 +531,7 @@ public override long GetSetCount(string key) throw new ArgumentNullException(nameof(key)); } - string query = $@"SELECT COUNT(""key"") FROM ""{_options.SchemaName}"".""set"" WHERE ""key"" = @Key"; + string query = $@"SELECT COUNT(""key"") FROM ""{_options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = @Key"; return _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingleOrDefault(query, new { Key = key })); @@ -542,7 +544,7 @@ public override List GetAllItemsFromList(string key) throw new ArgumentNullException(nameof(key)); } - string query = $@"SELECT ""value"" FROM ""{_options.SchemaName}"".""list"" WHERE ""key"" = @Key ORDER BY ""id"" DESC"; + string query = $@"SELECT ""value"" FROM ""{_options.SchemaName}"".""{TableNameHandler("list")}"" WHERE ""key"" = @Key ORDER BY ""id"" DESC"; return _storage.UseConnection(_dedicatedConnection, connection => connection .Query(query, new { Key = key }) @@ -557,9 +559,9 @@ public override long GetCounter(string key) } string query = $@"SELECT SUM(""value"") FROM ( - SELECT SUM(""value"") ""value"" FROM ""{_options.SchemaName}"".""counter"" WHERE ""key"" = @Key + SELECT SUM(""value"") ""value"" FROM ""{_options.SchemaName}"".""{TableNameHandler("counter")}"" WHERE ""key"" = @Key UNION ALL - SELECT SUM(""value"") ""value"" FROM ""{_options.SchemaName}"".""aggregatedcounter"" WHERE ""key"" = @Key) c"; + SELECT SUM(""value"") ""value"" FROM ""{_options.SchemaName}"".""{TableNameHandler("aggregatedcounter")}"" WHERE ""key"" = @Key) c"; return _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingleOrDefault(query, new { Key = key }) ?? 0); @@ -572,7 +574,7 @@ public override long GetListCount(string key) throw new ArgumentNullException(nameof(key)); } - string query = $@"SELECT COUNT(""id"") FROM ""{_options.SchemaName}"".""list"" WHERE ""key"" = @Key"; + string query = $@"SELECT COUNT(""id"") FROM ""{_options.SchemaName}"".""{TableNameHandler("list")}"" WHERE ""key"" = @Key"; return _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingleOrDefault(query, new { Key = key })); @@ -586,7 +588,7 @@ public override TimeSpan GetListTtl(string key) throw new ArgumentNullException(nameof(key)); } - string query = $@"SELECT min(""expireat"") FROM ""{_options.SchemaName}"".""list"" WHERE ""key"" = @Key"; + string query = $@"SELECT min(""expireat"") FROM ""{_options.SchemaName}"".""{TableNameHandler("list")}"" WHERE ""key"" = @Key"; DateTime? result = _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingleOrDefault(query, new { Key = key })); @@ -603,7 +605,7 @@ public override List GetRangeFromList(string key, int startingFrom, int string query = $@" SELECT ""value"" - FROM ""{_options.SchemaName}"".""list"" + FROM ""{_options.SchemaName}"".""{TableNameHandler("list")}"" WHERE ""key"" = @Key ORDER BY ""id"" DESC LIMIT @Limit OFFSET @Offset @@ -622,7 +624,7 @@ public override long GetHashCount(string key) throw new ArgumentNullException(nameof(key)); } - string query = $@"SELECT COUNT(""id"") FROM ""{_options.SchemaName}"".""hash"" WHERE ""key"" = @Key"; + string query = $@"SELECT COUNT(""id"") FROM ""{_options.SchemaName}"".""{TableNameHandler("hash")}"" WHERE ""key"" = @Key"; return _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingle(query, new { Key = key })); @@ -635,7 +637,7 @@ public override TimeSpan GetHashTtl(string key) throw new ArgumentNullException(nameof(key)); } - string query = $@"SELECT MIN(""expireat"") FROM ""{_options.SchemaName}"".""hash"" WHERE ""key"" = @Key"; + string query = $@"SELECT MIN(""expireat"") FROM ""{_options.SchemaName}"".""{TableNameHandler("hash")}"" WHERE ""key"" = @Key"; DateTime? result = _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingleOrDefault(query, new { Key = key })); @@ -652,7 +654,7 @@ public override List GetRangeFromSet(string key, int startingFrom, int e string query = $@" SELECT ""value"" - FROM ""{_options.SchemaName}"".""set"" + FROM ""{_options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = @Key ORDER BY ""id"" LIMIT @Limit OFFSET @Offset @@ -670,7 +672,7 @@ public override TimeSpan GetSetTtl(string key) throw new ArgumentNullException(nameof(key)); } - string query = $@"SELECT min(""expireat"") FROM ""{_options.SchemaName}"".""set"" WHERE ""key"" = @Key"; + string query = $@"SELECT min(""expireat"") FROM ""{_options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = @Key"; DateTime? result = _storage.UseConnection(_dedicatedConnection, connection => connection .QuerySingleOrDefault(query, new { Key = key })); @@ -690,7 +692,7 @@ public override string GetValueFromHash(string key, string name) throw new ArgumentNullException(nameof(name)); } - string query = $@"SELECT ""value"" FROM ""{_options.SchemaName}"".""hash"" WHERE ""key"" = @Key AND ""field"" = @Field"; + string query = $@"SELECT ""value"" FROM ""{_options.SchemaName}"".""{TableNameHandler("hash")}"" WHERE ""key"" = @Key AND ""field"" = @Field"; return _storage.UseConnection(_dedicatedConnection, connection => connection .Query(query, new { Key = key, Field = name }) @@ -761,6 +763,11 @@ private void ReleaseLock(string resource, Guid lockId, bool onDisposing) } } + private string TableNameHandler(string baseName) + { + return Utility.GetTableName(baseName, _options.UseTablePrefix, _options.TablePrefixName); + } + private class DisposableLock : IDisposable { private readonly PostgreSqlConnection _connection; diff --git a/src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs b/src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs index de2e7c6e..3216795c 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs @@ -29,6 +29,7 @@ using Hangfire.Logging; using Npgsql; using IsolationLevel = System.Data.IsolationLevel; +using Utility = Hangfire.PostgreSql.Utils.Utils; namespace Hangfire.PostgreSql { @@ -102,7 +103,7 @@ public static void Lock(string resource, TimeSpan timeout, IDbConnection connect Stopwatch lockAcquiringTime = Stopwatch.StartNew(); bool tryAcquireLock = true; - Func tryLock = options.UseNativeDatabaseTransactions + Func tryLock = options.UseNativeDatabaseTransactions ? TransactionLockHandler.TryLock : UpdateCountLockHandler.TryLock; @@ -117,7 +118,7 @@ public static void Lock(string resource, TimeSpan timeout, IDbConnection connect try { - if (tryLock(connection, options.SchemaName, resource)) + if (tryLock(connection, options, resource)) { return; } @@ -166,7 +167,7 @@ public static bool TryRemoveLock(string resource, IDbConnection connection, Post DateTime timeout = onlyExpired ? DateTime.UtcNow - options.DistributedLockTimeout : DateTime.MaxValue; - int rowsAffected = connection.Execute($@"DELETE FROM ""{options.SchemaName}"".""lock"" WHERE ""resource"" = @Resource AND ""acquired"" < @Timeout", + int rowsAffected = connection.Execute($@"DELETE FROM ""{options.SchemaName}"".""{TableNameHandler("lock", options)}"" WHERE ""resource"" = @Resource AND ""acquired"" < @Timeout", new { Resource = resource, Timeout = timeout, @@ -190,18 +191,19 @@ public static bool TryRemoveLock(string resource, IDbConnection connection, Post private static class TransactionLockHandler { - public static bool TryLock(IDbConnection connection, string schemaName, string resource) + public static bool TryLock(IDbConnection connection, PostgreSqlStorageOptions options, string resource) { IDbTransaction trx = null; try { + string schemaName = options.SchemaName; trx = BeginTransactionIfNotPresent(connection); int rowsAffected = connection.Execute($@" INSERT INTO ""{schemaName}"".""lock""(""resource"", ""acquired"") SELECT @Resource, @Acquired WHERE NOT EXISTS ( - SELECT 1 FROM ""{schemaName}"".""lock"" + SELECT 1 FROM ""{schemaName}"".""{TableNameHandler("lock", options)}"" WHERE ""resource"" = @Resource ) ON CONFLICT DO NOTHING; @@ -231,13 +233,16 @@ public static IDbTransaction BeginTransactionIfNotPresent(IDbConnection connecti private static class UpdateCountLockHandler { - public static bool TryLock(IDbConnection connection, string schemaName, string resource) + public static bool TryLock(IDbConnection connection, PostgreSqlStorageOptions options, string resource) { + + string schemaName = options.SchemaName; + connection.Execute($@" - INSERT INTO ""{schemaName}"".""lock""(""resource"", ""updatecount"", ""acquired"") + INSERT INTO ""{schemaName}"".""{TableNameHandler("lock", options)}""(""resource"", ""updatecount"", ""acquired"") SELECT @Resource, 0, @Acquired WHERE NOT EXISTS ( - SELECT 1 FROM ""{schemaName}"".""lock"" + SELECT 1 FROM ""{schemaName}"".""{TableNameHandler("lock", options)}"" WHERE ""resource"" = @Resource ) ON CONFLICT DO NOTHING; @@ -247,11 +252,16 @@ SELECT 1 FROM ""{schemaName}"".""lock"" }); int rowsAffected = connection.Execute( - $@"UPDATE ""{schemaName}"".""lock"" SET ""updatecount"" = 1 WHERE ""updatecount"" = 0 AND ""resource"" = @Resource", + $@"UPDATE ""{schemaName}"".""{TableNameHandler("lock", options)}"" SET ""updatecount"" = 1 WHERE ""updatecount"" = 0 AND ""resource"" = @Resource", new { Resource = resource }); return rowsAffected > 0; } } + + private static string TableNameHandler(string baseName, PostgreSqlStorageOptions options) + { + return Utility.GetTableName(baseName, options.UseTablePrefix, options.TablePrefixName); + } } } diff --git a/src/Hangfire.PostgreSql/PostgreSqlFetchedJob.cs b/src/Hangfire.PostgreSql/PostgreSqlFetchedJob.cs index 8528a1b6..0f4f43b3 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlFetchedJob.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlFetchedJob.cs @@ -1,4 +1,4 @@ -// This file is part of Hangfire.PostgreSql. +// This file is part of Hangfire.PostgreSql. // Copyright © 2014 Frank Hommers . // // Hangfire.PostgreSql is free software: you can redistribute it and/or modify @@ -25,6 +25,7 @@ using Hangfire.Logging; using Hangfire.PostgreSql.Utils; using Hangfire.Storage; +using Utility = Hangfire.PostgreSql.Utils.Utils; namespace Hangfire.PostgreSql { @@ -78,7 +79,7 @@ public void RemoveFromQueue() } _storage.UseConnection(null, connection => connection.Execute($@" - DELETE FROM ""{_storage.Options.SchemaName}"".""jobqueue"" WHERE ""id"" = @Id AND ""fetchedat"" = @FetchedAt; + DELETE FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" WHERE ""id"" = @Id AND ""fetchedat"" = @FetchedAt; ", new { Id, FetchedAt })); @@ -96,7 +97,7 @@ public void Requeue() } _storage.UseConnection(null, connection => connection.Execute($@" - UPDATE ""{_storage.Options.SchemaName}"".""jobqueue"" + UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" SET ""fetchedat"" = NULL WHERE ""id"" = @Id AND ""fetchedat"" = @FetchedAt; ", @@ -157,7 +158,7 @@ internal void ExecuteKeepAliveQueryIfRequired() } string updateFetchAtSql = $@" - UPDATE ""{_storage.Options.SchemaName}"".""jobqueue"" + UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" SET ""fetchedat"" = NOW() WHERE ""id"" = @id AND ""fetchedat"" = @fetchedAt RETURNING ""fetchedat"" AS ""FetchedAt""; @@ -186,5 +187,10 @@ internal void ExecuteKeepAliveQueryIfRequired() } } } + + private string TableNameHandler(string baseName) + { + return Utility.GetTableName(baseName, _storage.Options.UseTablePrefix, _storage.Options.TablePrefixName); + } } } diff --git a/src/Hangfire.PostgreSql/PostgreSqlJobQueue.cs b/src/Hangfire.PostgreSql/PostgreSqlJobQueue.cs index 89aa7244..47fd7718 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlJobQueue.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlJobQueue.cs @@ -30,6 +30,7 @@ using Hangfire.PostgreSql.Utils; using Hangfire.Storage; using Npgsql; +using Utility = Hangfire.PostgreSql.Utils.Utils; namespace Hangfire.PostgreSql { @@ -94,7 +95,7 @@ public IFetchedJob Dequeue(string[] queues, CancellationToken cancellationToken) public void Enqueue(IDbConnection connection, string queue, string jobId) { string enqueueJobSql = $@" - INSERT INTO ""{_storage.Options.SchemaName}"".""jobqueue"" (""jobid"", ""queue"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" (""jobid"", ""queue"") VALUES (@JobId, @Queue); "; @@ -133,11 +134,11 @@ internal IFetchedJob Dequeue_Transaction(string[] queues, CancellationToken canc FetchedJob fetchedJob; string fetchJobSql = $@" - UPDATE ""{_storage.Options.SchemaName}"".""jobqueue"" + UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" SET ""fetchedat"" = NOW() WHERE ""id"" = ( SELECT ""id"" - FROM ""{_storage.Options.SchemaName}"".""jobqueue"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" WHERE ""queue"" = ANY (@Queues) AND (""fetchedat"" IS NULL OR ""fetchedat"" < NOW() + INTERVAL '{timeoutSeconds.ToString(CultureInfo.InvariantCulture)} SECONDS') ORDER BY ""fetchedat"" NULLS FIRST, ""queue"", ""jobid"" @@ -217,7 +218,7 @@ internal IFetchedJob Dequeue_UpdateCount(string[] queues, CancellationToken canc string jobToFetchSql = $@" SELECT ""id"" AS ""Id"", ""jobid"" AS ""JobId"", ""queue"" AS ""Queue"", ""fetchedat"" AS ""FetchedAt"", ""updatecount"" AS ""UpdateCount"" - FROM ""{_storage.Options.SchemaName}"".""jobqueue"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" WHERE ""queue"" = ANY (@Queues) AND (""fetchedat"" IS NULL OR ""fetchedat"" < NOW() + INTERVAL '{timeoutSeconds.ToString(CultureInfo.InvariantCulture)} SECONDS') ORDER BY ""fetchedat"" NULLS FIRST, ""queue"", ""jobid"" @@ -225,7 +226,7 @@ internal IFetchedJob Dequeue_UpdateCount(string[] queues, CancellationToken canc "; string markJobAsFetchedSql = $@" - UPDATE ""{_storage.Options.SchemaName}"".""jobqueue"" + UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" SET ""fetchedat"" = NOW(), ""updatecount"" = (""updatecount"" + 1) % 2000000000 WHERE ""id"" = @Id @@ -317,6 +318,11 @@ private Task ListenForNotificationsAsync(CancellationToken cancellationToken) } } + private string TableNameHandler(string baseName) + { + return Utility.GetTableName(baseName, _storage.Options.UseTablePrefix, _storage.Options.TablePrefixName); + } + [UsedImplicitly(ImplicitUseTargetFlags.WithMembers)] private class FetchedJob { diff --git a/src/Hangfire.PostgreSql/PostgreSqlJobQueueMonitoringApi.cs b/src/Hangfire.PostgreSql/PostgreSqlJobQueueMonitoringApi.cs index 4ec1be6b..4860061a 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlJobQueueMonitoringApi.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlJobQueueMonitoringApi.cs @@ -23,6 +23,7 @@ using System.Collections.Generic; using System.Linq; using Dapper; +using Utility = Hangfire.PostgreSql.Utils.Utils; namespace Hangfire.PostgreSql { @@ -37,7 +38,7 @@ public PostgreSqlJobQueueMonitoringApi(PostgreSqlStorage storage) public IEnumerable GetQueues() { - string sqlQuery = $@"SELECT DISTINCT ""queue"" FROM ""{_storage.Options.SchemaName}"".""jobqueue"""; + string sqlQuery = $@"SELECT DISTINCT ""queue"" FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"""; return _storage.UseConnection(null, connection => connection.Query(sqlQuery).ToList()); } @@ -56,13 +57,13 @@ public EnqueuedAndFetchedCountDto GetEnqueuedAndFetchedCount(string queue) string sqlQuery = $@" SELECT ( SELECT COUNT(*) - FROM ""{_storage.Options.SchemaName}"".""jobqueue"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" WHERE ""fetchedat"" IS NULL AND ""queue"" = @Queue ) ""EnqueuedCount"", ( SELECT COUNT(*) - FROM ""{_storage.Options.SchemaName}"".""jobqueue"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" WHERE ""fetchedat"" IS NOT NULL AND ""queue"" = @Queue ) ""FetchedCount""; @@ -81,8 +82,8 @@ private IEnumerable GetQueuedOrFetchedJobIds(string queue, bool fetched, i { string sqlQuery = $@" SELECT j.""id"" - FROM ""{_storage.Options.SchemaName}"".""jobqueue"" jq - LEFT JOIN ""{_storage.Options.SchemaName}"".""job"" j ON jq.""jobid"" = j.""id"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" jq + LEFT JOIN ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" j ON jq.""jobid"" = j.""id"" WHERE jq.""queue"" = @Queue AND jq.""fetchedat"" {(fetched ? "IS NOT NULL" : "IS NULL")} AND j.""id"" IS NOT NULL @@ -94,5 +95,10 @@ AND j.""id"" IS NOT NULL new { Queue = queue, Offset = from, Limit = perPage }) .ToList()); } + + private string TableNameHandler(string baseName) + { + return Utility.GetTableName(baseName, _storage.Options.UseTablePrefix, _storage.Options.TablePrefixName); + } } } diff --git a/src/Hangfire.PostgreSql/PostgreSqlMonitoringApi.cs b/src/Hangfire.PostgreSql/PostgreSqlMonitoringApi.cs index 63a5e4c6..b3e17585 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlMonitoringApi.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlMonitoringApi.cs @@ -30,6 +30,7 @@ using Hangfire.States; using Hangfire.Storage; using Hangfire.Storage.Monitoring; +using Utility = Hangfire.PostgreSql.Utils.Utils; namespace Hangfire.PostgreSql { @@ -112,7 +113,7 @@ public IDictionary FailedByDatesCount() public IList Servers() { return UseConnection(connection => { - List<(Entities.Server Server, ServerData Data)> servers = connection.Query($@"SELECT * FROM ""{_storage.Options.SchemaName}"".""server""") + List<(Entities.Server Server, ServerData Data)> servers = connection.Query($@"SELECT * FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}""") .AsEnumerable() .Select(server => (server, SerializationHelper.Deserialize(server.Data))) .ToList(); @@ -227,15 +228,15 @@ public JobDetailsDto JobDetails(string jobId) return UseConnection(connection => { string sql = $@" SELECT ""id"" ""Id"", ""invocationdata"" ""InvocationData"", ""arguments"" ""Arguments"", ""createdat"" ""CreatedAt"", ""expireat"" ""ExpireAt"" - FROM ""{_storage.Options.SchemaName}"".""job"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" WHERE ""id"" = @Id; SELECT ""jobid"" ""JobId"", ""name"" ""Name"", ""value"" ""Value"" - FROM ""{_storage.Options.SchemaName}"".""jobparameter"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobparameter")}"" WHERE ""jobid"" = @Id; SELECT ""jobid"" ""JobId"", ""name"" ""Name"", ""reason"" ""Reason"", ""createdat"" ""CreatedAt"", ""data"" ""Data"" - FROM ""{_storage.Options.SchemaName}"".""state"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("state")}"" WHERE ""jobid"" = @Id ORDER BY ""id"" DESC; "; @@ -284,33 +285,33 @@ public StatisticsDto GetStatistics() return UseConnection(connection => { string sql = $@" SELECT ""statename"" ""State"", COUNT(""id"") ""Count"" - FROM ""{_storage.Options.SchemaName}"".""job"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" WHERE ""statename"" IS NOT NULL GROUP BY ""statename""; SELECT COUNT(*) - FROM ""{_storage.Options.SchemaName}"".""server""; + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("server")}""; SELECT SUM(""value"") FROM (SELECT SUM(""value"") AS value - FROM ""{_storage.Options.SchemaName}"".""counter"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("counter")}"" WHERE ""key"" = 'stats:succeeded' UNION ALL SELECT SUM(""value"") AS value - FROM ""{_storage.Options.SchemaName}"".""aggregatedcounter"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("aggregatedcounter")}"" WHERE ""key"" = 'stats:succeeded') c; SELECT SUM(""value"") FROM (SELECT SUM(""value"") AS value - FROM ""{_storage.Options.SchemaName}"".""counter"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("counter")}"" WHERE ""key"" = 'stats:deleted' UNION ALL SELECT SUM(""value"") AS value - FROM ""{_storage.Options.SchemaName}"".""aggregatedcounter"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("aggregatedcounter")}"" WHERE ""key"" = 'stats:deleted') c; SELECT COUNT(*) - FROM ""{_storage.Options.SchemaName}"".""set"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = 'recurring-jobs'; "; @@ -383,11 +384,11 @@ private Dictionary GetTimelineStats(IDictionary EnqueuedJobs(IEnumerable jobIds) string enqueuedJobsSql = $@" SELECT ""j"".""id"" ""Id"", ""j"".""invocationdata"" ""InvocationData"", ""j"".""arguments"" ""Arguments"", ""j"".""createdat"" ""CreatedAt"", ""j"".""expireat"" ""ExpireAt"", ""s"".""name"" ""StateName"", ""s"".""reason"" ""StateReason"", ""s"".""data"" ""StateData"" - FROM ""{_storage.Options.SchemaName}"".""job"" ""j"" - LEFT JOIN ""{_storage.Options.SchemaName}"".""state"" ""s"" ON ""s"".""id"" = ""j"".""stateid"" - LEFT JOIN ""{_storage.Options.SchemaName}"".""jobqueue"" ""jq"" ON ""jq"".""jobid"" = ""j"".""id"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" ""j"" + LEFT JOIN ""{_storage.Options.SchemaName}"".""{TableNameHandler("state")}"" ""s"" ON ""s"".""id"" = ""j"".""stateid"" + LEFT JOIN ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" ""jq"" ON ""jq"".""jobid"" = ""j"".""id"" WHERE ""j"".""id"" = ANY (@JobIds) AND ""jq"".""fetchedat"" IS NULL; "; @@ -458,7 +459,7 @@ private JobList EnqueuedJobs(IEnumerable jobIds) private long GetNumberOfJobsByStateName(string stateName) { - string sqlQuery = $@"SELECT COUNT(""id"") FROM ""{_storage.Options.SchemaName}"".""job"" WHERE ""statename"" = @StateName;"; + string sqlQuery = $@"SELECT COUNT(""id"") FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" WHERE ""statename"" = @StateName;"; return UseConnection(connection => connection.QuerySingle(sqlQuery, new { StateName = stateName })); @@ -484,8 +485,8 @@ private JobList GetJobs(int from, int count, string stateName, Func< string jobsSql = $@" SELECT ""j"".""id"" ""Id"", ""j"".""invocationdata"" ""InvocationData"", ""j"".""arguments"" ""Arguments"", ""j"".""createdat"" ""CreatedAt"", ""j"".""expireat"" ""ExpireAt"", NULL ""FetchedAt"", ""j"".""statename"" ""StateName"", ""s"".""reason"" ""StateReason"", ""s"".""data"" ""StateData"" - FROM ""{_storage.Options.SchemaName}"".""job"" ""j"" - LEFT JOIN ""{_storage.Options.SchemaName}"".""state"" ""s"" ON ""j"".""stateid"" = ""s"".""id"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" ""j"" + LEFT JOIN ""{_storage.Options.SchemaName}"".""{TableNameHandler("state")}"" ""s"" ON ""j"".""stateid"" = ""s"".""id"" WHERE ""j"".""statename"" = @StateName ORDER BY ""j"".""id"" DESC LIMIT @Limit OFFSET @Offset; @@ -531,9 +532,9 @@ private JobList FetchedJobs( SELECT ""j"".""id"" ""Id"", ""j"".""invocationdata"" ""InvocationData"", ""j"".""arguments"" ""Arguments"", ""j"".""createdat"" ""CreatedAt"", ""j"".""expireat"" ""ExpireAt"", ""jq"".""fetchedat"" ""FetchedAt"", ""j"".""statename"" ""StateName"", ""s"".""reason"" ""StateReason"", ""s"".""data"" ""StateData"" - FROM ""{_storage.Options.SchemaName}"".""job"" ""j"" - LEFT JOIN ""{_storage.Options.SchemaName}"".""state"" ""s"" ON ""j"".""stateid"" = ""s"".""id"" - LEFT JOIN ""{_storage.Options.SchemaName}"".""jobqueue"" ""jq"" ON ""jq"".""jobid"" = ""j"".""id"" + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" ""j"" + LEFT JOIN ""{_storage.Options.SchemaName}"".""{TableNameHandler("state")}"" ""s"" ON ""j"".""stateid"" = ""s"".""id"" + LEFT JOIN ""{_storage.Options.SchemaName}"".""{TableNameHandler("jobqueue")}"" ""jq"" ON ""jq"".""jobid"" = ""j"".""id"" WHERE ""j"".""id"" = ANY (@JobIds) AND ""jq"".""fetchedat"" IS NOT NULL; "; @@ -556,6 +557,11 @@ private T UseConnection(Func func) return _storage.UseConnection(null, func); } + private string TableNameHandler(string baseName) + { + return Utility.GetTableName(baseName, _storage.Options.UseTablePrefix, _storage.Options.TablePrefixName); + } + /// /// Overloaded dictionary that doesn't throw if given an invalid key /// Fixes issues such as https://github.com/frankhommers/Hangfire.PostgreSql/issues/79 diff --git a/src/Hangfire.PostgreSql/PostgreSqlObjectsInstaller.cs b/src/Hangfire.PostgreSql/PostgreSqlObjectsInstaller.cs index 00f586f8..1077ad92 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlObjectsInstaller.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlObjectsInstaller.cs @@ -20,6 +20,7 @@ // Special thanks goes to him. using System; +using System.Collections.Generic; using System.Data; using System.Globalization; using System.IO; @@ -34,7 +35,7 @@ public static class PostgreSqlObjectsInstaller { private static readonly ILog _logger = LogProvider.GetLogger(typeof(PostgreSqlStorage)); - public static void Install(NpgsqlConnection connection, string schemaName = "hangfire") + public static void Install(NpgsqlConnection connection, string schemaName = "hangfire", bool useTablePrefix = false, string tablePrefixName = "hangfire_") { if (connection == null) { @@ -102,6 +103,80 @@ public static void Install(NpgsqlConnection connection, string schemaName = "han version++; } while (true); + if(!useTablePrefix) + { + _logger.Warn("Hangfire SQL objects installed."); + return; + } + + _logger.Info($"Try to change table name with prefix {tablePrefixName}"); + + try + { + _logger.Info("Check if prefix change has already been recorded.."); + string prefixChangeKey = "hangfire:table_prefix_changed"; + using (var checkCmd = new NpgsqlCommand( + $@"SELECT 1 FROM ""{schemaName}"".""set"" WHERE ""key"" = @Key LIMIT 1;", connection)) + { + checkCmd.Parameters.AddWithValue("Key", prefixChangeKey); + var exists = checkCmd.ExecuteScalar(); + if (exists != null && exists != DBNull.Value) + { + _logger.Warn("Table prefix has already been changed. If you want to change it again, you must drop all Hangfire tables and reinstall them."); + return; + } + } + + var hangfireTableNames = new[] + { + "job", "state", "counter", "queue", "schema", "server", "set", "list", "hash", "lock", "aggregatedcounter", "jobparameter", "jobqueue" + }; + + List tablesToRename = new List(); + using (var cmd = new NpgsqlCommand($@" + SELECT table_name + FROM information_schema.tables + WHERE table_schema = @SchemaName + AND table_name = ANY(@TableNames)", connection)) + { + cmd.Parameters.AddWithValue("SchemaName", schemaName); + cmd.Parameters.AddWithValue("TableNames", hangfireTableNames); + using (var reader = cmd.ExecuteReader()) + { + while (reader.Read()) + { + tablesToRename.Add(reader.GetString(0)); + } + } + } + + foreach (string tableName in tablesToRename) + { + string newTableName = tablePrefixName + tableName; + using (var renameCmd = new NpgsqlCommand( + $@"ALTER TABLE ""{schemaName}"".""{tableName}"" RENAME TO ""{newTableName}"";", connection)) + { + renameCmd.ExecuteNonQuery(); + _logger.Info($"Table {tableName} renamed to {newTableName}"); + } + } + + using (var insertCmd = new NpgsqlCommand( + $@"INSERT INTO ""{schemaName}"".""{tablePrefixName + "set"}"" (""key"", ""value"", ""score"") VALUES (@Key, @Value, @Score);", connection)) + { + insertCmd.Parameters.AddWithValue("Key", prefixChangeKey); + insertCmd.Parameters.AddWithValue("Value", tablePrefixName); + insertCmd.Parameters.AddWithValue("Score", 0); // oppure un valore appropriato + insertCmd.ExecuteNonQuery(); + _logger.Info($"Table prefix change persisted in set table with key '{prefixChangeKey}' and value '{tablePrefixName}'"); + } + } + catch (Exception ex) + { + _logger.ErrorException("Error while changing table name with prefix", ex); + throw; + } + _logger.Info("Hangfire SQL objects installed."); } diff --git a/src/Hangfire.PostgreSql/PostgreSqlStorage.cs b/src/Hangfire.PostgreSql/PostgreSqlStorage.cs index f8f2969b..a661b08c 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlStorage.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlStorage.cs @@ -297,7 +297,7 @@ private void PerformSingleInitializationAttempt() NpgsqlConnection connection = CreateAndOpenConnection(); try { - PostgreSqlObjectsInstaller.Install(connection, Options.SchemaName); + PostgreSqlObjectsInstaller.Install(connection, Options.SchemaName, Options.UseTablePrefix, Options.TablePrefixName); } finally { diff --git a/src/Hangfire.PostgreSql/PostgreSqlStorageOptions.cs b/src/Hangfire.PostgreSql/PostgreSqlStorageOptions.cs index 3ce1b7d9..82f0c031 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlStorageOptions.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlStorageOptions.cs @@ -44,6 +44,8 @@ public PostgreSqlStorageOptions() JobExpirationCheckInterval = TimeSpan.FromHours(1); CountersAggregateInterval = TimeSpan.FromMinutes(5); SchemaName = "hangfire"; + UseTablePrefix = false; + TablePrefixName = "hangfire_"; AllowUnsafeValues = false; UseNativeDatabaseTransactions = true; PrepareSchemaIfNecessary = true; @@ -127,6 +129,8 @@ public int DeleteExpiredBatchSize public bool UseNativeDatabaseTransactions { get; set; } public bool PrepareSchemaIfNecessary { get; set; } public string SchemaName { get; set; } + public bool UseTablePrefix { get; set; } + public string TablePrefixName { get; set; } public bool EnableTransactionScopeEnlistment { get; set; } public bool EnableLongPolling { get; set; } diff --git a/src/Hangfire.PostgreSql/PostgreSqlWriteOnlyTransaction.cs b/src/Hangfire.PostgreSql/PostgreSqlWriteOnlyTransaction.cs index eb8ecba6..20c53887 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlWriteOnlyTransaction.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlWriteOnlyTransaction.cs @@ -30,6 +30,7 @@ using Hangfire.Common; using Hangfire.States; using Hangfire.Storage; +using Utility = Hangfire.PostgreSql.Utils.Utils; namespace Hangfire.PostgreSql { @@ -82,7 +83,7 @@ private TransactionScope CreateTransactionScope() public override void ExpireJob(string jobId, TimeSpan expireIn) { string sql = $@" - UPDATE ""{_storage.Options.SchemaName}"".""job"" + UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" SET ""expireat"" = NOW() + INTERVAL '{(long)expireIn.TotalSeconds} SECONDS' WHERE ""id"" = @Id; "; @@ -94,7 +95,7 @@ public override void ExpireJob(string jobId, TimeSpan expireIn) public override void PersistJob(string jobId) { string sql = $@" - UPDATE ""{_storage.Options.SchemaName}"".""job"" + UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" SET ""expireat"" = NULL WHERE ""id"" = @Id; "; @@ -107,10 +108,10 @@ public override void SetJobState(string jobId, IState state) { string addAndSetStateSql = $@" WITH ""s"" AS ( - INSERT INTO ""{_storage.Options.SchemaName}"".""state"" (""jobid"", ""name"", ""reason"", ""createdat"", ""data"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("state")}"" (""jobid"", ""name"", ""reason"", ""createdat"", ""data"") VALUES (@JobId, @Name, @Reason, @CreatedAt, @Data) RETURNING ""id"" ) - UPDATE ""{_storage.Options.SchemaName}"".""job"" ""j"" + UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("job")}"" ""j"" SET ""stateid"" = s.""id"", ""statename"" = @Name FROM ""s"" WHERE ""j"".""id"" = @Id; @@ -130,7 +131,7 @@ public override void SetJobState(string jobId, IState state) public override void AddJobState(string jobId, IState state) { string addStateSql = $@" - INSERT INTO ""{_storage.Options.SchemaName}"".""state"" (""jobid"", ""name"", ""reason"", ""createdat"", ""data"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("state")}"" (""jobid"", ""name"", ""reason"", ""createdat"", ""data"") VALUES (@JobId, @Name, @Reason, @CreatedAt, @Data); "; @@ -157,7 +158,7 @@ public override void AddToQueue(string queue, string jobId) public override void IncrementCounter(string key) { string sql = $@" - INSERT INTO ""{_storage.Options.SchemaName}"".""counter"" (""key"", ""value"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("counter")}"" (""key"", ""value"") VALUES (@Key, @Value); "; QueueCommand(con => con.Execute(sql, @@ -167,7 +168,7 @@ public override void IncrementCounter(string key) public override void IncrementCounter(string key, TimeSpan expireIn) { string sql = $@" - INSERT INTO ""{_storage.Options.SchemaName}"".""counter""(""key"", ""value"", ""expireat"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("counter")}""(""key"", ""value"", ""expireat"") VALUES (@Key, @Value, NOW() + INTERVAL '{(long)expireIn.TotalSeconds} SECONDS'); "; QueueCommand(con => con.Execute(sql, @@ -177,7 +178,7 @@ public override void IncrementCounter(string key, TimeSpan expireIn) public override void DecrementCounter(string key) { string sql = $@" - INSERT INTO ""{_storage.Options.SchemaName}"".""counter"" (""key"", ""value"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("counter")}"" (""key"", ""value"") VALUES (@Key, @Value); "; QueueCommand(con => con.Execute(sql, @@ -187,7 +188,7 @@ public override void DecrementCounter(string key) public override void DecrementCounter(string key, TimeSpan expireIn) { string sql = $@" - INSERT INTO ""{_storage.Options.SchemaName}"".""counter""(""key"", ""value"", ""expireat"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("counter")}""(""key"", ""value"", ""expireat"") VALUES (@Key, @Value, NOW() + INTERVAL '{((long)expireIn.TotalSeconds).ToString(CultureInfo.InvariantCulture)} SECONDS'); "; QueueCommand(con => con.Execute(sql, new { Key = key, Value = -1 })); @@ -201,7 +202,7 @@ public override void AddToSet(string key, string value) public override void AddToSet(string key, string value, double score) { string addSql = $@" - INSERT INTO ""{_storage.Options.SchemaName}"".""set""(""key"", ""value"", ""score"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("set")}""(""key"", ""value"", ""score"") VALUES(@Key, @Value, @Score) ON CONFLICT (""key"", ""value"") DO UPDATE SET ""score"" = EXCLUDED.""score"" @@ -213,7 +214,7 @@ ON CONFLICT (""key"", ""value"") public override void RemoveFromSet(string key, string value) { QueueCommand(con => con.Execute($@" - DELETE FROM ""{_storage.Options.SchemaName}"".""set"" + DELETE FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = @Key AND ""value"" = @Value; ", @@ -223,7 +224,7 @@ DELETE FROM ""{_storage.Options.SchemaName}"".""set"" public override void InsertToList(string key, string value) { QueueCommand(con => con.Execute($@" - INSERT INTO ""{_storage.Options.SchemaName}"".""list"" (""key"", ""value"") + INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("list")}"" (""key"", ""value"") VALUES (@Key, @Value); ", new { Key = key, Value = value })); @@ -232,7 +233,7 @@ public override void InsertToList(string key, string value) public override void RemoveFromList(string key, string value) { QueueCommand(con => con.Execute($@" - DELETE FROM ""{_storage.Options.SchemaName}"".""list"" + DELETE FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("list")}"" WHERE ""key"" = @Key AND ""value"" = @Value; ", new { Key = key, Value = value })); @@ -241,11 +242,11 @@ DELETE FROM ""{_storage.Options.SchemaName}"".""list"" public override void TrimList(string key, int keepStartingFrom, int keepEndingAt) { string trimSql = $@" - DELETE FROM ""{_storage.Options.SchemaName}"".""list"" AS source + DELETE FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("list")}"" AS source WHERE ""key"" = @Key AND ""id"" NOT IN ( SELECT ""id"" - FROM ""{_storage.Options.SchemaName}"".""list"" AS keep + FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("list")}"" AS keep WHERE keep.""key"" = source.""key"" ORDER BY ""id"" OFFSET @Offset LIMIT @Limit @@ -272,14 +273,14 @@ public override void SetRangeInHash(string key, IEnumerable con.Execute(sql, new { Key = key })); } @@ -315,7 +316,7 @@ public override void ExpireSet(string key, TimeSpan expireIn) throw new ArgumentNullException(nameof(key)); } - string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""set"" SET ""expireat"" = @ExpireAt WHERE ""key"" = @Key"; + string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("set")}"" SET ""expireat"" = @ExpireAt WHERE ""key"" = @Key"; QueueCommand(connection => connection.Execute(sql, new { Key = key, ExpireAt = DateTime.UtcNow.Add(expireIn) })); @@ -328,7 +329,7 @@ public override void ExpireList(string key, TimeSpan expireIn) throw new ArgumentNullException(nameof(key)); } - string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""list"" SET ""expireat"" = @ExpireAt WHERE ""key"" = @Key"; + string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("list")}"" SET ""expireat"" = @ExpireAt WHERE ""key"" = @Key"; QueueCommand(connection => connection.Execute(sql, new { Key = key, ExpireAt = DateTime.UtcNow.Add(expireIn) })); @@ -341,7 +342,7 @@ public override void ExpireHash(string key, TimeSpan expireIn) throw new ArgumentNullException(nameof(key)); } - string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""hash"" SET expireat = @ExpireAt WHERE ""key"" = @Key"; + string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("hash")}"" SET expireat = @ExpireAt WHERE ""key"" = @Key"; QueueCommand(connection => connection.Execute(sql, new { Key = key, ExpireAt = DateTime.UtcNow.Add(expireIn) })); @@ -354,7 +355,7 @@ public override void PersistSet(string key) throw new ArgumentNullException(nameof(key)); } - string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""set"" SET expireat = null WHERE ""key"" = @Key"; + string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("set")}"" SET expireat = null WHERE ""key"" = @Key"; QueueCommand(connection => connection.Execute(sql, new { Key = key })); } @@ -366,7 +367,7 @@ public override void PersistList(string key) throw new ArgumentNullException(nameof(key)); } - string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""list"" SET expireat = null WHERE ""key"" = @Key"; + string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("list")}"" SET expireat = null WHERE ""key"" = @Key"; QueueCommand(connection => connection.Execute(sql, new { Key = key })); } @@ -378,7 +379,7 @@ public override void PersistHash(string key) throw new ArgumentNullException(nameof(key)); } - string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""hash"" SET expireat = null WHERE ""key"" = @Key"; + string sql = $@"UPDATE ""{_storage.Options.SchemaName}"".""{TableNameHandler("hash")}"" SET expireat = null WHERE ""key"" = @Key"; QueueCommand(connection => connection.Execute(sql, new { Key = key })); @@ -396,7 +397,7 @@ public override void AddRangeToSet(string key, IList items) throw new ArgumentNullException(nameof(items)); } - string sql = $@"INSERT INTO ""{_storage.Options.SchemaName}"".""set"" (""key"", ""value"", ""score"") VALUES (@Key, @Value, 0.0)"; + string sql = $@"INSERT INTO ""{_storage.Options.SchemaName}"".""{TableNameHandler("set")}"" (""key"", ""value"", ""score"") VALUES (@Key, @Value, 0.0)"; QueueCommand(connection => connection.Execute(sql, items.Select(value => new { Key = key, Value = value }).ToList())); @@ -409,7 +410,7 @@ public override void RemoveSet(string key) throw new ArgumentNullException(nameof(key)); } - string sql = $@"DELETE FROM ""{_storage.Options.SchemaName}"".""set"" WHERE ""key"" = @Key"; + string sql = $@"DELETE FROM ""{_storage.Options.SchemaName}"".""{TableNameHandler("set")}"" WHERE ""key"" = @Key"; QueueCommand(connection => connection.Execute(sql, new { Key = key })); } @@ -418,5 +419,10 @@ internal void QueueCommand(Action action) { _commandQueue.Enqueue(action); } + + private string TableNameHandler(string baseName) + { + return Utility.GetTableName(baseName, _storage.Options.UseTablePrefix, _storage.Options.TablePrefixName); + } } } diff --git a/src/Hangfire.PostgreSql/Utils/TryExecute.cs b/src/Hangfire.PostgreSql/Utils/TryExecute.cs index 2bfa07a2..0ba5f3ea 100644 --- a/src/Hangfire.PostgreSql/Utils/TryExecute.cs +++ b/src/Hangfire.PostgreSql/Utils/TryExecute.cs @@ -50,5 +50,11 @@ public static bool TryExecute( result = default; return false; } + + public static string GetTableName(string baseName, bool useTablePrefix, string tablePrefixName) + { + return useTablePrefix ? tablePrefixName + baseName : baseName; + } + } } diff --git a/tests/Hangfire.PostgreSql.Tests/PostgreSqlInstallerFacts.cs b/tests/Hangfire.PostgreSql.Tests/PostgreSqlInstallerFacts.cs index 80d90025..16cc4f89 100644 --- a/tests/Hangfire.PostgreSql.Tests/PostgreSqlInstallerFacts.cs +++ b/tests/Hangfire.PostgreSql.Tests/PostgreSqlInstallerFacts.cs @@ -29,6 +29,26 @@ public void InstallingSchemaUpdatesVersionAndShouldNotThrowAnException() Assert.Null(ex); } + [Fact] + public void InstallingWithPrefixAndShouldNotThrowAnException() + { + Exception ex = Record.Exception(() => { + UseConnection(connection => { + string schemaName = "hangfire_tests_" + Guid.NewGuid().ToString().Replace("-", "_").ToLower(CultureInfo.InvariantCulture); + string tablePrefix = "hf_"; + + PostgreSqlObjectsInstaller.Install(connection, schemaName, true, tablePrefix); + + int lastVersion = connection.Query($@"SELECT version FROM ""{schemaName}"".""{ tablePrefix + "schema"}""").Single(); + Assert.Equal(23, lastVersion); + + connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + }); + + Assert.Null(ex); + } + [Fact] public void InstallingSchemaWithCapitalsUpdatesVersionAndShouldNotThrowAnException() { diff --git a/tests/Hangfire.PostgreSql.Tests/PostgreSqlTablePrefixFacts.cs b/tests/Hangfire.PostgreSql.Tests/PostgreSqlTablePrefixFacts.cs new file mode 100644 index 00000000..a3f49193 --- /dev/null +++ b/tests/Hangfire.PostgreSql.Tests/PostgreSqlTablePrefixFacts.cs @@ -0,0 +1,1027 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Linq; +using System.Text.Json; +using Dapper; +using Hangfire.Common; +using Hangfire.PostgreSql.Factories; +using Hangfire.PostgreSql.Tests.Entities; +using Hangfire.PostgreSql.Tests.Utils; +using Hangfire.States; +using Hangfire.Storage; +using Moq; +using Npgsql; +using Xunit; + +namespace Hangfire.PostgreSql.Tests +{ + public class PostgreSqlTablePrefixFacts : IClassFixture + { + private readonly PostgreSqlStorageFixture _fixture; + + public PostgreSqlTablePrefixFacts(PostgreSqlStorageFixture fixture) + { + _fixture = fixture; + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_ExpireJob_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string tableName = usePrefix ? prefix + "job" : "job"; + string arrangeSql = $@" + INSERT INTO ""{schemaName}"".""{tableName}"" (""invocationdata"", ""arguments"", ""createdat"") + VALUES ('{{}}', '[]', NOW()) RETURNING ""id"" + "; + + DateTime utcNow = DateTime.UtcNow; + string jobId = connection.QuerySingle(arrangeSql).ToString(CultureInfo.InvariantCulture); + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.ExpireJob(jobId, TimeSpan.FromDays(1)); + transaction.Commit(); + } + + // Assert + TestJob job = Helper.GetTestJob(connection, schemaName, jobId, usePrefix, prefix); + Assert.NotNull(job.ExpireAt); + Assert.True(utcNow.AddMinutes(-1) < job.ExpireAt && job.ExpireAt <= utcNow.AddDays(1).AddSeconds(5)); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_PersistJob_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string tableName = usePrefix ? prefix + "job" : "job"; + string arrangeSql = $@" + INSERT INTO ""{schemaName}"".""{tableName}"" (""invocationdata"", ""arguments"", ""createdat"", ""expireat"") + VALUES ('{{}}', '[]', NOW(), NOW()) RETURNING ""id"" + "; + + string jobId = connection.QuerySingle(arrangeSql).ToString(CultureInfo.InvariantCulture); + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.PersistJob(jobId); + transaction.Commit(); + } + + // Assert + TestJob job = Helper.GetTestJob(connection, schemaName, jobId, usePrefix, prefix); + Assert.Null(job.ExpireAt); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_SetJobState_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string tableName = usePrefix ? prefix + "job" : "job"; + string stateTableName = usePrefix ? prefix + "state" : "state"; + string arrangeSql = $@" + INSERT INTO ""{schemaName}"".""{tableName}"" (""invocationdata"", ""arguments"", ""createdat"") + VALUES ('{{}}', '[]', NOW()) RETURNING ""id"" + "; + + string jobId = connection.QuerySingle(arrangeSql).ToString(CultureInfo.InvariantCulture); + + Mock state = new(); + state.Setup(x => x.Name).Returns("TestState"); + state.Setup(x => x.Reason).Returns("TestReason"); + state.Setup(x => x.SerializeData()).Returns(new Dictionary { { "Key", "Value" } }); + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.SetJobState(jobId, state.Object); + transaction.Commit(); + } + + // Assert + TestJob job = Helper.GetTestJob(connection, schemaName, jobId, usePrefix, prefix); + Assert.Equal("TestState", job.StateName); + Assert.NotNull(job.StateId); + + dynamic jobState = connection.QuerySingle($@"SELECT * FROM ""{schemaName}"".""{stateTableName}"" WHERE ""jobid"" = @JobId", + new { JobId = long.Parse(jobId) }); + Assert.Equal("TestState", jobState.name); + Assert.Equal("TestReason", jobState.reason); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_AddToSet_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string key = "test-set"; + string value = "test-value"; + double score = 42.5; + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.AddToSet(key, value, score); + transaction.Commit(); + } + + // Assert + string setTableName = usePrefix ? prefix + "set" : "set"; + dynamic result = connection.QuerySingle( + $@"SELECT * FROM ""{schemaName}"".""{setTableName}"" WHERE ""key"" = @Key AND ""value"" = @Value", + new { Key = key, Value = value }); + + Assert.Equal(key, result.key); + Assert.Equal(value, result.value); + Assert.Equal(score, (double)result.score); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_AddToQueue_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string tableName = usePrefix ? prefix + "job" : "job"; + string queueTableName = usePrefix ? prefix + "jobqueue" : "jobqueue"; + string arrangeSql = $@" + INSERT INTO ""{schemaName}"".""{tableName}"" (""invocationdata"", ""arguments"", ""createdat"") + VALUES ('{{}}', '[]', NOW()) RETURNING ""id"" + "; + + string jobId = connection.QuerySingle(arrangeSql).ToString(CultureInfo.InvariantCulture); + string queue = "default"; + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.AddToQueue(queue, jobId); + transaction.Commit(); + } + + // Assert + dynamic result = connection.QuerySingle( + $@"SELECT * FROM ""{schemaName}"".""{queueTableName}"" WHERE ""jobid"" = @JobId", + new { JobId = long.Parse(jobId) }); + + Assert.Equal(queue, result.queue); + Assert.Equal(long.Parse(jobId), (long)result.jobid); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_IncrementCounter_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string key = "test-counter"; + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.IncrementCounter(key); + transaction.Commit(); + } + + // Assert + string counterTableName = usePrefix ? prefix + "counter" : "counter"; + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{counterTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + + Assert.Equal(1, count); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_SetJobParameter_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + // Arrange + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + string tableName = usePrefix ? prefix + "job" : "job"; + string hashTableName = usePrefix ? prefix + "hash" : "hash"; + string arrangeSql = $@" + INSERT INTO ""{schemaName}"".""{tableName}"" (""invocationdata"", ""arguments"", ""createdat"") + VALUES ('{{}}', '[]', NOW()) RETURNING ""id"" + "; + + string jobId = connection.QuerySingle(arrangeSql).ToString(CultureInfo.InvariantCulture); + string parameterName = "test-parameter"; + string parameterValue = "test-value"; + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.SetRangeInHash($"job:{jobId}:parameters", new[] { new KeyValuePair(parameterName, parameterValue) }); + transaction.Commit(); + } + + // Assert + string result = connection.QuerySingle( + $@"SELECT ""value"" FROM ""{schemaName}"".""{hashTableName}"" WHERE ""key"" = @Key AND ""field"" = @Field", + new { Key = $"job:{jobId}:parameters", Field = parameterName }); + + Assert.Equal(parameterValue, result); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_ComplexOperation_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string tableName = usePrefix ? prefix + "job" : "job"; + string hashTableName = usePrefix ? prefix + "hash" : "hash"; + string setTableName = usePrefix ? prefix + "set" : "set"; + string counterTableName = usePrefix ? prefix + "counter" : "counter"; + + string arrangeSql = $@" + INSERT INTO ""{schemaName}"".""{tableName}"" (""invocationdata"", ""arguments"", ""createdat"") + VALUES ('{{}}', '[]', NOW()) RETURNING ""id"" + "; + + string jobId = connection.QuerySingle(arrangeSql).ToString(CultureInfo.InvariantCulture); + + // Act - Perform multiple operations in a single transaction + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.SetRangeInHash($"job:{jobId}:metadata", new[] { + new KeyValuePair("key1", "value1"), + new KeyValuePair("key2", "value2") + }); + transaction.AddToSet("test-set", "value1", 1.0); + transaction.AddToSet("test-set", "value2", 2.0); + transaction.IncrementCounter("test-counter"); + transaction.IncrementCounter("test-counter"); + transaction.ExpireJob(jobId, TimeSpan.FromHours(1)); + transaction.Commit(); + } + + // Assert + int hashCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{hashTableName}"" WHERE ""key"" = @Key", + new { Key = $"job:{jobId}:metadata" }); + Assert.Equal(2, hashCount); + + int setCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{setTableName}"" WHERE ""key"" = @Key", + new { Key = "test-set" }); + Assert.Equal(2, setCount); + + int counterCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{counterTableName}"" WHERE ""key"" = @Key", + new { Key = "test-counter" }); + Assert.Equal(2, counterCount); + + TestJob job = Helper.GetTestJob(connection, schemaName, jobId, usePrefix, prefix); + Assert.NotNull(job.ExpireAt); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_MultipleJobsWithStates_WorksWithAndWithoutPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string tableName = usePrefix ? prefix + "job" : "job"; + string stateTableName = usePrefix ? prefix + "state" : "state"; + string jobQueueTableName = usePrefix ? prefix + "jobqueue" : "jobqueue"; + + List jobIds = new(); + for (int i = 0; i < 3; i++) + { + string sql = $@" + INSERT INTO ""{schemaName}"".""{tableName}"" (""invocationdata"", ""arguments"", ""createdat"") + VALUES ('{{}}', '[]', NOW()) RETURNING ""id"" + "; + jobIds.Add(connection.QuerySingle(sql).ToString(CultureInfo.InvariantCulture)); + } + + // Act - Create multiple jobs with different states and queue them + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + Mock enqueuedState = new(); + enqueuedState.Setup(x => x.Name).Returns("Enqueued"); + enqueuedState.Setup(x => x.Reason).Returns((string)null); + enqueuedState.Setup(x => x.SerializeData()).Returns(new Dictionary { { "Queue", "default" } }); + + Mock processingState = new(); + processingState.Setup(x => x.Name).Returns("Processing"); + processingState.Setup(x => x.Reason).Returns((string)null); + processingState.Setup(x => x.SerializeData()).Returns(new Dictionary { { "WorkerId", "worker1" } }); + + Mock failedState = new(); + failedState.Setup(x => x.Name).Returns("Failed"); + failedState.Setup(x => x.Reason).Returns("Exception occurred"); + failedState.Setup(x => x.SerializeData()).Returns(new Dictionary { { "ExceptionMessage", "Test error" } }); + + transaction.SetJobState(jobIds[0], enqueuedState.Object); + transaction.AddToQueue("default", jobIds[0]); + + transaction.SetJobState(jobIds[1], processingState.Object); + + transaction.SetJobState(jobIds[2], failedState.Object); + + transaction.Commit(); + } + + // Assert + int enqueuedCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{stateTableName}"" WHERE ""name"" = @Name", + new { Name = "Enqueued" }); + Assert.Equal(1, enqueuedCount); + + int processingCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{stateTableName}"" WHERE ""name"" = @Name", + new { Name = "Processing" }); + Assert.Equal(1, processingCount); + + int failedCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{stateTableName}"" WHERE ""name"" = @Name", + new { Name = "Failed" }); + Assert.Equal(1, failedCount); + + int queuedJobsCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{jobQueueTableName}"""); + Assert.Equal(1, queuedJobsCount); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_WithNonExistentJobId_DoesNotThrow(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Act - Should not throw even if job doesn't exist + Exception ex = Record.Exception(() => { + using IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction(); + transaction.ExpireJob("999999", TimeSpan.FromDays(1)); + transaction.Commit(); + }); + + // Assert + Assert.Null(ex); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_AddToSet_DuplicateValueUpdatesScore(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string key = "test-set"; + string value = "test-value"; + string setTableName = usePrefix ? prefix + "set" : "set"; + + // Act - Add same value twice with different scores + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.AddToSet(key, value, 1.0); + transaction.Commit(); + } + + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.AddToSet(key, value, 2.0); + transaction.Commit(); + } + + // Assert - Should have only one record with updated score + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{setTableName}"" WHERE ""key"" = @Key AND ""value"" = @Value", + new { Key = key, Value = value }); + Assert.Equal(1, count); + + double score = connection.QuerySingle( + $@"SELECT ""score"" FROM ""{schemaName}"".""{setTableName}"" WHERE ""key"" = @Key AND ""value"" = @Value", + new { Key = key, Value = value }); + Assert.Equal(2.0, score); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_RemoveFromSet_WorksWithPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string key = "test-set"; + string value1 = "value1"; + string value2 = "value2"; + string setTableName = usePrefix ? prefix + "set" : "set"; + + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.AddToSet(key, value1, 1.0); + transaction.AddToSet(key, value2, 2.0); + transaction.Commit(); + } + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.RemoveFromSet(key, value1); + transaction.Commit(); + } + + // Assert + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{setTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + Assert.Equal(1, count); + + string remainingValue = connection.QuerySingle( + $@"SELECT ""value"" FROM ""{schemaName}"".""{setTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + Assert.Equal(value2, remainingValue); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_InsertToList_WorksWithPrefix(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + + // Arrange + string key = "test-list"; + string listTableName = usePrefix ? prefix + "list" : "list"; + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.InsertToList(key, "value1"); + transaction.InsertToList(key, "value2"); + transaction.InsertToList(key, "value3"); + transaction.Commit(); + } + + // Assert + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{listTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + Assert.Equal(3, count); + + List values = connection.Query( + $@"SELECT ""value"" FROM ""{schemaName}"".""{listTableName}"" WHERE ""key"" = @Key ORDER BY ""id"" DESC", + new { Key = key }).ToList(); + + Assert.Equal(new[] { "value3", "value2", "value1" }, values); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_WithEmptyKey_DoesNotThrow(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + string setTableName = usePrefix ? prefix + "set" : "set"; + + // Act - Empty key should be allowed + Exception ex = Record.Exception(() => { + using IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction(); + transaction.AddToSet("", "value", 1.0); + transaction.Commit(); + }); + + // Assert + Assert.Null(ex); + + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{setTableName}"" WHERE ""key"" = @Key", + new { Key = "" }); + Assert.Equal(1, count); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_MultipleUpdatesToSameHash_WorksCorrectly(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + string hashTableName = usePrefix ? prefix + "hash" : "hash"; + string key = "test-hash"; + + // Act - Update same field multiple times in same transaction + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.SetRangeInHash(key, new[] { + new KeyValuePair("field1", "value1") + }); + transaction.SetRangeInHash(key, new[] { + new KeyValuePair("field1", "value2") + }); + transaction.SetRangeInHash(key, new[] { + new KeyValuePair("field1", "value3") + }); + transaction.Commit(); + } + + // Assert - Should have the last value + string result = connection.QuerySingle( + $@"SELECT ""value"" FROM ""{schemaName}"".""{hashTableName}"" WHERE ""key"" = @Key AND ""field"" = @Field", + new { Key = key, Field = "field1" }); + + Assert.Equal("value3", result); + + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{hashTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + Assert.Equal(1, count); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_IncrementCounter_WithTimeToLive_WorksCorrectly(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + string counterTableName = usePrefix ? prefix + "counter" : "counter"; + string key = "test-counter-ttl"; + DateTime utcNow = DateTime.UtcNow; + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.IncrementCounter(key, TimeSpan.FromHours(2)); + transaction.Commit(); + } + + // Assert + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{counterTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + Assert.Equal(1, count); + + DateTime? expireAt = connection.QuerySingle( + $@"SELECT ""expireat"" FROM ""{schemaName}"".""{counterTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + + Assert.NotNull(expireAt); + Assert.True(expireAt.Value > utcNow.AddHours(1).AddMinutes(59) && expireAt.Value <= utcNow.AddHours(2).AddMinutes(1)); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_DecrementCounter_WorksCorrectly(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + string counterTableName = usePrefix ? prefix + "counter" : "counter"; + string key = "test-decrement-counter"; + + // Arrange - Add some counters first + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.IncrementCounter(key); + transaction.IncrementCounter(key); + transaction.IncrementCounter(key); + transaction.Commit(); + } + + // Act - Decrement + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.DecrementCounter(key); + transaction.Commit(); + } + + // Assert + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{counterTableName}"" WHERE ""key"" = @Key AND ""value"" > 0", + new { Key = key }); + Assert.Equal(3, count); + + int negativeCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{counterTableName}"" WHERE ""key"" = @Key AND ""value"" < 0", + new { Key = key }); + Assert.Equal(1, negativeCount); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_RemoveFromList_WorksCorrectly(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + string listTableName = usePrefix ? prefix + "list" : "list"; + string key = "test-list"; + + // Arrange + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.InsertToList(key, "value1"); + transaction.InsertToList(key, "value2"); + transaction.InsertToList(key, "value3"); + transaction.InsertToList(key, "value2"); // duplicate + transaction.Commit(); + } + + // Act - Remove one instance of value2 + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.RemoveFromList(key, "value2"); + transaction.Commit(); + } + + // Assert + int totalCount = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{listTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + Assert.Equal(3, totalCount); + + int value2Count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{listTableName}"" WHERE ""key"" = @Key AND ""value"" = @Value", + new { Key = key, Value = "value2" }); + Assert.Equal(1, value2Count); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_TrimList_WorksCorrectly(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + string listTableName = usePrefix ? prefix + "list" : "list"; + string key = "test-list"; + + // Arrange - Add 10 items + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + for (int i = 0; i < 10; i++) + { + transaction.InsertToList(key, $"value{i}"); + } + transaction.Commit(); + } + + // Act - Keep only first 5 + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.TrimList(key, 0, 4); + transaction.Commit(); + } + + // Assert + int count = connection.QuerySingle( + $@"SELECT COUNT(*) FROM ""{schemaName}"".""{listTableName}"" WHERE ""key"" = @Key", + new { Key = key }); + Assert.Equal(5, count); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + [Theory] + [CleanDatabase] + [InlineData(false, "")] + [InlineData(true, "hf_")] + public void WriteOnlyTransaction_AddJobState_WithLongData_WorksCorrectly(bool usePrefix, string prefix) + { + string schemaName = "hangfire_test_" + Guid.NewGuid().ToString(); + PostgreSqlStorageOptions options = new() { + SchemaName = schemaName, + EnableTransactionScopeEnlistment = true, + UseTablePrefix = usePrefix, + TablePrefixName = usePrefix ? prefix : string.Empty, + PrepareSchemaIfNecessary = true, + }; + + UseConnection(connection => { + PostgreSqlStorage storage = new(new ExistingNpgsqlConnectionFactory(connection, options), options); + string tableName = usePrefix ? prefix + "job" : "job"; + string stateTableName = usePrefix ? prefix + "state" : "state"; + + // Arrange + string arrangeSql = $@" + INSERT INTO ""{schemaName}"".""{tableName}"" (""invocationdata"", ""arguments"", ""createdat"") + VALUES ('{{}}', '[]', NOW()) RETURNING ""id"" + "; + string jobId = connection.QuerySingle(arrangeSql).ToString(CultureInfo.InvariantCulture); + + // Create large data + Dictionary largeData = new(); + for (int i = 0; i < 100; i++) + { + largeData[$"Key{i}"] = $"Value{i}_" + new string('X', 100); + } + + Mock state = new(); + state.Setup(x => x.Name).Returns("ProcessingWithLargeData"); + state.Setup(x => x.Reason).Returns("Testing with large data payload"); + state.Setup(x => x.SerializeData()).Returns(largeData); + + // Act + using (IWriteOnlyTransaction transaction = storage.GetConnection().CreateWriteTransaction()) + { + transaction.AddJobState(jobId, state.Object); + transaction.Commit(); + } + + // Assert + dynamic jobState = connection.QuerySingle( + $@"SELECT * FROM ""{schemaName}"".""{stateTableName}"" WHERE ""jobid"" = @JobId", + new { JobId = long.Parse(jobId) }); + + Assert.Equal("ProcessingWithLargeData", jobState.name); + Assert.NotNull(jobState.data); + + string dataJson = jobState.data; + Dictionary deserializedData = JsonSerializer.Deserialize>(dataJson); + Assert.Equal(100, deserializedData.Count); + + // connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;"); + }); + } + + private static void UseConnection(Action action) + { + using NpgsqlConnection connection = ConnectionUtils.CreateConnection(); + action(connection); + } + } +} \ No newline at end of file diff --git a/tests/Hangfire.PostgreSql.Tests/Utils/Helper.cs b/tests/Hangfire.PostgreSql.Tests/Utils/Helper.cs index 3e79b95b..e73f7880 100644 --- a/tests/Hangfire.PostgreSql.Tests/Utils/Helper.cs +++ b/tests/Hangfire.PostgreSql.Tests/Utils/Helper.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Data; using System.Globalization; using Dapper; @@ -8,12 +8,32 @@ namespace Hangfire.PostgreSql.Tests.Utils { public static class Helper { - public static TestJob GetTestJob(IDbConnection connection, string schemaName, string jobId) + public static TestJob GetTestJob(IDbConnection connection, string schemaName, string jobId, bool usePrefix = false, string prefix = "") { + string tableName = usePrefix ? prefix + "job" : "job"; return connection - .QuerySingle($@"SELECT ""id"" ""Id"", ""invocationdata"" ""InvocationData"", ""arguments"" ""Arguments"", ""expireat"" ""ExpireAt"", ""statename"" ""StateName"", ""stateid"" ""StateId"", ""createdat"" ""CreatedAt"" FROM ""{schemaName}"".""job"" WHERE ""id"" = @Id OR @Id = -1", + .QuerySingle($@"SELECT ""id"" ""Id"", ""invocationdata"" ""InvocationData"", ""arguments"" ""Arguments"", ""expireat"" ""ExpireAt"", ""statename"" ""StateName"", ""stateid"" ""StateId"", ""createdat"" ""CreatedAt"" FROM ""{schemaName}"".""{tableName}"" WHERE ""id"" = @Id OR @Id = -1", new { Id = Convert.ToInt64(jobId, CultureInfo.InvariantCulture) }); } + public static void DoTableRename(IDbConnection connection, string schemaName, bool useTablePrefix, string newPrefix) + { + if (!useTablePrefix) + { + return; + } + + var hangfireTableNames = new[] + { + "job", "state", "counter", "queue", "schema", "server", "set", "list", "hash", "lock", "aggregatedcounter", "jobparameter", "jobqueue" + }; + foreach (string tableName in hangfireTableNames) + { + string oldTableName = tableName; + string newTableName = newPrefix + tableName; + connection.Execute($@"ALTER TABLE ""{schemaName}"".""{oldTableName}"" RENAME TO ""{newTableName}"";"); + } + } + } } From 1acde247338c97b868ae2160e13d304be7f14b5b Mon Sep 17 00:00:00 2001 From: argxh Date: Mon, 2 Feb 2026 17:35:41 +0100 Subject: [PATCH 2/3] fix: typo --- src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs b/src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs index 3216795c..6a22c44b 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlDistributedLock.cs @@ -200,7 +200,7 @@ public static bool TryLock(IDbConnection connection, PostgreSqlStorageOptions op trx = BeginTransactionIfNotPresent(connection); int rowsAffected = connection.Execute($@" - INSERT INTO ""{schemaName}"".""lock""(""resource"", ""acquired"") + INSERT INTO ""{schemaName}"".""{TableNameHandler("lock", options)}""(""resource"", ""acquired"") SELECT @Resource, @Acquired WHERE NOT EXISTS ( SELECT 1 FROM ""{schemaName}"".""{TableNameHandler("lock", options)}"" From d29adf80dcc161bbea2414a043b91b3d8e495e25 Mon Sep 17 00:00:00 2001 From: argxh Date: Mon, 9 Feb 2026 09:43:39 +0100 Subject: [PATCH 3/3] fix: added table prefix on VersionAlreadyApplied, other fixes --- .../PostgreSqlObjectsInstaller.cs | 64 ++++++++++++++----- 1 file changed, 47 insertions(+), 17 deletions(-) diff --git a/src/Hangfire.PostgreSql/PostgreSqlObjectsInstaller.cs b/src/Hangfire.PostgreSql/PostgreSqlObjectsInstaller.cs index 1077ad92..741c3fab 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlObjectsInstaller.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlObjectsInstaller.cs @@ -67,7 +67,7 @@ public static void Install(NpgsqlConnection connection, string schemaName = "han script = script.Replace("'hangfire'", $"'{schemaName}'").Replace(@"""hangfire""", $@"""{schemaName}"""); } - if (!VersionAlreadyApplied(connection, schemaName, version)) + if (!VersionAlreadyApplied(connection, schemaName, version, useTablePrefix, tablePrefixName)) { string commandText = $@"{script}; UPDATE ""{schemaName}"".""schema"" SET ""version"" = @Version WHERE ""version"" = @PreviousVersion"; using NpgsqlTransaction transaction = connection.BeginTransaction(IsolationLevel.Serializable); @@ -115,17 +115,43 @@ public static void Install(NpgsqlConnection connection, string schemaName = "han { _logger.Info("Check if prefix change has already been recorded.."); string prefixChangeKey = "hangfire:table_prefix_changed"; - using (var checkCmd = new NpgsqlCommand( - $@"SELECT 1 FROM ""{schemaName}"".""set"" WHERE ""key"" = @Key LIMIT 1;", connection)) + string tableNameToCheck = tablePrefixName + "set"; + bool shouldInsertTrackingRecord = false; + + // Verifica se la tabella con prefisso esiste già + bool tableWithPrefixExists = false; + using (var checkTableCmd = new NpgsqlCommand( + $@"SELECT 1 FROM information_schema.tables + WHERE table_schema = @SchemaName + AND table_name = @TableName + LIMIT 1;", connection)) { - checkCmd.Parameters.AddWithValue("Key", prefixChangeKey); - var exists = checkCmd.ExecuteScalar(); - if (exists != null && exists != DBNull.Value) + checkTableCmd.Parameters.AddWithValue("SchemaName", schemaName); + checkTableCmd.Parameters.AddWithValue("TableName", tableNameToCheck); + var tableExists = checkTableCmd.ExecuteScalar(); + tableWithPrefixExists = (tableExists != null && tableExists != DBNull.Value); + } + + if (tableWithPrefixExists) + { + // La tabella con prefisso esiste già, controlla se il prefisso è già stato applicato + using (var checkCmd = new NpgsqlCommand( + $@"SELECT 1 FROM ""{schemaName}"".""{tableNameToCheck}"" WHERE ""key"" = @Key LIMIT 1;", connection)) { - _logger.Warn("Table prefix has already been changed. If you want to change it again, you must drop all Hangfire tables and reinstall them."); - return; + checkCmd.Parameters.AddWithValue("Key", prefixChangeKey); + var exists = checkCmd.ExecuteScalar(); + if (exists != null && exists != DBNull.Value) + { + _logger.Warn("Table prefix has already been changed. If you want to change it again, you must drop all Hangfire tables and reinstall them."); + return; + } } } + else + { + // La tabella con prefisso non esiste, bisognerà inserire il record dopo le rename + shouldInsertTrackingRecord = true; + } var hangfireTableNames = new[] { @@ -161,14 +187,17 @@ FROM information_schema.tables } } - using (var insertCmd = new NpgsqlCommand( - $@"INSERT INTO ""{schemaName}"".""{tablePrefixName + "set"}"" (""key"", ""value"", ""score"") VALUES (@Key, @Value, @Score);", connection)) + if (shouldInsertTrackingRecord) { - insertCmd.Parameters.AddWithValue("Key", prefixChangeKey); - insertCmd.Parameters.AddWithValue("Value", tablePrefixName); - insertCmd.Parameters.AddWithValue("Score", 0); // oppure un valore appropriato - insertCmd.ExecuteNonQuery(); - _logger.Info($"Table prefix change persisted in set table with key '{prefixChangeKey}' and value '{tablePrefixName}'"); + using (var insertCmd = new NpgsqlCommand( + $@"INSERT INTO ""{schemaName}"".""{tablePrefixName + "set"}"" (""key"", ""value"", ""score"") VALUES (@Key, @Value, @Score);", connection)) + { + insertCmd.Parameters.AddWithValue("Key", prefixChangeKey); + insertCmd.Parameters.AddWithValue("Value", tablePrefixName); + insertCmd.Parameters.AddWithValue("Score", 0); + insertCmd.ExecuteNonQuery(); + _logger.Info($"Table prefix change persisted in set table with key '{prefixChangeKey}' and value '{tablePrefixName}'"); + } } } catch (Exception ex) @@ -180,11 +209,12 @@ FROM information_schema.tables _logger.Info("Hangfire SQL objects installed."); } - private static bool VersionAlreadyApplied(NpgsqlConnection connection, string schemaName, int version) + private static bool VersionAlreadyApplied(NpgsqlConnection connection, string schemaName, int version, bool useTablePrefix, string tablePrefixName) { try { - using NpgsqlCommand command = new($@"SELECT true ""VersionAlreadyApplied"" FROM ""{schemaName}"".""schema"" WHERE ""version"" >= $1", connection); + string tableName = useTablePrefix ? (tablePrefixName + "schema") : "schema"; + using NpgsqlCommand command = new($@"SELECT true ""VersionAlreadyApplied"" FROM ""{schemaName}"".""{tableName}"" WHERE ""version"" >= $1", connection); command.Parameters.Add(new NpgsqlParameter { Value = version }); object result = command.ExecuteScalar(); if (true.Equals(result))