Coverage for src/lexigram/auth/storage/cached_user_store.py: 73%
122 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Cached user store implementation for performance optimization"""
3from __future__ import annotations
5import time
6from typing import TYPE_CHECKING, Any, cast
8from lexigram.auth.models.user import User, UserCredentials
9from lexigram.logging import get_logger
11if TYPE_CHECKING:
12 from lexigram.auth.storage.token_store import UserStoreProtocol
13 from lexigram.contracts.events.protocols import PubSubProtocol
14 from lexigram.contracts.infra.cache import CacheBackendProtocol
16logger = get_logger(__name__)
18#: PubSubProtocol topic on which user-cache invalidation events are broadcast.
19CACHE_INVALIDATION_TOPIC = "auth.user.cache.invalidate"
22class CachedUserStore:
23 """User store with multi-layer caching for high-performance user lookups.
25 Implements a two-layer caching strategy:
26 - L1: In-memory cache (ultra-fast, short-lived)
27 - L2: Distributed cache (Redis, etc., longer-lived)
29 This dramatically reduces database load for authentication middleware.
30 """
32 def __init__(
33 self,
34 user_store: UserStoreProtocol,
35 cache_service: CacheBackendProtocol,
36 cache_ttl: int = 300, # 5 minutes
37 memory_cache_ttl: int = 60, # 1 minute in memory
38 *,
39 pubsub: PubSubProtocol | None = None,
40 ):
41 """Initialize cached user store.
43 Args:
44 user_store: Underlying user store implementation.
45 cache_service: Cache service for distributed caching.
46 cache_ttl: TTL for distributed cache in seconds.
47 memory_cache_ttl: TTL for in-memory cache in seconds.
48 pubsub: Optional :class:`~lexigram.contracts.events.protocols.PubSubProtocol`
49 backend used for cross-instance cache invalidation. When
50 provided, :meth:`update_user` and :meth:`delete_user` publish
51 an invalidation event to :data:`CACHE_INVALIDATION_TOPIC`
52 so that all other running instances can evict their L1 caches.
53 Call :meth:`subscribe_to_invalidations` once at startup to
54 respond to those events on *this* instance.
55 """
56 self.user_store = user_store
57 self.cache_service = cache_service
58 self.cache_ttl = cache_ttl
59 self.memory_cache_ttl = memory_cache_ttl
60 self._pubsub: PubSubProtocol | None = pubsub
62 # In-memory L1 cache: {cache_key: {"user": User, "expires_at": timestamp}}
63 self._memory_cache: dict[str, dict[str, Any]] = {}
65 async def subscribe_to_invalidations(self) -> None:
66 """Subscribe to cross-instance cache invalidation events.
68 Should be called once during application startup when a
69 :class:`~lexigram.contracts.events.protocols.PubSubProtocol` backend is
70 configured. Each received event evicts the affected user from
71 the in-process L1 cache so stale data is never served.
73 A no-op if no pubsub backend was provided.
74 """
75 if self._pubsub is None:
76 return
77 await self._pubsub.subscribe(
78 CACHE_INVALIDATION_TOPIC, self._handle_invalidation
79 )
80 logger.info(
81 "cached_user_store.subscribed",
82 topic=CACHE_INVALIDATION_TOPIC,
83 )
85 async def _handle_invalidation(self, data: Any) -> None:
86 """Handle a ``user.cache.invalidate`` event from another instance.
88 Evicts the in-process L1 cache entries for the affected user.
90 Args:
91 data: Event payload. Expected to include a ``user_id``
92 field and optionally ``email`` and ``name`` fields.
93 """
94 if not isinstance(data, dict):
95 return
96 user_id: str | None = data.get("user_id")
97 email: str | None = data.get("email")
98 name: str | None = data.get("name")
100 evicted: list[str] = []
101 for key in (
102 f"user:id:{user_id}" if user_id else None,
103 f"user:email:{email}" if email else None,
104 f"user:name:{name}" if name else None,
105 ):
106 if key and self._memory_cache.pop(key, None) is not None:
107 evicted.append(key)
109 if evicted:
110 logger.debug(
111 "cached_user_store.l1_evicted_remote",
112 keys=evicted,
113 user_id=user_id,
114 )
116 async def get_user_by_id(self, user_id: str) -> User | None:
117 """Get user by ID with multi-layer caching.
119 Returns:
120 User object if found, None otherwise
121 """
122 cache_key = f"user:id:{user_id}"
124 # L1: Check in-memory cache
125 cached = self._memory_cache.get(cache_key)
126 if cached and cached["expires_at"] > time.monotonic():
127 logger.debug("L1 cache HIT for user ID: %s", user_id)
129 return cast("User", cached["user"])
131 # L2: Check distributed cache
132 cached_result = await self.cache_service.get(cache_key)
133 cached_data = (
134 cached_result.unwrap()
135 if hasattr(cached_result, "is_ok") and cached_result.is_ok()
136 else cached_result
137 if cached_result
138 else None
139 )
140 if cached_data:
141 logger.debug("L2 cache HIT for user ID: %s", user_id)
142 user = self._deserialize_user(cached_data) # type: ignore[arg-type]
144 # Store in L1 cache
145 self._memory_cache[cache_key] = {
146 "user": user,
147 "expires_at": time.monotonic() + self.memory_cache_ttl,
148 }
150 return user
152 # Cache MISS: Fetch from database
153 logger.debug("Cache MISS for user ID: %s", user_id)
154 db_user: User | None = await self.user_store.get_user_by_id(user_id)
156 if db_user:
157 # Store in distributed cache (L2)
158 await self.cache_service.set(
159 cache_key,
160 self._serialize_user(db_user),
161 ttl=self.cache_ttl,
162 )
164 # Store in memory cache (L1)
165 self._memory_cache[cache_key] = {
166 "user": db_user,
167 "expires_at": time.monotonic() + self.memory_cache_ttl,
168 }
170 return db_user
172 async def get_user_by_email(self, email: str) -> User | None:
173 """Get user by email with caching.
175 Note: Email lookups are less cacheable than ID lookups since
176 emails can change, but we still cache them for performance.
177 """
178 cache_key = f"user:email:{email}"
180 # L1: Check in-memory cache
181 cached = self._memory_cache.get(cache_key)
182 if cached and cached["expires_at"] > time.monotonic():
183 logger.debug("L1 cache HIT for email: %s", email)
185 return cast("User", cached["user"])
187 # L2: Check distributed cache
188 cached_result = await self.cache_service.get(cache_key)
189 cached_data = (
190 cached_result.unwrap()
191 if hasattr(cached_result, "is_ok") and cached_result.is_ok()
192 else cached_result
193 if cached_result
194 else None
195 )
196 if cached_data:
197 logger.debug("L2 cache HIT for email: %s", email)
198 user = self._deserialize_user(cached_data) # type: ignore[arg-type]
200 # Store in L1 cache
201 self._memory_cache[cache_key] = {
202 "user": user,
203 "expires_at": time.monotonic() + self.memory_cache_ttl,
204 }
206 return user
208 # Cache MISS: Fetch from database
209 logger.debug("Cache MISS for email: %s", email)
210 db_user: User | None = await self.user_store.get_user_by_email(email)
212 if db_user:
213 # Store in both caches
214 await self.cache_service.set(
215 cache_key,
216 self._serialize_user(db_user),
217 ttl=self.cache_ttl,
218 )
220 self._memory_cache[cache_key] = {
221 "user": db_user,
222 "expires_at": time.monotonic() + self.memory_cache_ttl,
223 }
225 return db_user
227 async def update_user(self, user: User) -> None:
228 """Update user and invalidate all related caches."""
229 # Update the underlying store
230 await self.user_store.update_user(user)
232 # Invalidate all caches for this user
233 user_id = user.user_id
234 name = user.name
235 email = user.email
237 cache_keys = [
238 f"user:id:{user_id}",
239 f"user:name:{name}",
240 f"user:email:{email}",
241 ]
243 # Invalidate L1 cache
244 for key in cache_keys:
245 self._memory_cache.pop(key, None)
247 # Invalidate L2 cache
248 await self.cache_service.delete_many(cache_keys)
250 logger.debug("Cache INVALIDATE for user: %s", user_id)
252 # Broadcast cross-instance invalidation event
253 if self._pubsub is not None:
254 await self._pubsub.publish(
255 CACHE_INVALIDATION_TOPIC,
256 {"user_id": user_id, "email": email, "name": name},
257 )
259 async def delete_user(self, user_id: str) -> None:
260 """Delete user and invalidate caches."""
261 # Get user info before deletion for cache invalidation
262 user = await self.get_user_by_id(user_id)
263 if user:
264 # Delete from underlying store
265 await self.user_store.delete_user(user_id)
267 # Invalidate caches
268 cache_keys = [
269 f"user:id:{user_id}",
270 f"user:name:{user.name}",
271 f"user:email:{user.email}",
272 ]
274 for key in cache_keys:
275 self._memory_cache.pop(key, None)
277 await self.cache_service.delete_many(cache_keys)
279 logger.debug("Cache INVALIDATE for deleted user: %s", user_id)
281 # Broadcast cross-instance invalidation event
282 if self._pubsub is not None:
283 await self._pubsub.publish(
284 CACHE_INVALIDATION_TOPIC,
285 {"user_id": user_id, "email": user.email, "name": user.name},
286 )
287 else:
288 # User not found, just delete from underlying store
289 await self.user_store.delete_user(user_id)
291 async def create_user(
292 self,
293 name: str,
294 email: str,
295 hashed_password: str | None,
296 roles: list[str] | None = None,
297 permissions: list[str] | None = None,
298 profile: dict[str, Any] | None = None,
299 **kwargs: Any,
300 ) -> User:
301 """Create user (no caching needed for new users)."""
302 return await self.user_store.create_user(
303 name=name,
304 email=email,
305 hashed_password=hashed_password,
306 roles=roles,
307 permissions=permissions,
308 profile=profile,
309 **kwargs,
310 )
312 async def list_users(self, skip: int = 0, limit: int = 100) -> list[User]:
313 """List users (not cached for simplicity)."""
314 return await self.user_store.list_users(skip, limit)
316 async def count_users(self) -> int:
317 """Count users (not cached for simplicity)."""
318 return await self.user_store.count_users()
320 async def get_credentials(self, user_id: str) -> UserCredentials | None:
321 """Return credential data from the underlying store.
323 Credentials are never cached — they are always read from the
324 primary store to ensure the latest password hash is used.
325 """
326 return await self.user_store.get_credentials(user_id)
328 async def update_credentials(self, creds: UserCredentials) -> None:
329 """Persist updated credentials and invalidate the user cache."""
330 await self.user_store.update_credentials(creds)
331 # Invalidate the cached user entry so the next read reflects any
332 # side-effects the underlying store may have applied.
333 cache_key = f"user:id:{creds.user_id}"
334 self._memory_cache.pop(cache_key, None)
335 await self.cache_service.delete_many([cache_key])
337 def _serialize_user(self, user: User) -> dict[str, Any]:
338 """Serialize user for caching (credential fields excluded)."""
339 return {
340 "id": user.user_id,
341 "name": user.name,
342 "email": user.email,
343 "is_active": user.is_active,
344 "is_verified": user.is_verified,
345 "roles": user.roles,
346 "permissions": user.permissions,
347 "profile": user.profile,
348 "created_at": user.created_at.isoformat() if user.created_at else None,
349 "updated_at": user.updated_at.isoformat() if user.updated_at else None,
350 "last_login_at": user.last_login_at.isoformat()
351 if user.last_login_at
352 else None,
353 "login_count": user.login_count,
354 }
356 def _deserialize_user(self, data: dict[str, Any]) -> User:
357 """Deserialize user from cache (credential fields excluded)."""
358 from datetime import datetime
360 def parse_dt(val: Any) -> datetime | None:
361 if not val:
362 return None
363 if isinstance(val, datetime):
364 return val
365 try:
366 return datetime.fromisoformat(str(val))
367 except (ValueError, TypeError):
368 return None
370 return User(
371 user_id=data["id"],
372 name=data.get("name") or "",
373 email=data["email"],
374 is_active=data.get("is_active", True),
375 is_verified=data.get("is_verified", False),
376 roles=data.get("roles", []),
377 permissions=data.get("permissions", []),
378 profile=data.get("profile", {}),
379 created_at=parse_dt(data.get("created_at")),
380 updated_at=parse_dt(data.get("updated_at")),
381 last_login_at=parse_dt(data.get("last_login_at")),
382 login_count=data.get("login_count", 0),
383 )
386__all__ = ["CACHE_INVALIDATION_TOPIC", "CachedUserStore"]