fix: plus sub updated race condition

__tests__/paddle.ts

@@ -1081,7 +1081,7 @@ describe('plus subscription', () => {
     const data = getSubscriptionData({
       user_id: userId,
     });
-    await updateUserSubscription({ event: data, state: true });
+    await updateUserSubscription({ event: data });
 
     const updatedUser = await con
       .getRepository(User)
@@ -1093,6 +1093,120 @@ describe('plus subscription', () => {
     );
   });
 
+  it('should activate subscription when status is trialing', async () => {
+    const userId = 'whp-1';
+    const user = await con.getRepository(User).findOneByOrFail({ id: userId });
+    const isInitiallyPlus = isPlusMember(user.subscriptionFlags?.cycle);
+    expect(isInitiallyPlus).toBe(false);
+
+    const data = getSubscriptionData(
+      {
+        user_id: userId,
+      },
+      'trialing',
+    );
+    await updateUserSubscription({ event: data });
+
+    const updatedUser = await con
+      .getRepository(User)
+      .findOneByOrFail({ id: userId });
+    const isFinallyPlus = isPlusMember(updatedUser.subscriptionFlags?.cycle);
+    expect(isFinallyPlus).toBe(true);
+    expect(updatedUser.subscriptionFlags?.provider).toEqual(
+      SubscriptionProvider.Paddle,
+    );
+    expect(updatedUser.subscriptionFlags?.status).toEqual(
+      SubscriptionStatus.Active,
+    );
+  });
+
+  it('should revoke subscription when status is canceled', async () => {
+    const userId = 'whp-1';
+    await con.getRepository(User).update(
+      { id: userId },
+      {
+        subscriptionFlags: {
+          cycle: SubscriptionCycles.Yearly,
+          provider: SubscriptionProvider.Paddle,
+          status: SubscriptionStatus.Active,
+          subscriptionId: '1',
+        },
+      },
+    );
+
+    const data = getSubscriptionData(
+      {
+        user_id: userId,
+      },
+      'canceled',
+    );
+    await updateUserSubscription({ event: data });
+
+    const updatedUser = await con
+      .getRepository(User)
+      .findOneByOrFail({ id: userId });
+    const isFinallyPlus = isPlusMember(updatedUser.subscriptionFlags?.cycle);
+    expect(isFinallyPlus).toBe(false);
+  });
+
+  it('should revoke subscription when status is paused', async () => {
+    const userId = 'whp-1';
+    await con.getRepository(User).update(
+      { id: userId },
+      {
+        subscriptionFlags: {
+          cycle: SubscriptionCycles.Yearly,
+          provider: SubscriptionProvider.Paddle,
+          status: SubscriptionStatus.Active,
+          subscriptionId: '1',
+        },
+      },
+    );
+
+    const data = getSubscriptionData(
+      {
+        user_id: userId,
+      },
+      'paused',
+    );
+    await updateUserSubscription({ event: data });
+
+    const updatedUser = await con
+      .getRepository(User)
+      .findOneByOrFail({ id: userId });
+    const isFinallyPlus = isPlusMember(updatedUser.subscriptionFlags?.cycle);
+    expect(isFinallyPlus).toBe(false);
+  });
+
+  it('should revoke subscription when status is past_due', async () => {
+    const userId = 'whp-1';
+    await con.getRepository(User).update(
+      { id: userId },
+      {
+        subscriptionFlags: {
+          cycle: SubscriptionCycles.Yearly,
+          provider: SubscriptionProvider.Paddle,
+          status: SubscriptionStatus.Active,
+          subscriptionId: '1',
+        },
+      },
+    );
+
+    const data = getSubscriptionData(
+      {
+        user_id: userId,
+      },
+      'past_due',
+    );
+    await updateUserSubscription({ event: data });
+
+    const updatedUser = await con
+      .getRepository(User)
+      .findOneByOrFail({ id: userId });
+    const isFinallyPlus = isPlusMember(updatedUser.subscriptionFlags?.cycle);
+    expect(isFinallyPlus).toBe(false);
+  });
+
   it('should add an anonymous subscription to the claimable_items table', async () => {
     const mockCustomer = { email: '[email protected]' };
 
@@ -1104,7 +1218,7 @@ describe('plus subscription', () => {
       user_id: undefined,
     });
 
-    await updateUserSubscription({ event: data, state: true });
+    await updateUserSubscription({ event: data });
 
     const claimableItem = await con
       .getRepository(ClaimableItem)
@@ -1152,9 +1266,9 @@ it('should throw an error if the email already has a claimable subscription', as
     },
   });
 
-  await expect(
-    updateUserSubscription({ event: data, state: true }),
-  ).rejects.toThrow(`User already has a claimable subscription`);
+  await expect(updateUserSubscription({ event: data })).rejects.toThrow(
+    `User already has a claimable subscription`,
+  );
 });
 
 it('should not throw an error if the email has claimed a previously claimable subscription', async () => {
@@ -1177,9 +1291,7 @@ it('should not throw an error if the email has claimed a previously claimable su
     user_id: undefined,
   });
 
-  await expect(
-    updateUserSubscription({ event: data, state: true }),
-  ).resolves.not.toThrow();
+  await expect(updateUserSubscription({ event: data })).resolves.not.toThrow();
 });
 
 describe('anonymous subscription', () => {
@@ -1194,7 +1306,7 @@ describe('anonymous subscription', () => {
       user_id: undefined,
     });
 
-    await updateUserSubscription({ event: data, state: true });
+    await updateUserSubscription({ event: data });
 
     const claimableItem = await con
       .getRepository(ClaimableItem)
@@ -1244,9 +1356,9 @@ describe('anonymous subscription', () => {
       },
     });
 
-    await expect(
-      updateUserSubscription({ event: data, state: true }),
-    ).rejects.toThrow(`User already has a claimable subscription`);
+    await expect(updateUserSubscription({ event: data })).rejects.toThrow(
+      `User already has a claimable subscription`,
+    );
   });
 
   it('should not throw an error if the email has claimed a previously claimable subscription', async () => {
@@ -1273,7 +1385,7 @@ describe('anonymous subscription', () => {
     });
 
     await expect(
-      updateUserSubscription({ event: data, state: true }),
+      updateUserSubscription({ event: data }),
     ).resolves.not.toThrow();
   });
 
@@ -1302,7 +1414,7 @@ describe('anonymous subscription', () => {
       'canceled',
     );
 
-    await updateUserSubscription({ event: data, state: false });
+    await updateUserSubscription({ event: data });
 
     const claimableItem = await con
       .getRepository(ClaimableItem)

src/common/paddle/plus/eventHandler.ts

@@ -9,19 +9,27 @@ import { logPaddleAnalyticsEvent, planChanged } from '../index';
 import { AnalyticsEventName } from '../../../integrations/analytics';
 
 export const processPlusPaddleEvent = async (event: EventEntity) => {
-  switch (event?.eventType) {
+  // log all incoming events for monitoring
+  logger.info(
+    {
+      provider: SubscriptionProvider.Paddle,
+      purchaseType: PurchaseType.Plus,
+      occurredAt: event.occurredAt,
+    },
+    event.eventType,
+  );
+
+  switch (event.eventType) {
     case EventName.SubscriptionCreated:
       await updateUserSubscription({
         event,
-        state: true,
       });
 
       break;
     case EventName.SubscriptionCanceled:
       await Promise.all([
         updateUserSubscription({
           event,
-          state: false,
         }),
         logPaddleAnalyticsEvent(event, AnalyticsEventName.CancelSubscription),
       ]);
@@ -31,7 +39,6 @@ export const processPlusPaddleEvent = async (event: EventEntity) => {
       if (didPlanChange) {
         await updateUserSubscription({
           event,
-          state: true,
         });
         await logPaddleAnalyticsEvent(
           event,
@@ -46,12 +53,6 @@ export const processPlusPaddleEvent = async (event: EventEntity) => {
       ]);
       break;
     default:
-      logger.info(
-        {
-          provider: SubscriptionProvider.Paddle,
-          purchaseType: PurchaseType.Plus,
-        },
-        event?.eventType,
-      );
+      break;
   }
 };

src/common/paddle/plus/processing.ts

@@ -1,4 +1,7 @@
-import type { TransactionCompletedEvent } from '@paddle/paddle-node-sdk';
+import type {
+  TransactionCompletedEvent,
+  SubscriptionStatus as PaddleSubscriptionStatus,
+} from '@paddle/paddle-node-sdk';
 import {
   dropClaimableItem,
   extractSubscriptionCycle,
@@ -23,12 +26,31 @@ import {
 import { addMilliseconds } from 'date-fns';
 import { notifyNewPaddlePlusTransaction } from '../slack';
 
+/**
+ * Maps Paddle subscription status to internal SubscriptionStatus.
+ * - 'active' and 'trialing' map to Active (user has access)
+ * - 'canceled', 'past_due', and 'paused' map to Cancelled (access revoked)
+ */
+export const getPaddleSubscriptionStatus = (
+  paddleStatus: PaddleSubscriptionStatus,
+): SubscriptionStatus => {
+  switch (paddleStatus) {
+    case 'active':
+    case 'trialing':
+      return SubscriptionStatus.Active;
+    case 'canceled':
+    case 'past_due':
+    case 'paused':
+      return SubscriptionStatus.Cancelled;
+    default:
+      return SubscriptionStatus.None;
+  }
+};
+
 export const updateUserSubscription = async ({
   event,
-  state,
 }: {
   event: PaddleSubscriptionEvent | undefined;
-  state: boolean;
 }) => {
   if (!event) {
     return;
@@ -41,6 +63,8 @@ export const updateUserSubscription = async ({
   const userId = customData?.user_id;
 
   const subscriptionType = extractSubscriptionCycle(data.items);
+  const mappedStatus = getPaddleSubscriptionStatus(data.status);
+  const isActive = mappedStatus === SubscriptionStatus.Active;
 
   if (!subscriptionType) {
     logger.error(
@@ -54,7 +78,7 @@ export const updateUserSubscription = async ({
     return false;
   }
   if (!userId) {
-    if (state) {
+    if (isActive) {
       await updateClaimableItem(con, data);
     } else {
       await dropClaimableItem(con, data);
@@ -88,17 +112,39 @@ export const updateUserSubscription = async ({
       throw new Error('User already has a StoreKit subscription');
     }
 
+    const currentUpdatedAt = user.subscriptionFlags?.updatedAt
+      ? new Date(user.subscriptionFlags.updatedAt)
+      : new Date(0);
+    const eventUpdatedAt = data?.updatedAt ? new Date(data.updatedAt) : null;
+
+    if (eventUpdatedAt && eventUpdatedAt <= currentUpdatedAt) {
+      logger.warn(
+        {
+          provider: SubscriptionProvider.Paddle,
+          purchaseType: PurchaseType.Plus,
+          userId,
+          eventUpdatedAt,
+          currentUpdatedAt,
+          eventType: event.eventType,
+          subscriptionId: data?.id,
+        },
+        'Stale Paddle subscription event received, skipping',
+      );
+      return;
+    }
+
     await con.getRepository(User).update(
       {
         id: userId,
       },
       {
         subscriptionFlags: updateSubscriptionFlags({
-          cycle: state ? subscriptionType : null,
-          createdAt: state ? data?.startedAt : null,
-          subscriptionId: state ? data?.id : null,
-          provider: state ? SubscriptionProvider.Paddle : null,
-          status: state ? SubscriptionStatus.Active : null,
+          cycle: isActive ? subscriptionType : null,
+          createdAt: isActive ? data?.startedAt : null,
+          subscriptionId: isActive ? data?.id : null,
+          provider: isActive ? SubscriptionProvider.Paddle : null,
+          status: isActive ? mappedStatus : null,
+          updatedAt: data?.updatedAt ?? null,
         }),
       },
     );

src/entity/user/User.ts

@@ -66,6 +66,7 @@ export type UserSubscriptionFlags = Partial<{
   subscriptionId: string;
   cycle: SubscriptionCycles;
   createdAt: Date;
+  updatedAt: Date;
   provider: SubscriptionProvider;
   status: SubscriptionStatus;
 }> &