perf: defer user sync serialization from request path

Author: x0sina

app/db/crud/group.py

@@ -28,12 +28,20 @@ async def get_inbounds_by_tags(db: AsyncSession, tags: list[str]) -> list[ProxyI
     return [inbounds_map[tag] for tag in tags]
 
 
-async def load_group_attrs(group: Group):
-    await group.awaitable_attrs.users
-    await group.awaitable_attrs.inbounds
+async def load_group_attrs(group: Group, *, load_users: bool = True, load_inbounds: bool = True):
+    if load_users:
+        await group.awaitable_attrs.users
+    if load_inbounds:
+        await group.awaitable_attrs.inbounds
 
 
-async def get_group_by_id(db: AsyncSession, group_id: int) -> Group | None:
+async def get_group_by_id(
+    db: AsyncSession,
+    group_id: int,
+    *,
+    load_users: bool = True,
+    load_inbounds: bool = True,
+) -> Group | None:
     """
     Retrieves a group by its ID.
 
@@ -46,7 +54,7 @@ async def get_group_by_id(db: AsyncSession, group_id: int) -> Group | None:
     """
     group = (await db.execute(select(Group).where(Group.id == group_id))).unique().scalar_one_or_none()
     if group:
-        await load_group_attrs(group)
+        await load_group_attrs(group, load_users=load_users, load_inbounds=load_inbounds)
     return group
 
 
@@ -162,7 +170,13 @@ async def get_groups_simple(
     return rows, total
 
 
-async def get_groups_by_ids(db: AsyncSession, group_ids: list[int]) -> list[Group]:
+async def get_groups_by_ids(
+    db: AsyncSession,
+    group_ids: list[int],
+    *,
+    load_users: bool = True,
+    load_inbounds: bool = True,
+) -> list[Group]:
     """
     Retrieves a list of groups by their IDs.
 
@@ -176,7 +190,7 @@ async def get_groups_by_ids(db: AsyncSession, group_ids: list[int]) -> list[Grou
     groups = (await db.execute(select(Group).where(Group.id.in_(group_ids)))).scalars().all()
 
     for group in groups:
-        await load_group_attrs(group)
+        await load_group_attrs(group, load_users=load_users, load_inbounds=load_inbounds)
 
     return groups
 

app/db/crud/user.py

@@ -41,14 +41,33 @@
 _USER_AGENT_MAX_LEN = UserSubscriptionUpdate.__table__.columns.user_agent.type.length or 512
 
 
-async def load_user_attrs(user: User):
-    await user.awaitable_attrs.admin
-    await user.awaitable_attrs.next_plan
-    await user.awaitable_attrs.usage_logs
-    await user.awaitable_attrs.groups
+async def load_user_attrs(
+    user: User,
+    *,
+    load_admin: bool = True,
+    load_next_plan: bool = True,
+    load_usage_logs: bool = True,
+    load_groups: bool = True,
+):
+    if load_admin:
+        await user.awaitable_attrs.admin
+    if load_next_plan:
+        await user.awaitable_attrs.next_plan
+    if load_usage_logs:
+        await user.awaitable_attrs.usage_logs
+    if load_groups:
+        await user.awaitable_attrs.groups
 
 
-async def get_user(db: AsyncSession, username: str) -> Optional[User]:
+async def get_user(
+    db: AsyncSession,
+    username: str,
+    *,
+    load_admin: bool = True,
+    load_next_plan: bool = True,
+    load_usage_logs: bool = True,
+    load_groups: bool = True,
+) -> Optional[User]:
     """
     Retrieves a user by username.
 
@@ -63,11 +82,25 @@ async def get_user(db: AsyncSession, username: str) -> Optional[User]:
 
     user = (await db.execute(stmt)).unique().scalar_one_or_none()
     if user:
-        await load_user_attrs(user)
+        await load_user_attrs(
+            user,
+            load_admin=load_admin,
+            load_next_plan=load_next_plan,
+            load_usage_logs=load_usage_logs,
+            load_groups=load_groups,
+        )
     return user
 
 
-async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:
+async def get_user_by_id(
+    db: AsyncSession,
+    user_id: int,
+    *,
+    load_admin: bool = True,
+    load_next_plan: bool = True,
+    load_usage_logs: bool = True,
+    load_groups: bool = True,
+) -> User | None:
     """
     Retrieves a user by user ID.
 
@@ -82,10 +115,29 @@ async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:
 
     user = (await db.execute(stmt)).unique().scalar_one_or_none()
     if user:
-        await load_user_attrs(user)
+        await load_user_attrs(
+            user,
+            load_admin=load_admin,
+            load_next_plan=load_next_plan,
+            load_usage_logs=load_usage_logs,
+            load_groups=load_groups,
+        )
     return user
 
 
+async def get_user_lifetime_used_traffic(db: AsyncSession, user_id: int) -> int:
+    stmt = (
+        select(func.coalesce(func.sum(UserUsageResetLogs.used_traffic_at_reset), 0) + func.coalesce(User.used_traffic, 0))
+        .select_from(User)
+        .outerjoin(UserUsageResetLogs, UserUsageResetLogs.user_id == User.id)
+        .where(User.id == user_id)
+        .group_by(User.id)
+    )
+    result = await db.execute(stmt)
+    value = result.scalar_one_or_none()
+    return int(value or 0)
+
+
 async def get_existing_usernames(db: AsyncSession, usernames: Sequence[str]) -> set[str]:
     """
     Returns the set of usernames that already exist in the database.
@@ -594,7 +646,7 @@ async def create_user(db: AsyncSession, new_user: UserCreate, groups: list[Group
         await db.commit()
         await db.refresh(db_user)
 
-    await load_user_attrs(db_user)
+    await load_user_attrs(db_user, load_usage_logs=False)
     return db_user
 
 
@@ -634,7 +686,7 @@ async def create_users_bulk(
     await db.commit()
 
     for user in db_users:
-        await load_user_attrs(user)
+        await load_user_attrs(user, load_usage_logs=False)
 
     return db_users
 
@@ -780,7 +832,7 @@ async def modify_user(db: AsyncSession, db_user: User, modify: UserModify) -> Us
 
     await db.commit()
     await db.refresh(db_user)
-    await load_user_attrs(db_user)
+    await load_user_attrs(db_user, load_usage_logs=False)
     return db_user
 
 
@@ -819,7 +871,7 @@ async def reset_user_data_usage(db: AsyncSession, db_user: User) -> User:
 
     await db.commit()
     await db.refresh(db_user)
-    await load_user_attrs(db_user)
+    await load_user_attrs(db_user, load_usage_logs=False)
     return db_user
 
 
@@ -841,7 +893,7 @@ async def bulk_reset_user_data_usage(db: AsyncSession, users: list[User]) -> lis
     await db.commit()
     for user in users:
         await db.refresh(user)
-        await load_user_attrs(user)
+        await load_user_attrs(user, load_usage_logs=False)
     return users
 
 
@@ -903,7 +955,7 @@ async def reset_user_by_next(db: AsyncSession, db_user: User) -> User:
 
     await db.commit()
     await db.refresh(db_user)
-    await load_user_attrs(db_user)
+    await load_user_attrs(db_user, load_usage_logs=False)
     return db_user
 
 
@@ -926,7 +978,7 @@ async def revoke_user_sub(db: AsyncSession, db_user: User) -> User:
     await db.execute(stmt)
     await db.commit()
     await db.refresh(db_user)
-    await load_user_attrs(db_user)
+    await load_user_attrs(db_user, load_usage_logs=False)
     return db_user
 
 
@@ -1144,7 +1196,7 @@ async def update_users_status(db: AsyncSession, users: list[User], status: UserS
     await db.commit()
     for user in users:
         await db.refresh(user)
-        await load_user_attrs(user)
+        await load_user_attrs(user, load_usage_logs=False)
     return users
 
 
@@ -1164,7 +1216,7 @@ async def set_owner(db: AsyncSession, db_user: User, admin: Admin) -> User:
     await db.execute(stmt)
     await db.commit()
     await db.refresh(db_user)
-    await load_user_attrs(db_user)
+    await load_user_attrs(db_user, load_usage_logs=False)
     return db_user
 
 
@@ -1192,7 +1244,7 @@ async def start_users_expire(db: AsyncSession, users: list[User]) -> list[User]:
     await db.commit()
     for user in users:
         await db.refresh(user)
-        await load_user_attrs(user)
+        await load_user_attrs(user, load_usage_logs=False)
     return users
 
 

app/operation/__init__.py

@@ -15,6 +15,7 @@
     get_user_template,
 )
 from app.db.crud.admin import get_admin_by_id
+from app.db.crud.group import get_groups_by_ids
 from app.db.crud.user import get_user_by_id
 from app.db.models import Admin as DBAdmin, CoreConfig, Group, Node, ProxyHost, User, UserTemplate
 from app.models.admin import AdminDetails
@@ -98,7 +99,7 @@ async def get_validated_sub(self, db: AsyncSession, token: str) -> User:
         if not sub:
             await self.raise_error(message="Not Found", code=404)
 
-        db_user = await get_user(db, sub["username"])
+        db_user = await get_user(db, sub["username"], load_usage_logs=False)
         if not db_user or db_user.created_at.astimezone(tz.utc) > sub["created_at"]:
             await self.raise_error(message="Not Found", code=404)
 
@@ -108,7 +109,7 @@ async def get_validated_sub(self, db: AsyncSession, token: str) -> User:
         return db_user
 
     async def get_validated_user(self, db: AsyncSession, username: str, admin: AdminDetails) -> User:
-        db_user = await get_user(db, username)
+        db_user = await get_user(db, username, load_usage_logs=False)
         if not db_user:
             await self.raise_error(message="User not found", code=404)
 
@@ -118,7 +119,7 @@ async def get_validated_user(self, db: AsyncSession, username: str, admin: Admin
         return db_user
 
     async def get_validated_user_by_id(self, db: AsyncSession, user_id: int, admin: AdminDetails) -> User:
-        db_user = await get_user_by_id(db, user_id)
+        db_user = await get_user_by_id(db, user_id, load_usage_logs=False)
         if not db_user:
             await self.raise_error(message="User not found", code=404)
 
@@ -140,22 +141,31 @@ async def get_validated_admin_by_id(self, db: AsyncSession, id: int) -> DBAdmin:
         return db_admin
 
     async def get_validated_group(self, db: AsyncSession, group_id: int) -> Group:
-        db_group = await get_group_by_id(db, group_id)
+        db_group = await get_group_by_id(db, group_id, load_users=False, load_inbounds=True)
         if not db_group:
             await self.raise_error("Group not found", 404)
         return db_group
 
     async def validate_all_groups(self, db, model: UserCreate | UserModify | UserTemplate | BulkGroup) -> list[Group]:
-        all_groups: list[Group] = []
+        requested_group_ids: list[int] = []
         if model.group_ids:
-            for group_id in model.group_ids:
-                db_group = await self.get_validated_group(db, group_id)
-                all_groups.append(db_group)
+            requested_group_ids.extend(model.group_ids)
         if hasattr(model, "has_group_ids") and model.has_group_ids:
-            for group_id in model.has_group_ids:
-                db_group = await self.get_validated_group(db, group_id)
-                all_groups.append(db_group)
-        return all_groups
+            requested_group_ids.extend(model.has_group_ids)
+
+        if not requested_group_ids:
+            return []
+
+        unique_ids = list(dict.fromkeys(requested_group_ids))
+        groups = await get_groups_by_ids(db, unique_ids, load_users=False, load_inbounds=True)
+        groups_by_id = {group.id: group for group in groups}
+
+        missing_ids = [group_id for group_id in unique_ids if group_id not in groups_by_id]
+        if missing_ids:
+            await self.raise_error("Group not found", 404)
+
+        # Preserve the requested order and duplicate semantics.
+        return [groups_by_id[group_id] for group_id in requested_group_ids]
 
     async def get_validated_user_template(self, db: AsyncSession, template_id: int) -> UserTemplate:
         dbuser_template = await get_user_template(db, template_id)

app/operation/admin.py

@@ -20,8 +20,7 @@
 from app.db.crud.bulk import activate_all_disabled_users, disable_all_active_users
 from app.db.crud.user import get_users, remove_users
 from app.models.admin import AdminCreate, AdminDetails, AdminModify, AdminSimple, AdminsResponse, AdminsSimpleResponse
-from app.node.sync import schedule_sync_task, sync_remove_users, sync_users, sync_proto_users, remove_user as sync_remove_user
-from app.node.user import serialize_users_for_node
+from app.node.sync import schedule_sync_task, sync_remove_users, sync_users, remove_user as sync_remove_user
 from app.models.stats import Period, UserUsageStatsList
 from app.operation import BaseOperation, OperatorType
 from app.operation.user import UserOperation
@@ -172,8 +171,7 @@ async def disable_all_active_users(self, db: AsyncSession, username: str, admin:
 
         users = await get_users(db, admin=db_admin)
         if self._is_non_blocking_sync_operator(self.operator_type):
-            proto_users = await serialize_users_for_node(users)
-            schedule_sync_task(sync_proto_users(proto_users))
+            schedule_sync_task(sync_users(users))
         else:
             await sync_users(users)
 
@@ -190,8 +188,7 @@ async def activate_all_disabled_users(self, db: AsyncSession, username: str, adm
 
         users = await get_users(db, admin=db_admin)
         if self._is_non_blocking_sync_operator(self.operator_type):
-            proto_users = await serialize_users_for_node(users)
-            schedule_sync_task(sync_proto_users(proto_users))
+            schedule_sync_task(sync_users(users))
         else:
             await sync_users(users)
 
@@ -210,7 +207,9 @@ async def remove_all_users(self, db: AsyncSession, username: str, admin: AdminDe
             return 0
 
         user_operation = UserOperation(self.operator_type)
-        serialized_users = [await user_operation.validate_user(user) for user in users]
+        serialized_users = [
+            await user_operation.validate_user(db, user, include_subscription_url=False) for user in users
+        ]
 
         await remove_users(db, users)