Notification System with Apollo GraphQL and NestJS

chat-messages.resolver.ts

/*
 * This is just an example
*/

import { Inject, ParseIntPipe, UseGuards } from '@nestjs/common';
import {
  Args,
  Int,
  Mutation,
  Resolver,
  Subscription,
} from '@nestjs/graphql';

const MESSAGE_NOTIFICATION = 'messageNotification';

@Resolver(() => ChatMessageEntity)
export class ChatMessagesResolver {
  constructor(
    private readonly chatMessagesService: ChatMessagesService,
    private readonly commonService: CommonService,
    @Inject(PUB_SUB)
    private readonly pubSub: RedisPubSub,
  ) {}

  //-------------------- Mutation --------------------

  @UseGuards(WsGuard)
  @Mutation(() => ChatMessageEntity)
  public async sendPrivateMessage(
    @CurrentUser() userId: number,
    @Args('input') input: MessageInput,
  ): Promise<ChatMessageEntity> {
    const chatMessage = await this.chatMessagesService.sendPrivateMessage(
      userId,
      input,
    );
    this.publishNotification(chatMessage, NotificationTypeEnum.NEW);
    return chatMessage;
  }

  @UseGuards(WsGuard)
  @Mutation(() => ChatMessageEntity)
  public async sendChatMessage(
    @CurrentUser() userId: number,
    @Args('input') input: MessageInput,
  ): Promise<ChatMessageEntity> {
    const chatMessage = await this.chatMessagesService.sendChatMessage(
      userId,
      input,
    );
    this.publishNotification(chatMessage, NotificationTypeEnum.NEW);
    return chatMessage;
  }

  @Mutation(() => ChatMessageEntity)
  public async updateChatMessage(
    @GetQuery() query: string,
    @CurrentUser() userId: number,
    @Args('input') input: MessageInput,
  ): Promise<ChatMessageEntity> {
    const chatMessage = await this.updateChatMessage(query, userId, input);
    this.publishNotification(chatMessage, NotificationTypeEnum.UPDATE);
    return chatMessage;
  }

  @Mutation(() => LocalMessageType)
  public async deleteChatMessage(
    @CurrentUser() userId: number,
    @Args(
      {
        name: 'messageId',
        type: () => Int,
      },
      ParseIntPipe,
    )
    messageId: number,
  ): Promise<LocalMessageType> {
    const chatMessage = await this.chatMessagesService.deleteChatMessage(
      userId,
      messageId,
    );
    this.publishNotification(chatMessage, NotificationTypeEnum.DELETE);
    return new LocalMessageType('Message deleted successfully');
  }

  @Mutation(() => ChatMessageEntity)
  public async deleteChatMessageFile(
    @GetQuery() query: string,
    @CurrentUser() userId: number,
    @Args(
      {
        name: 'fileId',
        type: () => Int,
      },
      ParseIntPipe,
    )
    fileId: number,
  ): Promise<ChatMessageEntity> {
    const chatMessage = await this.chatMessagesService.deleteChatMessageFile(
      query,
      userId,
      fileId,
    );
    this.publishNotification(chatMessage, NotificationTypeEnum.UPDATE);
    return chatMessage;
  }

  @Mutation(() => ChatMessageEntity)
  public async likeChatMessage(
    @CurrentUser() userId: number,
    @Args(
      {
        name: 'messageId',
        type: () => Int,
      },
      ParseIntPipe,
    )
    messageId: number,
  ): Promise<ChatMessageEntity> {
    const chatMessage = await this.chatMessagesService.likeChatMessage(
      userId,
      messageId,
    );
    this.publishNotification(chatMessage, NotificationTypeEnum.UPDATE);
    return chatMessage;
  }

  @Mutation(() => ChatMessageEntity)
  public async removeLikeFromChatMessage(
    @CurrentUser() userId: number,
    @Args(
      {
        name: 'messageId',
        type: () => Int,
      },
      ParseIntPipe,
    )
    messageId: number,
  ): Promise<ChatMessageEntity> {
    const chatMessage =
      await this.chatMessagesService.removeLikeFromChatMessage(
        userId,
        messageId,
      );
    this.publishNotification(chatMessage, NotificationTypeEnum.UPDATE);
    return chatMessage;
  }

  //-------------------- Queries --------------------
  
  ...

  //-------------------- Subscriptions --------------------

  @Subscription(() => ChatMessageNotification, {
    filter: (
      payload: {
        [MESSAGE_NOTIFICATION]: INotification<ChatMessageEntity>;
      },
      { chatId }: { chatId?: number },
      ctx: ICtx,
    ) => {
      const userId = contextToUser(ctx);
      const chat = payload[MESSAGE_NOTIFICATION].edge.node.chat;
      const isInChat = !!chat.members
        .getItems()
        .find((m) => m.user.id === userId);

      if (chatId) return chatId === chat.id && isInChat;

      return isInChat;
    },
  })
  public chatMessageNotification(
    @Args(
      {
        name: 'chatId',
        type: () => Int,
        nullable: true,
      },
      ParseIntPipe,
    ) // eslint-disable-next-line @typescript-eslint/no-unused-vars
    _?: number,
  ) {
    return this.pubSub.asyncIterator(MESSAGE_NOTIFICATION);
  }

  //-------------------- Publishing --------------------

  private publishNotification(
    message: ChatMessageEntity,
    nType: NotificationTypeEnum,
  ): void {
    this.pubSub.publish(MESSAGE_NOTIFICATION, {
      [MESSAGE_NOTIFICATION]: this.commonService.generateNotification(
        message,
        nType,
      ),
    });
  }
}

common.service.ts

import { Dictionary, FilterQuery } from '@mikro-orm/core';
import { EntityRepository, QueryBuilder } from '@mikro-orm/postgresql';
import {
  BadRequestException,
  ConflictException,
  Injectable,
  InternalServerErrorException,
  NotFoundException,
} from '@nestjs/common';
import { NotificationTypeEnum } from './enums/notification-type.enum';
import { localQueryOrder, QueryOrderEnum } from './enums/query-order.enum';
import { ICountResult } from './interfaces/count-result.interface';
import { INotification } from './interfaces/notification.interface';
import { IEdge, IPaginated } from './interfaces/paginated.interface';

@Injectable()
export class CommonService {
  //-------------------- Cursor Pagination --------------------
  private readonly buff = Buffer;

  /**
   * Paginate
   *
   * Takes an entity array and returns the paginated type of that entity array
   * It uses cursor pagination as recomended in https://graphql.org/learn/pagination/
   */
  public paginate<T>(
    instances: T[],
    totalCount: number,
    cursor: keyof T,
    first: number,
    innerCursor?: string,
  ): IPaginated<T> {
    const pages: IPaginated<T> = {
      totalCount,
      edges: [],
      pageInfo: {
        endCursor: '',
        hasNextPage: false,
      },
    };

    const len = instances.length;
    if (len > 0) {
      for (let i = 0; i < len; i++) {
        pages.edges.push(this.createEdge(instances[i], cursor, innerCursor));
      }
      pages.pageInfo.endCursor = pages.edges[len - 1].cursor;
      pages.pageInfo.hasNextPage = totalCount > first;
    }

    return pages;
  }

  /**
   * Create Edge
   *
   * Takes an instance, the cursor key and a innerCursor,
   * and generates a GraphQL edge
   */
  private createEdge<T>(
    instance: T,
    cursor: keyof T,
    innerCursor?: string,
  ): IEdge<T> {
    try {
      return {
        node: instance,
        cursor: this.encodeCursor(
          innerCursor ? instance[cursor][innerCursor] : instance[cursor],
        ),
      };
    } catch (_) {
      throw new InternalServerErrorException('The given cursor is invalid');
    }
  }

  /**
   * Encode Cursor
   *
   * Takes a date, string or number and returns the base 64
   * representation of it
   */
  private encodeCursor(val: Date | string | number): string {
    let str: string;

    if (val instanceof Date) {
      str = val.getTime().toString();
    } else if (typeof val === 'number' || typeof val === 'bigint') {
      str = val.toString();
    } else {
      str = val;
    }

    return this.buff.from(str, 'utf-8').toString('base64');
  }

  /**
   * Decode Cursor
   *
   * Takes a base64 cursor and returns the string or number value
   */
  public decodeCursor(cursor: string, isNum = false): string | number {
    const str = this.buff.from(cursor, 'base64').toString('utf-8');

    if (isNum) {
      const num = parseInt(str, 10);

      if (isNaN(num))
        throw new BadRequestException(
          'Cursor does not reference a valid number',
        );

      return num;
    }

    return str;
  }

  //-------------------- Notification Generation --------------------

  /**
   * Generate Notification
   *
   * Generates an entity notification
   */
  public generateNotification<T>(
    entity: T,
    nType: NotificationTypeEnum,
    cursor: keyof T,
    innerCursor?: string,
  ): INotification<T> {
    return {
      edge: this.createEdge(entity, cursor, innerCursor),
      type: nType,
    };
  }
}

edge.type.ts

import { Type } from '@nestjs/common';
import { Field, ObjectType } from '@nestjs/graphql';

export function Edge<T>(classRef: Type<T>): any {
  @ObjectType({ isAbstract: true })
  abstract class EdgeType {
    @Field(() => String)
    public cursor: string;

    @Field(() => classRef)
    public node: T;
  }
  return EdgeType;
}

notification-type.enum.ts

import { registerEnumType } from '@nestjs/graphql';

export enum NotificationTypeEnum {
  NEW = 'NEW',
  UPDATE = 'UPDATE',
  DELETE = 'DELETE',
}

registerEnumType(NotificationTypeEnum, {
  name: 'NotificationType',
})

notification.type.ts

import { Type } from '@nestjs/common';
import { Field, ObjectType } from '@nestjs/graphql';
import { NotificationTypeEnum } from '../enums/notification-type.enum';
import { Edge } from './edge.type';

export function Notification<T>(classRef: Type<T>): any {
  @ObjectType(`${classRef.name}NotificationEdge`)
  abstract class EdgeType extends Edge(classRef) {}

  @ObjectType({ isAbstract: true })
  abstract class NotificationType {
    @Field(() => NotificationTypeEnum)
    public type: NotificationTypeEnum;

    @Field(() => EdgeType)
    public edge: EdgeType;
  }
  return NotificationType;
}