Billing - Migrate from Stripe metering (#20298)

**Overall strategy**
**1. Introduce “Billing V2” behind a workspace flag**
Gate the new model with FeatureFlagKey.IS_BILLING_V2_ENABLED so existing
workspaces stay on the old behavior until they’re migrated or explicitly
on V2.

**2. Replace workflow metered SKUs with a resource-credit product**
Conceptually, billable “workflow execution” usage is not the primary
subscription line item anymore. Add a RESOURCE_CREDIT product (and keep
WORKFLOW_NODE_EXECUTION as deprecated for the transition). Usage and
limits are expressed through credit buckets (e.g. price metadata like
credit_amount), so one product can represent pooled credits instead of a
narrow workflow-only meter.

**3. Migrate subscriptions in two layers**
Schema/catalog: persist extra price metadata (instance upgrade) so the
server knows credit amounts and can match Stripe prices to the new
model.
Per workspace: the registered workspace command
upgrade:2-2:migrate-to-billing-v2 finds subscriptions that still have
WORKFLOW_NODE_EXECUTION, swaps those items to the right RESOURCE_CREDIT
prices (using existing Stripe schedule +
BillingSubscriptionUpdateService stack), then treats the workspace as V2
(flag). Workspaces without that legacy item or without a subscription
are skipped.

**4. Unify subscription lifecycle + usage on the server**

**5. Refresh the product surface in Settings**

Test : 

- [x] Subscribe v1 +  Update subscribe + Migrate
- [x]  Subscribe v2 + Update subscribe

packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts

@@ -1,6 +1,7 @@
 import { Module } from '@nestjs/common';
 
 import { BillingModule } from 'src/engine/core-modules/billing/billing.module';
+import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module';
 import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
 import { ToolModule } from 'src/engine/core-modules/tool/tool.module';
@@ -30,6 +31,7 @@ import { WorkflowRunModule } from 'src/modules/workflow/workflow-runner/workflow
     RecordCRUDActionModule,
     FormActionModule,
     BillingModule,
+    WorkspaceCacheModule,
     FilterActionModule,
     IfElseActionModule,
     IteratorActionModule,

packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts

@@ -11,6 +11,7 @@ import { USAGE_RECORDED } from 'src/engine/core-modules/usage/constants/usage-re
 import { UsageOperationType } from 'src/engine/core-modules/usage/enums/usage-operation-type.enum';
 import { UsageResourceType } from 'src/engine/core-modules/usage/enums/usage-resource-type.enum';
 import { UsageUnit } from 'src/engine/core-modules/usage/enums/usage-unit.enum';
+import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
 import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
 import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory';
 import { shouldExecuteStep } from 'src/modules/workflow/workflow-executor/utils/should-execute-step.util';
@@ -120,6 +121,16 @@ describe('WorkflowExecutorWorkspaceService', () => {
           provide: BillingUsageService,
           useValue: mockBillingUsageService,
         },
+        {
+          provide: WorkspaceCacheService,
+          useValue: {
+            getOrRecompute: jest.fn().mockResolvedValue({
+              billingSubscription: {
+                currentPeriodStart: new Date('2026-04-01T00:00:00Z'),
+              },
+            }),
+          },
+        },
         {
           provide: ExceptionHandlerService,
           useValue: mockExceptionHandlerService,
@@ -225,6 +236,7 @@ describe('WorkflowExecutorWorkspaceService', () => {
             quantity: 1,
             unit: UsageUnit.INVOCATION,
             resourceId: 'workflow-id',
+            periodStart: new Date('2026-04-01T00:00:00.000Z'),
           },
         ],
         'workspace-id',
@@ -707,6 +719,7 @@ describe('WorkflowExecutorWorkspaceService', () => {
             quantity: 1,
             unit: UsageUnit.INVOCATION,
             resourceId: 'workflow-id',
+            periodStart: new Date('2026-04-01T00:00:00.000Z'),
           },
         ],
         'workspace-id',

packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts

@@ -22,6 +22,7 @@ import { UsageOperationType } from 'src/engine/core-modules/usage/enums/usage-op
 import { UsageResourceType } from 'src/engine/core-modules/usage/enums/usage-resource-type.enum';
 import { UsageUnit } from 'src/engine/core-modules/usage/enums/usage-unit.enum';
 import { type UsageEvent } from 'src/engine/core-modules/usage/types/usage-event.type';
+import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
 import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
 import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
 import { workflowHasRunningSteps } from 'src/modules/workflow/common/utils/workflow-has-running-steps.util';
@@ -60,6 +61,7 @@ export class WorkflowExecutorWorkspaceService {
     private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService,
     private readonly billingService: BillingService,
     private readonly billingUsageService: BillingUsageService,
+    private readonly workspaceCacheService: WorkspaceCacheService,
     private readonly exceptionHandlerService: ExceptionHandlerService,
     private readonly metricsService: MetricsService,
     @InjectMessageQueue(MessageQueue.workflowQueue)
@@ -363,6 +365,22 @@ export class WorkflowExecutorWorkspaceService {
     workspaceId: string,
     workflowId: string,
   ) {
+    let periodStart: Date | undefined;
+    if (this.billingService.isBillingEnabled()) {
+      const {
+        billingSubscription: { currentPeriodStart },
+      } = await this.workspaceCacheService.getOrRecompute(workspaceId, [
+        'billingSubscription',
+      ]);
+
+      periodStart = currentPeriodStart;
+
+      await this.billingUsageService.decrementAvailableCredits({
+        workspaceId,
+        usedCredits: 100,
+      });
+    }
+
     this.workspaceEventEmitter.emitCustomBatchEvent<UsageEvent>(
       USAGE_RECORDED,
       [
@@ -373,17 +391,11 @@ export class WorkflowExecutorWorkspaceService {
           quantity: 1,
           unit: UsageUnit.INVOCATION,
           resourceId: workflowId,
+          periodStart,
         },
       ],
       workspaceId,
     );
-
-    if (this.billingService.isBillingEnabled()) {
-      await this.billingUsageService.decrementAvailableCredits({
-        workspaceId,
-        usedCredits: 100,
-      });
-    }
   }
 
   private async processStepExecutionResult({