Update application functionality and supporting components

This commit is contained in:
lizerk 2026-08-17 13:54:35 +03:00
parent 99b6a7d4aa
commit 3c5a8c87e2
42 changed files with 1879 additions and 679 deletions

View File

@ -2,6 +2,9 @@
DEBUG=False
SECRET_KEY=change-me-django-secret-key
ALLOWED_HOSTS=localhost,127.0.0.1,web,nginx
WEBSOCKET_ALLOW_QUERY_JWT=false
# Set explicit frontend origins when WebSocket and frontend use different hosts.
# WEBSOCKET_ALLOWED_ORIGINS=https://app.example.com
# Database Settings
USE_SQLITE=False
@ -14,6 +17,9 @@ DB_PORT=5432
# Redis Settings
REDIS_HOST=redis
REDIS_PORT=6379
CHANNEL_LAYER_CAPACITY=1000
CHANNEL_LAYER_EXPIRY=60
CHANNEL_LAYER_GROUP_EXPIRY=86400
# Celery Settings
CELERY_BROKER_URL=redis://redis:6379/0

View File

@ -21,6 +21,7 @@ from .sessions import (
SessionListAPIView,
)
from .tokens import JWTTokenRefreshView, PhoneOrEmailTokenObtainPairView
from .websocket import WebSocketTicketAPIView
__all__ = [
"ChangePasswordAPIView",
@ -42,4 +43,5 @@ __all__ = [
"SessionDetailAPIView",
"SessionListAPIView",
"VerifyEmailAPIView",
"WebSocketTicketAPIView",
]

View File

@ -0,0 +1,31 @@
from drf_spectacular.utils import extend_schema
from rest_framework import serializers, status
from rest_framework.response import Response
from ..base import SessionBaseAPIView
from ..websocket_tickets import TICKET_MAX_AGE_SECONDS, issue_websocket_ticket
class WebSocketTicketResponseSerializer(serializers.Serializer):
ticket = serializers.CharField(read_only=True)
expires_in = serializers.IntegerField(read_only=True)
class WebSocketTicketAPIView(SessionBaseAPIView):
throttle_scope = "websocket_ticket"
@extend_schema(
tags=["Auth"],
summary="Получить одноразовый билет для WebSocket-подключения",
request=None,
responses={200: WebSocketTicketResponseSerializer},
)
def post(self, request):
ticket = issue_websocket_ticket(
user=request.user,
session_id=self.get_current_session_id(request),
)
return Response(
{"ticket": ticket, "expires_in": TICKET_MAX_AGE_SECONDS},
status=status.HTTP_200_OK,
)

View File

@ -80,6 +80,12 @@ class UserAuthSession(models.Model):
self.revoked_reason = reason
if commit:
self.save(update_fields=["revoked_at", "revoked_reason"])
from .realtime import schedule_auth_session_revoked
schedule_auth_session_revoked(
user_id=self.user_id,
session_id=str(self.session_id),
)
class QrLoginToken(models.Model):
@ -251,7 +257,8 @@ class QrLoginRequest(models.Model):
verbose_name_plural = "QR-запросы входа"
def __str__(self) -> str:
return f"{self.requester_device_name or 'unknown'}:{self.created_at:%Y-%m-%d %H:%M:%S}"
device = self.requester_device_name or "unknown"
return f"{device}:{self.created_at:%Y-%m-%d %H:%M:%S}"
@property
def is_expired(self) -> bool:

View File

@ -0,0 +1,38 @@
"""Realtime propagation of authentication-session revocation."""
from __future__ import annotations
import logging
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.db import transaction
logger = logging.getLogger(__name__)
def schedule_auth_session_revoked(*, user_id: int, session_id: str) -> None:
def _send() -> None:
channel_layer = get_channel_layer()
if channel_layer is None:
return
try:
async_to_sync(channel_layer.group_send)(
f"user_{user_id}",
{
"type": "auth_session_revoked",
"session_id": str(session_id),
},
)
except Exception as error:
logger.exception(
"Auth session revocation realtime update failed",
extra={
"event": "authentication.session.realtime_revocation_failed",
"user_id": user_id,
"auth_session_id": str(session_id),
"error_type": type(error).__name__,
},
)
transaction.on_commit(_send)

View File

@ -20,11 +20,13 @@ from .views import (
SessionDetailAPIView,
SessionListAPIView,
VerifyEmailAPIView,
WebSocketTicketAPIView,
)
urlpatterns = [
path("token/", PhoneOrEmailTokenObtainPairView.as_view(), name="token_obtain_pair"),
path("token/refresh/", JWTTokenRefreshView.as_view(), name="token_refresh"),
path("ws-ticket/", WebSocketTicketAPIView.as_view(), name="ws-ticket"),
path("qr-login/", QrLoginCreateAPIView.as_view(), name="qr-login"),
path(
"qr-login/exchange/",

View File

@ -23,6 +23,7 @@ from .endpoints import (
SessionDetailAPIView,
SessionListAPIView,
VerifyEmailAPIView,
WebSocketTicketAPIView,
)
__all__ = [
@ -47,5 +48,6 @@ __all__ = [
"SessionDetailAPIView",
"SessionListAPIView",
"VerifyEmailAPIView",
"WebSocketTicketAPIView",
"authenticate",
]

View File

@ -0,0 +1,85 @@
"""Short-lived, single-use credentials for browser WebSocket handshakes."""
from __future__ import annotations
import hashlib
import secrets
from dataclasses import dataclass
from django.contrib.auth import get_user_model
from django.core import signing
from django.core.cache import cache
from .models import UserAuthSession
User = get_user_model()
TICKET_MAX_AGE_SECONDS = 30
TICKET_SALT = "eusrr.websocket.ticket.v1"
class WebSocketTicketError(ValueError):
pass
@dataclass(frozen=True)
class WebSocketTicketIdentity:
user: object
session_id: str | None
def issue_websocket_ticket(*, user, session_id: str | None) -> str:
return signing.dumps(
{
"uid": user.pk,
"sid": str(session_id) if session_id else None,
"nonce": secrets.token_urlsafe(24),
},
salt=TICKET_SALT,
compress=True,
)
def consume_websocket_ticket(raw_ticket: str) -> WebSocketTicketIdentity:
try:
payload = signing.loads(
raw_ticket,
salt=TICKET_SALT,
max_age=TICKET_MAX_AGE_SECONDS,
)
user_id = int(payload["uid"])
session_id = payload.get("sid")
nonce = str(payload["nonce"])
except (signing.BadSignature, KeyError, TypeError, ValueError) as error:
raise WebSocketTicketError("invalid_or_expired_ticket") from error
replay_digest = hashlib.sha256(
f"{user_id}:{session_id}:{nonce}".encode()
).hexdigest()
replay_key = f"websocket-ticket-used:{replay_digest}"
try:
is_first_use = cache.add(
replay_key,
True,
timeout=TICKET_MAX_AGE_SECONDS * 2,
)
except Exception as error: # noqa: BLE001 - backend-neutral cache boundary
raise WebSocketTicketError("ticket_store_unavailable") from error
if not is_first_use:
raise WebSocketTicketError("ticket_already_used")
# The normal HTTP authentication already decided whether this account can
# issue a ticket. Session validity below remains authoritative for JWT
# sessions; Django-session tickets keep Django's authenticated identity.
user = User.objects.filter(pk=user_id).first()
if user is None:
raise WebSocketTicketError("user_unavailable")
if session_id and not UserAuthSession.objects.filter(
user=user,
session_id=session_id,
revoked_at__isnull=True,
).exists():
raise WebSocketTicketError("session_revoked")
return WebSocketTicketIdentity(user=user, session_id=session_id)

View File

@ -12,13 +12,16 @@ WebSocket Consumer Mixin для чатов.
"""
import logging
import rules
from channels.db import database_sync_to_async
from django.db.models import Prefetch, Q
from django.utils import timezone
from realtime.protocol import parse_positive_int, parse_text
from .models import (
Chat,
ChatMembership,
ChatReadState,
Message,
MessageReaction,
@ -80,11 +83,7 @@ class ChatConsumerMixin:
Устанавливает активный чат и отправляет начальную историю.
"""
chat_id = content.get("chat_id")
if not chat_id:
return
chat_id = int(chat_id)
chat_id = parse_positive_int(content.get("chat_id"), field="chat_id")
# Проверяем доступ
chat = await self._get_chat(chat_id)
@ -135,9 +134,9 @@ class ChatConsumerMixin:
async def _handle_close_chat(self, content):
"""Закрытие чата - сброс активного чата"""
chat_id = content.get("chat_id")
chat_id = parse_positive_int(content.get("chat_id"), field="chat_id")
if chat_id and int(chat_id) == self.active_chat_id:
if chat_id == self.active_chat_id:
# Сбрасываем индикатор печати
if self.active_chat_id:
chat = await self._get_chat(self.active_chat_id)
@ -376,12 +375,19 @@ class ChatConsumerMixin:
await self.send_json({"type": "error", "error": "No active chat"})
return
text = str(content.get("content") or content.get("text") or "").strip()
if not text:
return
text = parse_text(
content.get("content") or content.get("text"),
field="content",
max_length=20_000,
)
chat = await self._get_chat(self.active_chat_id)
if not chat:
if not chat or not await self._user_can_access(chat, self.user):
await self._deny_chat_action("send_message", self.active_chat_id)
return
if not await self._user_can_send(chat, self.user):
await self._deny_chat_action("send_message", self.active_chat_id)
return
# Проверка прав на отправку
@ -415,14 +421,21 @@ class ChatConsumerMixin:
if not self.active_chat_id:
return
message_id = content.get("message_id")
new_content = content.get("content", "").strip()
if not message_id or not new_content:
return
message_id = parse_positive_int(content.get("message_id"), field="message_id")
new_content = parse_text(
content.get("content"),
field="content",
max_length=20_000,
)
msg = await self._get_message(message_id)
if not msg or msg.author_id != self.user.id:
if (
not msg
or msg.chat_id != self.active_chat_id
or not await self._user_can_access(msg.chat, self.user)
or not await self._user_can_change_message(msg, self.user)
):
await self._deny_chat_action("edit_message", self.active_chat_id)
return
# Обновляем сообщение
@ -445,12 +458,16 @@ class ChatConsumerMixin:
if not self.active_chat_id:
return
message_id = content.get("message_id")
if not message_id:
return
message_id = parse_positive_int(content.get("message_id"), field="message_id")
msg = await self._get_message(message_id)
if not msg or msg.author_id != self.user.id:
if (
not msg
or msg.chat_id != self.active_chat_id
or not await self._user_can_access(msg.chat, self.user)
or not await self._user_can_delete_message(msg, self.user)
):
await self._deny_chat_action("delete_message", self.active_chat_id)
return
chat_id = msg.chat_id
@ -470,14 +487,16 @@ class ChatConsumerMixin:
async def _handle_add_reaction(self, content):
"""Добавление реакции"""
message_id = content.get("message_id")
emoji = content.get("emoji")
if not message_id or not emoji:
return
message_id = parse_positive_int(content.get("message_id"), field="message_id")
emoji = parse_text(content.get("emoji"), field="emoji", max_length=32)
msg = await self._get_message(message_id)
if not msg:
if (
not msg
or msg.chat_id != self.active_chat_id
or not await self._user_can_access(msg.chat, self.user)
):
await self._deny_chat_action("add_reaction", getattr(msg, "chat_id", None))
return
# Создаем реакцию
@ -502,14 +521,19 @@ class ChatConsumerMixin:
async def _handle_remove_reaction(self, content):
"""Удаление реакции"""
message_id = content.get("message_id")
emoji = content.get("emoji")
if not message_id or not emoji:
return
message_id = parse_positive_int(content.get("message_id"), field="message_id")
emoji = parse_text(content.get("emoji"), field="emoji", max_length=32)
msg = await self._get_message(message_id)
if not msg:
if (
not msg
or msg.chat_id != self.active_chat_id
or not await self._user_can_access(msg.chat, self.user)
):
await self._deny_chat_action(
"remove_reaction",
getattr(msg, "chat_id", None),
)
return
# Удаляем реакцию
@ -533,7 +557,11 @@ class ChatConsumerMixin:
return
chat = await self._get_chat(self.active_chat_id)
if not chat:
if (
not chat
or not await self._user_can_access(chat, self.user)
or not await self._user_can_send(chat, self.user)
):
return
await self._set_typing_status(chat, self.user, True)
@ -552,13 +580,13 @@ class ChatConsumerMixin:
},
)
async def _handle_stop_typing(self):
async def _handle_stop_typing(self, content=None):
"""Остановка индикатора печати"""
if not self.active_chat_id:
return
chat = await self._get_chat(self.active_chat_id)
if not chat:
if not chat or not await self._user_can_access(chat, self.user):
return
await self._set_typing_status(chat, self.user, False)
@ -580,16 +608,7 @@ class ChatConsumerMixin:
вместо этого.
Автоотметка происходит автоматически при загрузке сообщений.
"""
chat_id = content.get("chat_id")
if not chat_id:
logger.warning(
"Deprecated WebSocket read marker call rejected",
extra={
"event": "websocket.chat.deprecated_read_marker_rejected",
"reason": "chat_id_missing",
},
)
return
chat_id = parse_positive_int(content.get("chat_id"), field="chat_id")
logger.warning(
"Deprecated WebSocket read marker call ignored",
@ -606,9 +625,13 @@ class ChatConsumerMixin:
async def _handle_vote_poll(self, content):
"""Голосование в опросе"""
poll_id = content.get("poll_id")
poll_id = parse_positive_int(content.get("poll_id"), field="poll_id")
option_ids = content.get("option_ids")
if not self.active_chat_id or not poll_id or not isinstance(option_ids, list):
if (
not self.active_chat_id
or not isinstance(option_ids, list)
or len(option_ids) > 50
):
await self.send_json(
{
"type": "error",
@ -618,6 +641,15 @@ class ChatConsumerMixin:
)
return
option_ids = [
parse_positive_int(option_id, field="option_ids")
for option_id in option_ids
]
chat = await self._get_chat(self.active_chat_id)
if not chat or not await self._user_can_access(chat, self.user):
await self._deny_chat_action("vote_poll", self.active_chat_id)
return
result = await self._cast_poll_vote(
poll_id,
option_ids,
@ -678,24 +710,19 @@ class ChatConsumerMixin:
@database_sync_to_async
def _get_available_chat_ids(self, user):
"""Получить ID всех доступных чатов пользователя"""
# Получаем ID через membership
membership_chat_ids = list(
ChatMembership.objects.filter(user=user).values_list("chat_id", flat=True)
)
chat_ids = list(
candidates = (
Chat.objects.filter(
Q(type="global")
| Q(id__in=membership_chat_ids)
| Q(memberships__user=user, memberships__is_active=True)
| Q(participants=user)
| Q(created_by=user)
| Q(include_all_users=True)
| Q(type="comments", flags__show_in_messages=True)
)
.values_list("id", flat=True)
.select_related("created_by", "context_content_type")
.distinct()
)
return chat_ids
return [chat.id for chat in candidates if user_can_access_chat(chat, user)]
@database_sync_to_async
def _get_chat(self, chat_id):
@ -716,30 +743,30 @@ class ChatConsumerMixin:
@database_sync_to_async
def _user_can_access(self, chat, user):
"""Проверка доступа пользователя к чату"""
if chat.type == "comments":
return user_can_access_chat(chat, user)
return user_can_access_chat(chat, user)
if chat.type == "global":
return True
@database_sync_to_async
def _user_can_send(self, chat, user):
return rules.test_rule("communications.send_message", user, chat)
if chat.type == "private":
return ChatMembership.objects.filter(
chat=chat, user=user, is_active=True
).exists()
@database_sync_to_async
def _user_can_change_message(self, message, user):
return rules.test_rule("communications.change_message", user, message)
if chat.type == "group":
return ChatMembership.objects.filter(
chat=chat, user=user, is_active=True
).exists()
@database_sync_to_async
def _user_can_delete_message(self, message, user):
return rules.test_rule("communications.delete_message", user, message)
if chat.type in ("channel", "announcement", "comments"):
if chat.include_all_users:
return user.is_active
return ChatMembership.objects.filter(
chat=chat, user=user, is_active=True
).exists()
return False
async def _deny_chat_action(self, action, chat_id):
await self.send_json(
{
"type": "error",
"code": "access_denied",
"action": action,
"chat_id": chat_id,
"error": "Чат не найден или доступ запрещён.",
}
)
@database_sync_to_async
def _create_message(self, chat, user, content):

View File

@ -0,0 +1,60 @@
"""Realtime synchronization for chat access changes.
The channel layer is an integration boundary: model signals publish only a
small access invalidation event, while the consumer owns group membership.
"""
from __future__ import annotations
import logging
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.db import transaction
logger = logging.getLogger(__name__)
def schedule_chat_access_changed(
*,
chat_id: int,
user_id: int,
membership_active: bool,
) -> None:
"""Notify every live session after the membership transaction commits."""
def _send() -> None:
channel_layer = get_channel_layer()
if channel_layer is None:
logger.warning(
"Chat access update skipped because channel layer is unavailable",
extra={
"event": "chat.realtime_access.channel_layer_unavailable",
"chat_id": chat_id,
"participant_id": user_id,
},
)
return
try:
async_to_sync(channel_layer.group_send)(
f"user_{user_id}",
{
"type": "chat_access_changed",
"chat_id": chat_id,
"membership_active": membership_active,
},
)
except Exception as error:
logger.exception(
"Chat access realtime update failed",
extra={
"event": "chat.realtime_access.send_failed",
"chat_id": chat_id,
"participant_id": user_id,
"membership_active": membership_active,
"error_type": type(error).__name__,
},
)
transaction.on_commit(_send)

View File

@ -5,9 +5,10 @@ django-rules: декларативные правила доступа для co
https://github.com/dfunckt/django-rules
"""
import rules
import logging
import rules
logger = logging.getLogger(__name__)
@ -25,36 +26,9 @@ def is_superuser(user):
@rules.predicate
def is_chat_member(user, chat):
"""Пользователь является участником чата"""
if chat is None:
return False
from .utils import user_can_access_chat
# Используем прямые запросы к БД в обход prefetch cache
from .models import ChatMembership
# Глобальный чат доступен всем
if chat.type == "global":
return True
if chat.type == "comments" and chat.context_object_id:
try:
return chat.get_participants().filter(pk=user.pk).exists()
except Exception as error:
logger.exception(
"Comments chat membership participants could not be resolved",
extra={
"event": "chat.permission.participant_resolution_failed",
"chat_id": chat.id,
"user_id": user.id,
"permission": "membership",
"error_type": type(error).__name__,
},
)
return False
return (
ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists()
or chat.participants.filter(pk=user.pk).exists()
)
return user_can_access_chat(chat, user)
@rules.predicate

View File

@ -1,15 +1,39 @@
# backend\communications\signals.py
import logging
from django.conf import settings
from django.core.files.base import ContentFile
from django.db.models import F
from django.db.models.signals import post_save, pre_save
from django.db.models.signals import post_delete, post_save, pre_save
from django.dispatch import receiver
from django.utils.module_loading import import_string
import logging
logger = logging.getLogger(__name__)
@receiver(post_save, sender="communications.ChatMembership")
def synchronize_chat_membership_realtime(sender, instance, **kwargs):
"""Keep live consumer group membership aligned with database access."""
from communications.realtime import schedule_chat_access_changed
schedule_chat_access_changed(
chat_id=instance.chat_id,
user_id=instance.user_id,
membership_active=instance.is_active,
)
@receiver(post_delete, sender="communications.ChatMembership")
def revoke_deleted_chat_membership_realtime(sender, instance, **kwargs):
from communications.realtime import schedule_chat_access_changed
schedule_chat_access_changed(
chat_id=instance.chat_id,
user_id=instance.user_id,
membership_active=False,
)
# NOTE: Project-specific chat creation signals should be in your project's app
# Example: employees/signals.py - create_main_department_chat()

View File

@ -6,14 +6,15 @@
from __future__ import annotations
import datetime
from datetime import datetime as dt
from datetime import timezone as dt_tz
import logging
from django.utils import timezone
from django.utils.dateparse import parse_datetime
from .models import Chat, ChatMembership
logger = logging.getLogger(__name__)
def _coerce_ts(val: str | None) -> datetime.datetime:
"""
@ -44,8 +45,8 @@ def _coerce_ts(val: str | None) -> datetime.datetime:
iv = int(val)
if iv > 10**12: # пришли миллисекунды
iv //= 1000
return dt.fromtimestamp(iv, tz=dt_tz.utc)
except Exception:
return datetime.datetime.fromtimestamp(iv, tz=datetime.UTC)
except (OSError, OverflowError, TypeError, ValueError):
pass
# Потом пробуем ISO-строку
@ -57,6 +58,60 @@ def _coerce_ts(val: str | None) -> datetime.datetime:
return d
def _membership_or_legacy_participant(chat: Chat, user) -> bool:
membership_state = (
ChatMembership.objects.filter(chat=chat, user=user)
.values_list("is_active", flat=True)
.first()
)
if membership_state is not None:
return membership_state
return chat.participants.filter(pk=user.pk).exists()
def _context_participant_access(chat: Chat, user) -> bool | None:
try:
return chat.get_participants().filter(pk=user.pk).exists()
except Exception as error: # noqa: BLE001 - configurable resolver boundary
logger.warning(
"Chat access participant resolution failed",
extra={
"event": "chat.access.participant_resolution_failed",
"chat_id": chat.id,
"user_id": user.id,
"chat_type": chat.type,
"error_type": type(error).__name__,
},
exc_info=True,
)
return None
def _group_chat_access(chat: Chat, user) -> bool:
if chat.context_object_id:
resolved = _context_participant_access(chat, user)
if resolved is not None:
return resolved
return _membership_or_legacy_participant(chat, user)
def _comments_chat_access(chat: Chat, user) -> bool:
if (chat.flags or {}).get("show_in_messages") is True:
return True
if chat.context_object_id:
return _context_participant_access(chat, user) is True
return _membership_or_legacy_participant(chat, user)
_CHAT_ACCESS_CHECKERS = {
"private": _membership_or_legacy_participant,
"group": _group_chat_access,
"channel": _membership_or_legacy_participant,
"announcement": _membership_or_legacy_participant,
"comments": _comments_chat_access,
}
def user_can_access_chat(chat: Chat, user) -> bool:
"""
Проверяет, имеет ли пользователь доступ к чату.
@ -81,78 +136,13 @@ def user_can_access_chat(chat: Chat, user) -> bool:
if not chat or not user:
return False
if chat.type == "global":
# Создатель является владельцем ресурса и не должен терять доступ из-за
# рассинхронизации legacy participants/ChatMembership.
if chat.created_by_id == user.pk:
return True
def _has_direct_participation() -> bool:
return chat.participants.filter(pk=user.pk).exists()
if chat.type == "global" or chat.include_all_users:
return True
if chat.type == "private":
return (
ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists()
or _has_direct_participation()
)
if chat.type == "group":
# Групповые чаты с контекстным объектом используют resolver
if chat.context_object_id:
try:
participants = chat.get_participants()
return participants.filter(pk=user.pk).exists()
except Exception as e:
import logging
logger = logging.getLogger(__name__)
logger.warning(
"Chat access participant resolution failed",
extra={
"event": "chat.access.participant_resolution_failed",
"chat_id": chat.id,
"user_id": user.id,
"chat_type": chat.type,
"error_type": type(e).__name__,
},
exc_info=True,
)
return (
ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists()
or _has_direct_participation()
)
if chat.type in ("channel", "announcement"):
if chat.include_all_users:
return user.is_active
return (
ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists()
or _has_direct_participation()
)
if chat.type == "comments":
# Чаты-комментарии используют context-based доступ через
# get_participants()
if chat.context_object_id:
try:
participants = chat.get_participants()
return participants.filter(pk=user.pk).exists()
except Exception as e:
import logging
logger = logging.getLogger(__name__)
logger.warning(
"Comments chat access participant resolution failed",
extra={
"event": "chat.access.participant_resolution_failed",
"chat_id": chat.id,
"user_id": user.id,
"chat_type": chat.type,
"error_type": type(e).__name__,
},
exc_info=True,
)
return False
return (
ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists()
or _has_direct_participation()
)
return False
checker = _CHAT_ACCESS_CHECKERS.get(chat.type)
return checker(chat, user) if checker else False

View File

@ -1,7 +1,7 @@
asyncapi: 2.6.0
info:
title: EUSRR Realtime WebSocket API
version: 1.2.0
version: 1.3.0
description: |
Отдельная AsyncAPI-спецификация для WebSocket-контура EUSRR.
@ -9,8 +9,8 @@ info:
Этот документ описывает только двусторонний realtime-канал /ws/.
Аутентификация:
- query-параметр ?token=<jwt_access_token>
- или заголовок Authorization: Bearer <jwt_access_token>
- browser: одноразовый ?ticket=<ticket> из POST /api/auth/ws-ticket/
- non-browser: заголовок Authorization: Bearer <jwt_access_token>
defaultContentType: application/json
servers:
@ -18,10 +18,16 @@ servers:
url: localhost:8000/ws/
protocol: ws
description: Локальная разработка
security:
- websocketTicket: []
- bearerAuth: []
production:
url: api.example.com/ws/
protocol: wss
description: Продакшен через TLS
security:
- websocketTicket: []
- bearerAuth: []
channels:
userRealtime:
address: /ws/
@ -51,6 +57,7 @@ channels:
- $ref: '#/components/messages/MarkReadAction'
- $ref: '#/components/messages/CalendarSubscribeAction'
- $ref: '#/components/messages/CalendarUnsubscribeAction'
- $ref: '#/components/messages/PongAction'
subscribe:
operationId: serverEvent
summary: События от сервера клиенту
@ -60,6 +67,7 @@ channels:
- $ref: '#/components/messages/ErrorEvent'
- $ref: '#/components/messages/ChatOpenedEvent'
- $ref: '#/components/messages/ChatClosedEvent'
- $ref: '#/components/messages/ChatAccessChangedEvent'
- $ref: '#/components/messages/InitialMessagesEvent'
- $ref: '#/components/messages/NewMessageEvent'
- $ref: '#/components/messages/ListUpdateEvent'
@ -81,6 +89,16 @@ channels:
- $ref: '#/components/messages/CalendarSnapshotEvent'
- $ref: '#/components/messages/CalendarErrorEvent'
components:
securitySchemes:
websocketTicket:
type: apiKey
in: query
name: ticket
description: Одноразовый билет со сроком жизни 30 секунд.
bearerAuth:
type: http
scheme: bearer
bearerFormat: JWT
schemas:
ChatId:
type: integer
@ -367,6 +385,19 @@ components:
type: string
minLength: 1
maxLength: 80
PongAction:
name: PongAction
title: Ответ на heartbeat
payload:
type: object
required: [action]
properties:
action:
type: string
enum: [pong]
timestamp:
type: string
format: date-time
PingEvent:
name: PingEvent
payload:
@ -414,6 +445,19 @@ components:
enum: [chat_closed]
chat_id:
$ref: '#/components/schemas/ChatId'
ChatAccessChangedEvent:
name: ChatAccessChangedEvent
payload:
type: object
required: [type, chat_id, can_access]
properties:
type:
type: string
enum: [chat_access_changed]
chat_id:
$ref: '#/components/schemas/ChatId'
can_access:
type: boolean
InitialMessagesEvent:
name: InitialMessagesEvent
payload:

View File

@ -131,7 +131,12 @@ comment = Message.objects.create(
```javascript
// Frontend обычно использует общий websocketManager.
const ws = new WebSocket(`ws://localhost:8000/ws/?token=${accessToken}`);
const { ticket } = await apiClient.request('/api/auth/ws-ticket/', {
method: 'POST'
});
const ws = new WebSocket(
`ws://localhost:8000/ws/?ticket=${encodeURIComponent(ticket)}`
);
// Сделать comments-chat активным в этой сессии.
ws.send(JSON.stringify({
@ -142,6 +147,10 @@ ws.send(JSON.stringify({
// Получать обновления в реальном времени
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.type === 'ping') {
ws.send(JSON.stringify({ action: 'pong', timestamp: data.timestamp }));
return;
}
if (data.type === 'new_message' && data.chat_id === commentsChat.id) {
// Новый комментарий → обновить UI
addCommentToUI(data.message);

View File

@ -3,6 +3,7 @@ from __future__ import annotations
import phonenumbers
from django.conf import settings
from django.contrib.auth import get_user_model
from phonenumbers import PhoneNumberFormat
from employees.models import (
Department,
@ -11,8 +12,6 @@ from employees.models import (
EmployeeDepartment,
RoleAssignment,
)
from phonenumbers import PhoneNumberFormat
Employee = get_user_model()
@ -370,6 +369,7 @@ def resolve_chat_participants_for_procurement_request(
отдел-исполнитель и администраторы.
"""
from django.db.models import Q
from procurement.models import ProcurementRequest
if not isinstance(context_object, ProcurementRequest):
@ -432,6 +432,7 @@ def resolve_chat_participants_for_task(context_object, **kwargs):
Доступ соответствует видимости доски задачи.
"""
from django.db.models import Q
from tasks.models import Task
if not isinstance(context_object, Task):
@ -539,9 +540,10 @@ def resolve_chat_participants(chat):
# 6. Fallback: ChatMembership или participants
from communications.models import ChatMembership
membership_ids = ChatMembership.objects.filter(chat=chat).values_list(
"user_id", flat=True
)
membership_ids = ChatMembership.objects.filter(
chat=chat,
is_active=True,
).values_list("user_id", flat=True)
participant_ids = chat.participants.values_list("id", flat=True)

View File

@ -3,6 +3,8 @@ from importlib import import_module
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator, OriginValidator
from django.conf import settings
from django.core.asgi import get_asgi_application
from eusrr_backend.logging_utils import WebSocketLoggingMiddleware
@ -14,12 +16,23 @@ django_asgi_app = get_asgi_application()
JWTAuthMiddleware = import_module("eusrr_backend.channels_jwt").JWTAuthMiddleware
websocket_urlpatterns = import_module("realtime.routing").websocket_urlpatterns
websocket_application = AuthMiddlewareStack(
JWTAuthMiddleware(URLRouter(websocket_urlpatterns))
)
if settings.WEBSOCKET_ALLOWED_ORIGINS:
websocket_application = OriginValidator(
websocket_application,
settings.WEBSOCKET_ALLOWED_ORIGINS,
)
else:
websocket_application = AllowedHostsOriginValidator(websocket_application)
application = ProtocolTypeRouter(
{
"http": django_asgi_app,
"websocket": WebSocketLoggingMiddleware(
AuthMiddlewareStack(JWTAuthMiddleware(URLRouter(websocket_urlpatterns)))
websocket_application
),
}
)

View File

@ -2,12 +2,19 @@
from __future__ import annotations
import logging
from typing import Optional
from urllib.parse import parse_qs
from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from rest_framework.exceptions import AuthenticationFailed
from rest_framework_simplejwt.exceptions import TokenError
from api.auth.websocket_tickets import (
WebSocketTicketError,
consume_websocket_ticket,
)
from .logging_utils import update_log_context
@ -17,20 +24,32 @@ logger = logging.getLogger(__name__)
class JWTAuthMiddleware(BaseMiddleware):
"""Auth для Channels по JWT.
Поддерживает query ?token=... и header Authorization: Bearer ...
Браузер использует одноразовый query ?ticket=..., программные клиенты могут
передать Authorization: Bearer. Долгоживущий JWT в URL выключен по умолчанию.
"""
def _get_raw_token(self, scope) -> Optional[str]:
# 1) query string
qs = parse_qs(scope.get("query_string", b"").decode())
if "token" in qs and qs["token"]:
return qs["token"][0]
# 2) header
def _get_query_params(self, scope) -> dict[str, list[str]]:
return parse_qs(
scope.get("query_string", b"").decode("utf-8"),
max_num_fields=10,
)
def _get_raw_token(self, scope, query_params) -> str | None:
if getattr(settings, "WEBSOCKET_ALLOW_QUERY_JWT", False) and query_params.get(
"token"
):
return query_params["token"][0]
for name, value in scope.get("headers", []):
if name == b"authorization" and value.lower().startswith(b"bearer "):
return value.split()[1].decode()
parts = value.split(maxsplit=1)
if len(parts) == 2 and parts[1]:
return parts[1].decode()
return None
@database_sync_to_async
def _authenticate_ticket(self, raw_ticket: str):
return consume_websocket_ticket(raw_ticket)
@database_sync_to_async
def _authenticate_raw_token(self, raw_token: str):
from api.auth.authentication import SessionAwareJWTAuthentication
@ -43,25 +62,41 @@ class JWTAuthMiddleware(BaseMiddleware):
async def __call__(self, scope, receive, send):
scope.setdefault("user", AnonymousUser())
credential_type = None
try:
raw = self._get_raw_token(scope)
if raw:
user, token = await self._authenticate_raw_token(raw)
query_params = self._get_query_params(scope)
raw_ticket = (query_params.get("ticket") or [None])[0]
raw_token = self._get_raw_token(scope, query_params)
if raw_ticket:
if len(raw_ticket) > 2048:
raise WebSocketTicketError("ticket_too_long")
credential_type = "ticket"
identity = await self._authenticate_ticket(raw_ticket)
scope["user"] = identity.user
scope["auth_session_id"] = identity.session_id
scope["auth"] = {"type": "websocket_ticket"}
elif raw_token:
credential_type = "bearer"
user, token = await self._authenticate_raw_token(raw_token)
scope["user"] = user
scope["auth"] = token
except Exception as error:
session_user = scope.get("user")
if getattr(session_user, "is_authenticated", False):
update_log_context(user_id=getattr(session_user, "pk", None))
scope["auth_session_id"] = token.get("session_id")
except (
AuthenticationFailed,
TokenError,
ValueError,
) as error:
scope["user"] = AnonymousUser()
scope.pop("auth", None)
scope.pop("auth_session_id", None)
logger.info(
"WebSocket authentication rejected",
extra={
"event": "websocket.authentication.rejected",
"error_type": type(error).__name__,
"credential_type": credential_type,
},
)
# Сохраняем уже определённого session/anonymous-пользователя.
pass
user = scope.get("user")
if getattr(user, "is_authenticated", False):
update_log_context(user_id=getattr(user, "pk", None))

View File

@ -23,6 +23,16 @@ def _split_env_list(value: str) -> list[str]:
ALLOWED_HOSTS = _split_env_list(os.getenv("ALLOWED_HOSTS", "localhost,127.0.0.1"))
# Browser clients authenticate WebSocket handshakes with a short-lived ticket.
# Long-lived JWTs in query strings are accepted only during an explicit legacy
# migration window because URLs can be recorded by reverse proxies.
WEBSOCKET_ALLOW_QUERY_JWT = (
os.getenv("WEBSOCKET_ALLOW_QUERY_JWT", "false").lower() == "true"
)
WEBSOCKET_ALLOWED_ORIGINS = _split_env_list(
os.getenv("WEBSOCKET_ALLOWED_ORIGINS", "")
)
INSTALLED_APPS = [
"daphne",
"channels",
@ -435,6 +445,13 @@ CHANNEL_LAYERS = {
int(os.getenv("REDIS_PORT", "6379")),
)
],
# Absorb short notification/task bursts without silently dropping
# events; expiry bounds memory when clients disappear abruptly.
"capacity": int(os.getenv("CHANNEL_LAYER_CAPACITY", "1000")),
"expiry": int(os.getenv("CHANNEL_LAYER_EXPIRY", "60")),
"group_expiry": int(
os.getenv("CHANNEL_LAYER_GROUP_EXPIRY", "86400")
),
},
}
}
@ -464,6 +481,7 @@ REST_FRAMEWORK = {
"auth_refresh": "60/min",
"auth_sensitive": "10/min",
"auth_email": "5/hour",
"websocket_ticket": "30/min",
"registration": "5/hour",
},
"DEFAULT_PAGINATION_CLASS": "api.pagination.StandardPagination",

View File

@ -35,6 +35,7 @@ REST_FRAMEWORK = {
"auth_refresh": "10000/min",
"auth_sensitive": "10000/min",
"auth_email": "10000/min",
"websocket_ticket": "10000/min",
"registration": "10000/min",
},
}

View File

@ -13,7 +13,10 @@ WebSocket Consumer для пользователя.
"""
import asyncio
import json
import logging
from time import monotonic
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from django.contrib.auth.models import AnonymousUser
@ -28,6 +31,8 @@ from scheduling.realtime import (
parse_calendar_subscription,
)
from .protocol import ConnectionRateLimiter, RealtimeProtocolError, parse_action
logger = logging.getLogger(__name__)
@ -43,6 +48,23 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
- Онлайн-статус
"""
ACTION_HANDLERS = {
"open_chat": "_handle_open_chat",
"close_chat": "_handle_close_chat",
"send_message": "_handle_send_message",
"edit_message": "_handle_edit_message",
"delete_message": "_handle_delete_message",
"add_reaction": "_handle_add_reaction",
"remove_reaction": "_handle_remove_reaction",
"typing": "_handle_typing",
"stop_typing": "_handle_stop_typing",
"mark_read": "_handle_mark_read",
"vote_poll": "_handle_vote_poll",
"calendar_subscribe": "_handle_calendar_subscribe",
"calendar_unsubscribe": "_handle_calendar_unsubscribe",
"pong": "_handle_pong",
}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.ping_task = None
@ -54,6 +76,43 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
self.calendar_subscriptions = {}
self.calendar_refresh_task = None
self.pending_calendar_refresh_ids = set()
self.last_pong_at = monotonic()
self.rate_limiter = ConnectionRateLimiter()
async def receive(self, text_data=None, bytes_data=None, **kwargs):
"""Reject malformed, binary and oversized commands without crashing."""
if bytes_data is not None:
await self.send_json(
{
"type": "error",
"code": "binary_not_supported",
"error": "Поддерживаются только JSON-команды.",
}
)
return
if text_data is None:
return
if len(text_data.encode("utf-8")) > 64 * 1024:
await self.send_json(
{
"type": "error",
"code": "payload_too_large",
"error": "WebSocket-команда превышает лимит 64 КиБ.",
}
)
return
try:
content = await self.decode_json(text_data)
except (json.JSONDecodeError, TypeError, ValueError):
await self.send_json(
{
"type": "error",
"code": "invalid_json",
"error": "Некорректный JSON.",
}
)
return
await self.receive_json(content, **kwargs)
@database_sync_to_async
def _mark_chat_notifications_as_read(self, chat_id):
@ -94,6 +153,7 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
await self.accept()
# Запускаем ping цикл для keepalive
self.last_pong_at = monotonic()
self.ping_task = asyncio.create_task(self._ping_loop())
logger.info(
@ -159,6 +219,16 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
try:
while True:
await asyncio.sleep(self.ping_interval)
if monotonic() - self.last_pong_at > self.ping_interval * 3:
logger.warning(
"WebSocket client heartbeat timed out",
extra={
"event": "websocket.heartbeat.timeout",
"user_id": self.user.id,
},
)
await self.close(code=4408)
return
await self.send_json(
{"type": "ping", "timestamp": timezone.now().isoformat()}
)
@ -176,10 +246,22 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
async def receive_json(self, content, **kwargs):
"""Обработка входящих сообщений от клиента"""
if not isinstance(content, dict):
try:
action = parse_action(content)
except RealtimeProtocolError as error:
await self._send_protocol_error(error)
return
action = content.get("action", "")
if not self.rate_limiter.allow(action):
await self.send_json(
{
"type": "error",
"code": "rate_limited",
"action": action,
"error": "Слишком много WebSocket-команд. Повторите позже.",
}
)
return
logger.debug(
"WebSocket action received",
@ -190,45 +272,72 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
},
)
# Управление активным чатом
if action == "open_chat":
await self._handle_open_chat(content)
elif action == "close_chat":
await self._handle_close_chat(content)
handler_name = self.ACTION_HANDLERS.get(action)
if handler_name is None:
await self.send_json(
{
"type": "error",
"code": "unknown_action",
"action": action,
"error": "Неизвестная WebSocket-команда.",
}
)
return
handler = getattr(self, handler_name)
# Работа с сообщениями (только для активного чата)
elif action == "send_message":
await self._handle_send_message(content)
elif action == "edit_message":
await self._handle_edit_message(content)
elif action == "delete_message":
await self._handle_delete_message(content)
try:
await handler(content)
except RealtimeProtocolError as error:
await self._send_protocol_error(error, action=action)
except (TypeError, ValueError) as error:
logger.info(
"Invalid WebSocket command payload rejected",
extra={
"event": "websocket.command.invalid",
"action": action,
"user_id": self.user.id,
"error_type": type(error).__name__,
},
)
await self.send_json(
{
"type": "error",
"code": "invalid_payload",
"action": action,
"error": "Некорректные данные WebSocket-команды.",
}
)
except Exception as error: # noqa: BLE001 - protocol boundary must survive
logger.exception(
"WebSocket command failed",
extra={
"event": "websocket.command.failed",
"action": action,
"user_id": self.user.id,
"error_type": type(error).__name__,
},
)
await self.send_json(
{
"type": "error",
"code": "internal_error",
"action": action,
"error": "Не удалось выполнить WebSocket-команду.",
}
)
# Реакции
elif action == "add_reaction":
await self._handle_add_reaction(content)
elif action == "remove_reaction":
await self._handle_remove_reaction(content)
async def _send_protocol_error(self, error, *, action=None):
await self.send_json(
{
"type": "error",
"code": error.code,
"action": action,
"error": error.detail,
}
)
# Индикатор печати
elif action == "typing":
await self._handle_typing(content)
elif action == "stop_typing":
await self._handle_stop_typing()
# Отметка прочитанного
elif action == "mark_read":
await self._handle_mark_read(content)
# Голосования
elif action == "vote_poll":
await self._handle_vote_poll(content)
# Подписки календаря на диапазоны событий
elif action == "calendar_subscribe":
await self._handle_calendar_subscribe(content)
elif action == "calendar_unsubscribe":
await self._handle_calendar_unsubscribe(content)
async def _handle_pong(self, content):
self.last_pong_at = monotonic()
@database_sync_to_async
def _build_calendar_snapshot(self, subscription):
@ -489,3 +598,61 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
self.calendar_refresh_task = asyncio.create_task(
self._refresh_calendar_subscriptions()
)
async def chat_access_changed(self, event):
"""Apply membership changes to a live connection without reconnecting."""
try:
chat_id = int(event.get("chat_id"))
except (TypeError, ValueError):
return
chat = await self._get_chat(chat_id)
can_access = bool(chat and await self._user_can_access(chat, self.user))
group_name = f"chat_{chat_id}"
if can_access:
if chat_id not in self.subscribed_chats:
await self.channel_layer.group_add(group_name, self.channel_name)
self.subscribed_chats.add(chat_id)
elif chat_id in self.subscribed_chats:
await self.channel_layer.group_discard(group_name, self.channel_name)
self.subscribed_chats.discard(chat_id)
if self.active_chat_id == chat_id:
self.active_chat_id = None
await self.send_json(
{
"type": "chat_access_changed",
"chat_id": chat_id,
"can_access": can_access,
}
)
logger.info(
"Live chat access synchronized",
extra={
"event": "websocket.chat.access_synchronized",
"chat_id": chat_id,
"user_id": self.user.id,
"allowed": can_access,
},
)
async def auth_session_revoked(self, event):
"""Immediately terminate the socket that belongs to a revoked session."""
connection_session_id = self.scope.get("auth_session_id")
revoked_session_id = event.get("session_id")
if not connection_session_id or str(connection_session_id) != str(
revoked_session_id
):
return
logger.info(
"WebSocket closed after authentication session revocation",
extra={
"event": "websocket.authentication.session_revoked",
"user_id": self.user.id,
"auth_session_id": str(connection_session_id),
},
)
await self.close(code=4401)

View File

@ -0,0 +1,105 @@
"""Validation and abuse protection for the realtime command protocol."""
from __future__ import annotations
from collections import defaultdict, deque
from dataclasses import dataclass
from time import monotonic
from typing import Any
class RealtimeProtocolError(ValueError):
def __init__(self, code: str, detail: str):
super().__init__(detail)
self.code = code
self.detail = detail
def parse_action(content: Any) -> str:
if not isinstance(content, dict):
raise RealtimeProtocolError("invalid_payload", "Ожидался JSON-объект.")
action = content.get("action")
if not isinstance(action, str) or not action.strip():
raise RealtimeProtocolError("action_required", "Не указано действие.")
action = action.strip()
if len(action) > 64:
raise RealtimeProtocolError("invalid_action", "Некорректное действие.")
return action
def parse_positive_int(value: Any, *, field: str) -> int:
if isinstance(value, bool):
raise RealtimeProtocolError("invalid_field", f"Поле {field} некорректно.")
try:
parsed = int(value)
except (TypeError, ValueError) as error:
raise RealtimeProtocolError(
"invalid_field",
f"Поле {field} должно быть положительным целым числом.",
) from error
if parsed <= 0:
raise RealtimeProtocolError(
"invalid_field",
f"Поле {field} должно быть положительным целым числом.",
)
return parsed
def parse_text(value: Any, *, field: str, max_length: int) -> str:
if not isinstance(value, str):
raise RealtimeProtocolError(
"invalid_field",
f"Поле {field} должно быть строкой.",
)
parsed = value.strip()
if not parsed:
raise RealtimeProtocolError(
"invalid_field",
f"Поле {field} не может быть пустым.",
)
if len(parsed) > max_length:
raise RealtimeProtocolError(
"field_too_long",
f"Поле {field} превышает лимит {max_length} символов.",
)
return parsed
@dataclass(frozen=True)
class RateLimit:
requests: int
window_seconds: float
class ConnectionRateLimiter:
"""Small per-connection limiter; infrastructure limits remain a second layer."""
DEFAULT = RateLimit(requests=120, window_seconds=60)
GLOBAL = RateLimit(requests=180, window_seconds=60)
POLICIES = {
"typing": RateLimit(requests=30, window_seconds=10),
"stop_typing": RateLimit(requests=30, window_seconds=10),
"send_message": RateLimit(requests=30, window_seconds=60),
"calendar_subscribe": RateLimit(requests=30, window_seconds=60),
}
def __init__(self) -> None:
self._events: dict[str, deque[float]] = defaultdict(deque)
self._all_events: deque[float] = deque()
@staticmethod
def _consume(events: deque[float], policy: RateLimit, now: float) -> bool:
threshold = now - policy.window_seconds
while events and events[0] <= threshold:
events.popleft()
if len(events) >= policy.requests:
return False
events.append(now)
return True
def allow(self, action: str) -> bool:
now = monotonic()
if not self._consume(self._all_events, self.GLOBAL, now):
return False
policy = self.POLICIES.get(action, self.DEFAULT)
return self._consume(self._events[action], policy, now)

View File

@ -120,12 +120,15 @@ export function initUserWebSocket(options = {}) {
function handleMessage(data) {
const { type } = data;
window.dispatchEvent(new CustomEvent('user:ws-message', { detail: data }));
console.log('[UserWebSocket] Received message:', type, data);
switch (type) {
case 'ping':
// Keepalive - просто логируем
if (ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify({ action: 'pong', timestamp: data.timestamp }));
}
break;
case 'chat_opened':

View File

@ -9,11 +9,6 @@ import {
class NotificationManager {
constructor() {
this.ws = null;
this.reconnectAttempts = 0;
this.maxReconnectAttempts = 5;
this.reconnectDelay = 1000;
this.isConnected = false;
this.notificationPermission = "default";
this.soundEnabled = this.loadSoundPreference();
@ -107,7 +102,9 @@ class NotificationManager {
init() {
this.requestNotificationPermission();
this.connectWebSocket();
window.addEventListener("user:ws-message", (event) => {
this.handleMessage(event.detail);
});
this.loadInitialNotifications();
this.setupDropdown();
}
@ -138,52 +135,8 @@ class NotificationManager {
}
}
connectWebSocket() {
const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
// Используем универсальный WebSocket endpoint (realtime architecture)
const wsUrl = `${protocol}//${window.location.host}/ws/`;
console.log("[Notifications] Connecting to:", wsUrl);
handleMessage(data) {
try {
this.ws = new WebSocket(wsUrl);
this.ws.onopen = () => {
console.log("[Notifications] WebSocket connected");
this.isConnected = true;
this.reconnectAttempts = 0;
this.updateConnectionStatus(true);
};
this.ws.onmessage = (event) => {
this.handleMessage(event);
};
this.ws.onclose = () => {
console.log("[Notifications] WebSocket disconnected");
this.isConnected = false;
this.updateConnectionStatus(false);
this.attemptReconnect();
};
this.ws.onerror = (error) => {
console.error("[Notifications] WebSocket error:", error);
};
} catch (error) {
console.error("[Notifications] Failed to create WebSocket:", error);
this.attemptReconnect();
}
}
handleMessage(event) {
try {
const data = JSON.parse(event.data);
// Игнорируем ping сообщения для keepalive (не логируем)
if (data.type === "ping") {
return;
}
// Игнорируем list_update - это для списка чатов, не для уведомлений
if (data.type === "list_update") {
return;
@ -238,35 +191,6 @@ class NotificationManager {
}
}
attemptReconnect() {
if (this.reconnectAttempts < this.maxReconnectAttempts) {
this.reconnectAttempts++;
const delay = this.reconnectDelay * this.reconnectAttempts;
console.log(
`[Notifications] Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts})`
);
setTimeout(() => {
this.connectWebSocket();
}, delay);
} else {
console.error("[Notifications] Max reconnect attempts reached");
}
}
updateConnectionStatus(connected) {
// Визуальная индикация статуса подключения (опционально)
const badge = document.getElementById("notificationBadge");
if (badge) {
if (!connected) {
badge.style.backgroundColor = "#dc3545"; // красный при отключении
} else {
badge.style.backgroundColor = ""; // вернуть стандартный цвет
}
}
}
updateBadge(count) {
const badge = document.getElementById("notificationBadge");
if (badge) {

View File

@ -105,8 +105,10 @@
<h2>Точка подключения</h2>
<p><code>{{ ws_endpoint_url }}</code></p>
<p>
Аутентификация поддерживает либо query-параметр
<code>?token=&lt;jwt_access_token&gt;</code>, либо заголовок
Browser получает одноразовый билет через
<code>POST /api/auth/ws-ticket/</code> и подключается с
<code>?ticket=&lt;one_time_ticket&gt;</code>. Non-browser
client может использовать заголовок
<code>Authorization: Bearer &lt;jwt_access_token&gt;</code>.
</p>
</article>

View File

@ -1,5 +1,4 @@
import pytest
from asgiref.sync import sync_to_async
from channels.testing import WebsocketCommunicator
from django.contrib.auth import get_user_model
@ -40,7 +39,8 @@ async def test_revoked_session_becomes_anonymous_for_websocket():
communicator = WebsocketCommunicator(
JWTAuthMiddleware(app),
f"/ws/?token={str(token)}",
"/ws/",
headers=[(b"authorization", f"Bearer {token}".encode())],
)
connected, _ = await communicator.connect()
assert connected is True
@ -52,7 +52,34 @@ async def test_revoked_session_becomes_anonymous_for_websocket():
captured.clear()
communicator = WebsocketCommunicator(
JWTAuthMiddleware(app),
f"/ws/?token={str(token)}",
"/ws/",
headers=[(b"authorization", f"Bearer {token}".encode())],
)
connected, _ = await communicator.connect()
assert connected is True
assert captured["user"].is_authenticated is False
await communicator.disconnect()
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_query_string_jwt_is_rejected_by_default():
user = await sync_to_async(User.objects.create_user)(
email="ws-query-auth@example.com",
password="testpass123",
phone_number="+79997770012",
)
token = AccessToken.for_user(user)
captured = {}
async def app(scope, receive, send):
captured["user"] = scope["user"]
await send({"type": "websocket.accept"})
await send({"type": "websocket.close"})
communicator = WebsocketCommunicator(
JWTAuthMiddleware(app),
f"/ws/?token={token}",
)
connected, _ = await communicator.connect()
assert connected is True

View File

@ -0,0 +1,275 @@
import time
from unittest.mock import patch
import pytest
from asgiref.sync import sync_to_async
from channels.layers import get_channel_layer
from channels.testing import WebsocketCommunicator
from api.auth.models import UserAuthSession
from api.auth.websocket_tickets import (
WebSocketTicketError,
consume_websocket_ticket,
issue_websocket_ticket,
)
from communications.models import Chat, ChatMembership, Message, MessageReaction
from eusrr_backend.asgi import application
from realtime.protocol import ConnectionRateLimiter
pytestmark = pytest.mark.django_db(transaction=True)
def test_connection_rate_limiter_has_action_specific_limits():
limiter = ConnectionRateLimiter()
assert all(limiter.allow("typing") for _ in range(30))
assert limiter.allow("typing") is False
limiter = ConnectionRateLimiter()
assert all(limiter.allow("open_chat") for _ in range(120))
assert limiter.allow("open_chat") is False
limiter = ConnectionRateLimiter()
assert all(limiter.allow(f"unknown-{index}") for index in range(180))
assert limiter.allow("one-more-unique-action") is False
@pytest.mark.asyncio
async def test_inactive_membership_is_not_subscribed(ws_communicator, user):
chat = await sync_to_async(Chat.objects.create)(name="Revoked", type="group")
await sync_to_async(ChatMembership.objects.create)(
chat=chat,
user=user,
role="member",
is_active=False,
)
communicator = await ws_communicator(user=user)
connected, _ = await communicator.connect()
assert connected
await get_channel_layer().group_send(
f"chat_{chat.id}",
{
"type": "chat_message",
"chat_id": chat.id,
"payload": {"id": 1, "content": "private"},
},
)
assert await communicator.receive_nothing(timeout=0.1)
await communicator.disconnect()
@pytest.mark.asyncio
async def test_reaction_requires_chat_access(ws_communicator, user, user2):
chat = await sync_to_async(Chat.objects.create)(name="Private", type="group")
await sync_to_async(ChatMembership.objects.create)(chat=chat, user=user2)
message = await sync_to_async(Message.objects.create)(
chat=chat,
author=user2,
content="secret",
)
communicator = await ws_communicator(user=user)
connected, _ = await communicator.connect()
assert connected
await communicator.send_json_to(
{"action": "add_reaction", "message_id": message.id, "emoji": "👍"}
)
response = await communicator.receive_json_from()
assert response["code"] == "access_denied"
reaction_exists = await sync_to_async(
MessageReaction.objects.filter(message=message).exists
)()
assert not reaction_exists
await communicator.disconnect()
@pytest.mark.asyncio
async def test_membership_revocation_removes_live_group(ws_communicator, user):
chat = await sync_to_async(Chat.objects.create)(name="Live", type="group")
membership = await sync_to_async(ChatMembership.objects.create)(
chat=chat,
user=user,
)
communicator = await ws_communicator(user=user)
connected, _ = await communicator.connect()
assert connected
membership.is_active = False
await sync_to_async(membership.save)(update_fields=["is_active"])
access_event = await communicator.receive_json_from()
assert access_event == {
"type": "chat_access_changed",
"chat_id": chat.id,
"can_access": False,
}
await get_channel_layer().group_send(
f"chat_{chat.id}",
{
"type": "chat_message",
"chat_id": chat.id,
"payload": {"id": 2, "content": "after revoke"},
},
)
assert await communicator.receive_nothing(timeout=0.1)
await communicator.disconnect()
@pytest.mark.asyncio
async def test_membership_addition_joins_live_group(ws_communicator, user):
chat = await sync_to_async(Chat.objects.create)(name="Added live", type="group")
communicator = await ws_communicator(user=user)
connected, _ = await communicator.connect()
assert connected
await sync_to_async(ChatMembership.objects.create)(chat=chat, user=user)
access_event = await communicator.receive_json_from()
assert access_event == {
"type": "chat_access_changed",
"chat_id": chat.id,
"can_access": True,
}
await get_channel_layer().group_send(
f"chat_{chat.id}",
{
"type": "chat_message",
"chat_id": chat.id,
"payload": {"id": 3, "content": "after grant"},
},
)
update = await communicator.receive_json_from()
assert update["type"] == "list_update"
await communicator.disconnect()
@pytest.mark.asyncio
async def test_send_message_honors_membership_permission(ws_communicator, user):
chat = await sync_to_async(Chat.objects.create)(name="Read only", type="group")
await sync_to_async(ChatMembership.objects.create)(
chat=chat,
user=user,
role="guest",
)
communicator = await ws_communicator(user=user)
connected, _ = await communicator.connect()
assert connected
await communicator.send_json_to({"action": "open_chat", "chat_id": chat.id})
assert (await communicator.receive_json_from())["type"] == "chat_opened"
await communicator.send_json_to({"action": "send_message", "content": "denied"})
response = await communicator.receive_json_from()
assert response["code"] == "access_denied"
message_exists = await sync_to_async(
Message.objects.filter(chat=chat, content="denied").exists
)()
assert not message_exists
await communicator.disconnect()
@pytest.mark.asyncio
async def test_invalid_command_does_not_terminate_connection(ws_communicator, user):
communicator = await ws_communicator(user=user)
connected, _ = await communicator.connect()
assert connected
await communicator.send_json_to({"action": "open_chat", "chat_id": "invalid"})
first_error = await communicator.receive_json_from()
assert first_error["code"] == "invalid_field"
await communicator.send_json_to({"action": "not_supported"})
second_error = await communicator.receive_json_from()
assert second_error["code"] == "unknown_action"
await communicator.send_to(text_data="{not-json")
malformed_error = await communicator.receive_json_from()
assert malformed_error["code"] == "invalid_json"
await communicator.send_to(bytes_data=b"binary")
binary_error = await communicator.receive_json_from()
assert binary_error["code"] == "binary_not_supported"
await communicator.disconnect()
def test_websocket_ticket_is_single_use(user):
ticket = issue_websocket_ticket(user=user, session_id=None)
identity = consume_websocket_ticket(ticket)
assert identity.user.pk == user.pk
with pytest.raises(WebSocketTicketError, match="ticket_already_used"):
consume_websocket_ticket(ticket)
def test_websocket_ticket_expires(user):
ticket = issue_websocket_ticket(user=user, session_id=None)
future = time.time() + 31
with (
patch("django.core.signing.time.time", return_value=future),
pytest.raises(WebSocketTicketError, match="invalid_or_expired_ticket"),
):
consume_websocket_ticket(ticket)
def test_websocket_ticket_rejects_revoked_session(user):
session = UserAuthSession.objects.create(user=user, refresh_token_hash="stub")
ticket = issue_websocket_ticket(user=user, session_id=str(session.session_id))
session.revoke(reason="test")
with pytest.raises(WebSocketTicketError, match="session_revoked"):
consume_websocket_ticket(ticket)
def test_websocket_ticket_endpoint_requires_authentication(
api_client,
auth_client_factory,
user,
):
anonymous_response = api_client.post("/api/auth/ws-ticket/")
assert anonymous_response.status_code == 401
response = auth_client_factory(user).post("/api/auth/ws-ticket/")
assert response.status_code == 200
assert response.data["expires_in"] == 30
assert response.data["ticket"]
@pytest.mark.asyncio
async def test_untrusted_websocket_origin_is_rejected(user):
ticket = await sync_to_async(issue_websocket_ticket)(user=user, session_id=None)
communicator = WebsocketCommunicator(
application,
f"/ws/?ticket={ticket}",
headers=[(b"origin", b"https://attacker.example")],
)
connected, _ = await communicator.connect()
assert connected is False
@pytest.mark.asyncio
async def test_ticket_connects_from_trusted_origin(user):
ticket = await sync_to_async(issue_websocket_ticket)(user=user, session_id=None)
communicator = WebsocketCommunicator(
application,
f"/ws/?ticket={ticket}",
headers=[(b"origin", b"http://localhost:3000")],
)
connected, _ = await communicator.connect()
assert connected is True
await communicator.disconnect()
@pytest.mark.asyncio
async def test_matching_session_revocation_closes_live_socket(ws_communicator, user):
communicator = await ws_communicator(user=user)
communicator.scope["auth_session_id"] = "session-to-revoke"
connected, _ = await communicator.connect()
assert connected
await get_channel_layer().group_send(
f"user_{user.id}",
{
"type": "auth_session_revoked",
"session_id": "session-to-revoke",
},
)
close_event = await communicator.receive_output()
assert close_event == {"type": "websocket.close", "code": 4401}

View File

@ -3,7 +3,7 @@
EUSRR использует одно авторизованное WebSocket-соединение на вкладку приложения:
```text
ws://localhost:8000/ws/?token=ACCESS_TOKEN
POST /api/auth/ws-ticket/ → ws://localhost:8000/ws/?ticket=ONE_TIME_TICKET
```
Отдельных `/ws/chat/`, `/ws/chats/`, `/ws/notifications/` и `/ws/realtime/` в текущем routing нет.
@ -43,21 +43,23 @@ NEXT_PUBLIC_WS_URL=ws://localhost:8000/ws/
## Аутентификация
Browser передаёт access token как `?token=...`. Middleware также понимает `Authorization: Bearer ...`, что удобно для non-browser clients.
Browser сначала выполняет авторизованный `POST /api/auth/ws-ticket/`, затем передаёт полученный одноразовый билет как `?ticket=...`. Билет действует 30 секунд, подписан сервером и может быть использован только один раз. Поэтому долгоживущий JWT не попадает в URL WebSocket handshake.
JWT проходит `SessionAwareJWTAuthentication`: проверяется подпись, пользователь и связанная `UserAuthSession`. При ошибке scope остаётся anonymous, после чего consumer закрывает соединение кодом `4401`.
Non-browser clients могут передать `Authorization: Bearer <access_token>`. Legacy `?token=<access_token>` по умолчанию отключён; временно включить его можно только явной настройкой `WEBSOCKET_ALLOW_QUERY_JWT=true`.
Токен в query string может попасть в access logs. В production маскируйте query parameters в reverse proxy и application monitoring.
ASGI проверяет `Origin` через `AllowedHostsOriginValidator`. Если frontend и WebSocket работают на разных host, разрешённые frontend origins задаются явно через `WEBSOCKET_ALLOWED_ORIGINS=https://app.example.com`. JWT и ticket также проверяют связанную `UserAuthSession`; отзыв сессии немедленно закрывает принадлежащее ей соединение кодом `4401`.
## Подписки consumer-а
После connect `UserConsumer`:
- подписывается на все доступные `chat_{id}`;
- подписывается только на доступные активные `chat_{id}`;
- подписывается на `user_{user_id}`;
- подписывается на `notifications_{user_id}`;
- запускает server ping каждые 20 секунд.
Изменение `ChatMembership` синхронизируется с уже открытыми соединениями: при добавлении consumer входит в группу, при отзыве доступа немедленно выходит из неё и закрывает активный чат.
`active_chat_id` определяет, получает соединение полное chat event или только обновление списка. При открытии чата связанные message notifications отмечаются прочитанными; уведомление об уже открытом чате подавляется.
## Команды клиента
@ -77,6 +79,7 @@ JWT проходит `SessionAwareJWTAuthentication`: проверяется п
| `mark_read` | `chat_id` | Deprecated; используйте HTTP mark-read action |
| `calendar_subscribe` | `subscription_id`, `start`, `end` | Подписаться на snapshot событий диапазона; optional `calendar_id`, `scope` |
| `calendar_unsubscribe` | `subscription_id` | Закрыть подписку диапазона календаря |
| `pong` | optional `timestamp` | Ответ на server heartbeat |
Пример:
@ -89,6 +92,7 @@ JWT проходит `SessionAwareJWTAuthentication`: проверяется п
Основные публичные `type`:
- connection: `ping`, `error`;
- access lifecycle: `chat_access_changed`;
- chat lifecycle: `chat_opened`, `chat_closed`, `initial_messages`;
- messages: `new_message`, `list_update`, `message_edited`, `message_deleted`;
- reactions: `reaction_added`, `reaction_removed`;
@ -108,16 +112,18 @@ Channel-layer handler names вроде `chat_message`, `notification_message` и
`websocketManager` поддерживает один shared connection, пока есть message/status subscribers. Он:
- подключается только при наличии access token;
- повторно подключается после обновления JWT;
- подключается только при наличии auth-сессии и получает новый одноразовый ticket на каждую попытку;
- отключается после очистки auth-сессии;
- синхронизирует active chat после reconnect;
- использует exponential backoff до пяти попыток;
- отвечает `pong`, контролирует heartbeat и закрывает зависшее соединение;
- использует exponential backoff с jitter, после пяти быстрых попыток продолжает восстановление раз в минуту;
- немедленно повторяет подключение после browser `online` и возвращения вкладки;
- закрывает socket, когда исчезает последний subscriber.
Не создавайте WebSocket напрямую в page/component. Chat-specific hook подписывается на shared manager и не владеет отдельным соединением.
После reconnect frontend повторяет активные календарные подписки. Если snapshot не приходит за три секунды или исчерпаны попытки reconnect, выполняется HTTP fallback для текущего диапазона.
Чаты синхронизируют сообщения после локального watermark, список чатов перечитывается, а задачи и список календарей выполняют контрольную REST-синхронизацию. Поэтому WebSocket используется как сигнал ускорения, а не как единственный источник истины.
## Изменение протокола
@ -139,10 +145,13 @@ pytest tests/realtime tests/api/test_documentation.py -q
Ручная проверка через `wscat`:
```bash
wscat -c 'ws://localhost:8000/ws/?token=ACCESS_TOKEN'
TICKET=$(curl -s -X POST http://localhost:8000/api/auth/ws-ticket/ \
-H 'Authorization: Bearer ACCESS_TOKEN' | jq -r .ticket)
wscat -c "ws://localhost:8000/ws/?ticket=${TICKET}" \
-H 'Origin: http://localhost:3000'
```
После подключения server отправляет `ping`; client-side `pong` action не требуется.
После подключения server отправляет `ping`; client должен ответить `{"action":"pong"}`. При отсутствии ответа в течение трёх интервалов соединение закрывается кодом `4408`.
## Production proxy
@ -154,5 +163,6 @@ Reverse proxy должен:
- иметь idle timeout больше 20 секунд;
- маршрутизировать все worker-ы к общему Redis channel layer;
- передавать корректные `Host`/forwarded headers.
- не подменять и не отбрасывать допустимый `Origin`.
`1006` — общий abnormal closure, а не самостоятельная причина. Сначала проверяйте close code/log consumer-а, JWT/session, proxy Upgrade, доступность Redis и совпадение endpoint-а.

View File

@ -34,6 +34,7 @@ import {
Link2,
} from "lucide-react";
import { DocumentAcknowledgementsReport } from "@/components/documents/DocumentAcknowledgementsReport";
import { DocumentAcknowledgementAudienceBadges } from "@/components/documents/DocumentAcknowledgementAudience";
import { DocumentMetadataEditor } from "@/components/documents/DocumentMetadataEditor";
import { DocumentDetailModal } from "@/components/documents/DocumentDetailModal";
import { DocumentTaskLinks } from "@/components/documents/DocumentTaskLinks";
@ -1601,8 +1602,6 @@ function DocumentsPageContent() {
const isSelected = selectionMode && isDocumentSelected;
const isDocumentMenuOpen = documentMenuOpenId === doc.id;
const fileSize = formatFileSize(doc.file_size);
const recipientsCount = doc.recipients?.length || 0;
const departmentsCount = doc.departments?.length || 0;
const hasPreview = Boolean(doc.file_url && canPreviewDocument(doc.file_name || doc.title));
const isAcknowledging = acknowledgingDocumentId === doc.id;
const isChangingMaximumHiding = maximallyHidingDocumentId === doc.id;
@ -1976,27 +1975,10 @@ function DocumentsPageContent() {
</div>
</div>
{(doc.sent_to_all || recipientsCount > 0 || departmentsCount > 0) && (
<div className="mt-3 flex flex-wrap items-center gap-1.5">
{doc.sent_to_all && (
<span className="app-selected-soft inline-flex items-center gap-1.5 rounded-full px-2.5 py-1 text-xs font-medium">
<Users size={12} />
Для всей компании
</span>
)}
{recipientsCount > 0 && (
<span className="app-badge inline-flex items-center gap-1.5 px-2.5 py-1 text-xs font-medium">
<Users size={12} />
{recipientsCount} получ.
</span>
)}
{departmentsCount > 0 && (
<span className="app-badge inline-flex items-center gap-1.5 px-2.5 py-1 text-xs font-medium">
{departmentsCount} отдел.
</span>
)}
</div>
)}
<DocumentAcknowledgementAudienceBadges
document={doc}
className="mt-3"
/>
{doc.description && (
<p className="app-text-wrap mt-3 line-clamp-3 text-sm leading-relaxed text-[var(--foreground)]">

View File

@ -413,7 +413,13 @@ function MessageDialogPageContent() {
// WS event handler
useEffect(() => {
handlers.current.onMessage = (data) => {
if (data.type === "new_message" && data.message) {
if (
data.type === "chat_access_changed"
&& data.chat_id === chatId
&& data.can_access === false
) {
router.replace("/messages");
} else if (data.type === "new_message" && data.message) {
const newMsg = data.message;
const isMyMessage = newMsg.author_id === user?.id;
const wasNearBottom = scroll.isNearBottom();
@ -512,7 +518,7 @@ function MessageDialogPageContent() {
handlers.current.onConnect = () => {
void cm.syncLatestMessages({ limit: 30, scrollOnSync: scroll.isNearBottom() });
};
}, [handlers, user?.id, chatId, scroll, cm, markRead, updateMessageReactionsSummary]);
}, [handlers, user?.id, chatId, scroll, cm, markRead, router, updateMessageReactionsSummary]);
/* ── cleanup timers ── */
useEffect(() => () => {

View File

@ -715,11 +715,14 @@ const TaskCardContent = memo(function TaskCardContent({
/>
) : null}
<div className="mb-2 min-w-0 overflow-hidden whitespace-normal text-left">
<div className="flex min-w-0 items-center gap-2">
<div className="flex min-w-0 items-start gap-2">
<span className="app-badge shrink-0 rounded-full px-1.5 py-0.5 text-[10px] font-semibold">
#{task.id}
</span>
<h3 className="min-w-0 flex-1 truncate text-sm font-semibold text-[var(--foreground)]">
<h3
className="line-clamp-4 min-w-0 flex-1 break-words text-sm font-semibold leading-5 text-[var(--foreground)]"
title={task.title}
>
{task.title}
</h3>
<div
@ -941,11 +944,14 @@ function TaskDragOverlayCard({ task }: { task: TaskCard }) {
<article className="app-surface-elevated w-[17rem] cursor-grabbing rounded-xl border border-[var(--accent-primary)] p-3 shadow-2xl ring-1 ring-[color:var(--accent-primary)]/20">
<div className="mb-2 flex items-start gap-2">
<div className="min-w-0 flex-1 overflow-hidden whitespace-normal text-left">
<div className="flex min-w-0 items-center gap-2">
<div className="flex min-w-0 items-start gap-2">
<span className="app-badge shrink-0 rounded-full px-1.5 py-0.5 text-[10px] font-semibold">
#{task.id}
</span>
<h3 className="min-w-0 max-w-[9rem] truncate text-sm font-semibold text-[var(--foreground)]">
<h3
className="line-clamp-4 min-w-0 max-w-[9rem] break-words text-sm font-semibold leading-5 text-[var(--foreground)]"
title={task.title}
>
{task.title}
</h3>
</div>

View File

@ -0,0 +1,180 @@
import { BookOpenCheck, Building2, User, Users } from "lucide-react";
import {
documentAudiencePersonName,
getDocumentAcknowledgementAudience,
} from "@/lib/document-acknowledgement-audience";
import type { Document } from "@/types/api";
type DocumentAcknowledgementAudienceBadgesProps = {
document: Document;
className?: string;
maxTargets?: number;
};
export function DocumentAcknowledgementAudienceBadges({
document,
className = "",
maxTargets = 3,
}: DocumentAcknowledgementAudienceBadgesProps) {
const audience = getDocumentAcknowledgementAudience(document);
if (!audience.required) {
return (
<div className={`flex flex-wrap items-center gap-1.5 ${className}`.trim()}>
<span
className="app-badge inline-flex items-center gap-1.5 rounded-full px-2.5 py-1 text-xs font-medium"
title="Документ не требует подтверждения ознакомления"
>
<BookOpenCheck size={12} />
Не требуется
</span>
</div>
);
}
if (audience.companyWide) {
return (
<div className={`flex flex-wrap items-center gap-1.5 ${className}`.trim()}>
<span
className="app-selected-soft inline-flex items-center gap-1.5 rounded-full px-2.5 py-1 text-xs font-medium"
title="Документ предназначен для ознакомления всей компании"
>
<Users size={12} />
Вся компания
</span>
</div>
);
}
const targets = [
...audience.departments.map((department) => ({
key: `department-${department.id}`,
label: department.name,
icon: <Building2 key="department-icon" size={12} />,
})),
...audience.recipients.map((recipient) => ({
key: `recipient-${recipient.id}`,
label: documentAudiencePersonName(recipient),
icon: <User key="recipient-icon" size={12} />,
})),
];
const visibleTargets = targets.slice(0, Math.max(1, maxTargets));
const hiddenTargets = Math.max(0, targets.length - visibleTargets.length);
return (
<div className={`flex flex-wrap items-center gap-1.5 ${className}`.trim()}>
{visibleTargets.map((target) => (
<span
key={target.key}
className="app-badge inline-flex items-center gap-1.5 rounded-full px-2.5 py-1 text-xs font-medium"
title={`Документ предназначен для ознакомления: ${target.label}`}
>
{target.icon}
{target.label}
</span>
))}
{hiddenTargets > 0 ? (
<span className="app-badge rounded-full px-2.5 py-1 text-xs font-medium">
+{hiddenTargets}
</span>
) : null}
{targets.length === 0 ? (
<span
className="app-badge inline-flex items-center gap-1.5 rounded-full px-2.5 py-1 text-xs font-medium"
title="Документ предназначен выбранной аудитории ознакомления"
>
<Users size={12} />
{audience.total > 0 ? `${audience.total} сотрудн.` : "Выбранные сотрудники"}
</span>
) : null}
</div>
);
}
export function DocumentAcknowledgementAudienceDetails({
document,
}: {
document: Document;
}) {
const audience = getDocumentAcknowledgementAudience(document);
return (
<div className="app-surface rounded-xl p-4">
<h3 className="app-text-muted mb-3 text-xs font-semibold uppercase tracking-wide">
Для ознакомления
</h3>
{!audience.required ? (
<div className="app-surface-muted flex items-center gap-2 rounded-lg px-3 py-2 text-xs text-[var(--foreground)]">
<BookOpenCheck size={14} className="app-text-muted" />
Подтверждение ознакомления не требуется
</div>
) : (
<div className="space-y-3">
{audience.companyWide ? (
<div className="app-selected app-accent-text flex items-center gap-2 rounded-lg px-3 py-2 text-xs font-medium">
<Users size={14} />
Вся компания
</div>
) : null}
{audience.departments.length > 0 ? (
<div>
<div className="app-text-muted mb-2 flex items-center gap-2 text-xs font-medium">
<Building2 size={14} />
Отделы для ознакомления ({audience.departments.length})
</div>
<div className="space-y-1">
{audience.departments.map((department) => (
<div
key={department.id}
className="app-surface-muted flex items-center gap-2 rounded-lg px-3 py-2 text-xs text-[var(--foreground)]"
>
<Building2 size={12} className="app-text-muted" />
{department.name}
</div>
))}
</div>
</div>
) : null}
{audience.recipients.length > 0 ? (
<div>
<div className="app-text-muted mb-2 flex items-center gap-2 text-xs font-medium">
<Users size={14} />
Сотрудники для ознакомления ({audience.recipients.length})
</div>
<div className="max-h-40 space-y-1 overflow-y-auto">
{audience.recipients.map((recipient) => (
<div
key={recipient.id}
className="app-surface-muted flex items-center gap-2 rounded-lg px-3 py-2 text-xs text-[var(--foreground)]"
>
<User size={12} className="app-text-muted" />
{documentAudiencePersonName(recipient)}
</div>
))}
</div>
</div>
) : null}
{!audience.companyWide
&& audience.departments.length === 0
&& audience.recipients.length === 0 ? (
<div className="app-surface-muted flex items-center gap-2 rounded-lg px-3 py-2 text-xs text-[var(--foreground)]">
<Users size={14} className="app-text-muted" />
{audience.total > 0
? `Выбрано сотрудников: ${audience.total}`
: "Выбранная аудитория ознакомления"}
</div>
) : null}
{audience.total > 0 ? (
<p className="app-text-muted text-xs">Всего должны ознакомиться: {audience.total}</p>
) : null}
</div>
)}
</div>
);
}

View File

@ -10,8 +10,6 @@ import {
Calendar,
HardDrive,
FolderOpen,
Users,
Building2,
Tag,
Clock,
CheckCircle,
@ -29,6 +27,7 @@ import {
X,
} from "lucide-react";
import { DocumentAcknowledgeButton, DocumentAcknowledgement } from "./DocumentAcknowledgement";
import { DocumentAcknowledgementAudienceDetails } from "./DocumentAcknowledgementAudience";
import { DocumentComments } from "./DocumentComments";
import { DocumentRelated } from "./DocumentRelated";
import { DocumentTaskLinks } from "./DocumentTaskLinks";
@ -411,69 +410,7 @@ export function DocumentDetailModal({
</div>
)}
{/* Recipients */}
{(document.sent_to_all ||
(document.departments && document.departments.length > 0) ||
(document.recipients && document.recipients.length > 0)) && (
<div className="app-surface rounded-xl p-4">
<h3 className="app-text-muted mb-3 text-xs font-semibold uppercase tracking-wide">
Получатели
</h3>
<div className="space-y-3">
{document.sent_to_all && (
<div className="app-selected app-accent-text flex items-center gap-2 rounded-lg px-3 py-2 text-xs font-medium">
<CheckCircle size={14} />
Все сотрудники
</div>
)}
{document.departments && document.departments.length > 0 && (
<div>
<div className="app-text-muted mb-2 flex items-center gap-2 text-xs font-medium">
<Building2 size={14} />
Отделы ({document.departments.length})
</div>
<div className="space-y-1">
{document.departments.map((dept) => (
<div
key={dept.id}
className="app-surface-muted flex items-center gap-2 rounded-lg px-3 py-2 text-xs text-[var(--foreground)]"
>
<Building2 size={12} className="app-text-muted" />
{dept.name}
</div>
))}
</div>
</div>
)}
{document.recipients && document.recipients.length > 0 && (
<div>
<div className="app-text-muted mb-2 flex items-center gap-2 text-xs font-medium">
<Users size={14} />
Получатели ({document.recipients.length})
</div>
<div className="max-h-40 space-y-1 overflow-y-auto">
{document.recipients.slice(0, 5).map((recipient) => (
<div
key={recipient.id}
className="app-surface-muted flex items-center gap-2 rounded-lg px-3 py-2 text-xs text-[var(--foreground)]"
>
<User size={12} className="app-text-muted" />
{recipient.last_name} {recipient.first_name}
</div>
))}
{document.recipients.length > 5 && (
<div className="app-text-muted px-3 py-2 text-xs">
... и ещё {document.recipients.length - 5}
</div>
)}
</div>
</div>
)}
</div>
</div>
)}
<DocumentAcknowledgementAudienceDetails document={document} />
</div>
</div>
)}
@ -687,69 +624,7 @@ export function DocumentDetailModal({
</div>
)}
{/* Recipients */}
{(document.sent_to_all ||
(document.departments && document.departments.length > 0) ||
(document.recipients && document.recipients.length > 0)) && (
<div className="app-surface rounded-xl p-4">
<h3 className="app-text-muted mb-3 text-xs font-semibold uppercase tracking-wide">
Получатели
</h3>
<div className="space-y-3">
{document.sent_to_all && (
<div className="app-selected app-accent-text flex items-center gap-2 rounded-lg px-3 py-2 text-xs font-medium">
<CheckCircle size={14} />
Все сотрудники
</div>
)}
{document.departments && document.departments.length > 0 && (
<div>
<div className="app-text-muted mb-2 flex items-center gap-2 text-xs font-medium">
<Building2 size={14} />
Отделы ({document.departments.length})
</div>
<div className="space-y-1">
{document.departments.map((dept) => (
<div
key={dept.id}
className="app-surface-muted flex items-center gap-2 rounded-lg px-3 py-2 text-xs text-[var(--foreground)]"
>
<Building2 size={12} className="app-text-muted" />
{dept.name}
</div>
))}
</div>
</div>
)}
{document.recipients && document.recipients.length > 0 && (
<div>
<div className="app-text-muted mb-2 flex items-center gap-2 text-xs font-medium">
<Users size={14} />
Получатели ({document.recipients.length})
</div>
<div className="max-h-40 space-y-1 overflow-y-auto">
{document.recipients.slice(0, 5).map((recipient) => (
<div
key={recipient.id}
className="app-surface-muted flex items-center gap-2 rounded-lg px-3 py-2 text-xs text-[var(--foreground)]"
>
<User size={12} className="app-text-muted" />
{recipient.last_name} {recipient.first_name}
</div>
))}
{document.recipients.length > 5 && (
<div className="app-text-muted px-3 py-2 text-xs">
... и ещё {document.recipients.length - 5}
</div>
)}
</div>
</div>
)}
</div>
</div>
)}
<DocumentAcknowledgementAudienceDetails document={document} />
</div>
</div>
)}

View File

@ -19,6 +19,7 @@ import {
} from "lucide-react";
import { ExpandableFeedText } from "@/components/feed/ExpandableFeedText";
import { DocumentAcknowledgementAudienceBadges } from "@/components/documents/DocumentAcknowledgementAudience";
import { RequestAvatar } from "@/components/requests/RequestAvatar";
import TaskLinkPill from "@/components/tasks/TaskLinkPill";
import { canPreviewDocument } from "@/lib/document-preview";
@ -299,16 +300,7 @@ export function FeedRegulationCard({
) : null}
<div className="app-text-muted mt-3 flex flex-wrap items-center gap-2 text-xs">
{document.sent_to_all ? (
<span className="app-badge inline-flex items-center gap-1 rounded-full px-2 py-1">
<Users size={11} /> Для всей компании
</span>
) : null}
{(document.departments || []).slice(0, 3).map((department) => (
<span key={department.id} className="app-badge rounded-full px-2 py-1">
{department.name}
</span>
))}
<DocumentAcknowledgementAudienceBadges document={document} />
{acknowledgementRequiredForUser ? (
<span className={`inline-flex items-center gap-1 rounded-full px-2 py-1 ${document.is_acknowledged ? "app-feedback-success" : "app-feedback-warning"}`}>
<CheckCircle2 size={11} />

View File

@ -78,7 +78,9 @@ export function CalendarProvider({ children }: { children: ReactNode }) {
useEffect(() => {
if (!user) return undefined;
const unsubscribe = wsManager.subscribe((event) => {
let hasConnected = false;
let wasConnected = false;
const unsubscribeMessages = wsManager.subscribe((event) => {
if (event.type !== "calendar_update") return;
const data = event.data as {
@ -100,9 +102,18 @@ export function CalendarProvider({ children }: { children: ReactNode }) {
}
}, 100);
});
const unsubscribeStatus = wsManager.subscribeStatus((status) => {
if (status.isConnected && hasConnected && !wasConnected) {
calendarService.clearCache();
void loadCalendars();
}
if (status.isConnected) hasConnected = true;
wasConnected = status.isConnected;
});
return () => {
unsubscribe();
unsubscribeMessages();
unsubscribeStatus();
if (realtimeSyncTimerRef.current) {
clearTimeout(realtimeSyncTimerRef.current);
realtimeSyncTimerRef.current = null;

View File

@ -135,6 +135,10 @@ function startRealtime() {
const unsubscribeMessages = wsManager.subscribe((event) => {
const data = event as RealtimeChatEvent;
if (data.type === "chat_access_changed") {
void loadAllChats(false);
return;
}
if (
(data.type === "list_update" || data.type === "message_deleted")
&& data.event === "message_deleted"

View File

@ -36,7 +36,9 @@ export function useTaskBoardRealtime(options: TaskBoardRealtimeOptions): void {
}, [options]);
useEffect(() => {
const unsubscribe = wsManager.subscribe((event) => {
let hasConnected = false;
let wasConnected = false;
const unsubscribeMessages = wsManager.subscribe((event) => {
const update = parseTaskBoardSocketUpdate(event);
if (!update) return;
@ -74,9 +76,17 @@ export function useTaskBoardRealtime(options: TaskBoardRealtimeOptions): void {
optionsRef.current.onBoardChange(update.boardId);
}, current.debounceMs ?? 250);
});
const unsubscribeStatus = wsManager.subscribeStatus((status) => {
if (status.isConnected && hasConnected && !wasConnected) {
optionsRef.current.onBoardChange(0);
}
if (status.isConnected) hasConnected = true;
wasConnected = status.isConnected;
});
return () => {
unsubscribe();
unsubscribeMessages();
unsubscribeStatus();
if (syncTimerRef.current) {
clearTimeout(syncTimerRef.current);
syncTimerRef.current = null;

View File

@ -0,0 +1,62 @@
import test from "node:test";
import assert from "node:assert/strict";
import { getDocumentAcknowledgementAudience } from "./document-acknowledgement-audience.ts";
const accounting = { id: 10, name: "Бухгалтерия" };
const procurement = { id: 20, name: "Закупки" };
const employee = { id: 7, first_name: "Иван", last_name: "Иванов" };
test("company-wide badge is based on acknowledgement audience, not access alone", () => {
assert.equal(
getDocumentAcknowledgementAudience({
sent_to_all: true,
acknowledgement_required: true,
acknowledgement_for_all: false,
acknowledgement_departments: [accounting],
}).companyWide,
false,
);
});
test("acknowledgement for everyone inherits a selectively shared audience", () => {
const audience = getDocumentAcknowledgementAudience({
sent_to_all: false,
departments: [accounting],
recipients: [employee],
acknowledgement_required: true,
acknowledgement_for_all: true,
acknowledgement_departments: [procurement],
});
assert.equal(audience.companyWide, false);
assert.deepEqual(audience.departments, [accounting]);
assert.deepEqual(audience.recipients, [employee]);
});
test("selective acknowledgement ignores the broader access audience", () => {
const audience = getDocumentAcknowledgementAudience({
sent_to_all: true,
departments: [accounting],
acknowledgement_required: true,
acknowledgement_for_all: false,
acknowledgement_departments: [procurement],
acknowledgement_recipients: [employee],
});
assert.deepEqual(audience.departments, [procurement]);
assert.deepEqual(audience.recipients, [employee]);
});
test("documents without acknowledgement have no presentation audience", () => {
const audience = getDocumentAcknowledgementAudience({
sent_to_all: true,
acknowledgement_required: false,
acknowledgement_for_all: true,
});
assert.equal(audience.required, false);
assert.equal(audience.companyWide, false);
assert.deepEqual(audience.departments, []);
assert.deepEqual(audience.recipients, []);
});

View File

@ -0,0 +1,66 @@
import type { Document, User } from "@/types/api";
type AudienceDepartment = NonNullable<Document["departments"]>[number];
export type DocumentAcknowledgementAudience = {
required: boolean;
companyWide: boolean;
departments: AudienceDepartment[];
recipients: User[];
total: number;
};
function uniqueById<T extends { id: number }>(items: T[]): T[] {
return [...new Map(items.map((item) => [item.id, item])).values()];
}
/**
* Resolves the business audience that must acknowledge a document.
* Access fields are used only when acknowledgement is configured for every
* employee who can open a selectively shared document.
*/
export function getDocumentAcknowledgementAudience(
document: Document,
): DocumentAcknowledgementAudience {
if (!document.acknowledgement_required) {
return {
required: false,
companyWide: false,
departments: [],
recipients: [],
total: 0,
};
}
const companyWide = Boolean(
document.acknowledgement_for_all && document.sent_to_all,
);
const inheritsAccessAudience = Boolean(document.acknowledgement_for_all);
return {
required: true,
companyWide,
departments: companyWide
? []
: uniqueById(
inheritsAccessAudience
? document.departments || []
: document.acknowledgement_departments || [],
),
recipients: companyWide
? []
: uniqueById(
inheritsAccessAudience
? document.recipients || []
: document.acknowledgement_recipients || [],
),
total: Math.max(0, Number(document.acknowledgement_total) || 0),
};
}
export function documentAudiencePersonName(person: User): string {
return [person.last_name, person.first_name, person.patronymic]
.filter(Boolean)
.join(" ")
.trim() || person.email || `Сотрудник #${person.id}`;
}

View File

@ -1,10 +1,13 @@
import assert from "node:assert/strict";
import test from "node:test";
import { apiClient } from "@/lib/api";
test("multiple realtime subscribers share one WebSocket connection", async () => {
const originalWindow = globalThis.window;
const originalLocalStorage = globalThis.localStorage;
const originalWebSocket = globalThis.WebSocket;
const originalRequest = apiClient.request;
class FakeWebSocket {
static CONNECTING = 0;
@ -29,6 +32,10 @@ test("multiple realtime subscribers share one WebSocket connection", async () =>
this.sent.push(JSON.parse(payload));
}
receive(payload) {
this.onmessage?.({ data: JSON.stringify(payload) });
}
close() {
this.readyState = FakeWebSocket.CLOSED;
this.onclose?.({ code: 1000, reason: "", wasClean: true });
@ -42,13 +49,21 @@ test("multiple realtime subscribers share one WebSocket connection", async () =>
},
};
globalThis.WebSocket = FakeWebSocket;
apiClient.request = async (endpoint) => {
assert.equal(endpoint, "/api/auth/ws-ticket/");
return { ticket: "single-use-ticket", expires_in: 30 };
};
try {
const { default: manager } = await import("./websocketManager.ts");
const unsubscribeMessages = manager.subscribe(() => {});
const unsubscribeStatus = manager.subscribeStatus(() => {});
await new Promise((resolve) => setTimeout(resolve, 0));
assert.equal(FakeWebSocket.instances.length, 1);
assert.match(FakeWebSocket.instances[0].url, /ticket=single-use-ticket/);
assert.doesNotMatch(FakeWebSocket.instances[0].url, /token=/);
const socket = FakeWebSocket.instances[0];
socket.open();
@ -61,11 +76,116 @@ test("multiple realtime subscribers share one WebSocket connection", async () =>
load_history: false,
});
socket.receive({ type: "ping", timestamp: "2026-08-17T12:00:00Z" });
assert.deepEqual(socket.sent.at(-1), {
action: "pong",
timestamp: "2026-08-17T12:00:00Z",
});
unsubscribeMessages();
unsubscribeStatus();
} finally {
globalThis.window = originalWindow;
globalThis.localStorage = originalLocalStorage;
globalThis.WebSocket = originalWebSocket;
apiClient.request = originalRequest;
}
});
test("connection recovers immediately when browser comes back online", async () => {
const originalWindow = globalThis.window;
const originalDocument = globalThis.document;
const originalLocalStorage = globalThis.localStorage;
const originalWebSocket = globalThis.WebSocket;
const originalRequest = apiClient.request;
const originalSetTimeout = globalThis.setTimeout;
const originalClearTimeout = globalThis.clearTimeout;
const originalSetInterval = globalThis.setInterval;
const originalClearInterval = globalThis.clearInterval;
const listeners = new Map();
const scheduledTimeouts = [];
class FakeWebSocket {
static OPEN = 1;
static instances = [];
constructor(url) {
this.url = url;
this.readyState = 0;
FakeWebSocket.instances.push(this);
}
open() {
this.readyState = FakeWebSocket.OPEN;
this.onopen?.();
}
send() {}
fail() {
this.readyState = 3;
this.onclose?.({ code: 1006, reason: "network", wasClean: false });
}
close() {
this.readyState = 3;
}
}
globalThis.window = {
addEventListener(type, handler) {
listeners.set(type, handler);
},
};
globalThis.document = {
visibilityState: "visible",
addEventListener(type, handler) {
listeners.set(type, handler);
},
};
globalThis.localStorage = {
getItem(key) {
return key === "access_token" ? "access" : null;
},
};
globalThis.WebSocket = FakeWebSocket;
globalThis.setTimeout = (handler, delay) => {
const timer = { handler, delay, cleared: false };
scheduledTimeouts.push(timer);
return timer;
};
globalThis.clearTimeout = (timer) => { timer.cleared = true; };
globalThis.setInterval = () => ({ interval: true });
globalThis.clearInterval = () => {};
apiClient.request = async () => ({ ticket: "recovery-ticket", expires_in: 30 });
try {
const { WebSocketManager } = await import("./websocketManager.ts");
const manager = new WebSocketManager();
const unsubscribe = manager.subscribe(() => {});
await new Promise((resolve) => originalSetTimeout(resolve, 0));
assert.equal(FakeWebSocket.instances.length, 1);
FakeWebSocket.instances[0].open();
FakeWebSocket.instances[0].fail();
assert.equal(scheduledTimeouts.length, 1);
assert.ok(scheduledTimeouts[0].delay >= 1600);
assert.ok(scheduledTimeouts[0].delay <= 2400);
listeners.get("online")?.();
await new Promise((resolve) => originalSetTimeout(resolve, 0));
assert.equal(scheduledTimeouts[0].cleared, true);
assert.equal(FakeWebSocket.instances.length, 2);
unsubscribe();
} finally {
globalThis.window = originalWindow;
globalThis.document = originalDocument;
globalThis.localStorage = originalLocalStorage;
globalThis.WebSocket = originalWebSocket;
globalThis.setTimeout = originalSetTimeout;
globalThis.clearTimeout = originalClearTimeout;
globalThis.setInterval = originalSetInterval;
globalThis.clearInterval = originalClearInterval;
apiClient.request = originalRequest;
}
});

View File

@ -1,8 +1,6 @@
/**
* Singleton WebSocket Manager
* Управляет единственным WebSocket соединением для всего приложения
*/
/** Shared, authenticated and self-healing WebSocket transport. */
import { apiClient } from '@/lib/api';
import { AUTH_SESSION_EVENT, type AuthSessionEventDetail } from '@/lib/api/client';
import { getWebSocketUrl } from '@/lib/url';
@ -18,28 +16,48 @@ export type ConnectionStatus = {
maxReconnectAttempts: number;
};
type StatusHandler = (status: ConnectionStatus) => void;
type WebSocketTicketResponse = { ticket: string; expires_in: number };
const RAPID_RECONNECT_LIMIT = 5;
const MAX_RECONNECT_DELAY_MS = 60_000;
const HEARTBEAT_TIMEOUT_MS = 70_000;
const HEARTBEAT_CHECK_MS = 15_000;
export class WebSocketManager {
private static instance: WebSocketManager | null = null;
private ws: WebSocket | null = null;
private reconnectTimeout: NodeJS.Timeout | null = null;
private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
private reconnectAttempts = 0;
private maxReconnectAttempts = 5;
private activeChatId: number | null = null;
private messageHandlers: Set<MessageHandler> = new Set();
private statusHandlers: Set<StatusHandler> = new Set();
private readonly messageHandlers = new Set<MessageHandler>();
private readonly statusHandlers = new Set<StatusHandler>();
private isConnecting = false;
private shouldReconnect = false;
private connectionGeneration = 0;
private lastServerMessageAt = 0;
private constructor() {
if (typeof window !== 'undefined') {
window.addEventListener(AUTH_SESSION_EVENT, (event) => {
const detail = (event as CustomEvent<AuthSessionEventDetail>).detail;
if (detail?.type === 'token-refreshed') this.reconnectWithLatestToken();
if (detail?.type === 'session-cleared') this.disconnect();
});
window.addEventListener('storage', (event) => {
if (event.key === 'access_token') this.reconnectWithLatestToken();
if (typeof window === 'undefined') return;
window.addEventListener(AUTH_SESSION_EVENT, (event) => {
const detail = (event as CustomEvent<AuthSessionEventDetail>).detail;
if (detail?.type === 'session-cleared') {
this.disconnect();
} else if (detail?.type === 'token-refreshed' && !this.ws && this.hasSubscribers()) {
this.reconnectAttempts = 0;
this.connect();
}
});
window.addEventListener('storage', (event) => {
if (event.key !== 'access_token') return;
if (!event.newValue) this.disconnect();
else if (!this.ws && this.hasSubscribers()) this.connectImmediately();
});
window.addEventListener('online', () => this.connectImmediately());
if (typeof document !== 'undefined') {
document.addEventListener('visibilitychange', () => {
if (document.visibilityState === 'visible') this.connectImmediately();
});
}
}
@ -60,7 +78,7 @@ export class WebSocketManager {
isConnected: this.ws?.readyState === WebSocket.OPEN,
isConnecting: this.isConnecting,
reconnectAttempts: this.reconnectAttempts,
maxReconnectAttempts: this.maxReconnectAttempts,
maxReconnectAttempts: RAPID_RECONNECT_LIMIT,
};
}
@ -75,20 +93,14 @@ export class WebSocketManager {
});
}
send(payload: Record<string, unknown>) {
if (this.ws?.readyState !== WebSocket.OPEN) {
return false;
}
send(payload: Record<string, unknown>): boolean {
if (this.ws?.readyState !== WebSocket.OPEN) return false;
this.ws.send(JSON.stringify(payload));
return true;
}
private syncActiveChatOnSocket() {
if (this.activeChatId === null) {
return;
}
if (this.activeChatId === null) return;
this.send({
action: 'open_chat',
chat_id: this.activeChatId,
@ -96,85 +108,72 @@ export class WebSocketManager {
});
}
private reconnectWithLatestToken() {
if (!this.hasSubscribers() || typeof window === 'undefined') return;
if (!localStorage.getItem('access_token')) {
this.disconnect();
return;
}
if (this.reconnectTimeout) {
clearTimeout(this.reconnectTimeout);
this.reconnectTimeout = null;
}
const previousSocket = this.ws;
this.ws = null;
if (previousSocket) {
previousSocket.onclose = null;
previousSocket.close();
}
this.isConnecting = false;
this.reconnectAttempts = 0;
this.shouldReconnect = true;
this.connect();
private async requestTicket(): Promise<string> {
const response = await apiClient.request<WebSocketTicketResponse>(
'/api/auth/ws-ticket/',
{ method: 'POST' },
);
if (!response.ticket) throw new Error('WebSocket ticket is missing');
return response.ticket;
}
connect() {
// Если уже подключены или подключаемся, не создаем новое соединение
if (this.ws?.readyState === WebSocket.OPEN || this.isConnecting) {
console.log('[WS Manager] Already connected or connecting');
return;
}
if (this.ws?.readyState === WebSocket.OPEN || this.isConnecting) return;
if (!this.hasSubscribers()) {
this.shouldReconnect = false;
return;
}
// Проверяем что мы в браузере и пользователь авторизован
if (typeof window === 'undefined' || !localStorage.getItem('access_token')) {
console.log('[WS Manager] Not in browser or not authenticated');
this.notifyStatus();
return;
}
if (this.reconnectTimeout) {
clearTimeout(this.reconnectTimeout);
this.reconnectTimeout = null;
}
this.clearReconnectTimer();
this.shouldReconnect = true;
this.isConnecting = true;
const generation = ++this.connectionGeneration;
this.notifyStatus();
void this.openConnection(generation);
}
private async openConnection(generation: number) {
try {
let wsUrl = getWebSocketUrl();
const ticket = await this.requestTicket();
if (
generation !== this.connectionGeneration
|| !this.shouldReconnect
|| !this.hasSubscribers()
) return;
const token = localStorage.getItem('access_token');
if (token) {
wsUrl += `${wsUrl.includes('?') ? '&' : '?'}token=${token}`;
}
this.ws = new WebSocket(wsUrl);
const baseUrl = getWebSocketUrl();
const wsUrl = `${baseUrl}${baseUrl.includes('?') ? '&' : '?'}ticket=${encodeURIComponent(ticket)}`;
const socket = new WebSocket(wsUrl);
this.ws = socket;
this.ws.onopen = () => {
console.log('[WS Manager] Connected');
socket.onopen = () => {
if (this.ws !== socket) return;
this.isConnecting = false;
this.reconnectAttempts = 0;
this.lastServerMessageAt = Date.now();
this.startHeartbeatWatchdog(socket);
this.syncActiveChatOnSocket();
this.notifyStatus();
};
this.ws.onmessage = (event) => {
socket.onmessage = (event) => {
if (this.ws !== socket) return;
this.lastServerMessageAt = Date.now();
try {
const data = JSON.parse(event.data);
// Отправляем сообщение всем подписчикам
this.messageHandlers.forEach(handler => {
const data = JSON.parse(event.data) as RealtimeEvent;
if (data.type === 'ping') {
this.send({ action: 'pong', timestamp: data.timestamp });
return;
}
this.messageHandlers.forEach((handler) => {
try {
handler(data);
} catch (err) {
console.error('[WS Manager] Handler error:', err);
} catch (error) {
console.error('[WS Manager] Handler error:', error);
}
});
} catch (error) {
@ -182,114 +181,118 @@ export class WebSocketManager {
}
};
this.ws.onerror = (error) => {
console.error('[WS Manager] Error event:', error);
console.error('[WS Manager] WebSocket state:', this.ws?.readyState);
socket.onerror = () => {
if (this.ws !== socket) return;
this.isConnecting = false;
this.notifyStatus();
};
this.ws.onclose = (event) => {
console.log('[WS Manager] Disconnected');
console.log('[WS Manager] Close event code:', event.code);
console.log('[WS Manager] Close event reason:', event.reason);
console.log('[WS Manager] Close event wasClean:', event.wasClean);
this.isConnecting = false;
socket.onclose = () => {
if (this.ws !== socket) return;
this.stopHeartbeatWatchdog();
this.ws = null;
this.isConnecting = false;
this.notifyStatus();
// Автоматический reconnect
if (this.shouldReconnect && this.hasSubscribers() && this.reconnectAttempts < this.maxReconnectAttempts) {
this.reconnectAttempts++;
this.notifyStatus();
const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
console.log(`[WS Manager] Reconnecting in ${delay}ms... (${this.reconnectAttempts}/${this.maxReconnectAttempts})`);
this.reconnectTimeout = setTimeout(() => this.connect(), delay);
}
this.scheduleReconnect();
};
} catch (error) {
console.error('[WS Manager] Connection failed:', error);
if (generation !== this.connectionGeneration) return;
console.error('[WS Manager] Connection preparation failed:', error);
this.isConnecting = false;
this.notifyStatus();
this.scheduleReconnect();
}
}
private scheduleReconnect() {
if (!this.shouldReconnect || !this.hasSubscribers() || this.reconnectTimeout) return;
this.reconnectAttempts += 1;
this.notifyStatus();
const exponentialAttempt = Math.min(this.reconnectAttempts, RAPID_RECONNECT_LIMIT);
const baseDelay = this.reconnectAttempts > RAPID_RECONNECT_LIMIT
? MAX_RECONNECT_DELAY_MS
: Math.min(1000 * 2 ** exponentialAttempt, 30_000);
const jitteredDelay = Math.round(baseDelay * (0.8 + Math.random() * 0.4));
this.reconnectTimeout = setTimeout(() => {
this.reconnectTimeout = null;
this.connect();
}, jitteredDelay);
}
private connectImmediately() {
if (!this.shouldReconnect || !this.hasSubscribers() || this.ws || this.isConnecting) return;
this.clearReconnectTimer();
this.reconnectAttempts = 0;
this.connect();
}
private startHeartbeatWatchdog(socket: WebSocket) {
this.stopHeartbeatWatchdog();
this.heartbeatTimer = setInterval(() => {
if (
this.ws === socket
&& socket.readyState === WebSocket.OPEN
&& Date.now() - this.lastServerMessageAt > HEARTBEAT_TIMEOUT_MS
) {
socket.close(4408, 'Heartbeat timeout');
}
}, HEARTBEAT_CHECK_MS);
}
private stopHeartbeatWatchdog() {
if (!this.heartbeatTimer) return;
clearInterval(this.heartbeatTimer);
this.heartbeatTimer = null;
}
private clearReconnectTimer() {
if (!this.reconnectTimeout) return;
clearTimeout(this.reconnectTimeout);
this.reconnectTimeout = null;
}
disconnect() {
this.shouldReconnect = false;
if (this.reconnectTimeout) {
clearTimeout(this.reconnectTimeout);
this.reconnectTimeout = null;
this.connectionGeneration += 1;
this.clearReconnectTimer();
this.stopHeartbeatWatchdog();
const socket = this.ws;
this.ws = null;
if (socket) {
socket.onclose = null;
socket.close(1000, 'Client disconnect');
}
if (this.ws) {
this.ws.close();
this.ws = null;
}
this.isConnecting = false;
this.reconnectAttempts = 0;
this.notifyStatus();
}
setActiveChat(chatId: number) {
if (!Number.isFinite(chatId) || chatId <= 0) {
return;
}
if (this.activeChatId === chatId) {
return;
}
if (!Number.isFinite(chatId) || chatId <= 0 || this.activeChatId === chatId) return;
const previousChatId = this.activeChatId;
this.activeChatId = chatId;
if (previousChatId !== null) {
this.send({
action: 'close_chat',
chat_id: previousChatId,
});
this.send({ action: 'close_chat', chat_id: previousChatId });
}
this.syncActiveChatOnSocket();
}
clearActiveChat(chatId?: number) {
if (this.activeChatId === null) {
return;
}
if (typeof chatId === 'number' && this.activeChatId !== chatId) {
return;
}
if (this.activeChatId === null) return;
if (typeof chatId === 'number' && this.activeChatId !== chatId) return;
const chatToClose = this.activeChatId;
this.activeChatId = null;
this.send({
action: 'close_chat',
chat_id: chatToClose,
});
this.send({ action: 'close_chat', chat_id: chatToClose });
}
subscribe(handler: MessageHandler) {
this.messageHandlers.add(handler);
this.shouldReconnect = true;
// Если есть подписчик и нет соединения, подключаемся
if (this.hasSubscribers() && !this.ws && !this.isConnecting) {
this.connect();
}
// Возвращаем функцию отписки
if (this.hasSubscribers() && !this.ws && !this.isConnecting) this.connect();
return () => {
this.messageHandlers.delete(handler);
// Если больше нет подписчиков, отключаемся
if (!this.hasSubscribers()) {
console.log('[WS Manager] No more subscribers, disconnecting');
this.disconnect();
}
if (!this.hasSubscribers()) this.disconnect();
};
}
@ -297,18 +300,10 @@ export class WebSocketManager {
this.statusHandlers.add(handler);
handler(this.getConnectionStatus());
this.shouldReconnect = true;
if (this.hasSubscribers() && !this.ws && !this.isConnecting) {
this.connect();
}
if (this.hasSubscribers() && !this.ws && !this.isConnecting) this.connect();
return () => {
this.statusHandlers.delete(handler);
if (!this.hasSubscribers()) {
console.log('[WS Manager] No more subscribers, disconnecting');
this.disconnect();
}
if (!this.hasSubscribers()) this.disconnect();
};
}

View File

@ -56,6 +56,7 @@ test("calendar service receives range snapshots through the shared WebSocket", a
const originalLocalStorage = globalThis.localStorage;
const originalWebSocket = globalThis.WebSocket;
const originalLoadEvents = calendarService.loadEvents;
const originalRequest = apiClient.request;
class FakeWebSocket {
static CONNECTING = 0;
@ -97,6 +98,10 @@ test("calendar service receives range snapshots through the shared WebSocket", a
},
};
globalThis.WebSocket = FakeWebSocket;
apiClient.request = async (endpoint) => {
assert.equal(endpoint, "/api/auth/ws-ticket/");
return { ticket: "calendar-ws-ticket", expires_in: 30 };
};
const received = [];
const loading = [];
@ -113,6 +118,8 @@ test("calendar service receives range snapshots through the shared WebSocket", a
onLoading: (value) => loading.push(value),
});
await new Promise((resolve) => setTimeout(resolve, 0));
assert.equal(FakeWebSocket.instances.length, 1);
const socket = FakeWebSocket.instances[0];
socket.open();
@ -170,5 +177,6 @@ test("calendar service receives range snapshots through the shared WebSocket", a
globalThis.window = originalWindow;
globalThis.localStorage = originalLocalStorage;
globalThis.WebSocket = originalWebSocket;
apiClient.request = originalRequest;
}
});