Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix notification lifecycle and commerce task scheduling

Browse Source
This commit is contained in:
lizerk 2026-08-18 16:43:50 +03:00
parent 138a1284cd
commit 214323760b
20 changed files with 731 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.env.example
View File

@ -97,3 +97,7 @@ PAYROLL_SIMPLE_ADMIN_ACCESS=true
# Optional Fernet key for encrypted commerce integration credentials.
# Generate: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
COMMERCE_CREDENTIAL_KEY=
# A queued import should normally be picked up within seconds. Celery Beat marks
# lost queue messages as failed after this timeout; running imports use a longer limit.
COMMERCE_QUEUED_STALE_SECONDS=120
COMMERCE_SYNCING_STALE_MINUTES=45

244
backend/commerce/tasks/__init__.py
View File

@ -2,6 +2,8 @@ import logging
from datetime import timedelta
from celery import shared_task
from celery.utils import uuid as celery_uuid
from django.conf import settings
from django.db import transaction
from django.db.models import Q
from django.utils import timezone
@ -16,6 +18,11 @@ from commerce.services.credentials import CredentialError
logger = logging.getLogger(__name__)
QUEUE_PUBLISH_ERROR = "Не удалось поставить синхронизацию в очередь."
MISSING_RUN_ERROR = "Запуск синхронизации не найден. Повторите импорт."
STALE_QUEUED_ERROR = "Задача не была получена обработчиком и снята с очереди."
STALE_SYNCING_ERROR = "Предыдущая синхронизация была прервана."
def _next_sync_at(connection, *, now=None):
if not connection.auto_sync:
@ -23,16 +30,96 @@ def _next_sync_at(connection, *, now=None):
return (now or timezone.now()) + timedelta(minutes=connection.sync_interval_minutes)
def _safe_send_connection_update(connection, *, run=None):
"""Realtime is best-effort and must never break the synchronization itself."""
try:
send_connection_update(connection, run=run)
except Exception:
logger.exception(
"Commerce realtime update failed",
extra={
"event": "commerce.connection.realtime_update_failed",
"connection_id": str(connection.pk),
"run_id": str(run.pk) if run is not None else None,
},
)
def _mark_enqueue_failed(*, connection_id, run_id, task_id):
now = timezone.now()
with transaction.atomic():
run = (
IntegrationSyncRun.objects.select_for_update()
.filter(pk=run_id, task_id=task_id)
.first()
)
connection = (
IntegrationConnection.objects.select_for_update()
.filter(pk=connection_id, task_id=task_id)
.first()
)
if run is not None and run.status == "queued":
run.status = "error"
run.error = QUEUE_PUBLISH_ERROR
run.finished_at = now
run.save(update_fields=["status", "error", "finished_at"])
if connection is not None and connection.status == IntegrationStatus.QUEUED:
connection.status = IntegrationStatus.ERROR
connection.last_error = QUEUE_PUBLISH_ERROR
connection.next_sync_at = (
now + timedelta(minutes=5) if connection.auto_sync else None
)
connection.save(
update_fields=[
"status",
"last_error",
"next_sync_at",
"updated_at",
]
)
if connection is not None:
_safe_send_connection_update(connection, run=run)
def _publish_connection_sync(*, connection_id, run_id, task_id):
connection = IntegrationConnection.objects.get(pk=connection_id)
run = IntegrationSyncRun.objects.get(pk=run_id)
_safe_send_connection_update(connection, run=run)
try:
sync_connection_run.apply_async(
args=[str(run_id), str(connection_id)],
task_id=task_id,
)
except Exception:
_mark_enqueue_failed(
connection_id=connection_id,
run_id=run_id,
task_id=task_id,
)
logger.exception(
"Commerce synchronization could not be published",
extra={
"event": "commerce.connection.enqueue_failed",
"connection_id": str(connection_id),
"run_id": str(run_id),
"task_id": task_id,
},
)
raise
def enqueue_connection_sync(connection):
"""Create and publish one run while preventing overlapping scans."""
task_id = celery_uuid()
with transaction.atomic():
locked = IntegrationConnection.objects.select_for_update().get(pk=connection.pk)
if locked.status in {IntegrationStatus.QUEUED, IntegrationStatus.SYNCING}:
return None
run = IntegrationSyncRun.objects.create(connection=locked)
run = IntegrationSyncRun.objects.create(connection=locked, task_id=task_id)
locked.status = IntegrationStatus.QUEUED
locked.task_id = ""
locked.task_id = task_id
locked.last_error = ""
locked.next_sync_at = _next_sync_at(locked)
locked.save(
@ -44,29 +131,13 @@ def enqueue_connection_sync(connection):
"updated_at",
]
)
send_connection_update(locked, run=run)
try:
async_result = sync_moysklad_run.delay(str(run.pk))
except Exception:
retry_at = timezone.now() + timedelta(minutes=5)
IntegrationSyncRun.objects.filter(pk=run.pk).update(
status="error",
error="Не удалось поставить синхронизацию в очередь.",
finished_at=timezone.now(),
transaction.on_commit(
lambda: _publish_connection_sync(
connection_id=locked.pk,
run_id=run.pk,
task_id=task_id,
)
)
IntegrationConnection.objects.filter(pk=connection.pk).update(
status=IntegrationStatus.ERROR,
last_error="Не удалось поставить синхронизацию в очередь.",
next_sync_at=retry_at,
)
connection.refresh_from_db()
run.refresh_from_db()
send_connection_update(connection, run=run)
raise
task_id = async_result.id or ""
IntegrationSyncRun.objects.filter(pk=run.pk).update(task_id=task_id)
IntegrationConnection.objects.filter(pk=connection.pk).update(task_id=task_id)
run.task_id = task_id
return run
@ -81,8 +152,43 @@ def sync_source(source_code: str, entity_type: str):
@shared_task(bind=True, name="commerce.sync_connection_run")
def sync_connection_run(self, run_id: str):
run = IntegrationSyncRun.objects.select_related("connection__source").get(pk=run_id)
def sync_connection_run(self, run_id: str, connection_id: str | None = None):
try:
run = IntegrationSyncRun.objects.select_related("connection__source").get(
pk=run_id
)
except IntegrationSyncRun.DoesNotExist:
task_id = self.request.id or ""
connection = None
if connection_id is not None:
connection = IntegrationConnection.objects.filter(pk=connection_id).first()
if connection is not None and (not task_id or connection.task_id == task_id):
finished_at = timezone.now()
connection.status = IntegrationStatus.ERROR
connection.last_error = MISSING_RUN_ERROR
connection.next_sync_at = (
finished_at + timedelta(minutes=5) if connection.auto_sync else None
)
connection.save(
update_fields=[
"status",
"last_error",
"next_sync_at",
"updated_at",
]
)
_safe_send_connection_update(connection)
logger.error(
"Commerce synchronization run does not exist",
extra={
"event": "commerce.connection.run_missing",
"connection_id": connection_id,
"run_id": run_id,
"task_id": task_id,
},
)
return {"status": "missing", "run_id": run_id}
connection = run.connection
started_at = timezone.now()
run.status = "running"
@ -96,7 +202,7 @@ def sync_connection_run(self, run_id: str):
connection.save(
update_fields=["status", "last_sync_started_at", "last_error", "updated_at"]
)
send_connection_update(connection, run=run)
_safe_send_connection_update(connection, run=run)
def report_progress(current_stats):
now = timezone.now()
@ -113,7 +219,7 @@ def sync_connection_run(self, run_id: str):
IntegrationConnection.objects.filter(pk=connection.pk).update(updated_at=now)
run.refresh_from_db()
connection.refresh_from_db()
send_connection_update(connection, run=run)
_safe_send_connection_update(connection, run=run)
try:
if connection.provider == "wildberries":
@ -156,7 +262,7 @@ def sync_connection_run(self, run_id: str):
"updated_at",
]
)
send_connection_update(connection, run=run)
_safe_send_connection_update(connection, run=run)
raise
finished_at = timezone.now()
@ -199,7 +305,7 @@ def sync_connection_run(self, run_id: str):
"updated_at",
]
)
send_connection_update(connection, run=run)
_safe_send_connection_update(connection, run=run)
return stats
@ -208,20 +314,76 @@ def sync_connection_run(self, run_id: str):
sync_moysklad_run = sync_connection_run
def _recover_stale_connections(*, now):
queued_cutoff = now - timedelta(
seconds=getattr(settings, "COMMERCE_QUEUED_STALE_SECONDS", 120)
)
syncing_cutoff = now - timedelta(
minutes=getattr(settings, "COMMERCE_SYNCING_STALE_MINUTES", 45)
)
recovered = 0
recovery_rules = (
(IntegrationStatus.QUEUED, queued_cutoff, ("queued",), STALE_QUEUED_ERROR),
(
IntegrationStatus.SYNCING,
syncing_cutoff,
("queued", "running"),
STALE_SYNCING_ERROR,
),
)
for connection_status, cutoff, run_statuses, message in recovery_rules:
candidate_ids = list(
IntegrationConnection.objects.filter(
status=connection_status,
updated_at__lte=cutoff,
).values_list("pk", flat=True)
)
for connection_id in candidate_ids:
with transaction.atomic():
connection = IntegrationConnection.objects.select_for_update().get(
pk=connection_id
)
if connection.status != connection_status or connection.updated_at > cutoff:
continue
run = (
connection.sync_runs.select_for_update()
.filter(status__in=run_statuses)
.order_by("-created_at")
.first()
)
if run is not None:
run.status = "error"
run.error = message
run.finished_at = now
run.save(update_fields=["status", "error", "finished_at"])
connection.status = IntegrationStatus.ERROR
connection.last_error = message
connection.next_sync_at = now if connection.auto_sync else None
connection.save(
update_fields=[
"status",
"last_error",
"next_sync_at",
"updated_at",
]
)
_safe_send_connection_update(connection, run=run)
recovered += 1
if recovered:
logger.warning(
"Recovered stale commerce synchronizations",
extra={
"event": "commerce.connection.stale_runs_recovered",
"recovered": recovered,
},
)
return recovered
@shared_task(name="commerce.dispatch_due_connection_syncs")
def dispatch_due_connection_syncs():
now = timezone.now()
stale_cutoff = now - timedelta(minutes=45)
IntegrationConnection.objects.filter(
auto_sync=True,
status__in=[IntegrationStatus.QUEUED, IntegrationStatus.SYNCING],
updated_at__lte=stale_cutoff,
).update(
status=IntegrationStatus.ERROR,
last_error="Предыдущая синхронизация была прервана и будет перезапущена.",
next_sync_at=now,
updated_at=now,
)
_recover_stale_connections(now=now)
due_ids = list(
IntegrationConnection.objects.filter(
auto_sync=True,

19
backend/communications/consumers.py
View File

@ -34,6 +34,7 @@ from .mentions import (
update_unread_mention_counters,
)
from .polls import PollActionError, cast_poll_vote
from .presence import clear_active_chat, set_active_chat, touch_active_chat
from .serialization import serialize_message, serialize_poll
from .services import soft_delete_message
from .utils import user_can_access_chat
@ -84,6 +85,22 @@ class ChatConsumerMixin:
# ==================== Управление активным чатом ====================
@database_sync_to_async
def _set_active_chat_presence(self, chat_id):
set_active_chat(
user_id=self.user.id,
chat_id=chat_id,
connection_id=self.channel_name,
)
@database_sync_to_async
def _touch_active_chat_presence(self):
touch_active_chat(connection_id=self.channel_name)
@database_sync_to_async
def _clear_active_chat_presence(self):
clear_active_chat(connection_id=self.channel_name)
async def _handle_open_chat(self, content):
"""Открытие чата.
@ -105,6 +122,7 @@ class ChatConsumerMixin:
# Устанавливаем активный чат
self.active_chat_id = chat_id
await self._set_active_chat_presence(chat_id)
# Отправляем начальную историю (если запрошена)
if content.get("load_history", False):
@ -150,6 +168,7 @@ class ChatConsumerMixin:
await self._set_typing_status(chat, self.user, False)
self.active_chat_id = None
await self._clear_active_chat_presence()
await self.send_json({"type": "chat_closed", "chat_id": chat_id})

53
backend/communications/migrations/0043_activechatsession.py Normal file
View File

@ -0,0 +1,53 @@
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
("communications", "0042_chatusersettings_notification_preferences"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name="ActiveChatSession",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("connection_id", models.CharField(max_length=255, unique=True)),
("last_seen_at", models.DateTimeField(auto_now=True)),
("expires_at", models.DateTimeField(db_index=True)),
(
"chat",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="active_sessions",
to="communications.chat",
),
),
(
"user",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="active_chat_sessions",
to=settings.AUTH_USER_MODEL,
),
),
],
),
migrations.AddIndex(
model_name="activechatsession",
index=models.Index(
fields=["user", "chat", "expires_at"],
name="communicati_user_id_412dae_idx",
),
),
]

24
backend/communications/models.py
View File

@ -89,6 +89,30 @@ class ChatReadState(models.Model):
return self.updated_at
class ActiveChatSession(models.Model):
"""Cross-worker presence for a user currently viewing a chat."""
connection_id = models.CharField(max_length=255, unique=True)
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name="active_chat_sessions",
)
chat = models.ForeignKey(
"Chat",
on_delete=models.CASCADE,
related_name="active_sessions",
)
last_seen_at = models.DateTimeField(auto_now=True)
expires_at = models.DateTimeField(db_index=True)
class Meta:
indexes = [models.Index(fields=["user", "chat", "expires_at"])]
def __str__(self):
return f"{self.user_id} @ {self.chat_id} ({self.connection_id})"
class Chat(models.Model):
CHAT_TYPE_CHOICES = [
("private", "Личный"),

42
backend/communications/presence.py Normal file
View File

@ -0,0 +1,42 @@
from datetime import timedelta
from django.utils import timezone
from communications.models import ActiveChatSession
ACTIVE_CHAT_SESSION_TTL = timedelta(seconds=75)
def set_active_chat(*, user_id: int, chat_id: int, connection_id: str) -> None:
now = timezone.now()
ActiveChatSession.objects.update_or_create(
connection_id=connection_id,
defaults={
"user_id": user_id,
"chat_id": chat_id,
"expires_at": now + ACTIVE_CHAT_SESSION_TTL,
},
)
def touch_active_chat(*, connection_id: str) -> None:
now = timezone.now()
ActiveChatSession.objects.filter(connection_id=connection_id).update(
last_seen_at=now,
expires_at=now + ACTIVE_CHAT_SESSION_TTL,
)
def clear_active_chat(*, connection_id: str) -> None:
ActiveChatSession.objects.filter(connection_id=connection_id).delete()
def user_is_viewing_chat(*, user_id: int, chat_id: int) -> bool:
now = timezone.now()
queryset = ActiveChatSession.objects.filter(
user_id=user_id,
chat_id=chat_id,
)
queryset.filter(expires_at__lte=now).delete()
return queryset.filter(expires_at__gt=now).exists()

6
backend/communications/services/chat_notifications.py
View File

@ -48,7 +48,11 @@ class ChatNotificationReadService:
def notify_after_commit():
invalidate_unread_summary(user.id)
send_notifications_read_all_event(user.id, notification_ids)
dismiss_chat_push(user_id=user.id, chat_id=chat_id)
dismiss_chat_push(
user_id=user.id,
chat_id=chat_id,
notification_ids=notification_ids,
)
transaction.on_commit(notify_after_commit)

6
backend/eusrr_backend/settings.py
View File

@ -754,6 +754,12 @@ CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60 # Мягкий лимит - 25 минут
# Integration credentials are encrypted at rest. Set an explicit Fernet key in
# production; when omitted, a stable key is derived from DJANGO_SECRET_KEY.
COMMERCE_CREDENTIAL_KEY = os.getenv("COMMERCE_CREDENTIAL_KEY", "")
COMMERCE_QUEUED_STALE_SECONDS = int(
os.getenv("COMMERCE_QUEUED_STALE_SECONDS", "120")
)
COMMERCE_SYNCING_STALE_MINUTES = int(
os.getenv("COMMERCE_SYNCING_STALE_MINUTES", "45")
)
MOYSKLAD_API_BASE_URL = os.getenv(
"MOYSKLAD_API_BASE_URL", "https://api.moysklad.ru/api/remap/1.2"
)

37
backend/notifications/chat_suppression.py Normal file
View File

@ -0,0 +1,37 @@
"""Suppress chat alerts when the recipient is already viewing that chat."""
from communications.presence import user_is_viewing_chat
from communications.services import ChatNotificationReadService
CHAT_NOTIFICATION_VERBS = {
"chat_new_message",
"chat_mention",
"chat_reply",
"announcement_new_message",
}
def notification_chat_id(notification) -> int | None:
if notification.verb not in CHAT_NOTIFICATION_VERBS:
return None
data = notification.data if isinstance(notification.data, dict) else {}
try:
return int(data.get("chat_id"))
except (TypeError, ValueError):
return None
def suppress_for_active_chat(notification) -> bool:
chat_id = notification_chat_id(notification)
if chat_id is None or not user_is_viewing_chat(
user_id=notification.recipient_id,
chat_id=chat_id,
):
return False
ChatNotificationReadService().mark_chat_notifications_as_read(
user=notification.recipient,
chat_id=chat_id,
)
return True

34
backend/notifications/push_lifecycle.py
View File

@ -10,6 +10,9 @@ from notifications.models import Notification
from notifications.realtime import send_notifications_deleted_event
PUSH_CONTROL_ID_BATCH_SIZE = 150
def queue_push_control(
*,
user_id: int,
@ -30,23 +33,36 @@ def queue_push_control(
transaction.on_commit(enqueue)
def dismiss_chat_push(*, user_id: int, chat_id: int) -> None:
def dismiss_chat_push(
*,
user_id: int,
chat_id: int,
notification_ids=None,
) -> None:
scope = {"chat_id": chat_id}
if notification_ids:
scope["notification_ids"] = [
int(item) for item in list(notification_ids)[:PUSH_CONTROL_ID_BATCH_SIZE]
]
queue_push_control(
user_id=user_id,
action="dismiss_chat",
scope={"chat_id": chat_id},
scope=scope,
)
def dismiss_notifications_push(*, user_id: int, notification_ids) -> None:
normalized_ids = [int(notification_id) for notification_id in notification_ids]
if not normalized_ids:
return
queue_push_control(
user_id=user_id,
action="dismiss_notifications",
scope={"notification_ids": normalized_ids},
)
for start in range(0, len(normalized_ids), PUSH_CONTROL_ID_BATCH_SIZE):
queue_push_control(
user_id=user_id,
action="dismiss_notifications",
scope={
"notification_ids": normalized_ids[
start : start + PUSH_CONTROL_ID_BATCH_SIZE
]
},
)
def delete_message_notifications(*, chat_id: int, message_id: int) -> None:

33
backend/notifications/senders/push.py
View File

@ -13,6 +13,7 @@ from .base import BaseNotificationSender
from ..config import get
from ..models import NotificationDeliveryRouting, NotificationInstallation
from ..routing import APP_MAIN, APP_MESSENGER, normalize_routes, notification_category
from notifications.chat_suppression import suppress_for_active_chat
class PushNotificationSender(BaseNotificationSender):
@ -169,10 +170,9 @@ class PushNotificationSender(BaseNotificationSender):
return any(marker in error_text for marker in permanent_markers)
def _deactivate_device(self, device, reason: str) -> None:
if not device.active:
return
device.active = False
device.save(update_fields=["active"])
if device.active:
device.active = False
device.save(update_fields=["active"])
NotificationInstallation.objects.filter(
push_device_id=device.id,
).update(endpoint=None, push_device_id=None)
@ -185,6 +185,12 @@ class PushNotificationSender(BaseNotificationSender):
},
)
@staticmethod
def _push_result_succeeded(result) -> bool:
if not isinstance(result, dict):
return True
return bool(result.get("success", 0)) and not result.get("failure", 0)
def _delivery_devices(self, notification):
user = notification.recipient
active_devices = list(WebPushDevice.objects.filter(user=user, active=True))
@ -271,6 +277,9 @@ class PushNotificationSender(BaseNotificationSender):
if notification.deleted or not notification.unread:
self.log_skip(notification, "notification no longer active")
return False
if suppress_for_active_chat(notification):
self.log_skip(notification, "recipient is viewing chat")
return False
chat_preferences = self._chat_preferences(notification)
if (
chat_preferences is not None
@ -298,7 +307,21 @@ class PushNotificationSender(BaseNotificationSender):
def send_to_device(device):
nonlocal sent_count, failed_count
try:
device.send_message(message)
result = device.send_message(message)
if not self._push_result_succeeded(result):
failed_count += 1
device.refresh_from_db(fields=["active"])
if not device.active:
self._deactivate_device(device, "provider rejected subscription")
self.logger.warning(
"Push provider rejected delivery",
extra={
"event": "notification.push.provider_rejected",
"device_id": device.id,
"notification_id": notification.id,
},
)
return False
sent_count += 1
sent_device_ids.add(device.id)
self.logger.debug(

19
backend/notifications/senders/push_control.py
View File

@ -44,7 +44,24 @@ class PushControlSender:
push_sender = PushNotificationSender()
for device in WebPushDevice.objects.filter(user_id=user_id, active=True):
try:
device.send_message(message)
result = device.send_message(message)
if not push_sender._push_result_succeeded(result):
device.refresh_from_db(fields=["active"])
if not device.active:
push_sender._deactivate_device(
device,
"provider rejected control command",
)
logger.warning(
"Push provider rejected control command",
extra={
"event": "notification.push.control_rejected",
"user_id": user_id,
"device_id": device.id,
"action": action,
},
)
continue
delivered = True
except Exception as error: # noqa: BLE001 - push clients raise vendor exceptions
logger.warning(

4
backend/notifications/senders/websocket.py
View File

@ -5,6 +5,7 @@ from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
from .base import BaseNotificationSender
from notifications.chat_suppression import suppress_for_active_chat
class WebSocketNotificationSender(BaseNotificationSender):
@ -33,6 +34,9 @@ class WebSocketNotificationSender(BaseNotificationSender):
True если отправлено успешно, False иначе
"""
try:
if suppress_for_active_chat(notification):
self.log_skip(notification, "recipient is viewing chat")
return False
channel_layer = get_channel_layer()
if not channel_layer:
self.log_skip(notification, "channel_layer not configured")

4
backend/realtime/consumers.py
View File

@ -198,6 +198,8 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
chat = await self._get_chat(self.active_chat_id)
if chat:
await self._set_typing_status(chat, self.user, False)
if self.user:
await self._clear_active_chat_presence()
# Отписываемся от всех групп чатов
for chat_id in self.subscribed_chats:
@ -351,6 +353,8 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer):
async def _handle_pong(self, content):
self.last_pong_at = monotonic()
if self.active_chat_id is not None:
await self._touch_active_chat_presence()
@database_sync_to_async
def _build_calendar_snapshot(self, subscription):

73
backend/tests/api/v1/notifications/test_push_lifecycle.py
View File

@ -6,12 +6,16 @@ import pytest
from django.utils import timezone
from communications.models import Chat, ChatUserSettings
from communications.presence import set_active_chat, user_is_viewing_chat
from communications.services.chat_notifications import ChatNotificationReadService
from notifications.models import Notification, UserChannelPreferences
from notifications.push_lifecycle import (
delete_message_notifications,
dismiss_notifications_push,
replace_message_notifications,
)
from notifications.senders.push import PushNotificationSender
from notifications.senders.websocket import WebSocketNotificationSender
pytestmark = pytest.mark.django_db(transaction=True)
@ -93,7 +97,11 @@ def test_reading_chat_schedules_native_notification_cleanup(user_factory):
notification.refresh_from_db()
assert result.notification_ids == [notification.id]
assert notification.unread is False
dismiss.assert_called_once_with(user_id=recipient.id, chat_id=chat.id)
dismiss.assert_called_once_with(
user_id=recipient.id,
chat_id=chat.id,
notification_ids=[notification.id],
)
def test_chat_policy_supports_mentions_full_mute_and_temporary_mute(user_factory):
@ -119,3 +127,66 @@ def test_chat_policy_supports_mentions_full_mute_and_temporary_mute(user_factory
settings.notifications_enabled = False
settings.muted_until = timezone.now() - timedelta(seconds=1)
assert settings.allows_notification("chat_new_message") is True
def test_active_chat_on_another_connection_suppresses_all_realtime_alerts(user_factory):
recipient = user_factory(email="presence-recipient@example.com")
chat = Chat.objects.create(type="private", created_by=recipient)
set_active_chat(
user_id=recipient.id,
chat_id=chat.id,
connection_id="specific.channel!device-one",
)
notification = _notification(
recipient,
chat_id=chat.id,
message_id=44,
)
with (
patch("push_notifications.models.WebPushDevice.send_message") as push_send,
patch("notifications.senders.websocket.async_to_sync") as websocket_send,
patch("notifications.tasks.send_push_control.delay"),
):
assert PushNotificationSender().send(notification) is False
assert WebSocketNotificationSender().send(notification) is False
notification.refresh_from_db()
assert notification.unread is False
push_send.assert_not_called()
websocket_send.assert_not_called()
def test_expired_active_chat_presence_is_not_considered_online(user_factory):
recipient = user_factory(email="expired-presence@example.com")
chat = Chat.objects.create(type="private", created_by=recipient)
set_active_chat(
user_id=recipient.id,
chat_id=chat.id,
connection_id="specific.channel!expired",
)
from communications.models import ActiveChatSession
ActiveChatSession.objects.filter(connection_id="specific.channel!expired").update(
expires_at=timezone.now() - timedelta(seconds=1),
)
assert user_is_viewing_chat(user_id=recipient.id, chat_id=chat.id) is False
assert ActiveChatSession.objects.filter(
connection_id="specific.channel!expired"
).exists() is False
def test_large_cleanup_is_split_into_web_push_sized_batches():
with patch("notifications.tasks.send_push_control.delay") as delay:
dismiss_notifications_push(
user_id=91,
notification_ids=range(1, 352),
)
assert delay.call_count == 3
assert [len(call.args[2]["notification_ids"]) for call in delay.call_args_list] == [
150,
150,
51,
]

168
backend/tests/commerce/test_connections.py
View File

@ -1,11 +1,13 @@
from datetime import timedelta
from decimal import Decimal
import io
import uuid
from unittest.mock import Mock, patch
import httpx
import pytest
from PIL import Image
from django.db import transaction
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
@ -13,6 +15,7 @@ from rest_framework import status
from commerce.models import (
IntegrationConnection,
IntegrationStatus,
IntegrationSyncRun,
Product,
ProductBinding,
ProductImage,
@ -40,7 +43,12 @@ from commerce.services.moysklad_sync import (
normalize_moysklad_stock_row,
reconcile_moysklad_variant_parents,
)
from commerce.tasks import dispatch_due_connection_syncs, sync_moysklad_run
from commerce.tasks import (
STALE_QUEUED_ERROR,
dispatch_due_connection_syncs,
enqueue_connection_sync,
sync_moysklad_run,
)
pytestmark = pytest.mark.django_db
@ -385,7 +393,11 @@ def test_connection_test_action_updates_account(api_client, user_factory):
assert response.data["account_name"] == "Иван Иванов"
def test_connection_sync_action_queues_background_run(api_client, user_factory):
def test_connection_sync_action_queues_background_run(
api_client,
user_factory,
django_capture_on_commit_callbacks,
):
user = user_factory(staff=True)
api_client.force_authenticate(user=user)
create_response = api_client.post(
@ -393,9 +405,11 @@ def test_connection_sync_action_queues_background_run(api_client, user_factory):
{"provider": "moysklad", "auth_type": "token", "token": "secret-token"},
format="json",
)
async_result = Mock(id="celery-task-1")
with patch("commerce.tasks.sync_moysklad_run.delay", return_value=async_result):
with (
patch("commerce.tasks.celery_uuid", return_value="celery-task-1"),
patch("commerce.tasks.sync_connection_run.apply_async") as apply_async,
django_capture_on_commit_callbacks(execute=True),
):
response = api_client.post(
reverse(
"api:v1:commerce:connection-sync",
@ -406,8 +420,113 @@ def test_connection_sync_action_queues_background_run(api_client, user_factory):
assert response.status_code == status.HTTP_202_ACCEPTED
assert response.data["status"] == "queued"
connection = IntegrationConnection.objects.get()
run = IntegrationSyncRun.objects.get()
assert connection.status == IntegrationStatus.QUEUED
assert connection.task_id == "celery-task-1"
assert run.task_id == "celery-task-1"
apply_async.assert_called_once_with(
args=[str(run.pk), str(connection.pk)],
task_id="celery-task-1",
)
@pytest.mark.django_db(transaction=True)
def test_connection_sync_is_published_only_after_outer_transaction_commit():
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
connection = IntegrationConnection.objects.create(
source=source,
provider="moysklad",
encrypted_credentials="unused",
status=IntegrationStatus.READY,
)
publication_checks = []
def verify_committed(*, args, task_id):
publication_checks.append(
{
"in_atomic_block": transaction.get_connection().in_atomic_block,
"run_exists": IntegrationSyncRun.objects.filter(
pk=args[0], task_id=task_id
).exists(),
"connection_has_task": IntegrationConnection.objects.filter(
pk=args[1], task_id=task_id
).exists(),
}
)
with (
patch("commerce.tasks.celery_uuid", return_value="commit-safe-task"),
patch(
"commerce.tasks.sync_connection_run.apply_async",
side_effect=verify_committed,
) as apply_async,
):
with transaction.atomic():
run = enqueue_connection_sync(connection)
assert run.task_id == "commit-safe-task"
assert apply_async.call_count == 0
assert apply_async.call_count == 1
assert publication_checks == [
{
"in_atomic_block": False,
"run_exists": True,
"connection_has_task": True,
}
]
def test_enqueue_failure_transitions_run_and_connection_to_error(
django_capture_on_commit_callbacks,
):
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
connection = IntegrationConnection.objects.create(
source=source,
provider="moysklad",
encrypted_credentials="unused",
status=IntegrationStatus.READY,
)
with (
patch("commerce.tasks.celery_uuid", return_value="failed-task"),
patch(
"commerce.tasks.sync_connection_run.apply_async",
side_effect=RuntimeError("broker unavailable"),
),
patch("commerce.tasks.send_connection_update"),
pytest.raises(RuntimeError, match="broker unavailable"),
django_capture_on_commit_callbacks(execute=True),
):
enqueue_connection_sync(connection)
connection.refresh_from_db()
run = IntegrationSyncRun.objects.get()
assert connection.status == IntegrationStatus.ERROR
assert run.status == "error"
assert run.finished_at is not None
def test_missing_sync_run_does_not_leave_connection_queued():
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
connection = IntegrationConnection.objects.create(
source=source,
provider="moysklad",
encrypted_credentials="unused",
status=IntegrationStatus.QUEUED,
task_id="missing-run-task",
)
with patch("commerce.tasks.send_connection_update"):
result = sync_moysklad_run.apply(
args=[str(uuid.uuid4()), str(connection.pk)],
task_id="missing-run-task",
).get()
connection.refresh_from_db()
assert result["status"] == "missing"
assert connection.status == IntegrationStatus.ERROR
assert connection.last_error
def test_sync_run_persists_product_image_progress():
@ -447,7 +566,7 @@ def test_sync_run_persists_product_image_progress():
assert connection.last_stats["images"] == 12
def test_beat_dispatcher_queues_due_connection():
def test_beat_dispatcher_queues_due_connection(django_capture_on_commit_callbacks):
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
connection = IntegrationConnection.objects.create(
source=source,
@ -458,9 +577,11 @@ def test_beat_dispatcher_queues_due_connection():
sync_interval_minutes=15,
next_sync_at=timezone.now() - timedelta(minutes=1),
)
async_result = Mock(id="scheduled-task-1")
with patch("commerce.tasks.sync_moysklad_run.delay", return_value=async_result):
with (
patch("commerce.tasks.celery_uuid", return_value="scheduled-task-1"),
patch("commerce.tasks.sync_connection_run.apply_async") as apply_async,
django_capture_on_commit_callbacks(execute=True),
):
result = dispatch_due_connection_syncs()
connection.refresh_from_db()
@ -468,6 +589,35 @@ def test_beat_dispatcher_queues_due_connection():
assert connection.status == IntegrationStatus.QUEUED
assert connection.task_id == "scheduled-task-1"
assert connection.next_sync_at > timezone.now()
apply_async.assert_called_once()
def test_dispatcher_recovers_old_queued_run_before_syncing_timeout(settings):
settings.COMMERCE_QUEUED_STALE_SECONDS = 60
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
connection = IntegrationConnection.objects.create(
source=source,
provider="moysklad",
encrypted_credentials="unused",
status=IntegrationStatus.QUEUED,
task_id="lost-task",
auto_sync=False,
)
run = connection.sync_runs.create(status="queued", task_id="lost-task")
stale_at = timezone.now() - timedelta(minutes=2)
IntegrationConnection.objects.filter(pk=connection.pk).update(updated_at=stale_at)
with patch("commerce.tasks.send_connection_update"):
result = dispatch_due_connection_syncs()
connection.refresh_from_db()
run.refresh_from_db()
assert result == {"queued": 0}
assert connection.status == IntegrationStatus.ERROR
assert connection.last_error == STALE_QUEUED_ERROR
assert run.status == "error"
assert run.error == STALE_QUEUED_ERROR
assert run.finished_at is not None
def test_schedule_update_does_not_disconnect_connection(api_client, user_factory):

23
backend/tests/realtime/test_user_consumer.py
View File

@ -14,7 +14,7 @@ from channels.layers import get_channel_layer
from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser
from communications.models import Chat, ChatMembership
from communications.models import ActiveChatSession, Chat, ChatMembership
from commerce.realtime import COMMERCE_CONNECTIONS_GROUP
from notifications.models import Notification
@ -160,8 +160,22 @@ class TestUserConsumerChatManagement:
assert response["type"] == "chat_opened"
assert response["chat_id"] == test_chat.id
presence_exists = await database_sync_to_async(
ActiveChatSession.objects.filter(
user=user,
chat=test_chat,
).exists
)()
assert presence_exists is True
await communicator.disconnect()
presence_exists = await database_sync_to_async(
ActiveChatSession.objects.filter(
user=user,
chat=test_chat,
).exists
)()
assert presence_exists is False
async def test_open_chat_marks_related_notifications_as_read(self, ws_communicator, user, test_chat):
"""Открытие конкретного чата читает только связанные с ним уведомления."""
@ -336,6 +350,13 @@ class TestUserConsumerChatManagement:
assert response["type"] == "chat_closed"
assert response["chat_id"] == test_chat.id
presence_exists = await database_sync_to_async(
ActiveChatSession.objects.filter(
user=user,
chat=test_chat,
).exists
)()
assert presence_exists is False
await communicator.disconnect()

2
docs/guides/NOTIFICATIONS.md
View File

@ -129,6 +129,8 @@ Push работает только в secure context, кроме localhost. Perm
Уведомления сообщений используют стабильный `tag=chat-{chat_id}` и несут `chat_id`, `message_id`, `notification_id`. Backend отправляет невидимые control push-команды, а `frontend/public/sw.js` применяет их через `getNotifications()`: прочтение закрывает уведомление на остальных устройствах, удаление сообщения убирает относящийся к нему push, редактирование заменяет только уже показанный push. Если исходная Celery-задача задержалась, sender повторно проверяет `unread/deleted` и не показывает устаревшее уведомление.
Активный чат хранится как `ActiveChatSession` с TTL 75 секунд и продлевается WebSocket heartbeat. Поэтому presence видна всем backend/Celery workers: если чат открыт хотя бы на одном устройстве пользователя, WebSocket-уведомление и Web Push для этого чата подавляются до доставки на остальные устройства, а запись сразу отмечается прочитанной. После добавления или изменения Celery tasks worker требуется перезапустить — Django autoreload задачи Celery не обновляет.
## Email
`instant` ставит отдельную Celery task при создании уведомления. `daily` и `weekly` обрабатываются Celery Beat в 09:00 по timezone проекта. `never` отключает email.

2
docs/guides/WEBSOCKET.md
View File

@ -62,7 +62,7 @@ ASGI проверяет `Origin` через `AllowedHostsOriginValidator`. Ес
Изменение `ChatMembership` синхронизируется с уже открытыми соединениями: при добавлении consumer входит в группу, при отзыве доступа немедленно выходит из неё и закрывает активный чат.
`active_chat_id` определяет, получает соединение полное chat event или только обновление списка. При открытии чата связанные message notifications отмечаются прочитанными; уведомление об уже открытом чате подавляется.
`active_chat_id` определяет, получает соединение полное chat event или только обновление списка. При открытии чата связанные message notifications отмечаются прочитанными. Состояние дополнительно публикуется в `ActiveChatSession` и продлевается heartbeat, поэтому WebSocket/Push-уведомление об уже открытом чате подавляется для всех устройств пользователя, а не только для текущего соединения.
## Команды клиента

5
frontend/public/sw.js
View File

@ -6,7 +6,7 @@
* нативные уведомления даже когда страница закрыта.
*/
const SW_VERSION = '2.2.0';
const SW_VERSION = '2.2.1';
const SW_APP_KIND = self.CORP_SW_APP_KIND || 'main';
const STATIC_CACHE = `eusrr-static-${SW_VERSION}`;
const PRECACHE_URLS = [
@ -61,9 +61,10 @@ function sameIdentity(actual, expected) {
function matchesNotificationScope(notification, scope = {}) {
const data = notification.data || {};
if (Array.isArray(scope.notification_ids)) {
return scope.notification_ids.some(
const matchesId = scope.notification_ids.some(
(id) => sameIdentity(data.notification_id, id),
);
if (matchesId) return true;
}
if (
scope.chat_id === undefined &&