perf(crud): eliminate N+1 queries by eagerly loading relations

app/db/crud/admin.py

@@ -379,6 +379,13 @@ async def get_admins(
     if load_role:
         stmt = stmt.options(selectinload(Admin.role))
 
+    # For non-compact path, eagerly load users and usage_logs to avoid N+1 queries
+    if not compact:
+        stmt = stmt.options(
+            selectinload(Admin.users),
+            selectinload(Admin.usage_logs),
+        )
+
     # Apply filters consistently
     if params.ids:
         stmt = stmt.where(Admin.id.in_(params.ids))
@@ -405,9 +412,8 @@ async def get_admins(
         for admin, total_users, reseted_usage in rows:
             admins.append(build_admin_details(admin, total_users=total_users, reseted_usage=reseted_usage))
     else:
-        admins = list((await db.execute(stmt)).scalars().all())
-        for admin in admins:
-            await load_admin_attrs(admin, load_role=load_role)
+        # users, usage_logs, and role already eagerly loaded via selectinload above
+        admins = list((await db.execute(stmt)).unique().scalars().all())
 
     if return_with_count:
         return admins, total, active, disabled, limited
@@ -524,20 +530,60 @@ async def get_usage_percentage_reached_admins(
     return list((await db.execute(stmt)).scalars().all())
 
 
-async def bulk_create_admin_notification_reminders(db: AsyncSession, reminder_data: list[dict]) -> None:
+async def bulk_create_admin_notification_reminders(
+    db: AsyncSession, reminder_data: list[dict]
+) -> list[dict]:
     """Bulk-insert admin reminder rows after successful sends."""
     if not reminder_data:
-        return
+        return []
 
-    db.add_all([AdminNotificationReminder(**data) for data in reminder_data])
-    await db.commit()
+    # De-duplicate input list to preserve only the first occurrence
+    seen_input = set()
+    unique_reminder_data = []
+    for data in reminder_data:
+        key = (data["admin_id"], data["type"], data["threshold"])
+        if key not in seen_input:
+            seen_input.add(key)
+            unique_reminder_data.append(data)
+
+    admin_ids = {d["admin_id"] for d in unique_reminder_data}
+    types = {d["type"] for d in unique_reminder_data}
+
+    # Lock the Admin rows to serialize reminder checks/creation for these admins
+    await db.execute(
+        select(Admin.id)
+        .where(Admin.id.in_(list(admin_ids)))
+        .with_for_update()
+    )
+
+    # Fetch existing reminders that match these criteria
+    stmt = select(AdminNotificationReminder).where(
+        AdminNotificationReminder.admin_id.in_(list(admin_ids)),
+        AdminNotificationReminder.type.in_(list(types)),
+    )
+    result = await db.execute(stmt)
+    existing = {(r.admin_id, r.type, r.threshold) for r in result.scalars().all()}
+
+    to_insert = []
+    inserted_data = []
+    for data in unique_reminder_data:
+        key = (data["admin_id"], data["type"], data["threshold"])
+        if key not in existing:
+            to_insert.append(AdminNotificationReminder(**data))
+            inserted_data.append(data)
+
+    if to_insert:
+        db.add_all(to_insert)
+        await db.commit()
+
+    return inserted_data
 
 
 async def create_admin_notification_reminder_if_absent(
     db: AsyncSession,
     admin_id: int,
     reminder_type: ReminderType,
-    threshold: int | None = None,
+    threshold: int = 0,
 ) -> bool:
     """Create an admin reminder row and return whether this call claimed it."""
     existing_reminder_id = await db.scalar(

app/db/crud/bulk.py

@@ -3,6 +3,7 @@
 from sqlalchemy import and_, case, cast, delete, func, or_, select, text, update
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import joinedload, selectinload
 
 from app.db.models import (
     Admin,
@@ -443,9 +444,20 @@ async def update_users_proxy_settings(
     await db.execute(update_stmt)
     await db.commit()
 
-    # Refresh the user objects to get updated values
-    for user in users_to_update:
-        await db.refresh(user)
-        await load_user_attrs(user, load_admin_role=True)
+    # Re-fetch all updated users in a single query instead of N individual refreshes
+    updated_user_ids = [u.id for u in users_to_update]
+    result = await db.execute(
+        select(User)
+        .where(User.id.in_(updated_user_ids))
+        .options(
+            joinedload(User.admin).selectinload(Admin.role),
+            joinedload(User.next_plan),
+            selectinload(User.usage_logs),
+            selectinload(User.groups),
+        )
+    )
+    refreshed_users = result.scalars().all()
+    refreshed_users_map = {user.id: user for user in refreshed_users}
+    ordered_refreshed_users = [refreshed_users_map[uid] for uid in updated_user_ids if uid in refreshed_users_map]
 
-    return users_to_update, count_effctive_users
+    return ordered_refreshed_users, count_effctive_users

app/db/crud/group.py

@@ -105,11 +105,15 @@ async def get_group(db: AsyncSession, query: GroupListQuery) -> tuple[list[Group
             - list[Group]: A list of Group objects
             - int: The total count of groups
     """
-    groups = select(Group)
+    groups = select(Group).options(selectinload(Group.users), selectinload(Group.inbounds))
     if query.ids:
         groups = groups.where(Group.id.in_(query.ids))
 
-    count_query = select(func.count()).select_from(groups.subquery())
+    # Build count on the base filter before adding pagination or eager loads
+    base_stmt = select(Group)
+    if query.ids:
+        base_stmt = base_stmt.where(Group.id.in_(query.ids))
+    count_query = select(func.count()).select_from(base_stmt.subquery())
 
     if query.offset:
         groups = groups.offset(query.offset)
@@ -118,10 +122,8 @@ async def get_group(db: AsyncSession, query: GroupListQuery) -> tuple[list[Group
 
     count = (await db.execute(count_query)).scalar_one()
 
-    all_groups = (await db.execute(groups)).scalars().all()
-
-    for group in all_groups:
-        await load_group_attrs(group)
+    # users and inbounds already eagerly loaded via selectinload above
+    all_groups = (await db.execute(groups)).unique().scalars().all()
 
     return all_groups, count
 

app/db/crud/node.py

@@ -3,6 +3,7 @@
 from sqlalchemy import and_, bindparam, case, delete, func, literal_column, or_, select, update
 from sqlalchemy.exc import InvalidRequestError
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
 from sqlalchemy.sql.functions import coalesce
 
 from app.db.compiles_types import DateDiff
@@ -142,9 +143,10 @@ async def get_nodes(
     # Order by created_at and id for consistent results
     stmt = stmt.order_by(Node.created_at.asc(), Node.id.asc())
 
-    db_nodes = (await db.execute(stmt)).scalars().all()
-    for node in db_nodes:
-        await load_node_attrs(node)
+    # Eagerly load usage_logs to avoid N+1 queries (one extra SELECT per node)
+    stmt = stmt.options(selectinload(Node.usage_logs))
+
+    db_nodes = (await db.execute(stmt)).unique().scalars().all()
 
     return db_nodes, count
 
@@ -206,15 +208,13 @@ async def get_limited_nodes(db: AsyncSession) -> list[Node]:
     Returns:
         list[Node]: Nodes that should be limited
     """
-    query = select(Node).where(
+    query = select(Node).options(selectinload(Node.usage_logs)).where(
         and_(
             Node.status.in_([NodeStatus.error, NodeStatus.connected, NodeStatus.connecting]),
             Node.is_limited,
         )
     )
-    nodes = (await db.execute(query)).scalars().all()
-    for node in nodes:
-        await load_node_attrs(node)
+    nodes = (await db.execute(query)).unique().scalars().all()
     return nodes
 
 
@@ -611,6 +611,7 @@ async def get_nodes_to_reset_usage(db: AsyncSession) -> list[Node]:
 
     stmt = (
         select(Node)
+        .options(selectinload(Node.usage_logs))
         .outerjoin(last_reset_subq, Node.id == last_reset_subq.c.node_id)
         .where(
             Node.status.in_([NodeStatus.connected, NodeStatus.limited, NodeStatus.error, NodeStatus.connecting]),
@@ -626,9 +627,7 @@ async def get_nodes_to_reset_usage(db: AsyncSession) -> list[Node]:
 
     nodes = list((await db.execute(stmt)).unique().scalars().all())
 
-    # Load node attributes to avoid greenlet errors
-    for node in nodes:
-        await load_node_attrs(node)
+    # usage_logs already eagerly loaded via selectinload — no extra queries needed
 
     # For nodes with reset_time >= 0, filter based on absolute time
 
@@ -796,10 +795,17 @@ async def bulk_reset_node_usage(db: AsyncSession, nodes: list[Node]) -> list[Nod
             db_node.status = NodeStatus.connecting
 
     await db.commit()
-    for node in nodes:
-        await db.refresh(node)
-        await load_node_attrs(node)
-    return nodes
+
+    # Re-fetch all nodes in a single query instead of N individual refreshes
+    node_ids = [node.id for node in nodes]
+    refreshed = (
+        await db.execute(
+            select(Node).options(selectinload(Node.usage_logs)).where(Node.id.in_(node_ids))
+        )
+    ).unique().scalars().all()
+    # Preserve input order
+    refreshed_by_id = {n.id: n for n in refreshed}
+    return [refreshed_by_id[nid] for nid in node_ids if nid in refreshed_by_id]
 
 
 async def remove_nodes(db: AsyncSession, node_ids: list[int]) -> None:

app/db/crud/user.py

@@ -619,17 +619,15 @@ async def get_usage_percentage_reached_users(db: AsyncSession, percentage: int)
     )
 
     stmt = (
-        select(User)
+        _build_user_select_stmt()
         .options(joinedload(User.notification_reminders))
         .where(User.status == UserStatus.active)
         .where(User.usage_percentage >= percentage)
         .where(not_(existing_reminder_subq))  # Only users without existing reminders
     )
 
-    users = list((await db.execute(stmt)).unique().scalars().all())
-    for user in users:
-        await load_user_attrs(user)
-    return users
+    # All relations (admin, next_plan, usage_logs, groups) eagerly loaded via _build_user_select_stmt
+    return list((await db.execute(stmt)).unique().scalars().all())
 
 
 async def get_days_left_reached_users(db: AsyncSession, days: int) -> list[User]:
@@ -649,18 +647,16 @@ async def get_days_left_reached_users(db: AsyncSession, days: int) -> list[User]
     )
 
     stmt = (
-        select(User)
+        _build_user_select_stmt()
         .options(joinedload(User.notification_reminders))
         .where(User.status == UserStatus.active)
         .where(User.expire.isnot(None))
         .where(User.days_left == days)
         .where(not_(existing_reminder_subq))  # Only users without existing reminders
     )
 
-    users = list((await db.execute(stmt)).unique().scalars().all())
-    for user in users:
-        await load_user_attrs(user)
-    return users
+    # All relations (admin, next_plan, usage_logs, groups) eagerly loaded via _build_user_select_stmt
+    return list((await db.execute(stmt)).unique().scalars().all())
 
 
 async def get_user_usages(

app/db/migrations/versions/bb4a32b7f5ce_add_admin_notification_reminders.py

@@ -18,6 +18,9 @@
 depends_on = None
 
 
+from app.db.compiles_types import SqliteCompatibleBigInteger
+
+
 def upgrade() -> None:
     bind = op.get_bind()
     dialect = bind.dialect.name
@@ -30,7 +33,7 @@ def upgrade() -> None:
 
     op.create_table(
         "admin_notification_reminders",
-        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("id", SqliteCompatibleBigInteger(), autoincrement=True, nullable=False),
         sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
         sa.Column("admin_id", sa.BigInteger(), nullable=False),
         sa.Column("type", reminder_type, nullable=False),

app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py

@@ -0,0 +1,64 @@
+"""add unique constraint to admin notification reminders
+
+Revision ID: fb32155473c1
+Revises: 3c1a7e5b9d20
+Create Date: 2026-07-26 19:32:42.096475
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'fb32155473c1'
+down_revision = '3c1a7e5b9d20'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # 1. Update any existing NULL thresholds to 0
+    op.execute(
+        "UPDATE admin_notification_reminders SET threshold = 0 WHERE threshold IS NULL"
+    )
+
+    # 2. Clean up duplicate rows before creating the constraint
+    op.execute(
+        """
+        DELETE FROM admin_notification_reminders
+        WHERE id NOT IN (
+            SELECT min_id FROM (
+                SELECT MIN(id) AS min_id
+                FROM admin_notification_reminders
+                GROUP BY admin_id, type, threshold
+            ) AS temp
+        )
+        """
+    )
+
+    # 3. Alter threshold column to be non-nullable with a server default of 0 and create the constraint
+    with op.batch_alter_table("admin_notification_reminders", schema=None) as batch_op:
+        batch_op.alter_column(
+            "threshold",
+            existing_type=sa.Integer(),
+            nullable=False,
+            server_default="0"
+        )
+        batch_op.create_unique_constraint(
+            "uq_admin_notification_reminders_admin_id_type_threshold",
+            ["admin_id", "type", "threshold"]
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("admin_notification_reminders", schema=None) as batch_op:
+        batch_op.drop_constraint(
+            "uq_admin_notification_reminders_admin_id_type_threshold",
+            type_="unique"
+        )
+        batch_op.alter_column(
+            "threshold",
+            existing_type=sa.Integer(),
+            nullable=True,
+            server_default=None
+        )

app/db/models.py

@@ -749,11 +749,14 @@ class NotificationReminder(Base, CreatedAtUTCMixin):
 
 class AdminNotificationReminder(Base, CreatedAtUTCMixin):
     __tablename__ = "admin_notification_reminders"
-    __table_args__ = (Index("ix_admin_notification_reminders_admin_id_type", "admin_id", "type"),)
+    __table_args__ = (
+        Index("ix_admin_notification_reminders_admin_id_type", "admin_id", "type"),
+        UniqueConstraint("admin_id", "type", "threshold", name="uq_admin_notification_reminders_admin_id_type_threshold"),
+    )
     admin_id: Mapped[int] = fk_id_column("admins.id", ondelete="CASCADE")
     admin: Mapped[Admin] = relationship(back_populates="notification_reminders", init=False)
     type: Mapped[ReminderType] = mapped_column(SQLEnum(ReminderType))
-    threshold: Mapped[int | None] = mapped_column(default=None)
+    threshold: Mapped[int] = mapped_column(server_default="0", default=0)
 
 
 class Group(Base, IdMixin):