feat(server): clean up dirty data from legacy version (#15078)

#### PR Dependency Tree * **PR #15078** 👈 This tree was auto-generated by [Charcoal](https://github.com/danerwilliams/charcoal)  ## Summary by CodeRabbit * **New Features** * Persist and replay incoming payment webhooks for reliability. * Track provider-level subscriptions, payment events, and per-target trial usage across providers. * Nightly replay job to reprocess stuck payment events. * Shadow backfill mode and emit-suppression options to control projection/backfill side effects. * Subscriptions now derived from entitlements + provider facts. * **Bug Fixes** * Improved error propagation, retry tracking, and safer owner-grant projection handling. * **Tests** * Added webhook failure/replay, provider integration, entitlement projection, and trial/checkout tests.
2026-07-24 05:48:59 +08:00 · 2026-06-04 16:38:44 +08:00
parent 489702eb66
commit 65c3271beb
24 changed files with 2359 additions and 175 deletions
@@ -23,6 +23,33 @@ test.after.always(async () => {
  await module.close();
 });

+test('payment provider facts migration makes nullable provider identities explicit', t => {
+  const migration = readFileSync(
+    join(
+      process.cwd(),
+      'migrations/20260604000000_payment_provider_facts/migration.sql'
+    ),
+    'utf8'
+  );
+
+  t.regex(
+    migration,
+    /provider_subscriptions_stripe_identity_check[\s\S]*"provider" <> 'stripe' OR "external_subscription_id" IS NOT NULL/
+  );
+  t.regex(
+    migration,
+    /provider_subscriptions_revenuecat_identity_check[\s\S]*"provider" <> 'revenuecat' OR \("iap_store" IS NOT NULL AND "external_ref" IS NOT NULL AND "external_product_id" IS NOT NULL AND "external_customer_id" IS NOT NULL\)/
+  );
+  t.regex(
+    migration,
+    /CREATE UNIQUE INDEX "provider_subscriptions_provider_external_subscription_id_key" ON "provider_subscriptions"\("provider", "external_subscription_id"\)/
+  );
+  t.regex(
+    migration,
+    /CREATE UNIQUE INDEX "provider_subscriptions_revenuecat_external_identity_key" ON "provider_subscriptions"\("provider", "iap_store", "external_ref", "external_product_id", "external_customer_id"\)/
+  );
+});
+
 class TestPermissionProjectionModel extends PermissionProjectionModel {
  constructor(private readonly fakeDb: unknown) {
    super();
@@ -1,13 +1,15 @@
 import { PrismaClient } from '@prisma/client';
 import ava, { TestFn } from 'ava';

-import { CryptoHelper, EventBus } from '../../base';
+import { CryptoHelper, EventBus, JobQueue } from '../../base';
 import { EntitlementService } from '../../core/entitlement';
 import { WorkspacePolicyService } from '../../core/permission';
 import { QuotaStateService } from '../../core/quota/state';
 import { WorkspaceService } from '../../core/workspaces';
 import { Models } from '../../models';
 import { licenseClient, LicenseService } from '../../plugins/license/service';
+import { StripeWebhookController } from '../../plugins/payment/controller';
+import { SubscriptionCronJobs } from '../../plugins/payment/cron';
 import { PaymentEventHandlers } from '../../plugins/payment/event';
 import {
  SubscriptionPlan,
@@ -196,3 +198,269 @@ test('recurring selfhost license activation returns activation projection withou
    },
  ]);
 });
+
+test('stripe webhook persists failed async processing for retry visibility', async t => {
+  const event = {
+    id: 'evt_1',
+    type: 'invoice.paid',
+    created: 1710000000,
+    data: { object: { id: 'in_1' } },
+  };
+  const updates: unknown[] = [];
+  const db = {
+    paymentEvent: {
+      findUnique: async () => null,
+      create: async (input: unknown) => {
+        updates.push(input);
+        return { id: 'payment_event_1' };
+      },
+      updateMany: async (input: unknown) => {
+        updates.push(input);
+        return { count: 1 };
+      },
+      update: async (input: unknown) => {
+        updates.push(input);
+        return {};
+      },
+    },
+  } as unknown as PrismaClient;
+  const controller = new StripeWebhookController(
+    { payment: { stripe: { webhookKey: 'whsec' } } } as never,
+    db,
+    {
+      stripe: {
+        webhooks: {
+          constructEvent: () => event,
+        },
+      },
+    } as never,
+    {
+      emitAsync: async () => {
+        throw new Error('handler failed');
+      },
+    } as unknown as EventBus
+  );
+
+  await controller.handleWebhook({
+    rawBody: Buffer.from('{}'),
+    headers: { 'stripe-signature': 'sig' },
+  } as never);
+  await new Promise(resolve => setImmediate(resolve));
+
+  t.like(updates[0], {
+    data: {
+      provider: 'stripe',
+      eventType: 'invoice.paid',
+      externalEventId: 'evt_1',
+    },
+  });
+  t.deepEqual(
+    updates.slice(1).map(update => (update as { data: unknown }).data),
+    [
+      {
+        processingStatus: 'processing',
+        processingAttempts: { increment: 1 },
+      },
+      {
+        processingStatus: 'failed',
+        lastError: 'handler failed',
+      },
+    ]
+  );
+});
+
+test('stripe webhook skips already processed events', async t => {
+  const event = {
+    id: 'evt_processed',
+    type: 'invoice.paid',
+    created: 1710000000,
+    data: { object: { id: 'in_1' } },
+  };
+  const controller = new StripeWebhookController(
+    { payment: { stripe: { webhookKey: 'whsec' } } } as never,
+    {
+      paymentEvent: {
+        findUnique: async () => ({
+          id: 'payment_event_processed',
+          processingStatus: 'processed',
+        }),
+      },
+    } as unknown as PrismaClient,
+    {
+      stripe: {
+        webhooks: {
+          constructEvent: () => event,
+        },
+      },
+    } as never,
+    {
+      emitAsync: async () => {
+        t.fail('processed event should not be emitted again');
+      },
+    } as unknown as EventBus
+  );
+
+  await controller.handleWebhook({
+    rawBody: Buffer.from('{}'),
+    headers: { 'stripe-signature': 'sig' },
+  } as never);
+  await new Promise(resolve => setImmediate(resolve));
+
+  t.pass();
+});
+
+test('stripe webhook skips events already claimed by another processor', async t => {
+  const event = {
+    id: 'evt_claimed',
+    type: 'invoice.paid',
+    created: 1710000000,
+    data: { object: { id: 'in_1' } },
+  };
+  const controller = new StripeWebhookController(
+    { payment: { stripe: { webhookKey: 'whsec' } } } as never,
+    {
+      paymentEvent: {
+        findUnique: async () => null,
+        create: async () => ({ id: 'payment_event_claimed' }),
+        updateMany: async () => ({ count: 0 }),
+      },
+    } as unknown as PrismaClient,
+    {
+      stripe: {
+        webhooks: {
+          constructEvent: () => event,
+        },
+      },
+    } as never,
+    {
+      emitAsync: async () => {
+        t.fail('unclaimed event should not be emitted');
+      },
+    } as unknown as EventBus
+  );
+
+  await controller.handleWebhook({
+    rawBody: Buffer.from('{}'),
+    headers: { 'stripe-signature': 'sig' },
+  } as never);
+  await new Promise(resolve => setImmediate(resolve));
+
+  t.pass();
+});
+
+test('stripe webhook replay job reprocesses pending events', async t => {
+  const updates: unknown[] = [];
+  const emitted: unknown[] = [];
+  let findManyInput: unknown;
+  const cron = new SubscriptionCronJobs(
+    {
+      paymentEvent: {
+        findMany: async (input: unknown) => {
+          findManyInput = input;
+          return [
+            {
+              id: 'payment_event_1',
+              eventType: 'invoice.paid',
+              metadata: { id: 'in_1' },
+            },
+          ];
+        },
+        updateMany: async (input: unknown) => {
+          updates.push(input);
+          return { count: 1 };
+        },
+        update: async (input: unknown) => {
+          updates.push(input);
+          return {};
+        },
+      },
+    } as unknown as PrismaClient,
+    {
+      emitAsync: async (name: string, payload: unknown) => {
+        emitted.push({ name, payload });
+      },
+    } as unknown as EventBus,
+    {} as unknown as JobQueue,
+    {} as never,
+    {} as never,
+    {} as never,
+    {} as never
+  );
+
+  await cron.replayStripeWebhookEvents();
+
+  t.deepEqual(emitted, [
+    { name: 'stripe.invoice.paid', payload: { id: 'in_1' } },
+  ]);
+  t.like(findManyInput, {
+    where: {
+      OR: [
+        { processingStatus: { in: ['pending', 'failed'] } },
+        { processingStatus: 'processing' },
+      ],
+    },
+  });
+  t.deepEqual((updates[0] as { data: unknown }).data, {
+    processingStatus: 'processing',
+    processingAttempts: { increment: 1 },
+  });
+  t.like((updates[1] as { data: unknown }).data, {
+    processingStatus: 'processed',
+    lastError: null,
+  });
+  t.true(
+    (updates[1] as { data: { processedAt: Date } }).data.processedAt instanceof
+      Date
+  );
+});
+
+test('stripe webhook replay job keeps failed events retryable', async t => {
+  const updates: unknown[] = [];
+  const cron = new SubscriptionCronJobs(
+    {
+      paymentEvent: {
+        findMany: async () => [
+          {
+            id: 'payment_event_1',
+            eventType: 'invoice.paid',
+            metadata: { id: 'in_1' },
+          },
+        ],
+        updateMany: async (input: unknown) => {
+          updates.push(input);
+          return { count: 1 };
+        },
+        update: async (input: unknown) => {
+          updates.push(input);
+          return {};
+        },
+      },
+    } as unknown as PrismaClient,
+    {
+      emitAsync: async () => {
+        throw new Error('handler still failing');
+      },
+    } as unknown as EventBus,
+    {} as unknown as JobQueue,
+    {} as never,
+    {} as never,
+    {} as never,
+    {} as never
+  );
+
+  await cron.replayStripeWebhookEvents();
+
+  t.deepEqual(
+    updates.map(update => (update as { data: unknown }).data),
+    [
+      {
+        processingStatus: 'processing',
+        processingAttempts: { increment: 1 },
+      },
+      {
+        processingStatus: 'failed',
+        lastError: 'handler still failing',
+      },
+    ]
+  );
+});
@@ -27,6 +27,7 @@ import { SubscriptionService } from '../../plugins/payment/service';
 import {
  SubscriptionPlan,
  SubscriptionRecurring,
+  SubscriptionStatus,
 } from '../../plugins/payment/types';
 import { createTestingApp, TestingApp } from '../utils';

@@ -1084,3 +1085,40 @@ test('user subscriptions ignore active rows after their current period ended', a
  });
  t.is(activeAI, null);
 });
+
+test('user subscriptions preserve provider trialing status', async t => {
+  const { db, models, subResolver } = t.context;
+  const trialUser = await models.user.create({
+    email: `${Date.now()}-trial-status@affine.pro`,
+  });
+
+  await db.subscription.create({
+    data: {
+      targetId: trialUser.id,
+      plan: SubscriptionPlan.Pro,
+      provider: 'stripe',
+      status: SubscriptionStatus.Trialing,
+      recurring: SubscriptionRecurring.Yearly,
+      start: new Date('2026-01-01T00:00:00.000Z'),
+      end: new Date('2099-01-01T00:00:00.000Z'),
+      stripeSubscriptionId: 'sub_trialing_status',
+    },
+  });
+  await db.providerSubscription.create({
+    data: {
+      provider: 'stripe',
+      targetType: 'user',
+      targetId: trialUser.id,
+      plan: SubscriptionPlan.Pro,
+      recurring: SubscriptionRecurring.Yearly,
+      status: SubscriptionStatus.Trialing,
+      externalSubscriptionId: 'sub_trialing_status',
+      periodStart: new Date('2026-01-01T00:00:00.000Z'),
+      periodEnd: new Date('2099-01-01T00:00:00.000Z'),
+    },
+  });
+
+  const subscriptions = await subResolver.subscriptions(trialUser, trialUser);
+
+  t.is(subscriptions[0]?.status, SubscriptionStatus.Trialing);
+});
@@ -11,6 +11,7 @@ import { ConfigFactory, ConfigModule } from '../../base/config';
 import { CurrentUser } from '../../core/auth';
 import { AuthService } from '../../core/auth/service';
 import { SubscriptionCronJobs } from '../../plugins/payment/cron';
+import { RevenueCatService } from '../../plugins/payment/revenuecat';
 import { SubscriptionService } from '../../plugins/payment/service';
 import { StripeFactory } from '../../plugins/payment/stripe';
 import {
@@ -135,6 +136,7 @@ const test = ava as TestFn<{
  app: TestingApp;
  service: SubscriptionService;
  event: Sinon.SinonStubbedInstance<EventBus>;
+  revenueCat: Sinon.SinonStubbedInstance<RevenueCatService>;
  stripe: {
    customers: Sinon.SinonStubbedInstance<Stripe.CustomersResource>;
    prices: Sinon.SinonStubbedInstance<Stripe.PricesResource>;
@@ -157,6 +159,11 @@ function getLastCheckoutPrice(checkoutStub: Sinon.SinonStub) {
  };
 }

+function getLastCheckoutParams(checkoutStub: Sinon.SinonStub) {
+  const call = checkoutStub.getCall(checkoutStub.callCount - 1);
+  return call.args[0] as Stripe.Checkout.SessionCreateParams;
+}
+
 test.before(async t => {
  const app = await createTestingApp({
    imports: [
@@ -179,6 +186,7 @@ test.before(async t => {

  t.context.event = app.get(EventBus);
  t.context.service = app.get(SubscriptionService);
+  t.context.revenueCat = Sinon.stub(app.get(RevenueCatService));
  t.context.db = app.get(PrismaClient);
  t.context.app = app;

@@ -209,6 +217,9 @@ test.beforeEach(async t => {
  app.get(ConfigFactory).override({
    payment: {
      showLifetimePrice: true,
+      revenuecat: {
+        enabled: false,
+      },
    },
  });

@@ -240,6 +251,8 @@ test.beforeEach(async t => {

  // @ts-expect-error stub
  stripe.subscriptions.list.resolves({ data: [] });
+  // @ts-expect-error stub
+  stripe.checkout.sessions.create.resolves({ id: 'cs_1' });
 });

 test.after.always(async t => {
@@ -409,6 +422,90 @@ test('should allow checkout after local subscription period ended', async t => {
  });
 });

+test('should reject checkout when stripe already has current subscription', async t => {
+  const { service, u1, stripe } = t.context;
+
+  stripe.subscriptions.list.resolves({
+    data: [
+      {
+        ...sub,
+        id: 'sub_pending_webhook',
+        status: SubscriptionStatus.Active,
+        items: {
+          data: [
+            {
+              // @ts-expect-error stub
+              price: {
+                lookup_key: PRO_YEARLY,
+              },
+            },
+          ],
+        },
+      },
+    ],
+  });
+
+  await t.throwsAsync(
+    () =>
+      service.checkout(
+        {
+          plan: SubscriptionPlan.Pro,
+          recurring: SubscriptionRecurring.Yearly,
+          successCallbackLink: '',
+        },
+        { user: u1 }
+      ),
+    { message: 'You have already subscribed to the pro plan.' }
+  );
+
+  t.false(stripe.checkout.sessions.create.called);
+});
+
+test('should reject checkout when revenuecat already has active subscription', async t => {
+  const { app, revenueCat, service, u1, stripe } = t.context;
+
+  app.get(ConfigFactory).override({
+    payment: {
+      revenuecat: {
+        enabled: true,
+      },
+    },
+  });
+
+  revenueCat.getSubscriptions.resolves([
+    {
+      identifier: 'Pro',
+      isTrial: false,
+      isActive: true,
+      latestPurchaseDate: new Date(),
+      expirationDate: new Date(Date.now() + 100000),
+      customerId: 'rc_customer',
+      productId: 'app.affine.pro.Annual',
+      store: 'app_store',
+      willRenew: true,
+      duration: 'P1Y',
+    },
+  ]);
+
+  await t.throwsAsync(
+    () =>
+      service.checkout(
+        {
+          plan: SubscriptionPlan.Pro,
+          recurring: SubscriptionRecurring.Yearly,
+          successCallbackLink: '',
+        },
+        { user: u1 }
+      ),
+    {
+      message:
+        'This subscription is managed by App Store or Google Play. Please manage it in the corresponding store.',
+    }
+  );
+
+  t.false(stripe.checkout.sessions.create.called);
+});
+
 test('should get correct pro plan price for checking out', async t => {
  const { app, service, u1, stripe } = t.context;
  // monthly
@@ -523,29 +620,15 @@ test('should get correct ai plan price for checking out', async t => {
      price: AI_YEARLY,
      coupon: undefined,
    });
+    t.is(
+      getLastCheckoutParams(stripe.checkout.sessions.create).subscription_data
+        ?.trial_period_days,
+      7
+    );
  }

-  // user with old subscription
+  // user with recorded trial usage
  {
-    stripe.subscriptions.list.resolves({
-      data: [
-        {
-          id: 'sub_1',
-          status: 'canceled',
-          items: {
-            data: [
-              {
-                // @ts-expect-error stub
-                price: {
-                  lookup_key: AI_YEARLY,
-                },
-              },
-            ],
-          },
-        },
-      ],
-    });
-
    await service.checkout(
      {
        plan: SubscriptionPlan.AI,
@@ -559,9 +642,38 @@ test('should get correct ai plan price for checking out', async t => {
      price: AI_YEARLY,
      coupon: undefined,
    });
+    t.is(
+      getLastCheckoutParams(stripe.checkout.sessions.create).subscription_data
+        ?.trial_period_days,
+      undefined
+    );
  }
 });

+test('should record AI trial usage when checkout grants trial', async t => {
+  const { db, service, u1 } = t.context;
+
+  await service.checkout(
+    {
+      plan: SubscriptionPlan.AI,
+      recurring: SubscriptionRecurring.Yearly,
+      successCallbackLink: '',
+    },
+    { user: u1 }
+  );
+
+  const usage = await db.subscriptionTrialUsage.findUnique({
+    where: {
+      targetType_targetId_plan: {
+        targetType: 'user',
+        targetId: u1.id,
+        plan: SubscriptionPlan.AI,
+      },
+    },
+  });
+  t.is(usage?.externalRef, 'cs_1');
+});
+
 test('should apply user coupon for checking out', async t => {
  const { service, u1, stripe } = t.context;

@@ -610,6 +722,22 @@ test('should be able to create subscription', async t => {
    })
  );
  t.is(subInDB?.stripeSubscriptionId, sub.id);
+
+  const providerFact = await db.providerSubscription.findUnique({
+    where: {
+      provider_externalSubscriptionId: {
+        provider: 'stripe',
+        externalSubscriptionId: sub.id,
+      },
+    },
+  });
+  t.like(providerFact, {
+    targetType: 'user',
+    targetId: u1.id,
+    plan: SubscriptionPlan.Pro,
+    recurring: SubscriptionRecurring.Monthly,
+    status: SubscriptionStatus.Active,
+  });
 });

 test('should be able to update subscription', async t => {
@@ -640,6 +768,49 @@ test('should be able to update subscription', async t => {
  t.is(subInDB?.canceledAt?.getTime(), canceledAt * 1000);
 });

+test('should replace old subscription row when stripe creates a new subscription for the same plan', async t => {
+  const { service, db, u1 } = t.context;
+
+  const old = await db.subscription.create({
+    data: {
+      targetId: u1.id,
+      stripeSubscriptionId: 'sub_old',
+      plan: SubscriptionPlan.Pro,
+      recurring: SubscriptionRecurring.Yearly,
+      status: SubscriptionStatus.Canceled,
+      start: new Date('2026-03-26T08:23:57.000Z'),
+      end: new Date('2027-03-26T08:23:57.000Z'),
+    },
+  });
+
+  await service.saveStripeSubscription({
+    ...sub,
+    id: 'sub_new',
+    status: SubscriptionStatus.Active,
+    items: {
+      ...sub.items,
+      data: [
+        {
+          ...sub.items.data[0],
+          // @ts-expect-error stub
+          price: {
+            lookup_key: PRO_YEARLY,
+          },
+        },
+      ],
+    },
+  });
+
+  const subscriptions = await db.subscription.findMany({
+    where: { targetId: u1.id, plan: SubscriptionPlan.Pro },
+  });
+
+  t.is(subscriptions.length, 1);
+  t.is(subscriptions[0].id, old.id);
+  t.is(subscriptions[0].stripeSubscriptionId, 'sub_new');
+  t.is(subscriptions[0].status, SubscriptionStatus.Active);
+});
+
 test('should be able to delete subscription', async t => {
  const { event, service, db, u1 } = t.context;
  await service.saveStripeSubscription(sub);
@@ -662,6 +833,19 @@ test('should be able to delete subscription', async t => {
  });

  t.is(subInDB, null);
+  t.like(
+    await db.providerSubscription.findUnique({
+      where: {
+        provider_externalSubscriptionId: {
+          provider: 'stripe',
+          externalSubscriptionId: sub.id,
+        },
+      },
+    }),
+    {
+      status: SubscriptionStatus.Canceled,
+    }
+  );
 });

 test('should be able to cancel subscription', async t => {
@@ -1118,6 +1302,23 @@ test('should be able to subscribe to lifetime recurring', async t => {
  t.is(subInDB?.recurring, SubscriptionRecurring.Lifetime);
  t.is(subInDB?.status, SubscriptionStatus.Active);
  t.is(subInDB?.stripeSubscriptionId, null);
+
+  const paymentFact = await db.paymentEvent.findUnique({
+    where: {
+      provider_externalEventId: {
+        provider: 'stripe',
+        externalEventId: `stripe_invoice:${lifetimeInvoice.id}`,
+      },
+    },
+  });
+  t.like(paymentFact, {
+    targetType: 'user',
+    targetId: u1.id,
+    plan: SubscriptionPlan.Pro,
+    amount: lifetimeInvoice.total,
+    currency: lifetimeInvoice.currency,
+    processingStatus: 'processed',
+  });
 });

 test('should be able to subscribe to lifetime recurring with old subscription', async t => {