fix(admin): apply audit logs and side effects to bulk actions (#2758)

2026-10-18 13:35:42 +09:00 · 2026-09-14 14:34:43 +02:00
1 parent a9dc74a520
commit 91a2604e9e
@@ -31,7 +31,6 @@ import {
 	resolveCronSchedulerEnabled,
 	resolveWorkerLanes,
 	validateLaneCompleteness,
-	type WorkerLaneDefinition,
 } from '@app/api/worker/WorkerLaneConfig';
 import {createWorkerProcessErrorHandler} from '@app/api/worker/WorkerProcessErrorHandler';
 import {WorkerQueueOverflowError} from '@app/api/worker/WorkerQueueOverflowError';
@@ -45,13 +44,6 @@ import {getDefaultPostgresClient, initPostgres, shutdownPostgres} from '@pkgs/po
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
 import {ms} from 'itty-time';

-const SEARCH_REQUIRED_TASKS = new Set<string>([
-	'indexChannelMessages',
-	'indexGuildMembers',
-	'refreshSearchIndex',
-	'syncDiscoveryIndex',
-]);
-
 function registerCronJobs(cron: CronScheduler): void {
 	cron.upsert('processAssetDeletionQueue', 'processAssetDeletionQueue', {}, '0 */5 * * * *', {ledger: false});
 	if (Config.cachePurge.adapter !== 'none') {
@@ -87,10 +79,6 @@ function registerCronJobs(cron: CronScheduler): void {
 	);
 }

-function workerLanesRequireSearch(activeWorkerLanes: ReadonlyArray<WorkerLaneDefinition>): boolean {
-	return activeWorkerLanes.some((lane) => lane.taskTypes.some((taskType) => SEARCH_REQUIRED_TASKS.has(taskType)));
-}
-
 export async function startWorkerMain(): Promise<void> {
 	Logger.info('Starting worker backend...');
 	let cassandraInitialized = false;
@@ -240,6 +228,14 @@ export async function startWorkerMain(): Promise<void> {
 		setInjectedWorkerService(workerService);
 		instanceConfigRepository = getInstanceConfigRepository();
 		limitConfigService = getLimitConfigService();
+		try {
+			await initializeSearch(getCacheService());
+			searchInitialized = true;
+			Logger.info('Search initialised for worker backend');
+		} catch (error) {
+			Logger.error({err: error}, 'Search initialisation failed for worker backend');
+			throw error;
+		}
 		dependencies = await initializeWorkerDependencies(snowflakeService);
 		setWorkerDependencies(dependencies);
 		if (Config.blocklistFeeds.enabled) {
@@ -284,18 +280,6 @@ export async function startWorkerMain(): Promise<void> {
 			});
 			runners.push(runner);
 		}
-		if (workerLanesRequireSearch(activeWorkerLanes)) {
-			try {
-				await initializeSearch(getCacheService());
-				searchInitialized = true;
-				Logger.info('Search initialised for worker backend');
-			} catch (error) {
-				Logger.error({err: error}, 'Search initialisation failed for worker backend');
-				throw error;
-			}
-		} else {
-			Logger.info('Search initialisation skipped for worker lanes without search tasks');
-		}
 		if (cronSchedulerEnabled) {
 			cron.start();
 			Logger.info('Cron scheduler started');
@@ -432,10 +432,10 @@ export class WorkerRunner {
 			},
 		};
 		try {
-			await task(jobPayload, helpers);
+			const result = await task(jobPayload, helpers);
 			if (ledgerJobId !== null) {
 				try {
-					await this.ledger.markSucceeded(ledgerJobId, null);
+					await this.ledger.markSucceeded(ledgerJobId, result ?? null);
 				} catch (err) {
 					Logger.warn({err, jobId: ledgerJobId.toString()}, 'Ledger markSucceeded failed');
 				}
@@ -0,0 +1,102 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import type {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
+import type {UserID} from '@app/api/BrandedTypes';
+import {JobCancelledError, type WorkerTaskHelpers, type WorkerTaskResult} from '@pkgs/worker/src/contracts/WorkerTask';
+
+const MAX_RECORDED_FAILURES = 100;
+
+interface AdminBulkFailure {
+	id: string;
+	error: string;
+}
+
+interface AdminBulkJobParams {
+	helpers: WorkerTaskHelpers;
+	ids: ReadonlyArray<string>;
+	progressEvery: number;
+	apply: (id: string) => Promise<void>;
+	afterItems?: () => Promise<ReadonlyArray<[string, string]>>;
+	summary: {
+		auditService: AdminAuditService;
+		adminUserId: UserID;
+		action: string;
+		auditLogReason: string | null;
+		metadata: ReadonlyArray<[string, string]>;
+		target?: {type: string; id: bigint};
+	};
+}
+
+function describeError(error: unknown): string {
+	return error instanceof Error ? error.message : String(error);
+}
+
+export async function runAdminBulkJob({
+	helpers,
+	ids,
+	progressEvery,
+	apply,
+	afterItems,
+	summary,
+}: AdminBulkJobParams): Promise<WorkerTaskResult> {
+	const total = ids.length;
+	const failed: Array<AdminBulkFailure> = [];
+	let successfulCount = 0;
+	let processed = 0;
+	let cancelled = false;
+	for (const id of ids) {
+		if (await helpers.shouldCancel()) {
+			cancelled = true;
+			break;
+		}
+		try {
+			await apply(id);
+			successfulCount++;
+		} catch (error) {
+			failed.push({id, error: describeError(error)});
+			helpers.logger.warn({id, err: error}, 'Admin bulk job item failed');
+		}
+		processed++;
+		if (processed % progressEvery === 0) {
+			await helpers.reportProgress(processed, total, null);
+		}
+	}
+	const finalMetadata: Array<[string, string]> = [];
+	let finalizeError: string | null = null;
+	if (afterItems) {
+		try {
+			finalMetadata.push(...(await afterItems()));
+		} catch (error) {
+			finalizeError = describeError(error);
+			helpers.logger.error({err: error}, 'Admin bulk job finalisation failed');
+		}
+	}
+	await summary.auditService.createAuditLog({
+		adminUserId: summary.adminUserId,
+		targetType: summary.target?.type ?? 'bulk_job',
+		targetId: summary.target?.id ?? helpers.jobId,
+		action: summary.action,
+		auditLogReason: summary.auditLogReason,
+		metadata: new Map([
+			...summary.metadata,
+			['job_id', helpers.jobId.toString()],
+			...finalMetadata,
+			...(finalizeError !== null ? ([['finalize_error', finalizeError]] as Array<[string, string]>) : []),
+			['processed', processed.toString()],
+			['successful', successfulCount.toString()],
+			['failed', failed.length.toString()],
+			...(cancelled ? ([['cancelled', 'true']] as Array<[string, string]>) : []),
+		]),
+	});
+	if (cancelled) {
+		throw new JobCancelledError();
+	}
+	await helpers.reportProgress(total, total, `+${successfulCount} ok, ${failed.length} failed`);
+	helpers.logger.info({successful: successfulCount, failed: failed.length}, 'Admin bulk job complete');
+	return {
+		successful_count: successfulCount,
+		failed_count: failed.length,
+		failed: failed.slice(0, MAX_RECORDED_FAILURES),
+		...(finalizeError !== null ? {finalize_error: finalizeError} : {}),
+	};
+}
@@ -0,0 +1,66 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
+import {AdminBanManagementService} from '@app/api/admin/services/AdminBanManagementService';
+import {AdminGuildService} from '@app/api/admin/services/AdminGuildService';
+import {AdminUserService} from '@app/api/admin/services/AdminUserService';
+import {createApiContext} from '@app/api/CreateApiContext';
+import {
+	getHistoricalOutcomeRepository,
+	getIpInfoService,
+	getReportServiceInstance,
+	getSuspiciousIpRepository,
+} from '@app/api/middleware/ServiceMiddleware';
+import {
+	getDiscriminatorService,
+	getEntityAssetService,
+	getGuildDiscoveryRepository,
+	getInviteRepository,
+} from '@app/api/middleware/ServiceSingletons';
+import type {WorkerDependencies} from '@app/api/worker/WorkerDependencies';
+
+interface AdminBulkServices {
+	auditService: AdminAuditService;
+	banManagementService: AdminBanManagementService;
+	userService: AdminUserService;
+	guildService: AdminGuildService;
+}
+
+export function createAdminBulkServices(deps: WorkerDependencies): AdminBulkServices {
+	const apiContext = createApiContext();
+	const auditService = new AdminAuditService(deps.adminRepository, deps.snowflakeService);
+	const banManagementService = new AdminBanManagementService({
+		apiContext,
+		adminRepository: deps.adminRepository,
+		auditService,
+		ipInfoService: getIpInfoService(),
+		suspiciousIpRepository: getSuspiciousIpRepository(),
+	});
+	const userService = new AdminUserService({
+		apiContext,
+		guildRepository: deps.guildRepository,
+		channelRepository: deps.channelRepository,
+		discriminatorService: getDiscriminatorService(),
+		entityAssetService: getEntityAssetService(),
+		auditService,
+		userCacheService: deps.userCacheService,
+		banManagementService,
+		kvDeletionQueue: deps.deletionQueueService,
+		bulkMessageDeletionQueue: deps.bulkMessageDeletionQueueService,
+		stripe: deps.stripe,
+		riskHistoryRepository: getHistoricalOutcomeRepository(),
+		reportService: getReportServiceInstance(),
+	});
+	const guildService = new AdminGuildService({
+		guildRepository: deps.guildRepository,
+		userRepository: deps.userRepository,
+		channelRepository: deps.channelRepository,
+		inviteRepository: getInviteRepository(),
+		guildService: deps.guildService,
+		gatewayService: deps.gatewayService,
+		entityAssetService: getEntityAssetService(),
+		auditService,
+		discoveryRepository: getGuildDiscoveryRepository(),
+	});
+	return {auditService, banManagementService, userService, guildService};
+}
@@ -1,12 +1,11 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later

-import {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
-import {createGuildID, createUserID} from '@app/api/BrandedTypes';
+import {createUserID} from '@app/api/BrandedTypes';
 import {createRequestCache} from '@app/api/middleware/RequestCacheMiddleware';
+import {runAdminBulkJob} from '@app/api/worker/tasks/admin_bulk/AdminBulkJob';
+import {createAdminBulkServices} from '@app/api/worker/tasks/admin_bulk/AdminBulkServices';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
-import {JoinSourceTypes} from '@fluxer/constants/src/GuildConstants';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';

 interface Payload {
 	guild_id: string;
@@ -22,57 +21,37 @@ const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 		admin_user_id: rawPayload.admin_user_id as string,
 		audit_log_reason: (rawPayload.audit_log_reason as string | null) ?? null,
 	};
-	const deps = getWorkerDependencies();
-	const auditService = new AdminAuditService(deps.adminRepository, deps.snowflakeService);
+	const {auditService, guildService} = createAdminBulkServices(getWorkerDependencies());
 	const adminUserId = createUserID(BigInt(payload.admin_user_id));
-	const guildId = createGuildID(BigInt(payload.guild_id));
-	const userIds = payload.user_ids.map((id) => BigInt(id));
-	const total = userIds.length;
-	const successful: Array<string> = [];
-	const failed: Array<{
-		id: string;
-		error: string;
-	}> = [];
+	const guildId = BigInt(payload.guild_id);
+	const total = payload.user_ids.length;
 	await helpers.setContextLink(`/guilds/${guildId}`);
 	await helpers.reportProgress(0, total, `Adding ${total} members to guild ${guildId}`);
-	for (let i = 0; i < userIds.length; i++) {
-		if (await helpers.shouldCancel()) throw new JobCancelledError();
-		const userIdBigInt = userIds[i]!;
-		const userId = createUserID(userIdBigInt);
-		try {
-			await deps.guildService.members.addUserToGuild({
-				skipRiskGate: true,
-				userId,
-				guildId,
-				sendJoinMessage: false,
-				skipBanCheck: true,
-				joinSourceType: JoinSourceTypes.ADMIN_FORCE_ADD,
+	return await runAdminBulkJob({
+		helpers,
+		ids: payload.user_ids,
+		progressEvery: 25,
+		apply: async (id) => {
+			await guildService.membershipService.forceAddUserToGuild({
+				data: {guild_id: guildId, user_id: BigInt(id)},
 				requestCache: createRequestCache(),
-				initiatorId: adminUserId,
+				adminUserId,
+				auditLogReason: payload.audit_log_reason,
+				sendJoinMessage: false,
 			});
-			successful.push(userId.toString());
-		} catch (err) {
-			failed.push({id: userIdBigInt.toString(), error: err instanceof Error ? err.message : String(err)});
-		}
-		if ((i + 1) % 25 === 0) {
-			await helpers.reportProgress(i + 1, total, null);
-		}
-	}
-	await auditService.createAuditLog({
-		adminUserId,
-		targetType: 'guild',
-		targetId: BigInt(guildId),
-		action: 'bulk_add_guild_members',
-		auditLogReason: payload.audit_log_reason,
-		metadata: new Map([
-			['guild_id', guildId.toString()],
-			['user_count', total.toString()],
-			['successful', successful.length.toString()],
-			['failed', failed.length.toString()],
-		]),
+		},
+		summary: {
+			auditService,
+			adminUserId,
+			action: 'bulk_add_guild_members',
+			auditLogReason: payload.audit_log_reason,
+			target: {type: 'guild', id: guildId},
+			metadata: [
+				['guild_id', guildId.toString()],
+				['user_count', total.toString()],
+			],
+		},
 	});
-	await helpers.reportProgress(total, total, `+${successful.length} ok, ${failed.length} failed`);
-	helpers.logger.info({successful: successful.length, failed: failed.length}, 'bulkAddGuildMembers complete');
 };

 export default handler;
@@ -1,15 +1,11 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later

-import {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
 import {createUserID} from '@app/api/BrandedTypes';
-import {ContentBlocklistCategory, ContentBlocklistSeverity} from '@app/api/constants/ContentModeration';
-import {fileShaCache} from '@app/api/middleware/FileShaCache';
-import {getCacheService} from '@app/api/middleware/ServiceSingletons';
+import {runAdminBulkJob} from '@app/api/worker/tasks/admin_bulk/AdminBulkJob';
+import {createAdminBulkServices} from '@app/api/worker/tasks/admin_bulk/AdminBulkServices';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';

-const BANNED_FILE_SHAS_REFRESH_CHANNEL = 'banned_file_shas:refresh';
 const SHA256_RE = /^[0-9a-fA-F]{64}$/;

 interface Payload {
@@ -24,60 +20,35 @@ const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 		admin_user_id: rawPayload.admin_user_id as string,
 		audit_log_reason: (rawPayload.audit_log_reason as string | null) ?? null,
 	};
-	const deps = getWorkerDependencies();
-	const cacheService = getCacheService();
-	const auditService = new AdminAuditService(deps.adminRepository, deps.snowflakeService);
+	const {auditService, banManagementService} = createAdminBulkServices(getWorkerDependencies());
 	const adminUserId = createUserID(BigInt(payload.admin_user_id));
-	const total = payload.sha256_list.length;
-	const successful: Array<string> = [];
-	const failed: Array<{
-		id: string;
-		error: string;
-	}> = [];
+	const shas = payload.sha256_list.map((sha) => sha.toLowerCase());
 	await helpers.setContextLink('/file-sha-bans');
-	await helpers.reportProgress(0, total, `Banning ${total} file SHAs`);
-	for (let i = 0; i < payload.sha256_list.length; i++) {
-		if (await helpers.shouldCancel()) throw new JobCancelledError();
-		const sha = payload.sha256_list[i]!.toLowerCase();
-		if (!SHA256_RE.test(sha)) {
-			failed.push({id: sha, error: 'invalid_sha256'});
-			continue;
-		}
-		try {
-			await deps.adminRepository.banFileSha({
-				sha256_hex: sha,
-				category: ContentBlocklistCategory.MANUAL,
-				severity: ContentBlocklistSeverity.BLOCK,
-				content_type: null,
-				source_url: null,
-				added_at: new Date(),
-				added_by: adminUserId,
-				notes: null,
+	await helpers.reportProgress(0, shas.length, `Banning ${shas.length} file SHAs`);
+	return runAdminBulkJob({
+		helpers,
+		ids: shas,
+		progressEvery: 50,
+		apply: async (sha) => {
+			if (!SHA256_RE.test(sha)) {
+				throw new Error('invalid_sha256');
+			}
+			await banManagementService.banFileSha({sha256_hex: sha}, adminUserId, payload.audit_log_reason, {
+				deferRefresh: true,
 			});
-			fileShaCache.add(sha);
-			successful.push(sha);
-		} catch (err) {
-			failed.push({id: sha, error: err instanceof Error ? err.message : String(err)});
-		}
-		if ((i + 1) % 50 === 0) {
-			await helpers.reportProgress(i + 1, total, null);
-		}
-	}
-	await cacheService.publish(BANNED_FILE_SHAS_REFRESH_CHANNEL, 'refresh');
-	await auditService.createAuditLog({
-		adminUserId,
-		targetType: 'file_sha',
-		targetId: BigInt(0),
-		action: 'bulk_ban_file_shas',
-		auditLogReason: payload.audit_log_reason,
-		metadata: new Map([
-			['count', total.toString()],
-			['successful', successful.length.toString()],
-			['failed', failed.length.toString()],
-		]),
+		},
+		afterItems: async () => {
+			await banManagementService.publishFileShaRefresh();
+			return [];
+		},
+		summary: {
+			auditService,
+			adminUserId,
+			action: 'bulk_ban_file_shas',
+			auditLogReason: payload.audit_log_reason,
+			metadata: [['count', shas.length.toString()]],
+		},
 	});
-	await helpers.reportProgress(total, total, `+${successful.length} ok, ${failed.length} failed`);
-	helpers.logger.info({successful: successful.length, failed: failed.length}, 'bulkBanFileShas complete');
 };

 export default handler;
@@ -3,9 +3,9 @@
 import {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
 import {createUserID} from '@app/api/BrandedTypes';
 import {UserMessageDeletionService} from '@app/api/channel/services/message/UserMessageDeletionService';
+import {runAdminBulkJob} from '@app/api/worker/tasks/admin_bulk/AdminBulkJob';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';

 interface Payload {
 	user_ids: Array<string>;
@@ -29,18 +29,14 @@ const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 	});
 	const adminUserId = createUserID(BigInt(payload.admin_user_id));
 	const total = payload.user_ids.length;
-	const successful: Array<string> = [];
-	const failed: Array<{
-		id: string;
-		error: string;
-	}> = [];
 	let deletedMessages = 0;
 	await helpers.setContextLink(`/users?ids=${payload.user_ids.slice(0, 50).join(',')}`);
 	await helpers.reportProgress(0, total, `Deleting all messages from ${total} users`);
-	for (let i = 0; i < payload.user_ids.length; i++) {
-		if (await helpers.shouldCancel()) throw new JobCancelledError();
-		const rawUserId = payload.user_ids[i]!;
-		try {
+	return runAdminBulkJob({
+		helpers,
+		ids: payload.user_ids,
+		progressEvery: 1,
+		apply: async (rawUserId) => {
 			const userId = createUserID(BigInt(rawUserId));
 			const deleted = await deletionService.deleteUserMessagesBulk(userId);
 			deletedMessages += deleted;
@@ -49,33 +45,22 @@ const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 				targetType: 'message_deletion',
 				targetId: BigInt(userId),
 				action: 'delete_all_user_messages',
-				auditLogReason: null,
-				metadata: new Map([['message_count', deleted.toString()]]),
+				auditLogReason: payload.audit_log_reason,
+				metadata: new Map([
+					['user_id', userId.toString()],
+					['message_count', deleted.toString()],
+				]),
 			});
-			successful.push(rawUserId);
-		} catch (err) {
-			failed.push({id: rawUserId, error: err instanceof Error ? err.message : String(err)});
-		}
-		await helpers.reportProgress(i + 1, total, `${deletedMessages} messages deleted`);
-	}
-	await auditService.createAuditLog({
-		adminUserId,
-		targetType: 'message_deletion',
-		targetId: BigInt(0),
-		action: 'bulk_delete_user_messages',
-		auditLogReason: payload.audit_log_reason,
-		metadata: new Map([
-			['user_count', total.toString()],
-			['message_count', deletedMessages.toString()],
-			['successful', successful.length.toString()],
-			['failed', failed.length.toString()],
-		]),
+		},
+		afterItems: async () => [['message_count', deletedMessages.toString()]] as Array<[string, string]>,
+		summary: {
+			auditService,
+			adminUserId,
+			action: 'bulk_delete_user_messages',
+			auditLogReason: payload.audit_log_reason,
+			metadata: [['user_count', total.toString()]],
+		},
 	});
-	await helpers.reportProgress(total, total, `${deletedMessages} messages deleted, ${failed.length} users failed`);
-	helpers.logger.info(
-		{successful: successful.length, failed: failed.length, deletedMessages},
-		'bulkDeleteMessagesForUsers complete',
-	);
 };

 export default handler;
@@ -1,125 +1,65 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later

-import {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
-import {AdminUserUpdatePropagator} from '@app/api/admin/services/AdminUserUpdatePropagator';
+import {resolveDeletionDays} from '@app/api/admin/services/AdminUserDeletionService';
 import {createUserID} from '@app/api/BrandedTypes';
-import {reschedulePendingDeletion} from '@app/api/user/services/PendingDeletionCoordinator';
+import {runAdminBulkJob} from '@app/api/worker/tasks/admin_bulk/AdminBulkJob';
+import {createAdminBulkServices} from '@app/api/worker/tasks/admin_bulk/AdminBulkServices';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
-import {DeletionReasons} from '@fluxer/constants/src/Core';
-import {UserFlags} from '@fluxer/constants/src/UserConstants';
+import type {ScheduleAccountDeletionRequest} from '@fluxer/schema/src/domains/admin/AdminUserSchemas';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';

 interface Payload {
 	user_ids: Array<string>;
-	reason_code: number;
+	reason_code: ScheduleAccountDeletionRequest['reason_code'];
 	days_until_deletion: number;
 	public_reason: string | null;
 	admin_user_id: string;
 	audit_log_reason: string | null;
 }

-const MIN_USER_REQUESTED_DAYS = 14;
-const MIN_STANDARD_DAYS = 60;
 const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 	const payload: Payload = {
 		user_ids: rawPayload.user_ids as Array<string>,
-		reason_code: rawPayload.reason_code as number,
+		reason_code: rawPayload.reason_code as ScheduleAccountDeletionRequest['reason_code'],
 		days_until_deletion: rawPayload.days_until_deletion as number,
 		public_reason: (rawPayload.public_reason as string | null) ?? null,
 		admin_user_id: rawPayload.admin_user_id as string,
 		audit_log_reason: (rawPayload.audit_log_reason as string | null) ?? null,
 	};
-	const deps = getWorkerDependencies();
-	const auditService = new AdminAuditService(deps.adminRepository, deps.snowflakeService);
-	const propagator = new AdminUserUpdatePropagator({
-		userCacheService: deps.userCacheService,
-		userRepository: deps.userRepository,
-		guildRepository: deps.guildRepository,
-		gatewayService: deps.gatewayService,
-	});
+	const {auditService, userService} = createAdminBulkServices(getWorkerDependencies());
 	const adminUserId = createUserID(BigInt(payload.admin_user_id));
-	const userIds = payload.user_ids.map((id) => BigInt(id));
-	const minDays = payload.reason_code === DeletionReasons.USER_REQUESTED ? MIN_USER_REQUESTED_DAYS : MIN_STANDARD_DAYS;
-	const daysUntilDeletion = Math.max(payload.days_until_deletion, minDays);
-	const total = userIds.length;
-	const successful: Array<string> = [];
-	const failed: Array<{
-		id: string;
-		error: string;
-	}> = [];
-	await helpers.setContextLink(`/users?ids=${userIds.slice(0, 50).join(',')}`);
+	const total = payload.user_ids.length;
+	const daysUntilDeletion = resolveDeletionDays(payload.reason_code, payload.days_until_deletion);
+	await helpers.setContextLink(`/users?ids=${payload.user_ids.slice(0, 50).join(',')}`);
 	await helpers.reportProgress(0, total, `Scheduling deletion of ${total} users in ${daysUntilDeletion} days`);
-	for (let i = 0; i < userIds.length; i++) {
-		if (await helpers.shouldCancel()) throw new JobCancelledError();
-		const userIdBigInt = userIds[i]!;
-		const userId = createUserID(userIdBigInt);
-		try {
-			const user = await deps.userRepository.findUnique(userId);
-			if (!user) throw new Error('user_not_found');
-			const pendingDeletionAt = new Date();
-			pendingDeletionAt.setDate(pendingDeletionAt.getDate() + daysUntilDeletion);
-			const updatedUser = await deps.userRepository.updateDeletionSchedule(user, {
-				flags: user.flags | UserFlags.DELETED,
-				pending_deletion_at: pendingDeletionAt,
-				deletion_reason_code: payload.reason_code,
-				deletion_public_reason: payload.public_reason ?? null,
-				deletion_audit_log_reason: payload.audit_log_reason ?? null,
-			});
-			await reschedulePendingDeletion({
-				userId,
-				currentPendingDeletionAt: user.pendingDeletionAt,
-				nextPendingDeletionAt: pendingDeletionAt,
-				deletionReasonCode: payload.reason_code,
-				userRepository: deps.userRepository,
-				deletionQueue: deps.deletionQueueService,
-			});
-			await propagator.propagateUserUpdate({userId, oldUser: user, updatedUser});
-			if (user.email) {
-				try {
-					await deps.emailService.sendAccountScheduledForDeletionEmail(
-						user.email,
-						user.username,
-						payload.public_reason ?? null,
-						pendingDeletionAt,
-						user.locale,
-					);
-				} catch (emailErr) {
-					helpers.logger.warn({err: emailErr, userId: userId.toString()}, 'Failed to send deletion email');
-				}
-			}
-			await auditService.createAuditLog({
+	return await runAdminBulkJob({
+		helpers,
+		ids: payload.user_ids,
+		progressEvery: 10,
+		apply: async (userId) => {
+			await userService.deletionService.applyScheduledDeletion(
+				{
+					user_id: BigInt(userId),
+					reason_code: payload.reason_code,
+					public_reason: payload.public_reason ?? undefined,
+					days_until_deletion: payload.days_until_deletion,
+				},
 				adminUserId,
-				targetType: 'user',
-				targetId: BigInt(userId),
-				action: 'schedule_deletion',
-				auditLogReason: null,
-				metadata: new Map([['days', daysUntilDeletion.toString()]]),
-			});
-			successful.push(userId.toString());
-		} catch (err) {
-			failed.push({id: userIdBigInt.toString(), error: err instanceof Error ? err.message : String(err)});
-		}
-		if ((i + 1) % 10 === 0) {
-			await helpers.reportProgress(i + 1, total, null);
-		}
-	}
-	await auditService.createAuditLog({
-		adminUserId,
-		targetType: 'user',
-		targetId: BigInt(0),
-		action: 'bulk_schedule_deletion',
-		auditLogReason: payload.audit_log_reason,
-		metadata: new Map([
-			['user_count', total.toString()],
-			['reason_code', payload.reason_code.toString()],
-			['days', daysUntilDeletion.toString()],
-			['successful', successful.length.toString()],
-			['failed', failed.length.toString()],
-		]),
+				payload.audit_log_reason,
+			);
+		},
+		summary: {
+			auditService,
+			adminUserId,
+			action: 'bulk_schedule_deletion',
+			auditLogReason: payload.audit_log_reason,
+			metadata: [
+				['user_count', total.toString()],
+				['reason_code', payload.reason_code.toString()],
+				['days', daysUntilDeletion.toString()],
+			],
+		},
 	});
-	await helpers.reportProgress(total, total, `+${successful.length} ok, ${failed.length} failed`);
-	helpers.logger.info({successful: successful.length, failed: failed.length}, 'bulkScheduleUserDeletion complete');
 };

 export default handler;
@@ -1,12 +1,10 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later

-import {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
-import {AdminGuildUpdatePropagator} from '@app/api/admin/services/guild/AdminGuildUpdatePropagator';
 import {createGuildID, createUserID} from '@app/api/BrandedTypes';
-import {getGuildDiscoveryRepository} from '@app/api/middleware/ServiceSingletons';
+import {runAdminBulkJob} from '@app/api/worker/tasks/admin_bulk/AdminBulkJob';
+import {createAdminBulkServices} from '@app/api/worker/tasks/admin_bulk/AdminBulkServices';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';

 interface Payload {
 	guild_ids: Array<string>;
@@ -24,73 +22,36 @@ const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 		admin_user_id: rawPayload.admin_user_id as string,
 		audit_log_reason: (rawPayload.audit_log_reason as string | null) ?? null,
 	};
-	const deps = getWorkerDependencies();
-	const auditService = new AdminAuditService(deps.adminRepository, deps.snowflakeService);
-	const propagator = new AdminGuildUpdatePropagator({
-		gatewayService: deps.gatewayService,
-		discoveryRepository: getGuildDiscoveryRepository(),
-	});
+	const {auditService, guildService} = createAdminBulkServices(getWorkerDependencies());
 	const adminUserId = createUserID(BigInt(payload.admin_user_id));
-	const guildIds = payload.guild_ids.map((id) => BigInt(id));
-	const total = guildIds.length;
-	const successful: Array<string> = [];
-	const failed: Array<{
-		id: string;
-		error: string;
-	}> = [];
-	await helpers.setContextLink(`/guilds?ids=${guildIds.slice(0, 50).join(',')}`);
+	const total = payload.guild_ids.length;
+	await helpers.setContextLink(`/guilds?ids=${payload.guild_ids.slice(0, 50).join(',')}`);
 	await helpers.reportProgress(0, total, `Updating features on ${total} guilds`);
-	for (let i = 0; i < guildIds.length; i++) {
-		if (await helpers.shouldCancel()) throw new JobCancelledError();
-		const guildIdBigInt = guildIds[i]!;
-		const guildId = createGuildID(guildIdBigInt);
-		try {
-			const guild = await deps.guildRepository.findUnique(guildId);
-			if (!guild) throw new Error('guild_not_found');
-			const newFeatures = new Set(guild.features);
-			for (const f of payload.add_features) newFeatures.add(f);
-			for (const f of payload.remove_features) newFeatures.delete(f);
-			const updatedGuild = await deps.guildRepository.upsertPartial(guildId, {features: newFeatures}, guild.toRow());
-			await propagator.dispatchGuildUpdate(guildId, updatedGuild, {
+	return await runAdminBulkJob({
+		helpers,
+		ids: payload.guild_ids,
+		progressEvery: 25,
+		apply: async (id) => {
+			await guildService.updateService.updateGuildFeatures({
+				guildId: createGuildID(BigInt(id)),
+				addFeatures: payload.add_features,
+				removeFeatures: payload.remove_features,
 				adminUserId,
-				reconcileDiscoveryFeature: true,
+				auditLogReason: payload.audit_log_reason,
 			});
-			await auditService.createAuditLog({
-				adminUserId,
-				targetType: 'guild',
-				targetId: BigInt(guildId),
-				action: 'update_features',
-				auditLogReason: null,
-				metadata: new Map([
-					['add_features', payload.add_features.join(',')],
-					['remove_features', payload.remove_features.join(',')],
-					['new_features', Array.from(newFeatures).join(',')],
-				]),
-			});
-			successful.push(guildId.toString());
-		} catch (err) {
-			failed.push({id: guildIdBigInt.toString(), error: err instanceof Error ? err.message : String(err)});
-		}
-		if ((i + 1) % 25 === 0) {
-			await helpers.reportProgress(i + 1, total, null);
-		}
-	}
-	await auditService.createAuditLog({
-		adminUserId,
-		targetType: 'guild',
-		targetId: BigInt(0),
-		action: 'bulk_update_guild_features',
-		auditLogReason: payload.audit_log_reason,
-		metadata: new Map([
-			['guild_count', total.toString()],
-			['add_features', payload.add_features.join(',')],
-			['remove_features', payload.remove_features.join(',')],
-			['successful', successful.length.toString()],
-			['failed', failed.length.toString()],
-		]),
+		},
+		summary: {
+			auditService,
+			adminUserId,
+			action: 'bulk_update_guild_features',
+			auditLogReason: payload.audit_log_reason,
+			metadata: [
+				['guild_count', total.toString()],
+				['add_features', payload.add_features.join(',')],
+				['remove_features', payload.remove_features.join(',')],
+			],
+		},
 	});
-	await helpers.reportProgress(total, total, `+${successful.length} ok, ${failed.length} failed`);
-	helpers.logger.info({successful: successful.length, failed: failed.length}, 'bulkUpdateGuildFeatures complete');
 };

 export default handler;
@@ -1,12 +1,12 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later

-import {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
-import {AdminUserUpdatePropagator} from '@app/api/admin/services/AdminUserUpdatePropagator';
 import {createUserID} from '@app/api/BrandedTypes';
+import {runAdminBulkJob} from '@app/api/worker/tasks/admin_bulk/AdminBulkJob';
+import {createAdminBulkServices} from '@app/api/worker/tasks/admin_bulk/AdminBulkServices';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
-import {imposePhoneRequirements, SuspiciousActivityFlags} from '@fluxer/constants/src/UserConstants';
+import {ALL_SUSPICIOUS_ACTIVITY_FLAGS, SuspiciousActivityFlags} from '@fluxer/constants/src/UserConstants';
+import {UnknownUserError} from '@fluxer/errors/src/domains/user/UnknownUserError';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';

 interface Payload {
 	user_ids: Array<string>;
@@ -16,6 +16,25 @@ interface Payload {
 	audit_log_reason: string | null;
 }

+const PROGRESS_EVERY = 25;
+
+function resolveFlagMask(flagNames: ReadonlyArray<string>): number {
+	let mask = 0;
+	const unknownNames: Array<string> = [];
+	for (const flagName of flagNames) {
+		const value = SuspiciousActivityFlags[flagName as keyof typeof SuspiciousActivityFlags];
+		if (value === undefined) {
+			unknownNames.push(flagName);
+			continue;
+		}
+		mask |= value;
+	}
+	if (unknownNames.length > 0) {
+		throw new Error(`Unknown suspicious activity flag names: ${unknownNames.join(', ')}`);
+	}
+	return mask;
+}
+
 const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 	const payload: Payload = {
 		user_ids: rawPayload.user_ids as Array<string>,
@@ -24,78 +43,48 @@ const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 		admin_user_id: rawPayload.admin_user_id as string,
 		audit_log_reason: (rawPayload.audit_log_reason as string | null) ?? null,
 	};
+	const addMask = resolveFlagMask(payload.add_flags);
+	const removeMask = resolveFlagMask(payload.remove_flags);
 	const deps = getWorkerDependencies();
-	const auditService = new AdminAuditService(deps.adminRepository, deps.snowflakeService);
-	const propagator = new AdminUserUpdatePropagator({
-		userCacheService: deps.userCacheService,
-		userRepository: deps.userRepository,
-		guildRepository: deps.guildRepository,
-		gatewayService: deps.gatewayService,
-	});
+	const {auditService, userService} = createAdminBulkServices(deps);
 	const adminUserId = createUserID(BigInt(payload.admin_user_id));
-	const userIds = payload.user_ids.map((id) => BigInt(id));
-	const addMask = payload.add_flags.reduce((mask, name) => {
-		const v = SuspiciousActivityFlags[name as keyof typeof SuspiciousActivityFlags];
-		return v !== undefined ? mask | v : mask;
-	}, 0);
-	const removeMask = payload.remove_flags.reduce((mask, name) => {
-		const v = SuspiciousActivityFlags[name as keyof typeof SuspiciousActivityFlags];
-		return v !== undefined ? mask | v : mask;
-	}, 0);
-	const total = userIds.length;
-	const successful: Array<string> = [];
-	const failed: Array<{
-		id: string;
-		error: string;
-	}> = [];
-	await helpers.setContextLink(`/users?ids=${userIds.slice(0, 50).join(',')}`);
+	const acls = new Set<string>();
+	const total = payload.user_ids.length;
+	await helpers.setContextLink(`/users?ids=${payload.user_ids.slice(0, 50).join(',')}`);
 	await helpers.reportProgress(0, total, `Updating suspicious flags on ${total} users`);
-	for (let i = 0; i < userIds.length; i++) {
-		if (await helpers.shouldCancel()) throw new JobCancelledError();
-		const userIdBigInt = userIds[i]!;
-		const userId = createUserID(userIdBigInt);
-		try {
-			const user = await deps.userRepository.findUnique(userId);
-			if (!user) throw new Error('user_not_found');
+	return await runAdminBulkJob({
+		helpers,
+		ids: payload.user_ids,
+		progressEvery: PROGRESS_EVERY,
+		apply: async (id) => {
+			const userIdBigInt = BigInt(id);
+			const user = await deps.userRepository.findUnique(createUserID(userIdBigInt));
+			if (!user) {
+				throw new UnknownUserError();
+			}
 			const currentFlags = user.suspiciousActivityFlags ?? 0;
-			const newFlags = imposePhoneRequirements(currentFlags, addMask) & ~removeMask;
-			const updatedUser = await deps.userRepository.patchUpsert(
-				userId,
-				{suspicious_activity_flags: newFlags},
-				user.toRow(),
+			const flags = (currentFlags | addMask) & ~removeMask & ALL_SUSPICIOUS_ACTIVITY_FLAGS;
+			await userService.securityService.updateSuspiciousActivityFlags(
+				{user_id: userIdBigInt, flags},
+				adminUserId,
+				payload.audit_log_reason,
+				acls,
 			);
-			await propagator.propagateUserUpdate({userId, oldUser: user, updatedUser});
-			successful.push(userId.toString());
-		} catch (err) {
-			failed.push({id: userIdBigInt.toString(), error: err instanceof Error ? err.message : String(err)});
-		}
-		if ((i + 1) % 25 === 0) {
-			await helpers.reportProgress(i + 1, total, null);
-		}
-	}
-	await auditService.createAuditLog({
-		adminUserId,
-		targetType: 'user',
-		targetId: BigInt(0),
-		action: 'bulk_update_suspicious_activity_flags',
-		auditLogReason: payload.audit_log_reason,
-		metadata: new Map(
-			(
+		},
+		summary: {
+			auditService,
+			adminUserId,
+			action: 'bulk_update_suspicious_activity_flags',
+			auditLogReason: payload.audit_log_reason,
+			metadata: (
 				[
 					['user_count', total.toString()],
 					['add_flags', payload.add_flags.join(',')],
 					['remove_flags', payload.remove_flags.join(',')],
-					['successful', successful.length.toString()],
-					['failed', failed.length.toString()],
 				] as Array<[string, string]>
-			).filter(([_, v]) => v.length > 0),
-		),
+			).filter(([, value]) => value.length > 0),
+		},
 	});
-	await helpers.reportProgress(total, total, `+${successful.length} ok, ${failed.length} failed`);
-	helpers.logger.info(
-		{successful: successful.length, failed: failed.length},
-		'bulkUpdateSuspiciousActivityFlags complete',
-	);
 };

 export default handler;
@@ -1,11 +1,10 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later

-import {AdminAuditService} from '@app/api/admin/services/AdminAuditService';
-import {AdminUserUpdatePropagator} from '@app/api/admin/services/AdminUserUpdatePropagator';
 import {createUserID} from '@app/api/BrandedTypes';
+import {runAdminBulkJob} from '@app/api/worker/tasks/admin_bulk/AdminBulkJob';
+import {createAdminBulkServices} from '@app/api/worker/tasks/admin_bulk/AdminBulkServices';
 import {getWorkerDependencies} from '@app/api/worker/WorkerContext';
 import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
-import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';

 interface BulkUpdateUserFlagsPayload {
 	user_ids: Array<string>;
@@ -15,6 +14,8 @@ interface BulkUpdateUserFlagsPayload {
 	audit_log_reason: string | null;
 }

+const PROGRESS_EVERY = 25;
+
 const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 	const payload: BulkUpdateUserFlagsPayload = {
 		user_ids: rawPayload.user_ids as Array<string>,
@@ -24,81 +25,41 @@ const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
 		audit_log_reason: (rawPayload.audit_log_reason as string | null) ?? null,
 	};
 	const deps = getWorkerDependencies();
-	const auditService = new AdminAuditService(deps.adminRepository, deps.snowflakeService);
-	const propagator = new AdminUserUpdatePropagator({
-		userCacheService: deps.userCacheService,
-		userRepository: deps.userRepository,
-		guildRepository: deps.guildRepository,
-		gatewayService: deps.gatewayService,
-	});
+	const {auditService, userService} = createAdminBulkServices(deps);
 	const adminUserId = createUserID(BigInt(payload.admin_user_id));
-	const userIds = payload.user_ids.map((id) => BigInt(id));
-	const addFlags = payload.add_flags.map((f) => BigInt(f));
-	const removeFlags = payload.remove_flags.map((f) => BigInt(f));
-	const total = userIds.length;
-	const successful: Array<string> = [];
-	const failed: Array<{
-		id: string;
-		error: string;
-	}> = [];
-	await helpers.setContextLink(`/users?ids=${userIds.slice(0, 50).join(',')}`);
+	const addFlags = payload.add_flags.map((flag) => BigInt(flag));
+	const removeFlags = payload.remove_flags.map((flag) => BigInt(flag));
+	const acls = new Set<string>();
+	const total = payload.user_ids.length;
+	await helpers.setContextLink(`/users?ids=${payload.user_ids.slice(0, 50).join(',')}`);
 	await helpers.reportProgress(0, total, `Updating flags on ${total} users`);
-	for (let i = 0; i < userIds.length; i++) {
-		if (await helpers.shouldCancel()) throw new JobCancelledError();
-		const userIdBigInt = userIds[i]!;
-		const userId = createUserID(userIdBigInt);
-		try {
-			const user = await deps.userRepository.findUnique(userId);
-			if (!user) throw new Error('user_not_found');
-			let newFlags = user.flags;
-			for (const f of addFlags) newFlags |= f;
-			for (const f of removeFlags) newFlags &= ~f;
-			const updatedUser = await deps.userRepository.patchUpsert(userId, {flags: newFlags}, user.toRow());
-			await propagator.propagateUserUpdate({userId, oldUser: user, updatedUser});
-			await auditService.createAuditLog({
+	return await runAdminBulkJob({
+		helpers,
+		ids: payload.user_ids,
+		progressEvery: PROGRESS_EVERY,
+		apply: async (id) => {
+			await userService.securityService.updateUserFlags({
+				userId: createUserID(BigInt(id)),
+				data: {addFlags, removeFlags},
 				adminUserId,
-				targetType: 'user',
-				targetId: BigInt(userId),
-				action: 'update_flags',
-				auditLogReason: null,
-				metadata: new Map(
-					(
-						[
-							['add_flags', addFlags.map((f) => f.toString()).join(',')],
-							['remove_flags', removeFlags.map((f) => f.toString()).join(',')],
-							['new_flags', newFlags.toString()],
-						] as Array<[string, string]>
-					).filter(([_, v]) => v.length > 0),
-				),
+				auditLogReason: payload.audit_log_reason,
+				acls,
 			});
-			successful.push(userId.toString());
-		} catch (err) {
-			failed.push({id: userIdBigInt.toString(), error: err instanceof Error ? err.message : String(err)});
-		}
-		if ((i + 1) % 25 === 0) {
-			await helpers.reportProgress(i + 1, total, null);
-		}
-	}
-	await auditService.createAuditLog({
-		adminUserId,
-		targetType: 'user',
-		targetId: BigInt(0),
-		action: 'bulk_update_user_flags',
-		auditLogReason: payload.audit_log_reason,
-		metadata: new Map(
-			(
+		},
+		summary: {
+			auditService,
+			adminUserId,
+			action: 'bulk_update_user_flags',
+			auditLogReason: payload.audit_log_reason,
+			metadata: (
 				[
 					['user_count', total.toString()],
-					['add_flags', addFlags.map((f) => f.toString()).join(',')],
-					['remove_flags', removeFlags.map((f) => f.toString()).join(',')],
-					['successful', successful.length.toString()],
-					['failed', failed.length.toString()],
+					['add_flags', addFlags.map((flag) => flag.toString()).join(',')],
+					['remove_flags', removeFlags.map((flag) => flag.toString()).join(',')],
 				] as Array<[string, string]>
-			).filter(([_, v]) => v.length > 0),
-		),
+			).filter(([, value]) => value.length > 0),
+		},
 	});
-	await helpers.reportProgress(total, total, `+${successful.length} ok, ${failed.length} failed`);
-	helpers.logger.info({successful: successful.length, failed: failed.length}, 'bulkUpdateUserFlags task complete');
 };

 export default handler;
@@ -0,0 +1,311 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import type {AdminAuditLog} from '@app/api/admin/IAdminRepository';
+import {createTestAccount, setUserACLs, type TestAccount} from '@app/api/auth/tests/AuthTestUtils';
+import {createGuildID, createUserID} from '@app/api/BrandedTypes';
+import {createApiContext} from '@app/api/CreateApiContext';
+import {GuildDiscoveryRepository} from '@app/api/guild/repositories/GuildDiscoveryRepository';
+import {createGuild} from '@app/api/guild/tests/GuildTestUtils';
+import {DisabledLiveKitService} from '@app/api/infrastructure/DisabledLiveKitService';
+import {InMemoryVoiceRoomStore} from '@app/api/infrastructure/InMemoryVoiceRoomStore';
+import {getMessages} from '@app/api/message/tests/MessageTestUtils';
+import {createGuildStackServices} from '@app/api/middleware/GuildStackServiceFactory';
+import {getIpInfoService} from '@app/api/middleware/ServiceMiddleware';
+import {getGatewayService, getSnowflakeService, getVoiceAvailabilityService} from '@app/api/middleware/ServiceRegistry';
+import {
+	getAdminRepository,
+	getAssetDeletionQueue,
+	getAttachmentUploadTraceRepository,
+	getAvatarService,
+	getChannelRepository,
+	getEmbedService,
+	getEntityAssetService,
+	getFavoriteMemeRepository,
+	getGuildAuditLogService,
+	getGuildRepository,
+	getInviteRepository,
+	getKVAccountDeletionQueue,
+	getKVBulkMessageDeletionQueue,
+	getLimitConfigService,
+	getPurgeQueue,
+	getReadStateService,
+	getStorageService,
+	getUserCacheService,
+	getUserRepository,
+	getVirusScanServiceInstance,
+	getWebhookRepository,
+} from '@app/api/middleware/ServiceSingletons';
+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';
+import {createBuilder} from '@app/api/test/TestRequestBuilder';
+import bulkAddGuildMembers from '@app/api/worker/tasks/admin_bulk/BulkAddGuildMembers';
+import bulkUpdateGuildFeatures from '@app/api/worker/tasks/admin_bulk/BulkUpdateGuildFeatures';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import {APIErrorCodes} from '@fluxer/constants/src/ApiErrorCodes';
+import {MessageTypes} from '@fluxer/constants/src/ChannelConstants';
+import {DiscoveryApplicationStatus, DiscoveryCategories} from '@fluxer/constants/src/DiscoveryConstants';
+import {GuildFeatures} from '@fluxer/constants/src/GuildConstants';
+import type {GuildResponse} from '@fluxer/schema/src/domains/guild/GuildResponseSchemas';
+import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterEach, beforeEach, describe, expect, test} from 'vitest';
+
+const JOB_ID = 4242n;
+const MISSING_ID = '123456789012345678';
+
+interface BulkJobResult {
+	successful_count: number;
+	failed_count: number;
+	failed: Array<{
+		id: string;
+		error: string;
+	}>;
+}
+
+function createHelpers(): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: JOB_ID,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+	};
+}
+
+function installWorkerDependencies(): void {
+	const guildStack = createGuildStackServices({
+		apiContext: createApiContext(),
+		channelRepository: getChannelRepository(),
+		userRepository: getUserRepository(),
+		guildRepository: getGuildRepository(),
+		inviteRepository: getInviteRepository(),
+		webhookRepository: getWebhookRepository(),
+		favoriteMemeRepository: getFavoriteMemeRepository(),
+		avatarService: getAvatarService(),
+		entityAssetService: getEntityAssetService(),
+		assetDeletionQueue: getAssetDeletionQueue(),
+		userCacheService: getUserCacheService(),
+		limitConfigService: getLimitConfigService(),
+		embedService: getEmbedService(),
+		readStateService: getReadStateService(),
+		storageService: getStorageService(),
+		attachmentUploadTraceRepository: getAttachmentUploadTraceRepository(),
+		virusScanService: getVirusScanServiceInstance(),
+		purgeQueue: getPurgeQueue(),
+		guildAuditLogService: getGuildAuditLogService(),
+		voiceRoomStore: new InMemoryVoiceRoomStore(),
+		liveKitService: new DisabledLiveKitService(),
+		voiceAvailabilityService: getVoiceAvailabilityService(),
+		ipInfoService: getIpInfoService(),
+	});
+	setWorkerDependenciesForTest({
+		adminRepository: getAdminRepository(),
+		snowflakeService: getSnowflakeService(),
+		userRepository: getUserRepository(),
+		guildRepository: getGuildRepository(),
+		channelRepository: getChannelRepository(),
+		userCacheService: getUserCacheService(),
+		gatewayService: getGatewayService(),
+		deletionQueueService: getKVAccountDeletionQueue(),
+		bulkMessageDeletionQueueService: getKVBulkMessageDeletionQueue(),
+		guildService: guildStack.guildService,
+		stripe: null,
+	});
+}
+
+async function listAuditLogs(): Promise<Array<AdminAuditLog>> {
+	return getAdminRepository().listAllAuditLogsPaginated(500);
+}
+
+function findAuditLog(logs: Array<AdminAuditLog>, action: string, targetId?: bigint): AdminAuditLog | undefined {
+	return logs.find((log) => log.action === action && (targetId === undefined || log.targetId === targetId));
+}
+
+describe('admin bulk guild worker tasks', () => {
+	let harness: ApiTestHarness;
+	beforeEach(async () => {
+		harness = await createApiTestHarness({search: 'enabled'});
+		installWorkerDependencies();
+	});
+	afterEach(async () => {
+		clearWorkerDependencies();
+		await harness.shutdown();
+	});
+
+	async function createAdmin(acls: Array<string>): Promise<TestAccount> {
+		const admin = await createTestAccount(harness);
+		return setUserACLs(harness, admin, ['admin:authenticate', ...acls]);
+	}
+
+	async function createDiscoveryApplicant(admin: TestAccount, name: string): Promise<GuildResponse> {
+		const guild = await createGuild(harness, admin.token, name);
+		await createBuilder(harness, '').post(`/test/guilds/${guild.id}/member-count`).body({member_count: 10}).execute();
+		await createBuilder(harness, admin.token)
+			.post(`/guilds/${guild.id}/discovery`)
+			.body({description: 'A guild worth discovering', category_type: DiscoveryCategories.GAMING})
+			.expect(HTTP_STATUS.OK)
+			.execute();
+		return guild;
+	}
+
+	async function runFeaturesJob(
+		admin: TestAccount,
+		guildIds: Array<string>,
+		auditLogReason: string | null,
+	): Promise<BulkJobResult> {
+		const result = await bulkUpdateGuildFeatures(
+			{
+				guild_ids: guildIds,
+				add_features: [GuildFeatures.DISCOVERABLE],
+				remove_features: [],
+				admin_user_id: admin.userId,
+				audit_log_reason: auditLogReason,
+			},
+			createHelpers(),
+		);
+		return result as unknown as BulkJobResult;
+	}
+
+	async function runAddMembersJob(
+		admin: TestAccount,
+		guildId: string,
+		userIds: Array<string>,
+		auditLogReason: string | null,
+	): Promise<BulkJobResult> {
+		const result = await bulkAddGuildMembers(
+			{
+				guild_id: guildId,
+				user_ids: userIds,
+				admin_user_id: admin.userId,
+				audit_log_reason: auditLogReason,
+			},
+			createHelpers(),
+		);
+		return result as unknown as BulkJobResult;
+	}
+
+	test('a bulk guild-features job writes a per-guild update_features row with the admin reason', async () => {
+		const admin = await createAdmin(['guild:update:features']);
+		const guild = await createGuild(harness, admin.token, `Bulk Features Guild ${Date.now()}`);
+		const reason = 'Lilith ticket 4821';
+
+		const result = await runFeaturesJob(admin, [guild.id], reason);
+
+		expect(result.successful_count).toBe(1);
+		expect(result.failed_count).toBe(0);
+		const logs = await listAuditLogs();
+		const perGuild = findAuditLog(logs, 'update_features', BigInt(guild.id));
+		expect(perGuild?.targetType).toBe('guild');
+		expect(perGuild?.auditLogReason).toBe(reason);
+		expect(perGuild?.metadata.get('add_features')).toBe(GuildFeatures.DISCOVERABLE);
+		expect(perGuild?.metadata.get('new_features')).toContain(GuildFeatures.DISCOVERABLE);
+		const summary = findAuditLog(logs, 'bulk_update_guild_features');
+		expect(summary?.targetType).toBe('bulk_job');
+		expect(summary?.targetId).toBe(JOB_ID);
+		expect(summary?.auditLogReason).toBe(reason);
+		expect(summary?.metadata.get('guild_count')).toBe('1');
+		expect(summary?.metadata.get('add_features')).toBe(GuildFeatures.DISCOVERABLE);
+		expect(summary?.metadata.get('remove_features')).toBe('');
+		expect(summary?.metadata.get('successful')).toBe('1');
+		expect(summary?.metadata.get('failed')).toBe('0');
+		const updatedGuild = await getGuildRepository().findUnique(createGuildID(BigInt(guild.id)));
+		expect(updatedGuild?.features.has(GuildFeatures.DISCOVERABLE)).toBe(true);
+	});
+
+	test('a bulk guild-features job reconciles discovery exactly like the single-guild endpoint', async () => {
+		const admin = await createAdmin(['guild:update:features']);
+		const bulkGuild = await createDiscoveryApplicant(admin, `Bulk Discovery Guild ${Date.now()}`);
+		const singleGuild = await createDiscoveryApplicant(admin, `Single Discovery Guild ${Date.now()}`);
+		const reason = 'Lilith ticket 4822';
+
+		await runFeaturesJob(admin, [bulkGuild.id], reason);
+		await createBuilder(harness, admin.token)
+			.patch(`/admin/guilds/${singleGuild.id}`)
+			.header('X-Audit-Log-Reason', reason)
+			.body({add_features: [GuildFeatures.DISCOVERABLE]})
+			.expect(HTTP_STATUS.OK)
+			.execute();
+
+		const discoveryRepository = new GuildDiscoveryRepository();
+		const bulkRow = await discoveryRepository.findByGuildId(createGuildID(BigInt(bulkGuild.id)));
+		const singleRow = await discoveryRepository.findByGuildId(createGuildID(BigInt(singleGuild.id)));
+		expect(bulkRow?.status).toBe(DiscoveryApplicationStatus.APPROVED);
+		expect(singleRow?.status).toBe(DiscoveryApplicationStatus.APPROVED);
+		expect(bulkRow?.reviewed_by?.toString()).toBe(admin.userId);
+		expect(singleRow?.reviewed_by?.toString()).toBe(admin.userId);
+		expect(bulkRow?.review_reason).toBe(singleRow?.review_reason);
+	});
+
+	test('a bulk guild-features job surfaces an unknown guild in the job result', async () => {
+		const admin = await createAdmin(['guild:update:features']);
+		const guild = await createGuild(harness, admin.token, `Partial Features Guild ${Date.now()}`);
+
+		const result = await runFeaturesJob(admin, [guild.id, MISSING_ID], 'Lilith ticket 4823');
+
+		expect(result.successful_count).toBe(1);
+		expect(result.failed_count).toBe(1);
+		expect(result.failed).toEqual([{id: MISSING_ID, error: APIErrorCodes.UNKNOWN_GUILD}]);
+		const summary = findAuditLog(await listAuditLogs(), 'bulk_update_guild_features');
+		expect(summary?.metadata.get('failed')).toBe('1');
+	});
+
+	test('a bulk add-members job adds each member with a per-user force_add_to_guild row and no join message', async () => {
+		const admin = await createAdmin(['guild:force_add_member']);
+		const guild = await createGuild(harness, admin.token, `Bulk Members Guild ${Date.now()}`);
+		const bulkTarget = await createTestAccount(harness);
+		const singleTarget = await createTestAccount(harness);
+		const reason = 'Lilith ticket 4824';
+
+		const result = await runAddMembersJob(admin, guild.id, [bulkTarget.userId], reason);
+		await createBuilder(harness, admin.token)
+			.put(`/admin/guilds/${guild.id}/members/${singleTarget.userId}`)
+			.header('X-Audit-Log-Reason', reason)
+			.body(null)
+			.expect(HTTP_STATUS.OK)
+			.execute();
+
+		expect(result.successful_count).toBe(1);
+		expect(result.failed_count).toBe(0);
+		const member = await getGuildRepository().getMember(
+			createGuildID(BigInt(guild.id)),
+			createUserID(BigInt(bulkTarget.userId)),
+		);
+		expect(member).not.toBeNull();
+		const logs = await listAuditLogs();
+		const perUser = findAuditLog(logs, 'force_add_to_guild', BigInt(bulkTarget.userId));
+		expect(perUser?.targetType).toBe('user');
+		expect(perUser?.auditLogReason).toBe(reason);
+		expect(perUser?.metadata.get('guild_id')).toBe(guild.id);
+		const summary = findAuditLog(logs, 'bulk_add_guild_members');
+		expect(summary?.targetType).toBe('guild');
+		expect(summary?.targetId).toBe(BigInt(guild.id));
+		expect(summary?.auditLogReason).toBe(reason);
+		expect(summary?.metadata.get('job_id')).toBe(JOB_ID.toString());
+		expect(summary?.metadata.get('guild_id')).toBe(guild.id);
+		expect(summary?.metadata.get('user_count')).toBe('1');
+		expect(summary?.metadata.get('successful')).toBe('1');
+		const systemChannelId = guild.system_channel_id;
+		expect(systemChannelId).toBeTruthy();
+		const messages = await getMessages(harness, admin.token, String(systemChannelId));
+		const joinAuthorIds = messages
+			.filter((message) => message.type === MessageTypes.USER_JOIN)
+			.map((message) => message.author?.id);
+		expect(joinAuthorIds).toContain(singleTarget.userId);
+		expect(joinAuthorIds).not.toContain(bulkTarget.userId);
+	});
+
+	test('a bulk add-members job reports an unknown user as an unknown user, not an unknown guild', async () => {
+		const admin = await createAdmin(['guild:force_add_member']);
+		const guild = await createGuild(harness, admin.token, `Partial Members Guild ${Date.now()}`);
+		const target = await createTestAccount(harness);
+
+		const result = await runAddMembersJob(admin, guild.id, [target.userId, MISSING_ID], 'Lilith ticket 4825');
+
+		expect(result.successful_count).toBe(1);
+		expect(result.failed_count).toBe(1);
+		expect(result.failed).toEqual([{id: MISSING_ID, error: APIErrorCodes.UNKNOWN_USER}]);
+		const summary = findAuditLog(await listAuditLogs(), 'bulk_add_guild_members');
+		expect(summary?.metadata.get('failed')).toBe('1');
+	});
+});
@@ -0,0 +1,161 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import type {AdminAuditLog} from '@app/api/admin/IAdminRepository';
+import {createTestAccount, setUserACLs} from '@app/api/auth/tests/AuthTestUtils';
+import {BANNED_FILE_SHAS_REFRESH_CHANNEL} from '@app/api/constants/ContentModeration';
+import {getSnowflakeService, setInjectedWorkerService} from '@app/api/middleware/ServiceRegistry';
+import {getAdminRepository} from '@app/api/middleware/ServiceSingletons';
+import {type ApiTestHarness, createApiTestHarness} from '@app/api/test/ApiTestHarness';
+import type {MockKVProvider} from '@app/api/test/mocks/MockKVProvider';
+import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
+import {NoopWorkerService} from '@app/api/test/NoopWorkerService';
+import {SyncTaskWorkerService} from '@app/api/test/SyncTaskWorkerService';
+import {createBuilder} from '@app/api/test/TestRequestBuilder';
+import bulkBanFileShas from '@app/api/worker/tasks/admin_bulk/BulkBanFileShas';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import {AdminACLs} from '@fluxer/constants/src/AdminACLs';
+import {JobCancelledError, type WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterAll, beforeAll, beforeEach, describe, expect, it} from 'vitest';
+
+interface BulkJobResult {
+	successful_count: number;
+	failed_count: number;
+	failed: Array<{
+		id: string;
+		error: string;
+	}>;
+}
+
+const FIRST_SHA = 'a1'.repeat(32);
+const SECOND_SHA = 'b2'.repeat(32);
+const NON_HEX_SHA = 'zz'.repeat(32);
+
+function createHelpers(overrides: Partial<WorkerTaskHelpers> = {}): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: 4242n,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+		...overrides,
+	};
+}
+
+describe('Bulk ban file SHAs', () => {
+	let harness: ApiTestHarness;
+
+	beforeAll(async () => {
+		harness = await createApiTestHarness();
+	});
+
+	beforeEach(async () => {
+		await harness.reset();
+		setWorkerDependenciesForTest({
+			adminRepository: getAdminRepository(),
+			snowflakeService: getSnowflakeService(),
+		});
+		setInjectedWorkerService(new SyncTaskWorkerService({bulkBanFileShas}));
+	});
+
+	afterAll(async () => {
+		clearWorkerDependencies();
+		setInjectedWorkerService(new NoopWorkerService());
+		await harness?.shutdown();
+	});
+
+	function refreshPublishes(): Array<Array<string>> {
+		return (harness.kvProvider as MockKVProvider).publishSpy.mock.calls.filter(
+			([channel]: Array<string>) => channel === BANNED_FILE_SHAS_REFRESH_CHANNEL,
+		);
+	}
+
+	async function listAuditLogs(): Promise<Array<AdminAuditLog>> {
+		return getAdminRepository().listAllAuditLogsPaginated(100);
+	}
+
+	async function runTask(payload: Record<string, unknown>, helpers = createHelpers()): Promise<BulkJobResult> {
+		return (await bulkBanFileShas(payload, helpers)) as unknown as BulkJobResult;
+	}
+
+	it('bans every hash with its own audit row and publishes one cache refresh', async () => {
+		const admin = await setUserACLs(harness, await createTestAccount(harness), [
+			AdminACLs.AUTHENTICATE,
+			AdminACLs.BAN_FILE_SHA_ADD,
+		]);
+
+		await createBuilder(harness, admin.token)
+			.put('/admin/blocklists/file-sha/entries')
+			.header('X-Audit-Log-Reason', 'CSAM hash feed')
+			.body({sha256_list: [FIRST_SHA, SECOND_SHA]})
+			.execute();
+
+		expect(refreshPublishes()).toEqual([[BANNED_FILE_SHAS_REFRESH_CHANNEL, 'refresh']]);
+		const logs = await listAuditLogs();
+		expect(
+			logs
+				.filter((log) => log.action === 'ban_file_sha')
+				.map((log) => ({
+					adminUserId: log.adminUserId.toString(),
+					targetType: log.targetType,
+					auditLogReason: log.auditLogReason,
+					sha256: log.metadata.get('sha256'),
+				})),
+		).toEqual([
+			{adminUserId: admin.userId, targetType: 'file_sha', auditLogReason: 'CSAM hash feed', sha256: FIRST_SHA},
+			{adminUserId: admin.userId, targetType: 'file_sha', auditLogReason: 'CSAM hash feed', sha256: SECOND_SHA},
+		]);
+		const summary = logs.find((log) => log.action === 'bulk_ban_file_shas');
+		expect(summary).toBeDefined();
+		expect(summary?.targetType).toBe('bulk_job');
+		expect(summary?.auditLogReason).toBe('CSAM hash feed');
+		expect(Object.fromEntries(summary!.metadata)).toMatchObject({
+			count: '2',
+			processed: '2',
+			successful: '2',
+			failed: '0',
+		});
+	});
+
+	it('reports a non-hexadecimal hash as a failed item and still bans the rest', async () => {
+		const result = await runTask({
+			sha256_list: [NON_HEX_SHA, FIRST_SHA],
+			admin_user_id: '7',
+			audit_log_reason: 'Feed import',
+		});
+
+		expect(result.successful_count).toBe(1);
+		expect(result.failed).toEqual([{id: NON_HEX_SHA, error: 'invalid_sha256'}]);
+		const logs = await listAuditLogs();
+		expect(logs.filter((log) => log.action === 'ban_file_sha').map((log) => log.metadata.get('sha256'))).toEqual([
+			FIRST_SHA,
+		]);
+	});
+
+	it('publishes the cache refresh for the hashes already banned when the job is cancelled', async () => {
+		let checks = 0;
+		const helpers = createHelpers({
+			shouldCancel: async () => {
+				checks += 1;
+				return checks > 1;
+			},
+		});
+
+		await expect(
+			bulkBanFileShas(
+				{sha256_list: [FIRST_SHA, SECOND_SHA], admin_user_id: '7', audit_log_reason: 'Feed import'},
+				helpers,
+			),
+		).rejects.toBeInstanceOf(JobCancelledError);
+
+		expect(refreshPublishes()).toEqual([[BANNED_FILE_SHAS_REFRESH_CHANNEL, 'refresh']]);
+		const logs = await listAuditLogs();
+		expect(logs.filter((log) => log.action === 'ban_file_sha').map((log) => log.metadata.get('sha256'))).toEqual([
+			FIRST_SHA,
+		]);
+		expect(Object.fromEntries(logs.find((log) => log.action === 'bulk_ban_file_shas')!.metadata)).toMatchObject({
+			successful: '1',
+			cancelled: 'true',
+		});
+	});
+});
@@ -0,0 +1,134 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import type {AdminAuditLog} from '@app/api/admin/IAdminRepository';
+import {sendChannelMessage, setupTestGuildWithMembers} from '@app/api/channel/tests/ChannelTestUtils';
+import {getGatewayService, getSnowflakeService} from '@app/api/middleware/ServiceRegistry';
+import {
+	getAdminRepository,
+	getChannelRepository,
+	getPurgeQueue,
+	getStorageService,
+} from '@app/api/middleware/ServiceSingletons';
+import {type ApiTestHarness, createApiTestHarness} from '@app/api/test/ApiTestHarness';
+import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
+import {createBuilder} from '@app/api/test/TestRequestBuilder';
+import bulkDeleteMessagesForUsers from '@app/api/worker/tasks/admin_bulk/BulkDeleteMessagesForUsers';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import type {MessageResponse} from '@fluxer/schema/src/domains/message/MessageResponseSchemas';
+import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterAll, beforeAll, beforeEach, describe, expect, it} from 'vitest';
+
+interface BulkJobResult {
+	successful_count: number;
+	failed_count: number;
+	failed: Array<{
+		id: string;
+		error: string;
+	}>;
+}
+
+const ADMIN_USER_ID = '9';
+
+function createHelpers(): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: 909n,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+	};
+}
+
+describe('Bulk delete messages for users', () => {
+	let harness: ApiTestHarness;
+
+	beforeAll(async () => {
+		harness = await createApiTestHarness();
+	});
+
+	beforeEach(async () => {
+		await harness.reset();
+		setWorkerDependenciesForTest({
+			adminRepository: getAdminRepository(),
+			snowflakeService: getSnowflakeService(),
+			channelRepository: getChannelRepository(),
+			gatewayService: getGatewayService(),
+			storageService: getStorageService(),
+			purgeQueue: getPurgeQueue(),
+		});
+	});
+
+	afterAll(async () => {
+		clearWorkerDependencies();
+		await harness?.shutdown();
+	});
+
+	async function listAuditLogs(): Promise<Array<AdminAuditLog>> {
+		return getAdminRepository().listAllAuditLogsPaginated(100);
+	}
+
+	async function runTask(userIds: Array<string>, auditLogReason: string | null): Promise<BulkJobResult> {
+		return (await bulkDeleteMessagesForUsers(
+			{user_ids: userIds, admin_user_id: ADMIN_USER_ID, audit_log_reason: auditLogReason},
+			createHelpers(),
+		)) as unknown as BulkJobResult;
+	}
+
+	async function listMessages(token: string, channelId: string): Promise<Array<MessageResponse>> {
+		return createBuilder<Array<MessageResponse>>(harness, token).get(`/channels/${channelId}/messages`).execute();
+	}
+
+	async function countMessagesBy(token: string, channelId: string, userId: string): Promise<number> {
+		const messages = await listMessages(token, channelId);
+		return messages.filter((message) => message.author.id === userId).length;
+	}
+
+	it('carries the admin reason and the message count on the per-user audit row', async () => {
+		const {owner, members, systemChannel} = await setupTestGuildWithMembers(harness, 1);
+		const member = members[0]!;
+		await sendChannelMessage(harness, member.token, systemChannel.id, 'first spam');
+		await sendChannelMessage(harness, member.token, systemChannel.id, 'second spam');
+		const authored = await countMessagesBy(owner.token, systemChannel.id, member.userId);
+
+		const result = await runTask([member.userId], 'Spam cleanup');
+
+		expect(result.successful_count).toBe(1);
+		const logs = await listAuditLogs();
+		const perUser = logs.filter((log) => log.action === 'delete_all_user_messages');
+		expect(perUser).toHaveLength(1);
+		expect(perUser[0]!.adminUserId.toString()).toBe(ADMIN_USER_ID);
+		expect(perUser[0]!.targetType).toBe('message_deletion');
+		expect(perUser[0]!.targetId.toString()).toBe(member.userId);
+		expect(perUser[0]!.auditLogReason).toBe('Spam cleanup');
+		expect(Object.fromEntries(perUser[0]!.metadata)).toEqual({
+			user_id: member.userId,
+			message_count: authored.toString(),
+		});
+		expect(await countMessagesBy(owner.token, systemChannel.id, member.userId)).toBe(0);
+	});
+
+	it('summarises the job and reports a user that could not be processed', async () => {
+		const {owner, members, systemChannel} = await setupTestGuildWithMembers(harness, 1);
+		const member = members[0]!;
+		await sendChannelMessage(harness, member.token, systemChannel.id, 'only spam');
+		const authored = await countMessagesBy(owner.token, systemChannel.id, member.userId);
+
+		const result = await runTask(['0', member.userId], 'Spam cleanup');
+
+		expect(result.successful_count).toBe(1);
+		expect(result.failed_count).toBe(1);
+		expect(result.failed[0]!.id).toBe('0');
+		const summary = (await listAuditLogs()).find((log) => log.action === 'bulk_delete_user_messages');
+		expect(summary).toBeDefined();
+		expect(summary?.targetType).toBe('bulk_job');
+		expect(summary?.auditLogReason).toBe('Spam cleanup');
+		expect(Object.fromEntries(summary!.metadata)).toMatchObject({
+			user_count: '2',
+			message_count: authored.toString(),
+			processed: '2',
+			successful: '1',
+			failed: '1',
+		});
+	});
+});
@@ -0,0 +1,215 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {AdminRepository} from '@app/api/admin/AdminRepository';
+import type {AdminAuditLog} from '@app/api/admin/IAdminRepository';
+import {createTestAccount, setUserACLs, type TestAccount} from '@app/api/auth/tests/AuthTestUtils';
+import {createReportID, createUserID} from '@app/api/BrandedTypes';
+import {getGatewayService, getSnowflakeService} from '@app/api/middleware/ServiceRegistry';
+import {
+	createUserCacheService,
+	getAdminRepository,
+	getChannelRepository,
+	getGuildRepository,
+	getKVAccountDeletionQueue,
+	getKVBulkMessageDeletionQueue,
+	getUserRepository,
+} from '@app/api/middleware/ServiceSingletons';
+import {ReportStatus} from '@app/api/report/IReportRepository';
+import {ReportRepository} from '@app/api/report/ReportRepository';
+import {drainSearchTasks} from '@app/api/search/SearchTaskTracker';
+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';
+import {createBuilder} from '@app/api/test/TestRequestBuilder';
+import bulkScheduleUserDeletion from '@app/api/worker/tasks/admin_bulk/BulkScheduleUserDeletion';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import {DeletionReasons} from '@fluxer/constants/src/Core';
+import {UserFlags} from '@fluxer/constants/src/UserConstants';
+import type {WorkerTaskHelpers, WorkerTaskResult} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterEach, beforeEach, describe, expect, test} from 'vitest';
+
+interface ReportResponse {
+	report_id: string;
+}
+
+interface BulkJobResult {
+	successful_count: number;
+	failed_count: number;
+	failed: Array<{id: string; error: string}>;
+}
+
+const AUDIT_LOG_REASON = 'Lilith spam sweep 2026-09-14';
+
+function createHelpers(): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: 4242n,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+	};
+}
+
+function installWorkerDependencies(): void {
+	setWorkerDependenciesForTest({
+		adminRepository: getAdminRepository(),
+		snowflakeService: getSnowflakeService(),
+		userRepository: getUserRepository(),
+		userCacheService: createUserCacheService(),
+		guildRepository: getGuildRepository(),
+		channelRepository: getChannelRepository(),
+		gatewayService: getGatewayService(),
+		deletionQueueService: getKVAccountDeletionQueue(),
+		bulkMessageDeletionQueueService: getKVBulkMessageDeletionQueue(),
+		stripe: null,
+	});
+}
+
+async function runBulkJob(
+	userIds: Array<string>,
+	adminUserId: string,
+	reasonCode: number = DeletionReasons.SPAM,
+): Promise<BulkJobResult> {
+	installWorkerDependencies();
+	const result = (await bulkScheduleUserDeletion(
+		{
+			user_ids: userIds,
+			reason_code: reasonCode,
+			days_until_deletion: 60,
+			public_reason: null,
+			admin_user_id: adminUserId,
+			audit_log_reason: AUDIT_LOG_REASON,
+		},
+		createHelpers(),
+	)) as WorkerTaskResult;
+	return result as unknown as BulkJobResult;
+}
+
+async function reportUser(harness: ApiTestHarness, reporter: TestAccount, targetUserId: string): Promise<string> {
+	const report = await createBuilder<ReportResponse>(harness, reporter.token)
+		.post('/reports/user')
+		.body({user_id: targetUserId, category: 'spam_account'})
+		.expect(HTTP_STATUS.OK)
+		.execute();
+	await drainSearchTasks();
+	return report.report_id;
+}
+
+async function getReportStatus(reportId: string): Promise<number | null> {
+	const report = await new ReportRepository().getReport(createReportID(BigInt(reportId)));
+	return report?.status ?? null;
+}
+
+async function listAuditLogs(action: string): Promise<Array<AdminAuditLog>> {
+	const logs = await new AdminRepository().listAllAuditLogsPaginated(500);
+	return logs.filter((log) => log.action === action);
+}
+
+async function isSessionAlive(harness: ApiTestHarness, token: string): Promise<boolean> {
+	const response = await harness.requestJson({path: '/users/@me', headers: {Authorization: token}});
+	return response.status === HTTP_STATUS.OK;
+}
+
+describe('bulkScheduleUserDeletion', () => {
+	let harness: ApiTestHarness;
+	beforeEach(async () => {
+		harness = await createApiTestHarness({search: 'enabled'});
+	});
+	afterEach(async () => {
+		clearWorkerDependencies();
+		await harness.shutdown();
+	});
+	test('runs the same side effects as the single-user endpoint for every user in the job', async () => {
+		const admin = await createTestAccount(harness);
+		await setUserACLs(harness, admin, ['admin:authenticate', 'bulk:delete:users']);
+		const reporter = await createTestAccount(harness);
+		const target = await createTestAccount(harness);
+		const reportId = await reportUser(harness, reporter, target.userId);
+		expect(await getReportStatus(reportId)).toBe(ReportStatus.PENDING);
+		const result = await runBulkJob([target.userId], admin.userId);
+		expect(result.successful_count).toBe(1);
+		expect(result.failed_count).toBe(0);
+		expect(await getReportStatus(reportId)).toBe(ReportStatus.RESOLVED);
+		const perUserLogs = await listAuditLogs('schedule_deletion');
+		const targetLog = perUserLogs.find((log) => log.targetId === BigInt(target.userId));
+		expect(targetLog).toBeDefined();
+		expect(targetLog!.auditLogReason).toBe(AUDIT_LOG_REASON);
+		expect(targetLog!.adminUserId.toString()).toBe(admin.userId);
+		expect(targetLog!.metadata.get('reason_code')).toBe(DeletionReasons.SPAM.toString());
+		expect(targetLog!.metadata.get('days')).toBe('60');
+		expect(await isSessionAlive(harness, target.token)).toBe(false);
+		expect(await new AdminRepository().isEmailBanned(target.email)).toBe(true);
+		const resolutionLogs = await listAuditLogs('auto_resolve_reports_on_deletion');
+		expect(resolutionLogs.some((log) => log.targetId === BigInt(target.userId))).toBe(true);
+		const summaryLogs = await listAuditLogs('bulk_schedule_deletion');
+		expect(summaryLogs).toHaveLength(1);
+		expect(summaryLogs[0]!.targetType).toBe('bulk_job');
+		expect(summaryLogs[0]!.targetId).toBe(4242n);
+		expect(summaryLogs[0]!.auditLogReason).toBe(AUDIT_LOG_REASON);
+		expect(summaryLogs[0]!.metadata.get('user_count')).toBe('1');
+		expect(summaryLogs[0]!.metadata.get('reason_code')).toBe(DeletionReasons.SPAM.toString());
+		expect(summaryLogs[0]!.metadata.get('days')).toBe('60');
+		expect(summaryLogs[0]!.metadata.get('successful')).toBe('1');
+		expect(summaryLogs[0]!.metadata.get('failed')).toBe('0');
+	});
+	test('the single-user endpoint produces the same report, audit, session and ban outcome', async () => {
+		const admin = await createTestAccount(harness);
+		await setUserACLs(harness, admin, ['admin:authenticate', 'user:delete']);
+		const reporter = await createTestAccount(harness);
+		const target = await createTestAccount(harness);
+		const reportId = await reportUser(harness, reporter, target.userId);
+		await createBuilder(harness, admin.token)
+			.put(`/admin/users/${target.userId}/deletion`)
+			.header('X-Audit-Log-Reason', AUDIT_LOG_REASON)
+			.body({reason_code: DeletionReasons.SPAM, days_until_deletion: 60})
+			.expect(HTTP_STATUS.OK)
+			.execute();
+		expect(await getReportStatus(reportId)).toBe(ReportStatus.RESOLVED);
+		const perUserLogs = await listAuditLogs('schedule_deletion');
+		const targetLog = perUserLogs.find((log) => log.targetId === BigInt(target.userId));
+		expect(targetLog).toBeDefined();
+		expect(targetLog!.auditLogReason).toBe(AUDIT_LOG_REASON);
+		expect(targetLog!.metadata.get('reason_code')).toBe(DeletionReasons.SPAM.toString());
+		expect(await isSessionAlive(harness, target.token)).toBe(false);
+		expect(await new AdminRepository().isEmailBanned(target.email)).toBe(true);
+	});
+	test('keeps deleting the remaining users after one of them fails and reports the failure', async () => {
+		const admin = await createTestAccount(harness);
+		await setUserACLs(harness, admin, ['admin:authenticate', 'bulk:delete:users']);
+		const reporter = await createTestAccount(harness);
+		const missingUserId = '999999999999999999';
+		const target = await createTestAccount(harness);
+		const reportId = await reportUser(harness, reporter, target.userId);
+		const result = await runBulkJob([missingUserId, target.userId], admin.userId);
+		expect(result.successful_count).toBe(1);
+		expect(result.failed_count).toBe(1);
+		expect(result.failed.map((failure) => failure.id)).toEqual([missingUserId]);
+		expect(result.failed[0]!.error).toBeTruthy();
+		const updatedTarget = await getUserRepository().findUnique(createUserID(BigInt(target.userId)));
+		expect(updatedTarget).not.toBeNull();
+		expect(updatedTarget!.flags & UserFlags.DELETED).toBe(UserFlags.DELETED);
+		expect(updatedTarget!.pendingDeletionAt).not.toBeNull();
+		expect(await getReportStatus(reportId)).toBe(ReportStatus.RESOLVED);
+		const perUserLogs = await listAuditLogs('schedule_deletion');
+		expect(perUserLogs.filter((log) => log.targetId === BigInt(target.userId))).toHaveLength(1);
+		expect(perUserLogs.some((log) => log.targetId === BigInt(missingUserId))).toBe(false);
+		const summaryLogs = await listAuditLogs('bulk_schedule_deletion');
+		expect(summaryLogs).toHaveLength(1);
+		expect(summaryLogs[0]!.metadata.get('successful')).toBe('1');
+		expect(summaryLogs[0]!.metadata.get('failed')).toBe('1');
+	});
+	test('a user-requested bulk deletion neither bans identifiers nor resolves reports', async () => {
+		const admin = await createTestAccount(harness);
+		await setUserACLs(harness, admin, ['admin:authenticate', 'bulk:delete:users']);
+		const reporter = await createTestAccount(harness);
+		const target = await createTestAccount(harness);
+		const reportId = await reportUser(harness, reporter, target.userId);
+		const result = await runBulkJob([target.userId], admin.userId, DeletionReasons.USER_REQUESTED);
+		expect(result.successful_count).toBe(1);
+		expect(await getReportStatus(reportId)).toBe(ReportStatus.PENDING);
+		expect(await new AdminRepository().isEmailBanned(target.email)).toBe(false);
+		const perUserLogs = await listAuditLogs('schedule_deletion');
+		expect(perUserLogs.filter((log) => log.targetId === BigInt(target.userId))).toHaveLength(1);
+	});
+});
@@ -0,0 +1,229 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {AdminRepository} from '@app/api/admin/AdminRepository';
+import {createTestAccount, type TestAccount} from '@app/api/auth/tests/AuthTestUtils';
+import {createUserID} from '@app/api/BrandedTypes';
+import {getHistoricalOutcomeRepository} from '@app/api/middleware/ServiceMiddleware';
+import {getGatewayService, getSnowflakeService} from '@app/api/middleware/ServiceRegistry';
+import {
+	createUserCacheService,
+	getAdminRepository,
+	getChannelRepository,
+	getGuildRepository,
+	getKVAccountDeletionQueue,
+	getKVBulkMessageDeletionQueue,
+	getUserRepository,
+} from '@app/api/middleware/ServiceSingletons';
+import {type ApiTestHarness, createApiTestHarness} from '@app/api/test/ApiTestHarness';
+import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
+import {createBuilder} from '@app/api/test/TestRequestBuilder';
+import bulkUpdateSuspiciousActivityFlags from '@app/api/worker/tasks/admin_bulk/BulkUpdateSuspiciousActivityFlags';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import {
+	DEFERRED_PHONE_ON_COMMUNITY_JOIN,
+	PHONE_GATE_PROMOTED_FROM_DEFERRAL,
+	SuspiciousActivityFlags,
+} from '@fluxer/constants/src/UserConstants';
+import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterAll, afterEach, beforeAll, beforeEach, describe, expect, test} from 'vitest';
+
+interface BulkJobResult {
+	successful_count: number;
+	failed_count: number;
+	failed: Array<{
+		id: string;
+		error: string;
+	}>;
+}
+
+const ADMIN_USER_ID = 4000000000000000000n;
+const JOB_ID = 7200000000000000000n;
+const MISSING_USER_ID = 4200000000000000002n;
+const RISK_CONTEXT_IP = '203.0.113.77';
+
+function createHelpers(): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: JOB_ID,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+	};
+}
+
+function installWorkerDependencies(): void {
+	setWorkerDependenciesForTest({
+		adminRepository: getAdminRepository(),
+		snowflakeService: getSnowflakeService(),
+		userRepository: getUserRepository(),
+		guildRepository: getGuildRepository(),
+		channelRepository: getChannelRepository(),
+		userCacheService: createUserCacheService(),
+		deletionQueueService: getKVAccountDeletionQueue(),
+		bulkMessageDeletionQueueService: getKVBulkMessageDeletionQueue(),
+		gatewayService: getGatewayService(),
+		stripe: null,
+	});
+}
+
+describe('bulkUpdateSuspiciousActivityFlags task', () => {
+	let harness: ApiTestHarness;
+	beforeAll(async () => {
+		harness = await createApiTestHarness({search: 'enabled'});
+	});
+	afterAll(async () => {
+		await harness.shutdown();
+	});
+	beforeEach(async () => {
+		await harness.resetData();
+		installWorkerDependencies();
+	});
+	afterEach(() => {
+		clearWorkerDependencies();
+	});
+
+	async function setSuspiciousFlags(account: TestAccount, flags: number): Promise<void> {
+		await createBuilder(harness, '')
+			.post(`/test/users/${account.userId}/security-flags`)
+			.body({
+				suspicious_activity_flags: flags,
+			})
+			.execute();
+	}
+
+	async function readSuspiciousFlags(account: TestAccount): Promise<number> {
+		const user = await getUserRepository().findUnique(createUserID(BigInt(account.userId)));
+		return user!.suspiciousActivityFlags ?? 0;
+	}
+
+	test('writes a per-user audit row with the admin reason, records the risk outcome, and reports failures', async () => {
+		const account = await createTestAccount(harness);
+		await setSuspiciousFlags(account, 0);
+		await getHistoricalOutcomeRepository().upsertLatestContext({
+			userId: account.userId,
+			ip: RISK_CONTEXT_IP,
+			subnet: null,
+			emailDomain: null,
+			asn: null,
+			updatedAt: new Date(),
+		});
+
+		const result = (await bulkUpdateSuspiciousActivityFlags(
+			{
+				user_ids: [account.userId, MISSING_USER_ID.toString()],
+				add_flags: ['REQUIRE_VERIFIED_EMAIL'],
+				remove_flags: [],
+				admin_user_id: ADMIN_USER_ID.toString(),
+				audit_log_reason: 'Lilith verification sweep',
+			},
+			createHelpers(),
+		)) as unknown as BulkJobResult;
+
+		expect(result.successful_count).toBe(1);
+		expect(result.failed).toEqual([{id: MISSING_USER_ID.toString(), error: 'UNKNOWN_USER'}]);
+		expect(await readSuspiciousFlags(account)).toBe(SuspiciousActivityFlags.REQUIRE_VERIFIED_EMAIL);
+
+		const auditLogs = await new AdminRepository().listAllAuditLogsPaginated(50);
+		const perUserLogs = auditLogs.filter((log) => log.action === 'update_suspicious_activity_flags');
+		expect(perUserLogs).toHaveLength(1);
+		expect(perUserLogs[0]!.targetType).toBe('user');
+		expect(perUserLogs[0]!.targetId).toBe(BigInt(account.userId));
+		expect(perUserLogs[0]!.adminUserId.toString()).toBe(ADMIN_USER_ID.toString());
+		expect(perUserLogs[0]!.auditLogReason).toBe('Lilith verification sweep');
+		expect(perUserLogs[0]!.metadata.get('flags')).toBe(SuspiciousActivityFlags.REQUIRE_VERIFIED_EMAIL.toString());
+
+		const summaryLogs = auditLogs.filter((log) => log.action === 'bulk_update_suspicious_activity_flags');
+		expect(summaryLogs).toHaveLength(1);
+		expect(summaryLogs[0]!.targetType).toBe('bulk_job');
+		expect(summaryLogs[0]!.targetId).toBe(JOB_ID);
+		expect(summaryLogs[0]!.auditLogReason).toBe('Lilith verification sweep');
+		expect(summaryLogs[0]!.metadata.get('user_count')).toBe('2');
+		expect(summaryLogs[0]!.metadata.get('add_flags')).toBe('REQUIRE_VERIFIED_EMAIL');
+		expect(summaryLogs[0]!.metadata.has('remove_flags')).toBe(false);
+		expect(summaryLogs[0]!.metadata.get('successful')).toBe('1');
+		expect(summaryLogs[0]!.metadata.get('failed')).toBe('1');
+
+		const outcomes = await getHistoricalOutcomeRepository().listByIp(RISK_CONTEXT_IP, new Date(0), 50);
+		expect(outcomes.map((outcome) => [outcome.outcomeCode, outcome.source])).toEqual([
+			['challenged', 'admin_update_suspicious_activity_flags'],
+		]);
+	});
+
+	test('removing a phone requirement clears the deferral bookkeeping bits', async () => {
+		const account = await createTestAccount(harness);
+		await setSuspiciousFlags(
+			account,
+			SuspiciousActivityFlags.REQUIRE_VERIFIED_PHONE |
+				DEFERRED_PHONE_ON_COMMUNITY_JOIN |
+				PHONE_GATE_PROMOTED_FROM_DEFERRAL,
+		);
+
+		const result = (await bulkUpdateSuspiciousActivityFlags(
+			{
+				user_ids: [account.userId],
+				add_flags: [],
+				remove_flags: ['REQUIRE_VERIFIED_PHONE'],
+				admin_user_id: ADMIN_USER_ID.toString(),
+				audit_log_reason: 'Phone gate lifted',
+			},
+			createHelpers(),
+		)) as unknown as BulkJobResult;
+
+		expect(result.failed).toEqual([]);
+		expect(await readSuspiciousFlags(account)).toBe(0);
+
+		const auditLogs = await new AdminRepository().listAllAuditLogsPaginated(50);
+		const perUserLogs = auditLogs.filter((log) => log.action === 'update_suspicious_activity_flags');
+		expect(perUserLogs).toHaveLength(1);
+		expect(perUserLogs[0]!.auditLogReason).toBe('Phone gate lifted');
+		expect(perUserLogs[0]!.metadata.get('flags')).toBe('0');
+	});
+
+	test('keeps the deferral bits when the deferrable phone flags are unchanged', async () => {
+		const account = await createTestAccount(harness);
+		await setSuspiciousFlags(
+			account,
+			SuspiciousActivityFlags.REQUIRE_VERIFIED_PHONE | DEFERRED_PHONE_ON_COMMUNITY_JOIN,
+		);
+
+		await bulkUpdateSuspiciousActivityFlags(
+			{
+				user_ids: [account.userId],
+				add_flags: ['REQUIRE_VERIFIED_EMAIL'],
+				remove_flags: [],
+				admin_user_id: ADMIN_USER_ID.toString(),
+				audit_log_reason: 'Add email requirement',
+			},
+			createHelpers(),
+		);
+
+		expect(await readSuspiciousFlags(account)).toBe(
+			SuspiciousActivityFlags.REQUIRE_VERIFIED_PHONE |
+				SuspiciousActivityFlags.REQUIRE_VERIFIED_EMAIL |
+				DEFERRED_PHONE_ON_COMMUNITY_JOIN,
+		);
+	});
+
+	test('rejects an unknown suspicious flag name instead of silently ignoring it', async () => {
+		const account = await createTestAccount(harness);
+		await setSuspiciousFlags(account, 0);
+
+		await expect(
+			bulkUpdateSuspiciousActivityFlags(
+				{
+					user_ids: [account.userId],
+					add_flags: ['REQUIRE_VERIFIED_EMAIL', 'REQUIRE_VERIFIED_EMAILL'],
+					remove_flags: [],
+					admin_user_id: ADMIN_USER_ID.toString(),
+					audit_log_reason: 'Typo sweep',
+				},
+				createHelpers(),
+			),
+		).rejects.toThrow('Unknown suspicious activity flag names: REQUIRE_VERIFIED_EMAILL');
+
+		expect(await readSuspiciousFlags(account)).toBe(0);
+		const auditLogs = await new AdminRepository().listAllAuditLogsPaginated(50);
+		expect(auditLogs).toEqual([]);
+	});
+});
@@ -0,0 +1,143 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {AdminRepository} from '@app/api/admin/AdminRepository';
+import {createTestAccount} from '@app/api/auth/tests/AuthTestUtils';
+import {createUserID} from '@app/api/BrandedTypes';
+import {getGatewayService, getSnowflakeService} from '@app/api/middleware/ServiceRegistry';
+import {
+	createUserCacheService,
+	getAdminRepository,
+	getChannelRepository,
+	getGuildRepository,
+	getKVAccountDeletionQueue,
+	getKVBulkMessageDeletionQueue,
+	getUserRepository,
+} from '@app/api/middleware/ServiceSingletons';
+import {type ApiTestHarness, createApiTestHarness} from '@app/api/test/ApiTestHarness';
+import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
+import bulkUpdateUserFlags from '@app/api/worker/tasks/admin_bulk/BulkUpdateUserFlags';
+import {clearWorkerDependencies, setWorkerDependenciesForTest} from '@app/api/worker/WorkerContext';
+import {UserFlags} from '@fluxer/constants/src/UserConstants';
+import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
+import {afterAll, afterEach, beforeAll, beforeEach, describe, expect, test} from 'vitest';
+
+interface BulkJobResult {
+	successful_count: number;
+	failed_count: number;
+	failed: Array<{
+		id: string;
+		error: string;
+	}>;
+}
+
+const ADMIN_USER_ID = 4000000000000000000n;
+const JOB_ID = 7100000000000000000n;
+const MISSING_USER_ID = 4200000000000000001n;
+
+function createHelpers(): WorkerTaskHelpers {
+	return {
+		logger: new NoopLogger(),
+		jobId: JOB_ID,
+		addJob: async () => 0n,
+		reportProgress: async () => {},
+		shouldCancel: async () => false,
+		setContextLink: async () => {},
+	};
+}
+
+function installWorkerDependencies(): void {
+	setWorkerDependenciesForTest({
+		adminRepository: getAdminRepository(),
+		snowflakeService: getSnowflakeService(),
+		userRepository: getUserRepository(),
+		guildRepository: getGuildRepository(),
+		channelRepository: getChannelRepository(),
+		userCacheService: createUserCacheService(),
+		deletionQueueService: getKVAccountDeletionQueue(),
+		bulkMessageDeletionQueueService: getKVBulkMessageDeletionQueue(),
+		gatewayService: getGatewayService(),
+		stripe: null,
+	});
+}
+
+describe('bulkUpdateUserFlags task', () => {
+	let harness: ApiTestHarness;
+	beforeAll(async () => {
+		harness = await createApiTestHarness({search: 'enabled'});
+	});
+	afterAll(async () => {
+		await harness.shutdown();
+	});
+	beforeEach(async () => {
+		await harness.resetData();
+		installWorkerDependencies();
+	});
+	afterEach(() => {
+		clearWorkerDependencies();
+	});
+
+	test('writes a per-user audit row carrying the admin reason and records failed items', async () => {
+		const first = await createTestAccount(harness);
+		const second = await createTestAccount(harness);
+		const result = (await bulkUpdateUserFlags(
+			{
+				user_ids: [first.userId, MISSING_USER_ID.toString(), second.userId],
+				add_flags: [UserFlags.SPAMMER.toString()],
+				remove_flags: [UserFlags.HAS_SESSION_STARTED.toString()],
+				admin_user_id: ADMIN_USER_ID.toString(),
+				audit_log_reason: 'Lilith spam sweep',
+			},
+			createHelpers(),
+		)) as unknown as BulkJobResult;
+
+		expect(result.successful_count).toBe(2);
+		expect(result.failed_count).toBe(1);
+		expect(result.failed).toEqual([{id: MISSING_USER_ID.toString(), error: 'UNKNOWN_USER'}]);
+
+		const userRepository = getUserRepository();
+		const updatedFirst = await userRepository.findUnique(createUserID(BigInt(first.userId)));
+		expect(updatedFirst!.flags & UserFlags.SPAMMER).toBe(UserFlags.SPAMMER);
+		expect(updatedFirst!.flags & UserFlags.HAS_SESSION_STARTED).toBe(0n);
+
+		const auditLogs = await new AdminRepository().listAllAuditLogsPaginated(50);
+		const perUserLogs = auditLogs.filter((log) => log.action === 'update_flags');
+		expect(perUserLogs.map((log) => log.targetId).sort()).toEqual([BigInt(first.userId), BigInt(second.userId)].sort());
+		for (const log of perUserLogs) {
+			expect(log.targetType).toBe('user');
+			expect(log.adminUserId.toString()).toBe(ADMIN_USER_ID.toString());
+			expect(log.auditLogReason).toBe('Lilith spam sweep');
+			expect(log.metadata.get('add_flags')).toBe(UserFlags.SPAMMER.toString());
+			expect(log.metadata.get('remove_flags')).toBe(UserFlags.HAS_SESSION_STARTED.toString());
+			expect(log.metadata.get('new_flags')).toBe(UserFlags.SPAMMER.toString());
+		}
+		expect(auditLogs.some((log) => log.action === 'update_flags' && log.targetId === MISSING_USER_ID)).toBe(false);
+	});
+
+	test('writes one summary row against the job id', async () => {
+		const account = await createTestAccount(harness);
+		await bulkUpdateUserFlags(
+			{
+				user_ids: [account.userId, MISSING_USER_ID.toString()],
+				add_flags: [UserFlags.STAFF.toString()],
+				remove_flags: [],
+				admin_user_id: ADMIN_USER_ID.toString(),
+				audit_log_reason: 'Staff grant',
+			},
+			createHelpers(),
+		);
+
+		const auditLogs = await new AdminRepository().listAllAuditLogsPaginated(50);
+		const summaryLogs = auditLogs.filter((log) => log.action === 'bulk_update_user_flags');
+		expect(summaryLogs).toHaveLength(1);
+		const summary = summaryLogs[0]!;
+		expect(summary.targetType).toBe('bulk_job');
+		expect(summary.targetId).toBe(JOB_ID);
+		expect(summary.auditLogReason).toBe('Staff grant');
+		expect(summary.metadata.get('user_count')).toBe('2');
+		expect(summary.metadata.get('add_flags')).toBe(UserFlags.STAFF.toString());
+		expect(summary.metadata.has('remove_flags')).toBe(false);
+		expect(summary.metadata.get('processed')).toBe('2');
+		expect(summary.metadata.get('successful')).toBe('1');
+		expect(summary.metadata.get('failed')).toBe('1');
+	});
+});