diff --git a/apps/server/src/routes/stripe/index.ts b/apps/server/src/routes/stripe/index.ts index 08a522f6d..d96446070 100644 --- a/apps/server/src/routes/stripe/index.ts +++ b/apps/server/src/routes/stripe/index.ts @@ -11,38 +11,32 @@ import type { HonoEnv } from '../../types/hono' import Stripe from 'stripe' -import { useLogger } from '@guiiai/logg' import { Hono } from 'hono' -import { safeParse } from 'valibot' import { authGuard } from '../../middlewares/auth' import { rateLimiter } from '../../middlewares/rate-limit' -import { captureSafe } from '../../services/adapters/posthog' import { createBadRequestError, createServiceUnavailableError } from '../../utils/error' -import { errorMessageFromUnknown } from '../../utils/error-message' import { resolveCheckoutRedirectBase } from '../../utils/origin' -import { redisKeyFrom } from '../../utils/redis-keys' -import { CheckoutBodySchema } from './schema' +import { createCheckoutOperation } from './operations/checkout' +import { createWebhookOperation } from './operations/webhook' +import { createStripePriceCatalog, formatPrice } from './price-catalog' -const logger = useLogger('stripe') - -const PRICES_CACHE_KEY = redisKeyFrom('cache', 'stripe', 'prices') -const PRICES_CACHE_TTL_SEC = 5 * 60 - -interface CachedCurrencyOption { - unitAmount: number | null -} - -interface CachedPrice { - id: string - unitAmount: number | null - currency: string - product: string - active: boolean - metadata: Record - currencyOptions: Record -} +export { formatPrice } from './price-catalog' +/** + * Creates Stripe HTTP routes for Flux purchase and billing records. + * + * Use when: + * - Mounting `/api/v1/stripe` in the server app. + * - Wiring Stripe checkout, customer portal, package catalog, and webhooks. + * + * Expects: + * - Auth middleware to populate `c.get('user')` for protected endpoints. + * - Stripe configuration to be present for checkout, portal, and webhook routes. + * + * Returns: + * - A Hono router scoped to Stripe endpoints. + */ export function createStripeRoutes( fluxService: FluxService, stripeService: StripeService, @@ -55,54 +49,27 @@ export function createStripeRoutes( posthog?: PostHog | null, ) { const stripe = env.STRIPE_SECRET_KEY ? new Stripe(env.STRIPE_SECRET_KEY) : null - - async function getActivePrices(productId: string): Promise { - // Try Redis cache first - const cached = await redis.get(PRICES_CACHE_KEY) - if (cached) { - try { - const parsed = JSON.parse(cached) as { productId: string, prices: CachedPrice[] } - if (parsed.productId === productId) - return parsed.prices - } - catch { /* corrupted cache, refetch */ } - } - - let result: Stripe.ApiList - try { - result = await stripe!.prices.list({ product: productId, active: true, expand: ['data.currency_options'] }) - } - catch (err) { - logger.withError(err).warn('Failed to fetch prices from Stripe') - return [] - } - const prices: CachedPrice[] = result.data - .sort((a, b) => (a.unit_amount ?? 0) - (b.unit_amount ?? 0)) - .map(p => ({ - id: p.id, - unitAmount: p.unit_amount, - currency: p.currency, - product: typeof p.product === 'string' ? p.product : p.product.id, - active: p.active, - metadata: p.metadata, - currencyOptions: Object.fromEntries( - Object.entries(p.currency_options ?? {}).map(([cur, opt]) => [cur, { unitAmount: opt.unit_amount }]), - ), - })) - - await redis.set(PRICES_CACHE_KEY, JSON.stringify({ productId, prices }), 'EX', PRICES_CACHE_TTL_SEC) - return prices - } + const priceCatalog = stripe ? createStripePriceCatalog(stripe, redis) : null + const checkout = createCheckoutOperation({ stripe, priceCatalog, stripeService, configKV, env, metrics }) + const webhook = createWebhookOperation({ + stripe, + webhookSecret: env.STRIPE_WEBHOOK_SECRET, + fluxService, + stripeService, + billingService, + metrics, + posthog, + }) return new Hono() .get('/packages', async (c) => { const fluxProductId = await configKV.getOptional('STRIPE_FLUX_PRODUCT_ID') - if (!stripe || !fluxProductId) + if (!priceCatalog || !fluxProductId) return c.json([]) - const prices = await getActivePrices(fluxProductId) + const prices = await priceCatalog.getActivePrices(fluxProductId) - // Build per-currency price map for each package + // Build per-currency price map for each package. return c.json(prices.map((p) => { const currencies: Record = { [p.currency]: formatPrice(p.unitAmount, p.currency), @@ -121,125 +88,23 @@ export function createStripeRoutes( })) }) .post('/checkout', authGuard, rateLimiter({ max: 10, windowSec: 60, metrics: rateLimitMetrics, routeLabel: 'stripe.checkout' }), async (c) => { - const fluxProductId = await configKV.getOptional('STRIPE_FLUX_PRODUCT_ID') - if (!stripe || !fluxProductId) - throw createServiceUnavailableError('Stripe is not configured', 'STRIPE_NOT_CONFIGURED') - - const user = c.get('user')! const body = await c.req.json() - - const result = safeParse(CheckoutBodySchema, body) - if (!result.success) - throw createBadRequestError('Invalid checkout request', 'INVALID_REQUEST', result.issues) - - const { stripePriceId, currency } = result.output - - // Validate against cached prices first, fall back to direct Stripe API - const cachedPrices = await getActivePrices(fluxProductId) - let price = cachedPrices.find(p => p.id === stripePriceId) - - if (!price) { - // Cache miss — price may have just been created - let fetched: Stripe.Price - try { - fetched = await stripe.prices.retrieve(stripePriceId) - } - catch { - throw createBadRequestError('Invalid price', 'INVALID_PACKAGE', { stripePriceId }) - } - - if (!fetched.active || (typeof fetched.product === 'string' ? fetched.product : fetched.product.id) !== fluxProductId) { - throw createBadRequestError('Invalid price', 'INVALID_PACKAGE', { stripePriceId }) - } - - price = { - id: fetched.id, - unitAmount: fetched.unit_amount, - currency: fetched.currency, - product: typeof fetched.product === 'string' ? fetched.product : fetched.product.id, - active: fetched.active, - metadata: fetched.metadata, - currencyOptions: Object.fromEntries( - Object.entries(fetched.currency_options ?? {}).map(([cur, opt]) => [cur, { unitAmount: opt.unit_amount }]), - ), - } - - // Invalidate cache so all instances pick up the new price - await redis.del(PRICES_CACHE_KEY) - } - - const fluxAmount = Number(price.metadata.fluxAmount) - if (!Number.isFinite(fluxAmount) || fluxAmount <= 0) { - throw createBadRequestError('Price is missing fluxAmount metadata', 'INVALID_PACKAGE', { stripePriceId }) - } - - // Reuse existing stripe customer if available - const customer = await stripeService.getCustomerByUserId(user.id) - const stripeCustomerId = customer?.stripeCustomerId - - const redirectBase = resolveCheckoutRedirectBase(c.req.raw, env.ADDITIONAL_TRUSTED_ORIGINS, env.WEB_APP_URL) - - const paymentMethods = await configKV.getOptional('STRIPE_PAYMENT_METHODS') - const paymentMethodOptions = await configKV.getOptional('STRIPE_PAYMENT_METHOD_OPTIONS') ?? {} - - const session = await stripe.checkout.sessions.create({ - // When STRIPE_PAYMENT_METHODS is not set, omit payment_method_types to let Stripe - // automatically determine available methods based on currency and Dashboard settings - ...(paymentMethods && { payment_method_types: paymentMethods as any }), - ...(Object.keys(paymentMethodOptions).length > 0 && { payment_method_options: paymentMethodOptions as any }), - // When currency is specified, Stripe uses the matching currency_options on the Price - ...(currency && { currency }), - line_items: [{ price: stripePriceId, quantity: 1 }], - mode: 'payment', - allow_promotion_codes: true, - success_url: `${redirectBase}/settings/flux?success=true`, - cancel_url: `${redirectBase}/settings/flux?canceled=true`, - customer: stripeCustomerId, - customer_email: stripeCustomerId ? undefined : user.email, - metadata: { - userId: user.id, - fluxAmount: String(fluxAmount), - }, - }) - - // Persist the checkout session - await stripeService.upsertCheckoutSession({ - userId: user.id, - stripeSessionId: session.id, - stripeCustomerId: typeof session.customer === 'string' ? session.customer : session.customer?.id, - mode: session.mode ?? 'payment', - status: session.status, - paymentStatus: session.payment_status, - amountTotal: session.amount_total, - currency: session.currency, - successUrl: session.success_url, - cancelUrl: session.cancel_url, - stripePaymentIntentId: typeof session.payment_intent === 'string' ? session.payment_intent : session.payment_intent?.id, - stripeSubscriptionId: typeof session.subscription === 'string' ? session.subscription : session.subscription?.id, - metadata: session.metadata ? JSON.stringify(session.metadata) : null, - expiresAt: session.expires_at ? new Date(session.expires_at * 1000) : null, - }) - - metrics?.stripeCheckoutCreated.add(1) - - return c.json({ url: session.url }) + return c.json(await checkout({ + user: c.get('user')!, + body, + request: c.req.raw, + })) }) - - // ---- Orders / checkout sessions history ---- .get('/orders', authGuard, async (c) => { const user = c.get('user')! const sessions = await stripeService.getCheckoutSessionsByUserId(user.id) return c.json(sessions) }) - - // ---- Invoices history ---- .get('/invoices', authGuard, async (c) => { const user = c.get('user')! const invoices = await stripeService.getInvoicesByUserId(user.id) return c.json(invoices) }) - - // ---- Customer portal ---- .post('/portal', authGuard, async (c) => { if (!stripe) throw createServiceUnavailableError('Stripe is not configured', 'STRIPE_NOT_CONFIGURED') @@ -258,344 +123,9 @@ export function createStripeRoutes( return c.json({ url: portalSession.url }) }) - - // ---- Webhook ---- .post('/webhook', async (c) => { - if (!stripe || !env.STRIPE_WEBHOOK_SECRET) - throw createServiceUnavailableError('Stripe is not configured', 'STRIPE_NOT_CONFIGURED') - - const sig = c.req.header('stripe-signature') - if (!sig) - throw createBadRequestError('No signature', 'MISSING_SIGNATURE') - - let event: Stripe.Event - try { - const body = await c.req.text() - event = stripe.webhooks.constructEvent(body, sig, env.STRIPE_WEBHOOK_SECRET) - } - catch (err: unknown) { - throw createBadRequestError(`Webhook Error: ${errorMessageFromUnknown(err)}`, 'WEBHOOK_ERROR') - } - - logger.withFields({ type: event.type, id: event.id }).log('Webhook event received') - metrics?.stripeEvents.add(1, { event_type: event.type }) - - switch (event.type) { - case 'checkout.session.completed': { - const result = await handleCheckoutSessionCompleted(event.id, event.data.object, fluxService, stripeService, billingService) - metrics?.stripeCheckoutCompleted.add(1) - // Revenue capture in smallest currency unit (e.g. cents). - // Cross-currency aggregation is meaningless, so always group by `currency` in queries. - if (event.data.object.amount_total != null && event.data.object.currency) { - metrics?.stripeRevenue.add(event.data.object.amount_total, { - currency: event.data.object.currency, - source: 'checkout', - }) - } - // PostHog: funnel terminator. Only fire when the handler actually - // processed the checkout — malformed sessions (missing userId, - // invalid fluxAmount) take the early-return path above and would - // otherwise poison the funnel with phantom conversions. distinctId - // is the Better Auth user id so it merges with the browser's - // `posthog.identify(userId)` and the prior `checkout_started` - // event lines up. See docs/ai-context/metrics-ownership.md. - if (result.processed) - await capturePaymentCompleted(posthog, event.data.object) - break - } - case 'customer.created': - case 'customer.updated': { - await handleCustomerEvent(event.data.object, stripeService) - break - } - case 'customer.subscription.created': - case 'customer.subscription.updated': - case 'customer.subscription.deleted': { - await handleSubscriptionEvent(event.data.object, stripeService) - metrics?.stripeSubscriptionEvent.add(1, { event_type: event.type.replace('customer.subscription.', '') }) - if (event.type === 'customer.subscription.deleted') - await captureSubscriptionCancelled(posthog, stripeService, event.data.object) - break - } - case 'invoice.created': - case 'invoice.updated': - case 'invoice.paid': - case 'invoice.payment_failed': { - await handleInvoiceEvent(event.data.object, stripeService) - if (event.type === 'invoice.payment_failed') { - metrics?.stripePaymentFailed.add(1) - } - if (event.type === 'invoice.paid' && event.data.object.amount_paid && event.data.object.currency) { - metrics?.stripeRevenue.add(event.data.object.amount_paid, { - currency: event.data.object.currency, - source: 'invoice', - }) - } - break - } - } - - return c.json({ received: true }) + const signature = c.req.header('stripe-signature') ?? null + const body = signature ? await c.req.text() : '' + return c.json(await webhook({ signature, body })) }) } - -// ---- Webhook handlers ---- - -async function handleCheckoutSessionCompleted( - stripeEventId: string, - session: Stripe.Checkout.Session, - fluxService: FluxService, - stripeService: StripeService, - billingService: BillingService, -): Promise<{ processed: boolean }> { - const userId = session.metadata?.userId - if (!userId) { - logger.withFields({ sessionId: session.id }).warn('Checkout session missing userId in metadata') - return { processed: false } - } - - logger.withFields({ userId, sessionId: session.id, mode: session.mode, amount: session.amount_total, currency: session.currency }).log('Processing checkout session') - - // Upsert customer record if we got a customer back - if (session.customer) { - const stripeCustomerId = typeof session.customer === 'string' ? session.customer : session.customer.id - await stripeService.upsertCustomer({ - userId, - stripeCustomerId, - email: session.customer_email ?? undefined, - }) - await fluxService.updateStripeCustomerId(userId, stripeCustomerId) - } - - // Update the checkout session record - await stripeService.upsertCheckoutSession({ - userId, - stripeSessionId: session.id, - stripeCustomerId: typeof session.customer === 'string' ? session.customer : session.customer?.id, - mode: session.mode ?? 'payment', - status: session.status, - paymentStatus: session.payment_status, - amountTotal: session.amount_total, - currency: session.currency, - successUrl: session.success_url, - cancelUrl: session.cancel_url, - stripePaymentIntentId: typeof session.payment_intent === 'string' ? session.payment_intent : session.payment_intent?.id, - stripeSubscriptionId: typeof session.subscription === 'string' ? session.subscription : session.subscription?.id, - metadata: session.metadata ? JSON.stringify(session.metadata) : null, - expiresAt: session.expires_at ? new Date(session.expires_at * 1000) : null, - }) - - // Idempotent flux credit: use fluxCredited flag inside a transaction - // to prevent double-crediting on webhook replay. - // - // For `payment` mode (one-time Flux purchase) `metadata.fluxAmount` is - // required — without it we can't credit anything, and the funnel must - // not see a `payment_completed` event for a checkout that didn't - // actually deliver Flux. Non-`payment` modes (e.g. `setup` for saving - // a card) deliberately skip crediting and still count as processed. - if (session.mode === 'payment') { - if (session.amount_total == null) { - logger.withFields({ userId, sessionId: session.id }).warn('Payment-mode checkout missing amount_total; skipping credit and capture') - return { processed: false } - } - const metadataFlux = session.metadata?.fluxAmount - if (!metadataFlux) { - logger.withFields({ userId, sessionId: session.id }).warn('Payment-mode checkout missing metadata.fluxAmount; skipping credit and capture') - return { processed: false } - } - const fluxAmount = Number(metadataFlux) - if (!Number.isFinite(fluxAmount) || fluxAmount <= 0) { - logger.withFields({ userId, sessionId: session.id, metadataFlux }).warn('Invalid fluxAmount in session metadata, skipping credit') - return { processed: false } - } - - const result = await billingService.creditFluxFromStripeCheckout({ - stripeEventId, - userId, - stripeSessionId: session.id, - amountTotal: session.amount_total, - currency: session.currency, - fluxAmount, - }) - - logger.withFields({ - userId, - fluxAmount, - amountTotal: session.amount_total, - applied: result.applied, - balanceAfter: result.balanceAfter, - }).log('Processed flux credit for one-time payment') - } - - return { processed: true } -} - -async function capturePaymentCompleted( - posthog: PostHog | null | undefined, - session: Stripe.Checkout.Session, -): Promise { - if (!posthog) - return - - const userId = session.metadata?.userId - const email = session.customer_email - || (typeof session.customer_details?.email === 'string' ? session.customer_details.email : null) - || null - - // distinctId fallback chain: userId (Better Auth, matches browser identify()) - // > email (PostHog will merge on identify later) > stripe session id (last - // resort — orphan event but at least we count it). - const distinctId = userId || email || session.id - - const fluxAmount = Number(session.metadata?.fluxAmount) - const stripeCustomerId = typeof session.customer === 'string' ? session.customer : session.customer?.id - - await captureSafe(posthog, { - distinctId, - event: 'payment_completed', - properties: { - amount_total: session.amount_total, - currency: session.currency, - flux_amount: Number.isFinite(fluxAmount) ? fluxAmount : null, - mode: session.mode, - stripe_session_id: session.id, - stripe_customer_id: stripeCustomerId, - stripe_subscription_id: typeof session.subscription === 'string' ? session.subscription : session.subscription?.id, - ...(userId ? { user_id: userId } : {}), - // $set populates the PostHog person profile so funnel joins work even - // when a user pays via direct checkout link before ever loading the SPA. - ...(email ? { $set: { email } } : {}), - }, - }) -} - -async function captureSubscriptionCancelled( - posthog: PostHog | null | undefined, - stripeService: StripeService, - subscription: Stripe.Subscription, -): Promise { - if (!posthog) - return - - const stripeCustomerId = typeof subscription.customer === 'string' ? subscription.customer : subscription.customer.id - const customer = await stripeService.getCustomerByStripeId(stripeCustomerId) - const distinctId = customer?.userId || stripeCustomerId - - await captureSafe(posthog, { - distinctId, - event: 'subscription_cancelled', - properties: { - stripe_subscription_id: subscription.id, - stripe_customer_id: stripeCustomerId, - cancel_at_period_end: subscription.cancel_at_period_end, - cancellation_reason: subscription.cancellation_details?.reason ?? null, - cancellation_comment: subscription.cancellation_details?.comment ?? null, - canceled_at: subscription.canceled_at, - ended_at: subscription.ended_at, - ...(customer?.userId ? { user_id: customer.userId } : {}), - }, - }) -} - -async function handleCustomerEvent( - customer: Stripe.Customer | Stripe.DeletedCustomer, - stripeService: StripeService, -) { - if (customer.deleted) - return - - // Try to find existing customer to get userId - const existing = await stripeService.getCustomerByStripeId(customer.id) - if (!existing) - return // We don't know the userId yet; will be linked on checkout - - await stripeService.upsertCustomer({ - userId: existing.userId, - stripeCustomerId: customer.id, - email: customer.email ?? undefined, - name: customer.name ?? undefined, - }) -} - -async function handleSubscriptionEvent( - subscription: Stripe.Subscription, - stripeService: StripeService, -) { - const stripeCustomerId = typeof subscription.customer === 'string' ? subscription.customer : subscription.customer.id - const customer = await stripeService.getCustomerByStripeId(stripeCustomerId) - if (!customer) - return - - // In newer Stripe API, period info is on subscription items - const firstItem = subscription.items.data[0] - await stripeService.upsertSubscription({ - userId: customer.userId, - stripeSubscriptionId: subscription.id, - stripeCustomerId, - stripePriceId: firstItem?.price?.id, - status: subscription.status, - currentPeriodStart: firstItem?.current_period_start ? new Date(firstItem.current_period_start * 1000) : null, - currentPeriodEnd: firstItem?.current_period_end ? new Date(firstItem.current_period_end * 1000) : null, - cancelAtPeriodEnd: subscription.cancel_at_period_end, - canceledAt: subscription.canceled_at ? new Date(subscription.canceled_at * 1000) : null, - endedAt: subscription.ended_at ? new Date(subscription.ended_at * 1000) : null, - metadata: subscription.metadata ? JSON.stringify(subscription.metadata) : null, - }) -} - -async function handleInvoiceEvent( - invoice: Stripe.Invoice, - stripeService: StripeService, -) { - const stripeCustomerId = typeof invoice.customer === 'string' ? invoice.customer : invoice.customer?.id - if (!stripeCustomerId) - return - - const customer = await stripeService.getCustomerByStripeId(stripeCustomerId) - if (!customer) - return - - // In newer Stripe API, subscription is under parent.subscription_details - const subDetails = invoice.parent?.subscription_details - const subscriptionId = subDetails - ? (typeof subDetails.subscription === 'string' ? subDetails.subscription : subDetails.subscription?.id) - : undefined - - await stripeService.upsertInvoice({ - userId: customer.userId, - stripeInvoiceId: invoice.id, - stripeCustomerId, - stripeSubscriptionId: subscriptionId, - status: invoice.status, - amountDue: invoice.amount_due, - amountPaid: invoice.amount_paid, - currency: invoice.currency, - invoiceUrl: invoice.hosted_invoice_url, - invoicePdf: invoice.invoice_pdf, - periodStart: new Date(invoice.period_start * 1000), - periodEnd: new Date(invoice.period_end * 1000), - paidAt: invoice.status_transitions?.paid_at ? new Date(invoice.status_transitions.paid_at * 1000) : null, - metadata: invoice.metadata ? JSON.stringify(invoice.metadata) : null, - }) - - // TODO: implement subscription-based flux crediting when subscriptions are enabled - if (invoice.status === 'paid' && invoice.amount_paid && subscriptionId) { - logger.withFields({ userId: customer.userId, invoiceId: invoice.id, amountPaid: invoice.amount_paid }).warn('Subscription invoice paid but flux crediting for subscriptions is not yet implemented') - } -} - -/** Format Stripe smallest-unit amount into a human-readable price string */ -export function formatPrice(unitAmount: number | null, currency: string): string { - if (unitAmount == null) - return currency.toUpperCase() - - try { - const formatter = new Intl.NumberFormat('en-US', { style: 'currency', currency }) - const fractionDigits = formatter.resolvedOptions().minimumFractionDigits ?? 2 - const amount = unitAmount / (10 ** fractionDigits) - return formatter.format(amount) - } - catch { - return `${unitAmount / 100} ${currency.toUpperCase()}` - } -} diff --git a/apps/server/src/routes/stripe/operations/checkout.ts b/apps/server/src/routes/stripe/operations/checkout.ts new file mode 100644 index 000000000..2122186ff --- /dev/null +++ b/apps/server/src/routes/stripe/operations/checkout.ts @@ -0,0 +1,127 @@ +import type Stripe from 'stripe' + +import type { Env } from '../../../libs/env' +import type { RevenueMetrics } from '../../../otel' +import type { ConfigKVService } from '../../../services/adapters/config-kv' +import type { StripeService } from '../../../services/domain/stripe' +import type { HonoEnv } from '../../../types/hono' +import type { StripePriceCatalog } from '../price-catalog' + +import { safeParse } from 'valibot' + +import { createBadRequestError, createServiceUnavailableError } from '../../../utils/error' +import { resolveCheckoutRedirectBase } from '../../../utils/origin' +import { CheckoutBodySchema } from '../schema' + +type AuthenticatedUser = NonNullable +type CheckoutSessionCreateParams = NonNullable[0]> + +export interface CheckoutOperationDeps { + stripe: Stripe | null + priceCatalog: StripePriceCatalog | null + stripeService: StripeService + configKV: ConfigKVService + env: Env + metrics?: RevenueMetrics | null +} + +export interface CheckoutOperationInput { + user: AuthenticatedUser + body: unknown + request: Request +} + +/** + * Creates Stripe checkout sessions for Flux packages. + * + * Use when: + * - A signed-in user starts a one-time Flux purchase. + * - The route already enforced auth and rate limiting. + * + * Expects: + * - ConfigKV has `STRIPE_FLUX_PRODUCT_ID`. + * - `body` matches {@link CheckoutBodySchema}. + * + * Returns: + * - A Stripe-hosted checkout URL. + */ +export function createCheckoutOperation(deps: CheckoutOperationDeps) { + return async (input: CheckoutOperationInput): Promise<{ url: string | null }> => { + const fluxProductId = await deps.configKV.getOptional('STRIPE_FLUX_PRODUCT_ID') + if (!deps.stripe || !deps.priceCatalog || !fluxProductId) + throw createServiceUnavailableError('Stripe is not configured', 'STRIPE_NOT_CONFIGURED') + + const result = safeParse(CheckoutBodySchema, input.body) + if (!result.success) + throw createBadRequestError('Invalid checkout request', 'INVALID_REQUEST', result.issues) + + const { stripePriceId, currency } = result.output + + const price = await deps.priceCatalog.findActivePrice(fluxProductId, stripePriceId) + if (!price) + throw createBadRequestError('Invalid price', 'INVALID_PACKAGE', { stripePriceId }) + + const fluxAmount = Number(price.metadata.fluxAmount) + if (!Number.isFinite(fluxAmount) || fluxAmount <= 0) + throw createBadRequestError('Price is missing fluxAmount metadata', 'INVALID_PACKAGE', { stripePriceId }) + + // Reuse existing stripe customer if available. + const customer = await deps.stripeService.getCustomerByUserId(input.user.id) + const stripeCustomerId = customer?.stripeCustomerId + + const redirectBase = resolveCheckoutRedirectBase(input.request, deps.env.ADDITIONAL_TRUSTED_ORIGINS, deps.env.WEB_APP_URL) + + const paymentMethods = await deps.configKV.getOptional('STRIPE_PAYMENT_METHODS') + const paymentMethodOptions = await deps.configKV.getOptional('STRIPE_PAYMENT_METHOD_OPTIONS') ?? {} + + const sessionParams: CheckoutSessionCreateParams = { + line_items: [{ price: stripePriceId, quantity: 1 }], + mode: 'payment', + allow_promotion_codes: true, + success_url: `${redirectBase}/settings/flux?success=true`, + cancel_url: `${redirectBase}/settings/flux?canceled=true`, + customer: stripeCustomerId, + customer_email: stripeCustomerId ? undefined : input.user.email, + metadata: { + userId: input.user.id, + fluxAmount: String(fluxAmount), + }, + } + + // When STRIPE_PAYMENT_METHODS is not set, omit payment_method_types to let Stripe + // automatically determine available methods based on currency and Dashboard settings. + if (paymentMethods) + sessionParams.payment_method_types = paymentMethods as CheckoutSessionCreateParams['payment_method_types'] + + if (Object.keys(paymentMethodOptions).length > 0) + sessionParams.payment_method_options = paymentMethodOptions as CheckoutSessionCreateParams['payment_method_options'] + + // When currency is specified, Stripe uses the matching currency_options on the Price. + if (currency) + sessionParams.currency = currency + + const session = await deps.stripe.checkout.sessions.create(sessionParams) + + // Persist the checkout session. + await deps.stripeService.upsertCheckoutSession({ + userId: input.user.id, + stripeSessionId: session.id, + stripeCustomerId: typeof session.customer === 'string' ? session.customer : session.customer?.id, + mode: session.mode ?? 'payment', + status: session.status, + paymentStatus: session.payment_status, + amountTotal: session.amount_total, + currency: session.currency, + successUrl: session.success_url, + cancelUrl: session.cancel_url, + stripePaymentIntentId: typeof session.payment_intent === 'string' ? session.payment_intent : session.payment_intent?.id, + stripeSubscriptionId: typeof session.subscription === 'string' ? session.subscription : session.subscription?.id, + metadata: session.metadata ? JSON.stringify(session.metadata) : null, + expiresAt: session.expires_at ? new Date(session.expires_at * 1000) : null, + }) + + deps.metrics?.stripeCheckoutCreated.add(1) + + return { url: session.url } + } +} diff --git a/apps/server/src/routes/stripe/operations/webhook.ts b/apps/server/src/routes/stripe/operations/webhook.ts new file mode 100644 index 000000000..0ed9418a9 --- /dev/null +++ b/apps/server/src/routes/stripe/operations/webhook.ts @@ -0,0 +1,361 @@ +import type { PostHog } from 'posthog-node' +import type Stripe from 'stripe' + +import type { RevenueMetrics } from '../../../otel' +import type { BillingService } from '../../../services/domain/billing/billing-service' +import type { FluxService } from '../../../services/domain/flux' +import type { StripeService } from '../../../services/domain/stripe' + +import { useLogger } from '@guiiai/logg' + +import { captureSafe } from '../../../services/adapters/posthog' +import { createBadRequestError, createServiceUnavailableError } from '../../../utils/error' +import { errorMessageFromUnknown } from '../../../utils/error-message' + +const logger = useLogger('stripe') + +export interface WebhookOperationDeps { + stripe: Stripe | null + webhookSecret: string | undefined + fluxService: FluxService + stripeService: StripeService + billingService: BillingService + metrics?: RevenueMetrics | null + posthog?: PostHog | null +} + +export interface WebhookOperationInput { + signature: string | null + body: string +} + +/** + * Processes Stripe webhook events. + * + * Use when: + * - The route has captured the raw request body. + * - Stripe signature verification must happen before event dispatch. + * + * Expects: + * - A configured Stripe client and webhook secret. + * + * Returns: + * - `{ received: true }` after known and unknown events are accepted. + */ +export function createWebhookOperation(deps: WebhookOperationDeps) { + return async (input: WebhookOperationInput): Promise<{ received: true }> => { + if (!deps.stripe || !deps.webhookSecret) + throw createServiceUnavailableError('Stripe is not configured', 'STRIPE_NOT_CONFIGURED') + + if (!input.signature) + throw createBadRequestError('No signature', 'MISSING_SIGNATURE') + + let event: Stripe.Event + try { + event = deps.stripe.webhooks.constructEvent(input.body, input.signature, deps.webhookSecret) + } + catch (err: unknown) { + throw createBadRequestError(`Webhook Error: ${errorMessageFromUnknown(err)}`, 'WEBHOOK_ERROR') + } + + logger.withFields({ type: event.type, id: event.id }).log('Webhook event received') + deps.metrics?.stripeEvents.add(1, { event_type: event.type }) + + switch (event.type) { + case 'checkout.session.completed': { + const result = await handleCheckoutSessionCompleted(event.id, event.data.object, deps.fluxService, deps.stripeService, deps.billingService) + deps.metrics?.stripeCheckoutCompleted.add(1) + // Revenue capture in smallest currency unit (e.g. cents). + // Cross-currency aggregation is meaningless, so always group by `currency` in queries. + if (event.data.object.amount_total != null && event.data.object.currency) { + deps.metrics?.stripeRevenue.add(event.data.object.amount_total, { + currency: event.data.object.currency, + source: 'checkout', + }) + } + // PostHog: funnel terminator. Only fire when the handler actually + // processed the checkout — malformed sessions (missing userId, + // invalid fluxAmount) take the early-return path above and would + // otherwise poison the funnel with phantom conversions. distinctId + // is the Better Auth user id so it merges with the browser's + // `posthog.identify(userId)` and the prior `checkout_started` + // event lines up. See docs/ai-context/metrics-ownership.md. + if (result.processed) + await capturePaymentCompleted(deps.posthog, event.data.object) + break + } + case 'customer.created': + case 'customer.updated': { + await handleCustomerEvent(event.data.object, deps.stripeService) + break + } + case 'customer.subscription.created': + case 'customer.subscription.updated': + case 'customer.subscription.deleted': { + await handleSubscriptionEvent(event.data.object, deps.stripeService) + deps.metrics?.stripeSubscriptionEvent.add(1, { event_type: event.type.replace('customer.subscription.', '') }) + if (event.type === 'customer.subscription.deleted') + await captureSubscriptionCancelled(deps.posthog, deps.stripeService, event.data.object) + break + } + case 'invoice.created': + case 'invoice.updated': + case 'invoice.paid': + case 'invoice.payment_failed': { + await handleInvoiceEvent(event.data.object, deps.stripeService) + if (event.type === 'invoice.payment_failed') + deps.metrics?.stripePaymentFailed.add(1) + if (event.type === 'invoice.paid' && event.data.object.amount_paid && event.data.object.currency) { + deps.metrics?.stripeRevenue.add(event.data.object.amount_paid, { + currency: event.data.object.currency, + source: 'invoice', + }) + } + break + } + } + + return { received: true } + } +} + +async function handleCheckoutSessionCompleted( + stripeEventId: string, + session: Stripe.Checkout.Session, + fluxService: FluxService, + stripeService: StripeService, + billingService: BillingService, +): Promise<{ processed: boolean }> { + const userId = session.metadata?.userId + if (!userId) { + logger.withFields({ sessionId: session.id }).warn('Checkout session missing userId in metadata') + return { processed: false } + } + + logger.withFields({ userId, sessionId: session.id, mode: session.mode, amount: session.amount_total, currency: session.currency }).log('Processing checkout session') + + // Upsert customer record if we got a customer back. + if (session.customer) { + const stripeCustomerId = typeof session.customer === 'string' ? session.customer : session.customer.id + await stripeService.upsertCustomer({ + userId, + stripeCustomerId, + email: session.customer_email ?? undefined, + }) + await fluxService.updateStripeCustomerId(userId, stripeCustomerId) + } + + await stripeService.upsertCheckoutSession({ + userId, + stripeSessionId: session.id, + stripeCustomerId: typeof session.customer === 'string' ? session.customer : session.customer?.id, + mode: session.mode ?? 'payment', + status: session.status, + paymentStatus: session.payment_status, + amountTotal: session.amount_total, + currency: session.currency, + successUrl: session.success_url, + cancelUrl: session.cancel_url, + stripePaymentIntentId: typeof session.payment_intent === 'string' ? session.payment_intent : session.payment_intent?.id, + stripeSubscriptionId: typeof session.subscription === 'string' ? session.subscription : session.subscription?.id, + metadata: session.metadata ? JSON.stringify(session.metadata) : null, + expiresAt: session.expires_at ? new Date(session.expires_at * 1000) : null, + }) + + // Idempotent flux credit: use fluxCredited flag inside a transaction + // to prevent double-crediting on webhook replay. + // + // For `payment` mode (one-time Flux purchase) `metadata.fluxAmount` is + // required — without it we can't credit anything, and the funnel must + // not see a `payment_completed` event for a checkout that didn't + // actually deliver Flux. Non-`payment` modes (e.g. `setup` for saving + // a card) deliberately skip crediting and still count as processed. + if (session.mode === 'payment') { + if (session.amount_total == null) { + logger.withFields({ userId, sessionId: session.id }).warn('Payment-mode checkout missing amount_total; skipping credit and capture') + return { processed: false } + } + const metadataFlux = session.metadata?.fluxAmount + if (!metadataFlux) { + logger.withFields({ userId, sessionId: session.id }).warn('Payment-mode checkout missing metadata.fluxAmount; skipping credit and capture') + return { processed: false } + } + const fluxAmount = Number(metadataFlux) + if (!Number.isFinite(fluxAmount) || fluxAmount <= 0) { + logger.withFields({ userId, sessionId: session.id, metadataFlux }).warn('Invalid fluxAmount in session metadata, skipping credit') + return { processed: false } + } + + const result = await billingService.creditFluxFromStripeCheckout({ + stripeEventId, + userId, + stripeSessionId: session.id, + amountTotal: session.amount_total, + currency: session.currency, + fluxAmount, + }) + + logger.withFields({ + userId, + fluxAmount, + amountTotal: session.amount_total, + applied: result.applied, + balanceAfter: result.balanceAfter, + }).log('Processed flux credit for one-time payment') + } + + return { processed: true } +} + +async function capturePaymentCompleted( + posthog: PostHog | null | undefined, + session: Stripe.Checkout.Session, +): Promise { + if (!posthog) + return + + const userId = session.metadata?.userId + const email = session.customer_email + || (typeof session.customer_details?.email === 'string' ? session.customer_details.email : null) + || null + + // distinctId fallback chain: userId (Better Auth, matches browser identify()) + // > email (PostHog will merge on identify later) > stripe session id (last + // resort — orphan event but at least we count it). + const distinctId = userId || email || session.id + + const fluxAmount = Number(session.metadata?.fluxAmount) + const stripeCustomerId = typeof session.customer === 'string' ? session.customer : session.customer?.id + + await captureSafe(posthog, { + distinctId, + event: 'payment_completed', + properties: { + amount_total: session.amount_total, + currency: session.currency, + flux_amount: Number.isFinite(fluxAmount) ? fluxAmount : null, + mode: session.mode, + stripe_session_id: session.id, + stripe_customer_id: stripeCustomerId, + stripe_subscription_id: typeof session.subscription === 'string' ? session.subscription : session.subscription?.id, + ...(userId ? { user_id: userId } : {}), + // $set populates the PostHog person profile so funnel joins work even + // when a user pays via direct checkout link before ever loading the SPA. + ...(email ? { $set: { email } } : {}), + }, + }) +} + +async function captureSubscriptionCancelled( + posthog: PostHog | null | undefined, + stripeService: StripeService, + subscription: Stripe.Subscription, +): Promise { + if (!posthog) + return + + const stripeCustomerId = typeof subscription.customer === 'string' ? subscription.customer : subscription.customer.id + const customer = await stripeService.getCustomerByStripeId(stripeCustomerId) + const distinctId = customer?.userId || stripeCustomerId + + await captureSafe(posthog, { + distinctId, + event: 'subscription_cancelled', + properties: { + stripe_subscription_id: subscription.id, + stripe_customer_id: stripeCustomerId, + cancel_at_period_end: subscription.cancel_at_period_end, + cancellation_reason: subscription.cancellation_details?.reason ?? null, + cancellation_comment: subscription.cancellation_details?.comment ?? null, + canceled_at: subscription.canceled_at, + ended_at: subscription.ended_at, + ...(customer?.userId ? { user_id: customer.userId } : {}), + }, + }) +} + +async function handleCustomerEvent( + customer: Stripe.Customer | Stripe.DeletedCustomer, + stripeService: StripeService, +) { + if (customer.deleted) + return + + // Try to find existing customer to get userId. + const existing = await stripeService.getCustomerByStripeId(customer.id) + if (!existing) + return + + await stripeService.upsertCustomer({ + userId: existing.userId, + stripeCustomerId: customer.id, + email: customer.email ?? undefined, + name: customer.name ?? undefined, + }) +} + +async function handleSubscriptionEvent( + subscription: Stripe.Subscription, + stripeService: StripeService, +) { + const stripeCustomerId = typeof subscription.customer === 'string' ? subscription.customer : subscription.customer.id + const customer = await stripeService.getCustomerByStripeId(stripeCustomerId) + if (!customer) + return + + // In newer Stripe API, period info is on subscription items. + const firstItem = subscription.items.data[0] + await stripeService.upsertSubscription({ + userId: customer.userId, + stripeSubscriptionId: subscription.id, + stripeCustomerId, + stripePriceId: firstItem?.price?.id, + status: subscription.status, + currentPeriodStart: firstItem?.current_period_start ? new Date(firstItem.current_period_start * 1000) : null, + currentPeriodEnd: firstItem?.current_period_end ? new Date(firstItem.current_period_end * 1000) : null, + cancelAtPeriodEnd: subscription.cancel_at_period_end, + canceledAt: subscription.canceled_at ? new Date(subscription.canceled_at * 1000) : null, + endedAt: subscription.ended_at ? new Date(subscription.ended_at * 1000) : null, + metadata: subscription.metadata ? JSON.stringify(subscription.metadata) : null, + }) +} + +async function handleInvoiceEvent( + invoice: Stripe.Invoice, + stripeService: StripeService, +) { + const stripeCustomerId = typeof invoice.customer === 'string' ? invoice.customer : invoice.customer?.id + if (!stripeCustomerId) + return + + const customer = await stripeService.getCustomerByStripeId(stripeCustomerId) + if (!customer) + return + + // In newer Stripe API, subscription is under parent.subscription_details. + const subDetails = invoice.parent?.subscription_details + const subscriptionId = subDetails + ? (typeof subDetails.subscription === 'string' ? subDetails.subscription : subDetails.subscription?.id) + : undefined + + await stripeService.upsertInvoice({ + userId: customer.userId, + stripeInvoiceId: invoice.id, + stripeCustomerId, + stripeSubscriptionId: subscriptionId, + status: invoice.status, + amountDue: invoice.amount_due, + amountPaid: invoice.amount_paid, + currency: invoice.currency, + invoiceUrl: invoice.hosted_invoice_url, + invoicePdf: invoice.invoice_pdf, + periodStart: new Date(invoice.period_start * 1000), + periodEnd: new Date(invoice.period_end * 1000), + paidAt: invoice.status_transitions?.paid_at ? new Date(invoice.status_transitions.paid_at * 1000) : null, + metadata: invoice.metadata ? JSON.stringify(invoice.metadata) : null, + }) + + // TODO: implement subscription-based flux crediting when subscriptions are enabled + if (invoice.status === 'paid' && invoice.amount_paid && subscriptionId) + logger.withFields({ userId: customer.userId, invoiceId: invoice.id, amountPaid: invoice.amount_paid }).warn('Subscription invoice paid but flux crediting for subscriptions is not yet implemented') +} diff --git a/apps/server/src/routes/stripe/price-catalog.ts b/apps/server/src/routes/stripe/price-catalog.ts new file mode 100644 index 000000000..65159f2fc --- /dev/null +++ b/apps/server/src/routes/stripe/price-catalog.ts @@ -0,0 +1,140 @@ +import type Redis from 'ioredis' +import type Stripe from 'stripe' + +import { useLogger } from '@guiiai/logg' + +import { redisKeyFrom } from '../../utils/redis-keys' + +const logger = useLogger('stripe') + +const PRICES_CACHE_KEY = redisKeyFrom('cache', 'stripe', 'prices') +const PRICES_CACHE_TTL_SEC = 5 * 60 + +interface CachedCurrencyOption { + unitAmount: number | null +} + +export interface CachedPrice { + id: string + unitAmount: number | null + currency: string + product: string + active: boolean + metadata: Record + currencyOptions: Record +} + +export interface StripePriceCatalog { + getActivePrices: (productId: string) => Promise + findActivePrice: (productId: string, stripePriceId: string) => Promise +} + +/** + * Creates a Stripe price catalog backed by Redis. + * + * Use when: + * - Listing public Flux packages. + * - Validating checkout price ids before creating Stripe sessions. + * + * Expects: + * - A configured Stripe client and Redis connection. + * + * Returns: + * - Cached active prices for a single configured product. + */ +export function createStripePriceCatalog(stripe: Stripe, redis: Redis): StripePriceCatalog { + return { + async getActivePrices(productId: string): Promise { + const cached = await redis.get(PRICES_CACHE_KEY) + if (cached) { + try { + const parsed = JSON.parse(cached) as { productId: string, prices: CachedPrice[] } + if (parsed.productId === productId) + return parsed.prices + } + catch { /* corrupted cache, refetch */ } + } + + let result: Stripe.ApiList + try { + result = await stripe.prices.list({ product: productId, active: true, expand: ['data.currency_options'] }) + } + catch (err) { + logger.withError(err).warn('Failed to fetch prices from Stripe') + return [] + } + + const prices = result.data + .sort((a, b) => (a.unit_amount ?? 0) - (b.unit_amount ?? 0)) + .map(toCachedPrice) + + await redis.set(PRICES_CACHE_KEY, JSON.stringify({ productId, prices }), 'EX', PRICES_CACHE_TTL_SEC) + return prices + }, + + async findActivePrice(productId: string, stripePriceId: string): Promise { + // Validate against cached prices first, fall back to direct Stripe API. + const cachedPrices = await this.getActivePrices(productId) + const cached = cachedPrices.find(p => p.id === stripePriceId) + if (cached) + return cached + + // Cache miss — price may have just been created. + let fetched: Stripe.Price + try { + fetched = await stripe.prices.retrieve(stripePriceId) + } + catch { + return null + } + + const fetchedProductId = typeof fetched.product === 'string' ? fetched.product : fetched.product.id + if (!fetched.active || fetchedProductId !== productId) + return null + + // Invalidate cache so all instances pick up the new price. + await redis.del(PRICES_CACHE_KEY) + return toCachedPrice(fetched) + }, + } +} + +function toCachedPrice(price: Stripe.Price): CachedPrice { + return { + id: price.id, + unitAmount: price.unit_amount, + currency: price.currency, + product: typeof price.product === 'string' ? price.product : price.product.id, + active: price.active, + metadata: price.metadata, + currencyOptions: Object.fromEntries( + Object.entries(price.currency_options ?? {}).map(([cur, opt]) => [cur, { unitAmount: opt.unit_amount }]), + ), + } +} + +/** + * Format Stripe smallest-unit amount into a human-readable price string. + * + * Before: + * - `300, "usd"` + * - `500, "jpy"` + * + * After: + * - `"$3.00"` + * - `"¥500"` + */ +export function formatPrice(unitAmount: number | null, currency: string): string { + if (unitAmount == null) + return currency.toUpperCase() + + try { + const formatter = new Intl.NumberFormat('en-US', { style: 'currency', currency }) + const fractionDigits = formatter.resolvedOptions().minimumFractionDigits ?? 2 + const amount = unitAmount / (10 ** fractionDigits) + return formatter.format(amount) + } + catch { + return `${unitAmount / 100} ${currency.toUpperCase()}` + } +}