refactor(nestjs): plaid banking syncing

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

@@ -1,43 +1,46 @@
-// import Container, { Service } from 'typedi';
-// import { PlaidUpdateTransactions } from './PlaidUpdateTransactions';
-// import { IPlaidItemCreatedEventPayload } from '@/interfaces';
+import { Process } from '@nestjs/bull';
+import { UseCls } from 'nestjs-cls';
+import { Processor, WorkerHost } from '@nestjs/bullmq';
+import { Scope } from '@nestjs/common';
+import { Job } from 'bullmq';
+import {
+  PlaidFetchTransitonsEventPayload,
+  UpdateBankingPlaidTransitionsJob,
+  UpdateBankingPlaidTransitionsQueueJob,
+} from '../types/BankingPlaid.types';
+import { PlaidUpdateTransactions } from '../command/PlaidUpdateTransactions';
+import { SetupPlaidItemTenantService } from '../command/SetupPlaidItemTenant.service';
 
-// @Service()
-// export class PlaidFetchTransactionsJob {
-//   /**
-//    * Constructor method.
-//    */
-//   constructor(agenda) {
-//     agenda.define(
-//       'plaid-update-account-transactions',
-//       { priority: 'high', concurrency: 2 },
-//       this.handler
-//     );
-//   }
+@Processor({
+  name: UpdateBankingPlaidTransitionsQueueJob,
+  scope: Scope.REQUEST,
+})
+export class PlaidFetchTransactionsProcessor extends WorkerHost {
+  constructor(
+    private readonly plaidFetchTransactionsService: PlaidUpdateTransactions,
+    private readonly setupPlaidItemService: SetupPlaidItemTenantService,
+  ) {
+    super();
+  }
 
-//   /**
-//    * Triggers the function.
-//    */
-//   private handler = async (job, done: Function) => {
-//     const { tenantId, plaidItemId } = job.attrs
-//       .data as IPlaidItemCreatedEventPayload;
+  /**
+   * Triggers the function.
+   */
+  @Process(UpdateBankingPlaidTransitionsJob)
+  @UseCls()
+  async process(job: Job<PlaidFetchTransitonsEventPayload>) {
+    const { plaidItemId } = job.data;
 
-//     const plaidFetchTransactionsService = Container.get(
-//       PlaidUpdateTransactions
-//     );
-//     const io = Container.get('socket');
-
-//     try {
-//       await plaidFetchTransactionsService.updateTransactions(
-//         tenantId,
-//         plaidItemId
-//       );
-//       // Notify the frontend to reflect the new transactions changes.
-//       io.emit('NEW_TRANSACTIONS_DATA', { plaidItemId });
-//       done();
-//     } catch (error) {
-//       console.log(error);
-//       done(error);
-//     }
-//   };
-// }
+    try {
+      await this.setupPlaidItemService.setupPlaidTenant(plaidItemId, () => {
+        return this.plaidFetchTransactionsService.updateTransactions(
+          plaidItemId,
+        );
+      });
+      // Notify the frontend to reflect the new transactions changes.
+      // io.emit('NEW_TRANSACTIONS_DATA', { plaidItemId });
+    } catch (error) {
+      console.log(error);
+    }
+  }
+}