From 4753f945f1cd6046919cd2ea4e04bbd386870f3e Mon Sep 17 00:00:00 2001 From: rookit Date: Mon, 29 Jun 2026 13:46:26 +0800 Subject: [PATCH] add websocket --- backend/apps/comments/services.py | 7 + backend/apps/comments/views.py | 14 ++ backend/apps/inbox/serializers.py | 16 +- backend/apps/inbox/services.py | 7 + backend/apps/inbox/views.py | 34 +++- backend/apps/realtime/__init__.py | 0 backend/apps/realtime/apps.py | 7 + backend/apps/realtime/auth.py | 48 ++++++ backend/apps/realtime/consumers.py | 88 ++++++++++ backend/apps/realtime/events.py | 171 +++++++++++++++++++ backend/apps/realtime/groups.py | 12 ++ backend/apps/realtime/routing.py | 7 + backend/asp/asgi.py | 11 +- backend/asp/settings.py | 16 ++ backend/pyproject.toml | 2 + backend/uv.lock | 62 +++++++ deploy/asp-compose/README.md | 2 + deploy/asp-compose/README.zh.md | 2 + frontend/nginx.conf.template | 12 ++ frontend/src/App.tsx | 3 +- frontend/src/components/DiscussionThread.tsx | 42 ++++- frontend/src/components/InboxDrawer.tsx | 82 +++++++-- frontend/src/realtime.tsx | 145 ++++++++++++++++ frontend/src/realtimeContext.ts | 31 ++++ frontend/vite.config.ts | 1 + 25 files changed, 802 insertions(+), 20 deletions(-) create mode 100644 backend/apps/realtime/__init__.py create mode 100644 backend/apps/realtime/apps.py create mode 100644 backend/apps/realtime/auth.py create mode 100644 backend/apps/realtime/consumers.py create mode 100644 backend/apps/realtime/events.py create mode 100644 backend/apps/realtime/groups.py create mode 100644 backend/apps/realtime/routing.py create mode 100644 frontend/src/realtime.tsx create mode 100644 frontend/src/realtimeContext.ts diff --git a/backend/apps/comments/services.py b/backend/apps/comments/services.py index 80254d4..e1e23a9 100644 --- a/backend/apps/comments/services.py +++ b/backend/apps/comments/services.py @@ -58,5 +58,12 @@ def create_record_comment( object_id=comment.object_id, metadata={"source": "comment", "comment_id": comment.id}, ) + transaction.on_commit(lambda comment_id=comment.id: _broadcast_comment_created(comment_id)) return comment + + +def _broadcast_comment_created(comment_id): + from apps.realtime.events import broadcast_comment_created + + broadcast_comment_created(comment_id) diff --git a/backend/apps/comments/views.py b/backend/apps/comments/views.py index 804aff4..9c02d54 100644 --- a/backend/apps/comments/views.py +++ b/backend/apps/comments/views.py @@ -1,4 +1,5 @@ from django.contrib.contenttypes.models import ContentType +from django.db import transaction from rest_framework import viewsets, permissions from rest_framework.exceptions import PermissionDenied from rest_framework.response import Response @@ -36,4 +37,17 @@ class CommentViewSet(viewsets.ModelViewSet): def perform_destroy(self, instance): if instance.author_id != self.request.user.id: raise PermissionDenied("You can only delete your own comments.") + comment_id = instance.id + content_type = instance.content_type.model + object_id = instance.object_id + actor_id = self.request.user.id instance.delete() + transaction.on_commit( + lambda: _broadcast_comment_deleted(comment_id, content_type, object_id, actor_id=actor_id) + ) + + +def _broadcast_comment_deleted(comment_id, content_type, object_id, *, actor_id): + from apps.realtime.events import broadcast_comment_deleted + + broadcast_comment_deleted(comment_id, content_type, object_id, actor_id=actor_id) diff --git a/backend/apps/inbox/serializers.py b/backend/apps/inbox/serializers.py index 0735076..654888e 100644 --- a/backend/apps/inbox/serializers.py +++ b/backend/apps/inbox/serializers.py @@ -237,8 +237,20 @@ class InboxReplySerializer(serializers.Serializer): def mark_message_read(message, user): - InboxMessageRecipient.objects.filter( + read_at = timezone.now() + updated = InboxMessageRecipient.objects.filter( message=message, user=user, read_at__isnull=True, - ).update(read_at=timezone.now()) + ).update(read_at=read_at) + if updated: + from django.db import transaction + from apps.realtime.events import broadcast_inbox_message_read + + transaction.on_commit( + lambda message_id=message.id, user_id=user.id, timestamp=read_at: broadcast_inbox_message_read( + message_id, + user_id, + timestamp, + ) + ) diff --git a/backend/apps/inbox/services.py b/backend/apps/inbox/services.py index a3d0931..a5f907c 100644 --- a/backend/apps/inbox/services.py +++ b/backend/apps/inbox/services.py @@ -188,9 +188,16 @@ def create_inbox_message( ]) if attachment_ids: message.attachments.set(Attachment.objects.filter(id__in=attachment_ids)) + transaction.on_commit(lambda message_id=message.id: _broadcast_inbox_message_created(message_id)) return message +def _broadcast_inbox_message_created(message_id): + from apps.realtime.events import broadcast_inbox_message_created + + broadcast_inbox_message_created(message_id) + + def send_system_message( *, recipients, diff --git a/backend/apps/inbox/views.py b/backend/apps/inbox/views.py index 8f95b92..c17cc07 100644 --- a/backend/apps/inbox/views.py +++ b/backend/apps/inbox/views.py @@ -1,3 +1,4 @@ +from django.db import transaction from django.db.models import Prefetch, Q from django.utils import timezone from rest_framework import permissions, status, viewsets @@ -75,7 +76,15 @@ class InboxMessageViewSet(viewsets.ModelViewSet): raise PermissionDenied("Viewer users cannot delete messages.") if instance.kind != InboxMessage.KIND_USER or instance.sender_id != self.request.user.id: raise PermissionDenied("You can only delete your own user messages.") + user_ids = set(instance.recipients.values_list("id", flat=True)) + if instance.sender_id: + user_ids.add(instance.sender_id) + message_id = instance.id + actor_id = self.request.user.id instance.delete() + transaction.on_commit( + lambda: _broadcast_inbox_message_deleted(message_id, list(user_ids), actor_id=actor_id) + ) @action(detail=False, methods=["get"], url_path="unread-count") def unread_count(self, request): @@ -96,10 +105,21 @@ class InboxMessageViewSet(viewsets.ModelViewSet): @action(detail=False, methods=["post"], url_path="mark-all-read") def mark_all_read(self, request): + message_ids = list( + InboxMessageRecipient.objects.filter( + user=request.user, + read_at__isnull=True, + ).values_list("message_id", flat=True) + ) + read_at = timezone.now() updated = InboxMessageRecipient.objects.filter( user=request.user, read_at__isnull=True, - ).update(read_at=timezone.now()) + ).update(read_at=read_at) + if updated: + transaction.on_commit( + lambda: _broadcast_inbox_all_read(request.user.id, message_ids, read_at) + ) return Response({"updated": updated}) @action(detail=True, methods=["post"]) @@ -112,3 +132,15 @@ class InboxMessageViewSet(viewsets.ModelViewSet): message = serializer.save() output = InboxMessageSerializer(message, context=self.get_serializer_context()) return Response(output.data, status=status.HTTP_201_CREATED) + + +def _broadcast_inbox_message_deleted(message_id, user_ids, *, actor_id): + from apps.realtime.events import broadcast_inbox_message_deleted + + broadcast_inbox_message_deleted(message_id, user_ids, actor_id=actor_id) + + +def _broadcast_inbox_all_read(user_id, message_ids, read_at): + from apps.realtime.events import broadcast_inbox_all_read + + broadcast_inbox_all_read(user_id, message_ids, read_at) diff --git a/backend/apps/realtime/__init__.py b/backend/apps/realtime/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/apps/realtime/apps.py b/backend/apps/realtime/apps.py new file mode 100644 index 0000000..4268826 --- /dev/null +++ b/backend/apps/realtime/apps.py @@ -0,0 +1,7 @@ +from django.apps import AppConfig + + +class RealtimeConfig(AppConfig): + default_auto_field = "django.db.models.BigAutoField" + name = "apps.realtime" + diff --git a/backend/apps/realtime/auth.py b/backend/apps/realtime/auth.py new file mode 100644 index 0000000..50a71b7 --- /dev/null +++ b/backend/apps/realtime/auth.py @@ -0,0 +1,48 @@ +from urllib.parse import parse_qs + +from channels.db import database_sync_to_async +from django.contrib.auth.models import AnonymousUser +from rest_framework_simplejwt.authentication import JWTAuthentication + + +class JWTAuthMiddleware: + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + scope = dict(scope) + scope["user"] = await self._authenticate(scope) + return await self.app(scope, receive, send) + + @database_sync_to_async + def _authenticate(self, scope): + token = self._token_from_scope(scope) + if not token: + return AnonymousUser() + + authenticator = JWTAuthentication() + try: + validated_token = authenticator.get_validated_token(token) + return authenticator.get_user(validated_token) + except Exception: + return AnonymousUser() + + def _token_from_scope(self, scope): + query_string = scope.get("query_string", b"").decode("utf-8") + token = parse_qs(query_string).get("token", [""])[0] + if token: + return token + + for name, value in scope.get("headers", []): + if name != b"sec-websocket-protocol": + continue + protocols = [ + part.strip() + for part in value.decode("utf-8").split(",") + if part.strip() + ] + for protocol in protocols: + if protocol.startswith("bearer."): + return protocol.removeprefix("bearer.") + return "" + diff --git a/backend/apps/realtime/consumers.py b/backend/apps/realtime/consumers.py new file mode 100644 index 0000000..d2857f9 --- /dev/null +++ b/backend/apps/realtime/consumers.py @@ -0,0 +1,88 @@ +from channels.db import database_sync_to_async +from channels.generic.websocket import AsyncJsonWebsocketConsumer +from django.contrib.contenttypes.models import ContentType +from django.core.exceptions import ValidationError + +from .groups import comments_group_name, inbox_group_name + + +class EventsConsumer(AsyncJsonWebsocketConsumer): + async def connect(self): + user = self.scope.get("user") + if not user or not user.is_authenticated: + await self.close(code=4401) + return + + self.comment_groups = set() + self.inbox_group = inbox_group_name(user.id) + await self.channel_layer.group_add(self.inbox_group, self.channel_name) + await self.accept() + await self.send_json({"type": "realtime.connected"}) + + async def disconnect(self, close_code): + if hasattr(self, "inbox_group"): + await self.channel_layer.group_discard(self.inbox_group, self.channel_name) + for group_name in getattr(self, "comment_groups", set()): + await self.channel_layer.group_discard(group_name, self.channel_name) + + async def receive_json(self, content, **kwargs): + message_type = content.get("type") + if message_type == "comments.subscribe": + await self._subscribe_comments(content) + return + if message_type == "comments.unsubscribe": + await self._unsubscribe_comments(content) + return + await self.send_json({"type": "realtime.error", "payload": {"detail": "Unknown message type."}}) + + async def _subscribe_comments(self, content): + content_type = str(content.get("content_type") or "") + object_id = str(content.get("object_id") or "") + if not await self._can_subscribe_to_record(content_type, object_id): + await self.send_json({ + "type": "realtime.error", + "payload": {"detail": "Cannot subscribe to record comments."}, + }) + return + + group_name = comments_group_name(content_type, object_id) + await self.channel_layer.group_add(group_name, self.channel_name) + self.comment_groups.add(group_name) + await self.send_json({ + "type": "comments.subscribed", + "payload": {"content_type": content_type, "object_id": object_id}, + }) + + async def _unsubscribe_comments(self, content): + content_type = str(content.get("content_type") or "") + object_id = str(content.get("object_id") or "") + group_name = comments_group_name(content_type, object_id) + if group_name in self.comment_groups: + self.comment_groups.remove(group_name) + await self.channel_layer.group_discard(group_name, self.channel_name) + await self.send_json({ + "type": "comments.unsubscribed", + "payload": {"content_type": content_type, "object_id": object_id}, + }) + + @database_sync_to_async + def _can_subscribe_to_record(self, content_type, object_id): + if not content_type or not object_id: + return False + try: + content_type_obj = ContentType.objects.get(model=content_type) + except ContentType.DoesNotExist: + return False + + model_class = content_type_obj.model_class() + if not model_class: + return False + + try: + return model_class._default_manager.filter(pk=object_id).exists() + except (ValidationError, ValueError): + return False + + async def realtime_event(self, event): + await self.send_json(event["event"]) + diff --git a/backend/apps/realtime/events.py b/backend/apps/realtime/events.py new file mode 100644 index 0000000..2e5f9c5 --- /dev/null +++ b/backend/apps/realtime/events.py @@ -0,0 +1,171 @@ +import uuid +import logging + +from asgiref.sync import async_to_sync +from channels.layers import get_channel_layer +from django.contrib.auth import get_user_model +from django.utils import timezone +from redis.exceptions import RedisError + +from .groups import comments_group_name, inbox_group_name + +User = get_user_model() +logger = logging.getLogger(__name__) + + +def _event(event_type, payload, *, actor_id=None): + return { + "type": event_type, + "event_id": str(uuid.uuid4()), + "occurred_at": timezone.now().isoformat(), + "actor_id": actor_id, + "payload": payload, + } + + +def _send_group(group_name, event): + channel_layer = get_channel_layer() + if channel_layer is None: + logger.warning("Realtime channel layer is not configured; skipped event type=%s group=%s", event.get("type"), group_name) + return + try: + async_to_sync(channel_layer.group_send)( + group_name, + { + "type": "realtime.event", + "event": event, + }, + ) + except RedisError: + logger.exception("Failed to publish realtime event type=%s group=%s", event.get("type"), group_name) + + +def _unread_count(user_id): + from apps.inbox.models import InboxMessageRecipient + + return InboxMessageRecipient.objects.filter(user_id=user_id, read_at__isnull=True).count() + + +def _send_unread_count_changed(user_id): + _send_group( + inbox_group_name(user_id), + _event( + "inbox.unread_count_changed", + {"count": _unread_count(user_id)}, + ), + ) + + +def _serialize_inbox_message(message, user): + from types import SimpleNamespace + + from apps.inbox.serializers import InboxMessageSerializer + + return InboxMessageSerializer(message, context={"request": SimpleNamespace(user=user)}).data + + +def broadcast_inbox_message_created(message_id): + from apps.inbox.models import InboxMessage + + message = ( + InboxMessage.objects + .select_related("sender", "content_type", "parent", "parent__sender") + .prefetch_related("attachments", "recipients") + .get(pk=message_id) + ) + users = list(message.recipients.all()) + if message.sender_id and all(user.id != message.sender_id for user in users): + users.append(message.sender) + + for user in users: + _send_group( + inbox_group_name(user.id), + _event( + "inbox.message_created", + {"message": _serialize_inbox_message(message, user)}, + actor_id=message.sender_id, + ), + ) + _send_unread_count_changed(user.id) + + +def broadcast_inbox_message_deleted(message_id, user_ids, *, actor_id=None): + for user_id in user_ids: + _send_group( + inbox_group_name(user_id), + _event( + "inbox.message_deleted", + {"message_id": message_id}, + actor_id=actor_id, + ), + ) + _send_unread_count_changed(user_id) + + +def broadcast_inbox_message_read(message_id, user_id, read_at): + _send_group( + inbox_group_name(user_id), + _event( + "inbox.message_read", + {"message_id": message_id, "read_at": read_at.isoformat()}, + actor_id=user_id, + ), + ) + _send_unread_count_changed(user_id) + + +def broadcast_inbox_all_read(user_id, message_ids, read_at): + _send_group( + inbox_group_name(user_id), + _event( + "inbox.all_read", + {"message_ids": message_ids, "read_at": read_at.isoformat()}, + actor_id=user_id, + ), + ) + _send_unread_count_changed(user_id) + + +def _serialize_comment(comment): + from apps.comments.serializers import CommentSerializer + + return CommentSerializer(comment).data + + +def broadcast_comment_created(comment_id): + from apps.comments.models import Comment + + comment = ( + Comment.objects + .select_related("author", "content_type", "parent", "parent__author") + .prefetch_related("mentions", "attachments") + .get(pk=comment_id) + ) + content_type = comment.content_type.model + _send_group( + comments_group_name(content_type, comment.object_id), + _event( + "comment.created", + { + "content_type": content_type, + "object_id": comment.object_id, + "comment": _serialize_comment(comment), + }, + actor_id=comment.author_id, + ), + ) + + +def broadcast_comment_deleted(comment_id, content_type, object_id, *, actor_id=None): + _send_group( + comments_group_name(content_type, object_id), + _event( + "comment.deleted", + { + "content_type": content_type, + "object_id": object_id, + "comment_id": comment_id, + }, + actor_id=actor_id, + ), + ) diff --git a/backend/apps/realtime/groups.py b/backend/apps/realtime/groups.py new file mode 100644 index 0000000..50764ed --- /dev/null +++ b/backend/apps/realtime/groups.py @@ -0,0 +1,12 @@ +import hashlib + + +def inbox_group_name(user_id): + return f"inbox.user.{user_id}" + + +def comments_group_name(content_type, object_id): + identity = f"{content_type}:{object_id}" + digest = hashlib.sha256(identity.encode("utf-8")).hexdigest() + return f"comments.record.{digest}" + diff --git a/backend/apps/realtime/routing.py b/backend/apps/realtime/routing.py new file mode 100644 index 0000000..d999dfc --- /dev/null +++ b/backend/apps/realtime/routing.py @@ -0,0 +1,7 @@ +from django.urls import path + +from .consumers import EventsConsumer + +websocket_urlpatterns = [ + path("ws/events/", EventsConsumer.as_asgi()), +] diff --git a/backend/asp/asgi.py b/backend/asp/asgi.py index ed41454..e80180f 100644 --- a/backend/asp/asgi.py +++ b/backend/asp/asgi.py @@ -8,6 +8,7 @@ import os from django.core.asgi import get_asgi_application from starlette.applications import Starlette from starlette.routing import Mount +from channels.routing import ProtocolTypeRouter, URLRouter from apps.common.logging import configure_process_file_logging @@ -17,6 +18,8 @@ django_application = get_asgi_application() configure_process_file_logging("asgi") from apps.mcp.asgi import mcp_asgi_app, mcp_server # noqa: E402 +from apps.realtime.auth import JWTAuthMiddleware # noqa: E402 +from apps.realtime.routing import websocket_urlpatterns # noqa: E402 @contextlib.asynccontextmanager @@ -25,10 +28,16 @@ async def lifespan(app): yield -application = Starlette( +http_application = Starlette( routes=[ Mount("/api/mcp", app=mcp_asgi_app), Mount("/", app=django_application), ], lifespan=lifespan, ) + +application = ProtocolTypeRouter({ + "http": http_application, + "lifespan": http_application, + "websocket": JWTAuthMiddleware(URLRouter(websocket_urlpatterns)), +}) diff --git a/backend/asp/settings.py b/backend/asp/settings.py index d324b50..ca895da 100644 --- a/backend/asp/settings.py +++ b/backend/asp/settings.py @@ -30,6 +30,7 @@ INSTALLED_APPS = [ "corsheaders", "django_filters", "storages", + "channels", # Local apps "apps.common", "apps.dashboard", @@ -45,6 +46,7 @@ INSTALLED_APPS = [ "apps.attachments", "apps.audit", "apps.inbox", + "apps.realtime", "apps.webhook", "apps.mcp", "apps.agentic", @@ -80,6 +82,7 @@ TEMPLATES = [ ] WSGI_APPLICATION = "asp.wsgi.application" +ASGI_APPLICATION = "asp.asgi.application" DATABASES = { "default": { @@ -98,6 +101,7 @@ REDIS_DB = os.environ.get("REDIS_DB", "1") REDIS_PASSWORD = os.environ.get("REDIS_PASSWORD", "") REDIS_AUTH = f":{quote(REDIS_PASSWORD, safe='')}@" if REDIS_PASSWORD else "" REDIS_URL = f"redis://{REDIS_AUTH}{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}" +CHANNEL_REDIS_SOCKET_TIMEOUT = int(os.environ.get("CHANNEL_REDIS_SOCKET_TIMEOUT", "10")) CACHES = { "default": { @@ -109,6 +113,18 @@ CACHES = { } } +CHANNEL_LAYERS = { + "default": { + "BACKEND": "channels_redis.core.RedisChannelLayer", + "CONFIG": { + "hosts": [{ + "address": REDIS_URL, + "socket_timeout": CHANNEL_REDIS_SOCKET_TIMEOUT, + }], + }, + }, +} + AUTH_USER_MODEL = "accounts.User" REST_FRAMEWORK = { diff --git a/backend/pyproject.toml b/backend/pyproject.toml index aedd6c0..99dd395 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -3,6 +3,8 @@ name = "asp" version = "0.1.0" requires-python = ">=3.14" dependencies = [ + "channels>=4.3.2", + "channels-redis>=4.3.0", "django>=6.0.6", "djangorestframework==3.17.1", "djangorestframework-simplejwt==5.5.1", diff --git a/backend/uv.lock b/backend/uv.lock index c3b280d..1e6dd00 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -46,6 +46,8 @@ name = "asp" version = "0.1.0" source = { virtual = "." } dependencies = [ + { name = "channels" }, + { name = "channels-redis" }, { name = "django" }, { name = "django-cors-headers" }, { name = "django-filter" }, @@ -71,6 +73,8 @@ dependencies = [ [package.metadata] requires-dist = [ + { name = "channels", specifier = ">=4.3.2" }, + { name = "channels-redis", specifier = ">=4.3.0" }, { name = "django", specifier = ">=6.0.6" }, { name = "django-cors-headers", specifier = "==4.9.0" }, { name = "django-filter", specifier = "==25.2" }, @@ -173,6 +177,34 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/ae/3a/dbeec9d1ee0844c679f6bb5d6ad4e9f198b1224f4e7a32825f47f6192b0c/cffi-2.0.0-cp314-cp314t-win_arm64.whl", hash = "sha256:0a1527a803f0a659de1af2e1fd700213caba79377e27e4693648c2923da066f9" }, ] +[[package]] +name = "channels" +version = "4.3.2" +source = { registry = "https://mirrors.aliyun.com/pypi/simple/" } +dependencies = [ + { name = "asgiref" }, + { name = "django" }, +] +sdist = { url = "https://mirrors.aliyun.com/pypi/packages/74/92/b18d4bb54d14986a8b35215a1c9e6a7f9f4d57ca63ac9aee8290ebb4957d/channels-4.3.2.tar.gz", hash = "sha256:f2bb6bfb73ad7fb4705041d07613c7b4e69528f01ef8cb9fb6c21d9295f15667" } +wheels = [ + { url = "https://mirrors.aliyun.com/pypi/packages/16/34/c32915288b7ef482377b6adc401192f98c6a99b3a145423d3b8aed807898/channels-4.3.2-py3-none-any.whl", hash = "sha256:fef47e9055a603900cf16cef85f050d522d9ac4b3daccf24835bd9580705c176" }, +] + +[[package]] +name = "channels-redis" +version = "4.3.0" +source = { registry = "https://mirrors.aliyun.com/pypi/simple/" } +dependencies = [ + { name = "asgiref" }, + { name = "channels" }, + { name = "msgpack" }, + { name = "redis" }, +] +sdist = { url = "https://mirrors.aliyun.com/pypi/packages/ab/69/fd3407ad407a80e72ca53850eb7a4c306273e67d5bbb71a86d0e6d088439/channels_redis-4.3.0.tar.gz", hash = "sha256:740ee7b54f0e28cf2264a940a24453d3f00526a96931f911fcb69228ef245dd2" } +wheels = [ + { url = "https://mirrors.aliyun.com/pypi/packages/df/fe/b7224a401ad227b263e5ba84753ffb5a88df048f3b15efd2797903543ce4/channels_redis-4.3.0-py3-none-any.whl", hash = "sha256:48f3e902ae2d5fef7080215524f3b4a1d3cea4e304150678f867a1a822c0d9f5" }, +] + [[package]] name = "charset-normalizer" version = "3.4.7" @@ -767,6 +799,36 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8" }, ] +[[package]] +name = "msgpack" +version = "1.2.1" +source = { registry = "https://mirrors.aliyun.com/pypi/simple/" } +sdist = { url = "https://mirrors.aliyun.com/pypi/packages/31/f9/c0a1c127f9049db9155afc316952ea571720dd01833ff5e4d7e8e6352dbb/msgpack-1.2.1.tar.gz", hash = "sha256:04c721c2c7448767e9e3f2520a475663d8ee0f09c31890f6d2bd70fd636a9647" } +wheels = [ + { url = "https://mirrors.aliyun.com/pypi/packages/77/58/cce442852c6b9e1639c7c8ac8fd9143121cb32dab0f308df4d1426a8eb9c/msgpack-1.2.1-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:05f340e47e7e47d2da8db9b53e1bb1d294369e9ef45a747441309f6650b8351d" }, + { url = "https://mirrors.aliyun.com/pypi/packages/60/5c/15b4c7a0182f75ffa90751958ba36a9c01cafee367d49a3edc10ed140b01/msgpack-1.2.1-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:810b916696c86ef0deb3b74588480224df4c1b071136c34183e4a2a4284d7ac7" }, + { url = "https://mirrors.aliyun.com/pypi/packages/b8/a6/99e58722feaffc5f2fbcc0c8c0d1451ab9f84097f7af87291b46af2390f4/msgpack-1.2.1-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ca0dacff965c47afdc3749a8469d7302a8f801d6a28758d55120d75e66ce6889" }, + { url = "https://mirrors.aliyun.com/pypi/packages/19/03/8c63e8cf52958534ef688625965ab04c269a6cadd8caef16758b380a821a/msgpack-1.2.1-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0e2bf9280bceb5efca998435904b5d3e9fdbcc11d90dc9df30aec7973252b720" }, + { url = "https://mirrors.aliyun.com/pypi/packages/63/d2/155d9e71b40e41fd934bc0c48b9b2770f22263e1ac20aad8e29fdca7be3f/msgpack-1.2.1-cp314-cp314-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:aa6c4be5d1c02a42b066ca6ddb71adf36432868fdcdb6ee87e634e86e0674190" }, + { url = "https://mirrors.aliyun.com/pypi/packages/98/48/deaf2326262a8d5ea3295ce9649912ecd3f551ba7ec8e33c665d2ba583f3/msgpack-1.2.1-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:ec0e675d59150a6269ddc9139087c722292664a37d071a849c05c473350f1f2d" }, + { url = "https://mirrors.aliyun.com/pypi/packages/10/2a/b4410f906c2ec0008f1608d3ab5143afc3ad3f4e6da0fed3ea2231d0bef4/msgpack-1.2.1-cp314-cp314-musllinux_1_2_riscv64.whl", hash = "sha256:dd3bfe82d53edfe4b7fc9a7ec9761e23a7a5b1dac22264505af428253c29ed24" }, + { url = "https://mirrors.aliyun.com/pypi/packages/59/86/1edc67270099a528fa2093ea60fe191233cd238e4bd30cfacf7db79fc959/msgpack-1.2.1-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:5ad5467fc3f68b5468e06c5f788d712e9f8ffc8b0cd1bcb160c105c1ee92dae7" }, + { url = "https://mirrors.aliyun.com/pypi/packages/82/90/8b630fef07d8c5ab457b71ff2c217910c83d333c7a68472c186e87cc504a/msgpack-1.2.1-cp314-cp314-win32.whl", hash = "sha256:98b58bdb89c46190e4609bb36abe17c6d4105ad13f9c5f8f6f64d320f8ced3fb" }, + { url = "https://mirrors.aliyun.com/pypi/packages/16/f1/467b81e98b24dd3885d7b1857728797b4ffc76a7a7483af4fb321a07de3c/msgpack-1.2.1-cp314-cp314-win_amd64.whl", hash = "sha256:74847557e28ce71bd3c438a447ca90e4b507e997ddbdef8a12a7b283b86c156b" }, + { url = "https://mirrors.aliyun.com/pypi/packages/a7/1d/5d8c4c89985feb6acefb82a09e501c60392261856d2408d20bfe4f0360b1/msgpack-1.2.1-cp314-cp314-win_arm64.whl", hash = "sha256:b50b727bd652bdc37d950336c848ef20ec54a4cafc38dce19b1cd86ad625d0f7" }, + { url = "https://mirrors.aliyun.com/pypi/packages/1b/02/ad2afb678b4de94496cd432b581759b756a92c1192d8c767edd6b132efdc/msgpack-1.2.1-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:8d00f177ca88a77c1cf848d204a38f249751650b601cb6532acc68805d8a8273" }, + { url = "https://mirrors.aliyun.com/pypi/packages/54/74/0b797484013128837f3b1cbb6cea019277c4de4e377dc512b4d9a0f92940/msgpack-1.2.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:5bb9c386f0a329c035ddbab4b72d1028bf9627add8dda41070288563d57ed1b1" }, + { url = "https://mirrors.aliyun.com/pypi/packages/a9/b4/b774d7eb95561739907fec675582f83203cf41c597a418c2589b4bfb8e9d/msgpack-1.2.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:20466cca18c49c7292a8984bc15d65857b171e7264bdcb5f96baf8be238791fc" }, + { url = "https://mirrors.aliyun.com/pypi/packages/b2/f9/3243191dc9937e00756c8bc1b0272fed8f23758e43df2a3b46f533e5090f/msgpack-1.2.1-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:196300e7e5d6e74d50f1607ab9c06c4a1484c383cd22defd727902591f7e8dde" }, + { url = "https://mirrors.aliyun.com/pypi/packages/23/c7/1693111db9944ba4ad4b67a1e788400d78a0b6af7a6523dc7e4e58f8274b/msgpack-1.2.1-cp314-cp314t-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:575957e79cd51903a4e8495a242442949641e08f1efd5197b43bebd3ea7682b4" }, + { url = "https://mirrors.aliyun.com/pypi/packages/3e/2b/92f86956a0c13e8662f7e2ad630c4eb4db07497b967589bd5245e018b2c1/msgpack-1.2.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:8c2ed1e48cc0f460bf3c7780e7137ff21a4e18433451916f2442c1b21036cd7d" }, + { url = "https://mirrors.aliyun.com/pypi/packages/da/ea/1479f72d200313a76fc2f823a79d1e07ed052ab7b8a0280640aa7b95de42/msgpack-1.2.1-cp314-cp314t-musllinux_1_2_riscv64.whl", hash = "sha256:5f6277e5f783c36786a145e0247fc189a03f35f84b251646e53592d2bc12b355" }, + { url = "https://mirrors.aliyun.com/pypi/packages/f5/4d/fa006060ffa1011d32bfae826fe766fe73e02982183601633b7121058ab3/msgpack-1.2.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:f9389552ecf4784886345ead0647e4edc96bee37cbab05b75540f542f766c48c" }, + { url = "https://mirrors.aliyun.com/pypi/packages/2f/e1/aab6c946570496b78e67804721f3d5e2d62a93081b9b37df77764ef56347/msgpack-1.2.1-cp314-cp314t-win32.whl", hash = "sha256:c1c79a604a2969a868a78b6ebd27a887e00c624f14f66b3038e0590cb23332d1" }, + { url = "https://mirrors.aliyun.com/pypi/packages/13/0a/e608956488a2af014cfe6e3d665e090b8ee42aa14b07f8f95b8880d66b09/msgpack-1.2.1-cp314-cp314t-win_amd64.whl", hash = "sha256:f12038a35fabd52e56a3547bab42401af49a45caa6dd00b34c44de235bc93ee2" }, + { url = "https://mirrors.aliyun.com/pypi/packages/d2/8a/27e2e57055176e366a46b85d02d68e7a5bcfbdd8474c9706375d965f24d3/msgpack-1.2.1-cp314-cp314t-win_arm64.whl", hash = "sha256:0adcf06ffde0777c0e1a9b771a2b1c4226ba1bbf748c8efcc02fcdeca3299107" }, +] + [[package]] name = "openai" version = "2.44.0" diff --git a/deploy/asp-compose/README.md b/deploy/asp-compose/README.md index e9056b1..68ca8fc 100644 --- a/deploy/asp-compose/README.md +++ b/deploy/asp-compose/README.md @@ -127,6 +127,8 @@ Restart only the Web/API entrypoints: docker compose restart asp-frontend asp-web asp-asgi ``` +The reverse proxy must forward `/ws/` to the ASGI service. + Restart only background workers: ```bash diff --git a/deploy/asp-compose/README.zh.md b/deploy/asp-compose/README.zh.md index e0b5278..d2c077d 100644 --- a/deploy/asp-compose/README.zh.md +++ b/deploy/asp-compose/README.zh.md @@ -127,6 +127,8 @@ docker compose restart docker compose restart asp-frontend asp-web asp-asgi ``` +反向代理需要将 `/ws/` 转发到 ASGI 服务。 + 只重启后台 Worker： ```bash diff --git a/frontend/nginx.conf.template b/frontend/nginx.conf.template index 1a606bd..cc05fa2 100644 --- a/frontend/nginx.conf.template +++ b/frontend/nginx.conf.template @@ -22,6 +22,18 @@ server { proxy_buffering off; } + location /ws/ { + proxy_pass http://asp-asgi:8001; + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "upgrade"; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_buffering off; + } + location /api/ { proxy_pass http://asp-web:8000; proxy_set_header Host $host; diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 66ef3e3..210a4e7 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -16,6 +16,7 @@ import {useAuthStore} from './stores/auth' import {hasPermission, type PermissionKey} from './utils/permissions' import {getMe} from './api/auth' import {buildLoginRedirectPath} from './utils/authRedirect' +import {RealtimeProvider} from './realtime' function ProtectedRoute({ children }: { children: React.ReactNode }) { const location = useLocation() @@ -31,7 +32,7 @@ function ProtectedRoute({ children }: { children: React.ReactNode }) { }, [setAuth, token, user?.role]) if (!token) return - return <>{children} + return {children} } function PermissionRoute({ permission, children }: { permission: PermissionKey; children: React.ReactNode }) { diff --git a/frontend/src/components/DiscussionThread.tsx b/frontend/src/components/DiscussionThread.tsx index 15a3b31..8ab0b94 100644 --- a/frontend/src/components/DiscussionThread.tsx +++ b/frontend/src/components/DiscussionThread.tsx @@ -1,4 +1,4 @@ -import {useCallback, useMemo, useState} from 'react' +import {useCallback, useEffect, useMemo, useState} from 'react' import {Alert, Button, Input, List, message, Popconfirm, Space, Tooltip} from 'antd' import {MessageOutlined, ReloadOutlined, SearchOutlined} from '@ant-design/icons' import dayjs from 'dayjs' @@ -10,6 +10,7 @@ import MessageAttachments from './MessageAttachments' import MessageComposer from './MessageComposer' import UserAvatar from './UserAvatar' import {tabularNumbersStyle, typography} from '../utils/typography' +import {useRealtime} from '../realtimeContext' dayjs.extend(relativeTime) @@ -121,6 +122,7 @@ export default function DiscussionThread({ contentType, objectId }: DiscussionTh const [replyTo, setReplyTo] = useState(null) const [hoveredCommentId, setHoveredCommentId] = useState(null) const [actionLoading, setActionLoading] = useState(false) + const {reconnectToken, subscribe, subscribeToComments} = useRealtime() const fetchCommentPage = useCallback((cursor?: string | null) => { return fetchComments(contentType, objectId, { cursor, pageSize: 20 }) @@ -128,6 +130,7 @@ export default function DiscussionThread({ contentType, objectId }: DiscussionTh const { items: comments, + setItems: setComments, loadingInitial, loadingMore, hasMore, @@ -140,6 +143,37 @@ export default function DiscussionThread({ contentType, objectId }: DiscussionTh errorMessage: 'Failed to load comments', }) + useEffect(() => { + return subscribeToComments(contentType, objectId) + }, [contentType, objectId, subscribeToComments]) + + useEffect(() => { + return subscribe((event) => { + if (event.type === 'comment.created') { + if (event.payload.content_type !== contentType || event.payload.object_id !== objectId) return + const nextComment = event.payload.comment + setComments((current) => { + if (current.some((comment) => comment.id === nextComment.id)) { + return current.map((comment) => comment.id === nextComment.id ? nextComment : comment) + } + return [...current, nextComment] + }) + return + } + if (event.type === 'comment.deleted') { + if (event.payload.content_type !== contentType || event.payload.object_id !== objectId) return + setComments((current) => current.filter((comment) => comment.id !== event.payload.comment_id)) + setReplyTo((current) => current?.id === event.payload.comment_id ? null : current) + } + }) + }, [contentType, objectId, setComments, subscribe]) + + useEffect(() => { + if (reconnectToken === 0) return + // eslint-disable-next-line react-hooks/set-state-in-effect + refreshComments() + }, [reconnectToken, refreshComments]) + const filteredComments = useMemo(() => { if (!search.trim()) return comments const keyword = search.toLowerCase() @@ -149,7 +183,7 @@ export default function DiscussionThread({ contentType, objectId }: DiscussionTh const submit = async (input: { body: string; mentionedIds: number[]; attachments: { id: number }[] }) => { setActionLoading(true) try { - await createComment({ + const createdComment = await createComment({ content_type: contentType, object_id: objectId, body: input.body, @@ -158,7 +192,7 @@ export default function DiscussionThread({ contentType, objectId }: DiscussionTh attachment_ids: input.attachments.map((attachment) => attachment.id), }) setReplyTo(null) - await refreshComments() + setComments((current) => current.some((comment) => comment.id === createdComment.id) ? current : [...current, createdComment]) message.success('Comment added') } catch { message.error('Failed to add comment') @@ -172,7 +206,7 @@ export default function DiscussionThread({ contentType, objectId }: DiscussionTh try { await deleteComment(comment.id) setReplyTo(null) - await refreshComments() + setComments((current) => current.filter((item) => item.id !== comment.id)) message.success('Comment deleted') } catch { message.error('Failed to delete comment') diff --git a/frontend/src/components/InboxDrawer.tsx b/frontend/src/components/InboxDrawer.tsx index 12020a7..9342a5c 100644 --- a/frontend/src/components/InboxDrawer.tsx +++ b/frontend/src/components/InboxDrawer.tsx @@ -21,6 +21,7 @@ import MessageAttachments from './MessageAttachments' import MessageComposer from './MessageComposer' import UserAvatar from './UserAvatar' import {tabularNumbersStyle, typography} from '../utils/typography' +import {useRealtime} from '../realtimeContext' dayjs.extend(relativeTime) @@ -28,8 +29,6 @@ interface InboxDrawerProps { onOpenResource?: (resourceKey: string, rowId: string | number) => void } -const INBOX_UNREAD_POLL_INTERVAL_MS = 60000 - function senderLabel(row: InboxMessage) { if (row.kind === 'system') return 'System' return row.sender_name || row.sender_username || 'Unknown user' @@ -138,6 +137,7 @@ export default function InboxDrawer({ onOpenResource }: InboxDrawerProps) { const [replyTo, setReplyTo] = useState(null) const [unreadCount, setUnreadCount] = useState(0) const [refreshingCount, setRefreshingCount] = useState(false) + const {reconnectToken, subscribe} = useRealtime() const fetchInboxPage = useCallback((cursor?: string | null) => ( fetchInboxMessages({ unread: filter === 'unread', cursor, pageSize: 20 }) @@ -173,9 +173,58 @@ export default function InboxDrawer({ onOpenResource }: InboxDrawerProps) { useEffect(() => { // eslint-disable-next-line react-hooks/set-state-in-effect loadUnreadCount() - const timer = window.setInterval(loadUnreadCount, INBOX_UNREAD_POLL_INTERVAL_MS) - return () => window.clearInterval(timer) - }, [loadUnreadCount]) + }, [loadUnreadCount, reconnectToken]) + + useEffect(() => { + return subscribe((event) => { + if (event.type === 'inbox.unread_count_changed') { + setUnreadCount(event.payload.count) + return + } + if (event.type === 'inbox.message_created') { + const nextMessage = event.payload.message + if (filter === 'unread' && nextMessage.is_read) return + setRows((current) => { + if (current.some((item) => item.id === nextMessage.id)) { + return current.map((item) => item.id === nextMessage.id ? nextMessage : item) + } + return [nextMessage, ...current] + }) + return + } + if (event.type === 'inbox.message_deleted') { + const deletedId = event.payload.message_id + setRows((current) => current.filter((item) => item.id !== deletedId)) + setSelected((current) => current?.id === deletedId ? null : current) + setReplyTo((current) => current?.id === deletedId ? null : current) + return + } + if (event.type === 'inbox.message_read') { + const {message_id: messageId, read_at: readAt} = event.payload + setRows((current) => { + const updated = current.map((item) => item.id === messageId ? {...item, is_read: true, read_at: readAt} : item) + return filter === 'unread' ? updated.filter((item) => item.id !== messageId) : updated + }) + setSelected((current) => current?.id === messageId ? {...current, is_read: true, read_at: readAt} : current) + return + } + if (event.type === 'inbox.all_read') { + const {read_at: readAt} = event.payload + if (filter === 'unread') { + setRows([]) + } else { + setRows((current) => current.map((item) => ({...item, is_read: true, read_at: readAt}))) + } + setSelected((current) => current ? {...current, is_read: true, read_at: readAt} : current) + } + }) + }, [filter, setRows, subscribe]) + + useEffect(() => { + if (!open || reconnectToken === 0) return + // eslint-disable-next-line react-hooks/set-state-in-effect + refreshMessages() + }, [open, reconnectToken, refreshMessages]) const canReply = (row: InboxMessage) => { return Boolean(row.kind === 'user' && row.sender && row.sender !== currentUser?.id) @@ -200,7 +249,8 @@ export default function InboxDrawer({ onOpenResource }: InboxDrawerProps) { setSelected(null) } if (replyTo?.id === row.id) setReplyTo(null) - await Promise.all([refreshMessages(), loadUnreadCount()]) + setRows((current) => current.filter((item) => item.id !== row.id)) + await loadUnreadCount() } catch { message.error('Failed to delete message') } @@ -208,13 +258,15 @@ export default function InboxDrawer({ onOpenResource }: InboxDrawerProps) { const submitMessage = async (input: { body: string; mentionedIds: number[]; attachments: { id: number }[] }) => { if (replyTo) { - await replyInboxMessage(replyTo.id, { + const sentMessage = await replyInboxMessage(replyTo.id, { body: input.body, attachments: input.attachments.map((attachment) => attachment.id), }) setReplyTo(null) + if (filter === 'all') { + setRows((current) => current.some((item) => item.id === sentMessage.id) ? current : [sentMessage, ...current]) + } message.success('Reply sent') - await Promise.all([refreshMessages(), loadUnreadCount()]) return } @@ -222,13 +274,15 @@ export default function InboxDrawer({ onOpenResource }: InboxDrawerProps) { message.warning('Mention at least one user to send a message') return } - await createInboxMessage({ + const sentMessage = await createInboxMessage({ body: input.body, recipients: input.mentionedIds, attachments: input.attachments.map((attachment) => attachment.id), }) + if (filter === 'all') { + setRows((current) => current.some((item) => item.id === sentMessage.id) ? current : [sentMessage, ...current]) + } message.success('Message sent') - await Promise.all([refreshMessages(), loadUnreadCount()]) } return ( @@ -271,7 +325,13 @@ export default function InboxDrawer({ onOpenResource }: InboxDrawerProps) { icon={} onClick={async () => { await markAllInboxMessagesRead() - await Promise.all([refreshMessages(), loadUnreadCount()]) + setUnreadCount(0) + if (filter === 'unread') { + setRows([]) + } else { + const readAt = new Date().toISOString() + setRows((current) => current.map((item) => ({...item, is_read: true, read_at: readAt}))) + } }} /> diff --git a/frontend/src/realtime.tsx b/frontend/src/realtime.tsx new file mode 100644 index 0000000..0e0b1ca --- /dev/null +++ b/frontend/src/realtime.tsx @@ -0,0 +1,145 @@ +import {useCallback, useEffect, useMemo, useRef, useState} from 'react' +import {useAuthStore} from './stores/auth' +import {RealtimeContext, type RealtimeEvent, type RealtimeStatus} from './realtimeContext' + +function websocketUrl(token: string) { + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:' + return `${protocol}//${window.location.host}/ws/events/?token=${encodeURIComponent(token)}` +} + +export function RealtimeProvider({ children }: { children: React.ReactNode }) { + const token = useAuthStore((state) => state.token) + const [status, setStatus] = useState('disconnected') + const [reconnectToken, setReconnectToken] = useState(0) + const socketRef = useRef(null) + const handlersRef = useRef(new Set<(event: RealtimeEvent) => void>()) + const commentSubscriptionsRef = useRef(new Map()) + const reconnectTimerRef = useRef(null) + const reconnectAttemptRef = useRef(0) + const shouldReconnectRef = useRef(false) + + const sendJson = useCallback((payload: unknown) => { + const socket = socketRef.current + if (socket?.readyState === WebSocket.OPEN) { + socket.send(JSON.stringify(payload)) + } + }, []) + + const sendCommentSubscription = useCallback((type: 'comments.subscribe' | 'comments.unsubscribe', contentType: string, objectId: string) => { + sendJson({ type, content_type: contentType, object_id: objectId }) + }, [sendJson]) + + useEffect(() => { + if (!token) { + shouldReconnectRef.current = false + socketRef.current?.close() + socketRef.current = null + return + } + + shouldReconnectRef.current = true + + const clearReconnectTimer = () => { + if (reconnectTimerRef.current !== null) { + window.clearTimeout(reconnectTimerRef.current) + reconnectTimerRef.current = null + } + } + + const connect = () => { + clearReconnectTimer() + setStatus('connecting') + const socket = new WebSocket(websocketUrl(token)) + socketRef.current = socket + + socket.onopen = () => { + reconnectAttemptRef.current = 0 + setStatus('connected') + setReconnectToken((current) => current + 1) + commentSubscriptionsRef.current.forEach(({ contentType, objectId }) => { + sendCommentSubscription('comments.subscribe', contentType, objectId) + }) + } + + socket.onmessage = (event) => { + let data: unknown + try { + data = JSON.parse(event.data) + } catch { + return + } + if (!data || typeof data !== 'object' || !('type' in data)) return + const realtimeEvent = data as RealtimeEvent + if (realtimeEvent.type.startsWith('realtime.') || realtimeEvent.type.startsWith('comments.subscribed')) return + handlersRef.current.forEach((handler) => handler(realtimeEvent)) + } + + socket.onclose = () => { + if (socketRef.current === socket) { + socketRef.current = null + } + setStatus('disconnected') + if (!shouldReconnectRef.current) return + const attempt = reconnectAttemptRef.current + 1 + reconnectAttemptRef.current = attempt + const delay = Math.min(30000, 1000 * 2 ** Math.min(attempt, 5)) + reconnectTimerRef.current = window.setTimeout(connect, delay) + } + + socket.onerror = () => { + socket.close() + } + } + + connect() + + return () => { + shouldReconnectRef.current = false + clearReconnectTimer() + socketRef.current?.close() + socketRef.current = null + } + }, [sendCommentSubscription, token]) + + const subscribe = useCallback((handler: (event: RealtimeEvent) => void) => { + handlersRef.current.add(handler) + return () => { + handlersRef.current.delete(handler) + } + }, []) + + const subscribeToComments = useCallback((contentType: string, objectId: string) => { + const key = `${contentType}:${objectId}` + const existing = commentSubscriptionsRef.current.get(key) + if (existing) { + existing.count += 1 + } else { + commentSubscriptionsRef.current.set(key, { contentType, objectId, count: 1 }) + sendCommentSubscription('comments.subscribe', contentType, objectId) + } + + return () => { + const current = commentSubscriptionsRef.current.get(key) + if (!current) return + if (current.count > 1) { + current.count -= 1 + return + } + commentSubscriptionsRef.current.delete(key) + sendCommentSubscription('comments.unsubscribe', contentType, objectId) + } + }, [sendCommentSubscription]) + + const value = useMemo(() => ({ + status, + reconnectToken, + subscribe, + subscribeToComments, + }), [reconnectToken, status, subscribe, subscribeToComments]) + + return ( + + {children} + + ) +} diff --git a/frontend/src/realtimeContext.ts b/frontend/src/realtimeContext.ts new file mode 100644 index 0000000..6b2eae6 --- /dev/null +++ b/frontend/src/realtimeContext.ts @@ -0,0 +1,31 @@ +import {createContext, useContext} from 'react' +import type {InboxMessage} from './api/inbox' +import type {RecordComment} from './api/comments' + +export type RealtimeStatus = 'disconnected' | 'connecting' | 'connected' + +export type RealtimeEvent = + | { type: 'inbox.message_created'; event_id: string; occurred_at: string; actor_id: number | null; payload: { message: InboxMessage } } + | { type: 'inbox.message_deleted'; event_id: string; occurred_at: string; actor_id: number | null; payload: { message_id: number } } + | { type: 'inbox.message_read'; event_id: string; occurred_at: string; actor_id: number | null; payload: { message_id: number; read_at: string } } + | { type: 'inbox.all_read'; event_id: string; occurred_at: string; actor_id: number | null; payload: { message_ids: number[]; read_at: string } } + | { type: 'inbox.unread_count_changed'; event_id: string; occurred_at: string; actor_id: number | null; payload: { count: number } } + | { type: 'comment.created'; event_id: string; occurred_at: string; actor_id: number | null; payload: { content_type: string; object_id: string; comment: RecordComment } } + | { type: 'comment.deleted'; event_id: string; occurred_at: string; actor_id: number | null; payload: { content_type: string; object_id: string; comment_id: number } } + +export interface RealtimeContextValue { + status: RealtimeStatus + reconnectToken: number + subscribe: (handler: (event: RealtimeEvent) => void) => () => void + subscribeToComments: (contentType: string, objectId: string) => () => void +} + +export const RealtimeContext = createContext(null) + +export function useRealtime() { + const context = useContext(RealtimeContext) + if (!context) { + throw new Error('useRealtime must be used within RealtimeProvider') + } + return context +} diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts index ebf7207..6e76599 100644 --- a/frontend/vite.config.ts +++ b/frontend/vite.config.ts @@ -6,6 +6,7 @@ export default defineConfig({ server: { proxy: { '/api': { target: 'http://localhost:8000', changeOrigin: true }, + '/ws': { target: 'ws://localhost:8001', ws: true }, }, }, })