Integration Events in Modular Monolith

Question

I am trying to implement a multitenant (using a database per tenant) modular monolith that uses integration events to communicate between modules.

I am attempting to implement the integration events functionality and have the following scenario:

1. A new integration event is raised via an HTTP request
2. The integration event is published to the event bus
3. An integration event handler receives the message from the bus and modifies a domain object
4. The modification to the domain object causes a domain event to be raised
5. A domain event handler receives the domain event and performs some read operations on the database

This all works fine if I ignore the multitenant requirement, I am struggling to understand how to implement this for multiple tenants. I am using Finbuckle in my API and constructor injecting ITenantInfo into my DbContext to retrieve the correct connection string for the current tenant.

When I try to move the scenario to multitenant:

• Step 1 is easy enough – I can simply pass a tenant identifier in the integration event.

• Step 2 is unaffected – the message is published to the event bus.

• Step 3 is tricky because it is asynchronous. If it was within the scope of the HTTP request, I would be able to inject the domain repository into the constructor of my integration event handler and retrieve the domain object from the database. However, because it is triggered from the event bus, the tenant identifier must be extracted from the message and manually resolved. Once I have resolved the tenant, I can manually create an instance of the domain repository and get the domain object from the database.

• Step 4 is not affected; the domain object is updated as usual.

• Step 5 is where I am struggling. I am constructor injecting the DbContext into the domain event handler which, before I introduced integration events, worked because ITenantInfo was resolved by the Finbuckle middleware and DI. The problem I have now is that I do not know how to resolve the tenant when the domain event is triggered via an integration event. My domain event does not contain a tenant identifier and there is no HttpContext therefore I cannot resolve ITenantInfo to provide the database connect string.

To clarify, the final step of my scenario requires the domain event handler to read data from the database via a DbContext. This is possible within the scope of an HTTP request because Finbuckle will resolve the ITenantInfo to provide the connection string. When outside the scope of an HTTP request, I do not know how to resolve the tenant because my domain event does not contain any tenant specific information.

How can I identify the tenant in a domain event handler when the domain event is raised as a side effect of an integration event?

Answer 1

For anybody that sees this in the future I was overthinking the problem.

I was able to resolve ITenantInfo using Scoped Filters in MassTransit.

For example:

public class MyTenantInfoConsumeFilter<T>
  : IFilter<ConsumeContext<T>>
  where T : class
{
  readonly MyTenantInfo _tenantInfo;
  private readonly MultiTenantStoreDbContext _dbContext;

  public MyTenantInfoConsumeFilter(MyTenantInfo tenantInfo, MultiTenantStoreDbContext dbContext)
  {
    _tenantInfo = tenantInfo;
    _dbContext = dbContext;
  }

  public Task Send(ConsumeContext<T> context, IPipe<ConsumeContext<T>> next)
  {
    if (context.Message is IntegrationEventBase message)
    {
      Map(_dbContext.TenantInfo.FirstOrDefault(tenant => tenant.Identifier == message.TenantIdentifier));
    }

    return next.Send(context);
  }

  public void Probe(ProbeContext context)
  {
  }

  private void Map(MyTenantInfo? tenantInfo)
  {
    if (tenantInfo is null) return;

    _tenantInfo.Id = tenantInfo.Id;
    _tenantInfo.Identifier = tenantInfo.Identifier;
    _tenantInfo.Name = tenantInfo.Name;
    _tenantInfo.ConnectionString = tenantInfo.ConnectionString;
  }
}

Then register the services using something like:

services.AddScoped<Finbuckle.MultiTenant.ITenantInfo>(sp => sp.GetService<MyTenantInfo>()!);

services.AddMassTransit(x =>
{
    x.AddScoped<MyTenantInfo>();

    x.UsingAzureServiceBus((context, cfg) =>
    {
      cfg.Host(hostContext.Configuration.GetConnectionString("AzureServiceBusConnection"));

      cfg.ConfigureEndpoints(context);

      cfg.UseConsumeFilter(typeof(MyTenantInfoConsumeFilter<>), context);
    });
});