diff --git a/src/modules/Elsa.Workflows.Runtime/Stores/MemoryTriggerStore.cs b/src/modules/Elsa.Workflows.Runtime/Stores/MemoryTriggerStore.cs index 88e454146..55aee2ad7 100644 --- a/src/modules/Elsa.Workflows.Runtime/Stores/MemoryTriggerStore.cs +++ b/src/modules/Elsa.Workflows.Runtime/Stores/MemoryTriggerStore.cs @@ -1,5 +1,6 @@ using Elsa.Common.Entities; using Elsa.Common.Models; +using Elsa.Common.Multitenancy; using Elsa.Common.Services; using Elsa.Extensions; using Elsa.Workflows.Runtime.Entities; @@ -14,26 +15,48 @@ namespace Elsa.Workflows.Runtime.Stores; public class MemoryTriggerStore : ITriggerStore { private readonly MemoryStore _store; + private readonly ITenantAccessor? _tenantAccessor; /// /// Initializes a new instance of the class. /// - public MemoryTriggerStore(MemoryStore store) + public MemoryTriggerStore(MemoryStore store, ITenantAccessor? tenantAccessor = null) { _store = store; + _tenantAccessor = tenantAccessor; } /// public ValueTask SaveAsync(StoredTrigger record, CancellationToken cancellationToken = default) { - _store.Save(record, x => x.Id); + lock (_store.Sync) + { + ApplyCurrentTenant(record); + EnsureLogicalKeyAvailable(record); + _store.Save(record, x => x.Id); + } + return new(); } /// public ValueTask SaveManyAsync(IEnumerable records, CancellationToken cancellationToken = default) { - _store.SaveMany(records, x => x.Id); + lock (_store.Sync) + { + var recordList = records.ToList(); + + foreach (var record in recordList) + ApplyCurrentTenant(record); + + var uniqueRecords = DistinctByLogicalKey(recordList).ToList(); + + foreach (var record in uniqueRecords) + EnsureLogicalKeyAvailable(record); + + _store.SaveMany(uniqueRecords, x => x.Id); + } + return new(); } @@ -66,15 +89,93 @@ public class MemoryTriggerStore : ITriggerStore /// public ValueTask ReplaceAsync(IEnumerable removed, IEnumerable added, CancellationToken cancellationToken = default) { - _store.DeleteMany(removed, x => x.Id); - _store.SaveMany(added, x => x.Id); + lock (_store.Sync) + { + var removedList = removed.ToList(); + var addedList = added.ToList(); + + foreach (var trigger in addedList) + ApplyCurrentTenant(trigger); + + addedList = DistinctByLogicalKey(addedList).ToList(); + + if (removedList.Count > 0) + _store.DeleteMany(removedList, x => x.Id); + + if (addedList.Count == 0) + return new(); + + var newTriggers = GetMissingLogicalTriggers(addedList); + + if (newTriggers.Count == 0) + return new(); + + _store.SaveMany(newTriggers, x => x.Id); + } + return new(); } /// - public async ValueTask DeleteManyAsync(TriggerFilter filter, CancellationToken cancellationToken = default) + public ValueTask DeleteManyAsync(TriggerFilter filter, CancellationToken cancellationToken = default) { - var ids = (await FindManyAsync(filter, cancellationToken)).Select(x => x.Id); - return _store.DeleteMany(ids); + lock (_store.Sync) + { + var ids = _store.Query(filter.Apply).Select(x => x.Id).ToList(); + return new(_store.DeleteMany(ids)); + } } -} \ No newline at end of file + + private void EnsureLogicalKeyAvailable(StoredTrigger record) + { + var logicalKey = GetLogicalKey(record); + var existing = _store.Find(x => x.Id != record.Id && GetLogicalKey(x) == logicalKey); + + if (existing is not null) + { + throw new InvalidOperationException( + $"A stored trigger already exists for workflow '{record.WorkflowDefinitionId}', hash '{record.Hash}', activity '{record.ActivityId}', tenant '{record.TenantId}'."); + } + } + + private List GetMissingLogicalTriggers(ICollection triggers) + { + var existingKeys = GetExistingLogicalKeys(triggers); + return triggers + .Where(trigger => !existingKeys.Contains(GetLogicalKey(trigger))) + .ToList(); + } + + private HashSet GetExistingLogicalKeys(ICollection triggers) + { + var workflowDefinitionIds = triggers.Select(x => x.WorkflowDefinitionId).Distinct().ToHashSet(); + return _store + .FindMany(trigger => workflowDefinitionIds.Contains(trigger.WorkflowDefinitionId)) + .Select(GetLogicalKey) + .ToHashSet(); + } + + private void ApplyCurrentTenant(StoredTrigger trigger) + { + if (trigger.TenantId == Tenant.AgnosticTenantId || _tenantAccessor is null) + return; + + trigger.TenantId ??= _tenantAccessor.TenantId; + } + + private static IEnumerable DistinctByLogicalKey(IEnumerable triggers) + { + var seen = new HashSet(); + + foreach (var trigger in triggers) + { + if (seen.Add(GetLogicalKey(trigger))) + yield return trigger; + } + } + + private static TriggerLogicalKey GetLogicalKey(StoredTrigger trigger) => + new(trigger.WorkflowDefinitionId, trigger.Hash, trigger.ActivityId, trigger.TenantId); + + private readonly record struct TriggerLogicalKey(string WorkflowDefinitionId, string? Hash, string ActivityId, string? TenantId); +} diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Stores/MemoryTriggerStoreTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Stores/MemoryTriggerStoreTests.cs new file mode 100644 index 000000000..624c2f4bc --- /dev/null +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Stores/MemoryTriggerStoreTests.cs @@ -0,0 +1,228 @@ +using Elsa.Common.Multitenancy; +using Elsa.Common.Services; +using Elsa.Testing.Shared.Multitenancy; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Stores; + +namespace Elsa.Workflows.Runtime.UnitTests.Stores; + +/// +/// Memory must enforce the same logical uniqueness that EF Core +/// already enforces via the unique index on (WorkflowDefinitionId, Hash, ActivityId, TenantId) +/// and that EFCoreTriggerStore.ReplaceAsync mirrors. +/// +public class MemoryTriggerStoreTests +{ + [Fact(DisplayName = "ReplaceAsync keeps the first added record when the incoming batch repeats a logical key")] + public async Task ReplaceAsync_WhenAddedBatchRepeatsLogicalKey_StoresOnlyTheFirst() + { + var store = CreateStore(); + var first = Trigger("id-1"); + var duplicate = Trigger("id-2"); + + await store.ReplaceAsync([], [first, duplicate]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + var match = Assert.Single(stored); + Assert.Equal("id-1", match.Id); + } + + [Fact(DisplayName = "ReplaceAsync skips an added record whose logical key is already present")] + public async Task ReplaceAsync_WhenLogicalKeyAlreadyPresent_SkipsTheAddedRecord() + { + var store = CreateStore(); + var existing = Trigger("existing-id", hash: "hash-1"); + await store.SaveAsync(existing); + + await store.ReplaceAsync([], [Trigger("new-id", hash: "hash-1")]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + var match = Assert.Single(stored); + Assert.Equal("existing-id", match.Id); + Assert.Equal("v1", match.WorkflowDefinitionVersionId); + } + + [Fact(DisplayName = "ReplaceAsync can insert a new Id after the existing logical-key row is removed")] + public async Task ReplaceAsync_WhenExistingLogicalKeyIsRemoved_InsertsTheReplacement() + { + var store = CreateStore(); + var existing = Trigger("existing-id", hash: "hash-1"); + await store.SaveAsync(existing); + + var replacement = Trigger("new-id", hash: "hash-1", workflowDefinitionVersionId: "v2"); + await store.ReplaceAsync([existing], [replacement]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + var match = Assert.Single(stored); + Assert.Equal("new-id", match.Id); + Assert.Equal("v2", match.WorkflowDefinitionVersionId); + } + + [Fact(DisplayName = "ReplaceAsync stamps the current tenant when TenantId is unset")] + public async Task ReplaceAsync_WhenTenantIdIsUnset_AppliesCurrentTenant() + { + var store = CreateStore(new TestTenantAccessor("tenant-a")); + var trigger = Trigger("id-1"); + trigger.TenantId = null; + + await store.ReplaceAsync([], [trigger]); + + var stored = await store.FindAsync(new TriggerFilter { Id = "id-1" }); + Assert.Equal("tenant-a", stored!.TenantId); + } + + [Fact(DisplayName = "ReplaceAsync leaves a tenant-agnostic TenantId untouched")] + public async Task ReplaceAsync_WhenTenantIdIsAgnostic_DoesNotOverwriteTenant() + { + var store = CreateStore(new TestTenantAccessor("tenant-a")); + var trigger = Trigger("id-1"); + trigger.TenantId = Tenant.AgnosticTenantId; + + await store.ReplaceAsync([], [trigger]); + + var stored = await store.FindAsync(new TriggerFilter { Id = "id-1" }); + Assert.Equal(Tenant.AgnosticTenantId, stored!.TenantId); + } + + [Fact(DisplayName = "ReplaceAsync treats different tenants as distinct logical keys")] + public async Task ReplaceAsync_WhenTenantIdsDiffer_StoresBothRecords() + { + var store = CreateStore(); + var tenantA = Trigger("id-a"); + tenantA.TenantId = "tenant-a"; + var tenantB = Trigger("id-b"); + tenantB.TenantId = "tenant-b"; + + await store.ReplaceAsync([], [tenantA, tenantB]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal(2, stored.Count); + Assert.Contains(stored, x => x.Id == "id-a"); + Assert.Contains(stored, x => x.Id == "id-b"); + } + + [Fact(DisplayName = "SaveAsync stamps the current tenant when TenantId is unset")] + public async Task SaveAsync_WhenTenantIdIsUnset_AppliesCurrentTenant() + { + var store = CreateStore(new TestTenantAccessor("tenant-a")); + var trigger = Trigger("id-1"); + trigger.TenantId = null; + + await store.SaveAsync(trigger); + + var stored = await store.FindAsync(new TriggerFilter { Id = "id-1" }); + Assert.Equal("tenant-a", stored!.TenantId); + } + + [Fact(DisplayName = "SaveAsync updates the existing row when the Id matches")] + public async Task SaveAsync_WhenIdMatches_UpdatesTheExistingRow() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + + await store.SaveAsync(Trigger("id-1", hash: "hash-2", workflowDefinitionVersionId: "v2")); + + var stored = await store.FindAsync(new TriggerFilter { Id = "id-1" }); + Assert.Equal("hash-2", stored!.Hash); + Assert.Equal("v2", stored.WorkflowDefinitionVersionId); + } + + [Fact(DisplayName = "SaveAsync rejects a different Id that repeats an existing logical key")] + public async Task SaveAsync_WhenLogicalKeyExistsUnderAnotherId_Throws() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + + var exception = await Assert.ThrowsAsync(() => + store.SaveAsync(Trigger("id-2", hash: "hash-1")).AsTask()); + + Assert.Contains("already exists", exception.Message); + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal("id-1", Assert.Single(stored).Id); + } + + [Fact(DisplayName = "SaveManyAsync keeps the first record when the batch repeats a logical key")] + public async Task SaveManyAsync_WhenBatchRepeatsLogicalKey_StoresOnlyTheFirst() + { + var store = CreateStore(); + + await store.SaveManyAsync([Trigger("id-1"), Trigger("id-2")]); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal("id-1", Assert.Single(stored).Id); + } + + [Fact(DisplayName = "SaveManyAsync rejects a batch whose logical key is already present under another Id")] + public async Task SaveManyAsync_WhenLogicalKeyExistsUnderAnotherId_Throws() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + + var exception = await Assert.ThrowsAsync(() => + store.SaveManyAsync([Trigger("id-2", hash: "hash-1")]).AsTask()); + + Assert.Contains("already exists", exception.Message); + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal("id-1", Assert.Single(stored).Id); + } + + [Fact(DisplayName = "FindAsync returns the first record when a filter matches several distinct triggers")] + public async Task FindAsync_WhenFilterMatchesMultipleTriggers_ReturnsTheFirst() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + await store.SaveAsync(Trigger("id-2", hash: "hash-2")); + + var found = await store.FindAsync(new TriggerFilter { WorkflowDefinitionId = "workflow-1" }); + + Assert.True(found!.Id is "id-1" or "id-2"); + } + + [Fact(DisplayName = "FindAsync returns the matching record when the filter is unique")] + public async Task FindAsync_WhenOneLogicalKeyMatches_ReturnsThatRecord() + { + var store = CreateStore(); + await store.SaveAsync(Trigger("id-1", hash: "hash-1")); + await store.SaveAsync(Trigger("id-2", hash: "hash-2")); + + var found = await store.FindAsync(new TriggerFilter { Hash = "hash-1" }); + + Assert.Equal("id-1", found!.Id); + } + + [Fact(DisplayName = "SaveAsync keeps distinct triggers whose fields contain the old delimiter character")] + public async Task SaveAsync_WhenFieldsContainUnitSeparator_DoesNotCollide() + { + var store = CreateStore(); + var first = Trigger("id-1", hash: "b\u001fc"); + var second = Trigger("id-2", workflowDefinitionId: "workflow-1\u001fb", hash: "c"); + + await store.SaveAsync(first); + await store.SaveAsync(second); + + var stored = (await store.FindManyAsync(new TriggerFilter())).ToList(); + Assert.Equal(2, stored.Count); + Assert.Contains(stored, x => x.Id == "id-1"); + Assert.Contains(stored, x => x.Id == "id-2"); + } + + private static MemoryTriggerStore CreateStore(ITenantAccessor? tenantAccessor = null) => + new(new MemoryStore(), tenantAccessor ?? TestTenantAccessor.Default); + + private static StoredTrigger Trigger( + string id, + string workflowDefinitionId = "workflow-1", + string workflowDefinitionVersionId = "v1", + string activityId = "activity-1", + string hash = "hash-1") => + new() + { + Id = id, + WorkflowDefinitionId = workflowDefinitionId, + WorkflowDefinitionVersionId = workflowDefinitionVersionId, + ActivityId = activityId, + Hash = hash, + Name = "Elsa.HttpEndpoint" + }; +}