feat(server): socket module

packages/server/package.json

@@ -38,10 +38,12 @@
     "@nestjs/jwt": "^10.2.0",
     "@nestjs/passport": "^11.0.5",
     "@nestjs/platform-express": "^10.0.0",
+    "@nestjs/platform-socket.io": "^10.0.0",
     "@nestjs/schedule": "^4.1.2",
     "@nestjs/serve-static": "^5.0.3",
     "@nestjs/swagger": "^7.4.2",
     "@nestjs/throttler": "^6.2.1",
+    "@nestjs/websockets": "^10.0.0",
     "@supercharge/promise-pool": "^3.2.0",
     "@types/multer": "^1.4.11",
     "@types/nodemailer": "^6.4.17",
@@ -100,6 +102,7 @@
     "remeda": "^2.19.2",
     "rxjs": "^7.8.1",
     "serialize-interceptor": "^1.1.7",
+    "socket.io": "^4.8.1",
     "strategy": "^1.1.1",
     "stripe": "^16.10.0",
     "uniqid": "^5.2.0",

packages/server/src/main.ts

@@ -18,6 +18,12 @@ async function bootstrap() {
   const app = await NestFactory.create(AppModule);
   app.setGlobalPrefix('/api');
 
+  // Enable CORS for socket.io
+  app.enableCors({
+    origin: true,
+    credentials: true,
+  });
+
   // create and mount the middleware manually here
   app.use(new ClsMiddleware({}).use);
 

packages/server/src/modules/App/App.module.ts

@@ -94,6 +94,7 @@ import { BankingPlaidModule } from '../BankingPlaid/BankingPlaid.module';
 import { BankingCategorizeModule } from '../BankingCategorize/BankingCategorize.module';
 import { TenantModelsInitializeModule } from '../Tenancy/TenantModelsInitialize.module';
 import { BillLandedCostsModule } from '../BillLandedCosts/BillLandedCosts.module';
+import { SocketModule } from '../Socket/Socket.module';
 
 @Module({
   imports: [
@@ -226,6 +227,7 @@ import { BillLandedCostsModule } from '../BillLandedCosts/BillLandedCosts.module
     MiscellaneousModule,
     UsersModule,
     ContactsModule,
+    SocketModule,
   ],
   controllers: [AppController],
   providers: [

packages/server/src/modules/BankingPlaid/BankingPlaid.module.ts

@@ -1,5 +1,6 @@
 import { BullModule } from '@nestjs/bullmq';
 import { Module } from '@nestjs/common';
+import { SocketModule } from '../Socket/Socket.module';
 import { PlaidUpdateTransactionsOnItemCreatedSubscriber } from './subscribers/PlaidUpdateTransactionsOnItemCreatedSubscriber';
 import { PlaidUpdateTransactions } from './command/PlaidUpdateTransactions';
 import { PlaidSyncDb } from './command/PlaidSyncDB';
@@ -26,6 +27,7 @@ const models = [RegisterTenancyModel(PlaidItem)];
 
 @Module({
   imports: [
+    SocketModule,
     PlaidModule,
     AccountsModule,
     BankingCategorizeModule,
@@ -49,4 +51,4 @@ const models = [RegisterTenancyModel(PlaidItem)];
   exports: [...models],
   controllers: [BankingPlaidController, BankingPlaidWebhooksController],
 })
-export class BankingPlaidModule {}
+export class BankingPlaidModule { }

packages/server/src/modules/BankingPlaid/jobs/PlaidFetchTransactionsJob.ts

@@ -10,6 +10,7 @@ import {
 } from '../types/BankingPlaid.types';
 import { PlaidUpdateTransactions } from '../command/PlaidUpdateTransactions';
 import { SetupPlaidItemTenantService } from '../command/SetupPlaidItemTenant.service';
+import { SocketGateway } from '../../Socket/Socket.gateway';
 
 @Processor({
   name: UpdateBankingPlaidTransitionsQueueJob,
@@ -19,6 +20,7 @@ export class PlaidFetchTransactionsProcessor extends WorkerHost {
   constructor(
     private readonly plaidFetchTransactionsService: PlaidUpdateTransactions,
     private readonly setupPlaidItemService: SetupPlaidItemTenantService,
+    private readonly socketGateway: SocketGateway,
   ) {
     super();
   }
@@ -38,7 +40,7 @@ export class PlaidFetchTransactionsProcessor extends WorkerHost {
         );
       });
       // Notify the frontend to reflect the new transactions changes.
-      // io.emit('NEW_TRANSACTIONS_DATA', { plaidItemId });
+      this.socketGateway.emitNewTransactionsData();
     } catch (error) {
       console.log(error);
     }

packages/server/src/modules/Import/Import.controller.ts

@@ -22,7 +22,7 @@ import { ApiCommonHeaders } from '@/common/decorators/ApiCommonHeaders';
 @ApiTags('Import')
 @ApiCommonHeaders()
 export class ImportController {
-  constructor(private readonly importResourceApp: ImportResourceApplication) {}
+  constructor(private readonly importResourceApp: ImportResourceApplication) { }
 
   /**
    * Imports xlsx/csv to the given resource type.

packages/server/src/modules/Socket/Socket.gateway.ts

@@ -0,0 +1,51 @@
+import {
+  WebSocketGateway,
+  WebSocketServer,
+  OnGatewayConnection,
+  OnGatewayDisconnect,
+  OnGatewayInit,
+} from '@nestjs/websockets';
+import { Server, Socket } from 'socket.io';
+import { Logger } from '@nestjs/common';
+
+@WebSocketGateway({
+  namespace: '/',
+  path: '/socket',
+  cors: {
+    origin: '*',
+    methods: ['GET', 'POST'],
+  },
+})
+export class SocketGateway
+  implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
+  @WebSocketServer()
+  server: Server;
+
+  private logger: Logger = new Logger('SocketGateway');
+
+  afterInit(server: Server) {
+    this.logger.log('Socket.IO Gateway initialized');
+  }
+
+  handleConnection(client: Socket, ...args: any[]) {
+    this.logger.log(`Client connected: ${client.id}`);
+  }
+
+  handleDisconnect(client: Socket) {
+    this.logger.log(`Client disconnected: ${client.id}`);
+  }
+
+  // Method to emit NEW_TRANSACTIONS_DATA event
+  emitNewTransactionsData() {
+    this.server.emit('NEW_TRANSACTIONS_DATA');
+    this.logger.log('Emitted NEW_TRANSACTIONS_DATA event');
+  }
+
+  // Method to emit SUBSCRIPTION_CHANGED event
+  emitSubscriptionChanged() {
+    this.server.emit('SUBSCRIPTION_CHANGED');
+    this.logger.log('Emitted SUBSCRIPTION_CHANGED event');
+  }
+}
+
+

packages/server/src/modules/Socket/Socket.module.ts

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { SocketGateway } from './Socket.gateway';
+
+@Module({
+  providers: [SocketGateway],
+  exports: [SocketGateway],
+})
+export class SocketModule { }
+
+

packages/server/src/modules/Subscription/Subscription.module.ts

@@ -1,4 +1,5 @@
 import { Module } from '@nestjs/common';
+import { SocketModule } from '../Socket/Socket.module';
 import { CancelLemonSubscription } from './commands/CancelLemonSubscription.service';
 import { ChangeLemonSubscription } from './commands/ChangeLemonSubscription.service';
 import { ResumeLemonSubscription } from './commands/ResumeLemonSubscription.service';
@@ -25,6 +26,7 @@ import { PlanSubscriptionRepository } from './repositories/PlanSubscription.repo
 const models = [InjectSystemModel(Plan), InjectSystemModel(PlanSubscription)];
 
 @Module({
+  imports: [SocketModule],
   providers: [
     ...models,
     TenancyContext,
@@ -48,4 +50,4 @@ const models = [InjectSystemModel(Plan), InjectSystemModel(PlanSubscription)];
   controllers: [SubscriptionsController, SubscriptionsLemonWebhook],
   exports: [...models],
 })
-export class SubscriptionModule {}
+export class SubscriptionModule { }

packages/server/src/modules/Subscription/subscribers/TriggerInvalidateCacheOnSubscriptionChange.ts

@@ -1,16 +1,17 @@
 import { events } from '@/common/events/events';
 import { Injectable } from '@nestjs/common';
 import { OnEvent } from '@nestjs/event-emitter';
+import { SocketGateway } from '../../Socket/Socket.gateway';
 
 @Injectable()
 export class TriggerInvalidateCacheOnSubscriptionChange {
+  constructor(private readonly socketGateway: SocketGateway) { }
+
   @OnEvent(events.subscription.onSubscriptionCancelled)
   @OnEvent(events.subscription.onSubscriptionResumed)
   @OnEvent(events.subscription.onSubscriptionPlanChanged)
   triggerInvalidateCache() {
-    // const io = Container.get('socket');
+    // Notify the frontend to reflect the subscription changes.
-
+    this.socketGateway.emitSubscriptionChanged();
-    // // Notify the frontend to reflect the new transactions changes.
-    // io.emit('SUBSCRIPTION_CHANGED', { subscriptionSlug: 'main' });
   }
 }