perf(worker): skip the premium sweep on self-hosted instances (#2195)

fluxer_api/src/api/worker/WorkerMain.ts

@@ -58,7 +58,9 @@ function registerCronJobs(cron: CronScheduler): void {
 		`*/${resolveScheduledJobDrainIntervalSeconds()} * * * * *`,
 		{ledger: false},
 	);
-	cron.upsert('processExpiredPremiumSweep', 'processExpiredPremiumSweep', {}, '0 0 * * * *', {ledger: false});
+	if (!Config.instance.selfHosted) {
+		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});
 	cron.upsert('prunePostgresKvTtl', 'prunePostgresKvTtl', {}, '0 */5 * * * *', {ledger: false});
@@ -69,7 +71,10 @@ function registerCronJobs(cron: CronScheduler): void {
 		cron.upsert('syncFileShaBlocklists', 'syncFileShaBlocklists', {}, '0 0 */12 * * *', {ledger: true});
 	}
 	cron.upsert('flushUserActivityBuffer', 'flushUserActivityBuffer', {}, '*/10 * * * * *', {ledger: false});
-	Logger.info({blocklistFeeds: Config.blocklistFeeds.enabled}, 'Cron jobs registered successfully');
+	Logger.info(
+		{blocklistFeeds: Config.blocklistFeeds.enabled, selfHosted: Config.instance.selfHosted},
+		'Cron jobs registered successfully',
+	);
 }
 
 function workerLanesRequireSearch(activeWorkerLanes: ReadonlyArray<WorkerLaneDefinition>): boolean {

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

@@ -105,10 +105,6 @@ async function processUser(user: User, deps: SweepDeps, result: SweepResult): Pr
 		result.skipped += 1;
 		return;
 	}
-	if (Config.instance.selfHosted) {
-		result.skipped += 1;
-		return;
-	}
 	if ((user.premiumType ?? 0) <= 0) {
 		result.skipped += 1;
 		return;
@@ -190,6 +186,10 @@ 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');
+		return result;
+	}
 	Logger.debug('Starting expired premium sweep');
 	let pageState: string | null = null;
 	while (true) {

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

@@ -0,0 +1,69 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterEach, describe, expect, test} from 'vitest';
+import {getConfig} from '../../Config';
+import {NoopLogger} from '../../test/mocks/NoopLogger';
+import type {UserRepository} from '../../user/repositories/UserRepository';
+import processExpiredPremiumSweep from '../tasks/ProcessExpiredPremiumSweep';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '../WorkerContext';
+
+function createHarness() {
+	const scanLimits: Array<number> = [];
+	const userRepository = {
+		async scanAllUsersPage(limit: number): Promise<{users: []; pageState: null}> {
+			scanLimits.push(limit);
+			return {users: [], pageState: null};
+		},
+	} as unknown as UserRepository;
+	setWorkerDependenciesForTest({userRepository});
+	return {scanLimits};
+}
+
+function createHelpers(): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: 1n,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+	};
+}
+
+async function withSelfHosted(selfHosted: boolean, callback: () => Promise<void>): Promise<void> {
+	const config = getConfig();
+	const originalSelfHosted = config.instance.selfHosted;
+	try {
+		config.instance.selfHosted = selfHosted;
+		await callback();
+	} finally {
+		config.instance.selfHosted = originalSelfHosted;
+	}
+}
+
+describe('processExpiredPremiumSweep', () => {
+	afterEach(() => {
+		clearWorkerDependencies();
+	});
+
+	test('scans no users on a self-hosted instance', async () => {
+		const harness = createHarness();
+
+		await withSelfHosted(true, async () => {
+			await processExpiredPremiumSweep({}, createHelpers());
+		});
+
+		expect(harness.scanLimits).toEqual([]);
+	});
+
+	test('scans users on a hosted instance', async () => {
+		const harness = createHarness();
+
+		await withSelfHosted(false, async () => {
+			await processExpiredPremiumSweep({}, createHelpers());
+		});
+
+		expect(harness.scanLimits).toEqual([100]);
+	});
+});