Coverage for src/lexigram/admin/realtime/subject_hub.py: 0%

1"""Subject-backed admin event hub. 

2 

3The hub the legacy SSE event stream used (``AdminEventHub`` in 

4``realtime/sse.py``) was retired in the live-widgets plan; this hub 

5backs both the inbox bridge and the ``/admin/_sse/widgets`` stream via 

6:class:`lexigram.reactive.Subject`, giving subscribers bounded 

7backpressure and the full operator toolbox. 

8""" 

9 

10from __future__ import annotations 

11 

12from collections.abc import AsyncGenerator 

13from dataclasses import dataclass 

14from typing import Any 

15 

16from lexigram.admin.realtime.sse import AdminEvent, AdminEventType 

17from lexigram.reactive import Subject, ops 

18 

19 

20@dataclass(frozen=True) 

21class _TargetedEvent: 

22 """Internal envelope pairing an event with its delivery scope. 

23 

24 ``target_users is None`` means broadcast to every subscriber; 

25 otherwise delivery is restricted to the listed user ids. This is 

26 what makes ``target_users``/``user_id`` filtering real instead of 

27 the no-op it would be if ``Subject`` only ever carried bare 

28 ``AdminEvent`` values. 

29 """ 

30 

31 event: AdminEvent 

32 target_users: tuple[Any, ...] | None 

33 

34 

35class SubjectAdminEventHub: 

36 """Fan-out hub for admin events over a reactive Subject. 

37 

38 ``on_overflow="drop_latest"`` is deliberate: the default 

39 ``"block"`` mode would suspend ``publish()`` itself — and thus every 

40 caller awaiting it, including a write-action HTTP response — on one 

41 slow subscriber. A dropped live delta is recoverable by the next 

42 reconcile-on-load snapshot; a blocked publisher is not. 

43 

44 Example: 

45 ```python 

46 hub = SubjectAdminEventHub() 

47 

48 async for event in hub.subscribe(resources=["users"]): 

49 publish_sse(event) 

50 

51 await hub.publish(AdminEvent(event_type=AdminEventType.RESOURCE_UPDATED, data={}, resource_type="users", resource_id=1)) 

52 ``` 

53 """ 

54 

55 def __init__(self, subject: Subject[_TargetedEvent] | None = None) -> None: 

56 """Initialize the hub. 

57 

58 Args: 

59 subject: Optional shared Subject; defaults to a private one 

60 with ``on_overflow="drop_latest"``. 

61 """ 

62 self._subject = subject or Subject[_TargetedEvent](on_overflow="drop_latest") 

63 

64 async def subscribe( 

65 self, 

66 user_id: Any | None = None, 

67 resources: list[str] | None = None, 

68 event_types: list[AdminEventType] | None = None, 

69 tenant_id: str | None = None, 

70 ) -> AsyncGenerator[AdminEvent, None]: 

71 """Subscribe to filtered admin events. 

72 

73 Args: 

74 user_id: Restricts delivery to broadcast events 

75 (``target_users is None``) plus events explicitly 

76 targeted at this user. ``None`` (the default) sees only 

77 broadcast events — matches the legacy hub's targeting 

78 semantics, which ``action_executor.py`` relies on to keep 

79 a caller's own action-result notification private to 

80 that caller. 

81 resources: Optional resource-type filter. Caller is 

82 responsible for authorizing which resources the 

83 subscriber may request — this hub applies the filter, 

84 it does not authorize it (see the SSE route handler). 

85 event_types: Optional event-type filter. 

86 tenant_id: Restricts delivery to events with no ``tenant_id`` 

87 (untenanted / framework-level) plus events whose 

88 ``tenant_id`` matches. ``None`` sees only untenanted 

89 events. 

90 

91 Yields: 

92 Matching AdminEvent objects as they are published. 

93 """ 

94 stream = self._subject.pipe( 

95 ops.filter(lambda te: te.target_users is None or user_id in te.target_users) 

96 ) 

97 stream = stream.pipe( 

98 ops.filter( 

99 lambda te: te.event.tenant_id is None or te.event.tenant_id == tenant_id 

100 ) 

101 ) 

102 if resources: 

103 stream = stream.pipe( 

104 ops.filter(lambda te: te.event.resource_type in resources) 

105 ) 

106 if event_types: 

107 stream = stream.pipe( 

108 ops.filter(lambda te: te.event.event_type in event_types) 

109 ) 

110 async for targeted in stream: 

111 yield targeted.event 

112 

113 async def publish( 

114 self, 

115 event: AdminEvent, 

116 target_users: list[Any] | None = None, 

117 ) -> int: 

118 """Publish an event to active subscribers. 

119 

120 Args: 

121 event: Admin event to publish. 

122 target_users: Restrict delivery to these user ids; ``None`` 

123 (the default) broadcasts to every subscriber. 

124 

125 Returns: 

126 Number of active subscriber channels (approximate). 

127 """ 

128 await self._subject.publish( 

129 _TargetedEvent( 

130 event=event, 

131 target_users=( 

132 tuple(target_users) if target_users is not None else None 

133 ), 

134 ) 

135 ) 

136 return 1 

137 

138 async def publish_notification( 

139 self, 

140 title: str, 

141 message: str, 

142 level: str = "info", 

143 target_users: list[Any] | None = None, 

144 ) -> int: 

145 """Publish a notification event. 

146 

147 Produces an ``AdminEventType.NOTIFICATION`` broadcast to the 

148 given users (or everyone when ``target_users`` is ``None``), 

149 mirroring the retired legacy hub's call signature so 

150 callers (the inbox bridge, action-result notifications) can move 

151 to this hub with no call-site changes beyond the import. 

152 """ 

153 event = AdminEvent( 

154 event_type=AdminEventType.NOTIFICATION, 

155 data={"title": title, "message": message, "level": level}, 

156 ) 

157 return await self.publish(event, target_users=target_users)