如何从在 NestJS CQRS 的传奇中失败的后续命令引发 HTTP 异常？

Question

我正在使用 NestJS CQRS 配方来管理两个实体之间的交互：用户和用户配置文件。该架构是一个 API Gateway NestJS 服务器 + 每个微服务（User、UserProfile 等）的 NestJS 服务器。

我已经通过 API Gateway 上的 User 和 UserProfile 模块使用他们自己的 sagas/events/commands 设置了基本交互：

• 创建用户时，会创建用户配置文件
• 当用户配置文件创建失败时，之前创建的用户被删除

详细说明：

在 User module 中，CreateUser command 引发一个 UserCreated event 被 User saga 拦截，这将触发 CreateUserProfile command（来自 UserProfile module）。

如果后者失败，UserProfileFailedToCreate event 会引发并被 UserProfile saga 拦截，这将触发 DeleteUser command（来自 User模块）。

一切正常。

如果 CreateUser 命令失败，我resolve(Promise.reject(new HttpException(error, error.status)) 向最终用户表明在用户创建过程中出现了问题。

我的问题是我无法为 CreateUserProfile 命令复制相同的行为，因为显然 HTTP 请求承诺已经从第一个命令中解决了。

所以我的问题是：如果后续命令在 saga 中失败，有什么方法可以使命令失败？我知道 HTTP 请求与 saga 触发的任何后续命令完全断开，但我想知道是否有人已经在这里玩过事件或其他东西来复制这个数据流？

我使用 CQRS 的原因之一是，除了为微服务之间的数据交互提供更简洁的代码之外，还能够在任何链接命令失败的情况下回滚存储库操作，这可以正常工作。 但我需要一种方法向最终用户表明该链遇到问题并已回滚。

UserController.ts

@Post('createUser')
async createUser(@Body() createUserDto: CreateUserDto): Promise<{user: IAuthUser, token: string}> {
  const { authUser } = await this.authService.createAuthUser(createUserDto);
  // this is executed after resolve() in CreateUserCommand
  return {user: authUser, token: this.authService.createAccessTokenFromUser(authUser)};
}

UserService.ts

async createAuthUser(createUserDto: CreateUserDto): Promise<{authUser: IAuthUser}> {
  return await this.commandBus
    .execute(new CreateAuthUserCommand(createUserDto))
    .catch(error => { throw new HttpException(error, error.status); });
}

CreateUserCommand.ts

async execute(command: CreateAuthUserCommand, resolve: (value?) => void) {
    const { createUserDto } = command;
    const createAuthUserDto: CreateAuthUserDto = {
      email: createUserDto.email,
      password: createUserDto.password,
      phoneNumber: createUserDto.phoneNumber,
    };

    try {
      const user = this.publisher.mergeObjectContext(
        await this.client
          .send<IAuthUser>({ cmd: 'createAuthUser' }, createAuthUserDto)
          .toPromise()
          .then((dbUser: IAuthUser) => {
            const {password, passwordConfirm, ...publicUser} = Object.assign(dbUser, createUserDto);
            return new AuthUser(publicUser);
          }),
      );
      user.notifyCreated();
      user.commit();
      resolve(user); // <== This makes the HTTP request return its reponse
    } catch (error) {
      resolve(Promise.reject(error));
    }
  }

UserSagas.ts

authUserCreated = (event$: EventObservable<any>): Observable<ICommand> => {
    return event$
      .ofType(AuthUserCreatedEvent)
      .pipe(
        map(event => {
          const createUserProfileDto: CreateUserProfileDto = {
            avatarUrl: '',
            firstName: event.authUser.firstName,
            lastName: event.authUser.lastName,
            nationality: '',
            userId: event.authUser.id,
            username: event.authUser.username,
          };
          return new CreateUserProfileCommand(createUserProfileDto);
        }),
      );
  }

CreateUserProfileCommand.ts

async execute(command: CreateUserProfileCommand, resolve: (value?) => void) {
    const { createUserProfileDto } = command;

    try {
      const userProfile = this.publisher.mergeObjectContext(
        await this.client
          .send<IUserProfile>({ cmd: 'createUserProfile' }, createUserProfileDto)
          .toPromise()
          .then((dbUserProfile: IUserProfile) => new UserProfile(dbUserProfile)),
      );
      userProfile.notifyCreated();
      userProfile.commit();
      resolve(userProfile);
    } catch (error) {
      const userProfile = this.publisher.mergeObjectContext(new UserProfile({id: createUserProfileDto.userId} as IUserProfile));
      userProfile.notifyFailedToCreate();
      userProfile.commit();
      resolve(Promise.reject(new HttpException(error, 500)).catch(() => {}));
    }
  }

UserProfileSagas.ts

userProfileFailedToCreate = (event$: EventObservable<any>): Observable<ICommand> => {
    return event$
      .ofType(UserProfileFailedToCreateEvent)
      .pipe(
        map(event => {
          return new DeleteAuthUserCommand(event.userProfile);
        }),
      );
  }

DeleteUserCommand.ts

async execute(command: DeleteAuthUserCommand, resolve: (value?) => void) {
    const { deleteAuthUserDto } = command;

    try {
      const user = this.publisher.mergeObjectContext(
        await this.client
          .send<IAuthUser>({ cmd: 'deleteAuthUser' }, deleteAuthUserDto)
          .toPromise()
          .then(() => new AuthUser({} as IAuthUser)),
      );
      user.notifyDeleted();
      user.commit();
      resolve(user);
    } catch (error) {
      resolve(Promise.reject(new HttpException(error, error.status)).catch(() => {}));
    }
  }

Answer 1

在 DDD 术语中，User 和 UserProfile 的创建构成业务事务 - 一组必须一致的业务操作/规则 - spans multiple microservices。

在这种情况下，在创建 UserProfile 之前返回数据库 User 意味着您返回的数据状态不一致。这不一定是错误的，但如果你这样做，你应该在客户端适当地处理它。

我看到了三种可能的方式来处理这种情况：

1. 您让 Sagas 一直运行，直到它们执行指示业务事务已结束的命令，然后才为客户端解析指示成功或失败的结果（例如，在错误详细信息中，您可以报告哪些步骤成功，哪些没有） .所以你还没有解决CreateAuthUserCommand。

2. 如果创建 UserProfile 可能需要很长时间（甚至可能必须由版主手动验证），那么您可能需要解析 CreateAuthUserCommand 中的 User，然后再使用客户端订阅与 UserProfile 相关的事件。为此你需要一种机制，但它将客户端与正在运行的事务分离，并且它可以做其他事情。

3. 或者，您可以将业务事务分解为客户端发送单独请求的两个部分：一个创建/返回经过身份验证的User，另一个返回创建的UserProfile。尽管User + UserProfile 似乎属于同一个有界上下文，但它们驻留在两个不同的微服务中的事实可能表明它们不是（在这种情况下，我认为第一个微服务确实用于身份验证，另一个用于 UserProfiles这向我表明了不同的有界上下文）。最佳实践是让微服务实现自己封装的有界上下文。

（注意：回答了一个老问题，希望对其他人有所帮助）