fix(user): revert cross-loop async db changes and stabilize reset/delete paths

Author: x0sina

app/db/crud/user.py

@@ -1,4 +1,3 @@
-import asyncio
 from copy import deepcopy
 from datetime import UTC, datetime, timedelta, timezone
 from enum import Enum
@@ -9,7 +8,6 @@
 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,
@@ -698,32 +696,12 @@ async def _delete_user_dependencies(db: AsyncSession, user_ids: list[int]):
     if not user_ids:
         return
 
-    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
+    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)))
 
 
 async def remove_user(db: AsyncSession, db_user: User) -> User:
@@ -858,16 +836,12 @@ async def modify_user(db: AsyncSession, db_user: User, modify: UserModify) -> Us
     return db_user
 
 
-async def _reset_user_traffic_and_log(db: AsyncSession, db_user: User, *, reset_at: datetime | None = None) -> datetime:
+async def _reset_user_traffic_and_log(db: AsyncSession, db_user: User):
     """Helper to reset user traffic and log the action."""
-    if reset_at is None:
-        reset_at = datetime.now(timezone.utc)
-
     await db_user.awaitable_attrs.next_plan
     usage_log = UserUsageResetLogs(
         user_id=db_user.id,
         used_traffic_at_reset=db_user.used_traffic,
-        reset_at=reset_at,
     )
     db.add(usage_log)
 
@@ -885,13 +859,7 @@ async def clear_user_node_usages(db: AsyncSession, user_id: int, *, before: date
     await db.execute(stmt)
 
 
-async def reset_user_data_usage(
-    db: AsyncSession,
-    db_user: User,
-    *,
-    skip_node_usage_cleanup: bool = False,
-    node_usage_cleanup_before: datetime | None = None,
-) -> User:
+async def reset_user_data_usage(db: AsyncSession, db_user: User) -> User:
     """
     Resets the data usage of a user and logs the reset.
 
@@ -902,9 +870,8 @@ async def reset_user_data_usage(
     Returns:
         User: The updated user object.
     """
-    reset_at = await _reset_user_traffic_and_log(db, db_user, reset_at=node_usage_cleanup_before)
-    if not skip_node_usage_cleanup:
-        await clear_user_node_usages(db, db_user.id, before=reset_at)
+    await _reset_user_traffic_and_log(db, db_user)
+    await clear_user_node_usages(db, db_user.id)
 
     if db_user.status not in [UserStatus.expired, UserStatus.disabled]:
         db_user.status = UserStatus.active.value
@@ -927,8 +894,8 @@ async def bulk_reset_user_data_usage(db: AsyncSession, users: list[User]) -> lis
         list[User]: The updated list of user objects.
     """
     for db_user in users:
-        reset_at = await _reset_user_traffic_and_log(db, db_user)
-        await clear_user_node_usages(db, db_user.id, before=reset_at)
+        await _reset_user_traffic_and_log(db, db_user)
+        await clear_user_node_usages(db, db_user.id)
         if db_user.status not in [UserStatus.expired, UserStatus.disabled]:
             db_user.status = UserStatus.active.value
     await db.commit()
@@ -991,8 +958,8 @@ async def reset_user_by_next(db: AsyncSession, db_user: User) -> User:
             db_user.proxy_settings = proxy_settings
         db_user.data_limit_reset_strategy = db_user.next_plan.user_template.data_limit_reset_strategy
 
-    reset_at = await _reset_user_traffic_and_log(db, db_user)
-    await clear_user_node_usages(db, db_user.id, before=reset_at)
+    await _reset_user_traffic_and_log(db, db_user)
+    await clear_user_node_usages(db, db_user.id)
     db_user.status = UserStatus.active
 
     await db.commit()

app/db/models.py

@@ -20,6 +20,7 @@
     func,
     or_,
 )
+from sqlalchemy.ext.asyncio import async_object_session
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 from sqlalchemy.sql.expression import select, text
@@ -195,15 +196,28 @@ def last_traffic_reset_time(self):
         return self.usage_logs[-1].reset_at if self.usage_logs else self.created_at
 
     async def inbounds(self) -> list[str]:
-        """Returns a flat list of all included inbound tags across all proxies"""
+        """Returns a flat list of all included inbound tags for enabled groups."""
+        session = async_object_session(self)
+        if session is not None:
+            stmt = (
+                select(ProxyInbound.tag)
+                .select_from(users_groups_association)
+                .join(Group, users_groups_association.c.groups_id == Group.id)
+                .join(inbounds_groups_association, Group.id == inbounds_groups_association.c.group_id)
+                .join(ProxyInbound, inbounds_groups_association.c.inbound_id == ProxyInbound.id)
+                .where(users_groups_association.c.user_id == self.id, Group.is_disabled.is_(False))
+                .distinct()
+            )
+            result = await session.execute(stmt)
+            return list(result.scalars().all())
+
+        # Fallback for detached instances: use already-loaded attrs only.
         included_tags = set()
-        for group in self.groups:
+        for group in self.__dict__.get("groups") or []:
             if group.is_disabled:
                 continue
-
-            await group.awaitable_attrs.inbounds
-            for inbound in group.inbound_tags:
-                included_tags.add(inbound)
+            for inbound in group.__dict__.get("inbounds") or []:
+                included_tags.add(inbound.tag)
         return list(included_tags)
 
     @property

app/operation/user.py

@@ -10,7 +10,6 @@
 
 from app import notification
 from app.db import AsyncSession
-from app.db.base import GetDB, IS_SQLITE
 from app.db.crud.admin import get_admin
 from app.db.crud.bulk import (
     reset_all_users_data_usage,
@@ -21,13 +20,11 @@
 from app.db.crud.user import (
     UsersSortingOptions,
     UsersSortingOptionsSimple,
-    clear_user_node_usages,
     create_user,
     create_users_bulk,
     get_all_users_usages,
     get_existing_usernames,
     get_expired_users,
-    get_user_by_id,
     get_user_lifetime_used_traffic,
     get_user_usages,
     get_users,
@@ -85,28 +82,6 @@ class UserOperation(BaseOperation):
     def _is_non_blocking_sync_operator(operator_type: OperatorType) -> bool:
         return operator_type in (OperatorType.API, OperatorType.WEB)
 
-    @staticmethod
-    async def _sync_user_by_id(user_id: int) -> None:
-        async with GetDB() as sync_db:
-            db_user = await get_user_by_id(
-                sync_db,
-                user_id,
-                load_admin=False,
-                load_next_plan=False,
-                load_usage_logs=False,
-                load_groups=True,
-            )
-            if not db_user:
-                return
-            proto_user = await serialize_user(db_user)
-            await sync_proto_user(proto_user)
-
-    @staticmethod
-    async def _clear_user_node_usages_by_id(user_id: int, cleanup_before: dt) -> None:
-        async with GetDB() as cleanup_db:
-            await clear_user_node_usages(cleanup_db, user_id, before=cleanup_before)
-            await cleanup_db.commit()
-
     @staticmethod
     def _format_validation_errors(error: ValidationError) -> str:
         return "; ".join(
@@ -262,11 +237,8 @@ async def update_user(
         self, db: AsyncSession, db_user: User, *, include_lifetime_used_traffic: bool = True
     ) -> UserNotificationResponse:
         if self._is_non_blocking_sync_operator(self.operator_type):
-            if IS_SQLITE:
-                proto_user = await serialize_user(db_user)
-                schedule_sync_task(sync_proto_user(proto_user))
-            else:
-                schedule_sync_task(self._sync_user_by_id(db_user.id))
+            proto_user = await serialize_user(db_user)
+            schedule_sync_task(sync_proto_user(proto_user))
         else:
             await sync_user(db_user)
 
@@ -345,17 +317,7 @@ async def remove_user(self, db: AsyncSession, username: str, admin: AdminDetails
     async def _reset_user_data_usage(self, db: AsyncSession, db_user: User, admin: AdminDetails):
         old_status = db_user.status
 
-        skip_node_usage_cleanup = self._is_non_blocking_sync_operator(self.operator_type) and not IS_SQLITE
-        cleanup_before = dt.now(tz.utc)
-        db_user = await reset_user_data_usage(
-            db=db,
-            db_user=db_user,
-            skip_node_usage_cleanup=skip_node_usage_cleanup,
-            node_usage_cleanup_before=cleanup_before,
-        )
-        if skip_node_usage_cleanup:
-            schedule_sync_task(self._clear_user_node_usages_by_id(db_user.id, cleanup_before))
-
+        db_user = await reset_user_data_usage(db=db, db_user=db_user)
         user = await self.update_user(db, db_user, include_lifetime_used_traffic=False)
 
         if user.status != old_status: