add command to sync plan data from stripe, doing testing (#9177)

**Solves**
[ https://github.com/twentyhq/private-issues/issues/211 ]

**TLDR:**

Add a command that fetches the plans product, meters and price in stripe
and whrites it to the DataBase. For now it fetches only active products.

**In order to test**

- Set IS_BILLING_ENABLED=true
- Run `npx nx database:reset twenty-server` if you don't have the
billing tables in your data base schema
-  run `npx nx run twenty-server:command billing:sync-plans-data -v`

---------

Co-authored-by: Félix Malfait <felix.malfait@gmail.com>

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

@@ -4,6 +4,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 { 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';
 import { BillingEntitlement } from 'src/engine/core-modules/billing/entities/billing-entitlement.entity';
 import { BillingMeter } from 'src/engine/core-modules/billing/entities/billing-meter.entity';
@@ -61,6 +62,7 @@ import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
     BillingWebhookPriceService,
     BillingRestApiExceptionFilter,
     BillingSyncCustomerDataCommand,
+    BillingSyncPlansDataCommand,
   ],
   exports: [
     BillingSubscriptionService,

packages/twenty-server/src/engine/core-modules/billing/commands/billing-sync-plans-data.command.ts

@@ -0,0 +1,160 @@
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Command } from 'nest-commander';
+import Stripe from 'stripe';
+import { Repository } from 'typeorm';
+
+import {
+  BaseCommandOptions,
+  BaseCommandRunner,
+} from 'src/database/commands/base.command';
+import { BillingMeter } from 'src/engine/core-modules/billing/entities/billing-meter.entity';
+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 { StripeService } from 'src/engine/core-modules/billing/stripe/stripe.service';
+import { isStripeValidProductMetadata } from 'src/engine/core-modules/billing/utils/is-stripe-valid-product-metadata.util';
+import { transformStripeMeterDataToMeterRepositoryData } from 'src/engine/core-modules/billing/utils/transform-stripe-meter-data-to-meter-repository-data.util';
+import { transformStripePriceDataToPriceRepositoryData } from 'src/engine/core-modules/billing/utils/transform-stripe-price-data-to-price-repository-data.util';
+import { transformStripeProductDataToProductRepositoryData } from 'src/engine/core-modules/billing/utils/transform-stripe-product-data-to-product-repository-data.util';
+@Command({
+  name: 'billing:sync-plans-data',
+  description:
+    'Fetches from stripe the plans data (meter, product and price) and upserts it into the database',
+})
+export class BillingSyncPlansDataCommand extends BaseCommandRunner {
+  private readonly batchSize = 5;
+  constructor(
+    @InjectRepository(BillingPrice, 'core')
+    private readonly billingPriceRepository: Repository<BillingPrice>,
+    @InjectRepository(BillingProduct, 'core')
+    private readonly billingProductRepository: Repository<BillingProduct>,
+    @InjectRepository(BillingMeter, 'core')
+    private readonly billingMeterRepository: Repository<BillingMeter>,
+    private readonly stripeService: StripeService,
+  ) {
+    super();
+  }
+
+  private async upsertMetersRepositoryData(
+    meters: Stripe.Billing.Meter[],
+    options: BaseCommandOptions,
+  ) {
+    meters.map(async (meter) => {
+      try {
+        if (!options.dryRun) {
+          await this.billingMeterRepository.upsert(
+            transformStripeMeterDataToMeterRepositoryData(meter),
+            {
+              conflictPaths: ['stripeMeterId'],
+            },
+          );
+        }
+        this.logger.log(`Upserted meter: ${meter.id}`);
+      } catch (error) {
+        this.logger.error(`Error upserting meter ${meter.id}: ${error}`);
+      }
+    });
+  }
+
+  private async upsertProductRepositoryData(
+    product: Stripe.Product,
+    options: BaseCommandOptions,
+  ) {
+    try {
+      if (!options.dryRun) {
+        await this.billingProductRepository.upsert(
+          transformStripeProductDataToProductRepositoryData(product),
+          {
+            conflictPaths: ['stripeProductId'],
+          },
+        );
+      }
+      this.logger.log(`Upserted product: ${product.id}`);
+    } catch (error) {
+      this.logger.error(`Error upserting product ${product.id}: ${error}`);
+    }
+  }
+
+  private async getBillingPrices(
+    products: Stripe.Product[],
+    options: BaseCommandOptions,
+  ): Promise<Stripe.Price[][]> {
+    return await Promise.all(
+      products.map(async (product) => {
+        if (!isStripeValidProductMetadata(product.metadata)) {
+          this.logger.log(
+            `Product: ${product.id} purposefully not inserted, invalid metadata format: ${JSON.stringify(
+              product.metadata,
+            )}`,
+          );
+
+          return [];
+        }
+        await this.upsertProductRepositoryData(product, options);
+
+        const prices = await this.stripeService.getPricesByProductId(
+          product.id,
+        );
+
+        this.logger.log(
+          `${prices.length} prices found for product: ${product.id}`,
+        );
+
+        return prices;
+      }),
+    );
+  }
+
+  private async processBillingPricesByProductBatches(
+    products: Stripe.Product[],
+    options: BaseCommandOptions,
+  ) {
+    const prices: Stripe.Price[][] = [];
+
+    for (let start = 0; start < products.length; start += this.batchSize) {
+      const end =
+        start + this.batchSize > products.length
+          ? products.length
+          : start + this.batchSize;
+
+      const batch = products.slice(start, end);
+      const batchPrices = await this.getBillingPrices(batch, options);
+
+      prices.push(...batchPrices);
+      this.logger.log(
+        `Processed batch ${start / this.batchSize + 1} of products`,
+      );
+    }
+
+    return prices;
+  }
+
+  override async executeBaseCommand(
+    passedParams: string[],
+    options: BaseCommandOptions,
+  ): Promise<void> {
+    const billingMeters = await this.stripeService.getAllMeters();
+
+    await this.upsertMetersRepositoryData(billingMeters, options);
+
+    const billingProducts = await this.stripeService.getAllProducts();
+
+    const billingPrices = await this.processBillingPricesByProductBatches(
+      billingProducts,
+      options,
+    );
+    const transformedPrices = billingPrices.flatMap((prices) =>
+      prices.map((price) =>
+        transformStripePriceDataToPriceRepositoryData(price),
+      ),
+    );
+
+    this.logger.log(`Upserting ${transformedPrices.length} transformed prices`);
+
+    if (!options.dryRun) {
+      await this.billingPriceRepository.upsert(transformedPrices, {
+        conflictPaths: ['stripePriceId'],
+      });
+    }
+  }
+}

packages/twenty-server/src/engine/core-modules/billing/enums/billing-plan-key.enum.ts

@@ -1,7 +1,6 @@
 import { registerEnumType } from '@nestjs/graphql';
 
 export enum BillingPlanKey {
-  BASE = 'BASE',
   PRO = 'PRO',
   ENTERPRISE = 'ENTERPRISE',
 }

packages/twenty-server/src/engine/core-modules/billing/services/billing-webhook-product.service.ts

@@ -8,6 +8,7 @@ import { BillingProduct } from 'src/engine/core-modules/billing/entities/billing
 import { BillingPlanKey } from 'src/engine/core-modules/billing/enums/billing-plan-key.enum';
 import { BillingUsageType } from 'src/engine/core-modules/billing/enums/billing-usage-type.enum';
 import { BillingProductMetadata } from 'src/engine/core-modules/billing/types/billing-product-metadata.type';
+import { isStripeValidProductMetadata } from 'src/engine/core-modules/billing/utils/is-stripe-valid-product-metadata.util';
 import { transformStripeProductEventToProductRepositoryData } from 'src/engine/core-modules/billing/utils/transform-stripe-product-event-to-product-repository-data.util';
 @Injectable()
 export class BillingWebhookProductService {
@@ -21,9 +22,7 @@ export class BillingWebhookProductService {
     data: Stripe.ProductCreatedEvent.Data | Stripe.ProductUpdatedEvent.Data,
   ) {
     const metadata = data.object.metadata;
-    const isStripeValidProductMetadata =
+    const productRepositoryData = isStripeValidProductMetadata(metadata)
-      this.isStripeValidProductMetadata(metadata);
-    const productRepositoryData = isStripeValidProductMetadata
       ? {
           ...transformStripeProductEventToProductRepositoryData(data),
           metadata,
@@ -51,24 +50,12 @@ export class BillingWebhookProductService {
   }
 
   isValidBillingPlanKey(planKey?: string) {
-    switch (planKey) {
+    return Object.values(BillingPlanKey).includes(planKey as BillingPlanKey);
-      case BillingPlanKey.BASE:
-        return true;
-      case BillingPlanKey.PRO:
-        return true;
-      default:
-        return false;
-    }
   }
 
   isValidPriceUsageBased(priceUsageBased?: string) {
-    switch (priceUsageBased) {
+    return Object.values(BillingUsageType).includes(
-      case BillingUsageType.METERED:
+      priceUsageBased as BillingUsageType,
-        return true;
+    );
-      case BillingUsageType.LICENSED:
-        return true;
-      default:
-        return false;
-    }
   }
 }

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

@@ -168,10 +168,6 @@ export class StripeService {
     });
   }
 
-  async getCustomer(stripeCustomerId: string) {
-    return await this.stripe.customers.retrieve(stripeCustomerId);
-  }
-
   async getMeter(stripeMeterId: string) {
     return await this.stripe.billing.meters.retrieve(stripeMeterId);
   }
@@ -214,4 +210,24 @@ export class StripeService {
 
     return stripeCustomerId;
   }
+
+  async getAllProducts() {
+    const products = await this.stripe.products.list();
+
+    return products.data;
+  }
+
+  async getPricesByProductId(productId: string) {
+    const prices = await this.stripe.prices.search({
+      query: `product:'${productId}'`,
+    });
+
+    return prices.data;
+  }
+
+  async getAllMeters() {
+    const meters = await this.stripe.billing.meters.list();
+
+    return meters.data;
+  }
 }

packages/twenty-server/src/engine/core-modules/billing/utils/is-stripe-valid-product-metadata.util.ts

@@ -0,0 +1,39 @@
+import Stripe from 'stripe';
+
+import { BillingPlanKey } from 'src/engine/core-modules/billing/enums/billing-plan-key.enum';
+import { BillingUsageType } from 'src/engine/core-modules/billing/enums/billing-usage-type.enum';
+import { BillingProductMetadata } from 'src/engine/core-modules/billing/types/billing-product-metadata.type';
+
+export function isStripeValidProductMetadata(
+  metadata: Stripe.Metadata,
+): metadata is BillingProductMetadata {
+  if (Object.keys(metadata).length === 0) {
+    return true;
+  }
+  const hasBillingPlanKey = isValidBillingPlanKey(metadata.planKey);
+  const hasPriceUsageBased = isValidPriceUsageBased(metadata.priceUsageBased);
+
+  return hasBillingPlanKey && hasPriceUsageBased;
+}
+
+const isValidBillingPlanKey = (planKey?: string) => {
+  switch (planKey) {
+    case BillingPlanKey.ENTERPRISE:
+      return true;
+    case BillingPlanKey.PRO:
+      return true;
+    default:
+      return false;
+  }
+};
+
+const isValidPriceUsageBased = (priceUsageBased?: string) => {
+  switch (priceUsageBased) {
+    case BillingUsageType.METERED:
+      return true;
+    case BillingUsageType.LICENSED:
+      return true;
+    default:
+      return false;
+  }
+};

packages/twenty-server/src/engine/core-modules/billing/utils/transform-stripe-price-data-to-price-repository-data.util.ts

@@ -0,0 +1,104 @@
+import Stripe from 'stripe';
+
+import { BillingPriceBillingScheme } from 'src/engine/core-modules/billing/enums/billing-price-billing-scheme.enum';
+import { BillingPriceTaxBehavior } from 'src/engine/core-modules/billing/enums/billing-price-tax-behavior.enum';
+import { BillingPriceTiersMode } from 'src/engine/core-modules/billing/enums/billing-price-tiers-mode.enum';
+import { BillingPriceType } from 'src/engine/core-modules/billing/enums/billing-price-type.enum';
+import { SubscriptionInterval } from 'src/engine/core-modules/billing/enums/billing-subscription-interval.enum';
+import { BillingUsageType } from 'src/engine/core-modules/billing/enums/billing-usage-type.enum';
+
+export const transformStripePriceDataToPriceRepositoryData = (
+  data: Stripe.Price,
+) => {
+  return {
+    stripePriceId: data.id,
+    active: data.active,
+    stripeProductId: String(data.product),
+    stripeMeterId: data.recurring?.meter,
+    currency: data.currency.toUpperCase(),
+    nickname: data.nickname === null ? undefined : data.nickname,
+    taxBehavior: data.tax_behavior
+      ? getTaxBehavior(data.tax_behavior)
+      : undefined,
+    type: getBillingPriceType(data.type),
+    billingScheme: getBillingPriceBillingScheme(data.billing_scheme),
+    unitAmountDecimal:
+      data.unit_amount_decimal === null ? undefined : data.unit_amount_decimal,
+    unitAmount: data.unit_amount ? Number(data.unit_amount) : undefined,
+    transformQuantity:
+      data.transform_quantity === null ? undefined : data.transform_quantity,
+    usageType: data.recurring?.usage_type
+      ? getBillingPriceUsageType(data.recurring.usage_type)
+      : undefined,
+    interval: data.recurring?.interval
+      ? getBillingPriceInterval(data.recurring.interval)
+      : undefined,
+    currencyOptions:
+      data.currency_options === null ? undefined : data.currency_options,
+    tiers: data.tiers === null ? undefined : data.tiers,
+    tiersMode: data.tiers_mode
+      ? getBillingPriceTiersMode(data.tiers_mode)
+      : undefined,
+    recurring: data.recurring === null ? undefined : data.recurring,
+  };
+};
+
+const getTaxBehavior = (data: Stripe.Price.TaxBehavior) => {
+  switch (data) {
+    case 'exclusive':
+      return BillingPriceTaxBehavior.EXCLUSIVE;
+    case 'inclusive':
+      return BillingPriceTaxBehavior.INCLUSIVE;
+    case 'unspecified':
+      return BillingPriceTaxBehavior.UNSPECIFIED;
+  }
+};
+
+const getBillingPriceType = (data: Stripe.Price.Type) => {
+  switch (data) {
+    case 'one_time':
+      return BillingPriceType.ONE_TIME;
+    case 'recurring':
+      return BillingPriceType.RECURRING;
+  }
+};
+
+const getBillingPriceBillingScheme = (data: Stripe.Price.BillingScheme) => {
+  switch (data) {
+    case 'per_unit':
+      return BillingPriceBillingScheme.PER_UNIT;
+    case 'tiered':
+      return BillingPriceBillingScheme.TIERED;
+  }
+};
+
+const getBillingPriceUsageType = (data: Stripe.Price.Recurring.UsageType) => {
+  switch (data) {
+    case 'licensed':
+      return BillingUsageType.LICENSED;
+    case 'metered':
+      return BillingUsageType.METERED;
+  }
+};
+
+const getBillingPriceTiersMode = (data: Stripe.Price.TiersMode) => {
+  switch (data) {
+    case 'graduated':
+      return BillingPriceTiersMode.GRADUATED;
+    case 'volume':
+      return BillingPriceTiersMode.VOLUME;
+  }
+};
+
+const getBillingPriceInterval = (data: Stripe.Price.Recurring.Interval) => {
+  switch (data) {
+    case 'month':
+      return SubscriptionInterval.Month;
+    case 'day':
+      return SubscriptionInterval.Day;
+    case 'week':
+      return SubscriptionInterval.Week;
+    case 'year':
+      return SubscriptionInterval.Year;
+  }
+};

packages/twenty-server/src/engine/core-modules/billing/utils/transform-stripe-product-data-to-product-repository-data.util.ts

@@ -0,0 +1,21 @@
+import Stripe from 'stripe';
+
+export const transformStripeProductDataToProductRepositoryData = (
+  data: Stripe.Product,
+) => {
+  return {
+    stripeProductId: data.id,
+    name: data.name,
+    active: data.active,
+    description: data.description,
+    images: data.images,
+    marketingFeatures: data.marketing_features,
+    defaultStripePriceId: data.default_price
+      ? String(data.default_price)
+      : undefined,
+    unitLabel: data.unit_label === null ? undefined : data.unit_label,
+    url: data.url === null ? undefined : data.url,
+    taxCode: data.tax_code ? String(data.tax_code) : undefined,
+    metadata: data.metadata,
+  };
+};