task-stream/libs/api/tracker/application/src/lib/changes.service.ts

import { Injectable } from '@nestjs/common';
import { QueryBus } from '@nestjs/cqrs';
import { NotFoundError } from '@task-stream/api/shared/domain';
import { ActivityReadModel } from '@task-stream/api/tracker/domain';
import {
  ChangeFeed,
  CommentChange,
  IssueChange,
  ProjectChange,
  ReleaseChange,
  SprintChange,
  Typing,
} from './ports/change-feed';
import { CanViewProjectQuery } from './queries/can-view-project.query';
import { GetIssueQuery } from './queries/get-issue.query';
import { GetTicketScopeQuery } from './queries/get-ticket-scope.query';
import { involves } from './queries/read-access';

/** The changes of the tickets `userId` opened, has or follows. */
async function* onlyInvolving(
  changes: AsyncIterableIterator<IssueChange>,
  userId: string,
): AsyncIterableIterator<IssueChange> {
  for await (const change of changes)
    if (involves(change.issue, userId)) yield change;
}

/** Realtime subscriptions, authorised at subscribe time. */
@Injectable()
export class ChangesService {
  constructor(
    private readonly queryBus: QueryBus,
    private readonly changeFeed: ChangeFeed,
  ) {}

  async watchIssues(
    viewerId: string,
    projectId: string,
  ): Promise<AsyncIterableIterator<IssueChange>> {
    await this.assertCanView(viewerId, projectId);
    const changes = this.changeFeed.issues(projectId);
    const scope: string | null = await this.queryBus.execute(
      new GetTicketScopeQuery(viewerId, projectId),
    );
    // A guest hears only about the tickets they take part in.
    return scope ? onlyInvolving(changes, scope) : changes;
  }

  async watchComments(
    viewerId: string,
    issueId: string,
  ): Promise<AsyncIterableIterator<CommentChange>> {
    await this.queryBus.execute(new GetIssueQuery(viewerId, issueId)); // access check
    return this.changeFeed.comments(issueId);
  }

  /** Who is writing on the ticket (the client leaves itself out). */
  async watchTyping(
    viewerId: string,
    issueId: string,
  ): Promise<AsyncIterableIterator<Typing>> {
    await this.queryBus.execute(new GetIssueQuery(viewerId, issueId)); // access check
    return this.changeFeed.typing(issueId);
  }

  async watchActivity(
    viewerId: string,
    projectId: string,
  ): Promise<AsyncIterableIterator<ActivityReadModel>> {
    await this.assertCanView(viewerId, projectId, true);
    return this.changeFeed.activity(projectId);
  }

  async watchReleases(
    viewerId: string,
    projectId: string,
  ): Promise<AsyncIterableIterator<ReleaseChange>> {
    await this.assertCanView(viewerId, projectId, true);
    return this.changeFeed.releases(projectId);
  }

  async watchSprints(
    viewerId: string,
    projectId: string,
  ): Promise<AsyncIterableIterator<SprintChange>> {
    await this.assertCanView(viewerId, projectId, true);
    return this.changeFeed.sprints(projectId);
  }

  async watchProject(
    viewerId: string,
    projectId: string,
  ): Promise<AsyncIterableIterator<ProjectChange>> {
    await this.assertCanView(viewerId, projectId);
    return this.changeFeed.project(projectId);
  }

  /** Projects the viewer is added to or removed from. */
  watchMyProjects(viewerId: string): AsyncIterableIterator<ProjectChange> {
    return this.changeFeed.userProjects(viewerId);
  }

  private async assertCanView(
    viewerId: string,
    projectId: string,
    wholeProject = false,
  ): Promise<void> {
    const allowed = await this.queryBus.execute(
      new CanViewProjectQuery(viewerId, projectId, wholeProject),
    );
    if (!allowed) throw new NotFoundError('Project', projectId);
  }
}