mirror of
https://github.com/yusufipk/OpenFrame.git
synced 2026-09-11 17:46:06 +00:00
fix(billing): serialize reconciliation and record accepted cancellations
This commit is contained in:
@@ -0,0 +1,339 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import type Stripe from 'stripe';
|
||||
import { Pool } from 'pg';
|
||||
import {
|
||||
buildBillingAccessWhereInput,
|
||||
hasBillingAccess,
|
||||
syncStripeCustomerSubscriptions,
|
||||
} from '@/lib/billing';
|
||||
import { getStripe } from '@/lib/stripe';
|
||||
import { db } from '../helpers/db';
|
||||
import { createUser } from '../factories';
|
||||
|
||||
// Real PostgreSQL persistence and advisory locks; only Stripe responses are emulated.
|
||||
// The held response models transport delay, not Stripe's actual webhook scheduling.
|
||||
const CUSTOMER = 'cus_sync_concurrency';
|
||||
const PRICE = 'price_sync_concurrency';
|
||||
|
||||
function deferred() {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((done) => {
|
||||
resolve = done;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
function subscription(
|
||||
status: 'active' | 'canceled',
|
||||
id = 'sub_paid',
|
||||
customer = CUSTOMER
|
||||
): Stripe.Subscription {
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
return {
|
||||
id,
|
||||
customer,
|
||||
status,
|
||||
created: now - 86_400,
|
||||
trial_end: null,
|
||||
cancel_at: null,
|
||||
cancel_at_period_end: false,
|
||||
ended_at: status === 'canceled' ? now - 60 : null,
|
||||
canceled_at: status === 'canceled' ? now - 60 : null,
|
||||
items: {
|
||||
data: [
|
||||
{
|
||||
price: { id: PRICE },
|
||||
current_period_start: now - 86_400,
|
||||
current_period_end: now + 30 * 86_400,
|
||||
},
|
||||
],
|
||||
},
|
||||
} as Stripe.Subscription;
|
||||
}
|
||||
|
||||
async function seed(status: 'ACTIVE' | 'CANCELED' = 'CANCELED', customer = CUSTOMER) {
|
||||
return createUser({
|
||||
stripeCustomerId: customer,
|
||||
stripeSubscriptionId: `sub_seed_${customer}`,
|
||||
stripePriceId: PRICE,
|
||||
subscriptionStatus: status,
|
||||
stripeCurrentPeriodEnd: new Date(Date.now() + 30 * 86_400_000),
|
||||
trialEndsAt: null,
|
||||
billingTrialConsumedAt: new Date(Date.now() - 60 * 86_400_000),
|
||||
billingAccessEndedAt: status === 'CANCELED' ? new Date(Date.now() - 60_000) : null,
|
||||
});
|
||||
}
|
||||
|
||||
function installStripe() {
|
||||
const list = vi.fn<
|
||||
(params: Stripe.SubscriptionListParams) => Promise<{
|
||||
data: Stripe.Subscription[];
|
||||
has_more: boolean;
|
||||
}>
|
||||
>();
|
||||
vi.mocked(getStripe).mockReturnValue({ subscriptions: { list } } as unknown as Stripe);
|
||||
return list;
|
||||
}
|
||||
|
||||
async function waitForQueuedSync(customer = CUSTOMER) {
|
||||
// Observe a real waiter rather than sleeping and assuming the other request ran.
|
||||
// Replacing the database lock with a process-local mutex fails this assertion.
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const rows = await db.$queryRaw<{ waiting: boolean }[]>`
|
||||
SELECT EXISTS (
|
||||
SELECT 1 FROM pg_locks
|
||||
WHERE locktype = 'advisory' AND NOT granted
|
||||
AND classid = hashtext('stripe-subscription-sync')::oid
|
||||
AND objid = hashtext(${customer})::oid AND objsubid = 2
|
||||
) AS waiting
|
||||
`;
|
||||
expect(rows).toEqual([{ waiting: true }]);
|
||||
},
|
||||
{ timeout: 2_000, interval: 20 }
|
||||
);
|
||||
}
|
||||
|
||||
async function assertAccess(userId: string, expected: boolean) {
|
||||
const stored = await db.user.findUniqueOrThrow({ where: { id: userId } });
|
||||
expect(hasBillingAccess(stored)).toBe(expected);
|
||||
expect(
|
||||
await db.user.findMany({
|
||||
where: { AND: [{ id: userId }, buildBillingAccessWhereInput()] },
|
||||
select: { id: true },
|
||||
})
|
||||
).toEqual(expected ? [{ id: userId }] : []);
|
||||
return stored;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.stubEnv('STRIPE_PRICE_ID', PRICE);
|
||||
vi.stubEnv('OPENFRAME_ENABLE_STRIPE', 'true');
|
||||
vi.stubEnv('OPENFRAME_ENABLE_ANALYTICS', 'true');
|
||||
});
|
||||
|
||||
describe('customer-wide Stripe sync serialization', () => {
|
||||
it.each(['canceled-then-paid', 'paid-then-canceled', 'empty-then-paid'] as const)(
|
||||
'keeps the newer snapshot for overlapping %s reads',
|
||||
async (order) => {
|
||||
const paidFirst = order === 'paid-then-canceled';
|
||||
const user = await seed(paidFirst ? 'ACTIVE' : 'CANCELED');
|
||||
const stale =
|
||||
order === 'empty-then-paid'
|
||||
? []
|
||||
: [subscription(paidFirst ? 'active' : 'canceled', paidFirst ? 'sub_paid' : 'sub_old')];
|
||||
const fresh = subscription(paidFirst ? 'canceled' : 'active');
|
||||
const list = installStripe();
|
||||
const entered = deferred();
|
||||
const release = deferred();
|
||||
list.mockImplementationOnce(async () => {
|
||||
entered.resolve();
|
||||
await release.promise;
|
||||
return { data: stale, has_more: false };
|
||||
});
|
||||
list.mockImplementationOnce(async () => {
|
||||
// Reading Stripe for the next sync must wait for the previous mirror commit.
|
||||
const previous = await db.user.findUniqueOrThrow({ where: { id: user.id } });
|
||||
expect(previous.stripeSubscriptionId).toBe(stale[0]?.id ?? null);
|
||||
return { data: [fresh], has_more: false };
|
||||
});
|
||||
const first = syncStripeCustomerSubscriptions(CUSTOMER);
|
||||
let second: ReturnType<typeof syncStripeCustomerSubscriptions> | undefined;
|
||||
// Attach handlers immediately so assertion failures still drain both requests.
|
||||
void first.catch(() => {});
|
||||
try {
|
||||
await entered.promise;
|
||||
second = syncStripeCustomerSubscriptions(CUSTOMER);
|
||||
void second.catch(() => {});
|
||||
await waitForQueuedSync();
|
||||
expect(list).toHaveBeenCalledTimes(1);
|
||||
const unchanged = await db.user.findUniqueOrThrow({ where: { id: user.id } });
|
||||
expect(unchanged.stripeSubscriptionId).toBe(user.stripeSubscriptionId);
|
||||
} finally {
|
||||
release.resolve();
|
||||
await Promise.allSettled([first, ...(second ? [second] : [])]);
|
||||
}
|
||||
await expect(first).resolves.not.toBeNull();
|
||||
await expect(second!).resolves.not.toBeNull();
|
||||
expect(list).toHaveBeenCalledTimes(2);
|
||||
const stored = await assertAccess(user.id, !paidFirst);
|
||||
expect(stored.subscriptionStatus).toBe(paidFirst ? 'CANCELED' : 'ACTIVE');
|
||||
expect(stored.stripeSubscriptionId).toBe('sub_paid');
|
||||
expect(
|
||||
await db.analyticsEvent.findMany({
|
||||
where: { userId: user.id },
|
||||
select: { name: true },
|
||||
})
|
||||
).toEqual([{ name: paidFirst ? 'SUBSCRIPTION_CANCELED' : 'SUBSCRIPTION_STARTED' }]);
|
||||
}
|
||||
);
|
||||
|
||||
it('honors the customer lock held by an independent database connection before reading Stripe', async () => {
|
||||
const user = await seed();
|
||||
const list = installStripe().mockResolvedValue({
|
||||
data: [subscription('active')],
|
||||
has_more: false,
|
||||
});
|
||||
const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 1 });
|
||||
const connection = await pool.connect();
|
||||
let pending: ReturnType<typeof syncStripeCustomerSubscriptions> | undefined;
|
||||
try {
|
||||
await connection.query('BEGIN');
|
||||
await connection.query('SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))', [
|
||||
'stripe-subscription-sync',
|
||||
CUSTOMER,
|
||||
]);
|
||||
pending = syncStripeCustomerSubscriptions(CUSTOMER);
|
||||
void pending.catch(() => {});
|
||||
await waitForQueuedSync();
|
||||
expect(list).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
await connection.query('ROLLBACK');
|
||||
connection.release();
|
||||
await pool.end();
|
||||
if (pending) await Promise.allSettled([pending]);
|
||||
}
|
||||
await expect(pending!).resolves.not.toBeNull();
|
||||
expect(list).toHaveBeenCalledTimes(1);
|
||||
await assertAccess(user.id, true);
|
||||
});
|
||||
|
||||
it('lets a different customer sync while the first customer waits on Stripe', async () => {
|
||||
await seed();
|
||||
const otherCustomer = 'cus_sync_independent';
|
||||
const otherUser = await seed('CANCELED', otherCustomer);
|
||||
const entered = deferred();
|
||||
const release = deferred();
|
||||
const list = installStripe().mockImplementation(async ({ customer }) => {
|
||||
if (customer === CUSTOMER) {
|
||||
entered.resolve();
|
||||
await release.promise;
|
||||
}
|
||||
return { data: [subscription('active', `sub_${customer}`, customer)], has_more: false };
|
||||
});
|
||||
const first = syncStripeCustomerSubscriptions(CUSTOMER);
|
||||
void first.catch(() => {});
|
||||
let second: ReturnType<typeof syncStripeCustomerSubscriptions> | undefined;
|
||||
let secondFinished = false;
|
||||
try {
|
||||
await entered.promise;
|
||||
second = syncStripeCustomerSubscriptions(otherCustomer);
|
||||
void second.then(
|
||||
() => {
|
||||
secondFinished = true;
|
||||
},
|
||||
() => {
|
||||
secondFinished = true;
|
||||
}
|
||||
);
|
||||
await vi.waitFor(() => expect(secondFinished).toBe(true), { timeout: 2_000 });
|
||||
await expect(second).resolves.not.toBeNull();
|
||||
await assertAccess(otherUser.id, true);
|
||||
expect(list).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
release.resolve();
|
||||
await Promise.allSettled([first, ...(second ? [second] : [])]);
|
||||
}
|
||||
await expect(first).resolves.not.toBeNull();
|
||||
});
|
||||
|
||||
it('releases a failed sync for its queued successor without changing the original mirror', async () => {
|
||||
const user = await seed();
|
||||
const entered = deferred();
|
||||
const release = deferred();
|
||||
const failure = new Error('Emulated Stripe read failure');
|
||||
const list = installStripe();
|
||||
list.mockImplementationOnce(async () => {
|
||||
entered.resolve();
|
||||
await release.promise;
|
||||
throw failure;
|
||||
});
|
||||
list.mockImplementationOnce(async () => {
|
||||
expect(await db.user.findUniqueOrThrow({ where: { id: user.id } })).toEqual(user);
|
||||
return { data: [subscription('active')], has_more: false };
|
||||
});
|
||||
const first = syncStripeCustomerSubscriptions(CUSTOMER);
|
||||
void first.catch(() => {});
|
||||
let second: ReturnType<typeof syncStripeCustomerSubscriptions> | undefined;
|
||||
try {
|
||||
await entered.promise;
|
||||
second = syncStripeCustomerSubscriptions(CUSTOMER);
|
||||
void second.catch(() => {});
|
||||
await waitForQueuedSync();
|
||||
} finally {
|
||||
release.resolve();
|
||||
await Promise.allSettled([first, ...(second ? [second] : [])]);
|
||||
}
|
||||
await expect(first).rejects.toBe(failure);
|
||||
await expect(second!).resolves.not.toBeNull();
|
||||
expect(list).toHaveBeenCalledTimes(2);
|
||||
await assertAccess(user.id, true);
|
||||
});
|
||||
|
||||
it('cannot overwrite a newer mirror when a Stripe response arrives after transaction expiry', async () => {
|
||||
const user = await seed();
|
||||
const entered = deferred();
|
||||
const release = deferred();
|
||||
const callbackFinished = deferred();
|
||||
const list = installStripe();
|
||||
list.mockImplementationOnce(async () => {
|
||||
entered.resolve();
|
||||
await release.promise;
|
||||
return { data: [subscription('canceled', 'sub_stale')], has_more: false };
|
||||
});
|
||||
list.mockResolvedValueOnce({ data: [subscription('active', 'sub_new')], has_more: false });
|
||||
|
||||
// Keep the real transaction and expiry machinery, shortening only the first
|
||||
// request's deadline. Its callback can outlive rollback while Stripe is held.
|
||||
const transact = db.$transaction.bind(db);
|
||||
const transaction = vi.spyOn(db, '$transaction').mockImplementationOnce((callback, options) =>
|
||||
transact(
|
||||
async (tx) => {
|
||||
try {
|
||||
return await callback(tx);
|
||||
} finally {
|
||||
callbackFinished.resolve();
|
||||
}
|
||||
},
|
||||
{ ...options, timeout: 200 }
|
||||
)
|
||||
);
|
||||
const first = syncStripeCustomerSubscriptions(CUSTOMER);
|
||||
void first.catch(() => {});
|
||||
let second: ReturnType<typeof syncStripeCustomerSubscriptions> | undefined;
|
||||
let committed: Awaited<ReturnType<typeof assertAccess>> | undefined;
|
||||
try {
|
||||
await entered.promise;
|
||||
// Wait for PostgreSQL to release A's lock, not an assumed sleep duration.
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const rows = await db.$queryRaw<{ held: boolean }[]>`
|
||||
SELECT EXISTS (
|
||||
SELECT 1 FROM pg_locks
|
||||
WHERE locktype = 'advisory' AND granted
|
||||
AND classid = hashtext('stripe-subscription-sync')::oid
|
||||
AND objid = hashtext(${CUSTOMER})::oid AND objsubid = 2
|
||||
) AS held
|
||||
`;
|
||||
expect(rows).toEqual([{ held: false }]);
|
||||
},
|
||||
{ timeout: 3_000, interval: 20 }
|
||||
);
|
||||
second = syncStripeCustomerSubscriptions(CUSTOMER);
|
||||
void second.catch(() => {});
|
||||
await expect(second).resolves.not.toBeNull();
|
||||
committed = await assertAccess(user.id, true);
|
||||
expect(committed.stripeSubscriptionId).toBe('sub_new');
|
||||
} finally {
|
||||
release.resolve();
|
||||
// The outer promise may reject on expiry before its callback finishes.
|
||||
// Drain both so a late global-client write cannot escape the assertions.
|
||||
await Promise.allSettled([first, ...(second ? [second] : [])]);
|
||||
await callbackFinished.promise;
|
||||
transaction.mockRestore();
|
||||
}
|
||||
await expect(first).rejects.toMatchObject({ code: 'P2028' });
|
||||
expect(list).toHaveBeenCalledTimes(2);
|
||||
expect(await assertAccess(user.id, true)).toEqual(committed);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user