diff --git a/fluxer_api/src/api/worker/WorkerMain.ts b/fluxer_api/src/api/worker/WorkerMain.ts index b66180072..58c907ec4 100644 --- a/fluxer_api/src/api/worker/WorkerMain.ts +++ b/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): boolean { diff --git a/fluxer_api/src/api/worker/tasks/ProcessExpiredPremiumSweep.ts b/fluxer_api/src/api/worker/tasks/ProcessExpiredPremiumSweep.ts index 7d1c28b49..a1314b9e2 100644 --- a/fluxer_api/src/api/worker/tasks/ProcessExpiredPremiumSweep.ts +++ b/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 = []; + 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): Promise { + 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]); + }); +});