feat: archive SUCCESS/FAILED notifications

apps/api/.env.example

@@ -14,6 +14,7 @@ NODE_ENV= # Use "development" for graphql playground to work
 
 # Notification configuration
 MAX_RETRY_COUNT=3 # Max retry count, default is 3
+ARCHIVE_LIMIT=1000 # Max notifications to archive, default is 1000
 
 # Logger configuration
 LOG_LEVEL=info # Log level, default is info

apps/api/src/app.module.ts

@@ -13,6 +13,7 @@ import { AuthModule } from './modules/auth/auth.module';
 import { WebhookModule } from './modules/webhook/webhook.module';
 import { APP_INTERCEPTOR } from '@nestjs/core';
 import { DatabaseErrorInterceptor } from './database/database-error.interceptor';
+import { ArchivedNotificationsModule } from './modules/archived-notifications/archived-notifications.module';
 
 const configService = new ConfigService();
 @Module({
@@ -39,6 +40,7 @@ const configService = new ConfigService();
       playground: configService.getOrThrow('NODE_ENV') === 'development',
     }),
     AuthModule,
+    ArchivedNotificationsModule,
   ],
   controllers: [AppController],
   providers: [

apps/api/src/database/migrations/1730724383210-ArchiveCompletedNotifications.ts

@@ -0,0 +1,152 @@
+import { MigrationInterface, QueryRunner, Table, TableForeignKey } from 'typeorm';
+
+export class ArchiveCompletedNotifications1730724383210 implements MigrationInterface {
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    const table = await queryRunner.getTable('notify_notification_retries');
+    const foreignKey = table.foreignKeys.find(
+      (fk) => fk.columnNames.indexOf('notification_id') !== -1,
+    );
+    await queryRunner.dropForeignKey('notify_notification_retries', foreignKey);
+
+    await queryRunner.createTable(
+      new Table({
+        name: 'archived_notifications',
+        columns: [
+          {
+            name: 'id',
+            type: 'int',
+            isPrimary: true,
+            isGenerated: true,
+            generationStrategy: 'increment',
+          },
+          {
+            name: 'notification_id',
+            type: 'int',
+          },
+          {
+            name: 'channel_type',
+            type: 'tinyint',
+          },
+          {
+            name: 'data',
+            type: 'text',
+            isNullable: true,
+          },
+          {
+            name: 'delivery_status',
+            type: 'tinyint',
+            default: 1,
+          },
+          {
+            name: 'result',
+            type: 'text',
+            isNullable: true,
+          },
+          {
+            name: 'created_on',
+            type: 'timestamp',
+            default: 'CURRENT_TIMESTAMP',
+          },
+          {
+            name: 'updated_on',
+            type: 'timestamp',
+            default: 'CURRENT_TIMESTAMP',
+            onUpdate: 'CURRENT_TIMESTAMP',
+          },
+          {
+            name: 'created_by',
+            type: 'varchar',
+          },
+          {
+            name: 'updated_by',
+            type: 'varchar',
+          },
+          {
+            name: 'status',
+            type: 'tinyint',
+            default: 1,
+          },
+          {
+            name: 'application_id',
+            type: 'int',
+            isNullable: true,
+          },
+          {
+            name: 'provider_id',
+            type: 'int',
+            isNullable: true,
+          },
+          {
+            name: 'retry_count',
+            type: 'int',
+            length: '3',
+            default: 0,
+          },
+        ],
+      }),
+    );
+
+    // Create Foreign keys for archived_notifications
+    await queryRunner.createForeignKey(
+      'archived_notifications',
+      new TableForeignKey({
+        columnNames: ['channel_type'],
+        referencedColumnNames: ['master_id'],
+        referencedTableName: 'notify_master_providers',
+        onDelete: 'CASCADE',
+      }),
+    );
+
+    await queryRunner.createForeignKey(
+      'archived_notifications',
+      new TableForeignKey({
+        columnNames: ['provider_id'],
+        referencedColumnNames: ['provider_id'],
+        referencedTableName: 'notify_providers',
+        onDelete: 'CASCADE',
+      }),
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    // To drop the auto generated foreign key for archived_notifications
+    const archived_notifications_table = await queryRunner.getTable('archived_notifications');
+    const archived_notifications_providerIdforeignKey =
+      archived_notifications_table?.foreignKeys.find(
+        (fk) => fk.columnNames.indexOf('provider_id') !== -1,
+      );
+
+    // providerIdforeignKey
+    if (archived_notifications_providerIdforeignKey) {
+      await queryRunner.dropForeignKey(
+        'archived_notifications',
+        archived_notifications_providerIdforeignKey,
+      );
+    }
+
+    // channelTypeforeignKey
+    const archived_notifications_channelTypeforeignKey =
+      archived_notifications_table?.foreignKeys.find(
+        (fk) => fk.columnNames.indexOf('channel_type') !== -1,
+      );
+
+    if (archived_notifications_channelTypeforeignKey) {
+      await queryRunner.dropForeignKey(
+        'archived_notifications',
+        archived_notifications_channelTypeforeignKey,
+      );
+    }
+
+    await queryRunner.dropTable('archived_notifications');
+
+    await queryRunner.createForeignKey(
+      'notify_notification_retries',
+      new TableForeignKey({
+        columnNames: ['notification_id'],
+        referencedColumnNames: ['id'],
+        referencedTableName: 'notify_notifications',
+        onDelete: 'CASCADE',
+      }),
+    );
+  }
+}

apps/api/src/modules/applications/entities/application.entity.ts

@@ -11,6 +11,7 @@ import { Status } from 'src/common/constants/database';
 import { Field, ObjectType } from '@nestjs/graphql';
 import { Notification } from 'src/modules/notifications/entities/notification.entity';
 import { ServerApiKey } from 'src/modules/server-api-keys/entities/server-api-key.entity';
+import { ArchivedNotification } from 'src/modules/archived-notifications/entities/archived-notification.entity';
 
 @Entity({ name: 'notify_applications' })
 @ObjectType()
@@ -52,6 +53,12 @@ export class Application {
   @OneToMany(() => ServerApiKey, (serverApiKey) => serverApiKey.applicationDetails)
   serverApiKey: ServerApiKey[];
 
+  @OneToMany(
+    () => ArchivedNotification,
+    (archivedNotification) => archivedNotification.applicationDetails,
+  )
+  archivedNotifications: ArchivedNotification[];
+
   constructor(application: Partial<Application>) {
     Object.assign(this, application);
   }

apps/api/src/modules/archived-notifications/archived-notifications.module.ts

@@ -0,0 +1,15 @@
+import { forwardRef, Logger, Module } from '@nestjs/common';
+import { ArchivedNotificationsService } from './archived-notifications.service';
+import { ArchivedNotification } from './entities/archived-notification.entity';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { NotificationsModule } from '../notifications/notifications.module';
+
+@Module({
+  imports: [
+    TypeOrmModule.forFeature([ArchivedNotification]),
+    forwardRef(() => NotificationsModule),
+  ],
+  providers: [ArchivedNotificationsService, Logger],
+  exports: [ArchivedNotificationsService],
+})
+export class ArchivedNotificationsModule {}

apps/api/src/modules/archived-notifications/archived-notifications.service.spec.ts

@@ -0,0 +1,18 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { ArchivedNotificationsService } from './archived-notifications.service';
+
+describe('ArchivedNotificationsService', () => {
+  let service: ArchivedNotificationsService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [ArchivedNotificationsService],
+    }).compile();
+
+    service = module.get<ArchivedNotificationsService>(ArchivedNotificationsService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+});

apps/api/src/modules/archived-notifications/archived-notifications.service.ts

@@ -0,0 +1,86 @@
+import { forwardRef, Inject, Injectable, Logger } from '@nestjs/common';
+import { ArchivedNotification } from './entities/archived-notification.entity';
+import { Cron, CronExpression } from '@nestjs/schedule';
+import { Repository } from 'typeorm';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Notification } from 'src/modules/notifications/entities/notification.entity';
+import { NotificationsService } from '../notifications/notifications.service';
+import { ConfigService } from '@nestjs/config';
+
+const configService = new ConfigService();
+
+@Injectable()
+export class ArchivedNotificationsService {
+  protected readonly logger = new Logger(ArchivedNotificationsService.name);
+
+  constructor(
+    @InjectRepository(ArchivedNotification)
+    private readonly archivedNotificationRepository: Repository<ArchivedNotification>,
+    @Inject(forwardRef(() => NotificationsService))
+    protected readonly notificationsService: NotificationsService,
+  ) {}
+
+  private convertToArchivedNotifications(notifications: Notification[]): ArchivedNotification[] {
+    return notifications.map((notification) => {
+      const archivedNotification = new ArchivedNotification();
+      archivedNotification.applicationId = notification.applicationId;
+      archivedNotification.channelType = notification.channelType;
+      archivedNotification.createdBy = notification.createdBy;
+      archivedNotification.createdOn = notification.createdOn;
+      archivedNotification.data = notification.data;
+      archivedNotification.deliveryStatus = notification.deliveryStatus;
+      archivedNotification.notification_id = notification.id;
+      archivedNotification.providerId = notification.providerId;
+      archivedNotification.result = notification.result;
+      archivedNotification.retryCount = notification.retryCount;
+      archivedNotification.updatedBy = notification.updatedBy;
+      archivedNotification.updatedOn = notification.updatedOn;
+      archivedNotification.status = notification.status;
+
+      this.logger.debug(
+        `Created ArchivedNotification array using Notification ID: ${notification.id}, deliveryStatus: ${notification.deliveryStatus}`,
+      );
+      return archivedNotification;
+    });
+  }
+
+  async moveNotificationsToArchive(): Promise<void> {
+    const archiveLimit = configService.get<number>('ARCHIVE_LIMIT', 1000);
+
+    try {
+      // Step 1: Retrieve the notifications to archive
+      this.logger.log(`Retrieve the top ${archiveLimit} notifications to archive`);
+      const notificationsToArchive =
+        await this.notificationsService.findNotificationsToArchive(archiveLimit);
+
+      if (notificationsToArchive.length === 0) {
+        this.logger.log('No notifications to archive at this time.');
+        return;
+      }
+
+      // Step 2: Convert notifications to archived notifications
+      const archivedNotificationsArray =
+        this.convertToArchivedNotifications(notificationsToArchive);
+
+      // Step 3: Insert notifications into the archive table
+      this.logger.log(`Inserting archived notifications into the archive table`);
+      await this.archivedNotificationRepository.save(archivedNotificationsArray, {
+        transaction: true,
+      });
+
+      // Step 4: Delete notifications from the main table by IDs
+      this.logger.log(`Deleting notifications from the main table by IDs`);
+      await this.notificationsService.deleteArchivedNotifications(notificationsToArchive);
+
+      this.logger.log(`Archive notifications task completed`);
+    } catch (error) {
+      this.logger.error('Failed to archive notifications:', error);
+    }
+  }
+
+  @Cron(CronExpression.EVERY_HOUR)
+  async ArchiveCompletedNotificationsCron(): Promise<void> {
+    this.logger.log('Running archive notifications task');
+    this.moveNotificationsToArchive();
+  }
+}