使用 eventstore db 的事件溯源和 cqrs

Event Sourcing and cqrs with eventstore db

存储数据库和事件源,但我对投影和 cqrs 有疑问。 到目前为止,这是我调用突击队和命令处理程序的方式:

创建用户命令

export class CreateUserCommand implements ICommand {
  constructor(
    public readonly userDto: UserStruct,
  ) {}
}

命令处理程序:

export class CreateUserHandler implements ICommandHandler<CreateUserCommand> {
  constructor(private readonly publisher: EventPublisher) {}

  async execute(command: CreateUserCommand) {
    const { userDto } = command;
    const user = User.create(userDto);
        console.log(user.value)
    if (user.isLeft()) throw user.value;
    const userPublisher = this.publisher.mergeObjectContext(user.value);
        userPublisher.commit()
  }
}

事件:

export class UserCreatedEvent implements IEvent {
  static readonly NAME = "UniFtcIdade/user-registered";
  readonly $name = UserCreatedEvent.NAME;
  readonly $version = 0;
  constructor(
    public readonly aggregateId: string,
    public readonly state: { email: string; name: string },
    public readonly date: Date
  ) {
  }
}

域:

export class User extends AggregateRoot {
  public readonly name: string;
  public readonly email: string;

  private constructor (guid: string, name: string, email: string) {
    super()
    this.apply(new UserCreatedEvent(guid, {email, name}, new Date()));
  }
  static create(
    dto: UserStruct
  ): Either<InvalidNameError | InvalidEmailError, User> {
    const name: Either<InvalidNameError, Name> = Name.create(dto.name);
    const email: Either<InvalidEmailError, Email> = Email.create(dto.email);
    if (name.isLeft()) return left(name.value);
    if (email.isLeft()) return left(email.value);
    const user = new User(v4(),name.value.value, email.value.value);
    return right(user);
  }
}

但我对投影如何进入这种情况表示怀疑。 投影用于获取聚合体的当前状态 ??? 我应该有一个数据库作为 mongodb 来保存当前状态,也就是说,每次我调用命令处理程序并更改 mongodb 中的当前状态时??? eventstoredb 的投影是为了这个吗?保存聚合的当前状态 ??

在 CQRS 中,当使用 EventStoreDb 时,您的聚合必须设计为从事件恢复到状态。事件存储在具有唯一名称和标识符 (guid) 的流中。修改聚合时,您必须读取此流,并按顺序应用每个事件以恢复当前状态,然后再对聚合执行任何更改(这会生成更多事件)。为了保持完整性并处理乐观并发,您应该在聚合中进行简单的版本检查,计算旧事件 + 新事件以确定要保留的最新版本号。

我上面看到的问题如下。 您的聚合有一个构造函数和一个静态方法,该方法无需对当前状态进行任何验证即可生成事件,即:如果我使用相同的 guid 调用两次创建会怎样?

this.apply(new UserCreatedEvent(guid, {email, name}, new Date()));

您直接在此处申请状态。相反,您应该在 Create 方法中引发事件。

this.raiseEvent(new UserCreatedEvent(guid, {email, name}, new Date()));

应执行此操作以执行以下操作。

  • 添加到未提交事件列表
  • this.apply 调用

然后您应该将事件保存到命令处理程序中的 EventStoreDb。

async execute(command: CreateUserCommand) {
    const { userDto } = command;
    const user = eventRepository.Get<User>(command.Id);
    user.Create(userDto); // Can now check current state and fail if required.
    eventRepository.Save(user)
  }

这里的存储库很简单。它可以创建一个空的用户并在返回用户之前按顺序应用所有事件。 保存应该只读取未提交事件的列表并将它们保存到用户流。

这就是命令端完成的, 对于读取端,您可以使用所有用户的开箱即用类别投影,并将它们写入 mongo 以供不同的 API(不是您的命令处理程序)读取。