Fix Tenant ID propagation for MassTransit

src/modules/Elsa.Common/Multitenancy/Models/TenantScope.cs

@@ -8,21 +8,20 @@ namespace Elsa.Common.Multitenancy;
 /// </summary>
 public class TenantScope : IAsyncDisposable
 {
-    private readonly IServiceScope _serviceScope;
-
     public TenantScope(IServiceScope serviceScope, ITenantAccessor tenantAccessor, Tenant? tenant)
     {
-        _serviceScope = serviceScope;
+        ServiceScope = serviceScope;
         tenantAccessor.Tenant = tenant;
     }
 
-    public IServiceProvider ServiceProvider => _serviceScope.ServiceProvider;
+    public IServiceScope ServiceScope { get; }
+    public IServiceProvider ServiceProvider => ServiceScope.ServiceProvider;
 
     public async ValueTask DisposeAsync()
     {
-        if (_serviceScope is IAsyncDisposable asyncDisposable)
+        if (ServiceScope is IAsyncDisposable asyncDisposable)
             await asyncDisposable.DisposeAsync();
         else
-            _serviceScope.Dispose();
+            ServiceScope.Dispose();
     }
 }

src/modules/Elsa.MassTransit/Consumers/WorkflowMessageConsumer.cs

@@ -1,42 +1,34 @@
 using Elsa.MassTransit.Activities;
 using Elsa.Workflows.Helpers;
 using Elsa.Workflows.Runtime;
-using Elsa.Workflows.Runtime.Contracts;
-using Elsa.Workflows.Runtime.Requests;
 using MassTransit;
 
 namespace Elsa.MassTransit.Consumers;
 
 /// <summary>
-/// A consumer of various dispatch message types to asynchronously execute workflows.
+/// A consumer of various message types to trigger activities derived from these messages.
 /// </summary>
-public class WorkflowMessageConsumer<T> : IConsumer<T> where T : class
+public class WorkflowMessageConsumer<T>(IStimulusSender stimulusSender) : IConsumer<T>
+    where T : class
 {
-    private readonly IWorkflowDispatcher _workflowRuntime;
-
-    /// <summary>
-    /// Constructor.
-    /// </summary>
-    public WorkflowMessageConsumer(IWorkflowDispatcher workflowRuntime)
-    {
-        _workflowRuntime = workflowRuntime;
-    }
-
     /// <inheritdoc />
     public async Task Consume(ConsumeContext<T> context)
     {
         var cancellationToken = context.CancellationToken;
         var messageType = typeof(T);
         var message = context.Message;
         var activityTypeName = ActivityTypeNameHelper.GenerateTypeName(messageType);
-        var bookmark = new MessageReceivedBookmarkPayload(messageType);
+        var stimulus = new MessageReceivedBookmarkPayload(messageType);
         var correlationId = context.CorrelationId?.ToString();
-        var input = new Dictionary<string, object> { [MessageReceived.InputKey] = message };
-        var request = new DispatchTriggerWorkflowsRequest(activityTypeName, bookmark)
+        var input = new Dictionary<string, object>
+        {
+            [MessageReceived.InputKey] = message
+        };
+        var stimulusMetadata = new StimulusMetadata
         {
             CorrelationId = correlationId,
             Input = input
         };
-        await _workflowRuntime.DispatchAsync(request, cancellationToken: cancellationToken);
+        await stimulusSender.SendAsync(activityTypeName, stimulus, stimulusMetadata, cancellationToken);
     }
 }

src/modules/Elsa.MassTransit/Middleware/TenantConsumeMiddleware.cs

@@ -1,9 +1,10 @@
 using Elsa.Common.Multitenancy;
 using MassTransit;
+using MassTransit.DependencyInjection;
 
 namespace Elsa.MassTransit.Middleware;
 
-public class TenantConsumeMiddleware<T>(ITenantContextInitializer tenantContextInitializer) : IFilter<ConsumeContext<T>> where T : class
+public class TenantConsumeMiddleware<T>(ITenantFinder tenantFinder, ITenantScopeFactory tenantScopeFactory, Bind<IBus, ISetScopedConsumeContext> scopeSetter) : IFilter<ConsumeContext<T>> where T : class
 {
     public void Probe(ProbeContext context)
     {
@@ -12,9 +13,16 @@ public void Probe(ProbeContext context)
 
     public async Task Send(ConsumeContext<T> context, IPipe<ConsumeContext<T>> next)
     {
-        if (context.Headers.TryGetHeader(HeaderNames.TenantId, out var tenantId) && tenantId is string tenantIdString) 
-            await tenantContextInitializer.InitializeAsync(tenantIdString, context.CancellationToken);
-
-        await next.Send(context);
+        if (context.Headers.TryGetHeader(HeaderNames.TenantId, out var tenantId) && tenantId is string tenantIdString)
+        {
+            var tenant = await tenantFinder.FindByIdAsync(tenantIdString);
+            await using var tenantScope = tenantScopeFactory.CreateScope(tenant);
+            using var scope = scopeSetter.Value.PushContext(tenantScope.ServiceScope, context);
+            await next.Send(context);
+        }
+        else
+        {
+            await next.Send(context);
+        }
     }
 }