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

Question

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

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

创建用户时，会创建用户配置文件

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

详细说明:

在用户 模块 , 创建用户 命令 引发 UserCreated 事件 被用户 截获传奇 ，这将触发 CreateUserProfile 命令 (来自 UserProfile 模块 )。

如果后者失败，则 UserProfileFailedToCreate 事件 被 UserProfile 提出并拦截传奇 ，这将触发 DeleteUser 命令 (来自用户 模块 )。

一切正常。

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

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

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

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

用户 Controller .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)};
}

用户服务.ts

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

创建用户命令.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);
        }),
      );
  }

创建用户配置文件命令.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(() => {}));
    }
  }

用户配置文件Sagas.ts

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

删除用户命令.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 .
在这种情况下返回数据库 User在 UserProfile 之前已创建意味着您以不一致的状态返回数据。这不一定是错误的，但如果你这样做，你应该在客户端适本地处理这个问题。
我看到了三种可能的方法来处理这种情况:

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

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

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

(注:回答了一个老问题，希望对其他人有所帮助)