@mbc-cqrs-serverless/task

@mbc-cqrs-serverless/task

Asynchronous task processing for the MBC CQRS Serverless framework. Execute long-running operations, batch processes, and background jobs with status tracking and SNS notifications.

Features

• Async Task Execution: Process long-running operations in background
• Step Functions Integration: Orchestrate complex workflows with AWS Step Functions
• Sub-task Support: Break large tasks into smaller parallel sub-tasks
• Status Tracking: Real-time task status updates via SNS notifications
• Multi-tenant: Isolated task queues per tenant
• Error Handling: Built-in alarm notifications for failed tasks

Installation

npm install @mbc-cqrs-serverless/task

Quick Start

1. Register the Module

import { Module } from '@nestjs/common';
import { TaskModule, ITaskQueueEventFactory, TaskQueueEvent, StepFunctionTaskEvent } from '@mbc-cqrs-serverless/task';
import { IEvent } from '@mbc-cqrs-serverless/core';

// Implement factory to transform task events into your custom events
class MyTaskQueueEventFactory implements ITaskQueueEventFactory {
  async transformTask(event: TaskQueueEvent): Promise<IEvent[]> {
    // Transform task queue event into your custom events
    return [event];
  }

  async transformStepFunctionTask(event: StepFunctionTaskEvent): Promise<IEvent[]> {
    // Transform step function task event into your custom events
    return [event];
  }
}

@Module({
  imports: [
    TaskModule.register({
      taskQueueEventFactory: MyTaskQueueEventFactory,
      enableController: true, // Optional: enable REST endpoints
    }),
  ],
})
export class AppModule {}

Global module and single registration

TaskModule.register() returns a global dynamic module (global: true). That means:

• TaskService and the module’s exports are available in every Nest module without listing TaskModule in each feature module’s imports.
• You must call TaskModule.register() exactly once per application so the TASK_QUEUE_EVENT_FACTORY token and task providers are registered a single time. Multiple TaskModule.register() calls can cause conflicting bindings and runtime errors (for example transformTask is not a function on the task queue handler).

Implement ITaskQueueEventFactory with:

• transformTask — maps TaskQueueEvent (standard async tasks from SQS) to your domain events.
• transformStepFunctionTask — maps StepFunctionTaskEvent (Step Functions–backed tasks) to your domain events.

Omit either method only if you do not use that execution path; the handlers still expect the factory object to expose the methods your deployment uses.

Using @mbc-cqrs-serverless/master

If you use MasterModule.register({ enableController: true }) from @mbc-cqrs-serverless/master, that package does not call TaskModule.register() internally. You must register TaskModule in the host application as above so MasterSettingService, MyTaskService, and other providers can inject TaskService. See the master package README.

2. Create and Monitor Tasks

import { Injectable } from '@nestjs/common';
import { TaskService, TaskStatusEnum } from '@mbc-cqrs-serverless/task';
import { getUserContext, IInvoke } from '@mbc-cqrs-serverless/core';

@Injectable()
export class BatchService {
  constructor(private readonly taskService: TaskService) {}

  async startBatchProcess(data: any[], opts: { invokeContext: IInvoke }) {
    const { tenantCode } = getUserContext(opts.invokeContext);

    // Create a new task
    const task = await this.taskService.createTask(
      {
        tenantCode,
        taskType: 'BATCH_IMPORT',
        name: 'Import Customer Data',
        input: data,
      },
      opts,
    );

    console.log(task.id);     // Task identifier
    console.log(task.status); // "CREATED"
    console.log(task.code);   // ULID code

    return task;
  }

  async checkStatus(pk: string, sk: string) {
    const task = await this.taskService.getTask({ pk, sk });
    return task.status;
  }
}

API Reference

TaskService

Method	Description
createTask(dto, options)	Create a new async task
createStepFunctionTask(dto, options)	Create a task for Step Functions workflow
createSubTask(event)	Create sub-tasks from parent task input
getTask(key)	Get task by pk/sk
updateStatus(key, status, attributes?, notifyId?)	Update task status with SNS notification
updateSubTaskStatus(key, status, attributes?, notifyId?)	Update sub-task status
listItemsByPk(tenantCode, type?, options?)	List tasks by tenant
getAllSubTask(subTaskKey)	Get all sub-tasks of a parent task
publishAlarm(event, errorDetails)	Send alarm notification for failed tasks

CreateTaskDto

Property	Type	Required	Description
tenantCode	string	Yes	Tenant identifier
taskType	string	Yes	Task type identifier
name	string	No	Human-readable task name
input	object	Yes	Task input data

TaskStatusEnum

Status	Description
CREATED	Task created, not yet queued
QUEUED	Task queued for processing
STARTED	Task execution started
PROCESSING	Task is being processed
FINISHED	Task execution finished
COMPLETED	Task completed successfully
ERRORED	Task encountered an error
FAILED	Task failed permanently

TaskEntity

{
  id: string;          // Full identifier (pk#sk)
  pk: string;          // Partition key (TASK#tenantCode)
  sk: string;          // Sort key (taskType#taskCode)
  code: string;        // ULID task code
  type: string;        // Task type
  name: string;        // Task name
  tenantCode: string;  // Tenant identifier
  status: string;      // Current status
  input: any;          // Task input data
  attributes?: any;    // Result/error data
  createdAt: Date;
  updatedAt: Date;
  createdBy: string;
  updatedBy: string;
}

Usage Examples

Standard Task Processing

Create and process tasks with SQS queue:

@Injectable()
export class ReportService {
  constructor(private readonly taskService: TaskService) {}

  async generateReport(reportType: string, opts: { invokeContext: IInvoke }) {
    const { tenantCode } = getUserContext(opts.invokeContext);

    // Create task - triggers SNS/SQS workflow
    const task = await this.taskService.createTask(
      {
        tenantCode,
        taskType: 'REPORT_GENERATION',
        name: `Generate ${reportType} Report`,
        input: { reportType, filters: {} },
      },
      opts,
    );

    return { taskId: task.id, status: task.status };
  }
}

Step Functions Task

Create tasks that integrate with AWS Step Functions:

@Injectable()
export class WorkflowService {
  constructor(private readonly taskService: TaskService) {}

  async startWorkflow(data: any, opts: { invokeContext: IInvoke }) {
    const { tenantCode } = getUserContext(opts.invokeContext);

    // Create Step Functions task
    const task = await this.taskService.createStepFunctionTask(
      {
        tenantCode,
        taskType: 'ORDER_PROCESSING',
        name: 'Process Order Workflow',
        input: data,
      },
      opts,
    );

    // Task key for Step Functions: SFN_TASK#tenantCode
    return task;
  }
}

Sub-task Processing

Split large tasks into parallel sub-tasks:

@Injectable()
export class BulkImportHandler implements IEventHandler<TaskQueueEvent> {
  constructor(private readonly taskService: TaskService) {}

  async handle(event: TaskQueueEvent) {
    // Create sub-tasks from parent task input array
    const subTasks = await this.taskService.createSubTask(event);

    // Each sub-task processes one item from the input array
    console.log(`Created ${subTasks.length} sub-tasks`);

    // Update parent task status
    await this.taskService.updateStatus(
      event.taskEvent.taskKey,
      TaskStatusEnum.PROCESSING,
    );
  }
}

Task Status Updates

Update task status with SNS notifications:

@Injectable()
export class TaskProcessor {
  constructor(private readonly taskService: TaskService) {}

  async processTask(pk: string, sk: string) {
    try {
      // Mark as processing
      await this.taskService.updateStatus(
        { pk, sk },
        TaskStatusEnum.PROCESSING,
      );

      // ... do work ...

      // Mark as completed with result
      await this.taskService.updateStatus(
        { pk, sk },
        TaskStatusEnum.COMPLETED,
        { result: { processedCount: 100 } },
      );
    } catch (error) {
      // Mark as failed with error
      await this.taskService.updateStatus(
        { pk, sk },
        TaskStatusEnum.FAILED,
        { error: error.message },
      );
    }
  }
}

List Tasks

Query tasks by tenant and type:

@Injectable()
export class TaskDashboard {
  constructor(private readonly taskService: TaskService) {}

  async getTaskList(tenantCode: string) {
    // List standard tasks
    const tasks = await this.taskService.listItemsByPk(
      tenantCode,
      'TASK',
      { limit: 20, order: 'desc' },
    );

    // List Step Function tasks
    const sfnTasks = await this.taskService.listItemsByPk(
      tenantCode,
      'SFN_TASK',
      { limit: 20, order: 'desc' },
    );

    return { tasks: tasks.items, sfnTasks: sfnTasks.items };
  }
}

Event Handlers

The package provides built-in event handlers for task processing:

TaskQueueEventHandler

Handles task queue events from SQS:

import { EventsHandler, IEventHandler } from '@mbc-cqrs-serverless/core';
import { TaskQueueEvent } from '@mbc-cqrs-serverless/task';

@EventsHandler(TaskQueueEvent)
export class MyTaskHandler implements IEventHandler<TaskQueueEvent> {
  async handle(event: TaskQueueEvent) {
    const { taskKey, taskEntity } = event.taskEvent;
    // Process the task...
  }
}

StepFunctionTaskEventHandler

Handles Step Functions task events:

import { EventsHandler, IEventHandler } from '@mbc-cqrs-serverless/core';
import { StepFunctionTaskEvent } from '@mbc-cqrs-serverless/task';

@EventsHandler(StepFunctionTaskEvent)
export class MySfnHandler implements IEventHandler<StepFunctionTaskEvent> {
  async handle(event: StepFunctionTaskEvent) {
    const { taskKey } = event;
    // Process the Step Function task...
  }
}

Related Packages

Package	Description
@mbc-cqrs-serverless/core	Core CQRS framework
@mbc-cqrs-serverless/import	Data import with task processing

Documentation

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

• Task Service Guide
• Step Functions Integration

License

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

This project is under the MIT License.