From 6951f1002f589fd1393839618d89bd610af3b258 Mon Sep 17 00:00:00 2001 From: DarkSky <25152247+darkskygit@users.noreply.github.com> Date: Mon, 29 Dec 2025 18:54:59 +0800 Subject: [PATCH] feat: improve event handle (#14177) --- .../src/__tests__/payment/service.spec.ts | 84 +++++++++++ .../backend/server/src/base/utils/unit.ts | 6 +- .../server/src/plugins/payment/cron.ts | 73 +++++++++- .../server/src/plugins/payment/event.ts | 13 +- .../src/plugins/payment/manager/user.ts | 100 ++++++++++++- .../server/src/plugins/payment/service.ts | 137 +++++++++++++++++- .../server/src/plugins/payment/types.ts | 3 + .../server/src/plugins/payment/webhook.ts | 57 ++++++++ 8 files changed, 465 insertions(+), 8 deletions(-) diff --git a/packages/backend/server/src/__tests__/payment/service.spec.ts b/packages/backend/server/src/__tests__/payment/service.spec.ts index e1852e494..b96e1ed35 100644 --- a/packages/backend/server/src/__tests__/payment/service.spec.ts +++ b/packages/backend/server/src/__tests__/payment/service.spec.ts @@ -172,6 +172,7 @@ const test = ava as TestFn<{ checkout: { sessions: Sinon.SinonStubbedInstance; }; + invoices: Sinon.SinonStubbedInstance; promotionCodes: Sinon.SinonStubbedInstance; }; }>; @@ -223,6 +224,7 @@ test.before(async t => { prices: Sinon.stub(stripe.prices), subscriptions: Sinon.stub(stripe.subscriptions), subscriptionSchedules: Sinon.stub(stripe.subscriptionSchedules), + invoices: Sinon.stub(stripe.invoices), checkout: { sessions: Sinon.stub(stripe.checkout.sessions), }, @@ -1766,6 +1768,88 @@ test('should be able to update team subscription', async t => { ); }); +test('should suspend on dispute and restore when dispute won', async t => { + const { service, db, stripe, event } = t.context; + + const invoice: Stripe.Invoice = { + id: 'in_dispute_1', + object: 'invoice', + status: 'paid', + customer_email: 'u1@affine.pro', + subscription: 'sub_1', + lines: { + object: 'list', + data: [ + { + id: 'il_1', + object: 'line_item', + amount: 799, + currency: 'usd', + description: '', + discount_amounts: [], + discountable: false, + livemode: false, + metadata: {}, + period: { + start: unixNow() - 60 * 60 * 24, + end: unixNow() + 60 * 60 * 24 * 30, + }, + price: { + ...PRICES[PRO_MONTHLY], + } as any, + quantity: 1, + } as any, + ], + has_more: false, + total_count: 1, + url: '', + }, + } as any; + + stripe.invoices.retrieve.resolves(invoice as any); + stripe.subscriptions.retrieve.resolves(sub as any); + stripe.subscriptions.cancel.resolves(sub as any); + + await service.saveStripeSubscription(sub as any); + + event.emit.resetHistory(); + stripe.subscriptions.cancel.resetHistory(); + + await service.handleRefundedInvoice(invoice.id, 'dispute_open'); + + const removed = await db.subscription.findFirst({ + where: { stripeSubscriptionId: 'sub_1' }, + }); + + t.is(removed, null); + t.true( + event.emit.calledWith('user.subscription.canceled', { + userId: t.context.u1.id, + plan: SubscriptionPlan.Pro, + recurring: SubscriptionRecurring.Monthly, + }) + ); + t.false(stripe.subscriptions.cancel.called); + + event.emit.resetHistory(); + + await service.handleRefundedInvoice(invoice.id, 'dispute_won'); + + const restored = await db.subscription.findFirst({ + where: { stripeSubscriptionId: 'sub_1' }, + }); + + t.truthy(restored); + t.is(restored?.status, SubscriptionStatus.Active); + t.true( + event.emit.calledWith('user.subscription.activated', { + userId: t.context.u1.id, + plan: SubscriptionPlan.Pro, + recurring: SubscriptionRecurring.Monthly, + }) + ); +}); + // NOTE(@forehalo): cancel and resume a team subscription share the same logic with user subscription test.skip('should be able to cancel team subscription', async () => {}); test.skip('should be able to resume team subscription', async () => {}); diff --git a/packages/backend/server/src/base/utils/unit.ts b/packages/backend/server/src/base/utils/unit.ts index 2605c46e9..86984e7b4 100644 --- a/packages/backend/server/src/base/utils/unit.ts +++ b/packages/backend/server/src/base/utils/unit.ts @@ -2,4 +2,8 @@ export const OneKB = 1024; export const OneMB = OneKB * OneKB; export const OneGB = OneKB * OneMB; export const OneMinute = 1000 * 60; -export const OneDay = OneMinute * 60 * 24; +export const OneHour = OneMinute * 60; +export const OneDay = OneHour * 24; +export const OneWeek = OneDay * 7; +export const OneMonth = OneDay * 30; +export const OneYear = OneDay * 365; diff --git a/packages/backend/server/src/plugins/payment/cron.ts b/packages/backend/server/src/plugins/payment/cron.ts index eb2d5d1ce..49a822f07 100644 --- a/packages/backend/server/src/plugins/payment/cron.ts +++ b/packages/backend/server/src/plugins/payment/cron.ts @@ -2,8 +2,10 @@ import { Injectable } from '@nestjs/common'; import { Cron, CronExpression } from '@nestjs/schedule'; import { PrismaClient, Provider } from '@prisma/client'; -import { EventBus, JobQueue, OnJob } from '../../base'; +import { EventBus, JobQueue, OneHour, OnJob } from '../../base'; import { RevenueCatWebhookHandler } from './revenuecat'; +import { SubscriptionService } from './service'; +import { StripeFactory } from './stripe'; import { SubscriptionPlan, SubscriptionRecurring, @@ -16,6 +18,7 @@ declare global { 'nightly.cleanExpiredOnetimeSubscriptions': {}; 'nightly.notifyAboutToExpireWorkspaceSubscriptions': {}; 'nightly.reconcileRevenueCatSubscriptions': {}; + 'nightly.reconcileStripeRefunds': {}; 'nightly.revenuecat.syncUser': { userId: string }; } } @@ -26,7 +29,9 @@ export class SubscriptionCronJobs { private readonly db: PrismaClient, private readonly event: EventBus, private readonly queue: JobQueue, - private readonly rcHandler: RevenueCatWebhookHandler + private readonly rcHandler: RevenueCatWebhookHandler, + private readonly stripeFactory: StripeFactory, + private readonly subscription: SubscriptionService ) {} private getDateRange(after: number, base: number | Date = Date.now()) { @@ -56,6 +61,12 @@ export class SubscriptionCronJobs { { jobId: 'nightly-payment-reconcile-revenuecat-subscriptions' } ); + await this.queue.add( + 'nightly.reconcileStripeRefunds', + {}, + { jobId: 'nightly-payment-reconcile-stripe-refunds' } + ); + // FIXME(@forehalo): the strategy is totally wrong, for monthly plan. redesign required // await this.queue.add( // 'nightly.notifyAboutToExpireWorkspaceSubscriptions', @@ -190,4 +201,62 @@ export class SubscriptionCronJobs { async reconcileRevenueCatSubscriptionOfUser(payload: { userId: string }) { await this.rcHandler.syncAppUser(payload.userId); } + + @OnJob('nightly.reconcileStripeRefunds') + async reconcileStripeRefunds() { + const stripe = this.stripeFactory.stripe; + const since = Math.floor((Date.now() - 36 * OneHour) / 1000); + const seen = new Set(); + + const refunds = await stripe.refunds.list({ + created: { gte: since }, + limit: 100, + expand: ['data.charge'], + }); + + for (const refund of refunds.data) { + const charge = refund.charge; + const invoiceId = + typeof charge !== 'string' + ? typeof charge?.invoice === 'string' + ? charge.invoice + : charge?.invoice?.id + : undefined; + if (invoiceId && !seen.has(invoiceId)) { + seen.add(invoiceId); + await this.subscription.handleRefundedInvoice(invoiceId, 'refund'); + } + } + + const disputes = await stripe.disputes.list({ + created: { gte: since }, + limit: 100, + expand: ['data.charge'], + }); + + for (const dispute of disputes.data) { + const charge = dispute.charge; + const invoiceId = + typeof charge !== 'string' + ? typeof charge?.invoice === 'string' + ? charge.invoice + : charge?.invoice?.id + : undefined; + + if (!invoiceId || seen.has(invoiceId)) { + continue; + } + + seen.add(invoiceId); + + const reason = + dispute.status === 'won' + ? 'dispute_won' + : dispute.status === 'lost' + ? 'dispute_lost' + : ('dispute_open' as const); + + await this.subscription.handleRefundedInvoice(invoiceId, reason); + } + } } diff --git a/packages/backend/server/src/plugins/payment/event.ts b/packages/backend/server/src/plugins/payment/event.ts index 76d962e5b..a5607a30d 100644 --- a/packages/backend/server/src/plugins/payment/event.ts +++ b/packages/backend/server/src/plugins/payment/event.ts @@ -3,7 +3,7 @@ import { Injectable } from '@nestjs/common'; import { EventBus, OnEvent } from '../../base'; import { WorkspaceService } from '../../core/workspaces'; import { Models } from '../../models'; -import { SubscriptionPlan } from './types'; +import { SubscriptionPlan, SubscriptionRecurring } from './types'; @Injectable() export class PaymentEventHandlers { @@ -91,12 +91,23 @@ export class PaymentEventHandlers { async onUserSubscriptionCanceled({ userId, plan, + recurring, }: Events['user.subscription.canceled']) { switch (plan) { case SubscriptionPlan.AI: await this.models.userFeature.remove(userId, 'unlimited_copilot'); break; case SubscriptionPlan.Pro: { + // if user disputed a lifetime plan, we just switch them to free plan directly + if (recurring === SubscriptionRecurring.Lifetime) { + await this.models.userFeature.switchQuota( + userId, + 'free_plan_v1', + 'lifetime subscription canceled' + ); + break; + } + // edge case: when user switch from recurring Pro plan to `Lifetime` plan, // a subscription canceled event will be triggered because `Lifetime` plan is not subscription based const isLifetimeUser = await this.models.userFeature.has( diff --git a/packages/backend/server/src/plugins/payment/manager/user.ts b/packages/backend/server/src/plugins/payment/manager/user.ts index 83e167373..fbec833fa 100644 --- a/packages/backend/server/src/plugins/payment/manager/user.ts +++ b/packages/backend/server/src/plugins/payment/manager/user.ts @@ -1,5 +1,5 @@ import { Injectable } from '@nestjs/common'; -import { PrismaClient, UserStripeCustomer } from '@prisma/client'; +import { PrismaClient, Provider, UserStripeCustomer } from '@prisma/client'; import { omit, pick } from 'lodash-es'; import Stripe from 'stripe'; import { z } from 'zod'; @@ -11,7 +11,9 @@ import { InvalidCheckoutParameters, ManagedByAppStoreOrPlay, Mutex, + OneMonth, OnEvent, + OneYear, SubscriptionAlreadyExists, SubscriptionPlanNotFound, TooManyRequest, @@ -557,6 +559,102 @@ export class UserSubscriptionManager extends SubscriptionManager { return subscription; } + async revokeOnetimeOrLifetime(knownInvoice: KnownStripeInvoice) { + this.assertUserIdExists(knownInvoice.userId); + const { userId, lookupKey } = knownInvoice; + + const subscription = await this.db.subscription.findFirst({ + where: { + targetId: userId, + plan: lookupKey.plan, + provider: Provider.stripe, + }, + }); + + if (!subscription) { + return; + } + + await this.db.subscription.update({ + where: { + id: subscription.id, + }, + data: { + status: SubscriptionStatus.Canceled, + nextBillAt: null, + canceledAt: new Date(), + }, + }); + + this.event.emit('user.subscription.canceled', { + userId, + plan: lookupKey.plan, + recurring: lookupKey.recurring, + }); + } + + async restoreOnetimeOrLifetime(knownInvoice: KnownStripeInvoice) { + this.assertUserIdExists(knownInvoice.userId); + const { userId, lookupKey, stripeInvoice } = knownInvoice; + + const subscription = await this.db.subscription.findFirst({ + where: { + targetId: userId, + plan: lookupKey.plan, + provider: Provider.stripe, + }, + }); + + const start = + stripeInvoice.lines.data[0]?.period?.start ?? + (typeof stripeInvoice.created === 'number' + ? stripeInvoice.created + : Date.now() / 1000); + + let end: Date | null = null; + + if (lookupKey.recurring === SubscriptionRecurring.Lifetime) { + end = null; + } else if (lookupKey.variant === SubscriptionVariant.Onetime) { + const isMonthly = lookupKey.recurring === SubscriptionRecurring.Monthly; + const duration = isMonthly ? OneMonth : OneYear; + end = subscription?.end ?? new Date(start * 1000 + duration); + } else { + end = subscription?.end ?? null; + } + + if (subscription) { + await this.db.subscription.update({ + where: { id: subscription.id }, + data: { + status: SubscriptionStatus.Active, + canceledAt: null, + nextBillAt: null, + start: subscription.start ?? new Date(start * 1000), + end, + }, + }); + } else { + await this.db.subscription.create({ + data: { + targetId: userId, + stripeSubscriptionId: null, + ...lookupKey, + start: new Date(start * 1000), + end, + status: SubscriptionStatus.Active, + nextBillAt: null, + }, + }); + } + + this.event.emit('user.subscription.activated', { + userId, + plan: lookupKey.plan, + recurring: lookupKey.recurring, + }); + } + private async autoPrice(lookupKey: LookupKey, strategy: PriceStrategyStatus) { // auto select ea variant when available if not specified let variant: SubscriptionVariant | null = lookupKey.variant; diff --git a/packages/backend/server/src/plugins/payment/service.ts b/packages/backend/server/src/plugins/payment/service.ts index 9d80cf283..d24f3ad27 100644 --- a/packages/backend/server/src/plugins/payment/service.ts +++ b/packages/backend/server/src/plugins/payment/service.ts @@ -55,6 +55,7 @@ import { SubscriptionPlan, SubscriptionRecurring, SubscriptionStatus, + SubscriptionVariant, } from './types'; export const CheckoutExtraArgs = z.union([ @@ -509,6 +510,128 @@ export class SubscriptionService { await manager.deleteStripeSubscription(knownSubscription); } + async handleRefundedInvoice( + invoiceId: string, + reason: 'refund' | 'dispute_open' | 'dispute_lost' | 'dispute_won' + ) { + try { + const invoice = await this.stripe.invoices.retrieve(invoiceId, { + expand: ['subscription', 'customer', 'lines.data.price'], + }); + + const knownInvoice = await this.parseStripeInvoice(invoice); + + if (!knownInvoice) { + this.logger.warn( + `Skip handling ${reason}: unable to parse invoice ${invoiceId}` + ); + return; + } + + const cancelOnStripe = reason === 'refund' || reason === 'dispute_lost'; + const revokeLocal = + reason === 'refund' || + reason === 'dispute_open' || + reason === 'dispute_lost'; + const restore = reason === 'dispute_won'; + const isOneTimeOrLifetime = + knownInvoice.lookupKey.recurring === SubscriptionRecurring.Lifetime || + knownInvoice.lookupKey.variant === SubscriptionVariant.Onetime; + + if (restore) { + if (invoice.subscription) { + const subscription = + typeof invoice.subscription === 'string' + ? await this.stripe.subscriptions.retrieve(invoice.subscription, { + expand: ['customer'], + }) + : invoice.subscription; + + const knownSubscription = + await this.parseStripeSubscription(subscription); + + if (!knownSubscription) { + this.logger.warn( + `Skip restore: unable to parse subscription ${invoice.subscription} from invoice ${invoiceId}` + ); + return; + } + + await this.saveStripeSubscription(subscription); + return; + } + + if ( + isOneTimeOrLifetime && + (knownInvoice.lookupKey.plan === SubscriptionPlan.Pro || + knownInvoice.lookupKey.plan === SubscriptionPlan.AI) + ) { + await this.userManager.restoreOnetimeOrLifetime(knownInvoice); + } + + return; + } + + if (!revokeLocal) { + return; + } + + if (invoice.subscription) { + const subscription = + typeof invoice.subscription === 'string' + ? await this.stripe.subscriptions.retrieve(invoice.subscription, { + expand: ['customer'], + }) + : invoice.subscription; + + const knownSubscription = + await this.parseStripeSubscription(subscription); + + if (!knownSubscription) { + this.logger.warn( + `Skip handling ${reason}: unable to parse subscription ${invoice.subscription} from invoice ${invoiceId}` + ); + return; + } + + if (cancelOnStripe) { + try { + await this.stripe.subscriptions.cancel( + knownSubscription.stripeSubscription.id + ); + } catch (e) { + this.logger.warn( + `Failed to cancel refunded subscription on Stripe ${knownSubscription.stripeSubscription.id}`, + e + ); + } + } + + const manager = this.select(knownSubscription.lookupKey.plan); + await manager.deleteStripeSubscription(knownSubscription); + return; + } + + if ( + isOneTimeOrLifetime && + (knownInvoice.lookupKey.plan === SubscriptionPlan.Pro || + knownInvoice.lookupKey.plan === SubscriptionPlan.AI) + ) { + await this.userManager.revokeOnetimeOrLifetime(knownInvoice); + return; + } + + this.logger.warn( + `Handled ${reason} for invoice ${invoiceId}, but no local subscription to cancel` + ); + } catch (e) { + this.logger.error( + `Failed to handle ${reason} for invoice ${invoiceId}`, + e + ); + } + } + async getOrCreateCustomer({ userId, userEmail, @@ -618,8 +741,16 @@ export class SubscriptionService { private async parseStripeInvoice( invoice: Stripe.Invoice ): Promise { + const customerEmail = + invoice.customer_email ?? + (typeof invoice.customer !== 'string' && + invoice.customer && + !invoice.customer.deleted + ? (invoice.customer.email ?? null) + : null); + // we can't do anything if we can't recognize the customer - if (!invoice.customer_email) { + if (!customerEmail) { return null; } @@ -638,11 +769,11 @@ export class SubscriptionService { return null; } - const user = await this.models.user.getUserByEmail(invoice.customer_email); + const user = await this.models.user.getUserByEmail(customerEmail); return { userId: user?.id, - userEmail: invoice.customer_email, + userEmail: customerEmail, stripeInvoice: invoice, lookupKey, metadata: invoice.subscription_details?.metadata ?? {}, diff --git a/packages/backend/server/src/plugins/payment/types.ts b/packages/backend/server/src/plugins/payment/types.ts index 4089c8613..2f26b38fe 100644 --- a/packages/backend/server/src/plugins/payment/types.ts +++ b/packages/backend/server/src/plugins/payment/types.ts @@ -88,6 +88,9 @@ declare global { 'stripe.customer.subscription.created': Stripe.CustomerSubscriptionCreatedEvent; 'stripe.customer.subscription.updated': Stripe.CustomerSubscriptionUpdatedEvent; 'stripe.customer.subscription.deleted': Stripe.CustomerSubscriptionDeletedEvent; + 'stripe.charge.refunded': Stripe.ChargeRefundedEvent; + 'stripe.charge.dispute.created': Stripe.ChargeDisputeCreatedEvent; + 'stripe.charge.dispute.closed': Stripe.ChargeDisputeClosedEvent; // RevenueCat integration 'revenuecat.webhook': { diff --git a/packages/backend/server/src/plugins/payment/webhook.ts b/packages/backend/server/src/plugins/payment/webhook.ts index 67bfd553c..7dddb9d34 100644 --- a/packages/backend/server/src/plugins/payment/webhook.ts +++ b/packages/backend/server/src/plugins/payment/webhook.ts @@ -36,6 +36,11 @@ export class StripeWebhook { | Stripe.InvoicePaidEvent ) { const invoice = await this.stripe.invoices.retrieve(event.data.object.id); + + if (invoice.status === 'void' || invoice.status === 'uncollectible') { + await this.service.handleRefundedInvoice(invoice.id, 'refund'); + } + await this.service.saveStripeInvoice(invoice); } @@ -60,4 +65,56 @@ export class StripeWebhook { async onSubscriptionDeleted(event: Stripe.CustomerSubscriptionDeletedEvent) { await this.service.deleteStripeSubscription(event.data.object); } + + private extractInvoiceId(charge: Stripe.Charge) { + return typeof charge.invoice === 'string' + ? charge.invoice + : charge.invoice?.id; + } + + @OnEvent('stripe.charge.refunded') + async onChargeRefunded(event: Stripe.ChargeRefundedEvent) { + const charge = event.data.object; + const invoiceId = this.extractInvoiceId(charge); + + if (invoiceId) { + await this.service.handleRefundedInvoice(invoiceId, 'refund'); + } + } + + @OnEvent('stripe.charge.dispute.created') + async onChargeDisputed(event: Stripe.ChargeDisputeCreatedEvent) { + const ref = event.data.object.charge; + if (!ref) return; + const chargeId = typeof ref === 'string' ? ref : ref.id; + + const charge = await this.stripe.charges.retrieve(chargeId, { + expand: ['invoice'], + }); + + const invoiceId = this.extractInvoiceId(charge); + if (invoiceId) { + await this.service.handleRefundedInvoice(invoiceId, 'dispute_open'); + } + } + + @OnEvent('stripe.charge.dispute.closed') + async onChargeDisputeClosed(event: Stripe.ChargeDisputeClosedEvent) { + const ref = event.data.object.charge; + if (!ref) return; + const chargeId = typeof ref === 'string' ? ref : ref.id; + const status = event.data.object.status; + + const charge = await this.stripe.charges.retrieve(chargeId, { + expand: ['invoice'], + }); + + const invoiceId = this.extractInvoiceId(charge); + + if (invoiceId) { + const reason = + status === 'won' ? 'dispute_won' : ('dispute_lost' as const); + await this.service.handleRefundedInvoice(invoiceId, reason); + } + } }