Multi-tenant messaging: filter out events from unknown tenants

src/Messaging/NBB.Messaging.MultiTenancy/NBB.Messaging.MultiTenancy.csproj

@@ -15,6 +15,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="$(MicrosoftExtensionsPackagesVersion)" />
     <PackageReference Include="Microsoft.Extensions.Options" Version="$(MicrosoftExtensionsPackagesVersion)" />
+		<PackageReference Include="MediatR.Contracts" Version="$(MediatRContractsPackageVersion)" />
   </ItemGroup>
 
 </Project>

src/Messaging/NBB.Messaging.MultiTenancy/TenantMiddleware.cs

@@ -13,6 +13,9 @@
 using NBB.MultiTenancy.Identification.Services;
 using NBB.MultiTenancy.Abstractions;
 using System.Diagnostics;
+using Microsoft.Extensions.Logging;
+using MediatR;
+using System.Linq;
 
 namespace NBB.Messaging.MultiTenancy
 {
@@ -21,45 +24,80 @@ namespace NBB.Messaging.MultiTenancy
     /// obtained from the current identification strategy and builds the tenant context.
     /// </summary>
     /// <seealso cref="IPipelineMiddleware{MessagingEnvelope}" />
-    public class TenantMiddleware : IPipelineMiddleware<MessagingContext>
-    {
-        private readonly ITenantContextAccessor _tenantContextAccessor;
-        private readonly ITenantIdentificationService _tenantIdentificationService;
-        private readonly IOptions<TenancyHostingOptions> _tenancyOptions;
-        private readonly ITenantRepository _tenantRepository;
-
-        public TenantMiddleware(ITenantContextAccessor tenantContextAccessor, ITenantIdentificationService tenantIdentificationService, IOptions<TenancyHostingOptions> tenancyOptions, ITenantRepository tenantRepository)
-        {
-            _tenantContextAccessor = tenantContextAccessor;
-            _tenantIdentificationService = tenantIdentificationService;
-            _tenancyOptions = tenancyOptions;
-            _tenantRepository = tenantRepository;
-        }
-
+    public class TenantMiddleware(
+        ITenantContextAccessor tenantContextAccessor,
+        ITenantIdentificationService tenantIdentificationService,
+        IOptions<TenancyHostingOptions> tenancyOptions,
+        ITenantRepository tenantRepository,
+        ILogger<TenantMiddleware> logger
+    ) : IPipelineMiddleware<MessagingContext>
+    {        
         public async Task Invoke(MessagingContext context, CancellationToken cancellationToken, Func<Task> next)
         {
-            if (_tenantContextAccessor.TenantContext != null)
+            if (tenantContextAccessor.TenantContext != null)
             {
                 throw new ApplicationException("Tenant context is already set");
             }
 
-            if (_tenancyOptions.Value.TenancyType == TenancyType.MonoTenant)
+            if (tenancyOptions.Value.TenancyType == TenancyType.MonoTenant)
             {
-                 _tenantContextAccessor.TenantContext = new TenantContext(Tenant.Default);
+                tenantContextAccessor.TenantContext = new TenantContext(Tenant.Default);
                 await next();
                 return;
             }
 
-            var tenantId = await _tenantIdentificationService.GetTenantIdAsync();
-            var tenant = await _tenantRepository.Get(tenantId, cancellationToken)
-                         ?? throw new ApplicationException($"Tenant {tenantId} not found");
+            Tenant tenant;
+
+            if (context.MessagingEnvelope.Payload is INotification)
+            {
+                tenant = await TryLoadTenant(context.TopicName, cancellationToken);
+                if (tenant == null)
+                {
+                    return;
+                }
+            }
+            else
+            {
+                tenant = await LoadTenant(cancellationToken);
+            }
 
-             _tenantContextAccessor.TenantContext = new TenantContext(tenant);
 
-            Activity.Current?.SetTag(TracingTags.TenantId, tenantId);
+            tenantContextAccessor.TenantContext = new TenantContext(tenant);
+
+            Activity.Current?.SetTag(TracingTags.TenantId, tenant.TenantId);
 
             await next();
         }
+
+        private async Task<Tenant> LoadTenant(CancellationToken cancellationToken)
+        {
+            var tenantId = await tenantIdentificationService.GetTenantIdAsync();
+            var tenant = await tenantRepository.Get(tenantId, cancellationToken)
+                            ?? throw new ApplicationException($"Tenant {tenantId} not found");
+
+            return tenant;
+        }
+
+
+        private async Task<Tenant> TryLoadTenant(string topic, CancellationToken cancellationToken)
+        {
+            var tenantId = await tenantIdentificationService.TryGetTenantIdAsync();
+            if (!tenantId.HasValue)
+            {
+                logger.LogDebug("Tenant could not be identified. Message {Topic} will be ignored.", topic);
+                return null;
+            }
+
+            var tenant = await tenantRepository.TryGet(tenantId.Value, cancellationToken);
+            if (tenant == null)
+            {
+                logger.LogDebug("Tenant {Tenant} not found or not enabled. Message {Topic} will be ignored.", tenantId.Value, topic);
+                return null;
+            }
+
+            return tenant;
+        }
+
     }
 
 

src/MultiTenancy/NBB.MultiTenancy.Abstractions/Repositories/CachedTenantRepositoryDecorator.cs

@@ -23,16 +23,16 @@ public CachedTenantRepositoryDecorator(ITenantRepository tenantRepository, IDist
             _cache = cache;
         }
 
-        public async Task<Tenant> Get(Guid id, CancellationToken token)
+        public async Task<Tenant> TryGet(Guid id, CancellationToken token)
         {
             var cacheKey = CacheTenantByIdKey(id);
             var cachedTenant = await GetTenantFromCache(cacheKey, token);
             if (cachedTenant != null)
             {
                 return cachedTenant;
             }
-                        
-            var dbTenant = await _tenantRepository.Get(id, token);
+
+            var dbTenant = await _tenantRepository.TryGet(id, token);
             if (dbTenant == null)
             {
                 return null;

src/MultiTenancy/NBB.MultiTenancy.Abstractions/Repositories/ConfigurationTenantRepository.cs

@@ -43,15 +43,14 @@ public ConfigurationTenantRepository(IConfiguration configuration, IOptions<Tena
             };
         }
 
-        public Task<Tenant> Get(Guid id, CancellationToken token = default)
+        public Task<Tenant> TryGet(Guid id, CancellationToken token = default)
         {
-            if (!tenantMap.TryGetValue(id, out var result))
+            if (!tenantMap.TryGetValue(id, out var result) || !result.Enabled)
             {
-                throw new TenantNotFoundException(id);
+                return Task.FromResult(default(Tenant));
             }
 
-
-            return Task.FromResult(result.Enabled ? result : throw new Exception($"Tenant {result.Code} is disabled "));
+            return Task.FromResult(result);
         }
 
         public Task<Tenant> GetByHost(string host, CancellationToken token = default)

src/MultiTenancy/NBB.MultiTenancy.Abstractions/Repositories/ITenantRepository.cs

@@ -10,8 +10,11 @@ namespace NBB.MultiTenancy.Abstractions.Repositories
 {
     public interface ITenantRepository
     {
-        Task<Tenant> Get(Guid id, CancellationToken token = default);
+        Task<Tenant> TryGet(Guid id, CancellationToken token = default);
         Task<Tenant> GetByHost(string host, CancellationToken token = default);
         Task<List<Tenant>> GetAll(CancellationToken token = default);
+
+        async Task<Tenant> Get(Guid id, CancellationToken token = default)
+            => await TryGet(id, token) ?? throw new TenantNotFoundException(id);
     }
 }

test/UnitTests/MultiTenancy/NBB.MultiTenancy.Configuration.Tests/ConfigurationTenantRepositoryTests.cs

@@ -190,14 +190,14 @@ public async Task get_tenant_should_throw_for_disabled_tenant()
 
             var options = new OptionsWrapper<TenancyHostingOptions>(tenancyHostingOptions);
 
-            var repo = new ConfigurationTenantRepository(configuration, options);
+            ITenantRepository repo = new ConfigurationTenantRepository(configuration, options);
 
             //Act
             Func<Task> action = async() => 
                 await repo.Get(Guid.Parse(tenantId));
 
             //Assert
-            await action.Should().ThrowAsync<Exception>().WithMessage("*disabled*");
+            await action.Should().ThrowAsync<TenantNotFoundException>();
         }
 
         [Fact]
@@ -232,7 +232,7 @@ public async Task get_should_bind_tenant_code_from_section_name()
             var repo = new ConfigurationTenantRepository(configuration, options);
 
             //arrange
-            var actual = await repo.Get(System.Guid.Parse("ef8d5362-9969-4e02-8794-0d1af56816f6"));
+            var actual = await repo.TryGet(Guid.Parse("ef8d5362-9969-4e02-8794-0d1af56816f6"));
 
             // Assert
             actual.Should().NotBeNull();