create workflow subscription item addition command (#11541)

closes https://github.com/twentyhq/core-team-issues/issues/541

packages/twenty-server/src/engine/core-modules/billing/billing.module.ts

@@ -5,6 +5,7 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 
 import { BillingController } from 'src/engine/core-modules/billing/billing.controller';
 import { BillingResolver } from 'src/engine/core-modules/billing/billing.resolver';
+import { BillingAddWorkflowSubscriptionItemCommand } from 'src/engine/core-modules/billing/commands/billing-add-workflow-subscription-item.command';
 import { BillingSyncCustomerDataCommand } from 'src/engine/core-modules/billing/commands/billing-sync-customer-data.command';
 import { BillingSyncPlansDataCommand } from 'src/engine/core-modules/billing/commands/billing-sync-plans-data.command';
 import { BillingCustomer } from 'src/engine/core-modules/billing/entities/billing-customer.entity';
@@ -82,6 +83,7 @@ import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permi
     BillingRestApiExceptionFilter,
     BillingSyncCustomerDataCommand,
     BillingSyncPlansDataCommand,
+    BillingAddWorkflowSubscriptionItemCommand,
     BillingUsageService,
   ],
   exports: [

packages/twenty-server/src/engine/core-modules/billing/commands/billing-add-workflow-subscription-item.command.ts

@@ -0,0 +1,153 @@
+/* @license Enterprise */
+
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Command } from 'nest-commander';
+import { Repository } from 'typeorm';
+
+import {
+  ActiveOrSuspendedWorkspacesMigrationCommandRunner,
+  RunOnWorkspaceArgs,
+} from 'src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner';
+import { BillingPrice } from 'src/engine/core-modules/billing/entities/billing-price.entity';
+import { BillingProduct } from 'src/engine/core-modules/billing/entities/billing-product.entity';
+import { BillingSubscription } from 'src/engine/core-modules/billing/entities/billing-subscription.entity';
+import { BillingProductKey } from 'src/engine/core-modules/billing/enums/billing-product-key.enum';
+import { StripeSubscriptionItemService } from 'src/engine/core-modules/billing/stripe/services/stripe-subscription-item.service';
+import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+
+@Command({
+  name: 'billing:add-workflow-subscription-item',
+  description: 'Add workflow subscription item to all workspaces subscriptions',
+})
+export class BillingAddWorkflowSubscriptionItemCommand extends ActiveOrSuspendedWorkspacesMigrationCommandRunner {
+  constructor(
+    @InjectRepository(Workspace, 'core')
+    protected readonly workspaceRepository: Repository<Workspace>,
+    protected readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+    @InjectRepository(BillingSubscription, 'core')
+    protected readonly billingSubscriptionRepository: Repository<BillingSubscription>,
+    @InjectRepository(BillingProduct, 'core')
+    protected readonly billingProductRepository: Repository<BillingProduct>,
+    private readonly stripeSubscriptionItemService: StripeSubscriptionItemService,
+  ) {
+    super(workspaceRepository, twentyORMGlobalManager);
+  }
+
+  override async runOnWorkspace({
+    workspaceId,
+    options,
+  }: RunOnWorkspaceArgs): Promise<void> {
+    const billingProducts = await this.billingProductRepository.find({
+      relations: ['billingPrices'],
+    });
+
+    const subscription = await this.billingSubscriptionRepository.findOneOrFail(
+      {
+        where: {
+          workspaceId,
+        },
+        relations: ['billingSubscriptionItems'],
+      },
+    );
+
+    const { basedProduct, basedPrice } =
+      this.getSubscriptionBasedProductAndPrice(billingProducts, subscription);
+
+    const associatedWorkflowMeteredPrice =
+      this.getAssociatedWorkflowMeteredBillingPrice(
+        billingProducts,
+        basedProduct,
+        basedPrice,
+      );
+
+    const hasAlreadyWorkflowSubscriptionItem =
+      subscription.billingSubscriptionItems.some(
+        (item) =>
+          item.stripePriceId === associatedWorkflowMeteredPrice.stripePriceId,
+      );
+
+    if (hasAlreadyWorkflowSubscriptionItem) {
+      this.logger.log(
+        `Workflow subscription item with price ${associatedWorkflowMeteredPrice.stripePriceId} already exists for ${workspaceId}`,
+      );
+
+      return;
+    }
+
+    if (!options.dryRun) {
+      await this.stripeSubscriptionItemService.createSubscriptionItem(
+        subscription.stripeSubscriptionId,
+        associatedWorkflowMeteredPrice.stripePriceId,
+      );
+    }
+
+    this.logger.log(
+      `Adding workflow subscription item with price ${associatedWorkflowMeteredPrice.stripePriceId} to ${workspaceId}`,
+    );
+  }
+
+  private getSubscriptionBasedProductAndPrice(
+    billingProducts: BillingProduct[],
+    subscription: BillingSubscription,
+  ) {
+    const basedProductSubscriptionItem =
+      subscription?.billingSubscriptionItems.find((item) => {
+        return billingProducts.some((product) => {
+          const isBasedProduct =
+            product.stripeProductId === item.stripeProductId &&
+            product.metadata.productKey === BillingProductKey.BASE_PRODUCT;
+
+          return isBasedProduct;
+        });
+      });
+
+    if (!basedProductSubscriptionItem) {
+      throw new Error('Based product subscription item not found');
+    }
+
+    const { stripeProductId, stripePriceId } = basedProductSubscriptionItem;
+
+    const basedProduct = billingProducts.find(
+      (product) => product.stripeProductId === stripeProductId,
+    );
+
+    const basedPrice = basedProduct?.billingPrices.find(
+      (price) => price.stripePriceId === stripePriceId,
+    );
+
+    if (!basedProduct || !basedPrice) {
+      throw new Error(
+        `Based product ${stripeProductId} or price ${stripePriceId} not found`,
+      );
+    }
+
+    return { basedProduct, basedPrice };
+  }
+
+  private getAssociatedWorkflowMeteredBillingPrice(
+    billingProducts: BillingProduct[],
+    basedProduct: BillingProduct,
+    basedPrice: BillingPrice,
+  ) {
+    const associatedMeteredProduct = billingProducts.find(
+      (product) =>
+        product.metadata.planKey === basedProduct.metadata.planKey &&
+        product.metadata.productKey ===
+          BillingProductKey.WORKFLOW_NODE_EXECUTION,
+    );
+
+    const associatedMeteredPrice = associatedMeteredProduct?.billingPrices.find(
+      (price) => price.interval === basedPrice.interval,
+    );
+
+    if (!associatedMeteredPrice) {
+      throw new Error(
+        `Associated metered price for ${basedProduct.name} ${basedPrice.interval} not found`,
+      );
+    }
+
+    return associatedMeteredPrice;
+  }
+}

packages/twenty-server/src/engine/core-modules/billing/stripe/services/stripe-subscription-item.service.ts

@@ -27,4 +27,14 @@ export class StripeSubscriptionItemService {
   async updateSubscriptionItem(stripeItemId: string, quantity: number) {
     await this.stripe.subscriptionItems.update(stripeItemId, { quantity });
   }
+
+  async createSubscriptionItem(
+    stripeSubscriptionId: string,
+    stripePriceId: string,
+  ) {
+    return this.stripe.subscriptionItems.create({
+      subscription: stripeSubscriptionId,
+      price: stripePriceId,
+    });
+  }
 }