Cell - CQRS Component

@celljs/cqrs separates the code that changes state (commands) from the code that reads it (queries), and lets the rest of the application react to what happened (events), in the way of @nestjs/cqrs. Handlers are Cell components: they are found by their decorators, get injections like any other component, and are created when the first message of their kind is dispatched. No dependencies besides @celljs/core; backend only.

npm install @celljs/cqrs

Commands and queries

A command or query is an instance of a class; its class selects its one handler. Extend CommandMessage<R> / QueryMessage<R> to give execute the result type (optional: without it, execute<R>() takes the type).

import { Autowired } from '@celljs/core';
import { Body, Controller, Get, Param, Post } from '@celljs/mvc/lib/node';
import { CommandBus, CommandHandler, CommandMessage, QueryBus, QueryHandler, QueryMessage } from '@celljs/cqrs';

export class CreateUser extends CommandMessage<string> {
    constructor(readonly name: string) {
        super();
    }
}

export class GetUser extends QueryMessage<User | undefined> {
    constructor(readonly id: string) {
        super();
    }
}

@CommandHandler(CreateUser)
export class CreateUserHandler implements CommandHandler<CreateUser> {
    @Autowired(UserRepository)
    protected readonly users: UserRepository;

    execute(command: CreateUser): Promise<string> {
        return this.users.insert(command.name);
    }
}

@QueryHandler(GetUser)
export class GetUserHandler implements QueryHandler<GetUser> {
    @Autowired(UserRepository)
    protected readonly users: UserRepository;

    execute(query: GetUser): Promise<User | undefined> {
        return this.users.findById(query.id);
    }
}

@Controller('users')
export class UserController {
    @Autowired(CommandBus)
    protected readonly commandBus: CommandBus;

    @Autowired(QueryBus)
    protected readonly queryBus: QueryBus;

    @Post()
    create(@Body() body: { name: string }): Promise<string> {
        return this.commandBus.execute(new CreateUser(body.name));
    }

    @Get(':id')
    get(@Param('id') id: string): Promise<User | undefined> {
        return this.queryBus.execute(new GetUser(id));
    }
}

execute resolves to what the handler returns and rejects with what it throws. The decorators take the options of @Component as second argument (scope, proxy for AOP...).

Events and sagas

An event class can have any number of @EventHandlers and @Sagas. A saga turns an event into the commands that follow from it: the event bus runs what it returns (a command, an array of commands or nothing) with the command bus, in order.

import { EventBus, EventHandler, Saga } from '@celljs/cqrs';

export class UserCreated {
    constructor(readonly id: string, readonly name: string) { }
}

@EventHandler(UserCreated)                    // or @EventHandler([UserCreated, UserRenamed])
export class UserProjection implements EventHandler<UserCreated> {
    async handle(event: UserCreated): Promise<void> { ... }
}

@Saga(UserCreated)
export class WelcomeSaga implements Saga<UserCreated> {
    handle(event: UserCreated): SendWelcomeMail {
        return new SendWelcomeMail(event.id);
    }
}

await eventBus.publish(new UserCreated(id, name));

publish(event) runs the handlers and sagas of the event concurrently and resolves when they are all done, so a serverless function does not return before them; publishAll(events) publishes one event after the other. What a handler or saga throws does not reject publish nor stop the other handlers: it is logged and handed to the UnhandledExceptionBus:

@Component(ApplicationLifecycle)
export class CqrsAlerts implements ApplicationLifecycle<Application> {
    @Autowired(UnhandledExceptionBus)
    protected readonly exceptions: UnhandledExceptionBus;

    initialize(): void {
        this.exceptions.onException(({ error, event, handler }) => { ... });
    }
}

Aggregates

An AggregateRoot records its changes as events: apply(event) records the event and calls the method on<EventClassName> of the aggregate, if there is one, to change the state; commit() publishes the recorded events with the event bus. Connect the aggregate to the event bus with EventPublisher first.

import { AggregateRoot, EventPublisher } from '@celljs/cqrs';

export class User extends AggregateRoot {
    name: string;

    constructor(readonly id: string) {
        super();
    }

    rename(name: string): void {
        this.apply(new UserRenamed(this.id, name));
    }

    onUserRenamed(event: UserRenamed): void {
        this.name = event.name;
    }
}

@CommandHandler(RenameUser)
export class RenameUserHandler implements CommandHandler<RenameUser> {
    @Autowired(EventPublisher)
    protected readonly publisher: EventPublisher;

    @Autowired(UserRepository)
    protected readonly users: UserRepository;

    async execute(command: RenameUser): Promise<void> {
        const user = this.publisher.mergeObjectContext(await this.users.load(command.id));
        user.rename(command.name);
        await this.users.save(user);
        await user.commit();
    }
}

• loadFromHistory(events) rebuilds the state from stored events without recording them again.
• getUncommittedEvents() returns the recorded events; uncommit() forgets them.
• eventPublisher.mergeClassContext(User) returns a subclass whose instances are connected when created.
• The recorded events are not properties of the aggregate: JSON.stringify(user) gives its state only.

Errors

The buses and commit() reject with a CqrsError whose code is stable and whose hint says how to fix it:

Code	When
CELL_CQRS_NO_HANDLER	No @CommandHandler / @QueryHandler for the class of the command or query
CELL_CQRS_DUPLICATE_HANDLER	Two handlers for one command or query class
CELL_CQRS_NOT_AN_INSTANCE	The message is a plain object, null or a primitive instead of an instance of a class
CELL_CQRS_AGGREGATE_NOT_MERGED	commit() on an aggregate that EventPublisher did not connect; the events stay recorded

Customizing

CommandBus, QueryBus, EventBus, EventPublisher and UnhandledExceptionBus are components: replace one with rebind: true, for example to also send the events to a message broker:

@Component({ id: EventBus, rebind: true })
export class BrokerEventBus extends EventBusImpl {
    override async publish(event: object): Promise<void> {
        await super.publish(event);
        await this.broker.send(event);
    }
}

Coming from @nestjs/cqrs

@nestjs/cqrs	@celljs/cqrs
CqrsModule.forRoot()	Add the dependency; the module is loaded with the backend
ICommandHandler, IQueryHandler, IEventHandler	CommandHandler, QueryHandler, EventHandler: one name for the decorator and the interface
@EventsHandler(A, B)	@EventHandler([A, B])
Command<R>, Query<R>	CommandMessage<R>, QueryMessage<R> (@celljs/mvc has a @Query decorator)
@Saga() with RxJS streams	@Saga(Event) class whose handle(event) returns the commands; no RxJS
eventBus.publish() returns at once	await eventBus.publish() waits for the handlers and sagas
aggregate.commit()	await aggregate.commit(); rejects when the aggregate is not connected instead of doing nothing
autoCommit	Not supported: call commit()
UnhandledExceptionBus (RxJS subject)	UnhandledExceptionBus.onException (a Cell Event)