Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-feedback/src/lexigram/ai/feedback/storage/cache.py: 33%

1"""Write-through cache feedback store. 

2 

3Wraps a durable :class:`~lexigram.ai.feedback.storage.protocols.FeedbackStoreProtocol` 

4with a :class:`~lexigram.contracts.cache.CacheBackendProtocol` for fast reads. 

5All writes go to both the backing store and the cache in a write-through 

6fashion; cache entries are invalidated on any mutation. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING 

12 

13from lexigram.logging import ( 

14 get_logger, 

15) 

16 

17if TYPE_CHECKING: 

18 from lexigram.ai.feedback.storage.protocols import ( 

19 FeedbackStoreProtocol, 

20 FeedbackSummary, 

21 ) 

22 from lexigram.ai.feedback.types import FeedbackItem, FeedbackType 

23 from lexigram.contracts.infra.cache import CacheBackendProtocol 

24 from lexigram.result import Result 

25 

26logger = get_logger(__name__) 

27 

28_SESSION_TTL = 300 # 5 minutes 

29_TYPE_TTL = 60 # 1 minute 

30 

31 

32def _session_key(owner_id: str, session_id: str) -> str: 

33 return f"feedback:session:{owner_id}:{session_id}" 

34 

35 

36def _type_key(owner_id: str, feedback_type: FeedbackType) -> str: 

37 return f"feedback:type:{owner_id}:{feedback_type.value}" 

38 

39 

40class CachedFeedbackStore: 

41 """Write-through cache: cold reads from *store*, hot reads from *cache*. 

42 

43 Session-scoped and type-scoped result lists are cached with short TTLs. 

44 Any :meth:`save` call invalidates the relevant cache entries so subsequent 

45 reads always reflect the latest data. 

46 

47 Args: 

48 store: Durable backing store (e.g. :class:`~lexigram.ai.feedback.storage.database.DatabaseFeedbackStore`). 

49 cache: Cache backend used for hot reads. 

50 """ 

51 

52 def __init__( 

53 self, 

54 store: FeedbackStoreProtocol, 

55 cache: CacheBackendProtocol, 

56 ) -> None: 

57 self._store = store 

58 self._cache = cache 

59 

60 async def save(self, feedback: FeedbackItem) -> Result[str, Exception]: 

61 """Persist *feedback* and invalidate related cache entries. 

62 

63 Args: 

64 feedback: Item to persist. 

65 

66 Returns: 

67 Result from the backing store. 

68 """ 

69 result = await self._store.save(feedback) 

70 if result.is_ok(): 

71 session_id: str | None = feedback.context.get("session_id") 

72 if session_id: 

73 await self._cache.delete(_session_key(feedback.owner_id, session_id)) 

74 await self._cache.delete( 

75 _type_key(feedback.owner_id, feedback.feedback_type) 

76 ) 

77 logger.debug( 

78 "feedback_cache_invalidated", 

79 feedback_id=feedback.id, 

80 owner_id=feedback.owner_id, 

81 session_id=session_id, 

82 ) 

83 return result 

84 

85 async def find_by_session( 

86 self, session_id: str, *, owner_id: str 

87 ) -> list[FeedbackItem]: 

88 """Return items for *session_id* owned by *owner_id*, using cache when available. 

89 

90 Args: 

91 session_id: Session identifier. 

92 owner_id: Owner scope; only this owner's items are returned. 

93 

94 Returns: 

95 Feedback items for the session, newest first. 

96 """ 

97 key = _session_key(owner_id, session_id) 

98 cached = await self._cache.get(key) 

99 if cached is not None: 

100 logger.debug("feedback_cache_hit", key=key) 

101 return cached # type: ignore[return-value] 

102 

103 items = await self._store.find_by_session(session_id, owner_id=owner_id) 

104 await self._cache.set(key, items, ttl=_SESSION_TTL) 

105 return items 

106 

107 async def find_by_type( 

108 self, 

109 feedback_type: FeedbackType, 

110 *, 

111 owner_id: str, 

112 limit: int = 100, 

113 ) -> list[FeedbackItem]: 

114 """Return items of *feedback_type* owned by *owner_id*, using cache when available. 

115 

116 Args: 

117 feedback_type: Type to filter by. 

118 owner_id: Owner scope; only this owner's items are returned. 

119 limit: Maximum result count. 

120 

121 Returns: 

122 Matching feedback items, newest first. 

123 """ 

124 key = _type_key(owner_id, feedback_type) 

125 cached = await self._cache.get(key) 

126 if cached is not None: 

127 logger.debug("feedback_cache_hit", key=key) 

128 items: list[FeedbackItem] = cached # type: ignore[assignment] 

129 return items[:limit] 

130 

131 items = await self._store.find_by_type( 

132 feedback_type, owner_id=owner_id, limit=limit 

133 ) 

134 await self._cache.set(key, items, ttl=_TYPE_TTL) 

135 return items 

136 

137 async def aggregate( 

138 self, *, owner_id: str, window_hours: int = 24 

139 ) -> FeedbackSummary: 

140 """Delegate aggregation to the backing store — not cached. 

141 

142 Args: 

143 owner_id: Owner scope; only this owner's items are aggregated. 

144 window_hours: Look-back window in hours. 

145 

146 Returns: 

147 :class:`~lexigram.ai.feedback.storage.protocols.FeedbackSummary`. 

148 """ 

149 return await self._store.aggregate(owner_id=owner_id, window_hours=window_hours) 

150 

151 

152__all__ = ["CachedFeedbackStore"]