refactor: Metadata-based filtering + Memory optimization

packages/components/package.json

@@ -47,6 +47,7 @@
         "@google-cloud/storage": "^7.17.1",
         "@google/generative-ai": "^0.24.1",
         "@huggingface/inference": "^2.8.1",
+        "@jlinc/langchain": "^0.1.3",
         "@langchain/anthropic": "0.3.23",
         "@langchain/aws": "0.1.11",
         "@langchain/baidu-qianfan": "^0.1.0",
@@ -113,7 +114,6 @@
         "graphql": "^16.11.0",
         "html-to-text": "^9.0.5",
         "ioredis": "^5.7.0",
-        "@jlinc/langchain": "^0.1.3",
         "js-levenshtein": "^1.1.6",
         "jsdom": "^22.1.0",
         "json-logic-js": "^2.0.5",
@@ -123,7 +123,8 @@
         "jsonrepair": "^3.13.0",
         "jsonschema": "^1.5.0",
         "langchain": "0.3.28",
-        "langfuse": "3.37.6",
+        "langfuse": "3.38.6",
+        "langfuse-core": "^3.38.6",
         "langfuse-langchain": "3.37.6",
         "langsmith": "0.1.6",
         "langwatch": "^0.1.7",

packages/components/src/handler.ts

@@ -675,7 +675,7 @@ export const additionalCallbacks = async (nodeData: INodeData, options: ICommonO
                         metadata: metadata,
                         userId: options?.user?.id,
                         sessionId: options.sessionId,
-                        tags: [`Name:${chatflow.name}`, 'billing:pending'],
+                        tags: [`Name:${chatflow.name}`],
                         version: chatflow.updatedDate
                         // TODO: This is still causing an error
                         // This works to keep the root trace name and have everything else update on the root trace
@@ -691,7 +691,7 @@ export const additionalCallbacks = async (nodeData: INodeData, options: ICommonO
                         try {
                             if (parentLangfuseTrace) {
                                 parentLangfuseTrace.update({
-                                    tags: [`Name:${chatflow.name}`, 'billing:pending'],
+                                    tags: [`Name:${chatflow.name}`],
                                     metadata,
                                     userId: options?.user?.id,
                                     sessionId: options.sessionId,

packages/server/package.json

@@ -108,8 +108,9 @@
         "http-errors": "^2.0.0",
         "http-status-codes": "^2.3.0",
         "langchainhub": "^0.0.11",
-        "langfuse": "3.37.6",
-        "langfuse-core": "3.37.6",
+        "@langfuse/core": "^4.2.0",
+        "langfuse": "3.38.6",
+        "langfuse-core": "^3.38.6",
         "lodash": "^4.17.21",
         "mailparser": "^3.7.4",
         "mammoth": "^1.10.0",
@@ -137,9 +138,9 @@
         "stripe": "^17.7.0",
         "turndown": "^7.2.1",
         "typeorm": "^0.3.26",
+        "undici": "^6.19.8",
         "uuid": "^9.0.1",
-        "winston": "^3.17.0",
-        "undici": "^6.19.8"
+        "winston": "^3.17.0"
     },
     "devDependencies": {
         "@jest/globals": "^29.7.0",

packages/server/src/aai-utils/billing/config.ts

@@ -1,5 +1,4 @@
 import { Logger } from 'winston'
-import { Langfuse } from 'langfuse'
 
 // Initialize logger
 export const log = console as unknown as Logger
@@ -45,13 +44,7 @@ export const BILLING_CONFIG = {
     SYNC: {
         LOOKBACK_DAYS: parseInt(process.env.BILLING_SYNC_LOOKBACK_DAYS || '90'),
         PAGE_BATCH_SIZE: 15, // Number of pages to fetch in parallel
-        RATE_LIMIT_DELAY_MS: 1000, // Delay between page batches
-        // Enable tag-based filtering (requires backfill)
-        // WARNING: Only enable after backfill is complete AND you have a mechanism to tag new traces
-        // When enabled, only fetches traces with 'billing:pending' tag
-        // New traces created after backfill won't have tags and will be MISSED
-        // Options: (1) Run backfill periodically, (2) Tag new traces on creation, or (3) Keep disabled
-        USE_TAG_FILTERING: process.env.BILLING_USE_TAG_FILTERING === 'true'
+        RATE_LIMIT_DELAY_MS: 1000 // Delay between page batches
     },
 
     // Resource configuration
@@ -86,10 +79,3 @@ export const BILLING_CONFIG = {
         STORAGE: 'Usage from data storage and persistence (1 GB/month = 500 Credits)'
     }
 }
-
-// Initialize Langfuse client
-export const langfuse = new Langfuse({
-    publicKey: process.env.LANGFUSE_PUBLIC_KEY || '',
-    secretKey: process.env.LANGFUSE_SECRET_KEY || '',
-    baseUrl: process.env.LANGFUSE_HOST || 'https://cloud.langfuse.com'
-})

packages/server/src/aai-utils/billing/core/BillingService.ts

@@ -311,10 +311,10 @@ export class BillingService implements BillingProvider {
                 durationMs: totalTime,
                 traceId,
                 processedCount: result.processedTraces.length,
-                failedCount: result.failedTraces.length,
-                skippedCount: result.skippedTraces.length,
-                meterEventsCount: result.meterEvents.length,
-                skippedTraces: result.skippedTraces
+                failedCount: result.failedTraces?.length,
+                skippedCount: result.skippedTraces?.length,
+                meterEventsCount: result.meterEvents?.length,
+                skippedTraces: result.skippedTraces?.length
             })
             return result
         } catch (error: any) {

packages/server/src/aai-utils/billing/core/types.ts

@@ -232,6 +232,9 @@ export interface SyncUsageResponse {
     meterEvents?: Stripe.Billing.MeterEvent[]
     traces?: any[]
     creditsData?: CreditsData[]
+    processedCount?: number
+    failedCount?: number
+    skippedCount?: number
 }
 
 export interface UsageSummary {