feat: improve sub sync (#13932)

packages/backend/server/src/plugins/payment/revenuecat/service.ts

@@ -112,6 +112,10 @@ export const Subscription = z.object({
   duration: z.string().nullable(),
 });
 
+const IdentifyUserResponse = z.object({
+  was_created: z.boolean(),
+});
+
 export type Subscription = z.infer<typeof Subscription>;
 type Entitlement = z.infer<typeof zRcV2RawEntitlementItem>;
 type Product = z.infer<typeof zRcV2RawProduct>;
@@ -139,6 +143,41 @@ export class RevenueCatService {
     return id;
   }
 
+  async identifyUser(userId: string, newUserId: string): Promise<boolean> {
+    try {
+      const res = await fetch(
+        `https://api.revenuecat.com/v1/subscribers/identify`,
+        {
+          method: 'POST',
+          body: JSON.stringify({
+            app_user_id: userId,
+            new_app_user_id: newUserId,
+          }),
+          headers: {
+            Authorization: `Bearer ${this.apiKey}`,
+            'Content-Type': 'application/json',
+          },
+        }
+      );
+
+      const json = await res.json();
+      const parsed = IdentifyUserResponse.safeParse(json);
+      if (parsed.success) {
+        return parsed.data.was_created;
+      } else {
+        this.logger.error(
+          `RevenueCat identifyUser parse failed: ${JSON.stringify(
+            parsed.error.format()
+          )}`
+        );
+        return false;
+      }
+    } catch (e: any) {
+      this.logger.error(`RevenueCat identifyUser failed: ${e.message}`);
+      return false;
+    }
+  }
+
   async getProducts(ent: Entitlement): Promise<Product[] | null> {
     if (ent.products?.items && ent.products.items.length > 0) {
       return ent.products.items;
@@ -182,7 +221,10 @@ export class RevenueCatService {
     return null;
   }
 
-  async getCustomerAlias(customerId: string): Promise<string[] | null> {
+  async getCustomerAlias(
+    customerId: string,
+    filterAlias = true
+  ): Promise<string[] | null> {
     const res = await fetch(
       `https://api.revenuecat.com/v2/projects/${this.projectId}/customers/${customerId}/aliases`,
       {
@@ -204,9 +246,12 @@ export class RevenueCatService {
     const customerParsed = zRcV2RawCustomerAliasEnvelope.safeParse(json);
 
     if (customerParsed.success) {
-      return customerParsed.data.items
-        .map(alias => alias.id)
-        .filter(id => !id.startsWith('$RCAnonymousID:'));
+      const customer = customerParsed.data.items.map(alias => alias.id);
+      if (filterAlias) {
+        return customer.filter(id => !id.startsWith('$RCAnonymousID:'));
+      } else {
+        return customer;
+      }
     }
     this.logger.error(
       `RevenueCat customer ${customerId} parse failed: ${JSON.stringify(

packages/backend/server/src/plugins/payment/revenuecat/webhook.ts

@@ -70,8 +70,14 @@ export class RevenueCatWebhookHandler {
       externalRef,
       new Date(Date.now() + 10 * OneMinute) // expire after 10 minutes
     );
+    this.logger.log('Sync subscription by externalRef completed', {
+      appUserId,
+      externalRef,
+      subscriptions: subscriptions.map(s => s.identifier),
+    });
     await this.queue.add('nightly.revenuecat.subscription.refresh', {
       userId: appUserId,
+      externalRef: externalRef,
       startTime: Date.now(),
     });
 
@@ -311,10 +317,9 @@ export class RevenueCatWebhookHandler {
   }
 
   @OnJob('nightly.revenuecat.subscription.refresh.anonymous')
-  async onSubscriptionRefreshAnonymousUser(evt: {
-    externalRef: string;
-    startTime: number;
-  }) {
+  async onSubscriptionRefreshAnonymousUser(
+    evt: Jobs['nightly.revenuecat.subscription.refresh.anonymous']
+  ) {
     if (!this.config.payment.revenuecat?.enabled) return;
     if (Date.now() - evt.startTime > REFRESH_MAX_TIMES) {
       this.logger.warn(
@@ -377,17 +382,47 @@ export class RevenueCatWebhookHandler {
   }
 
   @OnJob('nightly.revenuecat.subscription.refresh')
-  async onSubscriptionRefresh(evt: { userId: string; startTime: number }) {
+  async onSubscriptionRefresh(
+    evt: Jobs['nightly.revenuecat.subscription.refresh']
+  ) {
     if (!this.config.payment.revenuecat?.enabled) return;
-    if (Date.now() - evt.startTime > REFRESH_MAX_TIMES) {
-      this.logger.warn(
-        `RevenueCat subscription refresh timed out for user ${evt.userId}`
-      );
-      return;
-    }
+    const isTimeout = Date.now() - evt.startTime > REFRESH_MAX_TIMES;
+
     const startTime = Date.now();
+    if (isTimeout) {
+      const subs = await this.rc.getSubscriptionByExternalRef(evt.externalRef);
+      const customers = Array.from(
+        new Set(
+          (subs?.map(sub => sub.customerId).filter(Boolean) as string[]) || []
+        )
+      );
+      const customerAliases = await Promise.all(
+        customers.map(custId =>
+          this.rc
+            .getCustomerAlias(custId, false)
+            .then(aliases =>
+              aliases?.length &&
+              aliases.filter(a => !a.startsWith('$RCAnonymousID:')).length === 0
+                ? aliases[0]
+                : null
+            )
+        )
+      );
+      for (const oldUserId of customerAliases) {
+        if (oldUserId) {
+          await this.rc.identifyUser(oldUserId, evt.userId);
+        }
+      }
+    }
     const success = await this.syncAppUser(evt.userId);
     if (success) return;
+    if (isTimeout) {
+      this.logger.warn(`RevenueCat subscription refresh timed out`, {
+        userId: evt.userId,
+        externalRef: evt.externalRef,
+      });
+      return;
+    }
 
     const elapsed = Date.now() - startTime;
     if (elapsed < REFRESH_INTERVAL) {

packages/backend/server/src/plugins/payment/types.ts

@@ -99,6 +99,7 @@ declare global {
   interface Jobs {
     'nightly.revenuecat.subscription.refresh': {
       userId: User['id'];
+      externalRef: string;
       startTime: number;
     };
     'nightly.revenuecat.subscription.refresh.anonymous': {