limit by UID and server (#1252)

BOT/utils.py

@@ -17,7 +17,7 @@ import time
 import unicodedata
 from datetime import datetime, time as dt_time, timedelta, timezone
 from pathlib import Path
-from typing import Any, Dict, List, Literal, Optional, TypedDict
+from typing import Any, Dict, Iterable, List, Literal, Optional, TypedDict
 
 # Third-Party Library Imports
 import aiofiles
@@ -223,11 +223,12 @@ def higher_tier(a: Optional[str], b: Optional[str]) -> Optional[str]:
         return None
     return a if ra >= rb else b
 # Free-tier /comp caps per timeslot.
-# Server-wide cap counts every invocation in the guild during the window.
-# Per-user cap counts each user's invocations and is enforced in addition to
-# the server cap, so one user maxing out can't drain the rest of the server.
+# Server-wide cap counts every invocation in a non-premium guild during the
+# window. Per-user cap counts each user's invocations across ALL non-premium
+# guilds in the window — premium-guild usage never counts toward either cap,
+# so subscribers (and their members) bypass both checks entirely.
 COMP_LIMIT_PER_TIMESLOT: int = 25
-COMP_LIMIT_PER_USER_PER_TIMESLOT: int = 10
+COMP_LIMIT_PER_USER_PER_TIMESLOT: int = 15
 
 # ── SQB schedule (UTC, DST-immune) ───────────────────────────────────────────
 # Edit SQB_SLOTS_POSTED and the margin constants when Gaijin changes the
@@ -1085,17 +1086,26 @@ async def get_comp_usage_in_timeslot(guild_id: int, since_ts: int) -> int:
 
 
 async def get_comp_usage_in_timeslot_by_user(
-    guild_id: int, user_id: int, since_ts: int
+    user_id: int, since_ts: int, *, exclude_guild_ids: Iterable[int] = ()
 ) -> int:
-    """Count /comp invocations for a specific user in a guild since a timestamp."""
+    """Count a user's /comp invocations since `since_ts`, excluding any guild
+    in `exclude_guild_ids` (used to drop premium/entitled guilds so their use
+    doesn't count toward the free per-user cap).
+    """
     try:
+        excluded = [str(g) for g in exclude_guild_ids]
+        sql = (
+            "SELECT COUNT(*) FROM command_usage "
+            "WHERE command_name='comp' AND user_id=? AND timestamp >= ?"
+        )
+        params: list[Any] = [str(user_id), since_ts]
+        if excluded:
+            placeholders = ",".join("?" for _ in excluded)
+            sql += f" AND (guild_id IS NULL OR guild_id NOT IN ({placeholders}))"
+            params.extend(excluded)
         async with aiosqlite.connect(COMMAND_DATA_DB_PATH, timeout=5.0) as db:
             await db.execute("PRAGMA busy_timeout=5000;")
-            cur = await db.execute(
-                "SELECT COUNT(*) FROM command_usage "
-                "WHERE command_name='comp' AND guild_id=? AND user_id=? AND timestamp >= ?",
-                (str(guild_id), str(user_id), since_ts),
-            )
+            cur = await db.execute(sql, params)
             row = await cur.fetchone()
             return row[0] if row else 0
     except Exception:
@@ -1103,6 +1113,11 @@ async def get_comp_usage_in_timeslot_by_user(
         return 0
 
 
+def get_entitled_guild_ids() -> list[int]:
+    """Snapshot of currently-entitled (premium) guild IDs from the cache."""
+    return list(_entitled_tiers.keys())
+
+
 def minutes_ago(unix_timestamp: int) -> str:
     """Convert a unix timestamp to a human-readable 'X minutes ago' string."""
     if not unix_timestamp: