chore: platform api usage based billing using queue (#15085)

* chore: platform api usage based billing using queue

* fixup! chore: platform api usage based billing using queue

* fixup! fixup! chore: platform api usage based billing using queue

* fixup! Merge branch 'chore-platformapi-billing-queue' of github.com:calcom/cal.com into chore-platformapi-billing-queue

* fixup! fixup! Merge branch 'chore-platformapi-billing-queue' of github.com:calcom/cal.com into chore-platformapi-billing-queue

* fixup! Merge branch 'main' into chore-platformapi-billing-queue

* chore: platform billing by userId

* fix: fix tests requirement team subscirption

* fixup! Merge branch 'main' into chore-platformapi-billing-queue

* fixup! fixup! Merge branch 'main' into chore-platformapi-billing-queue

* fixup! Merge branch 'main' into chore-platformapi-billing-queue

* improve tests logs

* fix: silently close redis and queue

* fixup! Merge branch 'main' into chore-platformapi-billing-queue

* fixup! Merge branch 'main' into chore-platformapi-billing-queue

* fixup! fixup! Merge branch 'main' into chore-platformapi-billing-queue

* fix: upgrade libraries version

apps/api/v2/src/modules/billing/billing.module.ts

@@ -1,3 +1,4 @@
+import { BillingProcessor } from "@/modules/billing/billing.processor";
 import { BillingRepository } from "@/modules/billing/billing.repository";
 import { BillingController } from "@/modules/billing/controllers/billing.controller";
 import { BillingConfigService } from "@/modules/billing/services/billing.config.service";
@@ -6,11 +7,24 @@ import { MembershipsModule } from "@/modules/memberships/memberships.module";
 import { OrganizationsModule } from "@/modules/organizations/organizations.module";
 import { PrismaModule } from "@/modules/prisma/prisma.module";
 import { StripeModule } from "@/modules/stripe/stripe.module";
+import { BullModule } from "@nestjs/bull";
 import { Module } from "@nestjs/common";
 
 @Module({
-  imports: [PrismaModule, StripeModule, MembershipsModule, OrganizationsModule],
-  providers: [BillingConfigService, BillingService, BillingRepository],
+  imports: [
+    PrismaModule,
+    StripeModule,
+    MembershipsModule,
+    OrganizationsModule,
+    BullModule.registerQueue({
+      name: "billing",
+      limiter: {
+        max: 1,
+        duration: 1000,
+      },
+    }),
+  ],
+  providers: [BillingConfigService, BillingService, BillingRepository, BillingProcessor],
   exports: [BillingService, BillingRepository],
   controllers: [BillingController],
 })

apps/api/v2/src/modules/billing/billing.processor.ts

@@ -0,0 +1,91 @@
+import { BillingRepository } from "@/modules/billing/billing.repository";
+import { OrganizationsRepository } from "@/modules/organizations/organizations.repository";
+import { StripeService } from "@/modules/stripe/stripe.service";
+import { Process, Processor } from "@nestjs/bull";
+import { Logger } from "@nestjs/common";
+import { Job } from "bull";
+
+export const INCREMENT_JOB = "increment";
+export const BILLING_QUEUE = "billing";
+export type IncrementJobDataType = {
+  userId: number;
+};
+
+export type DecrementJobDataType = IncrementJobDataType;
+
+@Processor(BILLING_QUEUE)
+export class BillingProcessor {
+  private readonly logger = new Logger(BillingProcessor.name);
+
+  constructor(
+    public readonly stripeService: StripeService,
+    private readonly billingRepository: BillingRepository,
+    private readonly teamsRepository: OrganizationsRepository
+  ) {}
+
+  @Process(INCREMENT_JOB)
+  async handleIncrement(job: Job<IncrementJobDataType>) {
+    const { userId } = job.data;
+    try {
+      // get the platform organization of the managed user
+      const team = await this.teamsRepository.findPlatformOrgFromUserId(userId);
+      const teamId = team.id;
+      if (!team.id) {
+        this.logger.error(`User (${userId}) is not part of the platform organization (${teamId}) `, {
+          teamId,
+          userId,
+        });
+        return;
+      }
+
+      const billingSubscription = await this.billingRepository.getBillingForTeam(teamId);
+      if (!billingSubscription || !billingSubscription?.subscriptionId) {
+        this.logger.error(`Team ${teamId} did not have stripe subscription associated to it`, {
+          teamId,
+        });
+        return;
+      }
+
+      const stripeSubscription = await this.stripeService.stripe.subscriptions.retrieve(
+        billingSubscription.subscriptionId
+      );
+      if (!stripeSubscription?.id) {
+        this.logger.error(`Failed to retrieve stripe subscription (${billingSubscription.subscriptionId})`, {
+          teamId,
+          subscriptionId: billingSubscription.subscriptionId,
+        });
+        return;
+      }
+
+      const meteredItem = stripeSubscription.items.data.find(
+        (item) => item.price?.recurring?.usage_type === "metered"
+      );
+      // no metered item found to increase usage, return early
+      if (!meteredItem) {
+        this.logger.error(`Stripe subscription (${stripeSubscription.id} is not usage based`, {
+          teamId,
+          subscriptionId: stripeSubscription.id,
+        });
+        return;
+      }
+
+      await this.stripeService.stripe.subscriptionItems.createUsageRecord(meteredItem.id, {
+        action: "increment",
+        quantity: 1,
+        timestamp: "now",
+      });
+      this.logger.log("Increased organization usage for subscription", {
+        subscriptionId: billingSubscription.subscriptionId,
+        teamId,
+        userId,
+        itemId: meteredItem.id,
+      });
+    } catch (err) {
+      this.logger.error("Failed to increase usage for Organization", {
+        userId,
+        err,
+      });
+    }
+    return;
+  }
+}

apps/api/v2/src/modules/billing/services/billing.config.service.ts

@@ -3,21 +3,38 @@ import { Injectable } from "@nestjs/common";
 
 @Injectable()
 export class BillingConfigService {
-  private readonly config: Map<PlatformPlan, string>;
+  private readonly config: Map<
+    PlatformPlan,
+    {
+      base: string;
+      overage: string;
+    }
+  >;
 
   constructor() {
-    this.config = new Map<PlatformPlan, string>();
+    this.config = new Map<
+      PlatformPlan,
+      {
+        base: string;
+        overage: string;
+      }
+    >();
 
     const planKeys = Object.keys(PlatformPlan).filter((key) => isNaN(Number(key)));
     for (const key of planKeys) {
-      this.config.set(
-        PlatformPlan[key.toUpperCase() as keyof typeof PlatformPlan],
-        process.env[`STRIPE_PRICE_ID_${key}`] ?? ""
-      );
+      this.config.set(PlatformPlan[key.toUpperCase() as keyof typeof PlatformPlan], {
+        base: process.env[`STRIPE_PRICE_ID_${key}`] ?? "",
+        overage: process.env[`STRIPE_PRICE_ID_${key}_OVERAGE`] ?? "",
+      });
     }
   }
 
-  get(plan: PlatformPlan): string | undefined {
+  get(plan: PlatformPlan):
+    | {
+        base: string;
+        overage: string;
+      }
+    | undefined {
     return this.config.get(plan);
   }
 }

apps/api/v2/src/modules/billing/services/billing.service.ts

@@ -1,16 +1,19 @@
 import { AppConfig } from "@/config/type";
+import { BILLING_QUEUE, INCREMENT_JOB, IncrementJobDataType } from "@/modules/billing/billing.processor";
 import { BillingRepository } from "@/modules/billing/billing.repository";
 import { BillingConfigService } from "@/modules/billing/services/billing.config.service";
 import { PlatformPlan } from "@/modules/billing/types";
 import { OrganizationsRepository } from "@/modules/organizations/organizations.repository";
 import { StripeService } from "@/modules/stripe/stripe.service";
-import { Injectable, InternalServerErrorException, Logger } from "@nestjs/common";
+import { InjectQueue } from "@nestjs/bull";
+import { Injectable, InternalServerErrorException, Logger, OnModuleDestroy } from "@nestjs/common";
 import { ConfigService } from "@nestjs/config";
+import { Queue } from "bull";
 import { DateTime } from "luxon";
 import Stripe from "stripe";
 
 @Injectable()
-export class BillingService {
+export class BillingService implements OnModuleDestroy {
   private logger = new Logger("BillingService");
   private readonly webAppUrl: string;
 
@@ -19,9 +22,10 @@ export class BillingService {
     public readonly stripeService: StripeService,
     private readonly billingRepository: BillingRepository,
     private readonly configService: ConfigService<AppConfig>,
-    private readonly billingConfigService: BillingConfigService
+    private readonly billingConfigService: BillingConfigService,
+    @InjectQueue(BILLING_QUEUE) private readonly billingQueue: Queue
   ) {
-    this.webAppUrl = configService.get("app.baseUrl", { infer: true }) ?? "https://app.cal.com";
+    this.webAppUrl = this.configService.get("app.baseUrl", { infer: true }) ?? "https://app.cal.com";
   }
 
   async getBillingData(teamId: number) {
@@ -58,7 +62,10 @@ export class BillingService {
         customer: customerId,
         line_items: [
           {
-            price: this.billingConfigService.get(plan),
+            price: this.billingConfigService.get(plan)?.overage,
+          },
+          {
+            price: this.billingConfigService.get(plan)?.base,
             quantity: 1,
           },
         ],
@@ -99,46 +106,56 @@ export class BillingService {
     );
   }
 
-  async increaseUsageForTeam(teamId: number) {
-    try {
-      const billingSubscription = await this.billingRepository.getBillingForTeam(teamId);
-      if (!billingSubscription || !billingSubscription?.subscriptionId) {
-        this.logger.error("Team did not have stripe subscription associated to it", {
-          teamId,
-        });
-        return void 0;
-      }
+  /**
+   *
+   * Adds a job to the queue to increment usage of a stripe subscription.
+   * we delay the job until the booking starts.
+   * the delay ensure we can adapt to cancel / reschedule.
+   */
+  async increaseUsageByUserId(
+    userId: number,
+    booking: {
+      uid: string;
+      startTime: Date;
+      fromReschedule?: string | null;
+    }
+  ) {
+    const { uid, startTime, fromReschedule } = booking;
 
-      const stripeSubscription = await this.stripeService.stripe.subscriptions.retrieve(
-        billingSubscription.subscriptionId
-      );
-      const item = stripeSubscription.items.data[0];
-      // legacy plans are licensed, we cannot create usage records against them
-      if (item.price?.recurring?.usage_type === "licensed") {
-        return void 0;
-      }
+    const delay = startTime.getTime() - Date.now();
+    if (fromReschedule) {
+      // cancel the usage increment job for the booking that is being rescheduled
+      await this.cancelUsageByBookingUid(fromReschedule);
+      this.logger.log(`Cancelled usage increment job for rescheduled booking uid: ${fromReschedule}`);
+    }
+    await this.billingQueue.add(
+      INCREMENT_JOB,
+      {
+        userId,
+      } satisfies IncrementJobDataType,
+      { delay: delay > 0 ? delay : 0, jobId: `increment-${uid}`, removeOnComplete: true }
+    );
+    this.logger.log(`Added stripe usage increment job for booking ${uid} and user ${userId}`);
+  }
 
-      await this.stripeService.stripe.subscriptionItems.createUsageRecord(item.id, {
-        action: "increment",
-        quantity: 1,
-        timestamp: "now",
-      });
-    } catch (error) {
-      // don't fail the request, log it.
-      this.logger.error("Failed to increase usage for team", {
-        teamId: teamId,
-        error,
-      });
+  /**
+   *
+   * Cancels the usage increment job for a booking when it is cancelled.
+   * Removing an attendee from a booking does not cancel the usage increment job.
+   */
+  async cancelUsageByBookingUid(bookingUid: string) {
+    const job = await this.billingQueue.getJob(`increment-${bookingUid}`);
+    if (job) {
+      await job.remove();
+      this.logger.log(`Removed increment job for cancelled booking ${bookingUid}`);
     }
   }
 
-  async increaseUsageByClientId(clientId: string) {
-    if (this.configService.get("e2e")) {
-      return void 0;
+  async onModuleDestroy() {
+    try {
+      await this.billingQueue.close();
+    } catch (err) {
+      this.logger.error(err);
     }
-    const team = await this.teamsRepository.findTeamIdFromClientId(clientId);
-    if (!team.id) return Promise.resolve(); // noop resolution.
-
-    return this.increaseUsageForTeam(team?.id);
   }
 }

apps/api/v2/src/modules/email/email.service.ts

@@ -1,6 +1,6 @@
 import { Injectable } from "@nestjs/common";
 
-import { sendSignupToOrganizationEmail, getTranslation } from "@calcom/platform-libraries-0.0.20";
+import { sendSignupToOrganizationEmail, getTranslation } from "@calcom/platform-libraries-0.0.21";
 
 @Injectable()
 export class EmailService {

apps/api/v2/src/modules/oauth-clients/controllers/oauth-clients/oauth-clients.controller.e2e-spec.ts

@@ -163,7 +163,7 @@ describe("OAuth Clients Endpoints", () => {
       await app.init();
     });
 
-    describe("User is not in an organization", () => {
+    describe("User is not part of an organization", () => {
       it(`/GET`, () => {
         return request(app.getHttpServer()).get("/api/v2/oauth-clients").expect(403);
       });
@@ -371,6 +371,7 @@ describe("OAuth Clients Endpoints", () => {
     afterAll(async () => {
       await teamFixtures.delete(org.id);
       await usersFixtures.delete(user.id);
+      await platformBillingRepositoryFixture.deleteSubscriptionForTeam(org.id);
       await app.close();
     });
   });

apps/api/v2/src/modules/oauth-clients/services/oauth-clients-users.service.ts

@@ -7,7 +7,7 @@ import { UsersRepository } from "@/modules/users/users.repository";
 import { BadRequestException, Injectable } from "@nestjs/common";
 import { User } from "@prisma/client";
 
-import { createNewUsersConnectToOrgIfExists, slugify } from "@calcom/platform-libraries-0.0.20";
+import { createNewUsersConnectToOrgIfExists, slugify } from "@calcom/platform-libraries-0.0.21";
 
 @Injectable()
 export class OAuthClientUsersService {

apps/api/v2/src/modules/organizations/organizations.repository.ts

@@ -70,6 +70,26 @@ export class OrganizationsRepository {
       },
     });
   }
+
+  async findPlatformOrgFromUserId(userId: number) {
+    return this.dbRead.prisma.team.findFirstOrThrow({
+      where: {
+        orgProfiles: {
+          some: {
+            userId: userId,
+          },
+        },
+        isPlatform: true,
+        isOrganization: true,
+      },
+      select: {
+        id: true,
+        isPlatform: true,
+        isOrganization: true,
+      },
+    });
+  }
+
   async findOrgUser(organizationId: number, userId: number) {
     return this.dbRead.prisma.user.findUnique({
       where: {

apps/api/v2/src/modules/organizations/services/event-types/organizations-event-types.service.ts

@@ -9,7 +9,7 @@ import { PrismaWriteService } from "@/modules/prisma/prisma-write.service";
 import { UserWithProfile } from "@/modules/users/users.repository";
 import { Injectable, NotFoundException } from "@nestjs/common";
 
-import { createEventType, updateEventType } from "@calcom/platform-libraries-0.0.20";
+import { createEventType, updateEventType } from "@calcom/platform-libraries-0.0.21";
 import {
   CreateTeamEventTypeInput_2024_06_14,
   UpdateTeamEventTypeInput_2024_06_14,

apps/api/v2/src/modules/organizations/services/organizations-users-service.ts

@@ -6,7 +6,7 @@ import { CreateUserInput } from "@/modules/users/inputs/create-user.input";
 import { Injectable, ConflictException } from "@nestjs/common";
 import { plainToInstance } from "class-transformer";
 
-import { createNewUsersConnectToOrgIfExists } from "@calcom/platform-libraries-0.0.20";
+import { createNewUsersConnectToOrgIfExists } from "@calcom/platform-libraries-0.0.21";
 import { Team } from "@calcom/prisma/client";
 
 @Injectable()

apps/api/v2/src/modules/redis/redis.service.ts

@@ -1,11 +1,12 @@
 import { AppConfig } from "@/config/type";
-import { Injectable, OnModuleDestroy } from "@nestjs/common";
+import { Injectable, OnModuleDestroy, Logger } from "@nestjs/common";
 import { ConfigService } from "@nestjs/config";
 import { Redis } from "ioredis";
 
 @Injectable()
 export class RedisService implements OnModuleDestroy {
   public redis: Redis;
+  private readonly logger = new Logger("RedisService");
 
   constructor(readonly configService: ConfigService<AppConfig>) {
     const dbUrl = configService.get<string>("db.redisUrl", { infer: true });
@@ -15,6 +16,10 @@ export class RedisService implements OnModuleDestroy {
   }
 
   async onModuleDestroy() {
-    await this.redis.disconnect();
+    try {
+      await this.redis.quit();
+    } catch (err) {
+      this.logger.error(err);
+    }
   }
 }

apps/api/v2/src/modules/slots/controllers/slots.controller.ts

@@ -5,8 +5,8 @@ import { ApiTags as DocsTags } from "@nestjs/swagger";
 import { Response as ExpressResponse, Request as ExpressRequest } from "express";
 
 import { SUCCESS_STATUS } from "@calcom/platform-constants";
-import { getAvailableSlots } from "@calcom/platform-libraries-0.0.20";
-import type { AvailableSlotsType } from "@calcom/platform-libraries-0.0.20";
+import { getAvailableSlots } from "@calcom/platform-libraries-0.0.21";
+import type { AvailableSlotsType } from "@calcom/platform-libraries-0.0.21";
 import { RemoveSelectedSlotInput, ReserveSlotInput } from "@calcom/platform-types";
 import { ApiResponse, GetAvailableSlotsInput } from "@calcom/platform-types";
 

apps/api/v2/src/modules/slots/slots.repository.ts

@@ -3,7 +3,7 @@ import { PrismaWriteService } from "@/modules/prisma/prisma-write.service";
 import { Injectable } from "@nestjs/common";
 import { DateTime } from "luxon";
 
-import { MINUTES_TO_BOOK } from "@calcom/platform-libraries-0.0.20";
+import { MINUTES_TO_BOOK } from "@calcom/platform-libraries-0.0.21";
 import { ReserveSlotInput } from "@calcom/platform-types";
 
 @Injectable()

apps/api/v2/src/modules/timezones/controllers/timezones.controller.ts

@@ -4,7 +4,7 @@ import { Controller, Get } from "@nestjs/common";
 import { ApiTags as DocsTags } from "@nestjs/swagger";
 
 import { SUCCESS_STATUS } from "@calcom/platform-constants";
-import type { CityTimezones } from "@calcom/platform-libraries-0.0.20";
+import type { CityTimezones } from "@calcom/platform-libraries-0.0.21";
 import { ApiResponse } from "@calcom/platform-types";
 
 @Controller({

apps/api/v2/src/modules/timezones/services/timezones.service.ts

@@ -1,8 +1,8 @@
 import { RedisService } from "@/modules/redis/redis.service";
 import { Injectable } from "@nestjs/common";
 
-import { cityTimezonesHandler } from "@calcom/platform-libraries-0.0.20";
-import type { CityTimezones } from "@calcom/platform-libraries-0.0.20";
+import { cityTimezonesHandler } from "@calcom/platform-libraries-0.0.21";
+import type { CityTimezones } from "@calcom/platform-libraries-0.0.21";
 
 @Injectable()
 export class TimezonesService {