OUT-3276: backfill script to update product info in mapping table

package.json

@@ -17,7 +17,8 @@
     "prepare": "husky",
     "supabase:dev": "supabase start --ignore-health-check",
     "cmd:rename-qb-accounts": "tsx src/cmd/renameQbAccount/index.ts",
-    "patch-assembly-node-sdk": "cp ./lib-patches/assembly-js-node-sdk.js ./node_modules/@assembly-js/node-sdk/dist/api/init.js"
+    "patch-assembly-node-sdk": "cp ./lib-patches/assembly-js-node-sdk.js ./node_modules/@assembly-js/node-sdk/dist/api/init.js",
+    "cmd:backfill-product-info": "tsx src/cmd/backfillProductInfo/index.ts"
   },
   "dependencies": {
     "@assembly-js/node-sdk": "^3.19.1",

src/app/api/quickbooks/webhook/webhook.service.ts

@@ -8,7 +8,6 @@ import { PaymentService } from '@/app/api/quickbooks/payment/payment.service'
 import { ProductService } from '@/app/api/quickbooks/product/product.service'
 import { SettingService } from '@/app/api/quickbooks/setting/setting.service'
 import { SyncLogService } from '@/app/api/quickbooks/syncLog/syncLog.service'
-import { AccountErrorCodes } from '@/constant/intuitErrorCode'
 import { QBSyncLog } from '@/db/schema/qbSyncLogs'
 import {
   InvoiceCreatedResponseSchema,
@@ -20,7 +19,7 @@ import {
   WebhookEventResponseSchema,
   WebhookEventResponseType,
 } from '@/type/dto/webhook.dto'
-import { refreshTokenExpireMessage, validateAccessToken } from '@/utils/auth'
+import { validateAccessToken } from '@/utils/auth'
 import { CopilotAPI } from '@/utils/copilotAPI'
 import { ErrorMessageAndCode, getMessageAndCodeFromError } from '@/utils/error'
 import { IntuitAPITokensType } from '@/utils/intuitAPI'
@@ -53,7 +52,10 @@ export class WebhookService extends BaseService {
     })
 
     // for webhook event price.create, terminate process if createNewProductFlag is false
-    if (WebhookEvents.PRICE_CREATED === payload.eventType) {
+    if (
+      WebhookEvents.PRICE_CREATED === payload.eventType ||
+      WebhookEvents.PRODUCT_UPDATED === payload.eventType
+    ) {
       const settingService = new SettingService(this.user)
       const setting = await settingService.getOneByPortalId([
         'createNewProductFlag',

src/cmd/backfillProductInfo/backfillProductInfo.service.ts

@@ -0,0 +1,150 @@
+import { MAX_PRODUCT_LIST_LIMIT } from '@/app/api/core/constants/limit'
+import APIError from '@/app/api/core/exceptions/api'
+import { BaseService } from '@/app/api/core/services/base.service'
+import { withRetry } from '@/app/api/core/utils/withRetry'
+import { AuthService } from '@/app/api/quickbooks/auth/auth.service'
+import { ProductService } from '@/app/api/quickbooks/product/product.service'
+import {
+  QBProductSelectSchemaType,
+  QBProductSync,
+} from '@/db/schema/qbProductSync'
+import { StatusableError } from '@/type/CopilotApiError'
+import { CopilotAPI } from '@/utils/copilotAPI'
+import IntuitAPI from '@/utils/intuitAPI'
+import { eq, isNotNull } from 'drizzle-orm'
+import { convert } from 'html-to-text'
+import httpStatus from 'http-status'
+
+export class BackfillProductInfoService extends BaseService {
+  async _backfillProductInfoForPortal() {
+    try {
+      console.info(
+        `BackfillProductInfoService#backfillProductInfoForPortal :: Backfilling product info for portal: ${this.user.workspaceId}`,
+      )
+
+      // 1. get all mapped products from our mapping table
+      const productService = new ProductService(this.user)
+      const mappedProducts: QBProductSelectSchemaType[] =
+        await productService.getAll(isNotNull(QBProductSync.qbItemId))
+      const mappedAssemblyProductIds = [
+        ...new Set(mappedProducts.map((product) => product.productId)),
+      ]
+
+      if (mappedAssemblyProductIds.length === 0) {
+        console.info(
+          `No mapped product found for portal: ${this.user.workspaceId}`,
+        )
+        return
+      }
+
+      // 2. get all products from assembly
+      const copilotApi = new CopilotAPI(this.user.token)
+      const assemblyProducts = (
+        await copilotApi.getProducts(
+          undefined,
+          undefined,
+          MAX_PRODUCT_LIST_LIMIT,
+        )
+      )?.data
+
+      if (!assemblyProducts) {
+        console.info('No product found in assembly')
+        return
+      }
+
+      const filteredAssemblyProducts = assemblyProducts.filter((product) =>
+        mappedAssemblyProductIds.includes(product.id),
+      )
+
+      const authService = new AuthService(this.user)
+      const qbTokenInfo = await authService.getQBPortalConnection(
+        this.user.workspaceId,
+      )
+
+      if (!qbTokenInfo.accessToken || !qbTokenInfo.refreshToken) {
+        console.info(
+          `No access token found for portal: ${this.user.workspaceId}`,
+        )
+        return
+      }
+
+      const intuitApi = new IntuitAPI(qbTokenInfo)
+      const allQbItems = await intuitApi.getAllItems(MAX_PRODUCT_LIST_LIMIT, [
+        'Id',
+        'Name',
+        'UnitPrice',
+        'Description',
+        'SyncToken',
+      ])
+
+      // 3. update the product info in our mapping table
+      for (const mappedProduct of mappedProducts) {
+        if (!mappedProduct.qbItemId) {
+          console.info(`Qb item id not found for product ${mappedProduct.name}`)
+          continue
+        }
+
+        const assemblyProduct = filteredAssemblyProducts.find(
+          (item) => item.id === mappedProduct.productId,
+        )
+
+        if (!assemblyProduct) {
+          console.info(
+            `Copilot product not found for product ${mappedProduct.name} ${mappedProduct.productId}`,
+          )
+          continue
+        }
+
+        // 4. get item from QB
+        const qbItem = allQbItems?.find(
+          (item) => item.Id === mappedProduct.qbItemId,
+        )
+        if (!qbItem) {
+          console.info(
+            `Item not found in Quickbooks for product with assembly ID ${mappedProduct.productId}`,
+          )
+        }
+
+        console.info(
+          `\nUpdating item info in mapping table for product with QB id ${mappedProduct.qbItemId}. Product map id ${mappedProduct.id}`,
+        )
+
+        const payload = {
+          name: qbItem?.Name || null,
+          copilotName: assemblyProduct.name,
+          description: assemblyProduct.description
+            ? convert(assemblyProduct.description)
+            : '',
+          ...(qbItem?.SyncToken && { qbSyncToken: qbItem.SyncToken }),
+        }
+        await productService.updateQBProduct(
+          payload,
+          eq(QBProductSync.id, mappedProduct.id),
+        )
+      }
+    } catch (error: unknown) {
+      if (error instanceof APIError) {
+        throw error
+      }
+      const AssemnblyError = error as StatusableError // no
+      const status = AssemnblyError.status || httpStatus.BAD_REQUEST
+      if (status === httpStatus.FORBIDDEN) {
+        console.info(
+          `Assembly sdk returns forbidden for the portal ${this.user.workspaceId}`,
+        )
+        return
+      }
+      throw error
+    }
+  }
+
+  private wrapWithRetry<Args extends unknown[], R>(
+    fn: (...args: Args) => Promise<R>,
+  ): (...args: Args) => Promise<R> {
+    return (...args: Args): Promise<R> => withRetry(fn.bind(this), args)
+  }
+
+  backfillProductInfoForPortal = this.wrapWithRetry(
+    this._backfillProductInfoForPortal,
+  )
+}

src/cmd/backfillProductInfo/index.ts

@@ -0,0 +1,69 @@
+import APIError from '@/app/api/core/exceptions/api'
+import User from '@/app/api/core/models/User.model'
+import { BackfillProductInfoService } from '@/cmd/backfillProductInfo/backfillProductInfo.service'
+import { copilotAPIKey } from '@/config'
+import { PortalConnectionWithSettingType } from '@/db/schema/qbPortalConnections'
+import { getAllActivePortalConnections } from '@/db/service/token.service'
+import { CopilotAPI } from '@/utils/copilotAPI'
+import { encodePayload } from '@/utils/crypto'
+import CustomLogger from '@/utils/logger'
+
+/**
+ * This script is used to backfill product info in our mapping table
+ */
+
+// command to run the script: `yarn run cmd:backfill-product-info`
+;(async function run() {
+  try {
+    console.info('BackfillProductInfo#initiateProcess')
+    const activeConnections = await getAllActivePortalConnections()
+
+    if (!activeConnections.length) {
+      console.info('No active connection found')
+      return
+    }
+
+    for (const connection of activeConnections) {
+      if (!connection.setting?.syncFlag || !connection.setting?.isEnabled) {
+        console.info(
+          'Skipping connection: ' + JSON.stringify(connection.portalId),
+        )
+        continue
+      }
+
+      console.info(
+        `\n\n\n ########### Processing for PORTAL: ${connection.portalId} #############`,
+      )
+
+      await initiateProcess(connection)
+    }
+
+    console.info('\n Backfilled product info to mapping table successfully 🎉')
+    process.exit(0)
+  } catch (error) {
+    console.error(error)
+    process.exit(1)
+  }
+})()
+
+async function initiateProcess(connection: PortalConnectionWithSettingType) {
+  // generate token for the portal
+  console.info('Generating token for the portal')
+  const payload = {
+    workspaceId: connection.portalId,
+  }
+  const token = encodePayload(copilotAPIKey, payload)
+
+  const copilot = new CopilotAPI(token)
+  const tokenPayload = await copilot.getTokenPayload()
+  CustomLogger.info({
+    obj: { copilotApiCronToken: token, tokenPayload },
+    message:
+      'backfillProductInfo#initiateProcess | Copilot API token and payload',
+  })
+  if (!tokenPayload) throw new APIError(500, 'Encoded token is not valid') // this should trigger p-retry and re-run the function
+
+  const user = new User(token, tokenPayload)
+  const syncMissedService = new BackfillProductInfoService(user)
+  await syncMissedService.backfillProductInfoForPortal()
+}

src/type/dto/intuitAPI.dto.ts

@@ -224,3 +224,28 @@ export const CustomerQueryResponseSchema = z.object({
 export type CustomerQueryResponseType = z.infer<
   typeof CustomerQueryResponseSchema
 >
+
+export const QBInvoiceResponseSchema = z.object({
+  Id: z.string(),
+  Balance: z.number(),
+  PrivateNote: z.string().optional(),
+  SyncToken: z.string(),
+})
+export type QBInvoiceResponseType = z.infer<typeof QBInvoiceResponseSchema>
+
+export const QBPurchaseResponseSchema = z.object({
+  Id: z.string(),
+  TotalAmt: z.number(),
+})
+export type QBPurchaseResponseType = z.infer<typeof QBPurchaseResponseSchema>
+
+export const QBItemsResponseSchema = z.array(
+  z.object({
+    Id: z.string(),
+    Name: z.string(),
+    UnitPrice: z.number(),
+    Description: z.string().nullish(),
+    SyncToken: z.string(),
+  }),
+)
+export type QBItemsResponseType = z.infer<typeof QBItemsResponseSchema>

src/utils/intuitAPI.ts

@@ -25,6 +25,7 @@ import {
   CompanyInfoSchema,
   CustomerQueryResponseType,
   CustomerQueryResponseSchema,
+  QBItemsResponseSchema,
 } from '@/type/dto/intuitAPI.dto'
 import { RetryableError } from '@/utils/error'
 import CustomLogger from '@/utils/logger'
@@ -408,7 +409,7 @@ export default class IntuitAPI {
       )
     }
 
-    return qbItems.Item
+    return QBItemsResponseSchema.parse(qbItems.Item)
   }
 
   async _invoiceSparseUpdate(payload: QBInvoiceSparseUpdatePayloadType) {