diff --git a/packages/twenty-server/src/database/clickHouse/migrations/004-create-billing-event-table.sql b/packages/twenty-server/src/database/clickHouse/migrations/004-create-billing-event-table.sql new file mode 100644 index 00000000000..83c42e9327f --- /dev/null +++ b/packages/twenty-server/src/database/clickHouse/migrations/004-create-billing-event-table.sql @@ -0,0 +1,16 @@ +CREATE TABLE IF NOT EXISTS billingEvent +( + `eventId` UUID DEFAULT generateUUIDv4(), + `timestamp` DateTime64(3) NOT NULL, + `workspaceId` String NOT NULL, + `userWorkspaceId` String DEFAULT '', + `eventType` LowCardinality(String) NOT NULL, + `creditsUsed` Int64 NOT NULL, + `costInMicroDollars` Int64 DEFAULT 0, + `resourceType` LowCardinality(String) DEFAULT '', + `resourceId` String DEFAULT '', + `metadata` JSON +) +ENGINE = MergeTree +ORDER BY (workspaceId, userWorkspaceId, timestamp, eventType) +TTL timestamp + INTERVAL 3 YEAR DELETE; diff --git a/packages/twenty-server/src/engine/core-modules/billing/billing.module.ts b/packages/twenty-server/src/engine/core-modules/billing/billing.module.ts index 4454d215e85..1fb0f31153f 100644 --- a/packages/twenty-server/src/engine/core-modules/billing/billing.module.ts +++ b/packages/twenty-server/src/engine/core-modules/billing/billing.module.ts @@ -27,6 +27,9 @@ import { BillingSubscriptionItemService } from 'src/engine/core-modules/billing/ import { BillingSubscriptionPhaseService } from 'src/engine/core-modules/billing/services/billing-subscription-phase.service'; import { BillingSubscriptionUpdateService } from 'src/engine/core-modules/billing/services/billing-subscription-update.service'; import { BillingSubscriptionService } from 'src/engine/core-modules/billing/services/billing-subscription.service'; +import { BillingAnalyticsService } from 'src/engine/core-modules/billing/services/billing-analytics.service'; +import { BillingBudgetGuardService } from 'src/engine/core-modules/billing/services/billing-budget-guard.service'; +import { BillingEventWriterService } from 'src/engine/core-modules/billing/services/billing-event-writer.service'; import { BillingUsageService } from 'src/engine/core-modules/billing/services/billing-usage.service'; import { BillingService } from 'src/engine/core-modules/billing/services/billing.service'; import { MeteredCreditService } from 'src/engine/core-modules/billing/services/metered-credit.service'; @@ -38,6 +41,7 @@ import { MessageQueueModule } from 'src/engine/core-modules/message-queue/messag import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module'; import { UserWorkspaceEntity } from 'src/engine/core-modules/user-workspace/user-workspace.entity'; import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; +import { ClickHouseModule } from 'src/database/clickHouse/clickHouse.module'; import { AiBillingModule } from 'src/engine/metadata-modules/ai/ai-billing/ai-billing.module'; import { AiModelsModule } from 'src/engine/metadata-modules/ai/ai-models/ai-models.module'; import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-source.module'; @@ -52,6 +56,7 @@ import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permi AiBillingModule, AiModelsModule, WorkspaceDomainsModule, + ClickHouseModule, TypeOrmModule.forFeature([ BillingSubscriptionEntity, BillingSubscriptionItemEntity, @@ -84,6 +89,9 @@ import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permi BillingUpdateSubscriptionPriceCommand, BillingSyncPlansDataCommand, BillingUsageService, + BillingEventWriterService, + BillingBudgetGuardService, + BillingAnalyticsService, BillingPriceService, BillingCreditRolloverService, MeteredCreditService, @@ -96,6 +104,9 @@ import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permi BillingPortalWorkspaceService, BillingService, BillingUsageService, + BillingEventWriterService, + BillingBudgetGuardService, + BillingAnalyticsService, BillingCreditRolloverService, MeteredCreditService, ], diff --git a/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts b/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts index 9fe7d9afc70..f19d651ca25 100644 --- a/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts +++ b/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts @@ -6,6 +6,7 @@ import { isDefined } from 'twenty-shared/utils'; import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator'; import { BILLING_FEATURE_USED } from 'src/engine/core-modules/billing/constants/billing-feature-used.constant'; +import { BillingEventWriterService } from 'src/engine/core-modules/billing/services/billing-event-writer.service'; import { BillingUsageService } from 'src/engine/core-modules/billing/services/billing-usage.service'; import { type BillingUsageEvent } from 'src/engine/core-modules/billing/types/billing-usage-event.type'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; @@ -15,6 +16,7 @@ import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/ty export class BillingFeatureUsedListener { constructor( private readonly billingUsageService: BillingUsageService, + private readonly billingEventWriterService: BillingEventWriterService, private readonly twentyConfigService: TwentyConfigService, ) {} @@ -38,9 +40,9 @@ export class BillingFeatureUsedListener { return; } - await this.billingUsageService.billUsage({ - workspaceId: payload.workspaceId, - billingEvents: payload.events, - }); + await this.billingEventWriterService.writeBillingEvents( + payload.workspaceId, + payload.events, + ); } } diff --git a/packages/twenty-server/src/engine/core-modules/billing/services/billing-event-writer.service.ts b/packages/twenty-server/src/engine/core-modules/billing/services/billing-event-writer.service.ts new file mode 100644 index 00000000000..54bb927f8ce --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/billing/services/billing-event-writer.service.ts @@ -0,0 +1,105 @@ +/* @license Enterprise */ + +import { Injectable, Logger } from '@nestjs/common'; + +import { isDefined } from 'twenty-shared/utils'; + +import { type BillingExecutionType } from 'src/engine/core-modules/billing/types/billing-dimensions.type'; +import { type BillingUsageEvent } from 'src/engine/core-modules/billing/types/billing-usage-event.type'; +import { buildUserUsageCacheKey } from 'src/engine/core-modules/billing/utils/build-user-usage-cache-key.util'; +import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; +import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; +import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +import { ClickHouseService } from 'src/database/clickHouse/clickHouse.service'; + +import { BillingUsageService } from './billing-usage.service'; + +@Injectable() +export class BillingEventWriterService { + private readonly logger = new Logger(BillingEventWriterService.name); + + constructor( + private readonly clickHouseService: ClickHouseService, + private readonly billingUsageService: BillingUsageService, + private readonly twentyConfigService: TwentyConfigService, + @InjectCacheStorage(CacheStorageNamespace.EngineWorkspace) + private readonly cacheStorage: CacheStorageService, + ) {} + + async writeBillingEvents( + workspaceId: string, + events: BillingUsageEvent[], + ): Promise { + if (this.twentyConfigService.get('CLICKHOUSE_URL')) { + this.writeToClickHouse(workspaceId, events).catch((error) => { + this.logger.error( + 'Failed to write billing events to ClickHouse', + error, + ); + }); + + this.invalidateUserUsageCache(workspaceId, events); + } + + await this.billingUsageService.billUsage({ + workspaceId, + billingEvents: events, + }); + } + + private async writeToClickHouse( + workspaceId: string, + events: BillingUsageEvent[], + ): Promise { + const rows = events.map((event) => ({ + timestamp: new Date(), + workspaceId, + userWorkspaceId: event.userWorkspaceId ?? '', + eventType: event.dimensions?.execution_type ?? '', + creditsUsed: event.value, + resourceType: this.mapExecutionTypeToResourceType( + event.dimensions?.execution_type, + ), + resourceId: event.dimensions?.resource_id ?? '', + metadata: { + executionContext1: event.dimensions?.execution_context_1 ?? null, + eventName: event.eventName, + }, + })); + + await this.clickHouseService.insert('billingEvent', rows); + } + + private invalidateUserUsageCache( + workspaceId: string, + events: BillingUsageEvent[], + ): void { + const userWorkspaceIds = new Set( + events.map((event) => event.userWorkspaceId).filter(isDefined), + ); + + for (const userWorkspaceId of userWorkspaceIds) { + this.cacheStorage + .del(buildUserUsageCacheKey(workspaceId, userWorkspaceId)) + .catch(() => {}); + } + } + + private mapExecutionTypeToResourceType( + executionType: BillingExecutionType | undefined, + ): string { + switch (executionType) { + case 'ai_token': + return 'agent'; + case 'workflow_execution': + return 'workflow'; + case 'app_invocation': + return 'app'; + case 'code_execution': + return 'code'; + default: + return ''; + } + } +} diff --git a/packages/twenty-server/src/engine/core-modules/billing/types/billing-dimensions.type.ts b/packages/twenty-server/src/engine/core-modules/billing/types/billing-dimensions.type.ts index 1fbdf424b8c..c16d4458298 100644 --- a/packages/twenty-server/src/engine/core-modules/billing/types/billing-dimensions.type.ts +++ b/packages/twenty-server/src/engine/core-modules/billing/types/billing-dimensions.type.ts @@ -3,7 +3,8 @@ export type BillingExecutionType = | 'workflow_execution' | 'code_execution' - | 'ai_token'; + | 'ai_token' + | 'app_invocation'; export type BillingDimensions = { execution_type: BillingExecutionType; diff --git a/packages/twenty-server/src/engine/core-modules/billing/types/billing-usage-event.type.ts b/packages/twenty-server/src/engine/core-modules/billing/types/billing-usage-event.type.ts index d560fe68439..2ae69017347 100644 --- a/packages/twenty-server/src/engine/core-modules/billing/types/billing-usage-event.type.ts +++ b/packages/twenty-server/src/engine/core-modules/billing/types/billing-usage-event.type.ts @@ -9,4 +9,5 @@ export type BillingUsageEvent = { eventName: BillingMeterEventName; value: NonNegative; dimensions?: BillingDimensions; + userWorkspaceId?: string; }; diff --git a/packages/twenty-server/src/engine/core-modules/billing/utils/build-user-usage-cache-key.util.ts b/packages/twenty-server/src/engine/core-modules/billing/utils/build-user-usage-cache-key.util.ts new file mode 100644 index 00000000000..3a862a21525 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/billing/utils/build-user-usage-cache-key.util.ts @@ -0,0 +1,6 @@ +/* @license Enterprise */ + +export const buildUserUsageCacheKey = ( + workspaceId: string, + userWorkspaceId: string, +): string => `billing:user-usage:${workspaceId}:${userWorkspaceId}`; diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-billing/services/__tests__/ai-billing.service.spec.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-billing/services/__tests__/ai-billing.service.spec.ts index 4510d0a3689..39d2c765b22 100644 --- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-billing/services/__tests__/ai-billing.service.spec.ts +++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-billing/services/__tests__/ai-billing.service.spec.ts @@ -332,6 +332,7 @@ describe('AIBillingService', () => { { usage: mockTokenUsage }, 'workspace-1', 'agent-id-123', + 'user-workspace-456', ); expect( @@ -347,6 +348,35 @@ describe('AIBillingService', () => { resource_id: 'agent-id-123', execution_context_1: 'gpt-4o', }, + userWorkspaceId: 'user-workspace-456', + }, + ], + 'workspace-1', + ); + }); + + it('should emit billing event without userWorkspaceId when not provided', () => { + service.calculateAndBillUsage( + 'gpt-4o', + { usage: mockTokenUsage }, + 'workspace-1', + 'agent-id-123', + ); + + expect( + mockWorkspaceEventEmitter.emitCustomBatchEvent, + ).toHaveBeenCalledWith( + BILLING_FEATURE_USED, + [ + { + eventName: BillingMeterEventName.WORKFLOW_NODE_RUN, + value: 7500, + dimensions: { + execution_type: 'ai_token', + resource_id: 'agent-id-123', + execution_context_1: 'gpt-4o', + }, + userWorkspaceId: undefined, }, ], 'workspace-1', diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-billing/services/ai-billing.service.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-billing/services/ai-billing.service.ts index 878f2c28d9e..ff383800570 100644 --- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-billing/services/ai-billing.service.ts +++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-billing/services/ai-billing.service.ts @@ -59,13 +59,20 @@ export class AIBillingService { billingInput: BillingUsageInput, workspaceId: string, agentId?: string | null, + userWorkspaceId?: string, ): void { const costInDollars = this.calculateCost(modelId, billingInput); const creditsUsed = Math.round( convertDollarsToBillingCredits(costInDollars), ); - this.sendAiTokenUsageEvent(workspaceId, creditsUsed, modelId, agentId); + this.sendAiTokenUsageEvent( + workspaceId, + creditsUsed, + modelId, + agentId, + userWorkspaceId, + ); } private sendAiTokenUsageEvent( @@ -73,6 +80,7 @@ export class AIBillingService { creditsUsed: number, modelId: ModelId, agentId?: string | null, + userWorkspaceId?: string, ): void { this.workspaceEventEmitter.emitCustomBatchEvent( BILLING_FEATURE_USED, @@ -85,6 +93,7 @@ export class AIBillingService { resource_id: agentId || null, execution_context_1: modelId, }, + userWorkspaceId, }, ], workspaceId, diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts index e5f10f57eab..5777e6f9933 100644 --- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts +++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts @@ -243,6 +243,7 @@ export class ChatExecutionService { { usage, cacheCreationTokens }, workspace.id, null, + userWorkspaceId, ); }) .catch((error) => { diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/ai-agent/ai-agent.workflow-action.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/ai-agent/ai-agent.workflow-action.ts index c5b4dec11c3..c0ad5fc8947 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/ai-agent/ai-agent.workflow-action.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/ai-agent/ai-agent.workflow-action.ts @@ -84,11 +84,17 @@ export class AiAgentWorkflowAction implements WorkflowAction { authContext: executionContext.authContext, }); + const userWorkspaceId = + executionContext.authContext.type === 'user' + ? executionContext.authContext.userWorkspaceId + : undefined; + await this.aiBillingService.calculateAndBillUsage( agent?.modelId ?? DEFAULT_SMART_MODEL, { usage, cacheCreationTokens }, workspaceId, agent?.id || null, + userWorkspaceId, ); return {