feat(premium): add App Store and Google Play purchases (#3052)

fluxer_api/src/api/worker/WorkerDependencies.ts

@@ -97,6 +97,8 @@ 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 type {StoreEntitlementService} from '@app/api/store_billing/StoreEntitlementService';
+import {createStoreEntitlementService} from '@app/api/store_billing/StoreEntitlementServiceFactory';
 import {getStripeClient} from '@app/api/stripe/StripeClient';
 import {PaymentRepository} from '@app/api/user/repositories/PaymentRepository';
 import type {UserRepository} from '@app/api/user/repositories/UserRepository';
@@ -168,6 +170,7 @@ export interface WorkerDependencies {
 	donationRepository: IDonationRepository;
 	guildService: GuildService;
 	billingRepository: BillingRepository;
+	storeEntitlementService: StoreEntitlementService;
 	stripe: Stripe | null;
 }
 
@@ -263,6 +266,14 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS
 		ipInfoService,
 	});
 	const billingRepository = new BillingRepository(snowflakeService, kvClient);
+	const storeEntitlementService = createStoreEntitlementService({
+		userRepository,
+		userCacheService,
+		gatewayService,
+		kvClient,
+		snowflakeService,
+		premiumStateReconciliationQueueService,
+	});
 	Logger.info('Worker dependencies initialized successfully');
 	return {
 		kvClient,
@@ -314,6 +325,7 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS
 		guildService,
 		donationRepository,
 		billingRepository,
+		storeEntitlementService,
 		guildAuditLogService,
 		contactChangeLogService,
 		ncmecSubmissionService,

fluxer_api/src/api/worker/WorkerLaneConfig.ts

@@ -49,6 +49,9 @@ const LANE_CONFIG = {
 			'harvestUserData',
 			'batchGuildAuditLogMessageDeletes',
 			'reconcileUserPayments',
+			'processAppStoreNotification',
+			'processGooglePlayNotification',
+			'refreshStorePurchase',
 			'bulkUpdateUserFlags',
 			'bulkUpdateSuspiciousActivityFlags',
 			'bulkScheduleUserDeletion',
@@ -77,6 +80,9 @@ const LANE_CONFIG = {
 			'processInactivityDeletions',
 			'processPendingBulkMessageDeletions',
 			'processPremiumStateReconciliationQueue',
+			'processStorePurchaseRefreshQueue',
+			'pollGooglePlayVoidedPurchases',
+			'pollAppStoreNotificationHistory',
 			'prunePostgresKvTtl',
 			'refreshSearchIndex',
 			'syncDiscoveryIndex',

fluxer_api/src/api/worker/WorkerMain.ts

@@ -68,6 +68,15 @@ function registerCronJobs(cron: CronScheduler, jobsStreamMaxAgeMs: number): void
 		ledger: false,
 	});
 	cron.upsert('processExpiredPremiumSweep', 'processExpiredPremiumSweep', {}, '0 0 * * * *', {ledger: false});
+	if (!Config.instance.selfHosted) {
+		cron.upsert('processStorePurchaseRefreshQueue', 'processStorePurchaseRefreshQueue', {}, '0 */5 * * * *', {
+			ledger: false,
+		});
+		cron.upsert('pollGooglePlayVoidedPurchases', 'pollGooglePlayVoidedPurchases', {}, '0 30 4 * * *', {ledger: false});
+		cron.upsert('pollAppStoreNotificationHistory', 'pollAppStoreNotificationHistory', {}, '0 45 4 * * *', {
+			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/WorkerTaskRegistry.ts

@@ -26,16 +26,22 @@ import harvestUserData from '@app/api/worker/tasks/HarvestUserData';
 import indexChannelMessages from '@app/api/worker/tasks/IndexChannelMessages';
 import indexGuildMembers from '@app/api/worker/tasks/IndexGuildMembers';
 import messageShred from '@app/api/worker/tasks/MessageShred';
+import pollAppStoreNotificationHistory from '@app/api/worker/tasks/PollAppStoreNotificationHistory';
+import pollGooglePlayVoidedPurchases from '@app/api/worker/tasks/PollGooglePlayVoidedPurchases';
+import processAppStoreNotification from '@app/api/worker/tasks/ProcessAppStoreNotification';
 import processAssetDeletionQueue from '@app/api/worker/tasks/ProcessAssetDeletionQueue';
 import processCachePurgeQueue from '@app/api/worker/tasks/ProcessCachePurgeQueue';
 import processExpiredPremiumSweep from '@app/api/worker/tasks/ProcessExpiredPremiumSweep';
+import processGooglePlayNotification from '@app/api/worker/tasks/ProcessGooglePlayNotification';
 import processInactivityDeletions from '@app/api/worker/tasks/ProcessInactivityDeletions';
 import processPendingBulkMessageDeletions from '@app/api/worker/tasks/ProcessPendingBulkMessageDeletions';
 import processPremiumStateReconciliationQueue from '@app/api/worker/tasks/ProcessPremiumStateReconciliationQueue';
+import processStorePurchaseRefreshQueue from '@app/api/worker/tasks/ProcessStorePurchaseRefreshQueue';
 import processStripeWebhook from '@app/api/worker/tasks/ProcessStripeWebhook';
 import prunePostgresKvTtl from '@app/api/worker/tasks/PrunePostgresKvTtl';
 import reconcileUserPayments from '@app/api/worker/tasks/ReconcileUserPayments';
 import refreshSearchIndex from '@app/api/worker/tasks/RefreshSearchIndex';
+import refreshStorePurchase from '@app/api/worker/tasks/RefreshStorePurchase';
 import {sendSystemDm} from '@app/api/worker/tasks/SendSystemDm';
 import syncDiscoveryIndex from '@app/api/worker/tasks/SyncDiscoveryIndex';
 import syncFileShaBlocklists from '@app/api/worker/tasks/SyncFileShaBlocklists';
@@ -79,6 +85,12 @@ export const workerTasks: Record<WorkerTaskName, WorkerTaskHandler> = {
 	processPendingBulkMessageDeletions,
 	processPremiumStateReconciliationQueue,
 	reconcileUserPayments,
+	processAppStoreNotification,
+	processGooglePlayNotification,
+	refreshStorePurchase,
+	processStorePurchaseRefreshQueue,
+	pollGooglePlayVoidedPurchases,
+	pollAppStoreNotificationHistory,
 	prunePostgresKvTtl,
 	refreshSearchIndex,
 	sendSystemDm,

fluxer_api/src/api/worker/tasks/HarvestUserData.test.ts

@@ -1,7 +1,11 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import {type ChannelID, createChannelID, createMessageID, createUserID, type MessageID} from '@app/api/BrandedTypes';
-import {harvestMessages} from '@app/api/worker/tasks/HarvestUserData';
+import {EMPTY_USER_ROW} from '@app/api/database/types/UserTypes';
+import {User} from '@app/api/models/User';
+import {mapStorePurchaseToResponse} from '@app/api/store_billing/StoreBillingMappers';
+import {buildStorePurchaseRow} from '@app/api/store_billing/tests/StoreBillingTestUtils';
+import {buildUserDataJson, harvestMessages} from '@app/api/worker/tasks/HarvestUserData';
 import {describe, expect, it} from 'vitest';
 
 const AUTHOR = createUserID(1000000000000000000n);
@@ -66,3 +70,35 @@ describe('harvestMessages', () => {
 		expect(result.channelMessagesMap.size).toBe(0);
 	});
 });
+
+describe('buildUserDataJson', () => {
+	it('exports the store purchases in their public shape', () => {
+		const purchase = buildStorePurchaseRow({user_id: AUTHOR});
+		const data = buildUserDataJson({
+			user: new User({...EMPTY_USER_ROW, user_id: AUTHOR, username: 'exporter', discriminator: 1}),
+			userId: AUTHOR,
+			productName: 'Fluxer',
+			authSessions: [],
+			relationships: [],
+			userNotes: new Map(),
+			userSettings: null,
+			guildMemberships: [],
+			guildSettings: [],
+			savedMessages: [],
+			privateChannels: [],
+			favoriteMemes: [],
+			pushSubscriptions: [],
+			webAuthnCredentials: [],
+			mfaBackupCodes: [],
+			createdGiftCodes: [],
+			payments: [],
+			storePurchases: [purchase],
+			oauthClients: [],
+			connections: [],
+			pinnedDms: [],
+			authorizedIps: [],
+			activityData: {last_active_at: null, last_active_ip: null},
+		});
+		expect(data.store_purchases).toEqual([mapStorePurchaseToResponse(purchase)]);
+	});
+});

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

@@ -29,6 +29,7 @@ import {
 	type SelfMessageFilter,
 } from '@app/api/channel/services/message/SelfMessageFilter';
 import type {UserConnectionRow} from '@app/api/database/types/ConnectionTypes';
+import type {StorePurchaseRow} from '@app/api/database/types/StoreBillingTypes';
 import type {IStorageService} from '@app/api/infrastructure/IStorageService';
 import {Logger} from '@app/api/Logger';
 import type {Application} from '@app/api/models/Application';
@@ -48,6 +49,7 @@ import type {User} from '@app/api/models/User';
 import type {UserGuildSettings} from '@app/api/models/UserGuildSettings';
 import type {UserSettings} from '@app/api/models/UserSettings';
 import type {WebAuthnCredential} from '@app/api/models/WebAuthnCredential';
+import {mapStorePurchaseToResponse} from '@app/api/store_billing/StoreBillingMappers';
 import {buildHarvestDownloadUrl} from '@app/api/user/services/HarvestDownloadUrl';
 import {mapWithConcurrency} from '@app/api/utils/ConcurrencyUtils';
 import {resolveSessionClientInfo} from '@app/api/utils/SessionClientIdentity';
@@ -132,6 +134,7 @@ interface UserDataJsonParams {
 	mfaBackupCodes: Array<MfaBackupCode>;
 	createdGiftCodes: Array<GiftCode>;
 	payments: Array<Payment>;
+	storePurchases: Array<StorePurchaseRow>;
 	oauthClients: Array<Application>;
 	connections: Array<UserConnectionRow>;
 	pinnedDms: Array<{
@@ -407,7 +410,7 @@ export async function harvestMessages(
 	return {channelMessagesMap, totalMessages};
 }
 
-function buildUserDataJson(params: UserDataJsonParams) {
+export function buildUserDataJson(params: UserDataJsonParams) {
 	const {
 		user,
 		userId,
@@ -426,6 +429,7 @@ function buildUserDataJson(params: UserDataJsonParams) {
 		mfaBackupCodes,
 		createdGiftCodes,
 		payments,
+		storePurchases,
 		oauthClients,
 		connections,
 		pinnedDms,
@@ -630,6 +634,7 @@ function buildUserDataJson(params: UserDataJsonParams) {
 			stripe_payment_intent_id: gift.stripePaymentIntentId,
 		})),
 		payments: payments.map(mapPayment),
+		store_purchases: storePurchases.map(mapStorePurchaseToResponse),
 		oauth_applications: oauthClients.map(mapOAuthApplication),
 		connections: connections.map((connection) => ({
 			id: connection.connection_id,
@@ -775,6 +780,7 @@ const harvestUserData: ArchiveTaskHandler = async (payload, helpers, attempt) =>
 		storageService,
 		emailService,
 		instanceConfigRepository,
+		storeEntitlementService,
 	} = dependencies;
 	const adminRequestedBy = validated.adminRequestedBy ? BigInt(validated.adminRequestedBy) : null;
 	const isAdminArchive = adminRequestedBy !== null;
@@ -846,6 +852,7 @@ const harvestUserData: ArchiveTaskHandler = async (payload, helpers, attempt) =>
 			mfaBackupCodes,
 			createdGiftCodes,
 			payments,
+			storePurchases,
 			oauthClients,
 			connections,
 			pinnedDms,
@@ -865,6 +872,7 @@ const harvestUserData: ArchiveTaskHandler = async (payload, helpers, attempt) =>
 			userRepository.listMfaBackupCodes(userId),
 			userRepository.findGiftCodesByCreator(userId),
 			paymentRepository.findPaymentsByUserId(userId),
+			storeEntitlementService.listStorePurchases(userId),
 			applicationRepository.listApplicationsByOwner(userId),
 			connectionRepository.findByUserId(userId),
 			userRepository.getPinnedDmsWithDetails(userId),
@@ -900,6 +908,7 @@ const harvestUserData: ArchiveTaskHandler = async (payload, helpers, attempt) =>
 			mfaBackupCodes,
 			createdGiftCodes,
 			payments,
+			storePurchases,
 			oauthClients,
 			connections,
 			pinnedDms,

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

@@ -0,0 +1,57 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import {getAppStoreServerApiClient} from '@app/api/middleware/ServiceSingletons';
+import {isAppStoreConfigured} from '@app/api/store_billing/StoreBillingConfig';
+import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {ms} from 'itty-time';
+
+const HISTORY_WINDOW_MS = ms('2 days');
+const MAX_PAGES_PER_QUERY = 50;
+const HISTORY_QUERIES: ReadonlyArray<{notificationType?: string; onlyFailures?: boolean}> = [
+	{onlyFailures: true},
+	{notificationType: 'REFUND'},
+	{notificationType: 'REFUND_REVERSED'},
+	{notificationType: 'REVOKE'},
+];
+
+const pollAppStoreNotificationHistory: WorkerTaskHandler = async (_payload, helpers) => {
+	if (Config.instance.selfHosted || !isAppStoreConfigured()) {
+		return;
+	}
+	const {workerService} = getWorkerDependencies();
+	const client = getAppStoreServerApiClient();
+	const endDate = new Date();
+	const startDate = new Date(endDate.getTime() - HISTORY_WINDOW_MS);
+	const seen = new Set<string>();
+	for (const app of Config.appStore.apps) {
+		for (const query of HISTORY_QUERIES) {
+			let paginationToken: string | null = null;
+			for (let page = 0; page < MAX_PAGES_PER_QUERY; page++) {
+				const result = await client.getNotificationHistory({
+					environment: 'production',
+					bundleId: app.bundleId,
+					startDate,
+					endDate,
+					paginationToken,
+					...query,
+				});
+				for (const signedPayload of result.signedPayloads) {
+					if (seen.has(signedPayload)) {
+						continue;
+					}
+					seen.add(signedPayload);
+					await workerService.addJob('processAppStoreNotification', {signedPayload});
+				}
+				if (!result.paginationToken) {
+					break;
+				}
+				paginationToken = result.paginationToken;
+			}
+		}
+	}
+	helpers.logger.info({enqueued: seen.size}, 'Polled App Store notification history');
+};
+
+export default pollAppStoreNotificationHistory;

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

@@ -0,0 +1,52 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import {getGooglePlayDeveloperApiClient} from '@app/api/middleware/ServiceSingletons';
+import {isGooglePlayConfigured} from '@app/api/store_billing/StoreBillingConfig';
+import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {ms} from 'itty-time';
+
+const VOIDED_WINDOW_MS = ms('2 days');
+const MAX_PAGES_PER_PACKAGE = 50;
+
+const pollGooglePlayVoidedPurchases: WorkerTaskHandler = async (_payload, helpers) => {
+	if (Config.instance.selfHosted || !isGooglePlayConfigured()) {
+		return;
+	}
+	const {storeEntitlementService} = getWorkerDependencies();
+	const client = getGooglePlayDeveloperApiClient();
+	const endTime = new Date();
+	const startTime = new Date(endTime.getTime() - VOIDED_WINDOW_MS);
+	let refreshed = 0;
+	for (const packageName of Config.googlePlay.packages) {
+		let pageToken: string | undefined;
+		for (let page = 0; page < MAX_PAGES_PER_PACKAGE; page++) {
+			const result = await client.listVoidedPurchases(packageName, {startTime, endTime, pageToken});
+			for (const voided of result.voidedPurchases) {
+				try {
+					const voidedAt = voided.voidedTimeMillis ? new Date(Number(voided.voidedTimeMillis)) : null;
+					const applied = await storeEntitlementService.applyGooglePlayVoided({
+						packageName,
+						purchaseToken: voided.purchaseToken,
+						orderId: voided.orderId ?? null,
+						productType: null,
+						voidedAt: voidedAt && !Number.isNaN(voidedAt.getTime()) ? voidedAt : null,
+					});
+					if (applied) {
+						refreshed += 1;
+					}
+				} catch (error) {
+					helpers.logger.warn({error, orderId: voided.orderId}, 'Failed to refresh a voided Google Play purchase');
+				}
+			}
+			if (!result.nextPageToken) {
+				break;
+			}
+			pageToken = result.nextPageToken;
+		}
+	}
+	helpers.logger.info({refreshed}, 'Polled Google Play voided purchases');
+};
+
+export default pollGooglePlayVoidedPurchases;

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

@@ -0,0 +1,112 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import {
+	AppStoreJwsVerificationError,
+	type AppStoreVerifiedNotification,
+	toStoreEnvironment,
+	verifyNotification,
+} from '@app/api/store_billing/app_store/AppStoreJwsVerifier';
+import {getAppStoreProductSlot, isAppStoreConfigured} from '@app/api/store_billing/StoreBillingConfig';
+import type {StoreEntitlementService} from '@app/api/store_billing/StoreEntitlementService';
+import {
+	appStoreNotificationClaimKey,
+	rescheduleStoreNotification,
+	StoreNotificationClaims,
+} from '@app/api/store_billing/StoreNotificationClaims';
+import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
+import {StorePurchaseInvalidError} from '@fluxer/errors/src/domains/payment/StorePurchaseInvalidError';
+import type {LoggerInterface} from '@fluxer/logger/src/LoggerInterface';
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {z} from 'zod';
+
+const PayloadSchema = z.object({
+	signedPayload: z.string().min(1),
+	retry: z.number().int().min(0).optional(),
+});
+
+async function handleNotification(
+	notification: AppStoreVerifiedNotification,
+	storeEntitlementService: StoreEntitlementService,
+	logger: LoggerInterface,
+): Promise<void> {
+	const context = {
+		notificationType: notification.notificationType,
+		subtype: notification.subtype,
+		notificationUUID: notification.notificationUUID,
+		environment: notification.environment,
+	};
+	if (notification.notificationType === 'TEST') {
+		logger.info(context, 'Received App Store test notification');
+		return;
+	}
+	const transaction = notification.transaction;
+	if (!transaction) {
+		logger.info(context, 'Acknowledged App Store notification without a transaction');
+		return;
+	}
+	if (!getAppStoreProductSlot(transaction.productId)) {
+		logger.info({...context, productId: transaction.productId}, 'Ignored App Store notification for another product');
+		return;
+	}
+	try {
+		await storeEntitlementService.syncAppStorePurchase({
+			environment: toStoreEnvironment(notification.environment),
+			bundleId: transaction.bundleId,
+			productId: transaction.productId,
+			transactionId: transaction.transactionId,
+			originalTransactionId: transaction.originalTransactionId,
+			hintAccountToken: notification.renewalInfo?.appAccountToken ?? transaction.appAccountToken ?? null,
+		});
+	} catch (error) {
+		if (error instanceof StorePurchaseInvalidError) {
+			logger.warn(context, 'App Store notification refers to a purchase that could not be verified');
+			return;
+		}
+		throw error;
+	}
+}
+
+const processAppStoreNotification: WorkerTaskHandler = async (payload, helpers) => {
+	const {signedPayload, retry = 0} = PayloadSchema.parse(payload);
+	if (Config.instance.selfHosted || !isAppStoreConfigured()) {
+		helpers.logger.warn('App Store billing is not configured; discarding notification');
+		return;
+	}
+	let notification: AppStoreVerifiedNotification;
+	try {
+		notification = await verifyNotification(signedPayload);
+	} catch (error) {
+		if (error instanceof AppStoreJwsVerificationError) {
+			helpers.logger.warn({reason: error.reason}, 'Discarded App Store notification that failed verification');
+			return;
+		}
+		throw error;
+	}
+	const {kvClient, storeEntitlementService} = getWorkerDependencies();
+	const claims = new StoreNotificationClaims(kvClient);
+	const claimKey = appStoreNotificationClaimKey(notification.notificationUUID);
+	const claim = await claims.tryClaim(claimKey);
+	if (claim === 'already_processed') {
+		return;
+	}
+	try {
+		if (claim === 'in_flight') {
+			throw new Error('App Store notification is in flight, retrying');
+		}
+		try {
+			await handleNotification(notification, storeEntitlementService, helpers.logger);
+			await claims.markProcessed(claimKey);
+		} catch (error) {
+			await claims.releaseClaim(claimKey);
+			throw error;
+		}
+	} catch (error) {
+		if (await rescheduleStoreNotification(helpers, 'processAppStoreNotification', {signedPayload}, retry, error)) {
+			return;
+		}
+		throw error;
+	}
+};
+
+export default processAppStoreNotification;

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

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import type {StorePurchaseRow} from '@app/api/database/types/StoreBillingTypes';
 import {mapGuildMemberToResponse} from '@app/api/guild/GuildModel';
 import type {IGuildRepositoryAggregate} from '@app/api/guild/repositories/IGuildRepositoryAggregate';
 import type {IGatewayService} from '@app/api/infrastructure/IGatewayService';
@@ -8,21 +9,35 @@ 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 {
+	isAppStoreConfigured,
+	isGooglePlayConfigured,
+	isGooglePlayPackageConfigured,
+} from '@app/api/store_billing/StoreBillingConfig';
+import {resolveStoreAccessEnd} from '@app/api/store_billing/StoreBillingMappers';
+import {isTerminalStorePurchaseState} from '@app/api/store_billing/StoreBillingTypes';
+import type {StoreEntitlementService} from '@app/api/store_billing/StoreEntitlementService';
 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';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
+import type {WorkerTaskName} from '@app/api/worker/WorkerLaneConfig';
 import {PremiumFlags, UserPremiumTypes} from '@fluxer/constants/src/UserConstants';
+import type {IWorkerService} from '@pkgs/worker/src/contracts/IWorkerService';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {ms} from 'itty-time';
 
 const BATCH_SIZE = 100;
+const STORE_REFRESH_STALE_MS = ms('30 minutes');
+const STORE_REFRESH_DEFER_LIMIT_MS = ms('1 day');
 
 interface SweepResult {
 	processed: number;
 	stripped: number;
 	sanitized: number;
 	reconcileEnqueued: number;
+	storeRefreshEnqueued: number;
 	skipped: number;
 	failed: number;
 }
@@ -33,6 +48,8 @@ interface SweepDeps {
 	userCacheService: UserCacheService;
 	gatewayService: IGatewayService;
 	premiumStateReconciliationQueueService: PremiumStateReconciliationQueueService;
+	storeEntitlementService: StoreEntitlementService;
+	workerService: IWorkerService<WorkerTaskName>;
 }
 
 async function sanitizeGuildMemberPerks(user: User, deps: SweepDeps): Promise<boolean> {
@@ -100,6 +117,44 @@ async function sanitizeGuildMemberPerks(user: User, deps: SweepDeps): Promise<bo
 	return true;
 }
 
+function isStoreProviderConfigured(row: StorePurchaseRow): boolean {
+	if (row.provider === 'app_store') {
+		return isAppStoreConfigured();
+	}
+	return isGooglePlayConfigured() && isGooglePlayPackageConfigured(row.app_id);
+}
+
+function shouldDeferToStoreRefresh(row: StorePurchaseRow, now: number): boolean {
+	if (row.kind !== 'subscription' || isTerminalStorePurchaseState(row.state) || !isStoreProviderConfigured(row)) {
+		return false;
+	}
+	const syncedMs = row.synced_at?.getTime() ?? 0;
+	const accessEndMs = resolveStoreAccessEnd(row)?.getTime();
+	if (accessEndMs === undefined) {
+		return false;
+	}
+	return (
+		syncedMs < now - STORE_REFRESH_STALE_MS &&
+		syncedMs < accessEndMs &&
+		accessEndMs > now - STORE_REFRESH_DEFER_LIMIT_MS
+	);
+}
+
+async function deferToStoreRefresh(user: User, deps: SweepDeps): Promise<number> {
+	const now = Date.now();
+	const rows = (await deps.storeEntitlementService.listStorePurchases(user.id)).filter((row) =>
+		shouldDeferToStoreRefresh(row, now),
+	);
+	for (const row of rows) {
+		await deps.workerService.addJob(
+			'refreshStorePurchase',
+			{storeKey: row.store_key},
+			{jobKey: `refreshStorePurchase:${row.store_key}`},
+		);
+	}
+	return rows.length;
+}
+
 async function processUser(user: User, deps: SweepDeps, result: SweepResult): Promise<void> {
 	if (user.isBot) {
 		result.skipped += 1;
@@ -136,6 +191,12 @@ async function processUser(user: User, deps: SweepDeps, result: SweepResult): Pr
 		result.skipped += 1;
 		return;
 	}
+	const storeRefreshes = await deferToStoreRefresh(user, deps);
+	if (storeRefreshes > 0) {
+		result.storeRefreshEnqueued += storeRefreshes;
+		result.skipped += 1;
+		return;
+	}
 	try {
 		let currentUser = await deps.userRepository.patchUpsert(user.id, createPremiumClearPatch(), user.toRow());
 		deps.userCacheService.setUserPartialResponseFromUserInBackground(currentUser);
@@ -183,6 +244,7 @@ async function processExpiredPremiumSweepCore(deps: SweepDeps): Promise<SweepRes
 		stripped: 0,
 		sanitized: 0,
 		reconcileEnqueued: 0,
+		storeRefreshEnqueued: 0,
 		skipped: 0,
 		failed: 0,
 	};
@@ -219,6 +281,7 @@ async function processExpiredPremiumSweepCore(deps: SweepDeps): Promise<SweepRes
 			stripped: result.stripped,
 			sanitized: result.sanitized,
 			reconcileEnqueued: result.reconcileEnqueued,
+			storeRefreshEnqueued: result.storeRefreshEnqueued,
 			skipped: result.skipped,
 			failed: result.failed,
 		},
@@ -229,14 +292,23 @@ async function processExpiredPremiumSweepCore(deps: SweepDeps): Promise<SweepRes
 
 const processExpiredPremiumSweep: WorkerTaskHandler = async (_payload, helpers) => {
 	helpers.logger.debug('Processing expired premium sweep task');
-	const {userRepository, guildRepository, userCacheService, gatewayService, premiumStateReconciliationQueueService} =
-		getWorkerDependencies();
+	const {
+		userRepository,
+		guildRepository,
+		userCacheService,
+		gatewayService,
+		premiumStateReconciliationQueueService,
+		storeEntitlementService,
+		workerService,
+	} = getWorkerDependencies();
 	await processExpiredPremiumSweepCore({
 		userRepository,
 		guildRepository,
 		userCacheService,
 		gatewayService,
 		premiumStateReconciliationQueueService,
+		storeEntitlementService,
+		workerService,
 	});
 };
 

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

@@ -0,0 +1,115 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import {
+	decodeGooglePlayDeveloperNotification,
+	type GooglePlayDeveloperNotification,
+} from '@app/api/store_billing/google_play/GooglePlayPushVerifier';
+import {isGooglePlayConfigured, isGooglePlayPackageConfigured} from '@app/api/store_billing/StoreBillingConfig';
+import type {StoreEntitlementService} from '@app/api/store_billing/StoreEntitlementService';
+import {
+	googlePlayMessageClaimKey,
+	rescheduleStoreNotification,
+	StoreNotificationClaims,
+} from '@app/api/store_billing/StoreNotificationClaims';
+import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
+import {StorePurchaseInvalidError} from '@fluxer/errors/src/domains/payment/StorePurchaseInvalidError';
+import type {LoggerInterface} from '@fluxer/logger/src/LoggerInterface';
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {z} from 'zod';
+
+const PayloadSchema = z.object({
+	messageId: z.string().min(1),
+	data: z.string().min(1),
+	retry: z.number().int().min(0).optional(),
+});
+
+async function handleNotification(
+	notification: GooglePlayDeveloperNotification,
+	storeEntitlementService: StoreEntitlementService,
+	logger: LoggerInterface,
+): Promise<void> {
+	const {packageName, event, eventTime} = notification;
+	switch (event.type) {
+		case 'test':
+			logger.info({packageName}, 'Received Google Play test notification');
+			return;
+		case 'subscription':
+			await storeEntitlementService.syncGooglePlaySubscription({packageName, purchaseToken: event.purchaseToken});
+			return;
+		case 'one_time_product':
+			await storeEntitlementService.syncGooglePlayProduct({
+				packageName,
+				purchaseToken: event.purchaseToken,
+				productId: event.productId,
+			});
+			return;
+		case 'voided_purchase':
+			await storeEntitlementService.applyGooglePlayVoided({
+				packageName,
+				purchaseToken: event.purchaseToken,
+				orderId: event.orderId,
+				productType: event.productType,
+				voidedAt: eventTime,
+			});
+			return;
+		case 'pending_refund_review':
+			logger.warn({packageName, orderId: event.orderId}, 'Google Play opened a refund review for an order');
+			return;
+		default:
+			logger.info({packageName}, 'Ignored Google Play notification of an unknown type');
+	}
+}
+
+const processGooglePlayNotification: WorkerTaskHandler = async (payload, helpers) => {
+	const {messageId, data, retry = 0} = PayloadSchema.parse(payload);
+	if (Config.instance.selfHosted || !isGooglePlayConfigured()) {
+		helpers.logger.warn('Google Play billing is not configured; discarding notification');
+		return;
+	}
+	const notification = decodeGooglePlayDeveloperNotification(data);
+	if (!notification) {
+		helpers.logger.warn({messageId}, 'Discarded a Google Play notification that could not be decoded');
+		return;
+	}
+	if (!isGooglePlayPackageConfigured(notification.packageName)) {
+		helpers.logger.warn(
+			{messageId, packageName: notification.packageName},
+			'Discarded a Google Play notification for another package',
+		);
+		return;
+	}
+	const {kvClient, storeEntitlementService} = getWorkerDependencies();
+	const claims = new StoreNotificationClaims(kvClient);
+	const claimKey = googlePlayMessageClaimKey(messageId);
+	const claim = await claims.tryClaim(claimKey);
+	if (claim === 'already_processed') {
+		return;
+	}
+	try {
+		if (claim === 'in_flight') {
+			throw new Error('Google Play notification is in flight, retrying');
+		}
+		try {
+			try {
+				await handleNotification(notification, storeEntitlementService, helpers.logger);
+			} catch (error) {
+				if (!(error instanceof StorePurchaseInvalidError)) {
+					throw error;
+				}
+				helpers.logger.warn({messageId}, 'Google Play notification refers to a purchase that could not be verified');
+			}
+			await claims.markProcessed(claimKey);
+		} catch (error) {
+			await claims.releaseClaim(claimKey);
+			throw error;
+		}
+	} catch (error) {
+		if (await rescheduleStoreNotification(helpers, 'processGooglePlayNotification', {messageId, data}, retry, error)) {
+			return;
+		}
+		throw error;
+	}
+};
+
+export default processGooglePlayNotification;

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

@@ -22,7 +22,14 @@ import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
 import type Stripe from 'stripe';
 
 interface ReconcileResult {
-	status: 'patched' | 'no_change' | 'skipped' | 'no_active_subscription' | 'stripped_no_subscription' | 'missing_user';
+	status:
+		| 'patched'
+		| 'no_change'
+		| 'skipped'
+		| 'no_active_subscription'
+		| 'stripped_no_subscription'
+		| 'store_entitled'
+		| 'missing_user';
 	patchedFields: Array<string>;
 }
 
@@ -274,6 +281,11 @@ async function reconcileUserPremiumStateFromStripe(params: {userId: UserID; stri
 		user,
 	);
 	if (!subscription) {
+		const {storeEntitlementService} = getWorkerDependencies();
+		if (await storeEntitlementService.getActiveStoreEntitlement(user.id)) {
+			await storeEntitlementService.applyStoreEntitlementToUser(user.id);
+			return {status: 'store_entitled', patchedFields: []};
+		}
 		const hasStalePremium = user.premiumType === UserPremiumTypes.SUBSCRIPTION;
 		const hasNonStripePremium =
 			(Config.instance.selfHosted && !isBillingActive()) || (user.premiumFlags & PremiumFlags.ENABLED_OVERRIDE) !== 0;
@@ -355,6 +367,7 @@ const processPremiumStateReconciliationQueue: WorkerTaskHandler = async (_payloa
 	let skippedCount = 0;
 	let noActiveSubscriptionCount = 0;
 	let strippedNoSubscriptionCount = 0;
+	let storeEntitledCount = 0;
 	let failedCount = 0;
 	let requeuedCount = 0;
 	let claimedElsewhereCount = 0;
@@ -393,6 +406,8 @@ const processPremiumStateReconciliationQueue: WorkerTaskHandler = async (_payloa
 					},
 					'Stripped expired premium with no active Stripe subscription via worker queue',
 				);
+			} else if (result.status === 'store_entitled') {
+				storeEntitledCount += 1;
 			} else if (result.status === 'no_change') {
 				noChangeCount += 1;
 			} else if (result.status === 'no_active_subscription') {
@@ -426,6 +441,7 @@ const processPremiumStateReconciliationQueue: WorkerTaskHandler = async (_payloa
 			skippedCount,
 			noActiveSubscriptionCount,
 			strippedNoSubscriptionCount,
+			storeEntitledCount,
 			failedCount,
 			requeuedCount,
 			claimedElsewhereCount,

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

@@ -0,0 +1,44 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import {isAppStoreConfigured, isGooglePlayConfigured} from '@app/api/store_billing/StoreBillingConfig';
+import {redactStoreKey} from '@app/api/store_billing/StoreBillingTypes';
+import {STORE_PURCHASE_REFRESH_QUEUE_KEY} from '@app/api/store_billing/StorePurchaseRefresh';
+import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {ms} from 'itty-time';
+
+const MAX_KEYS_PER_RUN = 200;
+const REFRESH_LEASE_MS = ms('10 minutes');
+
+const processStorePurchaseRefreshQueue: WorkerTaskHandler = async (_payload, helpers) => {
+	if (Config.instance.selfHosted || (!isAppStoreConfigured() && !isGooglePlayConfigured())) {
+		return;
+	}
+	const {kvClient, workerService} = getWorkerDependencies();
+	const now = Date.now();
+	const dueKeys = await kvClient.zrangebyscore(
+		STORE_PURCHASE_REFRESH_QUEUE_KEY,
+		'-inf',
+		now,
+		'LIMIT',
+		0,
+		MAX_KEYS_PER_RUN,
+	);
+	if (dueKeys.length === 0) {
+		return;
+	}
+	let enqueued = 0;
+	for (const storeKey of dueKeys) {
+		await kvClient.zadd(STORE_PURCHASE_REFRESH_QUEUE_KEY, now + REFRESH_LEASE_MS, storeKey);
+		try {
+			await workerService.addJob('refreshStorePurchase', {storeKey}, {jobKey: `refreshStorePurchase:${storeKey}`});
+			enqueued += 1;
+		} catch (error) {
+			helpers.logger.warn({error, storeKey: redactStoreKey(storeKey)}, 'Failed to enqueue a store purchase refresh');
+		}
+	}
+	helpers.logger.info({due: dueKeys.length, enqueued}, 'Enqueued store purchase refreshes');
+};
+
+export default processStorePurchaseRefreshQueue;

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

@@ -50,6 +50,7 @@ const processStripeWebhook: WorkerTaskHandler = async (payload, helpers) => {
 		productRegistry,
 		deps.cacheService,
 		deps.gatewayService,
+		deps.storeEntitlementService,
 	);
 	const giftService = new StripeGiftService(
 		stripe,
@@ -59,6 +60,7 @@ const processStripeWebhook: WorkerTaskHandler = async (payload, helpers) => {
 		checkoutService,
 		premiumService,
 		subscriptionService,
+		deps.storeEntitlementService,
 	);
 	const ageVerificationService = new AgeVerificationService(
 		stripe,
@@ -87,6 +89,7 @@ const processStripeWebhook: WorkerTaskHandler = async (payload, helpers) => {
 		deps.snowflakeService,
 		deps.billingRepository,
 		refundService,
+		deps.storeEntitlementService,
 	);
 	await webhookService.handleWebhook({body, signature});
 };

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

@@ -0,0 +1,30 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {Config} from '@app/api/Config';
+import {redactStoreKey} from '@app/api/store_billing/StoreBillingTypes';
+import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
+import {StorePurchaseInvalidError} from '@fluxer/errors/src/domains/payment/StorePurchaseInvalidError';
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {z} from 'zod';
+
+const PayloadSchema = z.object({
+	storeKey: z.string().min(1),
+});
+
+const refreshStorePurchase: WorkerTaskHandler = async (payload, helpers) => {
+	const {storeKey} = PayloadSchema.parse(payload);
+	if (Config.instance.selfHosted) {
+		return;
+	}
+	try {
+		await getWorkerDependencies().storeEntitlementService.refreshStorePurchase(storeKey);
+	} catch (error) {
+		if (error instanceof StorePurchaseInvalidError) {
+			helpers.logger.warn({storeKey: redactStoreKey(storeKey)}, 'Store purchase could not be verified during refresh');
+			return;
+		}
+		throw error;
+	}
+};
+
+export default refreshStorePurchase;

fluxer_api/src/api/worker/tasks/admin_bulk/AdminBulkServices.ts

@@ -43,6 +43,7 @@ export function createAdminBulkServices(deps: WorkerDependencies): AdminBulkServ
 		bulkMessageDeletionQueue: deps.bulkMessageDeletionQueueService,
 		stripe: deps.stripe,
 		reportService: getReportServiceInstance(),
+		storeEntitlementService: deps.storeEntitlementService,
 	});
 	const guildService = new AdminGuildService({
 		guildRepository: deps.guildRepository,

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

@@ -23,6 +23,7 @@ import {
 import {ReportStatus} from '@app/api/report/IReportRepository';
 import {ReportRepository} from '@app/api/report/ReportRepository';
 import {drainSearchTasks} from '@app/api/search/SearchTaskTracker';
+import {createTestStoreEntitlementService} from '@app/api/store_billing/tests/StoreBillingTestUtils';
 import {type ApiTestHarness, createApiTestHarness} from '@app/api/test/ApiTestHarness';
 import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
 import {HTTP_STATUS} from '@app/api/test/TestConstants';
@@ -70,6 +71,7 @@ function installWorkerDependencies(): void {
 		deletionQueueService: getKVAccountDeletionQueue(),
 		bulkMessageDeletionQueueService: getKVBulkMessageDeletionQueue(),
 		stripe: null,
+		storeEntitlementService: createTestStoreEntitlementService(),
 	});
 }
 

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

@@ -1,12 +1,22 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import {createUserID} from '@app/api/BrandedTypes';
 import {getConfig} from '@app/api/Config';
+import type {StorePurchaseRow} from '@app/api/database/types/StoreBillingTypes';
+import type {UserRow} from '@app/api/database/types/UserTypes';
+import {EMPTY_USER_ROW} from '@app/api/database/types/UserTypes';
 import {getCachedInstancePremiumMode, setCachedInstancePremiumMode} from '@app/api/limits/InstancePremiumModeCache';
+import {User} from '@app/api/models/User';
+import type {StoreEntitlementService} from '@app/api/store_billing/StoreEntitlementService';
+import {buildStorePurchaseRow} from '@app/api/store_billing/tests/StoreBillingTestUtils';
 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';
 import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import {UserPremiumTypes} from '@fluxer/constants/src/UserConstants';
+import type {IWorkerService} from '@pkgs/worker/src/contracts/IWorkerService';
 import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {ms} from 'itty-time';
 import {afterEach, describe, expect, test} from 'vitest';
 
 function createHarness() {
@@ -21,6 +31,54 @@ function createHarness() {
 	return {scanLimits};
 }
 
+const EXPIRED_USER_ID = createUserID(834271905123471362n);
+
+function createExpiredStoreUserHarness(storeRows: Array<StorePurchaseRow>) {
+	const user = new User({
+		...EMPTY_USER_ROW,
+		user_id: EXPIRED_USER_ID,
+		username: 'storelapse',
+		discriminator: 1,
+		premium_type: UserPremiumTypes.SUBSCRIPTION,
+		premium_since: new Date(Date.now() - ms('60 days')),
+		premium_until: new Date(Date.now() - ms('10 days')),
+		premium_grace_ends_at: new Date(Date.now() - ms('10 days')),
+	});
+	const patches: Array<Partial<UserRow>> = [];
+	const jobs: Array<{task: string; payload: unknown; jobKey: string | undefined}> = [];
+	const userRepository = {
+		async scanAllUsersPage(): Promise<{users: Array<User>; pageState: null}> {
+			return {users: [user], pageState: null};
+		},
+		async patchUpsert(_userId: unknown, patch: Partial<UserRow>): Promise<User> {
+			patches.push(patch);
+			return new User({...user.toRow(), ...patch});
+		},
+		async getUserGuildIds(): Promise<Array<never>> {
+			return [];
+		},
+	} as unknown as UserRepository;
+	const storeEntitlementService = {
+		async listStorePurchases() {
+			return storeRows;
+		},
+	} as unknown as StoreEntitlementService;
+	const workerService = {
+		async addJob(task: string, payload: unknown, options?: {jobKey?: string}) {
+			jobs.push({task, payload, jobKey: options?.jobKey});
+			return 0n;
+		},
+	} as unknown as IWorkerService<never>;
+	setWorkerDependenciesForTest({
+		userRepository,
+		storeEntitlementService,
+		workerService,
+		userCacheService: {setUserPartialResponseFromUserInBackground: () => {}} as never,
+		gatewayService: {dispatchPresence: async () => {}} as never,
+	});
+	return {patches, jobs};
+}
+
 function createHelpers(): WorkerTaskHelpers {
 	return {
 		logger: new NoopLogger(),
@@ -79,6 +137,110 @@ describe('processExpiredPremiumSweep', () => {
 		expect(harness.scanLimits).toEqual([100]);
 	});
 
+	test('refreshes a stale store subscription instead of stripping premium', async () => {
+		const row = buildStorePurchaseRow({
+			user_id: EXPIRED_USER_ID,
+			expires_at: new Date(Date.now() - ms('2 hours')),
+			synced_at: new Date(Date.now() - ms('3 hours')),
+		});
+		const harness = createExpiredStoreUserHarness([row]);
+
+		await withSelfHosted(false, async () => {
+			await processExpiredPremiumSweep({}, createHelpers());
+		});
+
+		expect(harness.patches).toEqual([]);
+		expect(harness.jobs).toEqual([
+			{
+				task: 'refreshStorePurchase',
+				payload: {storeKey: row.store_key},
+				jobKey: `refreshStorePurchase:${row.store_key}`,
+			},
+		]);
+	});
+
+	test('strips premium when the store subscription was synced recently', async () => {
+		const harness = createExpiredStoreUserHarness([
+			buildStorePurchaseRow({
+				user_id: EXPIRED_USER_ID,
+				expires_at: new Date(Date.now() - ms('10 days')),
+				synced_at: new Date(Date.now() - ms('5 minutes')),
+			}),
+			buildStorePurchaseRow({
+				user_id: EXPIRED_USER_ID,
+				state: 'expired',
+				entitled: false,
+				synced_at: new Date(Date.now() - ms('2 days')),
+			}),
+		]);
+
+		await withSelfHosted(false, async () => {
+			await processExpiredPremiumSweep({}, createHelpers());
+		});
+
+		expect(harness.jobs).toEqual([]);
+		expect(harness.patches[0]).toMatchObject({premium_type: null, premium_until: null});
+	});
+
+	test('strips premium once a sync after the access end confirmed the store state', async () => {
+		const harness = createExpiredStoreUserHarness([
+			buildStorePurchaseRow({
+				user_id: EXPIRED_USER_ID,
+				state: 'billing_retry',
+				entitled: false,
+				expires_at: new Date(Date.now() - ms('3 hours')),
+				synced_at: new Date(Date.now() - ms('1 hour')),
+			}),
+		]);
+
+		await withSelfHosted(false, async () => {
+			await processExpiredPremiumSweep({}, createHelpers());
+		});
+
+		expect(harness.jobs).toEqual([]);
+		expect(harness.patches[0]).toMatchObject({premium_type: null, premium_until: null});
+	});
+
+	test('strips premium when the store access ended more than a day ago', async () => {
+		const harness = createExpiredStoreUserHarness([
+			buildStorePurchaseRow({
+				user_id: EXPIRED_USER_ID,
+				expires_at: new Date(Date.now() - ms('2 days')),
+				synced_at: new Date(Date.now() - ms('3 days')),
+			}),
+		]);
+
+		await withSelfHosted(false, async () => {
+			await processExpiredPremiumSweep({}, createHelpers());
+		});
+
+		expect(harness.jobs).toEqual([]);
+		expect(harness.patches[0]).toMatchObject({premium_type: null, premium_until: null});
+	});
+
+	test('strips premium when the store provider is not configured', async () => {
+		const harness = createExpiredStoreUserHarness([
+			buildStorePurchaseRow({
+				user_id: EXPIRED_USER_ID,
+				expires_at: new Date(Date.now() - ms('2 hours')),
+				synced_at: new Date(Date.now() - ms('3 hours')),
+			}),
+		]);
+		const config = getConfig();
+		const originalEnabled = config.appStore.enabled;
+		try {
+			config.appStore.enabled = false;
+			await withSelfHosted(false, async () => {
+				await processExpiredPremiumSweep({}, createHelpers());
+			});
+		} finally {
+			config.appStore.enabled = originalEnabled;
+		}
+
+		expect(harness.jobs).toEqual([]);
+		expect(harness.patches[0]).toMatchObject({premium_type: null, premium_until: null});
+	});
+
 	test('scans users on a hosted instance', async () => {
 		const harness = createHarness();
 

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

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-import {createUserID} from '@app/api/BrandedTypes';
+import {createUserID, type UserID} from '@app/api/BrandedTypes';
 import type {UserRow} from '@app/api/database/types/UserTypes';
 import {EMPTY_USER_ROW} from '@app/api/database/types/UserTypes';
 import {PremiumStateReconciliationQueueService} from '@app/api/infrastructure/PremiumStateReconciliationQueueService';
@@ -91,11 +91,16 @@ function createStripeStub(subscription: Stripe.Subscription, invoices: Array<Str
 	} as unknown as Stripe;
 }
 
-function createCapturingDeps(user: User): {
+function createCapturingDeps(
+	user: User,
+	storeEntitlement: {expiresAt: Date} | null = null,
+): {
 	userRepository: UserRepository;
 	patches: Array<Partial<UserRow>>;
+	storeApplications: Array<UserID>;
 	extras: Record<string, unknown>;
 } {
+	const storeApplications: Array<UserID> = [];
 	const patches: Array<Partial<UserRow>> = [];
 	const userRepository = {
 		findUnique: async () => user,
@@ -107,8 +112,25 @@ function createCapturingDeps(user: User): {
 	const extras = {
 		userCacheService: {setUserPartialResponseFromUserInBackground: () => {}},
 		gatewayService: {dispatchPresence: async () => {}},
+		storeEntitlementService: {
+			getActiveStoreEntitlement: async () =>
+				storeEntitlement
+					? {
+							provider: 'app_store',
+							storePurchaseId: 1n,
+							entitledUntil: storeEntitlement.expiresAt,
+							expiresAt: storeEntitlement.expiresAt,
+							graceEndsAt: null,
+							autoRenew: true,
+							slot: 'monthly',
+						}
+					: null,
+			applyStoreEntitlementToUser: async (userId: UserID) => {
+				storeApplications.push(userId);
+			},
+		},
 	};
-	return {userRepository, patches, extras};
+	return {userRepository, patches, storeApplications, extras};
 }
 
 function createHelpers(): WorkerTaskHelpers {
@@ -322,6 +344,36 @@ describe('processPremiumStateReconciliationQueue', () => {
 		expect(patches[0].premium_until).toBeNull();
 		expect(patches[0].premium_since).toBeNull();
 	});
+	test('hands a user with an active App Store subscription and a stale Stripe customer to the store', async () => {
+		const queueService = createQueueService();
+		await queueService.enqueueUser(USER_ID, new Date(Date.now() - 1000));
+
+		const endedAtMs = Date.now() - 40 * ONE_DAY_MS;
+		const storeExpiresAt = new Date(Date.now() + 20 * ONE_DAY_MS);
+		const user = createPremiumUser({
+			premium_until: storeExpiresAt,
+			premium_billing_cycle: null,
+			premium_will_cancel: false,
+			stripe_subscription_id: null,
+		});
+		const {userRepository, patches, storeApplications, extras} = createCapturingDeps(user, {
+			expiresAt: storeExpiresAt,
+		});
+
+		setWorkerDependenciesForTest({
+			premiumStateReconciliationQueueService: queueService,
+			stripe: createStripeStub(createCancelledSubscription(endedAtMs), [createPaidInvoice(endedAtMs)]),
+			userRepository,
+			...extras,
+		});
+
+		await processPremiumStateReconciliationQueue({}, createHelpers());
+
+		expect(patches).toEqual([]);
+		expect(storeApplications).toEqual([USER_ID]);
+		expect(await queueService.getQueueSize()).toBe(0);
+	});
+
 	test('clears the perks-sanitized latch once the subscription is active again', async () => {
 		const queueService = createQueueService();
 		await queueService.enqueueUser(USER_ID, new Date(Date.now() - 1000));