fix: use single function to update node user

M03ED · M03ED · commit 92de1b746068 · 2025-11-18T21:36:14.000+03:30
diff --git a/app/db/crud/admin.py b/app/db/crud/admin.py
@@ -198,23 +198,23 @@ async def get_admins(
     total = None
     active = None
     disabled = None
-    
+
     if return_with_count:
         # Get total count
         count_stmt = select(func.count(Admin.id))
         if username:
             count_stmt = count_stmt.where(Admin.username.ilike(f"%{username}%"))
         result = await db.execute(count_stmt)
         total = result.scalar()
-        
+
         # Get active count (not disabled)
         active_stmt = select(func.count(Admin.id))
         if username:
             active_stmt = active_stmt.where(Admin.username.ilike(f"%{username}%"))
         active_stmt = active_stmt.where(Admin.is_disabled.is_(False))
         result = await db.execute(active_stmt)
         active = result.scalar()
-        
+
         # Get disabled count
         disabled_stmt = select(func.count(Admin.id))
         if username:
diff --git a/app/jobs/node_checker.py b/app/jobs/node_checker.py
@@ -92,7 +92,9 @@ async def process_node_health_check(db_node: Node, node: PasarGuardNode):
         health = await asyncio.wait_for(verify_node_backend_health(node, db_node.name), timeout=20)
     except asyncio.TimeoutError:
         if db_node.status == NodeStatus.connected:
-            logger.warning(f"Node {db_node.id} ({db_node.name}) health check timed out but was previously connected, will retry")
+            logger.warning(
+                f"Node {db_node.id} ({db_node.name}) health check timed out but was previously connected, will retry"
+            )
             return
         async with GetDB() as db:
             await NodeOperation._update_single_node_status(
diff --git a/app/jobs/reset_user_data_usage.py b/app/jobs/reset_user_data_usage.py
@@ -3,17 +3,16 @@
 
 from app import scheduler
 from app.db import GetDB
-from app.db.models import UserStatus
 from app.db.crud.user import get_users_to_reset_data_usage, bulk_reset_user_data_usage
-from app.models.user import UserNotificationResponse
+from app.operation import OperatorType
+from app.operation.user import UserOperation
 from app import notification
-from app.core.manager import core_manager
-from app.node import node_manager
 from app.jobs.dependencies import SYSTEM_ADMIN
 from app.utils.logger import get_logger
 from config import JOB_RESET_USER_DATA_USAGE_INTERVAL
 
 logger = get_logger("jobs")
+user_operator = UserOperation(operator_type=OperatorType.SYSTEM)
 
 
 async def reset_data_usage():
@@ -24,16 +23,12 @@ async def reset_data_usage():
         updated_users = await bulk_reset_user_data_usage(db, users)
 
         for db_user in updated_users:
-            user = UserNotificationResponse.model_validate(db_user)
+            user = await user_operator.update_user(db_user)
             asyncio.create_task(notification.reset_user_data_usage(user, SYSTEM_ADMIN))
 
             if old_statuses.get(user.id) != user.status:
                 asyncio.create_task(notification.user_status_change(user, SYSTEM_ADMIN))
 
-            # make user active if limited on usage reset
-            if user.status == UserStatus.active:
-                await node_manager.update_user(user=user, inbounds=await core_manager.get_inbounds())
-
             logger.info(f'User data usage reset for User "{user.username}"')
 
 
diff --git a/app/jobs/review_users.py b/app/jobs/review_users.py
@@ -17,6 +17,8 @@
     update_users_status,
     bulk_create_notification_reminders,
 )
+from app.operation import OperatorType
+from app.operation.user import UserOperation
 from app.jobs.dependencies import SYSTEM_ADMIN
 from app.models.settings import Webhook
 from app.models.user import UserNotificationResponse
@@ -26,28 +28,25 @@
 from config import JOB_REVIEW_USERS_INTERVAL
 
 logger = get_logger("review-users")
+user_operator = UserOperation(operator_type=OperatorType.SYSTEM)
 
 
 async def reset_user_by_next_report(db: AsyncSession, db_user: User):
     db_user = await reset_user_by_next(db, db_user)
-    inbounds = await db_user.inbounds()
-    user = UserNotificationResponse.model_validate(db_user)
+    user = await user_operator.update_user(db_user)
 
-    await node_manager.update_user(user, inbounds)
     asyncio.create_task(notification.user_data_reset_by_next(user, SYSTEM_ADMIN))
 
     logger.info(f'User "{db_user.username}" next plan activated')
 
 
 async def change_status(db: AsyncSession, db_user: User, status: UserStatus):
-    user = UserNotificationResponse.model_validate(db_user)
-    if user.status is not UserStatus.active:
-        await node_manager.remove_user(user)
+    user = await user_operator.update_user(db_user)
     asyncio.create_task(notification.user_status_change(user, SYSTEM_ADMIN))
 
-    logger.info(f'User "{db_user.username}" status changed to {status.value}')
+    logger.info(f'User "{user.username}" status changed to {status.value}')
 
-    if db_user.next_plan and db_user.status is not UserStatus.active:
+    if db_user.next_plan and db_user.status != UserStatus.active:
         await reset_user_by_next_report(db, db_user)
 
 
diff --git a/app/models/admin.py b/app/models/admin.py
@@ -115,7 +115,8 @@ class AdminValidationResult(BaseModel):
 
 class AdminsResponse(BaseModel):
     """Response model for admins list with pagination and statistics."""
+
     admins: list[AdminDetails]
     total: int
     active: int
-    disabled: int
+    disabled: int
diff --git a/app/operation/admin.py b/app/operation/admin.py
@@ -15,7 +15,6 @@
 )
 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.db.models import Admin as DBAdmin
 from app.models.admin import AdminCreate, AdminDetails, AdminModify, AdminsResponse
 from app.node import node_manager
 from app.operation import BaseOperation, OperatorType
@@ -97,7 +96,7 @@ async def get_admins(
         admins, total, active, disabled = await get_admins(
             db, offset, limit, username, sort_list if sort_list else None, return_with_count=True
         )
-        
+
         return AdminsResponse(
             admins=[AdminDetails.model_validate(admin) for admin in admins],
             total=total,
diff --git a/app/operation/user.py b/app/operation/user.py
@@ -89,7 +89,7 @@ async def update_user(self, db_user: User) -> UserNotificationResponse:
             user_inbounds = await db_user.inbounds()
             asyncio.create_task(node_manager.update_user(user, inbounds=user_inbounds))
         else:
-            asyncio.create_task(node_manager.update_user(user))
+            asyncio.create_task(node_manager.remove_user(user))
 
         return user
 
diff --git a/app/subscription/xray.py b/app/subscription/xray.py
@@ -248,27 +248,31 @@ def _transport_quic(self, config: QUICTransportConfig, path: str) -> dict:
         """Handle QUIC transport - only gets QUIC config"""
         host = config.host if isinstance(config.host, str) else (config.host[0] if config.host else "")
 
-        return self._normalize_and_remove_none_values({
-            "security": host,
-            "header": {"type": config.header_type},
-            "key": path,
-        })
+        return self._normalize_and_remove_none_values(
+            {
+                "security": host,
+                "header": {"type": config.header_type},
+                "key": path,
+            }
+        )
 
     def _transport_kcp(self, config: KCPTransportConfig, path: str) -> dict:
         """Handle KCP transport - only gets KCP config"""
         host = config.host if isinstance(config.host, str) else (config.host[0] if config.host else "")
 
-        return self._normalize_and_remove_none_values({
-            "header": {"type": config.header_type, "domain": host},
-            "mtu": config.mtu if config.mtu else 1350,
-            "tti": config.tti if config.tti else 50,
-            "uplinkCapacity": config.uplink_capacity if config.uplink_capacity else 12,
-            "downlinkCapacity": config.downlink_capacity if config.downlink_capacity else 100,
-            "congestion": config.congestion,
-            "readBufferSize": config.read_buffer_size if config.read_buffer_size else 2,
-            "writeBufferSize": config.write_buffer_size if config.write_buffer_size else 2,
-            "seed": path,
-        })
+        return self._normalize_and_remove_none_values(
+            {
+                "header": {"type": config.header_type, "domain": host},
+                "mtu": config.mtu if config.mtu else 1350,
+                "tti": config.tti if config.tti else 50,
+                "uplinkCapacity": config.uplink_capacity if config.uplink_capacity else 12,
+                "downlinkCapacity": config.downlink_capacity if config.downlink_capacity else 100,
+                "congestion": config.congestion,
+                "readBufferSize": config.read_buffer_size if config.read_buffer_size else 2,
+                "writeBufferSize": config.write_buffer_size if config.write_buffer_size else 2,
+                "seed": path,
+            }
+        )
 
     def _apply_transport(self, network: str, inbound: SubscriptionInboundData, path: str) -> dict | None:
         """Apply transport settings using registry pattern"""
@@ -284,15 +288,17 @@ def _apply_tls(self, tls_config: TLSConfig, security: str) -> dict:
         sni = tls_config.sni if isinstance(tls_config.sni, str) else (tls_config.sni[0] if tls_config.sni else None)
 
         if security == "reality":
-            return self._normalize_and_remove_none_values({
-                "serverName": sni,
-                "fingerprint": tls_config.fingerprint,
-                "show": False,
-                "publicKey": tls_config.reality_public_key,
-                "shortId": tls_config.reality_short_id,
-                "spiderX": tls_config.reality_spx,
-                "mldsa65Verify": tls_config.mldsa65_verify,
-            })
+            return self._normalize_and_remove_none_values(
+                {
+                    "serverName": sni,
+                    "fingerprint": tls_config.fingerprint,
+                    "show": False,
+                    "publicKey": tls_config.reality_public_key,
+                    "shortId": tls_config.reality_short_id,
+                    "spiderX": tls_config.reality_spx,
+                    "mldsa65Verify": tls_config.mldsa65_verify,
+                }
+            )
         else:  # tls
             config = {
                 "serverName": sni,
@@ -356,11 +362,13 @@ def _download_config(self, download_settings: SubscriptionInboundData, link_form
             sockopt=sockopt,
         )
 
-        return self._normalize_and_remove_none_values({
-            "address": download_settings.address,
-            "port": self._select_port(download_settings.port),
-            **stream_settings,
-        })
+        return self._normalize_and_remove_none_values(
+            {
+                "address": download_settings.address,
+                "port": self._select_port(download_settings.port),
+                **stream_settings,
+            }
+        )
 
     # ========== Protocol Builders (Registry Methods) ==========
 
