feat(api): notifications feed filtering by partial payload object

apps/api/src/app/subscribers/dtos/get-in-app-notification-feed-for-subscriber.dto.ts

@@ -26,4 +26,12 @@ export class GetInAppNotificationsFeedForSubscriberDto extends PaginationRequest
 
   @ApiPropertyOptional({ required: false, type: Boolean })
   seen: boolean;
+
+  @ApiPropertyOptional({
+    required: false,
+    type: 'string',
+    description: 'Base64 encoded string of the partial payload JSON object',
+    example: 'btoa(JSON.stringify({ foo: 123 })) results in base64 encoded string like eyJmb28iOjEyM30=',
+  })
+  payload?: string;
 }

apps/api/src/app/subscribers/e2e/get-notifications-feed.e2e.ts

@@ -38,6 +38,57 @@ describe('Get Notifications feed - /:subscriberId/notifications/feed (GET)', fun
       );
     }
   });
+
+  it('should throw exception when invalid payload query param is passed', async function () {
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId);
+
+    await session.awaitRunningJobs(template._id);
+
+    try {
+      await getNotificationsFeed(subscriberId, session.apiKey, { limit: 5, payload: 'invalid' });
+    } catch (err) {
+      expect(err.response.status).to.equals(400);
+      expect(err.response.data.message).to.eq(`Invalid payload, the JSON object should be encoded to base64 string.`);
+
+      return;
+    }
+
+    expect.fail('Should have thrown an bad request exception');
+  });
+
+  it('should allow filtering by custom data from the payload', async function () {
+    const partialPayload = { foo: 123 };
+    const payload = { ...partialPayload, bar: 'bar' };
+
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId);
+    await session.awaitRunningJobs(template._id);
+
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId, payload);
+    await session.awaitRunningJobs(template._id);
+
+    const payloadQueryValue = Buffer.from(JSON.stringify(partialPayload)).toString('base64');
+    const { data } = await getNotificationsFeed(subscriberId, session.apiKey, { limit: 5, payload: payloadQueryValue });
+
+    expect(data.length).to.equal(1);
+    expect(data[0].payload).to.deep.equal(payload);
+  });
+
+  it('should allow filtering by custom nested data from the payload', async function () {
+    const partialPayload = { foo: { bar: 123 } };
+    const payload = { ...partialPayload, baz: 'baz' };
+
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId);
+    await session.awaitRunningJobs(template._id);
+
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId, payload);
+    await session.awaitRunningJobs(template._id);
+
+    const payloadQueryValue = Buffer.from(JSON.stringify(partialPayload)).toString('base64');
+    const { data } = await getNotificationsFeed(subscriberId, session.apiKey, { limit: 5, payload: payloadQueryValue });
+
+    expect(data.length).to.equal(1);
+    expect(data[0].payload).to.deep.equal(payload);
+  });
 });
 
 async function getNotificationsFeed(subscriberId: string, apiKey: string, query = {}) {

apps/api/src/app/subscribers/subscribers.controller.ts

@@ -371,6 +371,7 @@ export class SubscribersController {
       feedId: feedsQuery,
       query: { seen: query.seen, read: query.read },
       limit: query.limit != null ? parseInt(query.limit) : 10,
+      payload: query.payload,
     });
 
     return await this.getNotificationsFeedUsecase.execute(command);

apps/api/src/app/widgets/e2e/get-notification-feed.e2e.ts

@@ -157,6 +157,57 @@ describe('GET /widget/notifications/feed', function () {
     expect(feed.hasMore).to.be.equal(true);
   });
 
+  it('should throw exception when invalid payload query param is passed', async function () {
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId);
+
+    await session.awaitRunningJobs(template._id);
+
+    try {
+      await getSubscriberFeed({ payload: 'invalid' });
+    } catch (err) {
+      expect(err.response.status).to.equals(400);
+      expect(err.response.data.message).to.eq(`Invalid payload, the JSON object should be encoded to base64 string.`);
+
+      return;
+    }
+
+    expect.fail('Should have thrown an bad request exception');
+  });
+
+  it('should allow filtering by custom data from the payload', async function () {
+    const partialPayload = { foo: 123 };
+    const payload = { ...partialPayload, bar: 'bar' };
+
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId);
+    await session.awaitRunningJobs(template._id);
+
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId, payload);
+    await session.awaitRunningJobs(template._id);
+
+    const payloadQueryValue = Buffer.from(JSON.stringify(partialPayload)).toString('base64');
+    const { data } = await getSubscriberFeed({ payload: payloadQueryValue });
+
+    expect(data.length).to.equal(1);
+    expect(data[0].payload).to.deep.equal(payload);
+  });
+
+  it('should allow filtering by custom nested data from the payload', async function () {
+    const partialPayload = { foo: { bar: 123 } };
+    const payload = { ...partialPayload, baz: 'baz' };
+
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId);
+    await session.awaitRunningJobs(template._id);
+
+    await session.triggerEvent(template.triggers[0].identifier, subscriberId, payload);
+    await session.awaitRunningJobs(template._id);
+
+    const payloadQueryValue = Buffer.from(JSON.stringify(partialPayload)).toString('base64');
+    const { data } = await getSubscriberFeed({ payload: payloadQueryValue });
+
+    expect(data.length).to.equal(1);
+    expect(data[0].payload).to.deep.equal(payload);
+  });
+
   async function getSubscriberFeed(query = {}) {
     const response = await axios.get(`http://localhost:${process.env.PORT}/v1/widgets/notifications/feed`, {
       params: {

apps/api/src/app/widgets/usecases/get-notifications-feed/get-notifications-feed.command.ts

@@ -1,7 +1,7 @@
-import { IsArray, IsNumber, IsOptional, Max, Min } from 'class-validator';
+import { IsArray, IsNumber, IsOptional, IsString } from 'class-validator';
+
 import { EnvironmentWithSubscriber } from '../../../shared/commands/project.command';
 import { StoreQuery } from '../../queries/store.query';
-import { Transform } from 'class-transformer';
 
 export class GetNotificationsFeedCommand extends EnvironmentWithSubscriber {
   @IsNumber()
@@ -18,4 +18,8 @@ export class GetNotificationsFeedCommand extends EnvironmentWithSubscriber {
 
   @IsOptional()
   query: StoreQuery;
+
+  @IsOptional()
+  @IsString()
+  payload?: string;
 }

apps/api/src/app/widgets/usecases/get-notifications-feed/get-notifications-feed.usecase.ts

@@ -1,4 +1,4 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, BadRequestException } from '@nestjs/common';
 import { ActorTypeEnum, ChannelTypeEnum } from '@novu/shared';
 import {
   AnalyticsService,
@@ -21,6 +21,18 @@ export class GetNotificationsFeed {
     private subscriberRepository: SubscriberRepository
   ) {}
 
+  private getPayloadObject(payload?: string): object | undefined {
+    if (!payload) {
+      return;
+    }
+
+    try {
+      return JSON.parse(Buffer.from(payload, 'base64').toString());
+    } catch (e) {
+      throw new BadRequestException('Invalid payload, the JSON object should be encoded to base64 string.');
+    }
+  }
+
   @CachedQuery({
     builder: ({ environmentId, subscriberId, ...command }: GetNotificationsFeedCommand) =>
       buildFeedKey().cache({
@@ -30,6 +42,8 @@ export class GetNotificationsFeed {
       }),
   })
   async execute(command: GetNotificationsFeedCommand): Promise<MessagesResponseDto> {
+    const payload = this.getPayloadObject(command.payload);
+
     const subscriber = await this.fetchSubscriber({
       _environmentId: command.environmentId,
       subscriberId: command.subscriberId,
@@ -47,7 +61,7 @@ export class GetNotificationsFeed {
       command.environmentId,
       subscriber._id,
       ChannelTypeEnum.IN_APP,
-      { feedId: command.feedId, seen: command.query.seen, read: command.query.read },
+      { feedId: command.feedId, seen: command.query.seen, read: command.query.read, payload },
       {
         limit: command.limit,
         skip: command.page * command.limit,
@@ -80,6 +94,7 @@ export class GetNotificationsFeed {
           feedId: command.feedId,
           seen: command.query.seen,
           read: command.query.read,
+          payload,
         },
         { limit: command.limit + 1, skip }
       );

apps/api/src/app/widgets/widgets.controller.ts

@@ -112,6 +112,7 @@ export class WidgetsController {
       feedId: feedsQuery,
       query: { seen: query.seen, read: query.read },
       limit: query.limit != null ? parseInt(query.limit) : 10,
+      payload: query.payload,
     });
 
     return await this.getNotificationsFeedUsecase.execute(command);

libs/dal/src/repositories/message/message.repository.ts

@@ -11,6 +11,15 @@ import { EnforceEnvId } from '../../types/enforce';
 
 type MessageQuery = FilterQuery<MessageDBModel>;
 
+const getEntries = (obj: object, prefix = '') =>
+  Object.entries(obj).flatMap(([key, value]) =>
+    Object(value) === value ? getEntries(value, `${prefix}${key}.`) : [[`${prefix}${key}`, value]]
+  );
+
+const getFlatObject = (obj: object) => {
+  return Object.fromEntries(getEntries(obj));
+};
+
 export class MessageRepository extends BaseRepository<MessageDBModel, MessageEntity, EnforceEnvId> {
   private message: SoftDeleteModel;
   private feedRepository = new FeedRepository();
@@ -23,9 +32,9 @@ export class MessageRepository extends BaseRepository<MessageDBModel, MessageEnt
     environmentId: string,
     subscriberId: string,
     channel: ChannelTypeEnum,
-    query: { feedId?: string[]; seen?: boolean; read?: boolean } = {}
+    query: { feedId?: string[]; seen?: boolean; read?: boolean; payload?: object } = {}
   ): Promise<MessageQuery & EnforceEnvId> {
-    const requestQuery: MessageQuery & EnforceEnvId = {
+    let requestQuery: MessageQuery & EnforceEnvId = {
       _environmentId: environmentId,
       _subscriberId: subscriberId,
       channel,
@@ -62,14 +71,21 @@ export class MessageRepository extends BaseRepository<MessageDBModel, MessageEnt
       requestQuery.read = { $in: [true, false] };
     }
 
+    if (query.payload) {
+      requestQuery = {
+        ...requestQuery,
+        ...getFlatObject({ payload: query.payload }),
+      };
+    }
+
     return requestQuery;
   }
 
   async findBySubscriberChannel(
     environmentId: string,
     subscriberId: string,
     channel: ChannelTypeEnum,
-    query: { feedId?: string[]; seen?: boolean; read?: boolean } = {},
+    query: { feedId?: string[]; seen?: boolean; read?: boolean; payload?: object } = {},
     options: { limit: number; skip?: number } = { limit: 10 }
   ) {
     const requestQuery = await this.getFilterQueryForMessage(environmentId, subscriberId, channel, query);
@@ -90,13 +106,14 @@ export class MessageRepository extends BaseRepository<MessageDBModel, MessageEnt
     environmentId: string,
     subscriberId: string,
     channel: ChannelTypeEnum,
-    query: { feedId?: string[]; seen?: boolean; read?: boolean } = {},
+    query: { feedId?: string[]; seen?: boolean; read?: boolean; payload?: object } = {},
     options: { limit: number; skip?: number } = { limit: 100, skip: 0 }
   ) {
     const requestQuery = await this.getFilterQueryForMessage(environmentId, subscriberId, channel, {
       feedId: query.feedId,
       seen: query.seen,
       read: query.read,
+      payload: query.payload,
     });
 
     return this.MongooseModel.countDocuments(requestQuery, options).read('secondaryPreferred');

packages/client/src/api/api.service.ts

@@ -92,13 +92,16 @@ export class ApiService {
 
   async getNotificationsList(
     page: number,
-    query: IStoreQuery = {}
+    { payload, ...rest }: IStoreQuery = {}
   ): Promise<IPaginatedResponse<IMessage>> {
+    const payloadString = payload ? btoa(JSON.stringify(payload)) : undefined;
+
     return await this.httpClient.getFullResponse(
       `/widgets/notifications/feed`,
       {
         page,
-        ...query,
+        payload: payloadString,
+        ...rest,
       }
     );
   }

packages/client/src/index.ts

@@ -20,6 +20,7 @@ export interface IStoreQuery {
   seen?: boolean;
   read?: boolean;
   limit?: number;
+  payload?: Record<string, unknown>;
 }
 
 export interface ITabCountQuery {

packages/node/src/lib/subscribers/subscriber.interface.ts

@@ -66,6 +66,7 @@ export interface IGetSubscriberNotificationFeedParams {
   feedIdentifier?: string;
   seen?: boolean;
   read?: boolean;
+  payload?: Record<string, unknown>;
 }
 
 export interface IMarkFields {

packages/node/src/lib/subscribers/subscribers.ts

@@ -96,12 +96,19 @@ export class Subscribers extends WithHttp implements ISubscribers {
 
   async getNotificationsFeed(
     subscriberId: string,
-    params: IGetSubscriberNotificationFeedParams
+    { payload, ...rest }: IGetSubscriberNotificationFeedParams = {}
   ) {
+    const payloadString = payload
+      ? Buffer.from(JSON.stringify(payload)).toString('base64')
+      : undefined;
+
     return await this.http.get(
       `/subscribers/${subscriberId}/notifications/feed`,
       {
-        params,
+        params: {
+          payload: payloadString,
+          ...rest,
+        },
       }
     );
   }