/* * Backfill Stripe subscription end dates into public.stripe_info.canceled_at. * * Dry run for rows missing canceled_at: * bun run stripe:backfill-subscription-end-dates * * Apply missing end dates: * bun run stripe:backfill-subscription-end-dates --apply * * Refresh existing end dates and billing anchors too: * bun run stripe:backfill-subscription-end-dates --apply --refresh-existing */ import type Stripe from 'stripe' import type { Database } from '../supabase/functions/_backend/utils/supabase.types.ts' import { mkdir, writeFile } from 'node:fs/promises' import process from 'node:process' import { asyncPool, createStripeClient, createSupabaseServiceClient, DEFAULT_ENV_FILE, getArgValue, getRequiredEnv, isActionableStripeCustomerId, loadEnv, parsePositiveInteger } from './admin_stripe_backfill_utils.ts' const DEFAULT_CONCURRENCY = 8 const DEFAULT_PAGE_SIZE = 1000 const FAILURE_OUTPUT = './tmp/stripe_subscription_end_date_backfill_failures.json' type SupabaseClient = ReturnType type StripeInfoSubscriptionEndRow = Pick< Database['public']['Tables']['stripe_info']['Row'], 'canceled_at' | 'customer_id' | 'subscription_anchor_end' | 'subscription_anchor_start' | 'subscription_id' > export interface StripeSubscriptionEndBackfillCandidate { current_anchor_end: string | null current_anchor_start: string | null current_canceled_at: string | null customer_id: string next_anchor_end: string | null next_anchor_start: string | null next_canceled_at: string | null subscription_id: string } interface BackfillFailure { error: string subscriptionId: string | null customerId: string } function toIsoFromSeconds(seconds: number | null | undefined) { if (typeof seconds !== 'number' || !Number.isFinite(seconds)) return null return new Date(seconds * 1000).toISOString() } function getLicensedSubscriptionItem(subscription: Stripe.Subscription) { return subscription.items.data.find(item => item.plan?.usage_type === 'licensed') ?? subscription.items.data[0] ?? null } export function getStripeSubscriptionEndSnapshot(subscription: Stripe.Subscription) { const item = getLicensedSubscriptionItem(subscription) const anchorStart = toIsoFromSeconds(item?.current_period_start) const anchorEnd = toIsoFromSeconds(item?.current_period_end) const itemPeriodEnd = typeof item?.current_period_end === 'number' ? item.current_period_end : null const endedAtSeconds = subscription.ended_at ?? subscription.cancel_at ?? (subscription.cancel_at_period_end ? itemPeriodEnd : null) return { subscription_anchor_start: anchorStart, subscription_anchor_end: anchorEnd, canceled_at: toIsoFromSeconds(endedAtSeconds), } } function normalizeIso(value: string | null | undefined) { if (!value) return null const parsed = new Date(value) if (Number.isNaN(parsed.getTime())) return null return parsed.toISOString() } function hasSnapshotChanged(row: StripeInfoSubscriptionEndRow, snapshot: ReturnType, refreshExisting: boolean) { const canceledAtChanged = normalizeIso(row.canceled_at) !== snapshot.canceled_at const anchorsChanged = normalizeIso(row.subscription_anchor_start) !== snapshot.subscription_anchor_start || normalizeIso(row.subscription_anchor_end) !== snapshot.subscription_anchor_end if (refreshExisting) return canceledAtChanged || anchorsChanged return !row.canceled_at && !!snapshot.canceled_at } async function fetchStripeInfoRows( supabase: SupabaseClient, options: { customerId?: string | null missingOnly: boolean }, ) { const rows: StripeInfoSubscriptionEndRow[] = [] let lastSeenCustomerId: string | null = null while (true) { let query = supabase .from('stripe_info') .select('customer_id, subscription_id, subscription_anchor_start, subscription_anchor_end, canceled_at') .not('subscription_id', 'is', null) .order('customer_id', { ascending: true }) .limit(DEFAULT_PAGE_SIZE) if (options.customerId) query = query.eq('customer_id', options.customerId) else if (lastSeenCustomerId) query = query.gt('customer_id', lastSeenCustomerId) if (options.missingOnly) query = query.is('canceled_at', null) const { data, error } = await query if (error) throw error if (!data?.length) break rows.push(...data) if (options.customerId || data.length < DEFAULT_PAGE_SIZE) break lastSeenCustomerId = data.at(-1)?.customer_id ?? null } return rows } async function updateSubscriptionEndSnapshot(supabase: SupabaseClient, candidate: StripeSubscriptionEndBackfillCandidate) { const update: Database['public']['Tables']['stripe_info']['Update'] = { canceled_at: candidate.next_canceled_at, } if (candidate.next_anchor_start) update.subscription_anchor_start = candidate.next_anchor_start if (candidate.next_anchor_end) update.subscription_anchor_end = candidate.next_anchor_end const { error } = await supabase .from('stripe_info') .update(update) .eq('customer_id', candidate.customer_id) if (error) throw error } async function writeFailures(failures: BackfillFailure[]) { if (failures.length === 0) return await mkdir('./tmp', { recursive: true }) await writeFile(FAILURE_OUTPUT, `${JSON.stringify(failures, null, 2)}\n`) console.log(`Failure details written to ${FAILURE_OUTPUT}`) } async function main(args = process.argv.slice(2), runtimeEnv: Record = process.env) { const apply = args.includes('--apply') const refreshExisting = args.includes('--refresh-existing') const envFile = getArgValue(args, '--env-file') ?? DEFAULT_ENV_FILE const customerId = getArgValue(args, '--customer-id') const limitArg = getArgValue(args, '--limit') const limit = limitArg ? parsePositiveInteger(limitArg, '--limit', 0) : null const concurrency = parsePositiveInteger(getArgValue(args, '--concurrency'), '--concurrency', DEFAULT_CONCURRENCY) const fileEnv = await loadEnv(envFile) const env = { ...fileEnv, ...runtimeEnv, } const supabase = createSupabaseServiceClient(env) const stripe = createStripeClient( getRequiredEnv(env, 'STRIPE_SECRET_KEY'), env.STRIPE_API_BASE_URL?.trim(), ) const rows = await fetchStripeInfoRows(supabase, { customerId, missingOnly: !refreshExisting, }) const actionableRows = rows.filter(row => isActionableStripeCustomerId(row.customer_id) && !!row.subscription_id) const limitedRows = limit ? actionableRows.slice(0, limit) : actionableRows console.log(`Loaded ${rows.length} stripe_info rows (${actionableRows.length} actionable)`) console.log(`Env file: ${envFile}`) if (customerId) console.log(`Scoped to customer: ${customerId}`) if (refreshExisting) console.log('Mode: refresh canceled_at end dates and billing anchors') else console.log('Mode: fill missing canceled_at end dates only') if (!apply) console.log('Dry run only. Pass --apply to update stripe_info.') const failures: BackfillFailure[] = [] const candidates: StripeSubscriptionEndBackfillCandidate[] = [] let checked = 0 await asyncPool(concurrency, limitedRows, async (row) => { try { const subscriptionId = row.subscription_id! const subscription = await stripe.subscriptions.retrieve(subscriptionId, { expand: ['items.data.price'], }) const snapshot = getStripeSubscriptionEndSnapshot(subscription) if (hasSnapshotChanged(row, snapshot, refreshExisting)) { candidates.push({ customer_id: row.customer_id, subscription_id: subscriptionId, current_anchor_start: normalizeIso(row.subscription_anchor_start), current_anchor_end: normalizeIso(row.subscription_anchor_end), current_canceled_at: normalizeIso(row.canceled_at), next_anchor_start: snapshot.subscription_anchor_start, next_anchor_end: snapshot.subscription_anchor_end, next_canceled_at: snapshot.canceled_at, }) } checked++ if (checked % 100 === 0 || checked === limitedRows.length) console.log(`Checked ${checked}/${limitedRows.length}`) } catch (error) { failures.push({ customerId: row.customer_id, subscriptionId: row.subscription_id, error: error instanceof Error ? error.message : String(error), }) } }) console.log(`Candidates needing update: ${candidates.length}`) if (candidates.length > 0) { console.log('Sample updates:') for (const candidate of candidates.slice(0, 10)) { console.log(`${candidate.customer_id}: canceled_at ${candidate.current_canceled_at ?? 'null'} -> ${candidate.next_canceled_at ?? 'null'}, anchor_end ${candidate.current_anchor_end ?? 'null'} -> ${candidate.next_anchor_end ?? 'null'}`) } } if (!apply) { await writeFailures(failures) if (failures.length > 0) throw new Error(`Stripe subscription end-date backfill dry run had ${failures.length} failures`) return } let updated = 0 await asyncPool(concurrency, candidates, async (candidate) => { try { await updateSubscriptionEndSnapshot(supabase, candidate) updated++ if (updated % 100 === 0 || updated === candidates.length) console.log(`Updated ${updated}/${candidates.length}`) } catch (error) { failures.push({ customerId: candidate.customer_id, subscriptionId: candidate.subscription_id, error: error instanceof Error ? error.message : String(error), }) } }) await writeFailures(failures) if (failures.length > 0) throw new Error(`Stripe subscription end-date backfill had ${failures.length} failures`) console.log(`Done. Updated ${updated}/${candidates.length} stripe_info rows.`) } if (import.meta.main) await main()