perf(worker): batch the inactivity sweep activity lookups (#2215)

fluxer_api/src/api/infrastructure/KVActivityTracker.ts

@@ -11,6 +11,17 @@ const STATE_VERSION_KEY = 'activity_tracker:state_version';
 const STATE_VERSION_TTL_SECONDS = seconds('1 day');
 const REBUILD_BATCH_SIZE = 100;
 
+function parseActivityValue(value: string | null): Date | null {
+	if (!value) {
+		return null;
+	}
+	const timestamp = parseInt(value, 10);
+	if (Number.isNaN(timestamp)) {
+		return null;
+	}
+	return new Date(timestamp);
+}
+
 export class KVActivityTracker {
 	private kvClient: IKVProvider;
 	private isShuttingDown = false;
@@ -35,15 +46,19 @@ export class KVActivityTracker {
 
 	async getActivity(userId: UserID): Promise<Date | null> {
 		const key = this.getActivityKey(userId);
-		const value = await this.kvClient.get(key);
-		if (!value) {
-			return null;
+		return parseActivityValue(await this.kvClient.get(key));
+	}
+
+	async getActivities(userIds: ReadonlyArray<UserID>): Promise<Map<UserID, Date | null>> {
+		const activities = new Map<UserID, Date | null>();
+		if (userIds.length === 0) {
+			return activities;
 		}
-		const timestamp = parseInt(value, 10);
-		if (Number.isNaN(timestamp)) {
-			return null;
+		const values = await this.kvClient.mget(...userIds.map((userId) => this.getActivityKey(userId)));
+		for (const [index, userId] of userIds.entries()) {
+			activities.set(userId, parseActivityValue(values[index] ?? null));
 		}
-		return new Date(timestamp);
+		return activities;
 	}
 
 	async needsRebuild(): Promise<boolean> {

fluxer_api/src/api/user/tests/InactivityDeletion.test.ts

@@ -5,6 +5,7 @@ import {afterEach, beforeEach, describe, expect, test} from 'vitest';
 import {createTestAccount} from '../../auth/tests/AuthTestUtils';
 import {createUserID} from '../../BrandedTypes';
 import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness';
+import type {MockKVProvider} from '../../test/mocks/MockKVProvider';
 import {createBuilderWithoutAuth} from '../../test/TestRequestBuilder';
 import {UserRepository} from '../repositories/UserRepository';
 import {expectDataExists} from './UserTestUtils';
@@ -153,6 +154,30 @@ describe('Inactivity Deletion', () => {
 		expect(data1.hasSelfDeletedFlag).toBe(false);
 		expect(data2.hasSelfDeletedFlag).toBe(false);
 	});
+	test('a page issues one batched activity lookup instead of per-user reads', async () => {
+		const kvProvider = harness.kvProvider as MockKVProvider;
+		const accounts = [
+			await createTestAccount(harness),
+			await createTestAccount(harness),
+			await createTestAccount(harness),
+		];
+		const threeYearsAgo = new Date(Date.now() - 3 * 365 * 24 * 60 * 60 * 1000);
+		for (const account of accounts) {
+			await setUserActivity(harness, account.userId, threeYearsAgo);
+		}
+		kvProvider.mgetSpy.mockClear();
+		kvProvider.getSpy.mockClear();
+		await processInactivityDeletions(harness);
+		const activityBatches = kvProvider.mgetSpy.mock.calls.filter((keys) =>
+			keys.some((key) => String(key).startsWith('user_activity:')),
+		);
+		expect(activityBatches).toHaveLength(1);
+		for (const account of accounts) {
+			expect(activityBatches[0]).toContain(`user_activity:${account.userId}`);
+		}
+		const activityGets = kvProvider.getSpy.mock.calls.filter((call) => String(call[0]).startsWith('user_activity:'));
+		expect(activityGets).toHaveLength(0);
+	});
 	test('should return processing statistics', async () => {
 		const result = await processInactivityDeletions(harness);
 		expect(result).toHaveProperty('warnings_sent');

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

@@ -16,9 +16,11 @@ import type {User} from '../../models/User';
 import type {UserRepository} from '../../user/repositories/UserRepository';
 import {reschedulePendingDeletion} from '../../user/services/PendingDeletionCoordinator';
 import type {UserDeletionEligibilityService} from '../../user/services/UserDeletionEligibilityService';
+import {mapWithConcurrency} from '../../utils/ConcurrencyUtils';
 import {getWorkerDependencies} from '../WorkerContext';
 
 const BATCH_SIZE = 100;
+const USER_PROCESSING_CONCURRENCY = 8;
 
 interface InactivityCheckResult {
 	warningsSent: number;
@@ -63,7 +65,24 @@ interface ProcessUserDeps {
 	deletionEligibilityService: UserDeletionEligibilityService;
 }
 
-async function processUser(user: User, deps: ProcessUserDeps, result: InactivityCheckResult): Promise<void> {
+async function loadPageActivities(
+	activityTracker: KVActivityTracker,
+	users: ReadonlyArray<User>,
+): Promise<ReadonlyMap<UserID, Date | null> | null> {
+	try {
+		return await activityTracker.getActivities(users.map((user) => user.id));
+	} catch (error) {
+		Logger.warn({error}, 'Batched activity lookup failed, falling back to per-user activity lookups');
+		return null;
+	}
+}
+
+async function processUser(
+	user: User,
+	deps: ProcessUserDeps,
+	pageActivities: ReadonlyMap<UserID, Date | null> | null,
+	result: InactivityCheckResult,
+): Promise<void> {
 	const {
 		userRepository,
 		deletionQueueService,
@@ -85,7 +104,9 @@ async function processUser(user: User, deps: ProcessUserDeps, result: Inactivity
 		Logger.debug({userId}, 'User is an app store reviewer, skipping');
 		return;
 	}
-	const lastActivity = await activityTracker.getActivity(userId);
+	const prefetchedActivity = pageActivities?.get(userId);
+	const lastActivity =
+		prefetchedActivity !== undefined ? prefetchedActivity : await activityTracker.getActivity(userId);
 	const now = new Date();
 	const userInactiveMs = lastActivity ? now.getTime() - lastActivity.getTime() : Infinity;
 	if (userInactiveMs < ms('2 years')) {
@@ -182,14 +203,15 @@ export async function processInactivityDeletionsCore(
 		if (users.length === 0) {
 			break;
 		}
-		for (const user of users) {
+		const pageActivities = await loadPageActivities(activityTracker, users);
+		await mapWithConcurrency(users, USER_PROCESSING_CONCURRENCY, async (user) => {
 			try {
-				await processUser(user, userDeps, result);
+				await processUser(user, userDeps, pageActivities, result);
 			} catch (userError) {
 				Logger.error({error: userError, userId: user.id}, 'Failed to process inactive user');
 				result.errors++;
 			}
-		}
+		});
 		processedUsers += users.length;
 		pageState = page.pageState;
 		if (processedUsers % 1000 === 0) {