Coverage for src / lexigram / admin / di / sub_providers / realtime.py: 90%

1"""Admin realtime sub-provider — WebSocket, SSE, collaborative editing, events.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from lexigram.contracts.core.health import HealthCheckResult, HealthStatus 

8 

9if TYPE_CHECKING: 

10 from lexigram.admin.config import AdminConfig 

11 from lexigram.contracts.core.di import ( 

12 ContainerRegistrarProtocol, 

13 ContainerResolverProtocol, 

14 ) 

15 

16 

17class AdminRealtimeSubProvider: 

18 """Manages admin realtime infrastructure: WebSocket, SSE, collaborative editing. 

19 

20 Registers realtime services: WebSocket manager, event hub, SSE manager. 

21 """ 

22 

23 def __init__(self, config: AdminConfig, **kwargs: object) -> None: 

24 self._config = config 

25 self._kwargs = kwargs 

26 self._initialized = False 

27 

28 @property 

29 def config(self) -> AdminConfig: 

30 """Return current admin config.""" 

31 return self._config 

32 

33 async def register(self, container: ContainerRegistrarProtocol) -> None: 

34 """Register realtime services: WebSocket manager, event hub, SSE manager, collaborative.""" 

35 from lexigram.admin.realtime.sse import AdminEventHub 

36 from lexigram.admin.realtime.ws_handler_registry import WSMessageTypeRegistry 

37 from lexigram.admin.services.collaborative import CollaborativeEditingService 

38 from lexigram.admin.services.realtime import RealtimeService 

39 from lexigram.contracts.core.stores import LockStoreProtocol 

40 

41 realtime_svc = RealtimeService() 

42 

43 container.singleton(WSMessageTypeRegistry, WSMessageTypeRegistry()) 

44 container.singleton(RealtimeService, realtime_svc) 

45 container.singleton(AdminEventHub, AdminEventHub()) 

46 

47 # Lock store is provided externally through container bindings. 

48 lock_store: LockStoreProtocol | None = None 

49 

50 # Pre-build instance to avoid container trying to resolve RealtimeService 

51 # from TYPE_CHECKING-only annotation on CollaborativeEditingService.__init__. 

52 container.singleton( 

53 CollaborativeEditingService, 

54 CollaborativeEditingService( 

55 lock_store=lock_store, # type: ignore[arg-type] 

56 realtime_service=realtime_svc, 

57 ), 

58 ) 

59 

60 async def boot(self, container: ContainerResolverProtocol) -> None: 

61 """Boot realtime services: initialize WebSocket and SSE connections.""" 

62 self._initialized = True 

63 

64 async def shutdown(self) -> None: 

65 """Shut down realtime services.""" 

66 self._initialized = False 

67 

68 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

69 """Return realtime infrastructure health status.""" 

70 return HealthCheckResult( 

71 component="admin_realtime", 

72 status=HealthStatus.HEALTHY if self._initialized else HealthStatus.UNKNOWN, 

73 message="Admin realtime operational" 

74 if self._initialized 

75 else "Not yet initialized", 

76 ) 

77 

78 

79__all__ = ["AdminRealtimeSubProvider"]