diff --git a/src/modules/persistence/Elsa.Persistence.Dapper.Migrations/Runtime/V3_9.cs b/src/modules/persistence/Elsa.Persistence.Dapper.Migrations/Runtime/V3_9.cs new file mode 100644 index 00000000..7093264d --- /dev/null +++ b/src/modules/persistence/Elsa.Persistence.Dapper.Migrations/Runtime/V3_9.cs @@ -0,0 +1,39 @@ +using System.Diagnostics.CodeAnalysis; +using FluentMigrator; +using JetBrains.Annotations; +using static System.Int32; + +namespace Elsa.Persistence.Dapper.Migrations.Runtime; + +/// +/// Creates the KeyValues table used by DapperKeyValueStore (#263) +/// and adds BookmarkQueueItems.SerializedOptions (#264). +/// Both operations are existence-guarded so hand-created workarounds keep working. +/// +[Migration(20008, "Elsa:Runtime:V3.9")] +[PublicAPI] +[SuppressMessage("ReSharper", "InconsistentNaming")] +public class V3_9 : Migration +{ + /// + public override void Up() + { + if (!Schema.Table("KeyValues").Exists()) + Create.Table("KeyValues") + .WithColumn("Id").AsString().PrimaryKey() + .WithColumn("TenantId").AsString().Nullable() + .WithColumn("Value").AsString(MaxValue).Nullable(); + + if (!Schema.Table("BookmarkQueueItems").Column("SerializedOptions").Exists()) + Alter.Table("BookmarkQueueItems").AddColumn("SerializedOptions").AsString(MaxValue).Nullable(); + } + + /// + /// + /// Leave KeyValues and SerializedOptions in place. Rolling them back would + /// drop a hand-created KeyValues table or SerializedOptions column and their data. + /// + public override void Down() + { + } +} diff --git a/src/modules/persistence/Elsa.Persistence.Dapper/Extensions/ParameterizedQueryBuilderExtensions.cs b/src/modules/persistence/Elsa.Persistence.Dapper/Extensions/ParameterizedQueryBuilderExtensions.cs index 1739d6d8..028d0d0c 100644 --- a/src/modules/persistence/Elsa.Persistence.Dapper/Extensions/ParameterizedQueryBuilderExtensions.cs +++ b/src/modules/persistence/Elsa.Persistence.Dapper/Extensions/ParameterizedQueryBuilderExtensions.cs @@ -154,6 +154,17 @@ public static ParameterizedQuery IsNull(this ParameterizedQuery query, string fi return query; } + /// + /// Matches a default-tenant stamp: NULL or empty string. + /// Legacy rows may use either; is ''. + /// + public static ParameterizedQuery IsNullOrEmpty(this ParameterizedQuery query, string field) + { + var ident = query.QuoteIdent(field); + query.Sql.AppendLine($"and ({ident} is null or {ident} = '')"); + return query; + } + /// /// Appends an IS NOT NULL clause to the query. /// @@ -241,9 +252,9 @@ public static ParameterizedQuery StartsWith(this ParameterizedQuery query, strin if (!startsWith || value == null || string.IsNullOrWhiteSpace(value)) return query; - var searchTermLike = $"{value}%"; - query.Sql.AppendLine($"and {query.QuoteIdent(field)} like @SearchTermLike"); - query.Parameters.Add($"@{field}", searchTermLike); + var parameterName = $"@{field}StartsWith"; + query.Sql.AppendLine($"and {query.QuoteIdent(field)} like {parameterName}"); + query.Parameters.Add(parameterName, $"{value}%"); return query; } diff --git a/src/modules/persistence/Elsa.Persistence.Dapper/Modules/Runtime/Stores/KeyValueStore.cs b/src/modules/persistence/Elsa.Persistence.Dapper/Modules/Runtime/Stores/KeyValueStore.cs index 03725c44..5781fd56 100644 --- a/src/modules/persistence/Elsa.Persistence.Dapper/Modules/Runtime/Stores/KeyValueStore.cs +++ b/src/modules/persistence/Elsa.Persistence.Dapper/Modules/Runtime/Stores/KeyValueStore.cs @@ -42,12 +42,23 @@ public Task DeleteAsync(string key, CancellationToken cancellationToken) return store.DeleteAsync(query => query.Is(nameof(KeyValuePairRecord.Id), key), cancellationToken); } + /// + /// Atomic delete: returns true only when the DELETE affected a row. + /// Becomes IKeyValueStore.TryDeleteAsync once ElsaVersion includes elsa-core#8538. + /// + public async Task TryDeleteAsync(string key, CancellationToken cancellationToken = default) + { + return await store.DeleteAsync(query => query.Is(nameof(KeyValuePairRecord.Id), key), cancellationToken) > 0; + } + private void ApplyFilter(ParameterizedQuery query, KeyValueFilter filter) { - query - .Is(nameof(KeyValuePairRecord.Id), filter.Key) - .In(nameof(KeyValuePairRecord.Id), filter.Keys) - .StartsWith(nameof(KeyValuePairRecord.Id), filter.StartsWith, filter.Key); + if (filter.StartsWith) + query.StartsWith(nameof(KeyValuePairRecord.Id), true, filter.Key); + else + query.Is(nameof(KeyValuePairRecord.Id), filter.Key); + + query.In(nameof(KeyValuePairRecord.Id), filter.Keys); } private KeyValuePairRecord Map(SerializedKeyValuePair source) diff --git a/src/modules/persistence/Elsa.Persistence.Dapper/Services/Store.cs b/src/modules/persistence/Elsa.Persistence.Dapper/Services/Store.cs index 91d1db47..fe0857d2 100644 --- a/src/modules/persistence/Elsa.Persistence.Dapper/Services/Store.cs +++ b/src/modules/persistence/Elsa.Persistence.Dapper/Services/Store.cs @@ -542,7 +542,11 @@ private void ApplyTenantFilter(ParameterizedQuery query, bool tenantAgnostic = f var tenant = tenantAccessor.Tenant; var tenantId = tenant?.Id; - query.Is(nameof(Record.TenantId), (object?)tenantId ?? DBNull.Value); + // Default tenant is '' in Elsa and NULL on some legacy rows (#245 / #260). + if (string.IsNullOrEmpty(tenantId)) + query.IsNullOrEmpty(nameof(Record.TenantId)); + else + query.Is(nameof(Record.TenantId), tenantId); } private void SetTenantId(T record) diff --git a/src/modules/persistence/Elsa.Persistence.MongoDb/Common/MongoDbStore.cs b/src/modules/persistence/Elsa.Persistence.MongoDb/Common/MongoDbStore.cs index 19df4c48..85144a7f 100644 --- a/src/modules/persistence/Elsa.Persistence.MongoDb/Common/MongoDbStore.cs +++ b/src/modules/persistence/Elsa.Persistence.MongoDb/Common/MongoDbStore.cs @@ -509,7 +509,12 @@ private FilterDefinition CreateTenantOwnedUpsertFilter(TDocument docu & Builders.Filter.Where(x => (x as Entity)!.TenantId == writerTenantId || (x as Entity)!.TenantId == Tenant.AgnosticTenantId); } - private FilterDefinition ApplyTenantScope(FilterDefinition filter, bool tenantAgnostic) + /// + /// ANDs the current tenant onto . No-ops when + /// is true or is not an . + /// Writes using this filter match only the ambient tenant, not tenant-agnostic ("*") rows. + /// + public FilterDefinition ApplyTenantScope(FilterDefinition filter, bool tenantAgnostic = false) { if (tenantAgnostic || !typeof(Entity).IsAssignableFrom(typeof(TDocument))) return filter; diff --git a/src/modules/persistence/Elsa.Persistence.MongoDb/Modules/Runtime/KeyValueStore.cs b/src/modules/persistence/Elsa.Persistence.MongoDb/Modules/Runtime/KeyValueStore.cs index ccbf94b8..b3de372b 100644 --- a/src/modules/persistence/Elsa.Persistence.MongoDb/Modules/Runtime/KeyValueStore.cs +++ b/src/modules/persistence/Elsa.Persistence.MongoDb/Modules/Runtime/KeyValueStore.cs @@ -3,6 +3,7 @@ using Elsa.KeyValues.Models; using Elsa.Persistence.MongoDb.Common; using JetBrains.Annotations; +using MongoDB.Driver; using MongoDB.Driver.Linq; namespace Elsa.Persistence.MongoDb.Modules.Runtime; @@ -37,6 +38,17 @@ public Task DeleteAsync(string key, CancellationToken cancellationToken) return keyValueMongoDbStore.DeleteWhereAsync(x => x.Key == key, cancellationToken); } + /// + /// Atomic delete: returns true only when DeleteOneAsync removed a document. + /// Becomes IKeyValueStore.TryDeleteAsync once ElsaVersion includes elsa-core#8538. + /// + public async Task TryDeleteAsync(string key, CancellationToken cancellationToken = default) + { + var filter = keyValueMongoDbStore.ApplyTenantScope(Builders.Filter.Eq(x => x.Key, key)); + var result = await keyValueMongoDbStore.GetCollection().DeleteOneAsync(filter, cancellationToken); + return result.DeletedCount > 0; + } + private IQueryable Filter(IQueryable queryable, KeyValueFilter filter) { return filter.Apply(queryable); diff --git a/test/modules/persistence/Elsa.MongoDb.UnitTests/MongoKeyValueStoreTryDeleteTests.cs b/test/modules/persistence/Elsa.MongoDb.UnitTests/MongoKeyValueStoreTryDeleteTests.cs new file mode 100644 index 00000000..690a135e --- /dev/null +++ b/test/modules/persistence/Elsa.MongoDb.UnitTests/MongoKeyValueStoreTryDeleteTests.cs @@ -0,0 +1,158 @@ +using Elsa.Common.Multitenancy; +using Elsa.KeyValues.Contracts; +using Elsa.KeyValues.Entities; +using Elsa.KeyValues.Models; +using Elsa.Persistence.MongoDb.Common; +using Elsa.Persistence.MongoDb.Modules.Runtime; +using MongoDB.Driver; +using Testcontainers.MongoDb; + +namespace Elsa.MongoDb.UnitTests; + +/// +/// Atomic TryDeleteAsync against a shared Mongo collection (#260). +/// Calls the Mongo store method directly; IKeyValueStore.TryDeleteAsync is not +/// in 3.10.0-preview.5722 yet (waiting for a core pin at or after c1c935ce). +/// +public sealed class MongoKeyValueStoreTryDeleteTests : IAsyncLifetime +{ + private const string LegacyKey = "elsa.quiescence.pause.default"; + private readonly MongoDbContainer _container = new MongoDbBuilder().WithImage("mongo:7.0.24").Build(); + private readonly TestTenantAccessor _tenants = new(); + private MongoClient? _client; + private IMongoCollection _collection = null!; + private MongoDbStore _mongoStore = null!; + + public async Task InitializeAsync() + { + try + { + await _container.StartAsync(); + _client = new MongoClient(_container.GetConnectionString()); + var database = _client.GetDatabase($"elsa-trydelete-{Guid.NewGuid():N}"); + _collection = database.GetCollection("key_values"); + _mongoStore = new MongoDbStore(_collection, _tenants); + } + catch + { + await DisposeAsync(); + throw; + } + } + + public async Task DisposeAsync() + { + try + { + _client?.Dispose(); + } + finally + { + await _container.DisposeAsync(); + } + } + + [Fact(DisplayName = "#260: two Mongo stores racing TryDeleteAsync: exactly one returns true")] + public async Task TryDeleteAsync_TwoNodes_ExactlyOneReturnsTrue() + { + using var tenant = _tenants.PushContext(Tenant.Default); + var nodeA = new MongoKeyValueStore(_mongoStore); + var nodeB = new MongoKeyValueStore(_mongoStore); + await nodeA.SaveAsync(Pair(LegacyKey, "legacy-maintenance"), CancellationToken.None); + + var results = await Task.WhenAll(nodeA.TryDeleteAsync(LegacyKey), nodeB.TryDeleteAsync(LegacyKey)); + + Assert.Equal(1, results.Count(won => won)); + Assert.Equal(1, results.Count(won => !won)); + Assert.Null(await nodeA.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + } + + [Fact(DisplayName = "#260: the default find-then-delete lets both racers win")] + public async Task DefaultTryDeleteAsync_TwoNodes_BothReturnTrue() + { + using var tenant = _tenants.PushContext(Tenant.Default); + var inner = new MongoKeyValueStore(_mongoStore); + await inner.SaveAsync(Pair(LegacyKey, "legacy-maintenance"), CancellationToken.None); + var dim = new BarrierFindThenDelete(inner); + + var results = await Task.WhenAll(dim.TryDeleteAsync(LegacyKey), dim.TryDeleteAsync(LegacyKey)); + + Assert.Equal(2, results.Count(won => won)); + Assert.Null(await inner.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + } + + [Fact(DisplayName = "#260: TryDeleteAsync returns false when the key is missing")] + public async Task TryDeleteAsync_NotFound_ReturnsFalse() + { + using var tenant = _tenants.PushContext(Tenant.Default); + var store = new MongoKeyValueStore(_mongoStore); + + Assert.False(await store.TryDeleteAsync(LegacyKey)); + Assert.False(await store.TryDeleteAsync(LegacyKey)); + } + + [Fact(DisplayName = "#260: a NULL TenantId legacy row is found and TryDeleted by the default tenant")] + public async Task DefaultTenant_FindsAndTryDeletesNullTenantIdRow() + { + await _collection.InsertOneAsync(new SerializedKeyValuePair + { + Key = LegacyKey, + SerializedValue = "legacy-null", + TenantId = null + }); + + using var tenant = _tenants.PushContext(Tenant.Default); + var store = new MongoKeyValueStore(_mongoStore); + + var found = await store.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None); + Assert.Equal("legacy-null", found?.SerializedValue); + Assert.True(await store.TryDeleteAsync(LegacyKey)); + Assert.Null(await store.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + } + + private static SerializedKeyValuePair Pair(string key, string value) => new() + { + Key = key, + SerializedValue = value + }; + + private sealed class TestTenantAccessor : ITenantAccessor + { + public string TenantId => Tenant?.Id ?? Tenant.DefaultTenantId; + public Tenant? Tenant { get; private set; } + + public IDisposable PushContext(Tenant? tenant) + { + var previousTenant = Tenant; + Tenant = tenant; + return new Restore(() => Tenant = previousTenant); + } + + private sealed class Restore(Action restore) : IDisposable + { + public void Dispose() => restore(); + } + } + + /// + /// The core default TryDeleteAsync (find, then delete, always true if found). + /// Both Finds complete before either Delete so the race is deterministic. + /// + private sealed class BarrierFindThenDelete(IKeyValueStore inner) + { + private readonly TaskCompletionSource _bothFound = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _finds; + + public async Task TryDeleteAsync(string key) + { + var found = await inner.FindAsync(new KeyValueFilter { Key = key }, CancellationToken.None); + if (Interlocked.Increment(ref _finds) == 2) + _bothFound.TrySetResult(); + await _bothFound.Task; + if (found is null) + return false; + await inner.DeleteAsync(key, CancellationToken.None); + return true; + } + } +} diff --git a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValueStoreTryDeletePostgreSqlTests.cs b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValueStoreTryDeletePostgreSqlTests.cs new file mode 100644 index 00000000..1ccda52a --- /dev/null +++ b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValueStoreTryDeletePostgreSqlTests.cs @@ -0,0 +1,137 @@ +using Dapper; +using Elsa.Common.Multitenancy; +using Elsa.KeyValues.Contracts; +using Elsa.KeyValues.Entities; +using Elsa.KeyValues.Models; +using Elsa.Persistence.Dapper.Modules.Runtime.Records; +using Elsa.Persistence.Dapper.Modules.Runtime.Stores; +using Elsa.Persistence.Dapper.Services; +using FluentMigrator.Runner; +using Microsoft.Extensions.DependencyInjection; +using Npgsql; +using Testcontainers.PostgreSql; + +namespace Elsa.Persistence.Dapper.UnitTests; + +/// +/// Same #260 TryDelete cases as SQLite, on a fresh migration-built PostgreSQL via Testcontainers. +/// Calls the Dapper store method directly; IKeyValueStore.TryDeleteAsync is not +/// in 3.10.0-preview.5722 yet (waiting for a core pin at or after c1c935ce / #8538). +/// +public sealed class DapperKeyValueStoreTryDeletePostgreSqlTests : IAsyncLifetime +{ + private const string LegacyKey = "elsa.quiescence.pause.default"; + private readonly PostgreSqlContainer _container = new PostgreSqlBuilder() + .WithImage("postgres:16") + .WithDatabase("elsa") + .WithUsername("elsa") + .WithPassword("elsa") + .Build(); + private readonly TestTenantAccessor _tenants = new(); + private string _connectionString = null!; + + public async Task InitializeAsync() + { + try + { + await _container.StartAsync(); + _connectionString = _container.GetConnectionString(); + } + catch + { + await DisposeAsync(); + throw; + } + } + + public async Task DisposeAsync() + { + await _container.DisposeAsync(); + } + + [Fact(DisplayName = "#260 PG: two Dapper stores racing TryDeleteAsync: exactly one returns true")] + public async Task TryDeleteAsync_TwoNodes_ExactlyOneReturnsTrue() + { + MigrateUp(); + using var tenant = _tenants.PushContext(Tenant.Default); + var nodeA = CreateStore(); + var nodeB = CreateStore(); + await nodeA.SaveAsync(Pair(LegacyKey, "legacy-maintenance"), CancellationToken.None); + + var results = await Task.WhenAll(nodeA.TryDeleteAsync(LegacyKey), nodeB.TryDeleteAsync(LegacyKey)); + + Assert.Equal(1, results.Count(won => won)); + Assert.Equal(1, results.Count(won => !won)); + Assert.Null(await nodeA.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + } + + [Fact(DisplayName = "#260 PG: TryDeleteAsync returns false when the key is missing")] + public async Task TryDeleteAsync_NotFound_ReturnsFalse() + { + MigrateUp(); + using var tenant = _tenants.PushContext(Tenant.Default); + var store = CreateStore(); + + Assert.False(await store.TryDeleteAsync(LegacyKey)); + } + + [Fact(DisplayName = "#245 / #260 PG: a NULL TenantId legacy row is found and TryDeleted by the default tenant")] + public async Task DefaultTenant_FindsAndTryDeletesNullTenantIdRow() + { + MigrateUp(); + using (var connection = new NpgsqlConnection(_connectionString)) + { + connection.Execute( + """insert into "KeyValues" ("Id", "TenantId", "Value") values (@Id, null, @Value)""", + new { Id = LegacyKey, Value = "legacy-null" }); + } + + using var tenant = _tenants.PushContext(Tenant.Default); + var store = CreateStore(); + + var found = await store.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None); + Assert.Equal("legacy-null", found?.SerializedValue); + Assert.True(await store.TryDeleteAsync(LegacyKey)); + Assert.Null(await store.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + } + + private DapperKeyValueStore CreateStore() => + new(new Store(new PostgreSqlDbConnectionProvider(_connectionString), _tenants, "KeyValues")); + + private void MigrateUp() + { + using var services = new ServiceCollection() + .AddFluentMigratorCore() + .ConfigureRunner(rb => rb + .AddPostgres() + .WithGlobalConnectionString(_connectionString) + .ScanIn(typeof(Elsa.Persistence.Dapper.Migrations.Runtime.Initial).Assembly).For.Migrations()) + .BuildServiceProvider(false); + using var scope = services.CreateScope(); + scope.ServiceProvider.GetRequiredService().MigrateUp(); + } + + private static SerializedKeyValuePair Pair(string key, string value) => new() + { + Key = key, + SerializedValue = value + }; + + private sealed class TestTenantAccessor : ITenantAccessor + { + public string TenantId => Tenant?.Id ?? Tenant.DefaultTenantId; + public Tenant? Tenant { get; private set; } + + public IDisposable PushContext(Tenant? tenant) + { + var previousTenant = Tenant; + Tenant = tenant; + return new Restore(() => Tenant = previousTenant); + } + + private sealed class Restore(Action restore) : IDisposable + { + public void Dispose() => restore(); + } + } +} diff --git a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValueStoreTryDeleteTests.cs b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValueStoreTryDeleteTests.cs new file mode 100644 index 00000000..5085ab94 --- /dev/null +++ b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValueStoreTryDeleteTests.cs @@ -0,0 +1,193 @@ +using Dapper; +using Elsa.Common.Multitenancy; +using Elsa.KeyValues.Contracts; +using Elsa.KeyValues.Entities; +using Elsa.KeyValues.Models; +using Elsa.Persistence.Dapper.Extensions; +using Elsa.Persistence.Dapper.Modules.Runtime.Records; +using Elsa.Persistence.Dapper.Modules.Runtime.Stores; +using Elsa.Persistence.Dapper.Services; +using FluentMigrator.Runner; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Persistence.Dapper.UnitTests; + +/// +/// Atomic TryDeleteAsync on a migration-built SQLite DB (#260). +/// PostgreSQL coverage is in . +/// Calls the Dapper store method directly; IKeyValueStore.TryDeleteAsync is not +/// in 3.10.0-preview.5722 yet (waiting for a core pin at or after c1c935ce / #8538). +/// +public sealed class DapperKeyValueStoreTryDeleteTests : IDisposable +{ + private const string LegacyKey = "elsa.quiescence.pause.default"; + private readonly string _databasePath = Path.Combine(Path.GetTempPath(), $"elsa-dapper-trydelete-{Guid.NewGuid():N}.db"); + private readonly string _connectionString; + private readonly TestTenantAccessor _tenants = new(); + + public DapperKeyValueStoreTryDeleteTests() + { + _connectionString = new SqliteConnectionStringBuilder { DataSource = _databasePath, Pooling = false }.ToString(); + } + + [Fact(DisplayName = "#260: two Dapper stores racing TryDeleteAsync: exactly one returns true")] + public async Task TryDeleteAsync_TwoNodes_ExactlyOneReturnsTrue() + { + MigrateUp(); + using var tenant = _tenants.PushContext(Tenant.Default); + var nodeA = CreateStore(); + var nodeB = CreateStore(); + await nodeA.SaveAsync(Pair(LegacyKey, "legacy-maintenance"), CancellationToken.None); + + var results = await Task.WhenAll(nodeA.TryDeleteAsync(LegacyKey), nodeB.TryDeleteAsync(LegacyKey)); + + Assert.Equal(1, results.Count(won => won)); + Assert.Equal(1, results.Count(won => !won)); + Assert.Null(await nodeA.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + } + + [Fact(DisplayName = "#260: the default find-then-delete lets both racers win")] + public async Task DefaultTryDeleteAsync_TwoNodes_BothReturnTrue() + { + MigrateUp(); + using var tenant = _tenants.PushContext(Tenant.Default); + var inner = CreateStore(); + await inner.SaveAsync(Pair(LegacyKey, "legacy-maintenance"), CancellationToken.None); + var dim = new BarrierFindThenDelete(inner); + + var results = await Task.WhenAll(dim.TryDeleteAsync(LegacyKey), dim.TryDeleteAsync(LegacyKey)); + + Assert.Equal(2, results.Count(won => won)); + Assert.Null(await inner.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + } + + [Fact(DisplayName = "#260: TryDeleteAsync returns false when the key is missing")] + public async Task TryDeleteAsync_NotFound_ReturnsFalse() + { + MigrateUp(); + using var tenant = _tenants.PushContext(Tenant.Default); + var store = CreateStore(); + + Assert.False(await store.TryDeleteAsync(LegacyKey)); + Assert.False(await store.TryDeleteAsync(LegacyKey)); + } + + [Fact(DisplayName = "#245 / #260: a NULL TenantId legacy row is found and TryDeleted by the default tenant")] + public async Task DefaultTenant_FindsAndTryDeletesNullTenantIdRow() + { + MigrateUp(); + using (var connection = new SqliteConnection(_connectionString)) + { + connection.Execute( + "insert into KeyValues (Id, TenantId, Value) values (@Id, null, @Value)", + new { Id = LegacyKey, Value = "legacy-null" }); + } + + using var tenant = _tenants.PushContext(Tenant.Default); + var store = CreateStore(); + + var found = await store.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None); + Assert.Equal("legacy-null", found?.SerializedValue); + Assert.True(await store.TryDeleteAsync(LegacyKey)); + Assert.Null(await store.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + } + + [Fact(DisplayName = "#245 / #260: a named tenant cannot TryDelete a NULL TenantId default-tenant row")] + public async Task NamedTenant_DoesNotTryDeleteNullTenantIdRow() + { + MigrateUp(); + using (var connection = new SqliteConnection(_connectionString)) + { + connection.Execute( + "insert into KeyValues (Id, TenantId, Value) values (@Id, null, @Value)", + new { Id = LegacyKey, Value = "legacy-null" }); + } + + using (var named = _tenants.PushContext(new Tenant { Id = "tenant-a" })) + { + var namedStore = CreateStore(); + Assert.Null(await namedStore.FindAsync(new KeyValueFilter { Key = LegacyKey }, CancellationToken.None)); + Assert.False(await namedStore.TryDeleteAsync(LegacyKey)); + } + + using var tenant = _tenants.PushContext(Tenant.Default); + Assert.True(TableHasKey(LegacyKey)); + } + + public void Dispose() + { + File.Delete(_databasePath); + } + + private DapperKeyValueStore CreateStore() + { + var connection = new SqliteDbConnectionProvider(_connectionString); + return new DapperKeyValueStore(new Store(connection, _tenants, "KeyValues")); + } + + private void MigrateUp() + { + using var services = new ServiceCollection() + .AddFluentMigratorCore() + .ConfigureRunner(rb => rb + .AddSQLite() + .WithGlobalConnectionString(_connectionString) + .ScanIn(typeof(Elsa.Persistence.Dapper.Migrations.Runtime.Initial).Assembly).For.Migrations()) + .BuildServiceProvider(false); + using var scope = services.CreateScope(); + scope.ServiceProvider.GetRequiredService().MigrateUp(); + } + + private bool TableHasKey(string key) + { + using var connection = new SqliteConnection(_connectionString); + return connection.ExecuteScalar("select count(*) from KeyValues where Id = @key", new { key }) == 1; + } + + private static SerializedKeyValuePair Pair(string key, string value) => new() + { + Key = key, + SerializedValue = value + }; + + private sealed class TestTenantAccessor : ITenantAccessor + { + public string TenantId => Tenant?.Id ?? Tenant.DefaultTenantId; + public Tenant? Tenant { get; private set; } + + public IDisposable PushContext(Tenant? tenant) + { + var previousTenant = Tenant; + Tenant = tenant; + return new Restore(() => Tenant = previousTenant); + } + + private sealed class Restore(Action restore) : IDisposable + { + public void Dispose() => restore(); + } + } + + /// + /// The core default TryDeleteAsync (find, then delete, always true if found). + /// Both Finds complete before either Delete so the race is deterministic. + /// + private sealed class BarrierFindThenDelete(IKeyValueStore inner) + { + private readonly TaskCompletionSource _bothFound = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _finds; + + public async Task TryDeleteAsync(string key) + { + var found = await inner.FindAsync(new KeyValueFilter { Key = key }, CancellationToken.None); + if (Interlocked.Increment(ref _finds) == 2) + _bothFound.TrySetResult(); + await _bothFound.Task; + if (found is null) + return false; + await inner.DeleteAsync(key, CancellationToken.None); + return true; + } + } +} diff --git a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValuesAndBookmarkQueueTests.cs b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValuesAndBookmarkQueueTests.cs new file mode 100644 index 00000000..d1f8e5f7 --- /dev/null +++ b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/DapperKeyValuesAndBookmarkQueueTests.cs @@ -0,0 +1,287 @@ +using Dapper; +using Elsa.Common.Multitenancy; +using Elsa.Extensions; +using Elsa.KeyValues.Contracts; +using Elsa.KeyValues.Entities; +using Elsa.KeyValues.Models; +using Elsa.Persistence.Dapper.Extensions; +using Elsa.Persistence.Dapper.Services; +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Runtime; +using Elsa.Workflows.Runtime.Activities; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Models; +using Elsa.Workflows.Runtime.Options; +using FluentMigrator.Runner; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Persistence.Dapper.UnitTests; + +/// +/// Fresh DBs are built only by FluentMigrator. That is what issues #263 and #264 +/// actually break on: the stores write columns/tables the older migrations never created. +/// +public sealed class DapperKeyValuesAndBookmarkQueueTests : IDisposable +{ + private readonly string _databasePath = Path.Combine(Path.GetTempPath(), $"elsa-dapper-kv-bq-{Guid.NewGuid():N}.db"); + private readonly string _connectionString; + + public DapperKeyValuesAndBookmarkQueueTests() + { + _connectionString = new SqliteConnectionStringBuilder { DataSource = _databasePath, Pooling = false }.ToString(); + } + + [Fact(DisplayName = "#263: a migration-built DB can Save / Find / FindMany / Delete through IKeyValueStore")] + public async Task FreshMigratedDatabase_KeyValueStore_RoundTrips() + { + // Arrange + MigrateUp(); + await using var services = CreateElsaServices(); + using var scope = services.CreateScope(); + using var tenant = scope.ServiceProvider.GetRequiredService().PushContext(Tenant.Default); + var store = scope.ServiceProvider.GetRequiredService(); + + await store.SaveAsync(Pair("app:alpha", "one"), CancellationToken.None); + await store.SaveAsync(Pair("app:beta", "two"), CancellationToken.None); + await store.SaveAsync(Pair("other:gamma", "three"), CancellationToken.None); + + // Act + var alpha = await store.FindAsync(new KeyValueFilter { Key = "app:alpha" }, CancellationToken.None); + var many = (await store.FindManyAsync(new KeyValueFilter { Keys = ["app:alpha", "app:beta"] }, CancellationToken.None)).ToList(); + await store.DeleteAsync("app:beta", CancellationToken.None); + var deleted = await store.FindAsync(new KeyValueFilter { Key = "app:beta" }, CancellationToken.None); + var remaining = (await store.FindManyAsync(new KeyValueFilter { Keys = ["app:alpha", "app:beta", "other:gamma"] }, CancellationToken.None)).ToList(); + + // Assert + Assert.True(TableExists("KeyValues")); + Assert.Equal("one", alpha?.SerializedValue); + Assert.Equal(["app:alpha", "app:beta"], many.Select(x => x.Key).Order().ToArray()); + Assert.Null(deleted); + Assert.Equal(["app:alpha", "other:gamma"], remaining.Select(x => x.Key).Order().ToArray()); + } + + [Fact(DisplayName = "#263: prefix FindMany returns only matching keys and exact Find still works")] + public async Task FreshMigratedDatabase_KeyValueStore_PrefixFindMany_ReturnsMatchingKeys() + { + // Arrange + MigrateUp(); + await using var services = CreateElsaServices(); + using var scope = services.CreateScope(); + using var tenant = scope.ServiceProvider.GetRequiredService().PushContext(Tenant.Default); + var store = scope.ServiceProvider.GetRequiredService(); + + await store.SaveAsync(Pair("app:1", "one"), CancellationToken.None); + await store.SaveAsync(Pair("app:2", "two"), CancellationToken.None); + await store.SaveAsync(Pair("other:1", "other"), CancellationToken.None); + + // Act + var prefixed = (await store.FindManyAsync(new KeyValueFilter { Key = "app:", StartsWith = true }, CancellationToken.None)).ToList(); + var exact = await store.FindAsync(new KeyValueFilter { Key = "app:1" }, CancellationToken.None); + var missing = await store.FindAsync(new KeyValueFilter { Key = "app:" }, CancellationToken.None); + + // Assert + Assert.Equal(["app:1", "app:2"], prefixed.Select(x => x.Key).Order().ToArray()); + Assert.DoesNotContain(prefixed, x => x.Key == "other:1"); + Assert.Equal("one", exact?.SerializedValue); + Assert.Null(missing); + } + + [Fact(DisplayName = "#263: the KV outbox store can Save and FindMany on a migration-built DB")] + public async Task FreshMigratedDatabase_OutboxStore_SaveAndFindMany() + { + // Arrange + MigrateUp(); + await using var services = CreateElsaServices(); + using var scope = services.CreateScope(); + using var tenant = scope.ServiceProvider.GetRequiredService().PushContext(Tenant.Default); + var outbox = scope.ServiceProvider.GetRequiredService(); + var item = new WorkflowDispatchOutboxItem + { + Id = "outbox-1", + OwnerWorkflowInstanceId = "wf-owner", + Kind = WorkflowDispatchOutboxItemKind.WorkflowDefinition, + CreatedAt = DateTimeOffset.Parse("2026-09-28T00:00:00+00:00") + }; + + // Act + await outbox.SaveAsync(item, CancellationToken.None); + var found = (await outbox.FindManyAsync(CancellationToken.None)).ToList(); + + // Assert + var loaded = Assert.Single(found); + Assert.Equal("outbox-1", loaded.Id); + Assert.Equal("wf-owner", loaded.OwnerWorkflowInstanceId); + } + + [Fact(DisplayName = "#264: a migration-built DB can enqueue a bookmark-queue item with Options and read it back")] + public async Task FreshMigratedDatabase_BookmarkQueueStore_RoundTripsOptions() + { + // Arrange + MigrateUp(); + await using var services = CreateElsaServices(); + using var scope = services.CreateScope(); + using var tenant = scope.ServiceProvider.GetRequiredService().PushContext(Tenant.Default); + var store = scope.ServiceProvider.GetRequiredService(); + var item = new BookmarkQueueItem + { + Id = "bq-1", + WorkflowInstanceId = "wf-1", + BookmarkId = "bm-1", + StimulusHash = "hash-1", + ActivityInstanceId = "act-1", + ActivityTypeName = "Elsa.Event", + CreatedAt = DateTimeOffset.Parse("2026-09-28T00:00:00+00:00"), + Options = new ResumeBookmarkOptions + { + Input = new Dictionary { ["payload"] = "hello" }, + Properties = new Dictionary { ["source"] = "test" } + } + }; + + // Act + await store.AddAsync(item, CancellationToken.None); + var loaded = await store.FindAsync(new BookmarkQueueFilter { Id = item.Id }, CancellationToken.None); + + // Assert + Assert.Contains("SerializedOptions", TableColumns("BookmarkQueueItems")); + Assert.NotNull(loaded); + Assert.NotNull(loaded.Options); + Assert.Equal("hello", loaded.Options.Input?["payload"]?.ToString()); + Assert.Equal("test", loaded.Options.Properties?["source"]?.ToString()); + } + + [Fact(DisplayName = "#264: Event start + resume completes on a migration-built Dapper DB")] + public async Task FreshMigratedDatabase_EventResumeWorkflow_ReachesFinished() + { + // Arrange + MigrateUp(); + await using var services = CreateElsaServices(elsa => elsa.AddWorkflow()); + using var scope = services.CreateScope(); + var sp = scope.ServiceProvider; + using var tenant = sp.GetRequiredService().PushContext(Tenant.Default); + await sp.GetRequiredService().PopulateAsync(); + var publisher = sp.GetRequiredService(); + var instances = sp.GetRequiredService(); + + // Act + await publisher.PublishAsync("probe-start"); + var afterStart = (await instances.FindManyAsync(new WorkflowInstanceFilter())).ToList(); + await publisher.PublishAsync("probe-resume"); + var afterResume = (await instances.FindManyAsync(new WorkflowInstanceFilter())).ToList(); + + // Assert + var started = Assert.Single(afterStart); + Assert.Equal(WorkflowStatus.Running, started.Status); + var finished = Assert.Single(afterResume); + Assert.Equal(WorkflowStatus.Finished, finished.Status); + } + + [Fact(DisplayName = "V3_9 is a no-op when KeyValues and SerializedOptions already exist")] + public void Migration_IsNoOp_WhenKeyValuesAndSerializedOptionsAlreadyExist() + { + // Arrange: stop at V3_7, then hand-create the objects a user may already have. + MigrateUp(20007); + Assert.False(TableExists("KeyValues")); + Assert.DoesNotContain("SerializedOptions", TableColumns("BookmarkQueueItems")); + + using (var connection = new SqliteConnection(_connectionString)) + { + connection.Execute(""" + create table KeyValues ( + Id text not null primary key, + TenantId text null, + Value text null + ); + insert into KeyValues (Id, TenantId, Value) values ('kept-key', '', 'kept-value'); + alter table BookmarkQueueItems add column SerializedOptions text null; + """); + } + + // Act + MigrateUp(); + + // Assert + Assert.True(TableExists("KeyValues")); + Assert.Contains("SerializedOptions", TableColumns("BookmarkQueueItems")); + using var read = new SqliteConnection(_connectionString); + var value = read.ExecuteScalar("select Value from KeyValues where Id = 'kept-key'"); + Assert.Equal("kept-value", value); + } + + public void Dispose() + { + File.Delete(_databasePath); + } + + private ServiceProvider CreateElsaServices(Action? extra = null) + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddElsa(elsa => + { + elsa.UseDapper(d => d.DbConnectionProvider = _ => new SqliteDbConnectionProvider(_connectionString)); + elsa.UseWorkflowManagement(m => m.UseDapper()); + elsa.UseWorkflowRuntime(r => r.UseDapper()); + extra?.Invoke(elsa); + }); + return services.BuildServiceProvider(); + } + + private void MigrateUp(long? targetVersion = null) + { + using var services = new ServiceCollection() + .AddFluentMigratorCore() + .ConfigureRunner(rb => rb + .AddSQLite() + .WithGlobalConnectionString(_connectionString) + .ScanIn(typeof(Elsa.Persistence.Dapper.Migrations.Runtime.Initial).Assembly).For.Migrations()) + .BuildServiceProvider(false); + using var scope = services.CreateScope(); + var runner = scope.ServiceProvider.GetRequiredService(); + if (targetVersion is { } version) + runner.MigrateUp(version); + else + runner.MigrateUp(); + } + + private IReadOnlyCollection TableColumns(string tableName) + { + using var connection = new SqliteConnection(_connectionString); + return connection.Query($"SELECT name FROM pragma_table_info('{tableName}')").ToList(); + } + + private bool TableExists(string tableName) + { + using var connection = new SqliteConnection(_connectionString); + return connection.ExecuteScalar( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = @tableName", + new { tableName }) == 1; + } + + private static SerializedKeyValuePair Pair(string key, string value) => new() + { + Key = key, + SerializedValue = value + }; + + private sealed class EventResumeProbeWorkflow : WorkflowBase + { + protected override void Build(IWorkflowBuilder builder) + { + builder.Root = new Sequence + { + Activities = + { + new Event("probe-start") { CanStartWorkflow = true }, + new Event("probe-resume"), + new WriteLine("resumed") + } + }; + } + } +} diff --git a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/Elsa.Persistence.Dapper.UnitTests.csproj b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/Elsa.Persistence.Dapper.UnitTests.csproj index 6b1868be..47666a73 100644 --- a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/Elsa.Persistence.Dapper.UnitTests.csproj +++ b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/Elsa.Persistence.Dapper.UnitTests.csproj @@ -1,6 +1,7 @@  + diff --git a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/NonPgQuerySqlSnapshotTests.cs b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/NonPgQuerySqlSnapshotTests.cs index fecd4a1c..49ee90cc 100644 --- a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/NonPgQuerySqlSnapshotTests.cs +++ b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/NonPgQuerySqlSnapshotTests.cs @@ -185,7 +185,7 @@ private static void AssertSnapshots(ISqlDialect dialect, IReadOnlyDictionary("IdStartsWith")); + Assert.DoesNotContain("Id", query.Parameters.ParameterNames.Where(name => name == "Id")); + } + + [Fact(DisplayName = "IsNullOrEmpty matches NULL or empty string for the default tenant")] + public void IsNullOrEmpty_MatchesNullOrEmptyTenantId() + { + var query = new ParameterizedQuery(new SqliteDialect()) + .From("KeyValues") + .IsNullOrEmpty("TenantId"); + + Assert.Contains("and (TenantId is null or TenantId = '')", query.Sql.ToString(), StringComparison.Ordinal); + } + + [Fact(DisplayName = "StartsWith quotes the identifier on PostgreSQL and binds @{field}StartsWith")] + public void StartsWith_PostgreSql_QuotesIdentifierAndBindsFieldPrefixedParameter() + { + var query = new ParameterizedQuery(new PostgreSqlDialect()) + .From("KeyValues") + .StartsWith("Id", true, "app:"); + + var sql = query.Sql.ToString(); + + Assert.Contains("and \"Id\" like @IdStartsWith", sql, StringComparison.Ordinal); + Assert.DoesNotContain("@SearchTermLike", sql, StringComparison.Ordinal); + Assert.Equal("app:%", query.Parameters.Get("IdStartsWith")); + } + [Fact] public async Task In_WithTenantAgnosticFilter_ReturnsMatchingRowsAcrossTenants() { diff --git a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/PostgreSqlDialectTests.cs b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/PostgreSqlDialectTests.cs index 8bf7c1d4..1a92eada 100644 --- a/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/PostgreSqlDialectTests.cs +++ b/test/modules/persistence/Elsa.Persistence.Dapper.UnitTests/PostgreSqlDialectTests.cs @@ -68,6 +68,7 @@ public void QueryBuilder_QuotesInlinedIdentifiers() Assert.Contains("and \"IsLatest\" = true", queries["version-latest"], StringComparison.Ordinal); Assert.Contains("and \"IsPublished\" = false", queries["version-draft"], StringComparison.Ordinal); Assert.Contains("\"Name\" like @SearchTermLike", queries["definition-search"], StringComparison.Ordinal); + Assert.Contains("and \"Name\" like @NameStartsWith", queries["starts-with"], StringComparison.Ordinal); Assert.Contains("\"Id\" like @SearchTerm", queries["instance-search"], StringComparison.Ordinal); Assert.DoesNotContain("\"ID\"", queries["instance-search"], StringComparison.Ordinal); Assert.Contains("and \"Id\" in (", queries["paged-delete"], StringComparison.Ordinal);