@mbc-cqrs-serverless/import

Flexible data import module for the MBC CQRS Serverless framework. Import data from REST APIs, CSV files, and ZIP archives with validation, transformation, and async processing.

Features

• Multi-Source Import: Support for REST API, CSV files, and ZIP archives
• Strategy Pattern: Customizable validation and transformation per entity type
• Two-Phase Processing: Separate ingestion and business logic phases
• Dual Processing Modes: DIRECT mode for small files, STEP_FUNCTION for large-scale imports
• Progress Tracking: Real-time status updates via SNS notifications
• Error Handling: Built-in alarm notifications and row-level error tracking

Installation

npm install @mbc-cqrs-serverless/import

Quick Start

1. Register the Module

import { Module } from '@nestjs/common';
import { ImportModule } from '@mbc-cqrs-serverless/import';
import { ProductImportStrategy } from './strategies/product-import.strategy';
import { ProductProcessStrategy } from './strategies/product-process.strategy';
import { ProductModule } from './product/product.module';

@Module({
  imports: [
    ImportModule.register({
      profiles: [
        {
          tableName: 'product',
          importStrategy: ProductImportStrategy,
          processStrategy: ProductProcessStrategy,
        },
      ],
      imports: [ProductModule], // Optional: modules that provide strategy dependencies
      enableController: true, // Optional: enable REST endpoints
    }),
  ],
})
export class AppModule {}

2. Implement Import Strategy

import { Injectable } from '@nestjs/common';
import { IImportStrategy } from '@mbc-cqrs-serverless/import';

@Injectable()
export class ProductImportStrategy implements IImportStrategy<RawProductInput, ProductDto> {
  async transform(input: RawProductInput): Promise<ProductDto> {
    return {
      code: input.product_code?.trim(),
      name: input.product_name?.trim(),
      price: parseFloat(input.price),
      category: input.category?.toUpperCase(),
    };
  }

  async validate(dto: ProductDto): Promise<void> {
    if (!dto.code) throw new Error('Product code is required');
    if (!dto.name) throw new Error('Product name is required');
    if (isNaN(dto.price) || dto.price < 0) throw new Error('Invalid price');
  }
}

3. Implement Process Strategy

import { Injectable } from '@nestjs/common';
import {
  IProcessStrategy,
  ComparisonStatus,
  ComparisonResult,
} from '@mbc-cqrs-serverless/import';
import { CommandService, DataModel, CommandInputModel } from '@mbc-cqrs-serverless/core';

// Define your entity model
interface ProductModel extends DataModel {
  code: string;
  name: string;
  price: number;
}

@Injectable()
export class ProductProcessStrategy implements IProcessStrategy<ProductModel, ProductDto> {
  constructor(
    private readonly productService: ProductService,
    private readonly commandService: CommandService,
  ) {}

  async compare(dto: ProductDto, tenantCode: string): Promise<ComparisonResult<ProductModel>> {
    const existing = await this.productService.findByCode(dto.code, tenantCode);
    if (!existing) {
      return { status: ComparisonStatus.NOT_EXIST };
    }
    if (this.hasChanges(existing, dto)) {
      return { status: ComparisonStatus.CHANGED, existingData: existing };
    }
    return { status: ComparisonStatus.EQUAL };
  }

  async map(
    status: ComparisonStatus.NOT_EXIST | ComparisonStatus.CHANGED,
    dto: ProductDto,
    tenantCode: string,
    existingData?: ProductModel,
  ): Promise<CommandInputModel> {
    return {
      pk: `PRODUCT#${tenantCode}`,
      sk: `PRODUCT#${dto.code}`,
      code: dto.code,
      name: dto.name,
      attributes: { price: dto.price },
    };
  }

  getCommandService(): CommandService {
    return this.commandService;
  }
}

Architecture

The import module uses a two-phase architecture:

┌─────────────────────────────────────────────────────────────────┐
│                    Import Architecture                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  Phase 1: Ingestion                                              │
│  ┌──────────────┐     ┌──────────────┐     ┌──────────────┐     │
│  │ REST API /   │────▶│ ImportStrategy│────▶│  Temp Table  │     │
│  │ CSV / ZIP    │     │ transform()   │     │  (CREATED)   │     │
│  └──────────────┘     │ validate()    │     └──────────────┘     │
│                       └──────────────┘            │              │
│                                                   ▼ DynamoDB     │
│  Phase 2: Processing                              │ Streams      │
│  ┌──────────────┐     ┌──────────────┐     ┌─────┴────────┐     │
│  │ ProcessStrategy│◀───│  SNS/SQS     │◀────│   Lambda     │     │
│  │ compare()    │     │  Event       │     │   Trigger    │     │
│  │ map()        │     └──────────────┘     └──────────────┘     │
│  └──────────────┘                                                │
│         │                                                        │
│         ▼                                                        │
│  ┌──────────────┐     ┌──────────────┐                          │
│  │ CommandService│────▶│ Final Table  │                          │
│  │ publish()    │     │ (Data Store) │                          │
│  └──────────────┘     └──────────────┘                          │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

API Reference

ImportService

| Method | Description | |--------|-------------| | createWithApi(dto, options) | Import single record from REST API | | handleCsvImport(dto, options) | Import from CSV file (DIRECT or STEP_FUNCTION) | | createCsvJob(dto, options) | Create CSV import job for Step Functions | | createZipJob(dto, options) | Create ZIP import job for Step Functions | | createImport(dto, options) | Create import record in temp table | | updateStatus(key, status, payload?, attributes?, notifyId?) | Update import status | | getImportByKey(key) | Get import record by key | | incrementParentJobCounters(parentKey, childSucceeded) | Update parent job progress |

IImportStrategy Interface

interface IImportStrategy<TInput extends object, TAttributesDto extends object> {
  transform(input: TInput): Promise<TAttributesDto>;
  validate(data: TAttributesDto): Promise<void>;
}

IProcessStrategy Interface

interface IProcessStrategy<TEntity extends DataModel, TAttributesDto extends object> {
  compare(importAttributes: TAttributesDto, tenantCode: string): Promise<ComparisonResult<TEntity>>;
  map(status: ComparisonStatus, importAttributes: TAttributesDto, tenantCode: string, existingData?: TEntity): Promise<CommandInputModel | CommandPartialInputModel>;
  getCommandService(): CommandService;
}

interface ComparisonResult<TEntity> {
  status: ComparisonStatus;
  existingData?: TEntity; // Provided when status is CHANGED
}

enum ComparisonStatus {
  NOT_EXIST = 'NOT_EXIST',
  CHANGED = 'CHANGED',
  EQUAL = 'EQUAL',
}

ImportStatusEnum

| Status | Description | |--------|-------------| | CREATED | Import record created | | PROCESSING | Being processed | | COMPLETED | Successfully completed | | FAILED | Processing failed |

ImportStatusHandler

The ImportStatusHandler is an internal event handler that manages Step Functions callbacks for import jobs. When using Step Functions orchestration (ZIP imports or STEP_FUNCTION mode CSV imports), this handler ensures proper communication with the state machine.

Behavior

| Import Status | Action | Step Functions Command | |---------------|--------|----------------------| | COMPLETED | Send success callback | SendTaskSuccessCommand | | FAILED | Send failure callback | SendTaskFailureCommand | | Other statuses | Ignored | None |

Methods

| Method | Description | |--------|-------------| | sendTaskSuccess(taskToken, output) | Sends success signal to Step Functions with the import result | | sendTaskFailure(taskToken, error, cause) | Sends failure signal to Step Functions with error details |

Step Functions Integration

When an import job is created as part of a Step Functions workflow (e.g., ZIP import), a taskToken is stored in the job's attributes. The ImportStatusHandler listens for status change notifications and:

1. Retrieves the import job from DynamoDB
2. Checks if a taskToken exists in the job's attributes
3. Sends the appropriate callback based on the final status:
   • COMPLETED → SendTaskSuccessCommand with result data
   • FAILED → SendTaskFailureCommand with error details

This ensures Step Functions workflows properly handle both success and failure cases without hanging indefinitely.

Usage Examples

REST API Import

@Controller('import')
export class ImportController {
  constructor(private readonly importService: ImportService) {}

  @Post('product')
  async importProduct(
    @Body() dto: ImportProductDto,
    @InvokeContext() ctx: IInvoke,
  ) {
    return this.importService.createWithApi(
      {
        tableName: 'product',
        tenantCode: dto.tenantCode,
        attributes: dto,
      },
      { invokeContext: ctx },
    );
  }
}

CSV Import (Direct Mode)

Process small CSV files immediately:

async importSmallCsv(bucket: string, key: string, opts: { invokeContext: IInvoke }) {
  return this.importService.handleCsvImport(
    {
      tenantCode: 'MBC',
      tableName: 'product',
      bucket,
      key,
      processingMode: 'DIRECT',
    },
    opts,
  );
}

CSV Import (Step Functions Mode)

Process large CSV files with Step Functions orchestration:

async importLargeCsv(bucket: string, key: string, opts: { invokeContext: IInvoke }) {
  return this.importService.handleCsvImport(
    {
      tenantCode: 'MBC',
      tableName: 'product',
      bucket,
      key,
      processingMode: 'STEP_FUNCTION',
    },
    opts,
  );
}

ZIP Import

Import multiple CSV files from a ZIP archive:

async importZip(bucket: string, key: string, opts: { invokeContext: IInvoke }) {
  return this.importService.createZipJob(
    {
      tenantCode: 'MBC',
      bucket,
      key,
    },
    opts,
  );
}

Check Import Status

async checkStatus(pk: string, sk: string) {
  const importJob = await this.importService.getImportByKey({ pk, sk });
  return {
    status: importJob.status,
    totalRows: importJob.totalRows,
    processedRows: importJob.processedRows,
    succeededRows: importJob.succeededRows,
    failedRows: importJob.failedRows,
  };
}

Processing Modes

| Mode | Use Case | Processing | |------|----------|------------| | DIRECT | Small files (< 1000 rows) | Immediate, synchronous | | STEP_FUNCTION | Large files, mission-critical | Async, resilient, tracked |

Related Packages

| Package | Description | |---------|-------------| | @mbc-cqrs-serverless/core | Core CQRS framework | | @mbc-cqrs-serverless/task | Task processing for async jobs |

Documentation

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

• Import Service Guide

License

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

This project is under the MIT License.