feat(admin): add bulk delete user messages tool (#2472)

fluxer_admin/openapi-admin.json

@@ -2335,6 +2335,63 @@
 				}
 			}
 		},
+		"/admin/bulk/delete-user-messages": {
+			"post": {
+				"operationId": "bulk_delete_user_messages",
+				"summary": "Bulk delete user messages",
+				"tags": ["Admin"],
+				"responses": {
+					"200": {
+						"description": "Success",
+						"content": {"application/json": {"schema": {"$ref": "#/components/schemas/BulkJobResponse"}}}
+					},
+					"400": {
+						"description": "Bad Request - The request was malformed or contained invalid data",
+						"content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}}
+					},
+					"401": {
+						"description": "Unauthorized - Authentication is required or the token is invalid",
+						"content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}}
+					},
+					"403": {
+						"description": "Forbidden - You do not have permission to perform this action",
+						"content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}}
+					},
+					"429": {
+						"description": "Too Many Requests - You are being rate limited",
+						"content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}},
+						"headers": {
+							"Retry-After": {
+								"description": "Number of seconds to wait before retrying (only on 429)",
+								"schema": {"type": "integer"}
+							},
+							"X-RateLimit-Limit": {
+								"description": "The number of requests that can be made in the current window",
+								"schema": {"type": "integer"}
+							},
+							"X-RateLimit-Remaining": {
+								"description": "The number of remaining requests that can be made",
+								"schema": {"type": "integer"}
+							},
+							"X-RateLimit-Reset": {
+								"description": "Unix timestamp when the rate limit resets",
+								"schema": {"type": "integer"}
+							}
+						}
+					},
+					"500": {
+						"description": "Internal Server Error - An unexpected error occurred",
+						"content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}}
+					}
+				},
+				"description": "Enqueue a background job that deletes every message authored by each of the given users across all channels. Returns a job_id immediately; observe progress at /admin/jobs/:job_id.",
+				"security": [{"adminApiKey": []}],
+				"requestBody": {
+					"required": true,
+					"content": {"application/json": {"schema": {"$ref": "#/components/schemas/BulkDeleteUserMessagesRequest"}}}
+				}
+			}
+		},
 		"/admin/bulk/schedule-user-deletion": {
 			"post": {
 				"operationId": "schedule_bulk_user_deletion",
@@ -8808,7 +8865,7 @@
 					"acls": {
 						"type": "array",
 						"items": {"type": "string"},
-						"maxItems": 111,
+						"maxItems": 112,
 						"description": "List of access control permissions for the key"
 					}
 				},
@@ -9162,7 +9219,7 @@
 					"acls": {
 						"type": "array",
 						"items": {"type": "string"},
-						"maxItems": 111,
+						"maxItems": 112,
 						"description": "List of access control permissions for the key"
 					}
 				},
@@ -9192,7 +9249,7 @@
 					"acls": {
 						"type": "array",
 						"items": {"type": "string"},
-						"maxItems": 111,
+						"maxItems": 112,
 						"description": "List of access control permissions for the key"
 					}
 				},
@@ -9892,6 +9949,18 @@
 				},
 				"required": ["guild_id", "user_ids"]
 			},
+			"BulkDeleteUserMessagesRequest": {
+				"type": "object",
+				"properties": {
+					"user_ids": {
+						"type": "array",
+						"items": {"$ref": "#/components/schemas/SnowflakeType"},
+						"maxItems": 1000,
+						"description": "List of user IDs whose messages will be deleted"
+					}
+				},
+				"required": ["user_ids"]
+			},
 			"BulkScheduleUserDeletionRequest": {
 				"type": "object",
 				"properties": {
@@ -13581,7 +13650,7 @@
 					"pending_bulk_message_deletion_at": {"nullable": true, "type": "string"},
 					"deletion_reason_code": {"nullable": true, "allOf": [{"$ref": "#/components/schemas/Int32Type"}]},
 					"deletion_public_reason": {"nullable": true, "type": "string"},
-					"acls": {"type": "array", "items": {"type": "string"}, "maxItems": 111},
+					"acls": {"type": "array", "items": {"type": "string"}, "maxItems": 112},
 					"traits": {"type": "array", "items": {"type": "string"}, "maxItems": 100},
 					"has_totp": {"type": "boolean"},
 					"authenticator_types": {"type": "array", "items": {"$ref": "#/components/schemas/Int32Type"}, "maxItems": 10},
@@ -14149,7 +14218,7 @@
 					"acls": {
 						"type": "array",
 						"items": {"type": "string"},
-						"maxItems": 111,
+						"maxItems": 112,
 						"description": "List of access control permissions to assign"
 					}
 				},

fluxer_admin/src/acl.rs

@@ -43,6 +43,7 @@ pub const BAN_PROFILE_SUBSTRING_CHECK: &str = "ban:profile_substring:check";
 pub const BAN_PROFILE_SUBSTRING_REMOVE: &str = "ban:profile_substring:remove";
 pub const BULK_ADD_GUILD_MEMBERS: &str = "bulk:add:guild_members";
 pub const BULK_DELETE_USERS: &str = "bulk:delete:users";
+pub const BULK_DELETE_USER_MESSAGES: &str = "bulk:delete:user_messages";
 pub const BULK_UPDATE_GUILD_FEATURES: &str = "bulk:update:guild_features";
 pub const BULK_UPDATE_SUSPICIOUS_ACTIVITY: &str = "bulk:update:suspicious_activity";
 pub const BULK_UPDATE_USER_FLAGS: &str = "bulk:update:user_flags";
@@ -154,6 +155,7 @@ pub const ALL_ACLS: &[&str] = &[
     BAN_PROFILE_SUBSTRING_REMOVE,
     BULK_ADD_GUILD_MEMBERS,
     BULK_DELETE_USERS,
+    BULK_DELETE_USER_MESSAGES,
     BULK_UPDATE_GUILD_FEATURES,
     BULK_UPDATE_SUSPICIOUS_ACTIVITY,
     BULK_UPDATE_USER_FLAGS,

fluxer_admin/src/api/bulk.rs

@@ -72,6 +72,18 @@ impl AdminApiClient {
             .await
     }
 
+    pub async fn bulk_delete_user_messages(
+        &self,
+        user_ids: &[String],
+        audit_log_reason: Option<&str>,
+    ) -> ApiResult<BulkJobResponse> {
+        let body = generated_types::BulkDeleteUserMessagesRequest {
+            user_ids: snowflakes(user_ids),
+        };
+        self.post_typed_with_reason("/admin/bulk/delete-user-messages", &body, audit_log_reason)
+            .await
+    }
+
     pub async fn bulk_schedule_user_deletion(
         &self,
         user_ids: &[String],

fluxer_admin/src/routes/message_actions.rs

@@ -262,6 +262,12 @@ pub(crate) async fn bulk_actions_post(
                 )
                 .await
         }
+        "bulk-delete-user-messages" => {
+            let user_ids = form.list_values_any(&["user_ids[]", "user_ids"]);
+            client
+                .bulk_delete_user_messages(&user_ids, audit_log_reason.as_deref())
+                .await
+        }
         "bulk_delete_users" => {
             let user_ids = form.list_values_any(&["user_ids[]", "user_ids"]);
             client

fluxer_admin/src/templates/layout_sidebar_nav.rs

@@ -57,6 +57,7 @@ pub const NAV_SECTIONS: &[NavSection] = &[
                     acl::BULK_UPDATE_GUILD_FEATURES,
                     acl::BULK_ADD_GUILD_MEMBERS,
                     acl::BULK_DELETE_USERS,
+                    acl::BULK_DELETE_USER_MESSAGES,
                 ]
             ),
         ],

fluxer_admin/src/templates/pages/bulk_actions.rs

@@ -201,6 +201,9 @@ pub fn bulk_actions_page(config: &AdminConfig, auth: &AuthContext, csrf_token: &
             @if acl::has_permission(admin_acls, acl::BULK_DELETE_USERS) {
                 (bulk_schedule_deletion_section(base, csrf_token))
             }
+            @if acl::has_permission(admin_acls, acl::BULK_DELETE_USER_MESSAGES) {
+                (bulk_delete_user_messages_section(base, csrf_token))
+            }
         }
     };
     admin_layout(config, auth, "Bulk Actions", "bulk-actions", None, content)
@@ -384,3 +387,24 @@ fn bulk_schedule_deletion_section(base: &str, csrf_token: &str) -> Markup {
         },
     )
 }
+
+fn bulk_delete_user_messages_section(base: &str, csrf_token: &str) -> Markup {
+    section_card_simple(
+        "Bulk Delete User Messages",
+        html! {
+            form method="post" action={(base) "/bulk-actions?action=bulk-delete-user-messages"} {
+                (csrf_input(csrf_token))
+                div class="space-y-4" {
+                    p class="text-neutral-500 text-sm" {
+                        "Deletes every message authored by each user across all channels. This cannot be undone."
+                    }
+                    (textarea_input("user_ids", "User IDs (one per line)", "123456789\n987654321", "", 5, true))
+                    (text_input("audit_log_reason", "Audit Log Reason (optional)", "", "Reason for this bulk operation"))
+                    (form_actions(html! {
+                        (danger_button("Delete All Messages"))
+                    }))
+                }
+            }
+        },
+    )
+}

fluxer_api/src/api/admin/controllers/BulkAdminController.ts

@@ -5,6 +5,7 @@ import {
 	BulkAddGuildMembersRequest,
 	BulkUpdateGuildFeaturesRequest,
 } from '@fluxer/schema/src/domains/admin/AdminGuildSchemas';
+import {BulkDeleteUserMessagesRequest} from '@fluxer/schema/src/domains/admin/AdminMessageSchemas';
 import {BulkJobResponse} from '@fluxer/schema/src/domains/admin/AdminSchemas';
 import {
 	BulkScheduleUserDeletionRequest,
@@ -185,4 +186,35 @@ export function BulkAdminController(app: HonoApp) {
 			return ctx.json({job_id: jobId.toString()});
 		},
 	);
+	app.post(
+		'/admin/bulk/delete-user-messages',
+		RateLimitMiddleware(RateLimitConfigs.ADMIN_BULK_OPERATION),
+		requireAdminACL(AdminACLs.BULK_DELETE_USER_MESSAGES),
+		Validator('json', BulkDeleteUserMessagesRequest),
+		OpenAPI({
+			operationId: 'bulk_delete_user_messages',
+			summary: 'Bulk delete user messages',
+			description:
+				'Enqueue a background job that deletes every message authored by each of the given users across all channels. Returns a job_id immediately; observe progress at /admin/jobs/:job_id.',
+			responseSchema: BulkJobResponse,
+			statusCode: 200,
+			security: 'adminApiKey',
+			tags: 'Admin',
+		}),
+		async (ctx) => {
+			const adminUserId = ctx.get('adminUserId');
+			const auditLogReason = ctx.get('auditLogReason');
+			const body = ctx.req.valid('json');
+			const jobId = await getWorkerService().addJob(
+				'bulkDeleteMessagesForUsers',
+				{
+					user_ids: body.user_ids.map((id) => id.toString()),
+					admin_user_id: adminUserId.toString(),
+					audit_log_reason: auditLogReason,
+				},
+				{requestedByUserId: adminUserId, ...(auditLogReason && {auditLogReason})},
+			);
+			return ctx.json({job_id: jobId.toString()});
+		},
+	);
 }

fluxer_api/src/api/worker/WorkerLaneConfig.ts

@@ -56,6 +56,7 @@ const LANE_CONFIG = {
 			'bulkUpdateGuildFeatures',
 			'bulkAddGuildMembers',
 			'bulkBanFileShas',
+			'bulkDeleteMessagesForUsers',
 		] as const,
 		retiredTasks: ['sendScheduledMessage'],
 		concurrency: 8,

fluxer_api/src/api/worker/WorkerTaskRegistry.ts

@@ -4,6 +4,7 @@ import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
 import applicationProcessDeletion from './tasks/ApplicationProcessDeletion';
 import bulkAddGuildMembers from './tasks/admin_bulk/BulkAddGuildMembers';
 import bulkBanFileShas from './tasks/admin_bulk/BulkBanFileShas';
+import bulkDeleteMessagesForUsers from './tasks/admin_bulk/BulkDeleteMessagesForUsers';
 import bulkScheduleUserDeletion from './tasks/admin_bulk/BulkScheduleUserDeletion';
 import bulkUpdateGuildFeatures from './tasks/admin_bulk/BulkUpdateGuildFeatures';
 import bulkUpdateSuspiciousActivityFlags from './tasks/admin_bulk/BulkUpdateSuspiciousActivityFlags';
@@ -49,6 +50,7 @@ export const workerTasks: Record<WorkerTaskName, WorkerTaskHandler> = {
 	batchGuildAuditLogMessageDeletes,
 	bulkAddGuildMembers: bulkAddGuildMembers,
 	bulkBanFileShas: bulkBanFileShas,
+	bulkDeleteMessagesForUsers: bulkDeleteMessagesForUsers,
 	bulkDeleteSelfMessagesImmediate,
 	bulkDeleteUserMessages,
 	bulkDeleteUserMessagesScoped,

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

@@ -0,0 +1,81 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
+import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';
+import {AdminAuditService} from '../../../admin/services/AdminAuditService';
+import {createUserID} from '../../../BrandedTypes';
+import {UserMessageDeletionService} from '../../../channel/services/message/UserMessageDeletionService';
+import {getWorkerDependencies} from '../../WorkerContext';
+
+interface Payload {
+	user_ids: Array<string>;
+	admin_user_id: string;
+	audit_log_reason: string | null;
+}
+
+const handler: WorkerTaskHandler = async (rawPayload, helpers) => {
+	const payload: Payload = {
+		user_ids: rawPayload.user_ids as Array<string>,
+		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 deletionService = new UserMessageDeletionService({
+		channelRepository: deps.channelRepository,
+		gatewayService: deps.gatewayService,
+		storageService: deps.storageService,
+		purgeQueue: deps.purgeQueue,
+	});
+	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 {
+			const userId = createUserID(BigInt(rawUserId));
+			const deleted = await deletionService.deleteUserMessagesBulk(userId);
+			deletedMessages += deleted;
+			await auditService.createAuditLog({
+				adminUserId,
+				targetType: 'message_deletion',
+				targetId: BigInt(userId),
+				action: 'delete_all_user_messages',
+				auditLogReason: null,
+				metadata: new Map([['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()],
+		]),
+	});
+	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;

packages/constants/src/AdminACLs.ts

@@ -44,6 +44,7 @@ export const AdminACLs = {
 	BAN_PROFILE_SUBSTRING_REMOVE: 'ban:profile_substring:remove',
 	BULK_ADD_GUILD_MEMBERS: 'bulk:add:guild_members',
 	BULK_DELETE_USERS: 'bulk:delete:users',
+	BULK_DELETE_USER_MESSAGES: 'bulk:delete:user_messages',
 	BULK_UPDATE_GUILD_FEATURES: 'bulk:update:guild_features',
 	BULK_UPDATE_SUSPICIOUS_ACTIVITY: 'bulk:update:suspicious_activity',
 	BULK_UPDATE_USER_FLAGS: 'bulk:update:user_flags',

packages/schema/src/domains/admin/AdminMessageSchemas.ts

@@ -81,3 +81,9 @@ export const DeleteAllUserMessagesResponse = z.object({
 });
 
 export type DeleteAllUserMessagesResponse = z.infer<typeof DeleteAllUserMessagesResponse>;
+
+export const BulkDeleteUserMessagesRequest = z.object({
+	user_ids: z.array(SnowflakeType).max(1000).describe('List of user IDs whose messages will be deleted'),
+});
+
+export type BulkDeleteUserMessagesRequest = z.infer<typeof BulkDeleteUserMessagesRequest>;