Merge pull request #563 from bigcapitalhq/pause-resume-bank-feeds-syncing

feat: pause/resume bank account feeds syncing

packages/server/src/api/controllers/Banking/BankAccountsController.ts

@@ -1,9 +1,9 @@
 import { Inject, Service } from 'typedi';
 import { NextFunction, Request, Response, Router } from 'express';
 import BaseController from '@/api/controllers/BaseController';
-import { CashflowApplication } from '@/services/Cashflow/CashflowApplication';
 import { GetBankAccountSummary } from '@/services/Banking/BankAccounts/GetBankAccountSummary';
 import { BankAccountsApplication } from '@/services/Banking/BankAccounts/BankAccountsApplication';
+import { param } from 'express-validator';
 
 @Service()
 export class BankAccountsController extends BaseController {
@@ -25,6 +25,22 @@ export class BankAccountsController extends BaseController {
       this.disconnectBankAccount.bind(this)
     );
     router.post('/:bankAccountId/update', this.refreshBankAccount.bind(this));
+    router.post(
+      '/:bankAccountId/pause_feeds',
+      [
+        param('bankAccountId').exists().isNumeric().toInt(),
+      ],
+      this.validationResult,
+      this.pauseBankAccountFeeds.bind(this)
+    );
+    router.post(
+      '/:bankAccountId/resume_feeds',
+      [
+        param('bankAccountId').exists().isNumeric().toInt(),
+      ],
+      this.validationResult,
+      this.resumeBankAccountFeeds.bind(this)
+    );
 
     return router;
   }
@@ -109,4 +125,58 @@ export class BankAccountsController extends BaseController {
       next(error);
     }
   }
+
+  /**
+   * Resumes the bank account feeds sync.
+   * @param {Request} req 
+   * @param {Response} res 
+   * @param {NextFunction} next 
+   * @returns {Promise<Response | void>}
+   */
+  async resumeBankAccountFeeds(
+    req: Request<{ bankAccountId: number }>,
+    res: Response,
+    next: NextFunction
+  ) {
+    const { bankAccountId } = req.params;
+    const { tenantId } = req;
+
+    try {
+      await this.bankAccountsApp.resumeBankAccount(tenantId, bankAccountId);
+
+      return res.status(200).send({
+        message: 'The bank account feeds syncing has been resumed.',
+        id: bankAccountId,
+      });
+    } catch (error) {
+      next(error);
+    }
+  }
+
+  /**
+   * Pauses the bank account feeds sync.
+   * @param {Request} req 
+   * @param {Response} res 
+   * @param {NextFunction} next 
+   * @returns {Promise<Response | void>}
+   */
+  async pauseBankAccountFeeds(
+    req: Request<{ bankAccountId: number }>,
+    res: Response,
+    next: NextFunction
+  ) {
+    const { bankAccountId } = req.params;
+    const { tenantId } = req;
+
+    try {
+      await this.bankAccountsApp.pauseBankAccount(tenantId, bankAccountId);
+
+      return res.status(200).send({
+        message: 'The bank account feeds syncing has been paused.',
+        id: bankAccountId,
+      });
+    } catch (error) {
+      next(error);
+    }
+  }
 }

packages/server/src/database/migrations/20240804084709_create_paused_at_column_to_plaid_items_table.js

@@ -0,0 +1,11 @@
+exports.up = function (knex) {
+  return knex.schema.table('plaid_items', (table) => {
+    table.datetime('paused_at');
+  });
+};
+
+exports.down = function (knex) {
+  return knex.schema.table('plaid_items', (table) => {
+    table.dropColumn('paused_at');
+  });
+};

packages/server/src/models/PlaidItem.ts

@@ -1,6 +1,8 @@
 import TenantModel from 'models/TenantModel';
 
 export default class PlaidItem extends TenantModel {
+  pausedAt: Date;
+
   /**
    * Table name.
    */
@@ -21,4 +23,19 @@ export default class PlaidItem extends TenantModel {
   static get relationMappings() {
     return {};
   }
+
+  /**
+   * Virtual attributes.
+   */
+  static get virtualAttributes() {
+    return ['isPaused'];
+  }
+
+  /**
+   * Detarmines whether the Plaid item feeds syncing is paused.
+   * @return {boolean}
+   */
+  get isPaused() {
+    return !!this.pausedAt;
+  }
 }

packages/server/src/services/Accounts/AccountTransform.ts

@@ -18,9 +18,18 @@ export class AccountTransformer extends Transformer {
       'flattenName',
       'bankBalanceFormatted',
       'lastFeedsUpdatedAtFormatted',
+      'isFeedsPaused',
     ];
   };
 
+  /**
+   * Exclude attributes.
+   * @returns {string[]}
+   */
+  public excludeAttributes = (): string[] => {
+    return ['plaidItem'];
+  };
+
   /**
    * Retrieves the flatten name with all dependants accounts names.
    * @param {IAccount} account -
@@ -66,6 +75,15 @@ export class AccountTransformer extends Transformer {
     return this.formatDate(account.lastFeedsUpdatedAt);
   };
 
+  /**
+   * Detarmines whether the bank account connection is paused.
+   * @param account
+   * @returns {boolean}
+   */
+  protected isFeedsPaused = (account: any): boolean => {
+    return account.plaidItem?.isPaused || false;
+  };
+
   /**
    * Transformes the accounts collection to flat or nested array.
    * @param {IAccount[]}

packages/server/src/services/Accounts/GetAccount.ts

@@ -25,7 +25,10 @@ export class GetAccount {
     const { accountRepository } = this.tenancy.repositories(tenantId);
 
     // Find the given account or throw not found error.
-    const account = await Account.query().findById(accountId).throwIfNotFound();
+    const account = await Account.query()
+      .findById(accountId)
+      .withGraphFetched('plaidItem')
+      .throwIfNotFound();
 
     const accountsGraph = await accountRepository.getDependencyGraph();
 

packages/server/src/services/Banking/BankAccounts/BankAccountsApplication.tsx

@@ -1,6 +1,8 @@
 import { Inject, Service } from 'typedi';
 import { DisconnectBankAccount } from './DisconnectBankAccount';
 import { RefreshBankAccountService } from './RefreshBankAccount';
+import { PauseBankAccountFeeds } from './PauseBankAccountFeeds';
+import { ResumeBankAccountFeeds } from './ResumeBankAccountFeeds';
 
 @Service()
 export class BankAccountsApplication {
@@ -10,6 +12,12 @@ export class BankAccountsApplication {
   @Inject()
   private refreshBankAccountService: RefreshBankAccountService;
 
+  @Inject()
+  private resumeBankAccountFeedsService: ResumeBankAccountFeeds;
+
+  @Inject()
+  private pauseBankAccountFeedsService: PauseBankAccountFeeds;
+
   /**
    * Disconnects the given bank account.
    * @param {number} tenantId
@@ -27,7 +35,7 @@ export class BankAccountsApplication {
    * Refresh the bank transactions of the given bank account.
    * @param {number} tenantId
    * @param {number} bankAccountId
-   * @returns {Promise<void>} 
+   * @returns {Promise<void>}
    */
   async refreshBankAccount(tenantId: number, bankAccountId: number) {
     return this.refreshBankAccountService.refreshBankAccount(
@@ -35,4 +43,30 @@ export class BankAccountsApplication {
       bankAccountId
     );
   }
+
+  /**
+   * Pauses the feeds sync of the given bank account.
+   * @param {number} tenantId
+   * @param {number} bankAccountId
+   * @returns {Promise<void>}
+   */
+  async pauseBankAccount(tenantId: number, bankAccountId: number) {
+    return this.pauseBankAccountFeedsService.pauseBankAccountFeeds(
+      tenantId,
+      bankAccountId
+    );
+  }
+
+  /**
+   * Resumes the feeds sync of the given bank account.
+   * @param {number} tenantId
+   * @param {number} bankAccountId
+   * @returns {Promise<void>}
+   */
+  async resumeBankAccount(tenantId: number, bankAccountId: number) {
+    return this.resumeBankAccountFeedsService.resumeBankAccountFeeds(
+      tenantId,
+      bankAccountId
+    );
+  }
 }

packages/server/src/services/Banking/BankAccounts/PauseBankAccountFeeds.tsx

@@ -0,0 +1,44 @@
+import { Inject, Service } from 'typedi';
+import { Knex } from 'knex';
+import HasTenancyService from '@/services/Tenancy/TenancyService';
+import UnitOfWork from '@/services/UnitOfWork';
+import { ServiceError } from '@/exceptions';
+import { ERRORS } from './types';
+
+@Service()
+export class PauseBankAccountFeeds {
+  @Inject()
+  private tenancy: HasTenancyService;
+
+  @Inject()
+  private uow: UnitOfWork;
+
+  /**
+   * Pauses the bankfeed syncing of the given bank account.
+   * @param {number} tenantId
+   * @param {number} bankAccountId
+   * @returns {Promise<void>}
+   */
+  public async pauseBankAccountFeeds(tenantId: number, bankAccountId: number) {
+    const { Account, PlaidItem } = this.tenancy.models(tenantId);
+
+    const oldAccount = await Account.query()
+      .findById(bankAccountId)
+      .withGraphFetched('plaidItem')
+      .throwIfNotFound();
+
+    // Can't continue if the bank account is not connected.
+    if (!oldAccount.plaidItem) {
+      throw new ServiceError(ERRORS.BANK_ACCOUNT_NOT_CONNECTED);
+    }
+    // Cannot continue if the bank account feeds is already paused.
+    if (oldAccount.plaidItem.isPaused) {
+      throw new ServiceError(ERRORS.BANK_ACCOUNT_FEEDS_ALREADY_PAUSED);
+    }
+    return this.uow.withTransaction(tenantId, async (trx: Knex.Transaction) => {
+      await PlaidItem.query(trx).findById(oldAccount.plaidItem.id).patch({
+        pausedAt: new Date(),
+      });
+    });
+  }
+}

packages/server/src/services/Banking/BankAccounts/ResumeBankAccountFeeds.tsx

@@ -0,0 +1,43 @@
+import { Inject, Service } from 'typedi';
+import { Knex } from 'knex';
+import HasTenancyService from '@/services/Tenancy/TenancyService';
+import UnitOfWork from '@/services/UnitOfWork';
+import { ServiceError } from '@/exceptions';
+import { ERRORS } from './types';
+
+@Service()
+export class ResumeBankAccountFeeds {
+  @Inject()
+  private tenancy: HasTenancyService;
+
+  @Inject()
+  private uow: UnitOfWork;
+
+  /**
+   * Resumes the bank feeds syncing of the bank account.
+   * @param {number} tenantId
+   * @param {number} bankAccountId
+   * @returns {Promise<void>}
+   */
+  public async resumeBankAccountFeeds(tenantId: number, bankAccountId: number) {
+    const { Account, PlaidItem } = this.tenancy.models(tenantId);
+
+    const oldAccount = await Account.query()
+      .findById(bankAccountId)
+      .withGraphFetched('plaidItem');
+
+    // Can't continue if the bank account is not connected.
+    if (!oldAccount.plaidItem) {
+      throw new ServiceError(ERRORS.BANK_ACCOUNT_NOT_CONNECTED);
+    }
+    // Cannot continue if the bank account feeds is already paused.
+    if (!oldAccount.plaidItem.isPaused) {
+      throw new ServiceError(ERRORS.BANK_ACCOUNT_FEEDS_ALREADY_RESUMED);
+    }
+    return this.uow.withTransaction(tenantId, async (trx: Knex.Transaction) => {
+      await PlaidItem.query(trx).findById(oldAccount.plaidItem.id).patch({
+        pausedAt: null,
+      });
+    });
+  }
+}

packages/server/src/services/Banking/BankAccounts/types.ts

@@ -14,4 +14,6 @@ export interface IBankAccountDisconnectedEventPayload {
 
 export const ERRORS = {
   BANK_ACCOUNT_NOT_CONNECTED: 'BANK_ACCOUNT_NOT_CONNECTED',
+  BANK_ACCOUNT_FEEDS_ALREADY_PAUSED: 'BANK_ACCOUNT_FEEDS_ALREADY_PAUSED',
+  BANK_ACCOUNT_FEEDS_ALREADY_RESUMED: 'BANK_ACCOUNT_FEEDS_ALREADY_RESUMED',
 };

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

@@ -1,11 +1,15 @@
 import { Inject, Service } from 'typedi';
 import { PlaidUpdateTransactions } from './PlaidUpdateTransactions';
+import HasTenancyService from '@/services/Tenancy/TenancyService';
 
 @Service()
 export class PlaidWebooks {
   @Inject()
   private updateTransactionsService: PlaidUpdateTransactions;
 
+  @Inject()
+  private tenancy: HasTenancyService;
+
   /**
    * Listens to Plaid webhooks
    * @param {number} tenantId - Tenant Id.
@@ -61,7 +65,7 @@ export class PlaidWebooks {
     plaidItemId: string
   ): void {
     console.log(
-      `WEBHOOK: TRANSACTIONS: ${webhookCode}: Plaid_item_id ${plaidItemId}: ${additionalInfo}`
+      `PLAID WEBHOOK: TRANSACTIONS: ${webhookCode}: Plaid_item_id ${plaidItemId}: ${additionalInfo}`
     );
   }
 
@@ -78,8 +82,21 @@ export class PlaidWebooks {
     plaidItemId: string,
     webhookCode: string
   ): Promise<void> {
+    const { PlaidItem } = this.tenancy.models(tenantId);
+    const plaidItem = await PlaidItem.query()
+      .findById(plaidItemId)
+      .throwIfNotFound();
+
     switch (webhookCode) {
       case 'SYNC_UPDATES_AVAILABLE': {
+        if (plaidItem.isPaused) {
+          this.serverLogAndEmitSocket(
+            'Plaid item syncing is paused.',
+            webhookCode,
+            plaidItemId
+          );
+          return;
+        }
         // Fired when new transactions data becomes available.
         const { addedCount, modifiedCount, removedCount } =
           await this.updateTransactionsService.updateTransactions(