diff --git a/api/src/api/v1/websocket.py b/api/src/api/v1/websocket.py index 2f805c9..c4902c6 100644 --- a/api/src/api/v1/websocket.py +++ b/api/src/api/v1/websocket.py @@ -365,13 +365,71 @@ class ConnectionManager: self.cell_locks.pop(lock_key, None) +# AURORA-1845: срок жизни кэша пользователя/доступа к форме внутри одного +# WebSocket-соединения. Ограничивает устаревание (смена роли/орг-юнитов, +# удаление формы) при живом соединении; источник истины по правам - +# can_edit в SQL-обёртках, он проверяется на каждое сообщение. +CONNECTION_CACHE_TTL_SECONDS = 30.0 + + +class ConnectionCache: + """Кэш на время соединения: пользователь и факт доступа к форме. + + Хранится вне процессора (процессор создаётся на каждое сообщение вместе + с новой сессией БД). Кэшируются только положительные результаты. + """ + + def __init__(self, ttl: float = CONNECTION_CACHE_TTL_SECONDS, clock=time.monotonic): + self.ttl = ttl + self._clock = clock + self._user: tuple[float, AppUser] | None = None + self._forms: dict[int, float] = {} + + def get_user(self, user_id: int) -> AppUser | None: + if self._user is None: + return None + ts, user = self._user + if user.id != user_id or self._clock() - ts >= self.ttl: + self._user = None + return None + return user + + def set_user(self, user: AppUser | None) -> None: + if user is not None: + self._user = (self._clock(), user) + + def form_access_ok(self, form_id: int) -> bool: + ts = self._forms.get(form_id) + if ts is None: + return False + if self._clock() - ts >= self.ttl: + self._forms.pop(form_id, None) + return False + return True + + def set_form_access(self, form_id: int) -> None: + self._forms[form_id] = self._clock() + + class FormEventProcess: def __init__(self, db: AsyncSession): self.sheet_service: SheetService = SheetService(db) self.bf_service: BudgetFormService = BudgetFormService(db) self.bl_service: BudgetLineService = BudgetLineService(db) self.user_service: UserService = UserService(db) + self._cache: ConnectionCache | None = None + async def _check_form_access(self, form_id: int, user: AppUser) -> bool: + """Доступ пользователя к форме; положительный результат кэшируется.""" + cache = self._cache + if cache and cache.form_access_ok(form_id): + return True + form = await self.bf_service.get(budget_form_id=form_id, user=user) + if not form: + return False + if cache: + cache.set_form_access(form_id) + return True async def process( self, @@ -380,8 +438,14 @@ class FormEventProcess: user_id: int, sheet: str, direction: str | None = None, + cache: ConnectionCache | None = None, ) -> int | bool | dict: - curr_user: AppUser = await self.user_service.get(user_id) + self._cache = cache + curr_user: AppUser | None = cache.get_user(user_id) if cache else None + if curr_user is None: + curr_user = await self.user_service.get(user_id) + if cache: + cache.set_user(curr_user) match event_data["event"]: case "cell_updated": @@ -458,8 +522,7 @@ class FormEventProcess: "value": any, } """ - form = await self.bf_service.get(budget_form_id=form_id, user=user) - if not form: + if not await self._check_form_access(form_id=form_id, user=user): return None return await self.sheet_service.update_cell( @@ -495,8 +558,7 @@ class FormEventProcess: contract_end_date: Optional[datetime] = None } """ - form = await self.bf_service.get(budget_form_id=form_id, user=user) - if not form: + if not await self._check_form_access(form_id=form_id, user=user): return None ids = await self.bl_service.get_ids(budget_form_id=form_id) @@ -617,8 +679,7 @@ class FormEventProcess: "row_id": int, } """ - form = await self.bf_service.get(budget_form_id=form_id, user=user) - if not form: + if not await self._check_form_access(form_id=form_id, user=user): return None result = await self.sheet_service.delete_line( form_id=form_id, @@ -970,6 +1031,8 @@ async def process_websocket( ) return + conn_cache = ConnectionCache() + while True: data = loads(await websocket.receive_text()) @@ -1068,10 +1131,13 @@ async def process_websocket( processor = processor_cls(db) t0 = time.perf_counter() data_old = data.copy() + process_kwargs = dict(kwargs_process) + if processor_cls is FormEventProcess: + process_kwargs["cache"] = conn_cache data["result"] = await processor.process( event_data=data, user_id=user_id, - **kwargs_process + **process_kwargs ) should_refresh_ahr = ( con_key == ConnectionKeyEnum.FORM diff --git a/api/tests/integration/test_websocket_connection_cache_integration.py b/api/tests/integration/test_websocket_connection_cache_integration.py new file mode 100644 index 0000000..806bd1f --- /dev/null +++ b/api/tests/integration/test_websocket_connection_cache_integration.py @@ -0,0 +1,49 @@ +"""AURORA-1845: WebSocket листа формы читает user/форму один раз на соединение.""" +from src.services.budget_form_service import BudgetFormService +from src.services.user_service import UserService + +WS_URL = "/api/v1/ws/form/1/sheet/AHR?direction=Support" +CELL = {"line_id": 1, "column": "plan.q1"} + + +def _roundtrip(ws, event: str, data: dict) -> dict: + ws.send_json({"event": event, "data": data}) + return ws.receive_json() + + +def test_ws_reads_user_and_form_once_per_connection(client, admin_tokens, monkeypatch): + calls = {"user": 0, "form": 0} + orig_user_get = UserService.get + orig_form_get = BudgetFormService.get + + async def user_get(self, *a, **kw): + calls["user"] += 1 + return await orig_user_get(self, *a, **kw) + + async def form_get(self, *a, **kw): + calls["form"] += 1 + return await orig_form_get(self, *a, **kw) + + monkeypatch.setattr(UserService, "get", user_get) + monkeypatch.setattr(BudgetFormService, "get", form_get) + + n = 3 + with client.websocket_connect(WS_URL) as ws: + ws.send_json({"event": "user_login", "data": {"token": admin_tokens["access_token"]}}) + try: + for i in range(n): + started = _roundtrip(ws, "cell_edit_start", CELL) + assert not started.get("error") + updated = _roundtrip(ws, "cell_updated", {**CELL, "value": 100 + i}) + assert not updated.get("error"), updated + _roundtrip(ws, "cell_edit_end", CELL) + # Только счётчики до отката: откат ниже идёт тем же соединением. + user_calls, form_calls = calls["user"], calls["form"] + finally: + # WS коммитит: возвращаем исходное значение (в фикстуре plan.q1 = NULL). + _roundtrip(ws, "cell_edit_start", CELL) + _roundtrip(ws, "cell_updated", {**CELL, "value": None}) + _roundtrip(ws, "cell_edit_end", CELL) + + assert user_calls == 1 + assert form_calls == 1 diff --git a/api/tests/unit/test_websocket_connection_cache.py b/api/tests/unit/test_websocket_connection_cache.py new file mode 100644 index 0000000..d220c80 --- /dev/null +++ b/api/tests/unit/test_websocket_connection_cache.py @@ -0,0 +1,74 @@ +"""AURORA-1845: user/форма читаются из БД один раз на соединение, а не на каждое сообщение.""" +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from src.api.v1.websocket import ConnectionCache, FormEventProcess + + +def _processor(user=None, form=object()): + proc = FormEventProcess(MagicMock()) + proc.user_service = MagicMock(get=AsyncMock(return_value=user or SimpleNamespace(id=7, role_id=2))) + proc.bf_service = MagicMock(get=AsyncMock(return_value=form)) + proc.sheet_service = MagicMock(update_cell=AsyncMock(return_value=[])) + return proc + + +EVENT = {"event": "cell_updated", "data": {"line_id": 1, "column": "c", "value": 1}} + + +async def _send(proc, cache, n): + for _ in range(n): + await proc.process(event_data=EVENT, form_id=5, user_id=7, sheet="S", cache=cache) + + +@pytest.mark.asyncio +async def test_cache_reads_user_and_form_once_per_connection(): + proc = _processor() + cache = ConnectionCache() + await _send(proc, cache, 5) + assert proc.user_service.get.await_count == 1 + assert proc.bf_service.get.await_count == 1 + assert proc.sheet_service.update_cell.await_count == 5 + + +@pytest.mark.asyncio +async def test_without_cache_reads_every_message(): + proc = _processor() + await _send(proc, None, 3) + assert proc.user_service.get.await_count == 3 + assert proc.bf_service.get.await_count == 3 + + +@pytest.mark.asyncio +async def test_cache_expires_after_ttl(): + now = [0.0] + cache = ConnectionCache(ttl=30, clock=lambda: now[0]) + proc = _processor() + await _send(proc, cache, 2) + now[0] = 31 + await _send(proc, cache, 2) + assert proc.user_service.get.await_count == 2 + assert proc.bf_service.get.await_count == 2 + + +@pytest.mark.asyncio +async def test_missing_form_is_not_cached(): + proc = _processor(form=None) + cache = ConnectionCache() + for _ in range(3): + res = await proc.process(event_data=EVENT, form_id=5, user_id=7, sheet="S", cache=cache) + assert res is None + assert proc.bf_service.get.await_count == 3 + proc.sheet_service.update_cell.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_missing_user_is_not_cached(): + proc = _processor() + proc.user_service.get = AsyncMock(return_value=None) + cache = ConnectionCache() + assert cache.get_user(7) is None + cache.set_user(None) + assert cache.get_user(7) is None