diff --git a/packages/components/src/handler.ts b/packages/components/src/handler.ts index 3fcd3d2b497..d087fd216b1 100644 --- a/packages/components/src/handler.ts +++ b/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}`], + tags: [`Name:${chatflow.name}`, 'billing:pending'], 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}`], + tags: [`Name:${chatflow.name}`, 'billing:pending'], metadata, userId: options?.user?.id, sessionId: options.sessionId, diff --git a/packages/server/src/aai-utils/billing/config.ts b/packages/server/src/aai-utils/billing/config.ts index 5b3ea94434a..ac5f84cd50d 100644 --- a/packages/server/src/aai-utils/billing/config.ts +++ b/packages/server/src/aai-utils/billing/config.ts @@ -41,6 +41,19 @@ export const BILLING_CONFIG = { BATCH_DELAY_MS: 1000 }, + // Sync configuration + 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' + }, + // Resource configuration AI_TOKENS: { TOKENS_PER_CREDIT: 10, // 1,000 tokens = 100 Credits diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index 7cf14263da8..9e8bfff5d7b 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -13,43 +13,200 @@ export class LangfuseProvider { // Cache for flow platform status (resets each sync) private platformNodeCache = new Map() + // Future timestamp buffer (5 minutes) for timestamp validation + private static readonly FUTURE_TIMESTAMP_BUFFER_SECONDS = 300 + + /** + * Sync usage data from Langfuse to Stripe + * + * Auto-tagging: + * - New traces are automatically tagged with 'billing:pending' on creation (handler.ts:678,694) + * - Self-healing catches any edge cases where tags might be missing + * - Traces are marked 'billing:processed' after successful sync to Stripe + * + * Tag-based filtering (optional optimization): + * - When BILLING_USE_TAG_FILTERING=true, only fetches traces with 'billing:pending' tag + * - Significantly reduces API calls and memory usage for large datasets (40k → ~100s traces) + * - Self-healing logic still runs in-memory as defensive measure + * - SAFE to enable after backfill completes (auto-tagging ensures new traces won't be missed) + * + * ⚠️ IMPORTANT - Before enabling tag filtering: + * - Run backfill script ONCE to tag existing traces (packages/server/scripts/backfill-billing-tags.ts) + * - Backfill only needed for traces created BEFORE auto-tagging was implemented + * - After backfill, new traces will be automatically tagged on creation + * + * When disabled (default): + * - Fetches all traces from lookback period + * - Filters in-memory based on metadata.billing_status and tags + * - Self-healing catches ALL untagged traces automatically + * - Slower but guarantees no traces are missed (recommended until backfill completes) + * + * @param traceId - Optional specific trace ID to sync + */ async syncUsageToStripe(traceId?: string): Promise { - let traces: Trace[] = [] - let creditsDataWithTraces: Array<{ creditsData: CreditsData; fullTrace: any }> = [] let failedTraces: Array<{ traceId: string; error: string }> = [] let processedTraces: string[] = [] let skippedTraces: Array<{ traceId: string; reason: string }> = [] let meterEvents: Stripe.Billing.MeterEvent[] = [] try { - traces = await this.fetchUsageData(traceId) - // Track skipped traces from initial filtering - const skippedFromInitialFilter = traces.filter((trace) => { + // Handle single trace lookup separately (no pagination needed) + if (traceId) { + // Fetch single trace directly from Langfuse + const trace = await langfuse.fetchTrace(traceId) + const traces = trace.data + ? [ + { + ...trace.data, + observations: trace.data.observations?.map((obs: any) => obs?.id) + } as any + ] + : [] + const creditsDataWithTraces = await this.convertUsageToCredits(traces) + + // Sync to Stripe + const stripeProvider = new StripeProvider() + const stripeResponse = await stripeProvider.syncUsageToStripe( + creditsDataWithTraces.map((item) => ({ + ...item.creditsData, + fullTrace: item.fullTrace + })) + ) + + return { + processedTraces: stripeResponse.processedTraces, + failedTraces: stripeResponse.failedEvents, + skippedTraces: [], + meterEvents: stripeResponse.meterEvents + } + } + + // Calculate lookback timestamp + const fromTimestamp = new Date() + fromTimestamp.setDate(fromTimestamp.getDate() - BILLING_CONFIG.SYNC.LOOKBACK_DAYS) + + log.info('Starting streaming sync', { + lookbackDays: BILLING_CONFIG.SYNC.LOOKBACK_DAYS, + fromTimestamp: fromTimestamp.toISOString(), + useTagFiltering: BILLING_CONFIG.SYNC.USE_TAG_FILTERING + }) + + // Get total pages count with initial request + // If tag filtering is enabled, only fetch traces with 'billing:pending' tag + const fetchParams: any = { + fromTimestamp, + limit: 100, + page: 1 + } + + if (BILLING_CONFIG.SYNC.USE_TAG_FILTERING) { + fetchParams.tags = ['billing:pending'] + } + + const initialResponse = await langfuse.fetchTraces(fetchParams) + + const totalPages = initialResponse.meta.totalPages + log.info('Total pages to process', { totalPages }) + + // Process first page + let firstPageUntaggedCount = 0 + const firstPageTraces = initialResponse.data.filter((trace) => { + const hasTokenCost = trace.totalCost > 0 + const hasComputeTime = trace.latency > 0 const metadata = (trace.metadata as TraceMetadata) || {} - if (metadata.billing_status === 'processed') { - skippedTraces.push({ traceId: trace.id, reason: 'Already processed' }) - return true + const tags = this.getTraceTags(trace) + + // Self-healing: Check both metadata AND tags for processed status + // A trace is considered processed if EITHER: + // 1. metadata.billing_status === 'processed', OR + // 2. tags includes 'billing:processed' + const isProcessed = metadata.billing_status === 'processed' || tags.includes('billing:processed') + const isNotProcessed = !isProcessed + + // Track untagged traces for logging + const hasNoBillingTags = !tags.includes('billing:processed') && !tags.includes('billing:pending') + if (isNotProcessed && hasNoBillingTags && (hasTokenCost || hasComputeTime)) { + firstPageUntaggedCount++ } - if (!(trace.totalCost > 0 || trace.latency > 0)) { + + if (isProcessed) { + skippedTraces.push({ traceId: trace.id, reason: 'Already processed' }) + } else if (!(hasTokenCost || hasComputeTime)) { skippedTraces.push({ traceId: trace.id, reason: 'No billable usage' }) - return true } - return false + + return (hasTokenCost || hasComputeTime) && isNotProcessed }) - creditsDataWithTraces = await this.convertUsageToCredits(traces) + // Log if we found untagged traces on first page (self-healing scenario) + if (firstPageUntaggedCount > 0) { + log.info('Self-healing: Found untagged traces on first page', { + count: firstPageUntaggedCount + }) + } + + if (firstPageTraces.length > 0) { + const creditsDataWithTraces = await this.convertUsageToCredits(firstPageTraces) + const stripeProvider = new StripeProvider() + const stripeResponse = await stripeProvider.syncUsageToStripe( + creditsDataWithTraces.map((item) => ({ + ...item.creditsData, + fullTrace: item.fullTrace + })) + ) - // Create meter events in Stripe - const stripeProvider = new StripeProvider() - const stripeResponse = await stripeProvider.syncUsageToStripe( - creditsDataWithTraces.map((item) => ({ - ...item.creditsData, - fullTrace: item.fullTrace - })) - ) - meterEvents = stripeResponse.meterEvents - failedTraces = stripeResponse.failedEvents - processedTraces = stripeResponse.processedTraces + processedTraces.push(...stripeResponse.processedTraces) + failedTraces.push(...stripeResponse.failedEvents) + meterEvents.push(...stripeResponse.meterEvents) + } + + // Process remaining pages in groups + const PAGE_BATCH_SIZE = BILLING_CONFIG.SYNC.PAGE_BATCH_SIZE + const RATE_LIMIT_DELAY_MS = BILLING_CONFIG.SYNC.RATE_LIMIT_DELAY_MS + + for (let startPage = 2; startPage <= totalPages; startPage += PAGE_BATCH_SIZE) { + const endPage = Math.min(startPage + PAGE_BATCH_SIZE - 1, totalPages) + + log.info('Processing page group', { + startPage, + endPage, + progress: `${endPage}/${totalPages}` + }) + + // Fetch page group + const traces = await this.fetchPageGroup(startPage, endPage, fromTimestamp, undefined, skippedTraces) + + if (traces.length > 0) { + // Convert to credits + const creditsDataWithTraces = await this.convertUsageToCredits(traces) + + // Sync to Stripe + const stripeProvider = new StripeProvider() + const stripeResponse = await stripeProvider.syncUsageToStripe( + creditsDataWithTraces.map((item) => ({ + ...item.creditsData, + fullTrace: item.fullTrace + })) + ) + + // Accumulate results + processedTraces.push(...stripeResponse.processedTraces) + failedTraces.push(...stripeResponse.failedEvents) + meterEvents.push(...stripeResponse.meterEvents) + } + + // Rate limit between page groups (skip on last batch) + if (endPage < totalPages) { + await new Promise((resolve) => setTimeout(resolve, RATE_LIMIT_DELAY_MS)) + } + } + + log.info('Streaming sync completed', { + totalPagesProcessed: totalPages, + processedCount: processedTraces.length, + failedCount: failedTraces.length, + skippedCount: skippedTraces.length + }) } catch (error: any) { log.error('Error syncing usage data:', error) failedTraces = [{ traceId: traceId || 'unknown', error: error.message }] @@ -62,97 +219,115 @@ export class LangfuseProvider { processedTraces, failedTraces, skippedTraces, - traces, - creditsData: creditsDataWithTraces.map((item) => item.creditsData), meterEvents } } - private async fetchUsageData(traceId?: string): Promise { - try { - if (traceId) { - const trace = await langfuse.fetchTrace(traceId) - if (trace.data) { - return [ - { - ...trace.data, - observations: trace.data.observations?.map((obs: any) => obs?.id) - } as any - ] - } - } + /** + * Safely extract tags array from trace + * @param trace - Trace object to extract tags from + * @returns Array of tag strings + */ + private getTraceTags(trace: Trace): string[] { + return (trace.tags || []) as string[] + } - const startOfMonth = new Date() - startOfMonth.setDate(1) - startOfMonth.setHours(0, 0, 0, 0) + /** + * Fetch a group of pages in parallel (memory-efficient batch processing) + * @param startPage - Starting page number (1-indexed) + * @param endPage - Ending page number (inclusive) + * @param fromTimestamp - Timestamp to fetch traces from + * @param traceId - Optional specific trace ID to filter for + * @param skippedTraces - Optional array to track skipped traces + * @returns Array of filtered traces from the page range + */ + private async fetchPageGroup( + startPage: number, + endPage: number, + fromTimestamp: Date, + traceId?: string, + skippedTraces?: Array<{ traceId: string; reason: string }> + ): Promise { + const batch = [] + + // Build base fetch parameters + const baseFetchParams: any = { + fromTimestamp, + limit: 100 + } - // Get total pages first - const initialResponse = await langfuse.fetchTraces({ - fromTimestamp: startOfMonth, - limit: 100, - page: 1 - }) + // If tag filtering is enabled, only fetch traces with 'billing:pending' tag + if (BILLING_CONFIG.SYNC.USE_TAG_FILTERING) { + baseFetchParams.tags = ['billing:pending'] + } - const totalPages = initialResponse.meta.totalPages - let allTraces: GetLangfuseTracesResponse['data'] = [] + // Build batch of page requests + for (let page = startPage; page <= endPage; page++) { + batch.push( + langfuse.fetchTraces({ + ...baseFetchParams, + page + }) + ) + } + + // Fetch all pages in parallel + const responses = await Promise.all(batch) - // Process initial response - const validInitialTraces = initialResponse.data.filter((trace) => { + // Filter and combine traces from all responses + // Note: If USE_TAG_FILTERING is enabled, we already filtered at API level, + // but we still do in-memory filtering as a defensive measure for self-healing + const traces: GetLangfuseTracesResponse['data'] = [] + let untaggedCount = 0 + + responses.forEach((response) => { + const validTraces = response.data.filter((trace) => { if (traceId && trace.id !== traceId) { return false } const hasTokenCost = trace.totalCost > 0 const hasComputeTime = trace.latency > 0 const metadata = (trace.metadata as TraceMetadata) || {} - const isNotProcessed = metadata.billing_status !== 'processed' - return (hasTokenCost || hasComputeTime) && isNotProcessed - }) - allTraces = allTraces.concat(validInitialTraces) - - // Fetch remaining pages in parallel with rate limiting - const BATCH_SIZE = 15 // Number of concurrent requests - const RATE_LIMIT_DELAY = 1000 // 1 second delay between batches - - for (let i = 2; i <= totalPages; i += BATCH_SIZE) { - const batch = [] - for (let j = 0; j < BATCH_SIZE && i + j <= totalPages; j++) { - batch.push( - langfuse.fetchTraces({ - fromTimestamp: startOfMonth, - limit: 100, - page: i + j - }) - ) + const tags = this.getTraceTags(trace) + + // Self-healing: Check both metadata AND tags for processed status + // A trace is considered processed if EITHER: + // 1. metadata.billing_status === 'processed', OR + // 2. tags includes 'billing:processed' + const isProcessed = metadata.billing_status === 'processed' || tags.includes('billing:processed') + const isNotProcessed = !isProcessed + + // Track untagged traces for logging (self-healing scenario) + const hasNoBillingTags = !tags.includes('billing:processed') && !tags.includes('billing:pending') + if (isNotProcessed && hasNoBillingTags && (hasTokenCost || hasComputeTime)) { + untaggedCount++ } - const responses = await Promise.all(batch) - - responses.forEach((response) => { - const validTraces = response.data.filter((trace) => { - if (traceId && trace.id !== traceId) { - return false - } - const hasTokenCost = trace.totalCost > 0 - const hasComputeTime = trace.latency > 0 - const metadata = (trace.metadata as TraceMetadata) || {} - const isNotProcessed = metadata.billing_status !== 'processed' - return (hasTokenCost || hasComputeTime) && isNotProcessed - }) - allTraces = allTraces.concat(validTraces) - }) - - // Rate limit delay between batches - if (i + BATCH_SIZE <= totalPages) { - await new Promise((resolve) => setTimeout(resolve, RATE_LIMIT_DELAY)) + // Track skipped traces (if tracking array provided) + if (skippedTraces) { + if (isProcessed) { + skippedTraces.push({ traceId: trace.id, reason: 'Already processed' }) + } else if (!(hasTokenCost || hasComputeTime)) { + skippedTraces.push({ traceId: trace.id, reason: 'No billable usage' }) + } } - } - return allTraces - } catch (error) { - log.error('Error fetching usage data from Langfuse:', error) - console.log(error) - throw error + return (hasTokenCost || hasComputeTime) && isNotProcessed + }) + traces.push(...validTraces) + }) + + // Log if we found untagged traces (self-healing scenario) + // This should be rare when tag filtering is enabled + if (untaggedCount > 0) { + log.info('Self-healing: Found untagged traces with billable usage', { + count: untaggedCount, + pageRange: `${startPage}-${endPage}`, + tagFilteringEnabled: BILLING_CONFIG.SYNC.USE_TAG_FILTERING + }) } + + return traces } private async validateUsageData(trace: Trace): Promise { @@ -204,7 +379,7 @@ export class LangfuseProvider { const traceTimestampSeconds = Math.floor(traceDate.getTime() / 1000) // Skip future timestamps (with 5 min buffer) - if (traceTimestampSeconds > nowUtcSeconds + 300) { + if (traceTimestampSeconds > nowUtcSeconds + LangfuseProvider.FUTURE_TIMESTAMP_BUFFER_SECONDS) { log.warn('Skipping trace with future timestamp', { traceId: trace.id, timestamp: trace.timestamp, diff --git a/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts b/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts index c3491657b12..2aeeb525296 100644 --- a/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts +++ b/packages/server/src/aai-utils/billing/stripe/StripeProvider.ts @@ -307,6 +307,7 @@ export class StripeProvider { const meterEvents: Stripe.Billing.MeterEvent[] = [] const failedEvents: Array<{ traceId: string; error: string }> = [] const processedTraces: string[] = [] + let selfHealedCount = 0 // console.log('Syncing usage to Stripe', { // count: creditsData.length, @@ -364,7 +365,10 @@ export class StripeProvider { }) // Update trace metadata with billing info - await this.updateTraceMetadata(data, result, batchStartTime, BATCH_SIZE, i, meterId) + const wasSelfHealed = await this.updateTraceMetadata(data, result, batchStartTime, BATCH_SIZE, i, meterId) + if (wasSelfHealed) { + selfHealedCount++ + } meterEvents.push(result) processedTraces.push(data.traceId) @@ -389,6 +393,15 @@ export class StripeProvider { } } + // Log self-healing summary + if (selfHealedCount > 0) { + log.info('Self-healing: Processed untagged traces', { + count: selfHealedCount, + totalProcessed: processedTraces.length, + percentage: ((selfHealedCount / processedTraces.length) * 100).toFixed(2) + '%' + }) + } + return { meterEvents, failedEvents, @@ -446,14 +459,33 @@ export class StripeProvider { batchSize: number, batchIndex: number, meterId: string - ): Promise { + ): Promise { const totalCredits = data.credits.ai_tokens + data.credits.compute + data.credits.storage + // Self-healing: Prepare updated tags + // - Remove 'billing:pending' if it exists + // - Add 'billing:processed' to mark as completed + // - If trace was never tagged (missed by backfill), this will add the processed tag + // - If trace has 'billing:pending', it will be replaced with 'billing:processed' + const existingTags = (data.fullTrace?.tags || []) as string[] + const hasBillingProcessed = existingTags.includes('billing:processed') + const hasBillingPending = existingTags.includes('billing:pending') + const hasNoBillingTags = !hasBillingProcessed && !hasBillingPending + + // Only update tags if not already processed (defensive check) + // Use Set to prevent duplicate tags in case of concurrent updates (race condition) + const updatedTags = hasBillingProcessed + ? existingTags + : Array.from(new Set([...existingTags.filter((tag: string) => tag !== 'billing:pending'), 'billing:processed'])) + await langfuse.trace({ id: data.traceId, timestamp: data.fullTrace?.timestamp, + tags: updatedTags, metadata: { ...data.fullTrace?.metadata, + // Self-healing: Always set billing_status to processed + // This ensures consistency even if the trace was previously untagged billing_status: 'processed', meter_event_id: result.identifier, billing_details: { @@ -492,6 +524,9 @@ export class StripeProvider { } } }) + + // Return true if self-healing occurred (trace had no billing tags) + return hasNoBillingTags } private calculateResourceBreakdown(credits: number, cost: number, totalCredits: number) {