diff --git a/apps/sim/lib/auth/auth.ts b/apps/sim/lib/auth/auth.ts index a4aeafc2343..d5cdc948345 100644 --- a/apps/sim/lib/auth/auth.ts +++ b/apps/sim/lib/auth/auth.ts @@ -8,9 +8,9 @@ import * as schema from '@sim/db/schema' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' -import { betterAuth, type User } from 'better-auth' +import { type BetterAuthOptions, betterAuth, type User } from 'better-auth' import { drizzleAdapter } from 'better-auth/adapters/drizzle' -import { APIError, createAuthMiddleware } from 'better-auth/api' +import { APIError, createAuthMiddleware, getSessionFromCtx } from 'better-auth/api' import { nextCookies } from 'better-auth/next-js' import { admin, @@ -34,8 +34,13 @@ import { import { getAccessControlConfig, isEmailBlockedByAccessControl } from '@/lib/auth/access-control' import { createAnonymousSession, ensureAnonymousUserExists } from '@/lib/auth/anonymous' import { getRequestedSignInProviderId, isSignInProviderAllowed } from '@/lib/auth/constants' +import { guardSubscriptionPlanWrites } from '@/lib/auth/stripe-adapter-guard' import { sendPlanWelcomeEmail } from '@/lib/billing' -import { authorizeSubscriptionReference } from '@/lib/billing/authorization' +import { + assertPersonalCheckoutAllowed, + authorizeSubscriptionReference, + isPersonalCheckoutRequest, +} from '@/lib/billing/authorization' import { getOrganizationIdForSubscriptionReference, syncSubscriptionPlan, @@ -46,7 +51,8 @@ import { ensureOrganizationForTeamSubscription, syncSubscriptionUsageLimits, } from '@/lib/billing/organization' -import { isTeam } from '@/lib/billing/plan-helpers' +import { pauseProSubscriptionForOrgCoverage } from '@/lib/billing/organizations/membership' +import { isPro, isTeam } from '@/lib/billing/plan-helpers' import { getPlans, resolvePlanFromStripeSubscription } from '@/lib/billing/plans' import { syncSeatsFromStripeQuantity } from '@/lib/billing/validation/seat-management' import { handleAbandonedCheckout } from '@/lib/billing/webhooks/checkout' @@ -58,7 +64,6 @@ import { handleInvoicePaymentSucceeded, } from '@/lib/billing/webhooks/invoices' import { - handleOrganizationPlanDowngrade, handleSubscriptionCreated, handleSubscriptionDeleted, } from '@/lib/billing/webhooks/subscription' @@ -193,10 +198,13 @@ export const auth = betterAuth({ ...(env.NEXT_PUBLIC_SOCKET_URL ? [env.NEXT_PUBLIC_SOCKET_URL] : []), ...additionalTrustedOrigins, ].filter(Boolean), - database: drizzleAdapter(db, { - provider: 'pg', - schema, - }), + database: (options: BetterAuthOptions) => + guardSubscriptionPlanWrites( + drizzleAdapter(db, { + provider: 'pg', + schema, + })(options) + ), session: { cookieCache: { enabled: true, @@ -903,6 +911,21 @@ export const auth = betterAuth({ } } + /** + * Personal checkout guard. The Stripe plugin's `authorizeReference` + * only runs for organization references (it skips references equal to + * the session user), so duplicate-coverage enforcement for personal + * checkouts lives here: a member of an org with an entitled paid + * subscription must not buy a personal plan on top of it. + */ + if (isBillingEnabled && ctx.path === '/subscription/upgrade') { + const session = await getSessionFromCtx(ctx) + const sessionUserId = session?.user?.id + if (sessionUserId && isPersonalCheckoutRequest(ctx.body ?? {}, sessionUserId)) { + await assertPersonalCheckoutAllowed(sessionUserId) + } + } + return }), }, @@ -3140,8 +3163,21 @@ export const auth = betterAuth({ subscription: { enabled: true, plans: getPlans(), - authorizeReference: async ({ user, referenceId, action }) => { - return await authorizeSubscriptionReference(user.id, referenceId, action) + authorizeReference: async ({ user, referenceId, action }, ctx) => { + const body: unknown = ctx?.body + const requestedPlan = + typeof body === 'object' && + body !== null && + 'plan' in body && + typeof body.plan === 'string' + ? body.plan + : undefined + return await authorizeSubscriptionReference( + user.id, + referenceId, + action, + requestedPlan + ) }, getCheckoutSessionParams: async () => ({ params: { allow_promotion_codes: true }, @@ -3175,11 +3211,16 @@ export const auth = betterAuth({ ) } - await syncSubscriptionPlan(subscription.id, subscription.plan, planFromStripe) + const syncedPlan = await syncSubscriptionPlan( + subscription.id, + subscription.plan, + planFromStripe, + subscription.referenceId + ) const subscriptionForOrg = { ...subscription, - plan: planFromStripe ?? subscription.plan, + plan: syncedPlan ?? subscription.plan, enterpriseOperationId: stripeSubscription.metadata?.enterpriseOperationId ?? null, } @@ -3202,7 +3243,29 @@ export const auth = betterAuth({ throw orgError } - await handleSubscriptionCreated(resolvedSubscription, event.id) + /** + * Transactional fence behind the personal-checkout admission + * guard: if the user joined a paid organization while their + * checkout was in flight, pause the fresh personal Pro at + * period end (same state a paid-org joiner's personal Pro + * enters; restored automatically if they leave the org). + * + * Runs BEFORE the free→paid transition handling: a personal + * subscription born covered is not a free→paid transition — + * the org plan keeps governing the user — so the usage reset + * (which would wipe org-attributed current-period usage) and + * its instrumentation must not run. Gated on `covered`, not + * `paused`, so event retries decide identically even when the + * join path already paused the subscription. + */ + const coveredByOrganization = isPro(resolvedSubscription.plan) + ? (await pauseProSubscriptionForOrgCoverage(resolvedSubscription.referenceId)) + .covered + : false + + if (!coveredByOrganization) { + await handleSubscriptionCreated(resolvedSubscription, event.id) + } await syncSubscriptionUsageLimits(resolvedSubscription) @@ -3238,8 +3301,6 @@ export const auth = betterAuth({ const isUpgradeToTeam = isTeamPlan && !isTeam(subscription.plan) && referenceOrganizationId == null - const effectivePlanForTeamFeatures = planFromStripe ?? subscription.plan - logger.info('[onSubscriptionUpdate] Subscription updated', { subscriptionId: subscription.id, status: subscription.status, @@ -3258,11 +3319,24 @@ export const auth = betterAuth({ ) } - await syncSubscriptionPlan(subscription.id, subscription.plan, planFromStripe) + const syncedPlan = await syncSubscriptionPlan( + subscription.id, + subscription.plan, + planFromStripe, + subscription.referenceId + ) + + /** + * All downstream processing keys off the plan the DB actually + * holds after the sync — a plan write refused by the org/plan + * invariant must not leak the rejected Stripe plan into org + * resolution, seat sync, or usage limits. + */ + const effectivePlanForTeamFeatures = syncedPlan ?? subscription.plan const subscriptionForOrg = { ...subscription, - plan: planFromStripe ?? subscription.plan, + plan: effectivePlanForTeamFeatures, enterpriseOperationId: stripeSubscription.metadata?.enterpriseOperationId ?? null, } @@ -3333,16 +3407,6 @@ export const auth = betterAuth({ error, }) } - } else { - await handleOrganizationPlanDowngrade( - { - id: resolvedSubscription.id, - plan: effectivePlanForTeamFeatures ?? null, - referenceId: resolvedSubscription.referenceId, - status: resolvedSubscription.status ?? null, - }, - event.id - ) } await writeBillingInterval(resolvedSubscription.id, isAnnual ? 'year' : 'month') diff --git a/apps/sim/lib/auth/stripe-adapter-guard.test.ts b/apps/sim/lib/auth/stripe-adapter-guard.test.ts new file mode 100644 index 00000000000..8c8521041ee --- /dev/null +++ b/apps/sim/lib/auth/stripe-adapter-guard.test.ts @@ -0,0 +1,177 @@ +/** + * @vitest-environment node + */ +import { dbChainMock, dbChainMockFns, resetDbChainMock } from '@sim/testing' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +vi.mock('@sim/db', () => dbChainMock) + +import { guardSubscriptionPlanWrites } from '@/lib/auth/stripe-adapter-guard' + +function createBaseAdapter() { + return { + create: vi.fn(async (data: { data: unknown }) => data.data), + update: vi.fn(async (data: { update: unknown }) => data.update), + updateMany: vi.fn(async () => 1), + findOne: vi.fn(), + } +} + +/** + * The guard only relies on create/update/updateMany/findOne; the remaining + * adapter surface passes through the spread untouched. + */ +// double-cast-allowed: test double implements only the adapter subset the guard touches +const asAdapter = (base: ReturnType) => + guardSubscriptionPlanWrites(base as unknown as Parameters[0]) + +const ORG_ROW = { id: 'sub-1', referenceId: 'org-1', plan: 'team_6000' } +const WHERE = [{ field: 'id', value: 'sub-1' }] + +describe('guardSubscriptionPlanWrites', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + it('strips a non-org plan from an update targeting an org-referenced row but keeps the rest', async () => { + const base = createBaseAdapter() + base.findOne.mockResolvedValueOnce(ORG_ROW) + // org existence lookup + dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'org-1' }]) + + const guarded = asAdapter(base) + await guarded.update({ + model: 'subscription', + where: WHERE as never, + update: { plan: 'pro_6000', status: 'active', seats: 2 }, + }) + + expect(base.update).toHaveBeenCalledWith( + expect.objectContaining({ update: { status: 'active', seats: 2 } }) + ) + }) + + it('returns the current row without writing when stripping leaves an empty update', async () => { + const base = createBaseAdapter() + base.findOne.mockResolvedValueOnce(ORG_ROW).mockResolvedValueOnce(ORG_ROW) + dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'org-1' }]) + + const guarded = asAdapter(base) + const result = await guarded.update({ + model: 'subscription', + where: WHERE as never, + update: { plan: 'pro_6000' }, + }) + + expect(result).toEqual(ORG_ROW) + expect(base.update).not.toHaveBeenCalled() + }) + + it('strips the plan when the pre-update lookup finds no row (fail closed)', async () => { + const base = createBaseAdapter() + base.findOne.mockResolvedValueOnce(null) + + const guarded = asAdapter(base) + await guarded.update({ + model: 'subscription', + where: WHERE as never, + update: { plan: 'pro_6000', status: 'active' }, + }) + + expect(base.update).toHaveBeenCalledWith( + expect.objectContaining({ update: { status: 'active' } }) + ) + }) + + it('writes nothing when a plan-only update targets no readable row', async () => { + const base = createBaseAdapter() + base.findOne.mockResolvedValueOnce(null) + + const guarded = asAdapter(base) + const result = await guarded.update({ + model: 'subscription', + where: WHERE as never, + update: { plan: 'pro_6000' }, + }) + + expect(result).toBeNull() + expect(base.update).not.toHaveBeenCalled() + }) + + it('passes through non-org plan updates for user-referenced rows', async () => { + const base = createBaseAdapter() + base.findOne.mockResolvedValueOnce({ id: 'sub-2', referenceId: 'user-1', plan: 'pro_6000' }) + // org existence lookup: user id is not an organization + dbChainMockFns.limit.mockResolvedValueOnce([]) + + const guarded = asAdapter(base) + await guarded.update({ + model: 'subscription', + where: WHERE as never, + update: { plan: 'pro_25000', status: 'active' }, + }) + + expect(base.update).toHaveBeenCalledWith( + expect.objectContaining({ update: { plan: 'pro_25000', status: 'active' } }) + ) + }) + + it('passes through org-plan updates without any row lookup', async () => { + const base = createBaseAdapter() + + const guarded = asAdapter(base) + await guarded.update({ + model: 'subscription', + where: WHERE as never, + update: { plan: 'team_25000', seats: 5 }, + }) + + expect(base.findOne).not.toHaveBeenCalled() + expect(base.update).toHaveBeenCalledWith( + expect.objectContaining({ update: { plan: 'team_25000', seats: 5 } }) + ) + }) + + it('passes through updates on other models untouched', async () => { + const base = createBaseAdapter() + + const guarded = asAdapter(base) + await guarded.update({ + model: 'user', + where: WHERE as never, + update: { plan: 'pro_6000' }, + }) + + expect(base.findOne).not.toHaveBeenCalled() + expect(base.update).toHaveBeenCalled() + }) + + it('rejects creating an org-referenced subscription with a non-org plan', async () => { + const base = createBaseAdapter() + dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'org-1' }]) + + const guarded = asAdapter(base) + await expect( + guarded.create({ + model: 'subscription', + data: { plan: 'pro_6000', referenceId: 'org-1' } as never, + }) + ).rejects.toThrow(/must hold a Team or Enterprise plan/) + + expect(base.create).not.toHaveBeenCalled() + }) + + it('allows creating a personal pro subscription', async () => { + const base = createBaseAdapter() + dbChainMockFns.limit.mockResolvedValueOnce([]) + + const guarded = asAdapter(base) + await guarded.create({ + model: 'subscription', + data: { plan: 'pro_6000', referenceId: 'user-1' } as never, + }) + + expect(base.create).toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/auth/stripe-adapter-guard.ts b/apps/sim/lib/auth/stripe-adapter-guard.ts new file mode 100644 index 00000000000..194af9cd6f3 --- /dev/null +++ b/apps/sim/lib/auth/stripe-adapter-guard.ts @@ -0,0 +1,185 @@ +import { db } from '@sim/db' +import { organization } from '@sim/db/schema' +import { createLogger } from '@sim/logger' +import type { drizzleAdapter } from 'better-auth/adapters/drizzle' +import { eq } from 'drizzle-orm' +import { isOrgPlan } from '@/lib/billing/plan-helpers' + +const logger = createLogger('StripeAdapterGuard') + +type BetterAuthAdapter = ReturnType> + +type SubscriptionWriteSurface = Pick< + BetterAuthAdapter, + 'create' | 'update' | 'updateMany' | 'findOne' | 'findMany' +> + +/** + * The Better Auth Stripe plugin persists webhook state through the raw + * database adapter BEFORE invoking our subscription callbacks — including the + * `plan` column resolved from the Stripe price. That makes the adapter the + * only in-process seam that can enforce the billing invariant that + * organization-referenced subscriptions hold Team/Enterprise plans: by the + * time `syncSubscriptionPlan` runs in a callback, the plugin's write has + * already landed. + * + * Checkout admission blocks user-driven violations; this guard blocks the + * remaining vector — an operator swapping an org subscription onto a personal + * price in the Stripe dashboard. It never repairs state: an invalid `plan` + * write is refused (update: the field is stripped so status/period/seat sync + * still lands; create: the whole insert is rejected) and an error is logged + * so operators fix the price in Stripe. + * + * Transactions are wrapped recursively so writes inside + * `adapter.transaction(...)` callbacks go through the same guard. + */ +export function guardSubscriptionPlanWrites(adapter: BetterAuthAdapter): BetterAuthAdapter { + const guarded: BetterAuthAdapter = { + ...adapter, + ...guardWriteSurface(adapter), + } + + const transaction = adapter.transaction + if (typeof transaction === 'function') { + guarded.transaction = (callback) => + transaction((trx) => callback({ ...trx, ...guardWriteSurface(trx) })) + } + + return guarded +} + +function guardWriteSurface( + adapter: TAdapter +): SubscriptionWriteSurface { + return { + findOne: adapter.findOne, + findMany: adapter.findMany, + create: async (data) => { + if (data.model === 'subscription') { + const values = data.data as Record + const plan = values.plan + const referenceId = values.referenceId + if ( + typeof plan === 'string' && + !isOrgPlan(plan) && + typeof referenceId === 'string' && + (await isOrganizationReference(referenceId)) + ) { + logger.error( + 'Blocked creating an organization-referenced subscription with a non-org plan — fix the plan or reference in Stripe', + { referenceId, rejectedPlan: plan } + ) + throw new Error( + `Organization-referenced subscriptions must hold a Team or Enterprise plan (got '${plan}')` + ) + } + } + return adapter.create(data) + }, + update: async (data) => { + if (data.model === 'subscription' && hasNonOrgPlanWrite(data.update)) { + const row = await adapter.findOne({ + model: 'subscription', + where: data.where, + }) + const sanitized = await stripPlanWhenOrgReferenced( + row ? [row] : [], + data.update as Record + ) + if (sanitized.blockedAll) return row as never + return adapter.update({ ...data, update: sanitized.update as never }) + } + return adapter.update(data) + }, + updateMany: async (data) => { + if (data.model === 'subscription' && hasNonOrgPlanWrite(data.update)) { + const rows = await adapter.findMany({ + model: 'subscription', + where: data.where, + }) + const sanitized = await stripPlanWhenOrgReferenced(rows, data.update) + if (sanitized.blockedAll) return 0 + return adapter.updateMany({ ...data, update: sanitized.update }) + } + return adapter.updateMany(data) + }, + } +} + +interface SubscriptionRowSlice { + id: string + referenceId: string + plan: string +} + +function hasNonOrgPlanWrite(update: unknown): boolean { + if (!update || typeof update !== 'object' || !('plan' in update)) return false + const plan = (update as { plan: unknown }).plan + return typeof plan === 'string' && !isOrgPlan(plan) +} + +async function isOrganizationReference(referenceId: string): Promise { + const [referencedOrganization] = await db + .select({ id: organization.id }) + .from(organization) + .where(eq(organization.id, referenceId)) + .limit(1) + return Boolean(referencedOrganization) +} + +interface SanitizedSubscriptionUpdate { + update: Record + /** True when stripping the invalid plan left nothing to write. */ + blockedAll: boolean +} + +/** + * Strip a non-org `plan` (and its derived `limits`) from an update when ANY + * targeted row is organization-referenced. The remaining fields (status, + * periods, seats, cancellation state) are legitimate Stripe state and still + * sync. Evaluating every targeted row keeps multi-row updates safe: a mixed + * personal/org target set must not leak the invalid plan onto the org rows. + * + * Fails closed on an empty target set: a plan we could not verify against a + * readable row never forwards — the row may be mid-creation by a concurrent + * webhook delivery, and in the sequential no-row case the write was a no-op + * anyway. A legitimate user-referenced plan dropped this way re-syncs via + * `syncSubscriptionPlan` in the same webhook callback. + */ +async function stripPlanWhenOrgReferenced( + rows: SubscriptionRowSlice[], + update: Record +): Promise { + if (rows.length === 0) { + logger.warn( + 'Subscription plan write targeted no readable row; stripping the unverifiable plan', + { rejectedPlan: update.plan } + ) + const { plan: _plan, limits: _limits, ...rest } = update + return { update: rest, blockedAll: Object.keys(rest).length === 0 } + } + + let organizationRow: SubscriptionRowSlice | null = null + for (const row of rows) { + if (await isOrganizationReference(row.referenceId)) { + organizationRow = row + break + } + } + if (!organizationRow) { + return { update, blockedAll: false } + } + + logger.error( + 'Blocked writing a non-org plan onto an organization-referenced subscription — fix the price in Stripe', + { + subscriptionId: organizationRow.id, + organizationId: organizationRow.referenceId, + currentPlan: organizationRow.plan, + rejectedPlan: update.plan, + } + ) + + const { plan: _plan, limits: _limits, ...rest } = update + return { update: rest, blockedAll: Object.keys(rest).length === 0 } +} diff --git a/apps/sim/lib/billing/authorization.test.ts b/apps/sim/lib/billing/authorization.test.ts index 6025302fdb8..002bb2a4537 100644 --- a/apps/sim/lib/billing/authorization.test.ts +++ b/apps/sim/lib/billing/authorization.test.ts @@ -3,10 +3,16 @@ */ import { beforeEach, describe, expect, it, vi } from 'vitest' -const { mockHasPaidSubscription, mockIsOwnerOrAdmin, mockAssertNoUnresolved } = vi.hoisted(() => ({ +const { + mockHasPaidSubscription, + mockIsOwnerOrAdmin, + mockAssertNoUnresolved, + mockGetOrganizationCoverageForMember, +} = vi.hoisted(() => ({ mockHasPaidSubscription: vi.fn(), mockIsOwnerOrAdmin: vi.fn(), mockAssertNoUnresolved: vi.fn(), + mockGetOrganizationCoverageForMember: vi.fn(), })) vi.mock('@sim/db', () => ({ db: {} })) @@ -14,6 +20,9 @@ vi.mock('@/lib/billing', () => ({ hasPaidSubscription: mockHasPaidSubscription } vi.mock('@/lib/billing/core/organization', () => ({ isOrganizationOwnerOrAdmin: mockIsOwnerOrAdmin, })) +vi.mock('@/lib/billing/core/subscription', () => ({ + getOrganizationCoverageForMember: mockGetOrganizationCoverageForMember, +})) vi.mock('@/lib/billing/subscriptions/utils', () => ({ isOrgScopedSubscription: ({ referenceId }: { referenceId: string }, userId: string) => referenceId !== userId, @@ -26,32 +35,127 @@ vi.mock('@/lib/billing/enterprise-outbox', () => { } }) -import { authorizeSubscriptionReference } from '@/lib/billing/authorization' +import { + assertPersonalCheckoutAllowed, + authorizeSubscriptionReference, + isPersonalCheckoutRequest, +} from '@/lib/billing/authorization' import { EnterpriseIssuanceInProgressError } from '@/lib/billing/enterprise-outbox' +describe('isPersonalCheckoutRequest', () => { + it('classifies an explicit self reference as personal regardless of customerType', () => { + expect(isPersonalCheckoutRequest({ referenceId: 'user-1' }, 'user-1')).toBe(true) + expect( + isPersonalCheckoutRequest({ referenceId: 'user-1', customerType: 'organization' }, 'user-1') + ).toBe(true) + }) + + it('classifies an explicit foreign reference as not personal', () => { + expect(isPersonalCheckoutRequest({ referenceId: 'org-1' }, 'user-1')).toBe(false) + }) + + it('defaults to personal without a reference unless customerType selects the organization', () => { + expect(isPersonalCheckoutRequest({}, 'user-1')).toBe(true) + expect(isPersonalCheckoutRequest({ customerType: 'user' }, 'user-1')).toBe(true) + expect(isPersonalCheckoutRequest({ customerType: 'organization' }, 'user-1')).toBe(false) + }) +}) + describe('authorizeSubscriptionReference', () => { beforeEach(() => { vi.clearAllMocks() mockHasPaidSubscription.mockResolvedValue(false) mockAssertNoUnresolved.mockResolvedValue(undefined) mockIsOwnerOrAdmin.mockResolvedValue(true) + mockGetOrganizationCoverageForMember.mockResolvedValue({ status: 'not-covered' }) }) it('blocks an organization checkout while Enterprise issuance is unresolved', async () => { mockAssertNoUnresolved.mockRejectedValueOnce(new EnterpriseIssuanceInProgressError()) await expect( - authorizeSubscriptionReference('owner-1', 'org-1', 'upgrade-subscription') - ).resolves.toBe(false) + authorizeSubscriptionReference('owner-1', 'org-1', 'upgrade-subscription', 'team_6000') + ).rejects.toThrow(/Enterprise plan setup in progress/) expect(mockAssertNoUnresolved).toHaveBeenCalledWith(expect.anything(), 'org-1') expect(mockIsOwnerOrAdmin).not.toHaveBeenCalled() }) it('allows an authorized organization checkout when no paid or reserved entitlement exists', async () => { + await expect( + authorizeSubscriptionReference('owner-1', 'org-1', 'upgrade-subscription', 'team_6000') + ).resolves.toBe(true) + expect(mockIsOwnerOrAdmin).toHaveBeenCalledWith('owner-1', 'org-1') + }) + + it('rejects an organization checkout for a pro plan — org references only hold Team/Enterprise', async () => { + await expect( + authorizeSubscriptionReference('owner-1', 'org-1', 'upgrade-subscription', 'pro_6000') + ).rejects.toThrow('Organizations can only subscribe to Team or Enterprise plans.') + + expect(mockHasPaidSubscription).not.toHaveBeenCalled() + expect(mockIsOwnerOrAdmin).not.toHaveBeenCalled() + }) + + it('rejects an organization checkout when the plan cannot be determined (fail closed)', async () => { await expect( authorizeSubscriptionReference('owner-1', 'org-1', 'upgrade-subscription') + ).rejects.toThrow('Organizations can only subscribe to Team or Enterprise plans.') + }) + + it('blocks an organization checkout when the organization already has an active subscription', async () => { + mockHasPaidSubscription.mockResolvedValueOnce(true) + + await expect( + authorizeSubscriptionReference('owner-1', 'org-1', 'upgrade-subscription', 'team_6000') + ).rejects.toThrow(/already has an active subscription/) + }) + + it('does not apply checkout-only rules to other billing actions', async () => { + await expect( + authorizeSubscriptionReference('owner-1', 'org-1', 'cancel-subscription') ).resolves.toBe(true) + + expect(mockHasPaidSubscription).not.toHaveBeenCalled() expect(mockIsOwnerOrAdmin).toHaveBeenCalledWith('owner-1', 'org-1') }) + + it('allows personal references without invoking org checks', async () => { + await expect( + authorizeSubscriptionReference('user-1', 'user-1', 'upgrade-subscription', 'pro_6000') + ).resolves.toBe(true) + + expect(mockGetOrganizationCoverageForMember).not.toHaveBeenCalled() + expect(mockIsOwnerOrAdmin).not.toHaveBeenCalled() + }) +}) + +describe('assertPersonalCheckoutAllowed', () => { + beforeEach(() => { + vi.clearAllMocks() + mockGetOrganizationCoverageForMember.mockResolvedValue({ status: 'not-covered' }) + }) + + it('allows checkout when the user is not covered by any organization', async () => { + await expect(assertPersonalCheckoutAllowed('user-1')).resolves.toBeUndefined() + }) + + it('rejects checkout when an organization subscription already covers the user', async () => { + mockGetOrganizationCoverageForMember.mockResolvedValueOnce({ + status: 'covered', + organizationId: 'org-1', + }) + + await expect(assertPersonalCheckoutAllowed('user-1')).rejects.toThrow( + /already covered by your organization/ + ) + }) + + it('rejects checkout when coverage cannot be verified (fail closed)', async () => { + mockGetOrganizationCoverageForMember.mockResolvedValueOnce({ status: 'unknown' }) + + await expect(assertPersonalCheckoutAllowed('user-1')).rejects.toThrow( + /could not verify your billing status/ + ) + }) }) diff --git a/apps/sim/lib/billing/authorization.ts b/apps/sim/lib/billing/authorization.ts index 8354f768f8e..68586a1ff69 100644 --- a/apps/sim/lib/billing/authorization.ts +++ b/apps/sim/lib/billing/authorization.ts @@ -1,42 +1,127 @@ import { db } from '@sim/db' import { createLogger } from '@sim/logger' +import { APIError } from 'better-auth/api' import { hasPaidSubscription } from '@/lib/billing' import { isOrganizationOwnerOrAdmin } from '@/lib/billing/core/organization' +import { getOrganizationCoverageForMember } from '@/lib/billing/core/subscription' import { assertNoUnresolvedEnterpriseIssuance, EnterpriseIssuanceInProgressError, } from '@/lib/billing/enterprise-outbox' +import { isOrgPlan } from '@/lib/billing/plan-helpers' import { isOrgScopedSubscription } from '@/lib/billing/subscriptions/utils' const logger = createLogger('BillingAuthorization') +/** + * Classify a `/subscription/upgrade` request as a personal checkout using + * the same reference resolution as the Better Auth Stripe plugin + * (`@better-auth/stripe` 1.6.13): an explicit `referenceId` defines the + * reference (personal iff it is the session user); without one, the + * reference defaults to the user unless `customerType: 'organization'` + * selects the session's active organization. + * + * This mirror exists because the plugin does not expose its resolution and + * skips `authorizeReference` for personal references entirely. Re-verify + * against the plugin's `referenceMiddleware`/`getReferenceId` when + * upgrading better-auth. + */ +export function isPersonalCheckoutRequest( + body: { referenceId?: unknown; customerType?: unknown }, + sessionUserId: string +): boolean { + if (body.referenceId) return body.referenceId === sessionUserId + return body.customerType !== 'organization' +} + +/** + * Guard for personal (user-referenced) checkouts on `/subscription/upgrade`. + * + * A member of an organization with an entitled paid subscription is already + * covered by that org — their usage pools to it and personal Pro + * subscriptions are paused on join — so a personal checkout would bill the + * same human twice. Throws {@link APIError} with a user-facing message; the + * checkout UI surfaces it as-is. + * + * Called from the Better Auth `hooks.before` middleware, NOT from + * `authorizeReference`: the Stripe plugin skips `authorizeReference` + * entirely when the reference is the session user, so this is the only + * enforcement point that personal checkouts actually pass through. + * + * Fails closed: when coverage cannot be determined, the checkout is + * rejected rather than risking a duplicate subscription. + */ +export async function assertPersonalCheckoutAllowed(userId: string): Promise { + const coverage = await getOrganizationCoverageForMember(userId) + + if (coverage.status === 'covered') { + logger.warn( + 'Blocking personal checkout - user is already covered by an organization subscription', + { userId, organizationId: coverage.organizationId } + ) + throw new APIError('FORBIDDEN', { + message: + "You're already covered by your organization's plan, so a personal plan would bill you twice. Manage your plan from the organization's billing settings.", + }) + } + + if (coverage.status === 'unknown') { + logger.error( + 'Blocking personal checkout - could not verify organization coverage; failing closed', + { userId } + ) + throw new APIError('SERVICE_UNAVAILABLE', { + message: 'We could not verify your billing status. Please try again in a moment.', + }) + } +} + /** * Check if a user is authorized to manage billing for a given reference ID. - * Reference ID can be either a user ID (personal subscription) or an - * organization ID (org-scoped subscription — team, enterprise, or a - * `pro_*` plan transferred to an org). + * Only invoked for organization references — the Stripe plugin skips this + * callback when the reference is the session user itself (personal + * checkouts are guarded by {@link assertPersonalCheckoutAllowed} in the + * `hooks.before` middleware instead). * - * This function also performs duplicate subscription validation for - * organizations: - * - Rejects if an organization already has an active subscription (prevents - * duplicates). - * - Personal subscriptions skip this check to allow upgrades. + * For `upgrade-subscription` (checkout) this enforces, each thrown as an + * {@link APIError} with a descriptive message so callers see the real + * reason instead of a generic "Unauthorized": + * - Organizations can only check out Team or Enterprise plans — a `pro_*` + * plan can never become org-referenced. + * - Organizations cannot start a checkout while they already have an + * active subscription (prevents duplicates). + * - Checkout is deferred while an Enterprise issuance is unresolved. */ export async function authorizeSubscriptionReference( userId: string, referenceId: string, - action?: string + action?: string, + plan?: string ): Promise { if (!isOrgScopedSubscription({ referenceId }, userId)) { return true } + if (action === 'upgrade-subscription' && !isOrgPlan(plan)) { + logger.warn('Blocking checkout - organizations can only subscribe to Team or Enterprise', { + userId, + referenceId, + plan: plan ?? null, + }) + throw new APIError('FORBIDDEN', { + message: 'Organizations can only subscribe to Team or Enterprise plans.', + }) + } + if (action === 'upgrade-subscription' && (await hasPaidSubscription(referenceId))) { logger.warn('Blocking checkout - active subscription already exists for organization', { userId, referenceId, }) - return false + throw new APIError('CONFLICT', { + message: + 'This organization already has an active subscription. Manage it from the billing settings.', + }) } if (action === 'upgrade-subscription') { @@ -51,7 +136,10 @@ export async function authorizeSubscriptionReference( userId, referenceId, }) - return false + throw new APIError('CONFLICT', { + message: + 'This organization has an Enterprise plan setup in progress. Please try again in a few minutes or contact support.', + }) } } diff --git a/apps/sim/lib/billing/client/upgrade.ts b/apps/sim/lib/billing/client/upgrade.ts index 9c4e1472519..23729afe105 100644 --- a/apps/sim/lib/billing/client/upgrade.ts +++ b/apps/sim/lib/billing/client/upgrade.ts @@ -181,7 +181,12 @@ export function useSubscriptionUpgrade() { } ) - await betterAuthSubscription.upgrade(finalParams) + const upgradeResult = await betterAuthSubscription.upgrade(finalParams) + if (upgradeResult?.error) { + throw new Error( + upgradeResult.error.message || 'Checkout could not be started. Please try again.' + ) + } if (targetPlan === 'team' && currentSubscriptionRowId && referenceId !== userId) { try { diff --git a/apps/sim/lib/billing/core/billing.ts b/apps/sim/lib/billing/core/billing.ts index 68c78849aa7..8202e809d8d 100644 --- a/apps/sim/lib/billing/core/billing.ts +++ b/apps/sim/lib/billing/core/billing.ts @@ -86,8 +86,11 @@ export async function getOrganizationSubscription( /** * Check if a subscription is scoped to an organization by looking up its * `referenceId` in the organization table. This is the authoritative - * answer — the plan name alone is unreliable because `pro_*` plans can be - * attached to organizations (and we should treat them as org-scoped). + * answer — the plan name alone is unreliable because a team plan can be + * transiently user-referenced between checkout and webhook re-homing. + * (The converse cannot happen: org-referenced subscriptions only ever + * hold Team or Enterprise plans, enforced at checkout authorization and + * in the Stripe plan sync.) * * Use this in server contexts (webhooks, jobs) where we only have the * subscription row, not a user perspective. If you do have a user id, diff --git a/apps/sim/lib/billing/core/subscription.test.ts b/apps/sim/lib/billing/core/subscription.test.ts index ccfceb21889..f5b2c114c08 100644 --- a/apps/sim/lib/billing/core/subscription.test.ts +++ b/apps/sim/lib/billing/core/subscription.test.ts @@ -35,8 +35,11 @@ vi.mock('@/lib/billing/core/plan', () => ({ vi.mock('@/lib/billing/plan-helpers', () => ({ getPlanTierCredits: mockGetPlanTierCredits, isEnterprise: vi.fn().mockReturnValue(false), + isOrgPlan: (plan: string | null | undefined) => + plan === 'enterprise' || plan === 'team' || Boolean(plan?.startsWith('team_')), isPro: vi.fn(), isTeam: vi.fn(), + sqlIsPaid: vi.fn(() => ({ type: 'sqlIsPaid' })), })) vi.mock('@/lib/billing/subscriptions/utils', () => ({ @@ -63,10 +66,12 @@ vi.mock('@/lib/core/config/env-flags', () => ({ vi.mock('@/lib/core/utils/urls', () => urlsMock) import { + getOrganizationCoverageForMember, getOrganizationIdForSubscriptionReference, hasPaidSubscription, hasWorkspaceLiveSyncAccess, isWorkspaceOnEnterprisePlan, + syncSubscriptionPlan, } from '@/lib/billing/core/subscription' describe('hasPaidSubscription', () => { @@ -101,6 +106,79 @@ describe('hasPaidSubscription', () => { }) }) +describe('syncSubscriptionPlan', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('writes the resolved plan for a user-referenced subscription and returns it', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([]) + + await expect(syncSubscriptionPlan('sub-1', 'pro_6000', 'pro_25000', 'user-1')).resolves.toBe( + 'pro_25000' + ) + expect(dbChainMockFns.update).toHaveBeenCalled() + }) + + it('writes a team plan onto an org-referenced subscription without an org lookup', async () => { + await expect(syncSubscriptionPlan('sub-1', 'pro_6000', 'team_6000', 'org-1')).resolves.toBe( + 'team_6000' + ) + expect(dbChainMockFns.limit).not.toHaveBeenCalled() + expect(dbChainMockFns.update).toHaveBeenCalled() + }) + + it('refuses a pro plan on an org-referenced subscription and returns the current plan', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'org-1' }]) + + await expect(syncSubscriptionPlan('sub-1', 'team_6000', 'pro_6000', 'org-1')).resolves.toBe( + 'team_6000' + ) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + }) + + it('returns the current plan when the Stripe plan is unchanged or unresolved', async () => { + await expect(syncSubscriptionPlan('sub-1', 'team_6000', 'team_6000', 'org-1')).resolves.toBe( + 'team_6000' + ) + await expect(syncSubscriptionPlan('sub-1', 'team_6000', null, 'org-1')).resolves.toBe( + 'team_6000' + ) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + }) +}) + +describe('getOrganizationCoverageForMember', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('reports covered with the organization id when an entitled paid org exists', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([{ organizationId: 'org-1' }]) + + await expect(getOrganizationCoverageForMember('user-1')).resolves.toEqual({ + status: 'covered', + organizationId: 'org-1', + }) + }) + + it('reports not-covered when the user has no entitled paid org membership', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([]) + + await expect(getOrganizationCoverageForMember('user-1')).resolves.toEqual({ + status: 'not-covered', + }) + }) + + it('reports unknown on lookup errors so callers fail closed', async () => { + dbChainMockFns.limit.mockRejectedValueOnce(new Error('db unavailable')) + + await expect(getOrganizationCoverageForMember('user-1')).resolves.toEqual({ + status: 'unknown', + }) + }) +}) + describe('getOrganizationIdForSubscriptionReference', () => { beforeEach(() => { vi.clearAllMocks() diff --git a/apps/sim/lib/billing/core/subscription.ts b/apps/sim/lib/billing/core/subscription.ts index 09e7b47a950..5ab950c0eb8 100644 --- a/apps/sim/lib/billing/core/subscription.ts +++ b/apps/sim/lib/billing/core/subscription.ts @@ -10,9 +10,11 @@ import { } from '@/lib/billing/core/plan' import { getPlanTierCredits, + isOrgPlan, isEnterprise as isPlanEnterprise, isPro as isPlanPro, isTeam as isPlanTeam, + sqlIsPaid, } from '@/lib/billing/plan-helpers' import { checkEnterprisePlan, @@ -87,16 +89,53 @@ export async function writeBillingInterval( /** * Sync the subscription's `plan` column to match Stripe. Closes a gap * where plan changes (Pro → Team upgrades, tier swaps) updated price, - * seats, and referenceId at Stripe but left the DB plan stale. Returns - * `true` if a write was issued, `false` if no change was needed. + * seats, and referenceId at Stripe but left the DB plan stale. + * + * Enforces the billing invariant that organization-referenced + * subscriptions only ever hold Team or Enterprise plans: when Stripe + * resolves to a non-org plan (e.g. a Pro price was manually swapped onto + * an org subscription in the Stripe dashboard), the write is refused and + * an error is logged so operators fix the price in Stripe — the DB row + * never becomes an org-scoped Pro subscription. + * + * Returns the plan the DB row holds after the call. Callers must drive all + * downstream processing (org ensure, seat sync, usage limits) from this + * value — never from the raw Stripe plan — so a refused write cannot leak + * the rejected plan into the rest of the webhook handler. + * + * The organization lookup is inlined rather than delegated to + * `isSubscriptionOrgScoped` because that helper lives in `core/billing.ts`, + * which imports this module — delegating would create an import cycle. */ export async function syncSubscriptionPlan( subscriptionId: string, currentPlan: string | null, - planFromStripe: string | null -): Promise { - if (!planFromStripe) return false - if (currentPlan === planFromStripe) return false + planFromStripe: string | null, + referenceId: string +): Promise { + if (!planFromStripe) return currentPlan + if (currentPlan === planFromStripe) return currentPlan + + if (!isOrgPlan(planFromStripe)) { + const [referencedOrganization] = await db + .select({ id: organization.id }) + .from(organization) + .where(eq(organization.id, referenceId)) + .limit(1) + + if (referencedOrganization) { + logger.error( + 'Refusing to sync a non-org plan onto an organization-referenced subscription — fix the price in Stripe', + { + subscriptionId, + organizationId: referenceId, + currentPlan, + rejectedPlan: planFromStripe, + } + ) + return currentPlan + } + } await db .update(subscription) @@ -109,7 +148,7 @@ export async function syncSubscriptionPlan( newPlan: planFromStripe, }) - return true + return planFromStripe } /** @@ -188,6 +227,47 @@ export async function hasPaidSubscription( } } +export type OrganizationCoverageResult = + | { status: 'covered'; organizationId: string } + | { status: 'not-covered' } + | { status: 'unknown' } + +/** + * Check whether an organization already covers this user with an entitled + * paid subscription (the user is a member of the org, any role). + * + * Used to block redundant personal checkouts: a member of a paid org has + * their usage pooled to the org (personal Pro subscriptions are paused on + * join), so buying a personal plan would double-bill the same human. + * + * Returns `'unknown'` on error so callers can fail closed (block checkout + * rather than risk a duplicate subscription) with accurate messaging. + */ +export async function getOrganizationCoverageForMember( + userId: string +): Promise { + try { + const [row] = await db + .select({ organizationId: member.organizationId }) + .from(member) + .innerJoin(subscription, eq(subscription.referenceId, member.organizationId)) + .where( + and( + eq(member.userId, userId), + inArray(subscription.status, ENTITLED_SUBSCRIPTION_STATUSES), + sqlIsPaid(subscription.plan) + ) + ) + .limit(1) + + if (row) return { status: 'covered', organizationId: row.organizationId } + return { status: 'not-covered' } + } catch (error) { + logger.error('Error checking organization coverage for member', { error, userId }) + return { status: 'unknown' } + } +} + export async function getOrganizationIdForSubscriptionReference( referenceId: string ): Promise { diff --git a/apps/sim/lib/billing/organization.test.ts b/apps/sim/lib/billing/organization.test.ts index 8e2a566741b..00709265f20 100644 --- a/apps/sim/lib/billing/organization.test.ts +++ b/apps/sim/lib/billing/organization.test.ts @@ -12,6 +12,7 @@ const { mockAcquireOrganizationMutationLock, mockAssertNoCompetingEnterpriseIssuance, mockGetOrganizationIdForSubscriptionReference, + mockIsSubscriptionOrgScoped, } = vi.hoisted(() => ({ mockCreateOrganizationWithOwner: vi.fn(), mockGetPlanPricing: vi.fn(), @@ -20,12 +21,14 @@ const { mockAcquireOrganizationMutationLock: vi.fn(), mockAssertNoCompetingEnterpriseIssuance: vi.fn(), mockGetOrganizationIdForSubscriptionReference: vi.fn(), + mockIsSubscriptionOrgScoped: vi.fn(), })) vi.mock('@sim/db', () => dbChainMock) vi.mock('@/lib/billing/core/billing', () => ({ getPlanPricing: mockGetPlanPricing, + isSubscriptionOrgScoped: mockIsSubscriptionOrgScoped, })) vi.mock('@/lib/billing/core/subscription', () => ({ @@ -75,8 +78,10 @@ function queueWhereResponses(responses: unknown[][]) { const result = queue.shift() ?? [] const thenable = Promise.resolve(result) as Promise & { limit: ReturnType + for: ReturnType } thenable.limit = vi.fn(() => Promise.resolve(result)) + thenable.for = vi.fn(() => Promise.resolve(result)) return thenable as ReturnType }) } @@ -85,11 +90,11 @@ describe('ensureOrganizationForTeamSubscription', () => { beforeEach(() => { vi.clearAllMocks() resetDbChainMock() - mockGetOrganizationIdForSubscriptionReference.mockResolvedValue(null) + mockIsSubscriptionOrgScoped.mockResolvedValue(false) }) - it('treats existing legacy organization ids as organization references', async () => { - mockGetOrganizationIdForSubscriptionReference.mockResolvedValue('legacy-org-id') + it('treats existing organization references as already homed and takes no write', async () => { + mockIsSubscriptionOrgScoped.mockResolvedValueOnce(true) const result = await ensureOrganizationForTeamSubscription({ id: 'sub-1', @@ -108,6 +113,7 @@ describe('ensureOrganizationForTeamSubscription', () => { }) expect(mockCreateOrganizationWithOwner).not.toHaveBeenCalled() expect(mockAttachOwnedWorkspacesToOrganization).not.toHaveBeenCalled() + expect(dbChainMockFns.update).not.toHaveBeenCalled() expect(mockAcquireOrganizationMutationLock).toHaveBeenCalledWith( expect.anything(), 'legacy-org-id' @@ -120,7 +126,7 @@ describe('ensureOrganizationForTeamSubscription', () => { }) it('allows the authoritative Enterprise webhook to apply its own unresolved issuance', async () => { - mockGetOrganizationIdForSubscriptionReference.mockResolvedValue('org-enterprise') + mockIsSubscriptionOrgScoped.mockResolvedValueOnce(true) const result = await ensureOrganizationForTeamSubscription({ id: 'sub-enterprise', @@ -139,6 +145,39 @@ describe('ensureOrganizationForTeamSubscription', () => { ) }) + it('transfers a user-referenced team subscription onto the org the user administers', async () => { + mockIsSubscriptionOrgScoped.mockResolvedValueOnce(false) + queueWhereResponses([ + // membership lookup: user owns an org + [{ id: 'member-1', organizationId: 'org-owned', role: 'owner' }], + // locked membership re-read inside the transfer transaction + [{ organizationId: 'org-owned', role: 'owner' }], + // locked subscription re-read inside the transfer transaction + [{ id: 'sub-1', referenceId: 'user-1', plan: 'team' }], + // locked organization re-read + [{ id: 'org-owned' }], + // duplicate check: org has no entitled subscription + [], + ]) + + const result = await ensureOrganizationForTeamSubscription({ + id: 'sub-1', + plan: 'team', + referenceId: 'user-1', + status: 'active', + seats: 2, + }) + + expect(result.referenceId).toBe('org-owned') + expect(dbChainMockFns.update).toHaveBeenCalled() + expect(mockAttachOwnedWorkspacesToOrganization).toHaveBeenCalledWith({ + ownerUserId: 'user-1', + organizationId: 'org-owned', + externalMemberPolicy: 'keep-external', + }) + expect(mockCreateOrganizationWithOwner).not.toHaveBeenCalled() + }) + it('keeps org creation, subscription transfer, and workspace attachment on the caller transaction', async () => { queueWhereResponses([[], [], [{ name: 'Owner', email: 'owner@example.com' }]]) mockAttachOwnedWorkspacesToOrganizationTx.mockRejectedValueOnce( diff --git a/apps/sim/lib/billing/organization.ts b/apps/sim/lib/billing/organization.ts index 56a720708f5..a7af195844b 100644 --- a/apps/sim/lib/billing/organization.ts +++ b/apps/sim/lib/billing/organization.ts @@ -4,7 +4,7 @@ import { createLogger } from '@sim/logger' import { isOrgAdminRole } from '@sim/platform-authz/workspace' import { generateId } from '@sim/utils/id' import { and, eq, inArray, ne, sql } from 'drizzle-orm' -import { getPlanPricing } from '@/lib/billing/core/billing' +import { getPlanPricing, isSubscriptionOrgScoped } from '@/lib/billing/core/billing' import { getOrganizationIdForSubscriptionReference } from '@/lib/billing/core/subscription' import { syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage' import { assertNoCompetingEnterpriseIssuance } from '@/lib/billing/enterprise-outbox' @@ -108,25 +108,26 @@ export async function ensureOrganizationForTeamSubscription( return subscription } - const referencedOrganizationId = await getOrganizationIdForSubscriptionReference( - subscription.referenceId - ) - - if (referencedOrganizationId) { + if (await isSubscriptionOrgScoped(subscription)) { await db.transaction(async (tx) => { - await acquireOrganizationMutationLock(tx, referencedOrganizationId) + await acquireOrganizationMutationLock(tx, subscription.referenceId) await assertNoCompetingEnterpriseIssuance( tx, - referencedOrganizationId, + subscription.referenceId, subscription.enterpriseOperationId ?? null ) }) - return { - ...subscription, - referenceId: referencedOrganizationId, - } + return subscription } + /** + * The subscription references a user. Team/Enterprise subscriptions must be + * org-referenced, so fall through to the membership resolution below: it + * transfers the row onto the org the user administers (with duplicate + * checks under the org mutation lock) or creates a new organization. This + * keeps re-homing deterministic in the webhook flow instead of depending on + * a client-side transfer call after checkout. + */ const userId = subscription.referenceId logger.info('Creating organization for team subscription', { @@ -165,10 +166,29 @@ export async function ensureOrganizationForTeamSubscription( subscription.enterpriseOperationId ?? null ) + /** + * Re-verify the pre-transaction membership read under the org + * mutation lock: a concurrent removal or role change must not let a + * stale admin membership authorize the transfer. + */ + const [lockedMembership] = await tx + .select({ organizationId: member.organizationId, role: member.role }) + .from(member) + .where( + and(eq(member.userId, userId), eq(member.organizationId, membership.organizationId)) + ) + .limit(1) + if (!lockedMembership || !isOrgAdminRole(lockedMembership.role)) { + throw new Error( + `User ${userId} no longer administers organization ${membership.organizationId}` + ) + } + const [lockedSub] = await tx .select({ id: subscriptionTable.id, referenceId: subscriptionTable.referenceId, + plan: subscriptionTable.plan, }) .from(subscriptionTable) .where(eq(subscriptionTable.id, subscription.id)) @@ -182,6 +202,12 @@ export async function ensureOrganizationForTeamSubscription( return } + if (!isOrgPlan(lockedSub.plan)) { + throw new Error( + `Subscription ${subscription.id} is no longer a team/enterprise plan (${lockedSub.plan})` + ) + } + const [lockedOrg] = await tx .select({ id: organization.id }) .from(organization) diff --git a/apps/sim/lib/billing/organizations/membership.ts b/apps/sim/lib/billing/organizations/membership.ts index c4a11c8e699..571c652b02e 100644 --- a/apps/sim/lib/billing/organizations/membership.ts +++ b/apps/sim/lib/billing/organizations/membership.ts @@ -312,6 +312,133 @@ export async function restoreUserProSubscription(userId: string): Promise { + const result: PauseProForOrgCoverageResult = { covered: false, paused: false } + + await db.transaction(async (tx) => { + await acquireUserBillingIdentityLock(tx, userId) + + /** + * Coverage is determined before (and independent of) the personal-sub + * lookup: `covered` reports the organization's coverage truth even when + * no entitled personal Pro row exists, exactly as the result contract + * promises. Callers gate free→paid transition handling on it, so a + * personal-sub lookup miss must not read as "not covered". + */ + const memberships = await tx + .select({ organizationId: member.organizationId }) + .from(member) + .where(eq(member.userId, userId)) + if (memberships.length === 0) return + + const organizationSubscriptions = await tx + .select({ plan: subscriptionTable.plan, referenceId: subscriptionTable.referenceId }) + .from(subscriptionTable) + .where( + and( + inArray( + subscriptionTable.referenceId, + memberships.map((membership) => membership.organizationId) + ), + inArray(subscriptionTable.status, ENTITLED_SUBSCRIPTION_STATUSES) + ) + ) + const coveringSubscription = organizationSubscriptions.find((organizationSubscription) => + isPaid(organizationSubscription.plan) + ) + if (!coveringSubscription) return + + result.covered = true + result.organizationId = coveringSubscription.referenceId + + const [personalPro] = await tx + .select() + .from(subscriptionTable) + .where( + and( + eq(subscriptionTable.referenceId, userId), + inArray(subscriptionTable.status, ENTITLED_SUBSCRIPTION_STATUSES), + sqlIsPro(subscriptionTable.plan) + ) + ) + .for('update') + .limit(1) + + if (!personalPro) return + + result.subscriptionId = personalPro.id + + if (personalPro.cancelAtPeriodEnd) return + + await tx + .update(subscriptionTable) + .set({ cancelAtPeriodEnd: true }) + .where(eq(subscriptionTable.id, personalPro.id)) + + if (personalPro.stripeSubscriptionId) { + await enqueueOutboxEvent(tx, OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END, { + stripeSubscriptionId: personalPro.stripeSubscriptionId, + subscriptionId: personalPro.id, + reason: 'covered-by-organization', + }) + } + + result.paused = true + }) + + if (result.paused) { + logger.warn( + 'Paused personal Pro created while covered by an organization subscription (kept until period end, Stripe sync queued)', + { + userId, + subscriptionId: result.subscriptionId, + organizationId: result.organizationId, + } + ) + } + + return result +} + export interface AddMemberParams { userId: string organizationId: string diff --git a/apps/sim/lib/billing/organizations/pause-pro-for-coverage.test.ts b/apps/sim/lib/billing/organizations/pause-pro-for-coverage.test.ts new file mode 100644 index 00000000000..4c8e50dedea --- /dev/null +++ b/apps/sim/lib/billing/organizations/pause-pro-for-coverage.test.ts @@ -0,0 +1,145 @@ +/** + * @vitest-environment node + */ +import { dbChainMock, dbChainMockFns, resetDbChainMock } from '@sim/testing' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockEnqueueOutboxEvent } = vi.hoisted(() => ({ + mockEnqueueOutboxEvent: vi.fn(), +})) + +vi.mock('@sim/db', () => dbChainMock) +vi.mock('@/lib/billing/storage/payer-transfer', () => ({ + changeOrganizationWorkspaceBilledAccountsInTx: vi.fn(), + changeWorkspaceStoragePayerInTx: vi.fn(), + changeWorkspaceStoragePayersInTx: vi.fn(), +})) +vi.mock('@/lib/core/outbox/service', () => ({ + enqueueOutboxEvent: mockEnqueueOutboxEvent, +})) + +import { pauseProSubscriptionForOrgCoverage } from '@/lib/billing/organizations/membership' +import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers' + +const ACTIVE_PERSONAL_PRO = { + id: 'sub-personal', + plan: 'pro_6000', + referenceId: 'user-1', + status: 'active', + cancelAtPeriodEnd: false, + stripeSubscriptionId: 'stripe-sub-personal', +} + +/** + * Queues per-`where()` results for the three reads in the pause path: + * personal sub (`.for('update').limit(1)`), memberships (awaited where), + * and org subscriptions (awaited where). + */ +function queueWhereResponses(responses: unknown[][]) { + const queue = [...responses] + dbChainMockFns.where.mockImplementation(() => { + const result = queue.shift() ?? [] + const limit = vi.fn(() => Promise.resolve(result)) + const forResult = Promise.resolve(result) as Promise & { limit: typeof limit } + forResult.limit = limit + const thenable = Promise.resolve(result) as Promise & { + limit: typeof limit + for: () => typeof forResult + } + thenable.limit = limit + thenable.for = vi.fn(() => forResult) + return thenable as ReturnType + }) +} + +describe('pauseProSubscriptionForOrgCoverage', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + it('pauses the personal Pro and queues the Stripe sync when an entitled paid org covers the user', async () => { + queueWhereResponses([ + [{ organizationId: 'org-1' }], + [{ plan: 'team_6000', referenceId: 'org-1' }], + [ACTIVE_PERSONAL_PRO], + // update ... set ... where consumes one more where() call + [], + ]) + + const result = await pauseProSubscriptionForOrgCoverage('user-1') + + expect(result).toEqual({ + covered: true, + paused: true, + subscriptionId: 'sub-personal', + organizationId: 'org-1', + }) + expect(dbChainMockFns.set).toHaveBeenCalledWith({ cancelAtPeriodEnd: true }) + expect(mockEnqueueOutboxEvent).toHaveBeenCalledWith( + expect.anything(), + OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END, + { + stripeSubscriptionId: 'stripe-sub-personal', + subscriptionId: 'sub-personal', + reason: 'covered-by-organization', + } + ) + }) + + it('reports covered even when no entitled personal Pro row exists', async () => { + queueWhereResponses([ + [{ organizationId: 'org-1' }], + [{ plan: 'team_6000', referenceId: 'org-1' }], + [], + ]) + + const result = await pauseProSubscriptionForOrgCoverage('user-1') + + expect(result).toEqual({ + covered: true, + paused: false, + organizationId: 'org-1', + }) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(mockEnqueueOutboxEvent).not.toHaveBeenCalled() + }) + + it('reports covered without pausing again when the personal Pro is already pausing', async () => { + queueWhereResponses([ + [{ organizationId: 'org-1' }], + [{ plan: 'team_6000', referenceId: 'org-1' }], + [{ ...ACTIVE_PERSONAL_PRO, cancelAtPeriodEnd: true }], + ]) + + const result = await pauseProSubscriptionForOrgCoverage('user-1') + + expect(result).toEqual({ + covered: true, + paused: false, + subscriptionId: 'sub-personal', + organizationId: 'org-1', + }) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(mockEnqueueOutboxEvent).not.toHaveBeenCalled() + }) + + it('reports not covered when the user is not a member of any organization', async () => { + queueWhereResponses([[]]) + + const result = await pauseProSubscriptionForOrgCoverage('user-1') + + expect(result).toEqual({ covered: false, paused: false }) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + }) + + it('reports not covered when no org subscription is an entitled paid plan', async () => { + queueWhereResponses([[{ organizationId: 'org-1' }], []]) + + const result = await pauseProSubscriptionForOrgCoverage('user-1') + + expect(result).toEqual({ covered: false, paused: false }) + expect(dbChainMockFns.update).not.toHaveBeenCalled() + expect(mockEnqueueOutboxEvent).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/billing/plan-helpers.ts b/apps/sim/lib/billing/plan-helpers.ts index b74ed0ff448..e905c671399 100644 --- a/apps/sim/lib/billing/plan-helpers.ts +++ b/apps/sim/lib/billing/plan-helpers.ts @@ -47,11 +47,13 @@ export function isPaid(plan: string | null | undefined): boolean { } /** - * True when the plan **name** is a team/enterprise plan. This is a - * plan-name check, NOT a scope check — a `pro_*` plan attached to an - * organization is org-scoped at the billing level even though this - * returns `false` for it. For scope decisions use - * `isOrgScopedSubscription` (sync) or `isSubscriptionOrgScoped` (async). + * True when the plan **name** is a team/enterprise plan — the only plans + * that may be referenced to an organization (org-referenced subscriptions + * never hold `pro_*` plans; checkout authorization and the Stripe plan + * sync both enforce this). This is a plan-name check, NOT a scope check: + * a team plan can be transiently user-referenced between checkout and + * webhook re-homing. For scope decisions use `isOrgScopedSubscription` + * (sync) or `isSubscriptionOrgScoped` (async). */ export function isOrgPlan(plan: string | null | undefined): boolean { return isTeam(plan) || isEnterprise(plan) diff --git a/apps/sim/lib/billing/webhooks/subscription.ts b/apps/sim/lib/billing/webhooks/subscription.ts index 321415adb23..6ce6c72f5de 100644 --- a/apps/sim/lib/billing/webhooks/subscription.ts +++ b/apps/sim/lib/billing/webhooks/subscription.ts @@ -150,75 +150,6 @@ async function transitionOrganizationToDormantState( } } -export async function handleOrganizationPlanDowngrade( - subscriptionData: { - id: string - plan: string | null - referenceId: string - status: string | null - }, - stripeEventId?: string -): Promise { - if (!(await isSubscriptionOrgScoped({ referenceId: subscriptionData.referenceId }))) { - return - } - - const stillTeamOrEnterprise = isTeam(subscriptionData.plan) || isEnterprise(subscriptionData.plan) - if (stillTeamOrEnterprise) { - return - } - - const [currentRow] = await db - .select({ plan: subscription.plan }) - .from(subscription) - .where(eq(subscription.id, subscriptionData.id)) - .limit(1) - - if (currentRow && (isTeam(currentRow.plan) || isEnterprise(currentRow.plan))) { - logger.info('Skipping plan downgrade transition - subscription is currently Team/Enterprise', { - subscriptionId: subscriptionData.id, - organizationId: subscriptionData.referenceId, - eventPlan: subscriptionData.plan, - currentPlan: currentRow.plan, - }) - return - } - - const idempotencyIdentifier = stripeEventId ?? `plan-downgrade:${subscriptionData.id}` - - try { - await stripeWebhookIdempotency.executeWithIdempotency( - 'organization-plan-downgrade', - idempotencyIdentifier, - async () => { - const result = await transitionOrganizationToDormantState( - subscriptionData.referenceId, - subscriptionData.id - ) - - if (result.workspacesDetached > 0 || result.restoredProCount > 0) { - logger.info('Transitioned organization to dormant state after plan downgrade', { - organizationId: subscriptionData.referenceId, - subscriptionId: subscriptionData.id, - plan: subscriptionData.plan, - ...result, - }) - } - - return result - } - ) - } catch (error) { - logger.error('Failed to transition organization to dormant state on plan downgrade', { - organizationId: subscriptionData.referenceId, - subscriptionId: subscriptionData.id, - plan: subscriptionData.plan, - error, - }) - throw error - } -} - /** * Handle new subscription creation - reset usage if transitioning from free to paid */ @@ -333,10 +264,10 @@ export async function handleSubscriptionCreated( /** * Handles a subscription deletion (cancel) event. Bills any final-period * overages, resets usage, and transitions the organization to a dormant - * state via `transitionOrganizationToDormantState` — the same path used - * by plan downgrades. Wrapped in `stripeWebhookIdempotency` so duplicate - * event deliveries collapse to one execution; if any step throws, the - * webhook retries from scratch. + * state via `transitionOrganizationToDormantState` — the sole trigger for + * detaching an organization's workspaces. Wrapped in + * `stripeWebhookIdempotency` so duplicate event deliveries collapse to one + * execution; if any step throws, the webhook retries from scratch. */ export async function handleSubscriptionDeleted( subscription: { diff --git a/apps/sim/lib/workspaces/organization-workspaces.ts b/apps/sim/lib/workspaces/organization-workspaces.ts index c44505bbd82..428610ac014 100644 --- a/apps/sim/lib/workspaces/organization-workspaces.ts +++ b/apps/sim/lib/workspaces/organization-workspaces.ts @@ -1,3 +1,4 @@ +import { AuditAction, AuditResourceType, recordAuditBatch } from '@sim/audit' import { db } from '@sim/db' import { member, permissions, workspace } from '@sim/db/schema' import { createLogger } from '@sim/logger' @@ -410,6 +411,32 @@ export async function detachOrganizationWorkspaces( return [...workspaceIds].sort() }) + const workspacesById = new Map( + organizationWorkspaces.map((organizationWorkspace) => [ + organizationWorkspace.id, + organizationWorkspace, + ]) + ) + recordAuditBatch( + detachedWorkspaceIds.map((detachedWorkspaceId) => { + const detachedWorkspace = workspacesById.get(detachedWorkspaceId) + return { + workspaceId: detachedWorkspaceId, + actorId: null, + actorName: 'Billing System', + action: AuditAction.WORKSPACE_UPDATED, + resourceType: AuditResourceType.WORKSPACE, + resourceId: detachedWorkspaceId, + description: 'Workspace detached from organization after its subscription ended', + metadata: { + organizationId, + previousBilledAccountUserId: detachedWorkspace?.billedAccountUserId ?? null, + newBilledAccountUserId: organizationOwnerId ?? detachedWorkspace?.ownerId ?? null, + }, + } + }) + ) + logger.info('Detached organization workspaces', { organizationId, detachedWorkspaceCount: detachedWorkspaceIds.length, diff --git a/packages/audit/src/index.ts b/packages/audit/src/index.ts index f3ec73f8c5d..2813a231de4 100644 --- a/packages/audit/src/index.ts +++ b/packages/audit/src/index.ts @@ -1,3 +1,3 @@ -export { recordAudit } from './log' +export { recordAudit, recordAuditBatch } from './log' export type { AuditActionType, AuditResourceTypeValue } from './types' export { AuditAction, AuditResourceType } from './types' diff --git a/packages/audit/src/log.test.ts b/packages/audit/src/log.test.ts index e47485ee085..98a71773e65 100644 --- a/packages/audit/src/log.test.ts +++ b/packages/audit/src/log.test.ts @@ -37,7 +37,7 @@ vi.mock('@sim/utils/id', () => ({ })) import { sleep } from '@sim/utils/helpers' -import { AuditAction, AuditResourceType, recordAudit } from './index' +import { AuditAction, AuditResourceType, recordAudit, recordAuditBatch } from './index' const flush = () => sleep(10) @@ -385,6 +385,68 @@ describe('recordAudit', () => { }) }) +describe('recordAuditBatch', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + it('writes all entries in a single insert', async () => { + recordAuditBatch([ + { + workspaceId: 'ws-1', + actorId: null, + actorName: 'Billing System', + action: AuditAction.WORKSPACE_UPDATED, + resourceType: AuditResourceType.WORKSPACE, + resourceId: 'ws-1', + }, + { + workspaceId: 'ws-2', + actorId: null, + actorName: 'Billing System', + action: AuditAction.WORKSPACE_UPDATED, + resourceType: AuditResourceType.WORKSPACE, + resourceId: 'ws-2', + }, + ]) + + await flush() + + expect(dbChainMockFns.insert).toHaveBeenCalledTimes(1) + expect(dbChainMockFns.values).toHaveBeenCalledWith([ + expect.objectContaining({ workspaceId: 'ws-1', actorId: null, actorName: 'Billing System' }), + expect.objectContaining({ workspaceId: 'ws-2', actorId: null, actorName: 'Billing System' }), + ]) + }) + + it('does nothing for an empty batch', async () => { + recordAuditBatch([]) + + await flush() + + expect(dbChainMockFns.insert).not.toHaveBeenCalled() + }) + + it('does not throw when the batch insert fails', async () => { + dbChainMockFns.values.mockImplementation(() => Promise.reject(new Error('DB connection lost'))) + + expect(() => { + recordAuditBatch([ + { + workspaceId: 'ws-1', + actorId: null, + actorName: 'Billing System', + action: AuditAction.WORKSPACE_UPDATED, + resourceType: AuditResourceType.WORKSPACE, + }, + ]) + }).not.toThrow() + + await flush() + }) +}) + describe('auditMock sync', () => { it('has the same AuditAction keys as the source', () => { const sourceKeys = Object.keys(AuditAction).sort() diff --git a/packages/audit/src/log.ts b/packages/audit/src/log.ts index 3a9a6ef66d4..93381ae43b7 100644 --- a/packages/audit/src/log.ts +++ b/packages/audit/src/log.ts @@ -44,10 +44,73 @@ export function recordAudit(params: AuditLogParams): void { }) } -async function insertAuditLog(params: AuditLogParams): Promise { - const ipAddress = params.request ? getClientIp(params.request) : undefined - const userAgent = params.request?.headers.get('user-agent') ?? undefined +/** + * Rows per INSERT statement. Keeps each statement's bind-parameter count + * far below Postgres's 65k limit while still writing large batches in a + * handful of round-trips. + */ +const AUDIT_BATCH_CHUNK_SIZE = 500 + +/** + * Fire-and-forget batch audit write: one INSERT per chunk instead of one + * pooled query per entry, so callers auditing many resources at once (e.g. + * a bulk workspace detach) do not fan out N concurrent pool checkouts. + * + * Unlike {@link recordAudit} there is no lazy actor resolution — entries + * are inserted exactly as provided. Callers must pass an `actorId` that is + * a real `user.id` or `null`, and should supply `actorName`/`actorEmail` + * labels for system actors. + */ +export function recordAuditBatch(entries: AuditLogParams[]): void { + insertAuditLogBatch(entries).catch((error) => { + logger.error('Failed to record audit log batch', { error, count: entries.length }) + }) +} +/** + * Build the `audit_log` row for an entry. Shared by the single and batch + * insert paths so the write shape cannot drift between them. Actor fields + * are taken as-is — lazy actor resolution is layered on top by + * {@link recordAudit} only. + */ +function buildAuditRow( + params: AuditLogParams, + actor: { actorId: string | null; actorName?: string | null; actorEmail?: string | null } +) { + return { + id: generateShortId(), + workspaceId: params.workspaceId || null, + actorId: actor.actorId, + action: params.action, + resourceType: params.resourceType, + resourceId: params.resourceId, + actorName: actor.actorName ?? undefined, + actorEmail: actor.actorEmail ?? undefined, + resourceName: params.resourceName, + description: params.description, + metadata: params.metadata ?? {}, + ipAddress: params.request ? getClientIp(params.request) : undefined, + userAgent: params.request?.headers.get('user-agent') ?? undefined, + } +} + +async function insertAuditLogBatch(entries: AuditLogParams[]): Promise { + if (entries.length === 0) return + + const rows = entries.map((params) => + buildAuditRow(params, { + actorId: params.actorId, + actorName: params.actorName, + actorEmail: params.actorEmail, + }) + ) + + for (let index = 0; index < rows.length; index += AUDIT_BATCH_CHUNK_SIZE) { + await db.insert(auditLog).values(rows.slice(index, index + AUDIT_BATCH_CHUNK_SIZE)) + } +} + +async function insertAuditLog(params: AuditLogParams): Promise { let { actorName, actorEmail } = params /** @@ -81,19 +144,5 @@ async function insertAuditLog(params: AuditLogParams): Promise { } } - await db.insert(auditLog).values({ - id: generateShortId(), - workspaceId: params.workspaceId || null, - actorId, - action: params.action, - resourceType: params.resourceType, - resourceId: params.resourceId, - actorName: actorName ?? undefined, - actorEmail: actorEmail ?? undefined, - resourceName: params.resourceName, - description: params.description, - metadata: params.metadata ?? {}, - ipAddress, - userAgent, - }) + await db.insert(auditLog).values(buildAuditRow(params, { actorId, actorName, actorEmail })) } diff --git a/packages/testing/src/mocks/audit.mock.ts b/packages/testing/src/mocks/audit.mock.ts index 6bf30bbc24d..6e8a1b49a4e 100644 --- a/packages/testing/src/mocks/audit.mock.ts +++ b/packages/testing/src/mocks/audit.mock.ts @@ -14,6 +14,7 @@ import { vi } from 'vitest' */ export const auditMockFns = { mockRecordAudit: vi.fn(), + mockRecordAuditBatch: vi.fn(), } /** @@ -26,6 +27,7 @@ export const auditMockFns = { */ export const auditMock = { recordAudit: auditMockFns.mockRecordAudit, + recordAuditBatch: auditMockFns.mockRecordAuditBatch, AuditAction: { API_KEY_CREATED: 'api_key.created', API_KEY_UPDATED: 'api_key.updated',