Best Approach for multi-tenancy

[ymoreiratiti]

What is the best approach for multi-tenancy on TypeORM?

Currently, I created a TenantModule (Credits to Esposito Medium post to create and change de TypeORM connection on the fly.

But this doesn't seem right, since this ties the TypeORM connection and other sources to a general module instead of let each data source handle its way to multitenancy.

What do you think is the best practice for this?

[Sairyss]

I used this package: https://github.com/Avallone-io/rls

[khunder]

There is no silver bullet for multi tenancy, if you want RLS avallone is your best/quick choice,
If you want a separate database/schema it's going to be more complicated.
you need to define a connection provider that accept teantId as parameter,

The example below is not complete and will help you understand the complexity of the changes if you want to go full tenancy, with a separate schema, database.

you need to get teantId from controllers then cascade it to the repository, to do so you need,

• parse Url to get tenantId, in case of message borker, you need to include teant Id in the message body
• send the message in the service bus
• query/command handler will unbox the tenant id from the message/model then it will use it to call a factory to get a repository instance.

example: controller

 async getAssetById(
    @Param('id') id: string,
    @TenantId() tenantId: string, // decorator used to resolve tenant id form url like operatorX.example.com --> split('.')[0]
  ): Promise<AssetHttpResponse> {
    const query = new GetAssetQuery({ tenantId, id });
    const result: Result<AssetEntity> = await this.queryBus.execute(query);

    /* Returning Response classes which are responsible
       for whitelisting data that is sent to the asset */
    return new AssetHttpResponse(result.unwrap());
  }

src/libs/decorators/tenant-id.decorator.ts

import { createParamDecorator, ExecutionContext } from '@nestjs/common';

function resolveOperatorNameFromURl(ctx: ExecutionContext) {
  const req = ctx.switchToHttp().getRequest();
  return req.headers.host.split('.')[0];
}

export const TenantId = createParamDecorator((data, ctx: ExecutionContext) => {
  if (process.env.PROFILE === 'TEST') {
    return process.env.DEFAULT_OPERATOR;
  }
  return resolveOperatorNameFromURl(ctx);
});

*** Query handler ***
Be Aware, Nest CQRS doesn't support Request scope injection, so instead we inject a singleton instance in a form of Factory .

import { QueryHandlerBase } from '@libs/ddd/domain/base-classes/query-handler.base';
import { Result } from '@libs/ddd/domain/utils/result.util';
import { AssetEntity } from '@modules/asset/domain/entities/asset.entity';
import { AssetFactoryPort } from '@modules/asset/ports/asset-factory.port';
import { GetAssetQuery } from '@modules/asset/queries/get-asset/get-asset.query';
import { Inject } from '@nestjs/common';
import { QueryHandler } from '@nestjs/cqrs';

@QueryHandler(GetAssetQuery)
export class GetAssetQueryHandler extends QueryHandlerBase {
  constructor(
    @Inject('AssetFactoryPort')
    private assetFactory: AssetFactoryPort, // assetFactory is a factory class we use to create tenant repositories/objects
  ) {
    super();
  }

  /* Since this is a simple query with no additional business
     logic involved, it bypasses application's core completely 
     and retrieves assets directly from a repository.
   */
  async handle(query: GetAssetQuery): Promise<Result<AssetEntity>> {
    const { tenantId } = query;
    const asset = await this.assetFactory
      .getAssetReadRepository(tenantId) // return a repository for a specefic teant id
      .findOneByIdOrThrow(query.id);
    return Result.ok(asset);
  }
}

Connection Provider

import { typeormConfig } from '@config/ormconfig';
import { ParameterServicePort } from '@infrastructure/parameters/parameter.service.port';
import { Inject, Injectable } from '@nestjs/common';
import { Connection, createConnection, getConnectionManager } from 'typeorm';
import { PostgresConnectionOptions } from 'typeorm/driver/postgres/PostgresConnectionOptions';

@Injectable()
export class TenancyService {
  protected tenantSchemas: Map<string, string> = new Map<string, string>();

  constructor(
    @Inject('ParameterServicePort')
    private parameterService: ParameterServicePort,
  ) {}

  async getTenantConnection(tenantId: string): Promise<Connection> {
    const connectionManager = getConnectionManager();
    const connectionName = tenantId;
    const schema = await this.getSchemaName(tenantId);

    if (connectionManager.has(connectionName)) {
      const connection = connectionManager.get(connectionName);
      return Promise.resolve(
        connection.isConnected ? connection : connection.connect(),
      );
    }

    return createConnection({
      ...(typeormConfig as PostgresConnectionOptions),
      name: connectionName,
      schema,
    });
  }

  private async getSchemaName(tenantId: string): Promise<string> {
    if (this.tenantSchemas.has(tenantId)) {
      return this.tenantSchemas.get(tenantId) as string;
    }
    const tenantSchema = await this.parameterService.getOperatorSchema(
      tenantId,
    );
    this.tenantSchemas.set(tenantId, tenantSchema);
    return tenantSchema;
  }
}

typeorm.repository.base.ts
declare an abstract get repository that accept tenantId as parameter to be implemented in each concrete repository

 import { NotFoundException } from '@exceptions';
import { AggregateRoot } from '@libs/ddd/domain/base-classes/aggregate-root.base';
import { DomainEvents } from '@libs/ddd/domain/domain-events';
import { LoggerPort } from '@libs/ddd/domain/ports/logger.port';
import {
  DataWithPaginationMeta,
  FindManyPaginatedParams,
  QueryParams,
  ReadRepositoryPort,
  WriteRepositoryPort,
} from '@libs/ddd/domain/ports/repository.ports';
import { ID } from '@libs/ddd/domain/value-objects/id.value-object';
import { OrmMapper } from '@libs/ddd/infrastructure/database/base-classes/orm-mapper.base';
import { FindConditions, ObjectLiteral, Repository } from 'typeorm';

export type WhereCondition<OrmEntity> =
  | FindConditions<OrmEntity>[]
  | FindConditions<OrmEntity>
  | ObjectLiteral
  | string;

export abstract class TypeormRepositoryBase<
  Entity extends AggregateRoot<unknown>,
  EntityProps,
  OrmEntity,
> implements
    WriteRepositoryPort<Entity>,
    ReadRepositoryPort<Entity, EntityProps>
{
  protected constructor(
    protected readonly tenantId: string,
    protected readonly mapper: OrmMapper<Entity, OrmEntity>,
    protected readonly logger: LoggerPort,
  ) {}

  // protected readonly repository: Repository<OrmEntity>;

  /**
   * Specify relations to other tables.
   * For example: `relations = ['user', ...]`
   */
  protected abstract relations: string[];

  abstract getRepository(tenantId: string): Promise<Repository<OrmEntity>>;

  protected abstract prepareQuery(
    params: QueryParams<EntityProps>,
  ): WhereCondition<OrmEntity>;

  async save(entity: Entity): Promise<Entity> {
    entity.validate(); // Protecting invariant before saving
    const ormEntity = this.mapper.toOrmEntity(entity);
    const result = await (
      await this.getRepository(this.tenantId)
    ).save(ormEntity);
    await DomainEvents.publishEvents(
      entity.id,
      this.logger,
      this.correlationId,
    );
    this.logger.debug(
      `[${entity.constructor.name}] persisted ${entity.id.value}`,
    );
    return this.mapper.toDomainEntity(result);
  }

  async saveMultiple(entities: Entity[]): Promise<Entity[]> {
    const ormEntities = entities.map((entity) => {
      entity.validate();
      return this.mapper.toOrmEntity(entity);
    });
    const result = await (
      await this.getRepository(this.tenantId)
    ).save(ormEntities);
    await Promise.all(
      entities.map((entity) =>
        DomainEvents.publishEvents(entity.id, this.logger, this.correlationId),
      ),
    );
    this.logger.debug(
      `[${entities}]: persisted ${entities.map((entity) => entity.id)}`,
    );
    return result.map((entity) => this.mapper.toDomainEntity(entity));
  }

  async findOne(
    params: QueryParams<EntityProps> = {},
  ): Promise<Entity | undefined> {
    const where = this.prepareQuery(params);
    const found = await (
      await this.getRepository(this.tenantId)
    ).findOne({
      where,
      relations: this.relations,
    });
    return found ? this.mapper.toDomainEntity(found) : undefined;
  }

  async findOneOrThrow(params: QueryParams<EntityProps> = {}): Promise<Entity> {
    const found = await this.findOne(params);
    if (!found) {
      throw new NotFoundException();
    }
    return found;
  }

  async findOneByIdOrThrow(id: ID | string): Promise<Entity> {
    const query = {
      where: { id: id instanceof ID ? id.value : id },
    };
    const found = await (
      await this.getRepository(this.tenantId)
    ).findOne(query);
    if (!found) {
      throw new NotFoundException();
    }
    return this.mapper.toDomainEntity(found);
  }

  async findMany(params: QueryParams<EntityProps> = {}): Promise<Entity[]> {
    const result = await (
      await this.getRepository(this.tenantId)
    ).find({
      where: this.prepareQuery(params),
      relations: this.relations,
    });

    return result.map((item) => this.mapper.toDomainEntity(item));
  }

  async findManyPaginated({
    params = {},
    pagination,
    orderBy,
  }: FindManyPaginatedParams<EntityProps>): Promise<
    DataWithPaginationMeta<Entity>
  > {
    const [data, count] = await (
      await this.getRepository(this.tenantId)
    ).findAndCount({
      skip: pagination?.skip,
      take: pagination?.limit,
      where: this.prepareQuery(params),
      order: orderBy,
      relations: this.relations,
    });

    const result: DataWithPaginationMeta<Entity> = {
      results: data.map((item) => this.mapper.toDomainEntity(item)),
      nbResultsPerPage: count,
      limit: pagination?.limit,
      page: pagination?.page,
      startCursor: '',
      endCursor: '',
      hasNextPage: false,
      hasPreviousPage: false,
    };

    return result;
  }

  async delete(entity: Entity): Promise<Entity> {
    entity.validate();
    await (
      await this.getRepository(this.tenantId)
    ).remove(this.mapper.toOrmEntity(entity));
    await DomainEvents.publishEvents(
      entity.id,
      this.logger,
      this.correlationId,
    );
    this.logger.debug(
      `[${entity.constructor.name}] deleted ${entity.id.value}`,
    );
    return entity;
  }

  protected correlationId?: string;

  setCorrelationId(correlationId: string): this {
    this.correlationId = correlationId;
    this.setContext();
    return this;
  }

  private setContext() {
    if (this.correlationId) {
      this.logger.setContext(`${this.constructor.name}:${this.correlationId}`);
    } else {
      this.logger.setContext(this.constructor.name);
    }
  }
}

And finally the impl of the repository
src/modules/asset/database/asset.orm-repository.ts

import { TenancyService } from '@infrastructure/tenancy/tenancy.service';
import { LoggerPort } from '@libs/ddd/domain/ports/logger.port';
import { QueryParams } from '@libs/ddd/domain/ports/repository.ports';
import {
  TypeormRepositoryBase,
  WhereCondition,
} from '@libs/ddd/infrastructure/database/base-classes/typeorm.repository.base';
import { final } from '@libs/decorators/final.decorator';
import { prepareQueryProps } from '@libs/utils/prepare-query-props.util';
import { AssetOrmEntity } from '@modules/asset/database/asset.orm-entity';
import { AssetOrmMapper } from '@modules/asset/database/asset.orm-mapper';
import {
  AssetEntity,
  AssetProps,
} from '@modules/asset/domain/entities/asset.entity';
import { AssetReadRepositoryPort } from '@modules/asset/ports/asset.repository.port';
import { FindAssetsQuery } from '@modules/asset/queries/find-assets/find-assets.query';
import { Repository } from 'typeorm';

@final
export class AssetOrmRepository
  extends TypeormRepositoryBase<AssetEntity, AssetProps, AssetOrmEntity>
  implements AssetReadRepositoryPort
{
  protected relations: string[] = [];

  constructor(
    tenantId: string,
    protected logger: LoggerPort,
    protected tenancyService: TenancyService, // inject tenancy service
  ) {
    super(tenantId, new AssetOrmMapper(AssetEntity, AssetOrmEntity), logger);
  }

  async getRepository(tenantId: string): Promise<Repository<AssetOrmEntity>> {
    return (
      await this.tenancyService.getTenantConnection(tenantId) // create a connection using tenantId
    ).getRepository(AssetOrmEntity);
  }

  async findAssets(query: FindAssetsQuery): Promise<AssetEntity[]> {
    const where: QueryParams<AssetOrmEntity> = prepareQueryProps(query);
    const assets = await (
      await this.getRepository(this.tenantId)
    ).find({ where });
    return assets.map((asset) => this.mapper.toDomainEntity(asset));
  }

  // Used to construct a query
  protected prepareQuery(
    params: QueryParams<AssetProps>,
  ): WhereCondition<AssetOrmEntity> {
    const where: QueryParams<AssetOrmEntity> = {};
    if (params.name) {
      where.name = params.name;
    }
    if (params.ownerId) {
      where.ownerId = params.ownerId;
    }
    if (params.assetTypeId) {
      where.assetTypeId = params.assetTypeId;
    }
    if (params.categoryId) {
      where.categoryId = params.categoryId;
    }
    if (params.active !== undefined) {
      where.active = params.active;
    }
    if (params.validated !== undefined) {
      where.validated = params.validated;
    }

    return where;
  }
}