Chore/Add telemetry for productId (#5188)

add telemetry for productId

Author: HenryHengZJ

packages/server/src/Interface.ts

@@ -395,6 +395,7 @@ export interface IExecuteFlowParams extends IPredictionQueueAppServer {
     orgId: string
     workspaceId: string
     subscriptionId: string
+    productId: string
     baseURL: string
     isInternal: boolean
     isEvaluation?: boolean

packages/server/src/enterprise/services/user.service.ts

@@ -97,7 +97,18 @@ export class UserService {
             data.updatedBy = data.id
         }
 
-        return queryRunner.manager.create(User, data)
+        const userObj = queryRunner.manager.create(User, data)
+
+        this.telemetry.sendTelemetry(
+            TelemetryEventType.USER_CREATED,
+            {
+                userId: userObj.id,
+                createdBy: userObj.createdBy
+            },
+            userObj.id
+        )
+
+        return userObj
     }
 
     public async saveUser(data: Partial<User>, queryRunner: QueryRunner) {
@@ -120,15 +131,6 @@ export class UserService {
             await queryRunner.release()
         }
 
-        this.telemetry.sendTelemetry(
-            TelemetryEventType.USER_CREATED,
-            {
-                userId: newUser.id,
-                createdBy: newUser.createdBy
-            },
-            newUser.id
-        )
-
         return newUser
     }
 

packages/server/src/services/chatflows/index.ts

@@ -292,12 +292,18 @@ const saveChatflow = async (
         const chatflow = appServer.AppDataSource.getRepository(ChatFlow).create(newChatFlow)
         dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(chatflow)
     }
+
+    const subscriptionDetails = await usageCacheManager.getSubscriptionDataFromCache(subscriptionId)
+    const productId = subscriptionDetails?.productId || ''
+
     await appServer.telemetry.sendTelemetry(
         'chatflow_created',
         {
             version: await getAppVersion(),
             chatflowId: dbResponse.id,
-            flowGraph: getTelemetryFlowObj(JSON.parse(dbResponse.flowData)?.nodes, JSON.parse(dbResponse.flowData)?.edges)
+            flowGraph: getTelemetryFlowObj(JSON.parse(dbResponse.flowData)?.nodes, JSON.parse(dbResponse.flowData)?.edges),
+            productId,
+            subscriptionId
         },
         orgId
     )

packages/server/src/utils/buildAgentflow.ts

@@ -133,6 +133,7 @@ interface IExecuteNodeParams {
     orgId: string
     workspaceId: string
     subscriptionId: string
+    productId: string
 }
 
 interface IExecuteAgentFlowParams extends Omit<IExecuteFlowParams, 'incomingInput'> {
@@ -838,7 +839,8 @@ const executeNode = async ({
     iterationContext,
     orgId,
     workspaceId,
-    subscriptionId
+    subscriptionId,
+    productId
 }: IExecuteNodeParams): Promise<{
     result: any
     shouldStop?: boolean
@@ -1060,7 +1062,8 @@ const executeNode = async ({
                             },
                             orgId,
                             workspaceId,
-                            subscriptionId
+                            subscriptionId,
+                            productId
                         })
 
                         // Store the result
@@ -1287,7 +1290,8 @@ export const executeAgentFlow = async ({
     isTool = false,
     orgId,
     workspaceId,
-    subscriptionId
+    subscriptionId,
+    productId
 }: IExecuteAgentFlowParams) => {
     logger.debug('\n🚀 Starting flow execution')
 
@@ -1754,7 +1758,8 @@ export const executeAgentFlow = async ({
                 iterationContext,
                 orgId,
                 workspaceId,
-                subscriptionId
+                subscriptionId,
+                productId
             })
 
             if (executionResult.agentFlowExecutedData) {
@@ -2020,7 +2025,9 @@ export const executeAgentFlow = async ({
             chatflowId: chatflowid,
             chatId,
             type: evaluationRunId ? ChatType.EVALUATION : isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
-            flowGraph: getTelemetryFlowObj(nodes, edges)
+            flowGraph: getTelemetryFlowObj(nodes, edges),
+            productId,
+            subscriptionId
         },
         orgId
     )

packages/server/src/utils/buildChatflow.ts

@@ -249,7 +249,8 @@ export const executeFlow = async ({
     isTool,
     orgId,
     workspaceId,
-    subscriptionId
+    subscriptionId,
+    productId
 }: IExecuteFlowParams) => {
     // Ensure incomingInput has all required properties with default values
     incomingInput = {
@@ -421,7 +422,8 @@ export const executeFlow = async ({
             isTool,
             orgId,
             workspaceId,
-            subscriptionId
+            subscriptionId,
+            productId
         })
     }
 
@@ -812,7 +814,9 @@ export const executeFlow = async ({
                 chatflowId: chatflowid,
                 chatId,
                 type: isEvaluation ? ChatType.EVALUATION : isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
-                flowGraph: getTelemetryFlowObj(nodes, edges)
+                flowGraph: getTelemetryFlowObj(nodes, edges),
+                productId,
+                subscriptionId
             },
             orgId
         )
@@ -954,6 +958,9 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals
         organizationId = orgId
         const subscriptionId = org.subscriptionId as string
 
+        const subscriptionDetails = await appServer.usageCacheManager.getSubscriptionDataFromCache(subscriptionId)
+        const productId = subscriptionDetails?.productId || ''
+
         await checkPredictions(orgId, subscriptionId, appServer.usageCacheManager)
 
         const executeData: IExecuteFlowParams = {
@@ -974,7 +981,8 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals
             usageCacheManager: appServer.usageCacheManager,
             orgId,
             workspaceId,
-            subscriptionId
+            subscriptionId,
+            productId
         }
 
         if (process.env.MODE === MODE.QUEUE) {

packages/server/src/utils/upsertVector.ts

@@ -277,6 +277,9 @@ export const upsertVector = async (req: Request, isInternal: boolean = false) =>
         const orgId = org.id
         const subscriptionId = org.subscriptionId as string
 
+        const subscriptionDetails = await appServer.usageCacheManager.getSubscriptionDataFromCache(subscriptionId)
+        const productId = subscriptionDetails?.productId || ''
+
         const executeData: IExecuteFlowParams = {
             componentNodes: appServer.nodesPool.componentNodes,
             incomingInput,
@@ -293,7 +296,8 @@ export const upsertVector = async (req: Request, isInternal: boolean = false) =>
             isUpsert: true,
             orgId,
             workspaceId,
-            subscriptionId
+            subscriptionId,
+            productId
         }
 
         if (process.env.MODE === MODE.QUEUE) {

packages/ui/src/ui-component/dialog/ViewMessagesDialog.jsx

@@ -185,7 +185,6 @@ const ViewMessagesDialog = ({ show, dialogProps, onCancel }) => {
     const closeSnackbar = (...args) => dispatch(closeSnackbarAction(...args))
 
     const [chatlogs, setChatLogs] = useState([])
-    const [allChatlogs, setAllChatLogs] = useState([])
     const [chatMessages, setChatMessages] = useState([])
     const [stats, setStats] = useState({})
     const [selectedMessageIndex, setSelectedMessageIndex] = useState(0)
@@ -758,7 +757,6 @@ const ViewMessagesDialog = ({ show, dialogProps, onCancel }) => {
         if (getChatmessageApi.data) {
             getStoragePathFromServer.request()
 
-            setAllChatLogs(getChatmessageApi.data)
             const chatPK = processChatLogs(getChatmessageApi.data)
             setSelectedMessageIndex(0)
             if (chatPK) {
@@ -794,7 +792,6 @@ const ViewMessagesDialog = ({ show, dialogProps, onCancel }) => {
 
         return () => {
             setChatLogs([])
-            setAllChatLogs([])
             setChatMessages([])
             setChatTypeFilter(['INTERNAL', 'EXTERNAL'])
             setFeedbackTypeFilter([])