diff --git a/packages/server/src/interfaces/CashflowService.ts b/packages/server/src/interfaces/CashflowService.ts index 8b0576d8e..9df08a9ad 100644 --- a/packages/server/src/interfaces/CashflowService.ts +++ b/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; diff --git a/packages/server/src/lib/Plaid/_types.ts b/packages/server/src/interfaces/Plaid.ts similarity index 73% rename from packages/server/src/lib/Plaid/_types.ts rename to packages/server/src/interfaces/Plaid.ts index 097ffc512..9b19e37a9 100644 --- a/packages/server/src/lib/Plaid/_types.ts +++ b/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; diff --git a/packages/server/src/interfaces/index.ts b/packages/server/src/interfaces/index.ts index 1b23eedd3..858acba51 100644 --- a/packages/server/src/interfaces/index.ts +++ b/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; diff --git a/packages/server/src/lib/Plaid/Plaid.ts b/packages/server/src/lib/Plaid/Plaid.ts index ccc2a7f10..d067757fb 100644 --- a/packages/server/src/lib/Plaid/Plaid.ts +++ b/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, diff --git a/packages/server/src/lib/Plaid/index.ts b/packages/server/src/lib/Plaid/index.ts index 4a580954e..9f3f903ca 100644 --- a/packages/server/src/lib/Plaid/index.ts +++ b/packages/server/src/lib/Plaid/index.ts @@ -1 +1 @@ -export * from './Plaid'; \ No newline at end of file +export * from './Plaid'; diff --git a/packages/server/src/loaders/eventEmitter.ts b/packages/server/src/loaders/eventEmitter.ts index fa1942b72..6502c044c 100644 --- a/packages/server/src/loaders/eventEmitter.ts +++ b/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 ]; }; diff --git a/packages/server/src/loaders/jobs.ts b/packages/server/src/loaders/jobs.ts index 74c7d6b6e..3215c9558 100644 --- a/packages/server/src/loaders/jobs.ts +++ b/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(); }; diff --git a/packages/server/src/services/Banking/Plaid/PlaidApplication.ts b/packages/server/src/services/Banking/Plaid/PlaidApplication.ts index f641963f7..1789eafdd 100644 --- a/packages/server/src/services/Banking/Plaid/PlaidApplication.ts +++ b/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 { diff --git a/packages/server/src/lib/Plaid/PlaidFetchTransactionsJob.ts b/packages/server/src/services/Banking/Plaid/PlaidFetchTransactionsJob.ts similarity index 53% rename from packages/server/src/lib/Plaid/PlaidFetchTransactionsJob.ts rename to packages/server/src/services/Banking/Plaid/PlaidFetchTransactionsJob.ts index 77f6ea597..6ff699fd7 100644 --- a/packages/server/src/lib/Plaid/PlaidFetchTransactionsJob.ts +++ b/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); diff --git a/packages/server/src/services/Banking/Plaid/PlaidItem.ts b/packages/server/src/services/Banking/Plaid/PlaidItem.ts index 1c63c1ee4..ddf434aae 100644 --- a/packages/server/src/services/Banking/Plaid/PlaidItem.ts +++ b/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; /** - * - * @param {number} tenantId - * @param {PlaidItemDTO} itemDTO + * Exchanges the public token to get access token and item id and then creates + * a new Plaid item. + * @param {number} tenantId + * @param {PlaidItemDTO} itemDTO + * @returns {Promise} */ public async item(tenantId: number, itemDTO: PlaidItemDTO) { const { PlaidItem } = this.tenancy.models(tenantId); @@ -36,7 +42,12 @@ export class PlaidItemService { plaidItemId, plaidInstitutionId: institutionId, }); - - this.plaidUpdateTranasctions.updateTransactions(tenantId, plaidItemId); + // Triggers `onPlaidItemCreated` event. + await this.eventPublisher.emitAsync(events.plaid.onItemCreated, { + tenantId, + plaidAccessToken, + plaidItemId, + plaidInstitutionId: institutionId, + } as IPlaidItemCreatedEventPayload); } } diff --git a/packages/server/src/lib/Plaid/PlaidSyncDB.ts b/packages/server/src/services/Banking/Plaid/PlaidSyncDB.ts similarity index 66% rename from packages/server/src/lib/Plaid/PlaidSyncDB.ts rename to packages/server/src/services/Banking/Plaid/PlaidSyncDB.ts index 62a978089..c104a4bf0 100644 --- a/packages/server/src/lib/Plaid/PlaidSyncDB.ts +++ b/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 { forOwn, groupBy } from 'lodash'; +import bluebird from 'bluebird'; +import { entries, groupBy } from 'lodash'; import { CreateAccount } from '@/services/Accounts/CreateAccount'; -import { - PlaidAccount, - PlaidTransaction, - SyncAccountsTransactionsTask, -} from './_types'; +import { PlaidAccount, PlaidTransaction } from '@/interfaces'; 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} */ - public syncBankAccounts(tenantId: number, plaidAccounts: PlaidAccount[]) { + public async syncBankAccounts( + tenantId: number, + plaidAccounts: PlaidAccount[] + ): Promise { const accountCreateDTOs = R.map(transformPlaidAccountToCreateAccount)( plaidAccounts ); - accountCreateDTOs.map((createDTO) => { - return this.createAccountService.createAccount(tenantId, createDTO); - }); + await bluebird.map( + accountCreateDTOs, + (createAccountDTO: any) => + this.createAccountService.createAccount(tenantId, createAccountDTO), + { concurrency: 10 } + ); } /** @@ -52,10 +54,10 @@ export class PlaidSyncDb { plaidAccountId: number, plaidTranasctions: PlaidTransaction[] ): Promise { - 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( - (cashflowDTO: ICashflowNewCommandDTO) => + await bluebird.map( + 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 { - const groupedTrnsxByAccountId = groupBy( - plaidAccountsTransactions, - 'account_id' + const groupedTrnsxByAccountId = entries( + groupBy(plaidAccountsTransactions, 'account_id') ); - const syncAccountsTrnsx = async.queue( - ({ - tenantId, - plaidAccountId, - plaidTransactions, - }: SyncAccountsTransactionsTask) => { + await bluebird.map( + groupedTrnsxByAccountId, + ([plaidAccountId, plaidTransactions]: [number, PlaidTransaction[]]) => { 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 {string} itemId - - * @param {string} lastCursor - + * @param {number} tenantId - Tenant ID. + * @param {string} itemId - Plaid item ID. + * @param {string} lastCursor - Last transaction cursor. */ public async syncTransactionsCursor( tenantId: number, diff --git a/packages/server/src/lib/Plaid/PlaidUpdateTransactions.ts b/packages/server/src/services/Banking/Plaid/PlaidUpdateTransactions.ts similarity index 95% rename from packages/server/src/lib/Plaid/PlaidUpdateTransactions.ts rename to packages/server/src/services/Banking/Plaid/PlaidUpdateTransactions.ts index 346d4fa13..d82bda770 100644 --- a/packages/server/src/lib/Plaid/PlaidUpdateTransactions.ts +++ b/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} */ - public async fetchTransactionUpdates( + private async fetchTransactionUpdates( tenantId: number, plaidItemId: string ): Promise { diff --git a/packages/server/src/services/Banking/Plaid/_types.ts b/packages/server/src/services/Banking/Plaid/_types.ts deleted file mode 100644 index d944618d0..000000000 --- a/packages/server/src/services/Banking/Plaid/_types.ts +++ /dev/null @@ -1,4 +0,0 @@ -export interface PlaidItemDTO { - publicToken: string; - institutionId: string; -} diff --git a/packages/server/src/services/Banking/Plaid/subscribers/PlaidUpdateTransactionsOnItemCreatedSubscriber.ts b/packages/server/src/services/Banking/Plaid/subscribers/PlaidUpdateTransactionsOnItemCreatedSubscriber.ts new file mode 100644 index 000000000..216eae2f2 --- /dev/null +++ b/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); + }; +} diff --git a/packages/server/src/lib/Plaid/utils.ts b/packages/server/src/services/Banking/Plaid/utils.ts similarity index 60% rename from packages/server/src/lib/Plaid/utils.ts rename to packages/server/src/services/Banking/Plaid/utils.ts index 0517f0446..c8b97d7b7 100644 --- a/packages/server/src/lib/Plaid/utils.ts +++ b/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: '', - description: '', + transactionType: 'OwnerContribution', + description: plaidTranasction.name, amount: plaidTranasction.amount, exchangeRate: 1, @@ -36,6 +48,7 @@ export const transformPlaidTrxsToCashflowCreate = R.curry( // transactionNumber: string; // referenceNo: string; + publish: true, }; } ); diff --git a/packages/server/src/subscribers/events.ts b/packages/server/src/subscribers/events.ts index e54f48152..b42cb09bf 100644 --- a/packages/server/src/subscribers/events.ts +++ b/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', }, };