From 8b7ea6fb4b1291747615a33b2281932a26cae1fa Mon Sep 17 00:00:00 2001
From: x0sina <bulletsina77@gmail.com>
Date: Sun, 26 Jul 2026 20:22:23 +0330
Subject: [PATCH 01/11] perf(crud): eliminate N+1 queries by eagerly loading
 relations

- Add selectinload for Admin.users, Admin.usage_logs, and Admin.role in get_admins to prevent sequential queries when not in compact mode
- Add selectinload for Group.users and Group.inbounds in get_group to eagerly load relations
- Add selectinload for Node.usage_logs in get_nodes, get_limited_nodes, and get_nodes_to_reset_usage to avoid loading attributes sequentially
- Replace individual db.refresh() calls with single batch re-fetch queries in bulk_reset_node_usage and update_users_proxy_settings
- Use .unique() on query results when eager loading relations to handle duplicates from joins
- Remove sequential load_admin_attrs, load_group_attrs, and load_node_attrs calls where relations are now eagerly loaded
- Use selectinload in _build_user_select_stmt to eagerly load all user relations (admin, next_plan, usage_logs, groups)
- Improves query performance by reducing database round trips and eliminating N+1 query patterns across admin, group, node, and user CRUD operations
---
 app/db/crud/admin.py      | 12 +++++++++---
 app/db/crud/bulk.py       | 10 ++++++----
 app/db/crud/group.py      | 14 ++++++++------
 app/db/crud/node.py       | 34 ++++++++++++++++++++--------------
 app/db/crud/user.py       | 16 ++++++----------
 app/jobs/record_usages.py | 12 ++++++++++--
 app/jobs/review_admins.py | 24 +++++++++++++++---------
 7 files changed, 74 insertions(+), 48 deletions(-)

diff --git a/app/db/crud/admin.py b/app/db/crud/admin.py
index d4d15be6b..235823a16 100644
--- a/app/db/crud/admin.py
+++ b/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
diff --git a/app/db/crud/bulk.py b/app/db/crud/bulk.py
index dd7258204..f5d5304e1 100644
--- a/app/db/crud/bulk.py
+++ b/app/db/crud/bulk.py
@@ -443,9 +443,11 @@ 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)
+    # 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)))
+    refreshed_users = result.scalars().all()
+    for user in refreshed_users:
         await load_user_attrs(user, load_admin_role=True)
 
-    return users_to_update, count_effctive_users
+    return refreshed_users, count_effctive_users
diff --git a/app/db/crud/group.py b/app/db/crud/group.py
index f74580b43..64140c0c6 100644
--- a/app/db/crud/group.py
+++ b/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
 
diff --git a/app/db/crud/node.py b/app/db/crud/node.py
index 2dac73408..94baf5900 100644
--- a/app/db/crud/node.py
+++ b/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:
diff --git a/app/db/crud/user.py b/app/db/crud/user.py
index f21f77a7b..1c85cd417 100644
--- a/app/db/crud/user.py
+++ b/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,7 +647,7 @@ 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))
@@ -657,10 +655,8 @@ async def get_days_left_reached_users(db: AsyncSession, days: int) -> list[User]
         .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(
diff --git a/app/jobs/record_usages.py b/app/jobs/record_usages.py
index 7a38e11d6..5db0c20fd 100644
--- a/app/jobs/record_usages.py
+++ b/app/jobs/record_usages.py
@@ -100,9 +100,17 @@ def _chunked(items: list, size: int):
 
 
 async def get_dialect() -> str:
-    """Get the database dialect name without holding the session open."""
+    """Get the database dialect name. Cached after first call since the dialect never changes."""
+    if _dialect_cache:
+        return _dialect_cache[0]
     async with GetDB() as db:
-        return db.bind.dialect.name
+        dialect = db.bind.dialect.name
+    _dialect_cache.append(dialect)
+    return dialect
+
+
+# Simple one-element list used as a mutable cache container (set once, read many times)
+_dialect_cache: list[str] = []
 
 
 def build_node_user_usage_upsert(dialect: str, upsert_params: list[dict]):
diff --git a/app/jobs/review_admins.py b/app/jobs/review_admins.py
index d79507ceb..ef00ac58a 100644
--- a/app/jobs/review_admins.py
+++ b/app/jobs/review_admins.py
@@ -14,7 +14,7 @@
 from app import notification, scheduler
 from app.db import GetDB
 from app.db.crud.admin import (
-    create_admin_notification_reminder_if_absent,
+    bulk_create_admin_notification_reminders,
     get_usage_percentage_reached_admins,
 )
 from app.db.models import Admin, ReminderType
@@ -69,19 +69,25 @@ async def _send_usage_limit_warning_notifications(db):
 
     for threshold in default_thresholds:
         threshold_admins = await get_usage_percentage_reached_admins(db, threshold)
+
+        reminder_data = []
+        notifications_to_send = []
+
         for admin in threshold_admins:
             if not admin.data_limit or admin.data_limit <= 0:
                 continue
-            reminder_created = await create_admin_notification_reminder_if_absent(
-                db,
-                admin.id,
-                ReminderType.data_usage,
-                threshold,
-            )
-            if not reminder_created:
-                continue
             usage_percentage = int((admin.used_traffic * 100) / admin.data_limit)
             admin_model = _admin_usage_warning_details(admin)
+            reminder_data.append(
+                {"admin_id": admin.id, "type": ReminderType.data_usage, "threshold": threshold}
+            )
+            notifications_to_send.append((admin_model, usage_percentage, threshold))
+
+        # Bulk-insert all reminders for this threshold in one query
+        await bulk_create_admin_notification_reminders(db, reminder_data)
+
+        # Send notifications after the bulk insert succeeds
+        for admin_model, usage_percentage, threshold in notifications_to_send:
             await notification.admin_usage_limit_reached(admin_model, usage_percentage, threshold)
 
 

From 2c60ddf62ebb10b8082dc28ad059c89f65f86932 Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Sun, 26 Jul 2026 18:18:35 +0000
Subject: [PATCH 02/11] =?UTF-8?q?Eager-load=20the=20refreshed=20users?=
 =?UTF-8?q?=E2=80=99=20dependent=20relationships.?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/db/crud/bulk.py | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/app/db/crud/bulk.py b/app/db/crud/bulk.py
index f5d5304e1..9427c285e 100644
--- a/app/db/crud/bulk.py
+++ b/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,
@@ -445,9 +446,16 @@ async def update_users_proxy_settings(
 
     # 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)))
+    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()
-    for user in refreshed_users:
-        await load_user_attrs(user, load_admin_role=True)
 
     return refreshed_users, count_effctive_users

From 1ea1c7c7000e97d1dadaed954374d958ec9e5766 Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Sun, 26 Jul 2026 18:50:50 +0000
Subject: [PATCH 03/11] Preserve the input order when returning refreshed
 users.

---
 app/db/crud/bulk.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/app/db/crud/bulk.py b/app/db/crud/bulk.py
index 9427c285e..b3fb27f0a 100644
--- a/app/db/crud/bulk.py
+++ b/app/db/crud/bulk.py
@@ -457,5 +457,7 @@ async def update_users_proxy_settings(
         )
     )
     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 refreshed_users, count_effctive_users
+    return ordered_refreshed_users, count_effctive_users

From adeef038e36b50ca38927af2efe10bce7e9019e9 Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Sun, 26 Jul 2026 18:59:28 +0000
Subject: [PATCH 04/11] Preserve idempotent reminder creation.

---
 app/db/crud/admin.py                          | 48 +++++++++-
 ...b7f5ce_add_admin_notification_reminders.py |  5 +-
 app/jobs/review_admins.py                     | 13 ++-
 tests/test_review_admins_unit.py              | 87 +++++++++++++++++++
 4 files changed, 145 insertions(+), 8 deletions(-)
 create mode 100644 tests/test_review_admins_unit.py

diff --git a/app/db/crud/admin.py b/app/db/crud/admin.py
index 235823a16..55e748af9 100644
--- a/app/db/crud/admin.py
+++ b/app/db/crud/admin.py
@@ -530,13 +530,53 @@ 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(
diff --git a/app/db/migrations/versions/bb4a32b7f5ce_add_admin_notification_reminders.py b/app/db/migrations/versions/bb4a32b7f5ce_add_admin_notification_reminders.py
index 60370c7e9..1c2029537 100644
--- a/app/db/migrations/versions/bb4a32b7f5ce_add_admin_notification_reminders.py
+++ b/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),
diff --git a/app/jobs/review_admins.py b/app/jobs/review_admins.py
index ef00ac58a..d4db514c6 100644
--- a/app/jobs/review_admins.py
+++ b/app/jobs/review_admins.py
@@ -71,7 +71,7 @@ async def _send_usage_limit_warning_notifications(db):
         threshold_admins = await get_usage_percentage_reached_admins(db, threshold)
 
         reminder_data = []
-        notifications_to_send = []
+        admin_info_map = {}
 
         for admin in threshold_admins:
             if not admin.data_limit or admin.data_limit <= 0:
@@ -81,10 +81,17 @@ async def _send_usage_limit_warning_notifications(db):
             reminder_data.append(
                 {"admin_id": admin.id, "type": ReminderType.data_usage, "threshold": threshold}
             )
-            notifications_to_send.append((admin_model, usage_percentage, threshold))
+            admin_info_map[admin.id] = (admin_model, usage_percentage)
 
         # Bulk-insert all reminders for this threshold in one query
-        await bulk_create_admin_notification_reminders(db, reminder_data)
+        inserted_reminders = await bulk_create_admin_notification_reminders(db, reminder_data)
+
+        notifications_to_send = []
+        for item in inserted_reminders:
+            admin_id = item["admin_id"]
+            if admin_id in admin_info_map:
+                admin_model, usage_percentage = admin_info_map[admin_id]
+                notifications_to_send.append((admin_model, usage_percentage, threshold))
 
         # Send notifications after the bulk insert succeeds
         for admin_model, usage_percentage, threshold in notifications_to_send:
diff --git a/tests/test_review_admins_unit.py b/tests/test_review_admins_unit.py
new file mode 100644
index 000000000..cdc1baf78
--- /dev/null
+++ b/tests/test_review_admins_unit.py
@@ -0,0 +1,87 @@
+import pytest
+from sqlalchemy import select
+from app.db.models import Admin, AdminNotificationReminder, ReminderType
+from app.db.crud.admin import bulk_create_admin_notification_reminders
+from app.jobs.review_admins import _send_usage_limit_warning_notifications
+from tests.api import TestSession
+from unittest.mock import AsyncMock, MagicMock, patch
+
+@pytest.mark.asyncio
+async def test_bulk_create_admin_notification_reminders_idempotency():
+    async with TestSession() as session:
+        # Create an admin to associate reminders with
+        admin = Admin(username="idempotent_admin", hashed_password="secret", role_id=3)
+        session.add(admin)
+        await session.flush()
+        
+        reminder_data = [
+            {"admin_id": admin.id, "type": ReminderType.data_usage, "threshold": 80},
+            {"admin_id": admin.id, "type": ReminderType.data_usage, "threshold": 80}, # duplicate input
+        ]
+        
+        # Call the bulk helper
+        inserted = await bulk_create_admin_notification_reminders(session, reminder_data)
+        
+        # Check that only one was inserted and returned
+        assert len(inserted) == 1
+        assert inserted[0]["admin_id"] == admin.id
+        assert inserted[0]["threshold"] == 80
+        
+        # Query DB to verify
+        db_reminders = (await session.execute(
+            select(AdminNotificationReminder).where(AdminNotificationReminder.admin_id == admin.id)
+        )).scalars().all()
+        assert len(db_reminders) == 1
+        assert db_reminders[0].threshold == 80
+        
+        # Run again with same data
+        inserted_second = await bulk_create_admin_notification_reminders(session, reminder_data)
+        assert len(inserted_second) == 0
+        
+        # DB count should still be 1
+        db_reminders_second = (await session.execute(
+            select(AdminNotificationReminder).where(AdminNotificationReminder.admin_id == admin.id)
+        )).scalars().all()
+        assert len(db_reminders_second) == 1
+
+@pytest.mark.asyncio
+@patch("app.jobs.review_admins.notification")
+@patch("app.jobs.review_admins.notification_enable")
+@patch("app.jobs.review_admins.get_usage_percentage_reached_admins")
+@patch("app.jobs.review_admins._admin_usage_warning_details")
+async def test_send_usage_limit_warning_notifications_idempotent(
+    mock_details, mock_get_admins, mock_notif_enable, mock_notification
+):
+    async with TestSession() as session:
+        # Create test admin
+        admin = Admin(
+            username="notif_admin",
+            hashed_password="secret",
+            role_id=3,
+            data_limit=1000,
+            used_traffic=850, # 85% usage
+        )
+        session.add(admin)
+        await session.flush()
+        
+        # Setup mocks
+        mock_details.return_value = MagicMock()
+        
+        mock_notif_enable.return_value = AsyncMock()
+        mock_notif_enable.return_value.admin.usage_limit_warning = True
+        mock_notif_enable.return_value.admin.usage_limit_warning_percentages = [80]
+        
+        mock_get_admins.return_value = [admin]
+        
+        mock_notification.admin_usage_limit_reached = AsyncMock()
+        
+        # First call: should insert reminder and send notification
+        await _send_usage_limit_warning_notifications(session)
+        assert mock_notification.admin_usage_limit_reached.call_count == 1
+        
+        # Reset mock
+        mock_notification.admin_usage_limit_reached.reset_mock()
+        
+        # Second call: should NOT send notification (since reminder exists)
+        await _send_usage_limit_warning_notifications(session)
+        assert mock_notification.admin_usage_limit_reached.call_count == 0

From b61f6faeb0bfad7e968005047eab581d01756be8 Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Sun, 26 Jul 2026 19:03:10 +0000
Subject: [PATCH 05/11] Do not commit reminders before sending notifications

---
 app/jobs/review_admins.py        | 64 +++++++++++++++++++++-----------
 tests/test_review_admins_unit.py | 55 +++++++++++++++++++++++++++
 2 files changed, 97 insertions(+), 22 deletions(-)

diff --git a/app/jobs/review_admins.py b/app/jobs/review_admins.py
index d4db514c6..f927d8a19 100644
--- a/app/jobs/review_admins.py
+++ b/app/jobs/review_admins.py
@@ -11,13 +11,15 @@
 
 from datetime import UTC, datetime as dt
 
+from sqlalchemy import select
+
 from app import notification, scheduler
 from app.db import GetDB
 from app.db.crud.admin import (
     bulk_create_admin_notification_reminders,
     get_usage_percentage_reached_admins,
 )
-from app.db.models import Admin, ReminderType
+from app.db.models import Admin, AdminNotificationReminder, ReminderType
 from app.models.admin import AdminDetails, AdminRoleData
 from app.models.admin_role import RoleLimits
 from app.models.validators import ListValidator
@@ -70,32 +72,50 @@ async def _send_usage_limit_warning_notifications(db):
     for threshold in default_thresholds:
         threshold_admins = await get_usage_percentage_reached_admins(db, threshold)
 
-        reminder_data = []
-        admin_info_map = {}
+        candidate_admins = [admin for admin in threshold_admins if admin.data_limit and admin.data_limit > 0]
+        if not candidate_admins:
+            continue
+
+        candidate_ids = [admin.id for admin in candidate_admins]
+
+        # Lock the Admin rows to serialize reminder checks/creation for these admins
+        await db.execute(
+            select(Admin.id)
+            .where(Admin.id.in_(candidate_ids))
+            .with_for_update()
+        )
+
+        # Fetch existing reminders for this threshold to avoid duplicate notifications
+        result = await db.execute(
+            select(AdminNotificationReminder.admin_id)
+            .where(
+                AdminNotificationReminder.admin_id.in_(candidate_ids),
+                AdminNotificationReminder.type == ReminderType.data_usage,
+                AdminNotificationReminder.threshold == threshold,
+            )
+        )
+        existing_ids = set(result.scalars().all())
 
-        for admin in threshold_admins:
-            if not admin.data_limit or admin.data_limit <= 0:
+        successfully_sent = []
+        for admin in candidate_admins:
+            if admin.id in existing_ids:
                 continue
+
             usage_percentage = int((admin.used_traffic * 100) / admin.data_limit)
             admin_model = _admin_usage_warning_details(admin)
-            reminder_data.append(
-                {"admin_id": admin.id, "type": ReminderType.data_usage, "threshold": threshold}
-            )
-            admin_info_map[admin.id] = (admin_model, usage_percentage)
-
-        # Bulk-insert all reminders for this threshold in one query
-        inserted_reminders = await bulk_create_admin_notification_reminders(db, reminder_data)
-
-        notifications_to_send = []
-        for item in inserted_reminders:
-            admin_id = item["admin_id"]
-            if admin_id in admin_info_map:
-                admin_model, usage_percentage = admin_info_map[admin_id]
-                notifications_to_send.append((admin_model, usage_percentage, threshold))
 
-        # Send notifications after the bulk insert succeeds
-        for admin_model, usage_percentage, threshold in notifications_to_send:
-            await notification.admin_usage_limit_reached(admin_model, usage_percentage, threshold)
+            try:
+                # Send the notification first
+                await notification.admin_usage_limit_reached(admin_model, usage_percentage, threshold)
+                successfully_sent.append(
+                    {"admin_id": admin.id, "type": ReminderType.data_usage, "threshold": threshold}
+                )
+            except Exception as e:
+                logger.error(f"Failed to send usage limit warning to admin {admin.id}: {e}")
+
+        # Bulk-insert only successfully sent reminders after sending
+        if successfully_sent:
+            await bulk_create_admin_notification_reminders(db, successfully_sent)
 
 
 async def limit_admins_job():
diff --git a/tests/test_review_admins_unit.py b/tests/test_review_admins_unit.py
index cdc1baf78..714ceee80 100644
--- a/tests/test_review_admins_unit.py
+++ b/tests/test_review_admins_unit.py
@@ -85,3 +85,58 @@ async def test_send_usage_limit_warning_notifications_idempotent(
         # Second call: should NOT send notification (since reminder exists)
         await _send_usage_limit_warning_notifications(session)
         assert mock_notification.admin_usage_limit_reached.call_count == 0
+
+@pytest.mark.asyncio
+@patch("app.jobs.review_admins.notification")
+@patch("app.jobs.review_admins.notification_enable")
+@patch("app.jobs.review_admins.get_usage_percentage_reached_admins")
+@patch("app.jobs.review_admins._admin_usage_warning_details")
+async def test_send_usage_limit_warning_notifications_failure_handling(
+    mock_details, mock_get_admins, mock_notif_enable, mock_notification
+):
+    async with TestSession() as session:
+        # Create test admins
+        admin_ok = Admin(
+            username="admin_ok",
+            hashed_password="secret",
+            role_id=3,
+            data_limit=1000,
+            used_traffic=850,
+        )
+        admin_fail = Admin(
+            username="admin_fail",
+            hashed_password="secret",
+            role_id=3,
+            data_limit=1000,
+            used_traffic=850,
+        )
+        session.add_all([admin_ok, admin_fail])
+        await session.flush()
+
+        mock_details.side_effect = lambda admin: MagicMock(id=admin.id)
+
+        mock_notif_enable.return_value = AsyncMock()
+        mock_notif_enable.return_value.admin.usage_limit_warning = True
+        mock_notif_enable.return_value.admin.usage_limit_warning_percentages = [80]
+
+        mock_get_admins.return_value = [admin_ok, admin_fail]
+
+        # Succeed for admin_ok, raise exception for admin_fail
+        async def side_effect(admin_model, usage_percentage, threshold):
+            if admin_model.id == admin_fail.id:
+                raise Exception("Network failure")
+            return
+
+        mock_notification.admin_usage_limit_reached = AsyncMock(side_effect=side_effect)
+
+        # Call notifications
+        await _send_usage_limit_warning_notifications(session)
+
+        # Query DB to verify reminders
+        reminders = (await session.execute(
+            select(AdminNotificationReminder)
+        )).scalars().all()
+
+        admin_ids_with_reminders = {r.admin_id for r in reminders}
+        assert admin_ok.id in admin_ids_with_reminders
+        assert admin_fail.id not in admin_ids_with_reminders

From 8e8c5f309e73094589aa810bba2e461228da4094 Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Sun, 26 Jul 2026 19:05:22 +0000
Subject: [PATCH 06/11] fix ruff

---
 tests/test_review_admins_unit.py | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/tests/test_review_admins_unit.py b/tests/test_review_admins_unit.py
index 714ceee80..a8fcef253 100644
--- a/tests/test_review_admins_unit.py
+++ b/tests/test_review_admins_unit.py
@@ -1,4 +1,5 @@
 import pytest
+from uuid import uuid4
 from sqlalchemy import select
 from app.db.models import Admin, AdminNotificationReminder, ReminderType
 from app.db.crud.admin import bulk_create_admin_notification_reminders
@@ -10,7 +11,7 @@
 async def test_bulk_create_admin_notification_reminders_idempotency():
     async with TestSession() as session:
         # Create an admin to associate reminders with
-        admin = Admin(username="idempotent_admin", hashed_password="secret", role_id=3)
+        admin = Admin(username=f"idempotent_{uuid4().hex[:8]}", hashed_password="secret", role_id=3)
         session.add(admin)
         await session.flush()
         
@@ -55,7 +56,7 @@ async def test_send_usage_limit_warning_notifications_idempotent(
     async with TestSession() as session:
         # Create test admin
         admin = Admin(
-            username="notif_admin",
+            username=f"notif_{uuid4().hex[:8]}",
             hashed_password="secret",
             role_id=3,
             data_limit=1000,
@@ -97,14 +98,14 @@ async def test_send_usage_limit_warning_notifications_failure_handling(
     async with TestSession() as session:
         # Create test admins
         admin_ok = Admin(
-            username="admin_ok",
+            username=f"ok_{uuid4().hex[:8]}",
             hashed_password="secret",
             role_id=3,
             data_limit=1000,
             used_traffic=850,
         )
         admin_fail = Admin(
-            username="admin_fail",
+            username=f"fail_{uuid4().hex[:8]}",
             hashed_password="secret",
             role_id=3,
             data_limit=1000,
@@ -124,7 +125,7 @@ async def test_send_usage_limit_warning_notifications_failure_handling(
         # Succeed for admin_ok, raise exception for admin_fail
         async def side_effect(admin_model, usage_percentage, threshold):
             if admin_model.id == admin_fail.id:
-                raise Exception("Network failure")
+                raise RuntimeError("Network failure")
             return
 
         mock_notification.admin_usage_limit_reached = AsyncMock(side_effect=side_effect)

From a3872d0c318a610c980ccdb7bf9ab023209abe92 Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Sun, 26 Jul 2026 19:11:28 +0000
Subject: [PATCH 07/11] fix tests

---
 tests/test_review_admins_unit.py | 23 +++++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)

diff --git a/tests/test_review_admins_unit.py b/tests/test_review_admins_unit.py
index a8fcef253..00b22b179 100644
--- a/tests/test_review_admins_unit.py
+++ b/tests/test_review_admins_unit.py
@@ -1,12 +1,31 @@
 import pytest
 from uuid import uuid4
 from sqlalchemy import select
-from app.db.models import Admin, AdminNotificationReminder, ReminderType
+from app.db.models import Admin, AdminNotificationReminder, AdminRole, ReminderType
 from app.db.crud.admin import bulk_create_admin_notification_reminders
 from app.jobs.review_admins import _send_usage_limit_warning_notifications
-from tests.api import TestSession
+from tests.api import TestSession, engine
 from unittest.mock import AsyncMock, MagicMock, patch
 
+@pytest.fixture(autouse=True)
+async def setup_database():
+    from app.db import base
+    async with engine.begin() as conn:
+        await conn.run_sync(base.Base.metadata.create_all)
+        
+    async with TestSession() as session:
+        # Seed default roles if they do not exist (to satisfy FK constraints)
+        existing_roles = (await session.execute(select(AdminRole))).scalars().all()
+        if not existing_roles:
+            session.add_all(
+                [
+                    AdminRole(id=1, name="owner", is_owner=True, permissions={}, limits={}, features={}, access={}),
+                    AdminRole(id=2, name="administrator", is_owner=False, permissions={}, limits={}, features={}, access={}),
+                    AdminRole(id=3, name="operator", is_owner=False, permissions={}, limits={}, features={}, access={}),
+                ]
+            )
+            await session.commit()
+
 @pytest.mark.asyncio
 async def test_bulk_create_admin_notification_reminders_idempotency():
     async with TestSession() as session:

From 228a452fd7a7db57d76f21dbff2a1178abad072c Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Sun, 26 Jul 2026 19:17:26 +0000
Subject: [PATCH 08/11] fix

---
 tests/test_review_admins_unit.py | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/tests/test_review_admins_unit.py b/tests/test_review_admins_unit.py
index 00b22b179..02d74dd5b 100644
--- a/tests/test_review_admins_unit.py
+++ b/tests/test_review_admins_unit.py
@@ -17,13 +17,13 @@ async def setup_database():
         # Seed default roles if they do not exist (to satisfy FK constraints)
         existing_roles = (await session.execute(select(AdminRole))).scalars().all()
         if not existing_roles:
-            session.add_all(
-                [
-                    AdminRole(id=1, name="owner", is_owner=True, permissions={}, limits={}, features={}, access={}),
-                    AdminRole(id=2, name="administrator", is_owner=False, permissions={}, limits={}, features={}, access={}),
-                    AdminRole(id=3, name="operator", is_owner=False, permissions={}, limits={}, features={}, access={}),
-                ]
-            )
+            owner = AdminRole(name="owner", is_owner=True, permissions={}, limits={}, features={}, access={})
+            owner.id = 1
+            administrator = AdminRole(name="administrator", is_owner=False, permissions={}, limits={}, features={}, access={})
+            administrator.id = 2
+            operator = AdminRole(name="operator", is_owner=False, permissions={}, limits={}, features={}, access={})
+            operator.id = 3
+            session.add_all([owner, administrator, operator])
             await session.commit()
 
 @pytest.mark.asyncio

From 3a91c4f2829ecd9d3cd3e48267de0058c32cd79d Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Sun, 26 Jul 2026 19:24:34 +0000
Subject: [PATCH 09/11] fix mysql error

---
 tests/test_review_admins_unit.py | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git a/tests/test_review_admins_unit.py b/tests/test_review_admins_unit.py
index 02d74dd5b..c31d20be1 100644
--- a/tests/test_review_admins_unit.py
+++ b/tests/test_review_admins_unit.py
@@ -10,8 +10,23 @@
 @pytest.fixture(autouse=True)
 async def setup_database():
     from app.db import base
-    async with engine.begin() as conn:
-        await conn.run_sync(base.Base.metadata.create_all)
+    
+    # MySQL/MariaDB do not allow defaults on JSON columns; strip them temporarily
+    proxy_default = None
+    proxy_column = None
+    is_mysql = engine.dialect.name == "mysql"
+    if is_mysql:
+        users_table = base.Base.metadata.tables["users"]
+        proxy_column = users_table.c.proxy_settings
+        proxy_default = proxy_column.server_default
+        proxy_column.server_default = None
+
+    try:
+        async with engine.begin() as conn:
+            await conn.run_sync(base.Base.metadata.create_all)
+    finally:
+        if is_mysql and proxy_column is not None:
+            proxy_column.server_default = proxy_default
         
     async with TestSession() as session:
         # Seed default roles if they do not exist (to satisfy FK constraints)

From ed56b2981b85379e581904d77ca19654086c2722 Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Mon, 27 Jul 2026 09:54:43 +0000
Subject: [PATCH 10/11] Consider a DB-level unique constraint in addition to
 the application lock

---
 ...55473c1_add_unique_constraint_to_admin_.py | 46 +++++++++++++++++++
 app/db/models.py                              |  5 +-
 2 files changed, 50 insertions(+), 1 deletion(-)
 create mode 100644 app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py

diff --git a/app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py b/app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py
new file mode 100644
index 000000000..0a14498a4
--- /dev/null
+++ b/app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py
@@ -0,0 +1,46 @@
+"""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
+
+
+# revision identifiers, used by Alembic.
+revision = 'fb32155473c1'
+down_revision = '3c1a7e5b9d20'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # 1. 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
+        )
+        """
+    )
+
+    # 2. Add the unique constraint covering (admin_id, type, threshold)
+    with op.batch_alter_table("admin_notification_reminders", schema=None) as batch_op:
+        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"
+        )
diff --git a/app/db/models.py b/app/db/models.py
index 88e433723..ee4fb9471 100644
--- a/app/db/models.py
+++ b/app/db/models.py
@@ -749,7 +749,10 @@ 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))

From d9536787f761a16b74b7e34f6fe9f1f3afa2b179 Mon Sep 17 00:00:00 2001
From: default <imMohammad20000@ProtonMail.com>
Date: Mon, 27 Jul 2026 10:20:49 +0000
Subject: [PATCH 11/11] fix: Make threshold nullable-uniqueness consistent

---
 app/db/crud/admin.py                          |  2 +-
 ...55473c1_add_unique_constraint_to_admin_.py | 22 +++++++++++++++++--
 app/db/models.py                              |  2 +-
 3 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/app/db/crud/admin.py b/app/db/crud/admin.py
index 55e748af9..890eef15b 100644
--- a/app/db/crud/admin.py
+++ b/app/db/crud/admin.py
@@ -583,7 +583,7 @@ 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(
diff --git a/app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py b/app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py
index 0a14498a4..22d60a891 100644
--- a/app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py
+++ b/app/db/migrations/versions/fb32155473c1_add_unique_constraint_to_admin_.py
@@ -6,6 +6,7 @@
 
 """
 from alembic import op
+import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
@@ -16,7 +17,12 @@
 
 
 def upgrade() -> None:
-    # 1. Clean up duplicate rows before creating the constraint
+    # 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
@@ -30,8 +36,14 @@ def upgrade() -> None:
         """
     )
 
-    # 2. Add the unique constraint covering (admin_id, type, threshold)
+    # 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"]
@@ -44,3 +56,9 @@ def downgrade() -> None:
             "uq_admin_notification_reminders_admin_id_type_threshold",
             type_="unique"
         )
+        batch_op.alter_column(
+            "threshold",
+            existing_type=sa.Integer(),
+            nullable=True,
+            server_default=None
+        )
diff --git a/app/db/models.py b/app/db/models.py
index ee4fb9471..dc7581e20 100644
--- a/app/db/models.py
+++ b/app/db/models.py
@@ -756,7 +756,7 @@ class AdminNotificationReminder(Base, CreatedAtUTCMixin):
     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):