fix: parallelize user dependency cleanup deletes

Author: x0sina

app/db/crud/user.py

@@ -1,3 +1,4 @@
+import asyncio
 from copy import deepcopy
 from datetime import UTC, datetime, timedelta, timezone
 from enum import Enum
@@ -8,6 +9,7 @@
 from sqlalchemy.orm import joinedload, selectinload
 from sqlalchemy.sql.functions import coalesce
 
+from app.db.base import GetDB, IS_SQLITE
 from app.db.compiles_types import DateDiff
 from app.db.models import (
     Admin,
@@ -39,8 +41,6 @@
 from .group import get_groups_by_ids
 
 _USER_AGENT_MAX_LEN = UserSubscriptionUpdate.__table__.columns.user_agent.type.length or 512
-
-
 async def load_user_attrs(
     user: User,
     *,
@@ -696,12 +696,32 @@ async def _delete_user_dependencies(db: AsyncSession, user_ids: list[int]):
     if not user_ids:
         return
 
-    await db.execute(delete(NodeUserUsage).where(NodeUserUsage.user_id.in_(user_ids)))
-    await db.execute(delete(NotificationReminder).where(NotificationReminder.user_id.in_(user_ids)))
-    await db.execute(delete(UserSubscriptionUpdate).where(UserSubscriptionUpdate.user_id.in_(user_ids)))
-    await db.execute(delete(UserUsageResetLogs).where(UserUsageResetLogs.user_id.in_(user_ids)))
-    await db.execute(delete(NextPlan).where(NextPlan.user_id.in_(user_ids)))
-    await db.execute(users_groups_association.delete().where(users_groups_association.c.user_id.in_(user_ids)))
+    statements = [
+        delete(NodeUserUsage).where(NodeUserUsage.user_id.in_(user_ids)),
+        delete(NotificationReminder).where(NotificationReminder.user_id.in_(user_ids)),
+        delete(UserSubscriptionUpdate).where(UserSubscriptionUpdate.user_id.in_(user_ids)),
+        delete(UserUsageResetLogs).where(UserUsageResetLogs.user_id.in_(user_ids)),
+        delete(NextPlan).where(NextPlan.user_id.in_(user_ids)),
+        users_groups_association.delete().where(users_groups_association.c.user_id.in_(user_ids)),
+    ]
+
+    if IS_SQLITE:
+        for stmt in statements:
+            await db.execute(stmt)
+        return
+
+    async def execute_in_isolated_session(statement):
+        async with GetDB() as isolated_db:
+            await isolated_db.execute(statement)
+            await isolated_db.commit()
+
+    results = await asyncio.gather(
+        *(execute_in_isolated_session(stmt) for stmt in statements),
+        return_exceptions=True,
+    )
+    for result in results:
+        if isinstance(result, Exception):
+            raise result
 
 
 async def remove_user(db: AsyncSession, db_user: User) -> User: