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 },
},
},
})