feat(billing): add ClickHouse billing event store with dual-write and userWorkspaceId threading

- Create billingEvent ClickHouse table (migration 004) for storing all billing events
- Add userWorkspaceId to BillingUsageEvent type for user-level tracking
- Create BillingEventWriterService with dual-write (ClickHouse + Stripe)
- Wire BillingEventWriterService into BillingFeatureUsedListener
- Thread userWorkspaceId through AIBillingService from ChatExecutionService and AiAgentWorkflowAction
- Add app_invocation to BillingExecutionType for future app billing
- Register new services and ClickHouseModule in BillingModule

Co-authored-by: felix <felix@twenty.com>

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;

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,
   ],

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,
+    );
   }
 }

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<void> {
+    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<void> {
+    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 '';
+    }
+  }
+}

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;

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<number>;
   dimensions?: BillingDimensions;
+  userWorkspaceId?: string;
 };

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}`;

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',

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<BillingUsageEvent>(
       BILLING_FEATURE_USED,
@@ -85,6 +93,7 @@ export class AIBillingService {
             resource_id: agentId || null,
             execution_context_1: modelId,
           },
+          userWorkspaceId,
         },
       ],
       workspaceId,

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) => {

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 {