refactor(redis): strengthen shared cache invalidation

2026-05-21 06:26:48 +03:00
parent 3405d12696
commit 044cb4de7e
10 changed files with 237 additions and 61 deletions
@@ -1,6 +1,7 @@
 # ruff: noqa: F401,F403,F405,I001
 from ._runtime import *  # noqa: F403,F405
 from .auth import _hash_email_password
+from .common import _invalidate_webapp_user_caches


 async def account_email_request_route(request: web.Request) -> web.Response:
@@ -174,6 +175,7 @@ async def account_email_verify_route(request: web.Request) -> web.Response:
            logger.exception("Email account link failed")
            return _json_error(500, "link_failed", "Link failed")

+    await _invalidate_webapp_user_caches(settings, user_id, final_user_id, include_devices=True)
    if should_notify_email_linked:
        try:
            from bot.services.notification_service import NotificationService
@@ -276,7 +278,7 @@ async def account_password_confirm_route(request: web.Request) -> web.Response:
            logger.exception("Email password setup failed")
            return _json_error(500, "password_setup_failed", "Password setup failed")

-    await cache_delete(settings, redis_key(settings, "cache", "webapp", "me", user_id))
+    await _invalidate_webapp_user_caches(settings, user_id)
    return web.json_response({"ok": True, "password_auth_enabled": True})


@@ -396,6 +398,7 @@ async def account_telegram_link_route(request: web.Request) -> web.Response:
            logger.exception("Telegram account link failed")
            return _json_error(500, "link_failed", "Link failed")

+    await _invalidate_webapp_user_caches(settings, user_id, final_user_id, include_devices=True)
    if should_notify_telegram_linked and final_telegram_id:
        try:
            from bot.services.notification_service import NotificationService
@@ -488,6 +491,7 @@ async def account_language_route(request: web.Request) -> web.Response:
            await session.flush()
        await session.commit()

+    await _invalidate_webapp_user_caches(settings, user_id)
    return web.json_response({"ok": True, "language": language})


@@ -820,25 +820,30 @@ async def _enforce_webapp_rate_limit(
        or "unknown"
    )
    key = f"{action}:{ip_address}:{int(user_id)}"
-    redis = await get_redis(settings)
-    if redis is not None:
-        redis_rate_key = redis_key(settings, "rate-limit", "webapp", key)
-        current = await redis.incr(redis_rate_key)
-        if current == 1:
-            await redis.expire(redis_rate_key, settings.WEBAPP_RATE_LIMIT_TTL_SECONDS)
-        if current > settings.WEBAPP_RATE_LIMIT_MAX_REQUESTS:
-            ttl = await redis.ttl(redis_rate_key)
-            retry_after = max(1, int(ttl if ttl and ttl > 0 else WEBAPP_RATE_LIMIT_WINDOW_SECONDS))
-            return web.json_response(
-                {
-                    "ok": False,
-                    "error": "rate_limited",
-                    "retry_after": retry_after,
-                },
-                status=429,
-                headers={"Retry-After": str(retry_after)},
-            )
-        return None
+    try:
+        redis = await get_redis(settings)
+        if redis is not None:
+            redis_rate_key = redis_key(settings, "rate-limit", "webapp", key)
+            current = await redis.incr(redis_rate_key)
+            if current == 1:
+                await redis.expire(redis_rate_key, settings.WEBAPP_RATE_LIMIT_TTL_SECONDS)
+            if current > settings.WEBAPP_RATE_LIMIT_MAX_REQUESTS:
+                ttl = await redis.ttl(redis_rate_key)
+                retry_after = max(
+                    1, int(ttl if ttl and ttl > 0 else WEBAPP_RATE_LIMIT_WINDOW_SECONDS)
+                )
+                return web.json_response(
+                    {
+                        "ok": False,
+                        "error": "rate_limited",
+                        "retry_after": retry_after,
+                    },
+                    status=429,
+                    headers={"Retry-After": str(retry_after)},
+                )
+            return None
+    except Exception as exc:
+        logger.warning("Redis webapp rate limiter unavailable; using local fallback: %s", exc)

    buckets: Dict[str, deque[float]] = request.app["webapp_rate_limit_buckets"]
    lock: asyncio.Lock = request.app["webapp_rate_limit_lock"]
@@ -1,5 +1,6 @@
 # ruff: noqa: F401,F403,F405,I001
 from ._runtime import *  # noqa: F403,F405
+from .common import _invalidate_webapp_user_caches


 def _resolve_telegram_bot_id(bot_token: str) -> Optional[int]:
@@ -386,6 +387,7 @@ async def telegram_oauth_callback_route(request: web.Request) -> web.Response:
            logger.exception("Telegram OAuth callback failed")
            raise redirect(redirect_path, "failed")

+    await _invalidate_webapp_user_caches(settings, final_user_id, include_devices=True)
    token = create_webapp_session_token(settings, int(final_user_id))
    response = web.HTTPFound(_telegram_oauth_redirect_url(redirect_path, status="success"))
    _clear_telegram_oauth_state_cookie(response)
@@ -480,6 +482,7 @@ async def auth_token_route(request: web.Request) -> web.Response:
            logger.exception("WebApp auth failed")
            return _json_error(500, "auth_failed", "Auth failed")

+    await _invalidate_webapp_user_caches(settings, authenticated_user_id, include_devices=True)
    token = create_webapp_session_token(settings, int(authenticated_user_id))
    return _build_webapp_auth_response(settings, {"ok": True}, token=token)

@@ -694,6 +697,7 @@ async def email_auth_verify_route(request: web.Request) -> web.Response:
            logger.exception("Email WebApp auth failed")
            return _json_error(500, "auth_failed", "Auth failed")

+    await _invalidate_webapp_user_caches(settings, int(db_user.user_id), include_devices=True)
    if created_user:
        try:
            from bot.services.notification_service import NotificationService
@@ -801,6 +805,7 @@ async def email_auth_magic_route(request: web.Request) -> web.Response:
            logger.exception("Email magic-link auth failed")
            return _json_error(500, "auth_failed", "Auth failed")

+    await _invalidate_webapp_user_caches(settings, int(db_user.user_id), include_devices=True)
    if created_user and verified_email:
        try:
            from bot.services.notification_service import NotificationService
@@ -17,6 +17,30 @@ def _json_error(status: int, code: str, message: str) -> web.Response:
    )


+async def _invalidate_webapp_user_caches(
+    settings: Settings,
+    *user_ids: Optional[int],
+    include_devices: bool = False,
+) -> None:
+    keys: List[str] = []
+    seen: set[int] = set()
+    for raw_user_id in user_ids:
+        if raw_user_id is None:
+            continue
+        try:
+            user_id = int(raw_user_id)
+        except (TypeError, ValueError):
+            continue
+        if user_id in seen:
+            continue
+        seen.add(user_id)
+        keys.append(redis_key(settings, "cache", "webapp", "me", user_id))
+        if include_devices:
+            keys.append(redis_key(settings, "cache", "webapp", "devices", user_id))
+    if keys:
+        await cache_delete(settings, *keys)
+
+
 def _validation_error_response(exc: ValidationError) -> web.Response:
    for error in exc.errors():
        loc = error.get("loc") or ()
@@ -51,7 +51,11 @@ async def cache_get_json(settings: Settings, key: str) -> Any:
    redis = await get_redis(settings)
    if redis is None:
        return None
-    raw = await redis.get(key)
+    try:
+        raw = await redis.get(key)
+    except Exception as exc:
+        logger.warning("Redis cache get failed for key %s: %s", key, exc)
+        return None
    if raw is None:
        return None
    try:
@@ -65,14 +69,39 @@ async def cache_set_json(settings: Settings, key: str, value: Any, ttl_seconds:
    redis = await get_redis(settings)
    if redis is None:
        return
-    await redis.set(key, json.dumps(value, ensure_ascii=False, default=str), ex=ttl_seconds)
+    try:
+        await redis.set(key, json.dumps(value, ensure_ascii=False, default=str), ex=ttl_seconds)
+    except Exception as exc:
+        logger.warning("Redis cache set failed for key %s: %s", key, exc)


 async def cache_delete(settings: Settings, *keys: str) -> None:
    redis = await get_redis(settings)
    if redis is None or not keys:
        return
-    await redis.delete(*keys)
+    try:
+        await redis.delete(*keys)
+    except Exception as exc:
+        logger.warning("Redis cache delete failed for %s key(s): %s", len(keys), exc)
+
+
+async def cache_delete_pattern(settings: Settings, pattern: str) -> int:
+    redis = await get_redis(settings)
+    if redis is None or not pattern:
+        return 0
+    deleted = 0
+    batch = []
+    try:
+        async for key in redis.scan_iter(match=pattern, count=100):
+            batch.append(key)
+            if len(batch) >= 100:
+                deleted += int(await redis.delete(*batch))
+                batch.clear()
+        if batch:
+            deleted += int(await redis.delete(*batch))
+    except Exception as exc:
+        logger.warning("Redis cache pattern delete failed for %s: %s", pattern, exc)
+    return deleted


@asynccontextmanager
@@ -24,28 +24,36 @@ async def enqueue_webhook_event(
    if redis is None:
        return False

-    dedupe_id = event_id or payload.get("id") or payload.get("event_id")
-    if dedupe_id:
-        dedupe_key = redis_key(settings, "webhook", "seen", provider, dedupe_id)
-        if not await redis.set(dedupe_key, "1", nx=True, ex=24 * 60 * 60):
-            logger.info("Skipping duplicate %s webhook event %s", provider, dedupe_id)
-            return True
+    try:
+        dedupe_id = event_id or payload.get("id") or payload.get("event_id")
+        if dedupe_id:
+            dedupe_key = redis_key(settings, "webhook", "seen", provider, dedupe_id)
+            if not await redis.set(dedupe_key, "1", nx=True, ex=24 * 60 * 60):
+                logger.info("Skipping duplicate %s webhook event %s", provider, dedupe_id)
+                return True

-    message = {
-        "provider": provider,
-        "event_id": dedupe_id,
-        "payload": payload,
-        "enqueued_at": time.time(),
-    }
-    await redis.lpush(webhook_queue_key(settings), json.dumps(message, ensure_ascii=False))
-    return True
+        message = {
+            "provider": provider,
+            "event_id": dedupe_id,
+            "payload": payload,
+            "enqueued_at": time.time(),
+        }
+        await redis.lpush(webhook_queue_key(settings), json.dumps(message, ensure_ascii=False))
+        return True
+    except Exception as exc:
+        logger.warning("Redis webhook enqueue failed for %s: %s", provider, exc)
+        return False


 async def pop_webhook_event(settings: Settings, timeout_seconds: int = 5) -> Optional[dict]:
    redis = await get_redis(settings)
    if redis is None:
        return None
-    item = await redis.brpop(webhook_queue_key(settings), timeout=timeout_seconds)
+    try:
+        item = await redis.brpop(webhook_queue_key(settings), timeout=timeout_seconds)
+    except Exception as exc:
+        logger.warning("Redis webhook pop failed: %s", exc)
+        return None
    if not item:
        return None
    _, raw = item
@@ -60,4 +68,8 @@ async def webhook_queue_depth(settings: Settings) -> int:
    redis = await get_redis(settings)
    if redis is None:
        return 0
-    return int(await redis.llen(webhook_queue_key(settings)))
+    try:
+        return int(await redis.llen(webhook_queue_key(settings)))
+    except Exception as exc:
+        logger.warning("Redis webhook queue depth failed: %s", exc)
+        return 0
@@ -496,7 +496,7 @@ class PanelApiService:
            "POST", "/users", json=payload, log_full_response=log_response
        )
        if response and not response.get("error") and "response" in response:
-            self._invalidate_all_users_cache()
+            await self._invalidate_all_users_cache()
            logging.info(
                f"Panel user '{username_on_panel}' created successfully (UUID: {response.get('response', {}).get('uuid')})."  # noqa: E501
            )
@@ -518,8 +518,8 @@ class PanelApiService:
        )
        if full_response and not full_response.get("error") and "response" in full_response:
            logging.debug("User %s details updated on panel.", user_uuid)
-            self._invalidate_user_cache(user_uuid)
-            self._invalidate_all_users_cache()
+            await self._invalidate_user_cache(user_uuid)
+            await self._invalidate_all_users_cache()
            return full_response.get("response")

        logging.error(
@@ -535,8 +535,8 @@ class PanelApiService:
        response_data = await self._request("POST", endpoint, log_full_response=log_response)

        if response_data and not response_data.get("error") and "response" in response_data:
-            self._invalidate_user_cache(user_uuid)
-            self._invalidate_all_users_cache()
+            await self._invalidate_user_cache(user_uuid)
+            await self._invalidate_all_users_cache()
            actual_status = response_data.get("response", {}).get("status")
            expected_status = "ACTIVE" if enable else "DISABLED"
            if actual_status == expected_status:
@@ -573,17 +573,17 @@ class PanelApiService:
                logging.info(
                    f"Panel user {user_uuid} already absent (errorCode {error_code}). Treating as deleted."  # noqa: E501
                )
-                self._invalidate_user_cache(user_uuid)
-                self._invalidate_devices_cache(user_uuid)
-                self._invalidate_all_users_cache()
+                await self._invalidate_user_cache(user_uuid)
+                await self._invalidate_devices_cache(user_uuid)
+                await self._invalidate_all_users_cache()
                return True
            logging.error(f"Failed to delete user {user_uuid} on panel. Response: {response_data}")
            return False

        logging.info(f"Panel user {user_uuid} deleted successfully.")
-        self._invalidate_user_cache(user_uuid)
-        self._invalidate_devices_cache(user_uuid)
-        self._invalidate_all_users_cache()
+        await self._invalidate_user_cache(user_uuid)
+        await self._invalidate_devices_cache(user_uuid)
+        await self._invalidate_all_users_cache()
        return True

    async def get_subscription_link(
@@ -618,7 +618,7 @@ class PanelApiService:
        payload = {"userUuid": user_uuid, "hwid": hwid}
        response_data = await self._request("POST", endpoint, json=payload, log_full_response=False)
        if response_data and not response_data.get("error") and "response" in response_data:
-            self._invalidate_devices_cache(user_uuid)
+            await self._invalidate_devices_cache(user_uuid)
            return True
        logging.error(
            f"Failed to disconnect device {hwid} for user {user_uuid}. Payload: {payload}, Response: {response_data}"  # noqa: E501
@@ -720,21 +720,21 @@ class PanelApiService:
        )
        return None

-    def _invalidate_squad_caches(self) -> None:
-        self._squads_cache.invalidate()
+    async def _invalidate_squad_caches(self) -> None:
+        await self._squads_cache.invalidate_remote()

-    def _invalidate_user_cache(self, user_uuid: Optional[str]) -> None:
+    async def _invalidate_user_cache(self, user_uuid: Optional[str]) -> None:
        if not user_uuid:
            return
-        self._users_cache.invalidate(f"uuid:{user_uuid}")
+        await self._users_cache.invalidate_remote(f"uuid:{user_uuid}")

-    def _invalidate_all_users_cache(self) -> None:
-        self._all_users_cache.invalidate()
+    async def _invalidate_all_users_cache(self) -> None:
+        await self._all_users_cache.invalidate_remote()

-    def _invalidate_devices_cache(self, user_uuid: Optional[str]) -> None:
+    async def _invalidate_devices_cache(self, user_uuid: Optional[str]) -> None:
        if not user_uuid:
            return
-        self._devices_cache.invalidate(f"user:{user_uuid}")
+        await self._devices_cache.invalidate_remote(f"user:{user_uuid}")

    async def get_internal_squads(self) -> Optional[List[Dict[str, Any]]]:
        return await self._squads_cache.get_or_load("list", self._get_internal_squads_uncached)
@@ -835,6 +835,8 @@ class PanelApiService:
        endpoint = f"/users/{user_uuid}/actions/reset-traffic"
        response_data = await self._request("POST", endpoint, log_full_response=False)
        if response_data and not response_data.get("error"):
+            await self._invalidate_user_cache(user_uuid)
+            await self._invalidate_all_users_cache()
            return True
        logging.error("Failed to reset traffic for user %s. Response: %s", user_uuid, response_data)
        return False
@@ -848,7 +850,10 @@ class PanelApiService:
            log_full_response=False,
        )
        if response_data and not response_data.get("error"):
-            self._invalidate_squad_caches()
+            await self._invalidate_squad_caches()
+            for user_uuid in user_uuids:
+                await self._invalidate_user_cache(user_uuid)
+            await self._invalidate_all_users_cache()
            return True
        logging.error("Failed to add users to squad %s. Response: %s", squad_uuid, response_data)
        return False
@@ -864,7 +869,10 @@ class PanelApiService:
            log_full_response=False,
        )
        if response_data and not response_data.get("error"):
-            self._invalidate_squad_caches()
+            await self._invalidate_squad_caches()
+            for user_uuid in user_uuids:
+                await self._invalidate_user_cache(user_uuid)
+            await self._invalidate_all_users_cache()
            return True
        logging.error(
            "Failed to remove users from squad %s. Response: %s", squad_uuid, response_data
@@ -83,3 +83,20 @@ class AsyncTTLCache:
            self._data.clear()
            return
        self._data.pop(key, None)
+
+    async def invalidate_remote(self, key: Optional[str] = None) -> None:
+        self.invalidate(key)
+        if self.settings is None or not self.namespace:
+            return
+        try:
+            from bot.infra.redis import cache_delete, cache_delete_pattern, redis_key
+
+            if key is None:
+                pattern = redis_key(self.settings, "cache", self.namespace, "*")
+                await cache_delete_pattern(self.settings, pattern)
+                return
+            await cache_delete(
+                self.settings, redis_key(self.settings, "cache", self.namespace, key)
+            )
+        except Exception:
+            return