feat(server): move updating plaid transactions to background job

packages/server/src/interfaces/CashflowService.ts

@@ -47,7 +47,9 @@ export interface ICashflowCommandDTO {
   branchId?: number;
 }
 
-export interface ICashflowNewCommandDTO extends ICashflowCommandDTO {}
+export interface ICashflowNewCommandDTO extends ICashflowCommandDTO {
+  plaidAccountId?: string;
+}
 
 export interface ICashflowTransaction {
   id?: number;

packages/server/src/interfaces/Plaid.ts

@@ -1,3 +1,15 @@
+export interface IPlaidItemCreatedEventPayload {
+  tenantId: number;
+  plaidAccessToken: string;
+  plaidItemId: string;
+  plaidInstitutionId: string;
+}
+
+export interface PlaidItemDTO {
+  publicToken: string;
+  institutionId: string;
+}
+
 export interface PlaidAccount {
   account_id: string;
   balances: {
@@ -16,9 +28,11 @@ export interface PlaidAccount {
 }
 
 export interface PlaidTransaction {
+  date: string;
   account_id: string;
   amount: number;
-  authorized_data: string;
+  authorized_date: string;
+  name: string;
   category: string[];
   check_number: number | null;
   iso_currency_code: string;

packages/server/src/interfaces/index.ts

@@ -74,6 +74,7 @@ export * from './Tasks';
 export * from './Times';
 export * from './ProjectProfitabilitySummary';
 export * from './TaxRate';
+export * from './Plaid';
 
 export interface I18nService {
   __: (input: string) => string;

packages/server/src/lib/Plaid/Plaid.ts

@@ -23,7 +23,7 @@ const defaultLogger = async (clientMethod, clientMethodArgs, response) => {
   // );
   // await createPlaidApiEvent(1, 1, clientMethod, clientMethodArgs, response);
 
-  console.log(response);
+  // console.log(response);
 };
 
 /**
@@ -39,7 +39,7 @@ const noAccessTokenLogger = async (
   clientMethodArgs,
   response
 ) => {
-  console.log(response);
+  // console.log(response);
 
   // await createPlaidApiEvent(
   //   undefined,

packages/server/src/lib/Plaid/index.ts

@@ -1 +1 @@
-export * from './Plaid';
+export * from './Plaid';

packages/server/src/loaders/eventEmitter.ts

@@ -84,6 +84,7 @@ import { WriteInvoiceTaxTransactionsSubscriber } from '@/services/TaxRates/subsc
 import { BillTaxRateValidateSubscriber } from '@/services/TaxRates/subscribers/BillTaxRateValidateSubscriber';
 import { WriteBillTaxTransactionsSubscriber } from '@/services/TaxRates/subscribers/WriteBillTaxTransactionsSubscriber';
 import { SyncItemTaxRateOnEditTaxSubscriber } from '@/services/TaxRates/SyncItemTaxRateOnEditTaxSubscriber';
+import { PlaidUpdateTransactionsOnItemCreatedSubscriber } from '@/services/Banking/Plaid/subscribers/PlaidUpdateTransactionsOnItemCreatedSubscriber';
 
 export default () => {
   return new EventPublisher();
@@ -199,6 +200,9 @@ export const susbcribers = () => {
     BillTaxRateValidateSubscriber,
     WriteBillTaxTransactionsSubscriber,
 
-    SyncItemTaxRateOnEditTaxSubscriber
+    SyncItemTaxRateOnEditTaxSubscriber,
+
+    // Plaid
+    PlaidUpdateTransactionsOnItemCreatedSubscriber
   ];
 };

packages/server/src/loaders/jobs.ts

@@ -10,6 +10,7 @@ import { SendSaleInvoiceReminderMailJob } from '@/services/Sales/Invoices/SendSa
 import { SendSaleEstimateMailJob } from '@/services/Sales/Estimates/SendSaleEstimateMailJob';
 import { SaleReceiptMailNotificationJob } from '@/services/Sales/Receipts/SaleReceiptMailNotificationJob';
 import { PaymentReceiveMailNotificationJob } from '@/services/Sales/PaymentReceives/PaymentReceiveMailNotificationJob';
+import { PlaidFetchTransactionsJob } from '@/services/Banking/Plaid/PlaidFetchTransactionsJob';
 
 export default ({ agenda }: { agenda: Agenda }) => {
   new ResetPasswordMailJob(agenda);
@@ -23,6 +24,7 @@ export default ({ agenda }: { agenda: Agenda }) => {
   new SendSaleEstimateMailJob(agenda);
   new SaleReceiptMailNotificationJob(agenda);
   new PaymentReceiveMailNotificationJob(agenda);
+  new PlaidFetchTransactionsJob(agenda);
 
   agenda.start();
 };

packages/server/src/services/Banking/Plaid/PlaidApplication.ts

@@ -1,7 +1,7 @@
 import { Inject, Service } from 'typedi';
 import { PlaidLinkTokenService } from './PlaidLinkToken';
 import { PlaidItemService } from './PlaidItem';
-import { PlaidItemDTO } from './_types';
+import { PlaidItemDTO } from '@/interfaces';
 
 @Service()
 export class PlaidApplication {

packages/server/src/services/Banking/Plaid/PlaidFetchTransactionsJob.ts

@@ -1,4 +1,6 @@
 import Container, { Service } from 'typedi';
+import { PlaidUpdateTransactions } from './PlaidUpdateTransactions';
+import { IPlaidItemCreatedEventPayload } from '@/interfaces';
 
 @Service()
 export class PlaidFetchTransactionsJob {
@@ -17,9 +19,17 @@ export class PlaidFetchTransactionsJob {
    * Triggers the function.
    */
   private handler = async (job, done: Function) => {
-    const {} = job.attrs.data;
+    const { tenantId, plaidItemId } = job.attrs
+      .data as IPlaidItemCreatedEventPayload;
 
+    const plaidFetchTransactionsService = Container.get(
+      PlaidUpdateTransactions
+    );
     try {
+      await plaidFetchTransactionsService.updateTransactions(
+        tenantId,
+        plaidItemId
+      );
       done();
     } catch (error) {
       console.log(error);

packages/server/src/services/Banking/Plaid/PlaidItem.ts

@@ -1,8 +1,12 @@
 import { Inject, Service } from 'typedi';
 import { PlaidClientWrapper } from '@/lib/Plaid';
-import { PlaidItemDTO } from './_types';
 import HasTenancyService from '@/services/Tenancy/TenancyService';
-import { PlaidUpdateTransactions } from '@/lib/Plaid/PlaidUpdateTransactions';
+import { EventPublisher } from '@/lib/EventPublisher/EventPublisher';
+import events from '@/subscribers/events';
+import {
+  IPlaidItemCreatedEventPayload,
+  PlaidItemDTO,
+} from '@/interfaces/Plaid';
 
 @Service()
 export class PlaidItemService {
@@ -10,12 +14,14 @@ export class PlaidItemService {
   private tenancy: HasTenancyService;
 
   @Inject()
-  private plaidUpdateTranasctions: PlaidUpdateTransactions;
+  private eventPublisher: EventPublisher;
 
   /**
-   * 
+   * Exchanges the public token to get access token and item id and then creates
-   * @param {number} tenantId 
+   * a new Plaid item.
-   * @param {PlaidItemDTO} itemDTO 
+   * @param {number} tenantId
+   * @param {PlaidItemDTO} itemDTO
+   * @returns {Promise<void>}
    */
   public async item(tenantId: number, itemDTO: PlaidItemDTO) {
     const { PlaidItem } = this.tenancy.models(tenantId);
@@ -36,7 +42,12 @@ export class PlaidItemService {
       plaidItemId,
       plaidInstitutionId: institutionId,
     });
-    
+    // Triggers `onPlaidItemCreated` event.
-    this.plaidUpdateTranasctions.updateTransactions(tenantId, plaidItemId);
+    await this.eventPublisher.emitAsync(events.plaid.onItemCreated, {
+      tenantId,
+      plaidAccessToken,
+      plaidItemId,
+      plaidInstitutionId: institutionId,
+    } as IPlaidItemCreatedEventPayload);
   }
 }

packages/server/src/services/Banking/Plaid/PlaidSyncDB.ts

@@ -1,20 +1,15 @@
 import * as R from 'ramda';
 import { Inject, Service } from 'typedi';
-import async from 'async';
+import bluebird from 'bluebird';
-import { forOwn, groupBy } from 'lodash';
+import { entries, groupBy } from 'lodash';
 import { CreateAccount } from '@/services/Accounts/CreateAccount';
-import {
+import { PlaidAccount, PlaidTransaction } from '@/interfaces';
-  PlaidAccount,
-  PlaidTransaction,
-  SyncAccountsTransactionsTask,
-} from './_types';
 import {
   transformPlaidAccountToCreateAccount,
   transformPlaidTrxsToCashflowCreate,
 } from './utils';
 import NewCashflowTransactionService from '@/services/Cashflow/NewCashflowTransactionService';
 import HasTenancyService from '@/services/Tenancy/TenancyService';
-import { ICashflowNewCommandDTO } from '@/interfaces';
 
 @Service()
 export class PlaidSyncDb {
@@ -31,14 +26,21 @@ export class PlaidSyncDb {
    * Syncs the plaid accounts to the system accounts.
    * @param {number} tenantId Tenant ID.
    * @param {PlaidAccount[]} plaidAccounts
+   * @returns {Promise<void>}
    */
-  public syncBankAccounts(tenantId: number, plaidAccounts: PlaidAccount[]) {
+  public async syncBankAccounts(
+    tenantId: number,
+    plaidAccounts: PlaidAccount[]
+  ): Promise<void> {
     const accountCreateDTOs = R.map(transformPlaidAccountToCreateAccount)(
       plaidAccounts
     );
-    accountCreateDTOs.map((createDTO) => {
+    await bluebird.map(
-      return this.createAccountService.createAccount(tenantId, createDTO);
+      accountCreateDTOs,
-    });
+      (createAccountDTO: any) =>
+        this.createAccountService.createAccount(tenantId, createAccountDTO),
+      { concurrency: 10 }
+    );
   }
 
   /**
@@ -52,10 +54,10 @@ export class PlaidSyncDb {
     plaidAccountId: number,
     plaidTranasctions: PlaidTransaction[]
   ): Promise<void> {
-    const { Account } = await this.tenancy.models(tenantId);
+    const { Account } = this.tenancy.models(tenantId);
 
     const cashflowAccount = await Account.query()
-      .findOne('plaidAccountId', plaidAccountId)
+      .findOne({ plaidAccountId })
       .throwIfNotFound();
 
     const openingEquityBalance = await Account.query().findOne(
@@ -70,18 +72,15 @@ export class PlaidSyncDb {
     const accountsCashflowDTO = R.map(transformTransaction)(plaidTranasctions);
 
     // Creating account transaction queue.
-    const createAccountTransactionsQueue = async.queue(
+    await bluebird.map(
-      (cashflowDTO: ICashflowNewCommandDTO) =>
+      accountsCashflowDTO,
+      (cashflowDTO) =>
         this.createCashflowTransactionService.newCashflowTransaction(
           tenantId,
           cashflowDTO
         ),
-      10
+      { concurrency: 10 }
     );
-    accountsCashflowDTO.forEach((cashflowDTO) => {
-      createAccountTransactionsQueue.push(cashflowDTO);
-    });
-    await createAccountTransactionsQueue.drain();
   }
 
   /**
@@ -93,35 +92,27 @@ export class PlaidSyncDb {
     tenantId: number,
     plaidAccountsTransactions: PlaidTransaction[]
   ): Promise<void> {
-    const groupedTrnsxByAccountId = groupBy(
+    const groupedTrnsxByAccountId = entries(
-      plaidAccountsTransactions,
+      groupBy(plaidAccountsTransactions, 'account_id')
-      'account_id'
     );
-    const syncAccountsTrnsx = async.queue(
+    await bluebird.map(
-      ({
+      groupedTrnsxByAccountId,
-        tenantId,
+      ([plaidAccountId, plaidTransactions]: [number, PlaidTransaction[]]) => {
-        plaidAccountId,
-        plaidTransactions,
-      }: SyncAccountsTransactionsTask) => {
         return this.syncAccountTranactions(
           tenantId,
           plaidAccountId,
           plaidTransactions
         );
       },
-      2
+      { concurrency: 10 }
     );
-    forOwn(groupedTrnsxByAccountId, (plaidTransactions, plaidAccountId) => {
-      syncAccountsTrnsx.push({ tenantId, plaidAccountId, plaidTransactions });
-    });
-    await syncAccountsTrnsx.drain();
   }
 
   /**
    * Syncs the Plaid item last transaction cursor.
-   * @param {number} tenantId -
+   * @param {number} tenantId - Tenant ID.
-   * @param {string} itemId -
+   * @param {string} itemId - Plaid item ID.
-   * @param {string} lastCursor -
+   * @param {string} lastCursor - Last transaction cursor.
    */
   public async syncTransactionsCursor(
     tenantId: number,

packages/server/src/services/Banking/Plaid/PlaidUpdateTransactions.ts

@@ -1,8 +1,8 @@
 import HasTenancyService from '@/services/Tenancy/TenancyService';
 import { Inject, Service } from 'typedi';
-import { PlaidClientWrapper } from './Plaid';
+import { PlaidClientWrapper } from '@/lib/Plaid/Plaid';
 import { PlaidSyncDb } from './PlaidSyncDB';
-import { PlaidFetchedTransactionsUpdates } from './_types';
+import { PlaidFetchedTransactionsUpdates } from '@/interfaces';
 
 @Service()
 export class PlaidUpdateTransactions {
@@ -49,7 +49,7 @@ export class PlaidUpdateTransactions {
    * @param {string} plaidItemId - The Plaid ID for the item.
    * @returns {Promise<PlaidFetchedTransactionsUpdates>}
    */
-  public async fetchTransactionUpdates(
+  private async fetchTransactionUpdates(
     tenantId: number,
     plaidItemId: string
   ): Promise<PlaidFetchedTransactionsUpdates> {

packages/server/src/services/Banking/Plaid/_types.ts

@@ -1,4 +0,0 @@
-export interface PlaidItemDTO {
-  publicToken: string;
-  institutionId: string;
-}

packages/server/src/services/Banking/Plaid/subscribers/PlaidUpdateTransactionsOnItemCreatedSubscriber.ts

@@ -0,0 +1,34 @@
+import { Inject, Service } from 'typedi';
+import { EventSubscriber } from '@/lib/EventPublisher/EventPublisher';
+import { IPlaidItemCreatedEventPayload } from '@/interfaces/Plaid';
+import events from '@/subscribers/events';
+
+@Service()
+export class PlaidUpdateTransactionsOnItemCreatedSubscriber extends EventSubscriber {
+  @Inject('agenda')
+  private agenda: any;
+
+  /**
+   * Constructor method.
+   */
+  public attach(bus) {
+    bus.subscribe(
+      events.plaid.onItemCreated,
+      this.handleUpdateTransactionsOnItemCreated
+    );
+  }
+
+  /**
+   * Updates the Plaid item transactions
+   * @param {IPlaidItemCreatedEventPayload} payload - Event payload.
+   */
+  private handleUpdateTransactionsOnItemCreated = async ({
+    tenantId,
+    plaidItemId,
+    plaidAccessToken,
+    plaidInstitutionId,
+  }: IPlaidItemCreatedEventPayload) => {
+    const payload = { tenantId, plaidItemId };
+    await this.agenda.now('plaid-update-account-transactions', payload);
+  };
+}

packages/server/src/services/Banking/Plaid/utils.ts

@@ -2,6 +2,11 @@ import * as R from 'ramda';
 import { IAccountCreateDTO, ICashflowNewCommandDTO } from '@/interfaces';
 import { PlaidAccount, PlaidTransaction } from './_types';
 
+/**
+ * Transformes the Plaid account to create cashflow account DTO.
+ * @param {PlaidAccount} plaidAccount
+ * @returns {IAccountCreateDTO}
+ */
 export const transformPlaidAccountToCreateAccount = (
   plaidAccount: PlaidAccount
 ): IAccountCreateDTO => {
@@ -16,17 +21,24 @@ export const transformPlaidAccountToCreateAccount = (
   };
 };
 
+/**
+ * Transformes the plaid transaction to cashflow create DTO.
+ * @param {number} cashflowAccountId - Cashflow account ID.
+ * @param {number} creditAccountId - Credit account ID.
+ * @param {PlaidTransaction} plaidTranasction - Plaid transaction.
+ * @returns {ICashflowNewCommandDTO}
+ */
 export const transformPlaidTrxsToCashflowCreate = R.curry(
   (
     cashflowAccountId: number,
     creditAccountId: number,
-    plaidTranasction: PlaidTransaction,
+    plaidTranasction: PlaidTransaction
   ): ICashflowNewCommandDTO => {
     return {
-      date: plaidTranasction.authorized_data,
+      date: plaidTranasction.date,
 
-      transactionType: '',
+      transactionType: 'OwnerContribution',
-      description: '',
+      description: plaidTranasction.name,
 
       amount: plaidTranasction.amount,
       exchangeRate: 1,
@@ -36,6 +48,7 @@ export const transformPlaidTrxsToCashflowCreate = R.curry(
 
       // transactionNumber: string;
       // referenceNo: string;
+      publish: true,
     };
   }
 );

packages/server/src/subscribers/events.ts

@@ -131,7 +131,7 @@ export default {
     onNotifiedSms: 'onSaleInvoiceNotifiedSms',
 
     onNotifyMail: 'onSaleInvoiceNotifyMail',
-    onNotifyReminderMail: 'onSaleInvoiceNotifyReminderMail'
+    onNotifyReminderMail: 'onSaleInvoiceNotifyReminderMail',
   },
 
   /**
@@ -164,7 +164,7 @@ export default {
     onRejecting: 'onSaleEstimateRejecting',
     onRejected: 'onSaleEstimateRejected',
 
-    onNotifyMail: 'onSaleEstimateNotifyMail'
+    onNotifyMail: 'onSaleEstimateNotifyMail',
   },
 
   /**
@@ -580,6 +580,10 @@ export default {
     onActivated: 'onTaxRateActivated',
 
     onInactivating: 'onTaxRateInactivating',
-    onInactivated: 'onTaxRateInactivated'
+    onInactivated: 'onTaxRateInactivated',
+  },
+
+  plaid: {
+    onItemCreated: 'onPlaidItemCreated',
   },
 };