feat(premium): let self-hosted instances sell premium and gifts (#3025)

fluxer_api/src/api/worker/WorkerDependencies.ts

@@ -97,7 +97,7 @@ import type {OAuth2TokenRepository} from '@app/api/oauth/repositories/OAuth2Toke
 import type {ReadStateRepository} from '@app/api/read_state/ReadStateRepository';
 import type {ReadStateService} from '@app/api/read_state/ReadStateService';
 import type {ReportRepository} from '@app/api/report/ReportRepository';
-import {STRIPE_API_VERSION} from '@app/api/stripe/StripeApiVersion';
+import {getStripeClient} from '@app/api/stripe/StripeClient';
 import {PaymentRepository} from '@app/api/user/repositories/PaymentRepository';
 import type {UserRepository} from '@app/api/user/repositories/UserRepository';
 import type {UserContactChangeLogService} from '@app/api/user/services/UserContactChangeLogService';
@@ -113,7 +113,7 @@ import type {IKVProvider} from '@pkgs/kv_client/src/IKVProvider';
 import type {RateLimitService} from '@pkgs/rate_limit/src/RateLimitService';
 import type {IVirusScanService} from '@pkgs/virus_scan/src/IVirusScanService';
 import type {IWorkerService} from '@pkgs/worker/src/contracts/IWorkerService';
-import Stripe from 'stripe';
+import type Stripe from 'stripe';
 
 export interface WorkerDependencies {
 	kvClient: IKVProvider;
@@ -263,14 +263,6 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS
 		ipInfoService,
 	});
 	const billingRepository = new BillingRepository(snowflakeService, kvClient);
-	let stripe: Stripe | null = null;
-	if (Config.stripe.enabled && Config.stripe.secretKey) {
-		stripe = new Stripe(Config.stripe.secretKey, {
-			apiVersion: STRIPE_API_VERSION,
-			httpClient: Config.dev.testModeEnabled ? Stripe.createFetchHttpClient() : undefined,
-		});
-		Logger.info('Stripe initialized');
-	}
 	Logger.info('Worker dependencies initialized successfully');
 	return {
 		kvClient,
@@ -325,6 +317,8 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS
 		guildAuditLogService,
 		contactChangeLogService,
 		ncmecSubmissionService,
-		stripe,
+		get stripe() {
+			return getStripeClient();
+		},
 	};
 }

fluxer_api/src/api/worker/WorkerMain.ts

@@ -56,9 +56,7 @@ function registerCronJobs(cron: CronScheduler, jobsStreamMaxAgeMs: number): void
 	cron.upsert('processPremiumStateReconciliationQueue', 'processPremiumStateReconciliationQueue', {}, '0 * * * * *', {
 		ledger: false,
 	});
-	if (!Config.instance.selfHosted) {
-		cron.upsert('processExpiredPremiumSweep', 'processExpiredPremiumSweep', {}, '0 0 * * * *', {ledger: false});
-	}
+	cron.upsert('processExpiredPremiumSweep', 'processExpiredPremiumSweep', {}, '0 0 * * * *', {ledger: false});
 	cron.upsert('processInactivityDeletions', 'processInactivityDeletions', {}, '0 0 */6 * * *', {ledger: false});
 	cron.upsert('expireAttachments', 'expireAttachments', {}, '0 0 */12 * * *', {ledger: false});
 	if (jobsStreamMaxAgeMs > 0 && jobsStreamMaxAgeMs <= JOBS_STREAM_MAX_AGE_MS) {

fluxer_api/src/api/worker/tasks/ProcessExpiredPremiumSweep.ts

@@ -1,6 +1,5 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-import {Config} from '@app/api/Config';
 import {mapGuildMemberToResponse} from '@app/api/guild/GuildModel';
 import type {IGuildRepositoryAggregate} from '@app/api/guild/repositories/IGuildRepositoryAggregate';
 import type {IGatewayService} from '@app/api/infrastructure/IGatewayService';
@@ -9,6 +8,7 @@ import type {UserCacheService} from '@app/api/infrastructure/UserCacheService';
 import {Logger} from '@app/api/Logger';
 import {createRequestCache} from '@app/api/middleware/RequestCacheMiddleware';
 import type {User} from '@app/api/models/User';
+import {isPremiumTieringActive} from '@app/api/stripe/BillingConfigCache';
 import type {IUserRepository} from '@app/api/user/IUserRepository';
 import {checkIsPremium, createPremiumClearPatch, shouldStripExpiredPremium} from '@app/api/user/UserHelpers';
 import {mapUserToPrivateResponse} from '@app/api/user/UserMappers';
@@ -186,8 +186,8 @@ async function processExpiredPremiumSweepCore(deps: SweepDeps): Promise<SweepRes
 		skipped: 0,
 		failed: 0,
 	};
-	if (Config.instance.selfHosted) {
-		Logger.debug('Skipping expired premium sweep on a self-hosted instance');
+	if (!isPremiumTieringActive()) {
+		Logger.debug('Skipping expired premium sweep because premium tiering is not active');
 		return result;
 	}
 	Logger.debug('Starting expired premium sweep');

fluxer_api/src/api/worker/tasks/ProcessPremiumStateReconciliationQueue.ts

@@ -6,6 +6,7 @@ import type {UserRow} from '@app/api/database/types/UserTypes';
 import {Logger} from '@app/api/Logger';
 import {getBillingRepository} from '@app/api/middleware/ServiceRegistry';
 import type {User} from '@app/api/models/User';
+import {isBillingActive} from '@app/api/stripe/BillingConfigCache';
 import {canProvisionPremiumFromSubscriptionStatus} from '@app/api/stripe/StripeSubscriptionAccessPolicy';
 import {
 	getInvoiceLatestLinePeriodEnd,
@@ -274,7 +275,8 @@ async function reconcileUserPremiumStateFromStripe(params: {userId: UserID; stri
 	);
 	if (!subscription) {
 		const hasStalePremium = user.premiumType === UserPremiumTypes.SUBSCRIPTION;
-		const hasNonStripePremium = Config.instance.selfHosted || (user.premiumFlags & PremiumFlags.ENABLED_OVERRIDE) !== 0;
+		const hasNonStripePremium =
+			(Config.instance.selfHosted && !isBillingActive()) || (user.premiumFlags & PremiumFlags.ENABLED_OVERRIDE) !== 0;
 		if (hasStalePremium && !hasNonStripePremium) {
 			const patch: Partial<UserRow> = {};
 			let effectivePremiumUntil = getEffectivePremiumUntil(user);

fluxer_api/src/api/worker/tasks/ProcessStripeWebhook.ts

@@ -2,7 +2,8 @@
 
 import {revokeAllAuthSessions} from '@app/api/auth/AuthSessionRevocation';
 import type {ISessionTerminator} from '@app/api/auth/ISessionTerminator';
-import {ProductRegistry} from '@app/api/stripe/ProductRegistry';
+import {Config} from '@app/api/Config';
+import {getProductRegistry} from '@app/api/stripe/ProductRegistry';
 import {AgeVerificationService} from '@app/api/stripe/services/AgeVerificationService';
 import {StripeCheckoutService} from '@app/api/stripe/services/StripeCheckoutService';
 import {StripeGiftService} from '@app/api/stripe/services/StripeGiftService';
@@ -22,11 +23,15 @@ const PayloadSchema = z.object({
 const processStripeWebhook: WorkerTaskHandler = async (payload, helpers) => {
 	const {body, signature} = PayloadSchema.parse(payload);
 	const deps = getWorkerDependencies();
-	if (!deps.stripe) {
+	const stripe = deps.stripe;
+	if (!stripe) {
+		if (Config.instance.selfHosted) {
+			throw new Error('Stripe is not configured on this worker yet; retrying webhook event');
+		}
 		helpers.logger.warn('Stripe is not configured; discarding webhook event');
 		return;
 	}
-	const productRegistry = new ProductRegistry();
+	const productRegistry = getProductRegistry();
 	const sessionTerminator: ISessionTerminator = {
 		async terminateAllUserSessions(userId) {
 			await revokeAllAuthSessions({users: deps.userRepository, gateway: deps.gatewayService}, userId);
@@ -38,21 +43,16 @@ const processStripeWebhook: WorkerTaskHandler = async (payload, helpers) => {
 		deps.guildRepository,
 		deps.guildService,
 	);
-	const checkoutService = new StripeCheckoutService(
-		deps.stripe,
-		deps.userRepository,
-		productRegistry,
-		deps.cacheService,
-	);
+	const checkoutService = new StripeCheckoutService(stripe, deps.userRepository, productRegistry, deps.cacheService);
 	const subscriptionService = new StripeSubscriptionService(
-		deps.stripe,
+		stripe,
 		deps.userRepository,
 		productRegistry,
 		deps.cacheService,
 		deps.gatewayService,
 	);
 	const giftService = new StripeGiftService(
-		deps.stripe,
+		stripe,
 		deps.userRepository,
 		deps.cacheService,
 		deps.gatewayService,
@@ -60,12 +60,15 @@ const processStripeWebhook: WorkerTaskHandler = async (payload, helpers) => {
 		premiumService,
 		subscriptionService,
 	);
-	const ageVerificationService = deps.stripe
-		? new AgeVerificationService(deps.stripe, deps.userRepository, deps.gatewayService, deps.cacheService)
-		: null;
-	const refundService = new StripeRefundService(deps.stripe, deps.userRepository, subscriptionService);
+	const ageVerificationService = new AgeVerificationService(
+		stripe,
+		deps.userRepository,
+		deps.gatewayService,
+		deps.cacheService,
+	);
+	const refundService = new StripeRefundService(stripe, deps.userRepository, subscriptionService);
 	const webhookService = new StripeWebhookService(
-		deps.stripe,
+		stripe,
 		checkoutService,
 		deps.userRepository,
 		deps.userCacheService,

fluxer_api/src/api/worker/tasks/ReconcileUserPayments.ts

@@ -1,12 +1,11 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import {createUserID} from '@app/api/BrandedTypes';
-import {Config} from '@app/api/Config';
 import {Logger} from '@app/api/Logger';
 import {mapGiftDurationMonthsToFields} from '@app/api/models/GiftCode';
 import type {Payment} from '@app/api/models/Payment';
 import type {User} from '@app/api/models/User';
-import {ProductRegistry} from '@app/api/stripe/ProductRegistry';
+import {getProductRegistry} from '@app/api/stripe/ProductRegistry';
 import {extractId} from '@app/api/stripe/StripeUtils';
 import type {PaymentRepository} from '@app/api/user/repositories/PaymentRepository';
 import {mapUserToPrivateResponse} from '@app/api/user/UserMappers';
@@ -71,7 +70,7 @@ async function reconcileCompletedGiftWithoutCode(payment: Payment, purchaser: Us
 			return;
 		}
 	}
-	const productRegistry = new ProductRegistry();
+	const productRegistry = getProductRegistry();
 	const productInfo = payment.priceId ? productRegistry.getProduct(payment.priceId) : null;
 	if (!productInfo) {
 		Logger.warn(
@@ -159,7 +158,7 @@ async function reconcileStuckGiftPayment(payment: Payment, purchaser: User, stri
 		}
 	}
 	if (!giftCode) {
-		const productRegistry = new ProductRegistry();
+		const productRegistry = getProductRegistry();
 		const productInfo = payment.priceId ? productRegistry.getProduct(payment.priceId) : null;
 		if (!productInfo) {
 			Logger.warn(
@@ -316,9 +315,6 @@ const reconcileUserPayments: WorkerTaskHandler = async (payload, helpers) => {
 		helpers.logger.debug('Stripe is disabled, skipping user payment reconciliation');
 		return;
 	}
-	if (!Config.stripe.enabled) {
-		return;
-	}
 	const userIdStr = payload.userId as string;
 	if (!userIdStr) {
 		helpers.logger.warn({payload}, 'Payment reconciliation task missing userId');

fluxer_api/src/api/worker/tests/ProcessExpiredPremiumSweep.test.ts

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import {getConfig} from '@app/api/Config';
+import {getCachedInstancePremiumMode, setCachedInstancePremiumMode} from '@app/api/limits/InstancePremiumModeCache';
 import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
 import type {UserRepository} from '@app/api/user/repositories/UserRepository';
 import processExpiredPremiumSweep from '@app/api/worker/tasks/ProcessExpiredPremiumSweep';
@@ -31,14 +32,21 @@ function createHelpers(): WorkerTaskHelpers {
 	};
 }
 
-async function withSelfHosted(selfHosted: boolean, callback: () => Promise<void>): Promise<void> {
+async function withSelfHosted(
+	selfHosted: boolean,
+	callback: () => Promise<void>,
+	premiumMode: 'mirror' | 'everyone' = 'everyone',
+): Promise<void> {
 	const config = getConfig();
 	const originalSelfHosted = config.instance.selfHosted;
+	const originalPremiumMode = getCachedInstancePremiumMode();
 	try {
 		config.instance.selfHosted = selfHosted;
+		setCachedInstancePremiumMode(premiumMode);
 		await callback();
 	} finally {
 		config.instance.selfHosted = originalSelfHosted;
+		setCachedInstancePremiumMode(originalPremiumMode);
 	}
 }
 
@@ -47,7 +55,7 @@ describe('processExpiredPremiumSweep', () => {
 		clearWorkerDependencies();
 	});
 
-	test('scans no users on a self-hosted instance', async () => {
+	test('scans no users on a self-hosted instance where everyone is premium', async () => {
 		const harness = createHarness();
 
 		await withSelfHosted(true, async () => {
@@ -57,6 +65,20 @@ describe('processExpiredPremiumSweep', () => {
 		expect(harness.scanLimits).toEqual([]);
 	});
 
+	test('scans users on a self-hosted instance in mirror mode', async () => {
+		const harness = createHarness();
+
+		await withSelfHosted(
+			true,
+			async () => {
+				await processExpiredPremiumSweep({}, createHelpers());
+			},
+			'mirror',
+		);
+
+		expect(harness.scanLimits).toEqual([100]);
+	});
+
 	test('scans users on a hosted instance', async () => {
 		const harness = createHarness();