@mbc-cqrs-serverless/core

@mbc-cqrs-serverless/core

The core package of the MBC CQRS Serverless framework, providing a complete implementation of CQRS (Command Query Responsibility Segregation) and Event Sourcing patterns for AWS serverless architectures.

Features

• CQRS Pattern: Separate command (write) and query (read) operations for better scalability
• Event Sourcing: Full audit trail with versioned commands and optimistic locking
• AWS Integration: Built-in support for DynamoDB, Step Functions, SNS, SQS, S3, and Cognito
• Multi-tenancy: Tenant isolation with automatic context management
• NestJS Framework: Leverage dependency injection, decorators, and modular architecture
• TypeScript First: Full type safety and excellent IDE support

Installation

npm install @mbc-cqrs-serverless/core

Quick Start

1. Configure the Module

import { Module } from '@nestjs/common';
import { CommandModule, DataService, CommandService } from '@mbc-cqrs-serverless/core';

@Module({
  imports: [
    CommandModule.register({
      tableName: 'todo',
    }),
  ],
})
export class TodoModule {}

2. Inject Services

import { Injectable } from '@nestjs/common';
import {
  CommandService,
  DataService,
  generateId,
  getUserContext,
  VERSION_FIRST,
  IInvoke,
} from '@mbc-cqrs-serverless/core';

@Injectable()
export class TodoService {
  constructor(
    private readonly commandService: CommandService,
    private readonly dataService: DataService,
  ) {}

  async create(dto: CreateTodoDto, opts: { invokeContext: IInvoke }) {
    const { tenantCode } = getUserContext(opts.invokeContext);
    const pk = `TODO#${tenantCode}`;
    const sk = `TODO#${Date.now()}`;

    const command = {
      pk,
      sk,
      id: generateId(pk, sk),
      tenantCode,
      code: sk,
      type: 'TODO',
      version: VERSION_FIRST,
      name: dto.name,
      attributes: dto.attributes,
    };

    return await this.commandService.publishAsync(command, opts);
  }

  async findOne(pk: string, sk: string) {
    return await this.dataService.getItem({ pk, sk });
  }
}

Key Concepts

CQRS Architecture

┌─────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   Client    │────▶│  CommandService │────▶│ DynamoDB        │
│   (Write)   │     │  (publishAsync) │     │ (Command Table) │
└─────────────┘     └─────────────────┘     └────────┬────────┘
                                                     │
                                                     ▼ DynamoDB Streams
                                            ┌─────────────────┐
                                            │ Step Functions  │
                                            │ (Data Sync)     │
                                            └────────┬────────┘
                                                     │
┌─────────────┐     ┌─────────────────┐     ┌────────▼────────┐
│   Client    │────▶│   DataService   │────▶│ DynamoDB        │
│   (Read)    │     │   (getItem)     │     │ (Data Table)    │
└─────────────┘     └─────────────────┘     └─────────────────┘

Command vs Data Tables

Aspect	Command Table	Data Table
Purpose	Event log / Audit trail	Current state
Versioning	All versions stored	Latest version only
Sort Key	Includes version (sk#v001)	No version suffix
Use Case	Write operations	Read operations

Version Control

Every command includes a version number for optimistic locking:

// VERSION_FIRST = 0 for new items
const command = { pk, sk, version: VERSION_FIRST, ... };
await commandService.publishAsync(command, opts);

// Updates require the current version
const updateCommand = { pk, sk, version: currentVersion, ... };
await commandService.publishPartialUpdateAsync(updateCommand, opts);

API Reference

CommandService

The primary service for write operations.

Method	Description
publishAsync(input, options)	Create or update an item asynchronously via Step Functions
publishSync(input, options)	Create or update an item synchronously (bypasses Step Functions)
publishPartialUpdateAsync(input, options)	Partial update with field merging (async)
publishPartialUpdateSync(input, options)	Partial update with field merging (sync)
getItem(key)	Get a specific command version
getLatestItem(key)	Get the latest command version
duplicate(key, options)	Duplicate an existing command
reSyncData()	Re-synchronize all data to sync handlers

DataService

The primary service for read operations.

Method	Description
getItem(key)	Retrieve a single item by pk and sk
listItemsByPk(pk, options)	List items by partition key with filtering and pagination

Helper Functions

import {
  generateId,        // Generate unique ID from pk and sk
  getUserContext,    // Extract user info from Lambda context
  addSortKeyVersion, // Add version suffix to sort key
  removeSortKeyVersion, // Remove version suffix from sort key
  getTenantCode,     // Extract tenant code from partition key
} from '@mbc-cqrs-serverless/core';

Constants

import {
  VERSION_FIRST,   // Initial version (0)
  VERSION_LATEST,  // Marker for latest version (-1)
  VER_SEPARATOR,   // Version separator in sort key ('#')
} from '@mbc-cqrs-serverless/core';

Data Sync Handlers

Extend data synchronization to custom destinations (e.g., RDS, Elasticsearch):

import { Injectable } from '@nestjs/common';
import {
  IDataSyncHandler,
  DataSyncHandler,
  CommandModel,
} from '@mbc-cqrs-serverless/core';
import { PrismaService } from '../prisma/prisma.service';

@DataSyncHandler({ tableName: 'todo' })
@Injectable()
export class TodoDataSyncHandler implements IDataSyncHandler {
  type = 'rds';

  constructor(private readonly prisma: PrismaService) {}

  async up(cmd: CommandModel): Promise<void> {
    await this.prisma.todo.upsert({
      where: { pk_sk: { pk: cmd.pk, sk: cmd.sk } },
      create: { /* ... */ },
      update: { /* ... */ },
    });
  }

  async down(cmd: CommandModel): Promise<void> {
    // Optional: handle rollback
  }
}

Group-based roles

RolesGuard checks direct roles from JWT custom:roles first, then roles from groups in custom:groups (tenant-scoped). Group → role mappings are not stored in the JWT; implement a resolver in your app.

JWT example:

{
  "custom:roles": "[{\"tenant\":\"tenant-a\",\"role\":\"admin\"}]",
  "custom:groups": "[{\"tenant\":\"tenant-a\",\"groups\":[\"sales-team\"]}]"
}

Resolver (one per app):

import {
  GroupRoleResolver,
  IGroupRoleResolver,
} from '@mbc-cqrs-serverless/core';

// Do not add @Injectable() — @GroupRoleResolver() already registers the provider.
@GroupRoleResolver()
export class AppGroupRoleResolver implements IGroupRoleResolver {
  async resolveRoles({ tenantCode, groupIds, claims }) {
    // Load roles from DynamoDB, RDS, config, etc.
    return ['viewer', 'reporter'];
  }
}

Register the class in your NestJS module providers. AuthModule is imported automatically via core AppModule.forRoot() and exports GroupRoleResolverRegistry globally for @Auth() / RolesGuard.

Extending authorization

RolesGuard evaluates tenantRoles from the JWT (not getUserRole()). The protected getUserRole() method is deprecated and not called by the default guard. Subclasses should override verifyRole or resolveGroupRoles for custom logic.

Environment Variables

Variable	Description	Default
DYNAMODB_ENDPOINT	DynamoDB endpoint URL	AWS default
DYNAMODB_REGION	DynamoDB region	AWS default
SFN_ENDPOINT	Step Functions endpoint	AWS default
SFN_REGION	Step Functions region	AWS default
SNS_ENDPOINT	SNS endpoint	AWS default
SQS_ENDPOINT	SQS endpoint	AWS default
COGNITO_ENDPOINT	Cognito endpoint	AWS default
COGNITO_REGION	Cognito region	AWS default

Related Packages

Package	Description
@mbc-cqrs-serverless/cli	CLI for project scaffolding
@mbc-cqrs-serverless/sequence	Sequence number generation
@mbc-cqrs-serverless/task	Async task processing
@mbc-cqrs-serverless/master	Master data management
@mbc-cqrs-serverless/tenant	Multi-tenancy support
@mbc-cqrs-serverless/import	Data import utilities
@mbc-cqrs-serverless/ui-setting	UI configuration

Documentation

Full documentation is available at https://mbc-cqrs-serverless.mbc-net.com/

• Getting Started
• Build a Todo App Tutorial
• Architecture Guide
• API Reference

License

Copyright 2024-2025, Murakami Business Consulting, Inc. https://www.mbc-net.com/

This project is under the MIT License.