JSPM

  • Created
  • Published
  • Downloads 950
  • Score
    100M100P100Q113727F
  • License MIT

CQRS and event base core

Package Exports

  • @mbc-cqrs-serverless/core
  • @mbc-cqrs-serverless/core/dist/index.js

This package does not declare an exports field, so the exports above have been automatically detected and optimized by JSPM instead. If any package subpath is missing, it is recommended to post an issue to the original package (@mbc-cqrs-serverless/core) to support the "exports" field. If that is not possible, create a JSPM override to customize the exports field for this package.

Readme

MBC CQRS serverless framework

@mbc-cqrs-serverless/core

npm version License: MIT

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.

Quick Start Demo

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
  }
}

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
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/

License

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

This project is under the MIT License.