feat(submissions): auto-replace player in scopes, cancel sync + info about the sync

polylan_submitter/submissions/admin.py

@@ -1,6 +1,7 @@
 from django.contrib import admin
 from django.urls import reverse
 from .models import Submission, SubmissionFile, ImageCrop, Scope
+from .signals import broadcast_scope_updates
 
 
 class ImageCropInline(admin.TabularInline):
@@ -77,6 +78,45 @@ class ScopeAdmin(admin.ModelAdmin):
             },
         )
 
+    # Players on a scope edited here get moved like with the AAndT sync (see
+    # broadcast_scope_updates): state before the save is kept on the instance,
+    # compared once the assigned users (m2m) are saved too in save_related.
+    def save_model(self, request, obj, form, change):
+        previous = Scope.objects.filter(pk=obj.pk).first() if change else None
+        obj._state_before_save = (
+            (
+                previous.is_active,
+                set(previous.assigned_users.values_list("id", flat=True)),
+            )
+            if previous
+            else (False, set())
+        )
+        super().save_model(request, obj, form, change)
+
+    def save_related(self, request, form, formsets, change):
+        super().save_related(request, form, formsets, change)
+
+        scope = form.instance
+        was_active, previous_user_ids = scope._state_before_save
+        user_ids = set(scope.assigned_users.values_list("id", flat=True))
+
+        if scope.is_active != was_active or user_ids != previous_user_ids:
+            broadcast_scope_updates({scope: user_ids | previous_user_ids})
+
+    def delete_model(self, request, obj):
+        self.delete_queryset(request, Scope.objects.filter(pk=obj.pk))
+
+    def delete_queryset(self, request, queryset):
+        changes = {
+            scope: {user.id for user in scope.assigned_users.all()}
+            for scope in queryset.prefetch_related("assigned_users")
+        }
+        for scope in changes:
+            scope.is_active = False
+
+        super().delete_queryset(request, queryset)
+        broadcast_scope_updates(changes)
+
 
 @admin.register(Submission)
 class SubmissionAdmin(admin.ModelAdmin):

polylan_submitter/submissions/api.py

@@ -14,8 +14,16 @@ from .schemas import (
     ScopeOut,
     SyncScopesIn,
     SyncScopesOut,
+    SyncScopesStatusOut,
+    SyncScopesStopOut,
+)
+from .tasks import (
+    get_scope_sync_state,
+    process_submission_file,
+    process_crop,
+    schedule_scope_sync,
+    stop_scope_sync,
 )
-from .tasks import process_submission_file, process_crop, schedule_scope_sync
 
 router = Router()
 
@@ -73,6 +81,37 @@ def list_available_scopes(request: HttpRequest, event_path: str, game_path: str)
     return list(scopes.order_by("name"))
 
 
+def _syncable_event_game(request: HttpRequest, game_path: str) -> EventGame:
+    return get_object_or_404(
+        EventGame,
+        event=request.event,
+        game__path=game_path,
+        game__custom_path=False,
+        enabled=True,
+    )
+
+
+@router.get(
+    f"{PREFIX}/{{game_path}}/sync-scopes",
+    response={200: SyncScopesStatusOut, 403: dict},
+)
+def sync_scopes_status(request: HttpRequest, event_path: str, game_path: str):
+    """State of the AAndT scope sync loop of a (non custom path) game,
+    superusers only."""
+    if not request.user.is_superuser:
+        return 403, {"detail": "Superuser required"}
+
+    event_game = _syncable_event_game(request, game_path)
+    scopes = Scope.objects.filter(event_game=event_game)
+
+    return {
+        **vars(get_scope_sync_state(event_game.id)),
+        "has_token": bool(event_game.apibot_token),
+        "scope_count": scopes.count(),
+        "active_scope_count": scopes.filter(is_active=True).count(),
+    }
+
+
 @router.post(
     f"{PREFIX}/{{game_path}}/sync-scopes",
     response={200: SyncScopesOut, 400: dict, 403: dict},
@@ -86,13 +125,7 @@ def sync_scopes(
     if not request.user.is_superuser:
         return 403, {"detail": "Superuser required"}
 
-    event_game = get_object_or_404(
-        EventGame,
-        event=request.event,
-        game__path=game_path,
-        game__custom_path=False,
-        enabled=True,
-    )
+    event_game = _syncable_event_game(request, game_path)
 
     token = (payload.apibot_token or "").strip()
     if token:
@@ -108,6 +141,21 @@ def sync_scopes(
     return {"started": schedule_scope_sync(event_game.id)}
 
 
+@router.delete(
+    f"{PREFIX}/{{game_path}}/sync-scopes",
+    response={200: SyncScopesStopOut, 403: dict},
+)
+def stop_sync_scopes(request: HttpRequest, event_path: str, game_path: str):
+    """Stop the AAndT scope sync loop of a (non custom path) game, superusers
+    only. Cancels its scheduled job, scopes are left as they are."""
+    if not request.user.is_superuser:
+        return 403, {"detail": "Superuser required"}
+
+    event_game = _syncable_event_game(request, game_path)
+
+    return {"stopped": stop_scope_sync(event_game.id)}
+
+
 @router.post(f"{PREFIX}/submission/create", response={200: SubmissionOut, 403: dict})
 def create_submission(
     request: HttpRequest,

polylan_submitter/submissions/consumers.py

@@ -29,3 +29,6 @@ class SubmissionsConsumer(AsyncWebsocketConsumer):
     async def submission_crop_updated(self, event):
         crop_data = event["crop"]
         await self.send(text_data=json.dumps(crop_data))
+
+    async def scope_updated(self, event):
+        await self.send(text_data=json.dumps({"scope": event["scope"]}))

polylan_submitter/submissions/schemas.py

@@ -19,6 +19,24 @@ class SyncScopesOut(Schema):
     started: bool
 
 
+class SyncScopesStopOut(Schema):
+    # False when no sync loop was running
+    stopped: bool
+
+
+class SyncScopesStatusOut(Schema):
+    running: bool
+    has_token: bool
+    # Last AAndT status seen (e.g. "7inprog"), see games' AAndT status values
+    aandt_status: str | None = None
+    last_run_at: datetime | None = None
+    next_run_at: datetime | None = None
+    # Last run's error, if it failed
+    error: str | None = None
+    scope_count: int
+    active_scope_count: int
+
+
 class ImageCropIn(Schema):
     x1: int
     y1: int

polylan_submitter/submissions/services/aandt_sync.py

@@ -12,6 +12,7 @@ from accounts.models import CustomUser
 from games.models import EventGame
 
 from ..models import Scope
+from ..signals import broadcast_scope_updates
 from .aandt_client import AuthenticatedClient
 from .aandt_client.api.default import (
     aandt_api_infos,
@@ -120,6 +121,9 @@ def sync_event_game_scopes(event_game: EventGame, /) -> ScopeSyncResult:
     Scopes whose match is no longer returned by the API (deleted, or from a
     previous AAndT run reusing the same token) are deactivated.
 
+    Users of scopes that opened/closed or got their players changed are
+    notified over the submissions websocket (see `broadcast_scope_updates`).
+
     Args:
         event_game: The EventGame to sync; must have `apibot_token` set.
 
@@ -173,10 +177,11 @@ def sync_event_game_scopes(event_game: EventGame, /) -> ScopeSyncResult:
     result = ScopeSyncResult(status=status.value if status is not None else None)
     now = timezone.now()
     synced_match_ids: set[int] = set()
+    changes: dict[Scope, set[int]] = {}
 
     for match in matches:
         match_id = _unwrap(match.id)
-        created, warnings = _sync_match(
+        created, warnings, change = _sync_match(
             event_game, client, match, userids_by_member_id, now
         )
         if created is None:
@@ -185,14 +190,22 @@ def sync_event_game_scopes(event_game: EventGame, /) -> ScopeSyncResult:
         result.created += created
         result.updated += not created
         result.warnings.extend(warnings)
+        if change is not None:
+            changes[change[0]] = change[1]
 
-    (
+    gone = list(
         Scope.objects.filter(
             event_game=event_game, is_active=True, external_id__isnull=False
         )
         .exclude(external_id__in=synced_match_ids)
-        .update(is_active=False)
+        .prefetch_related("assigned_users")
     )
+    Scope.objects.filter(id__in=[scope.id for scope in gone]).update(is_active=False)
+    for scope in gone:
+        scope.is_active = False
+        changes[scope] = {user.id for user in scope.assigned_users.all()}
+
+    broadcast_scope_updates(changes)
 
     return result
 
@@ -203,10 +216,17 @@ def _sync_match(
     match: Match,
     userids_by_member_id: dict[int, set[int]],
     now: datetime,
-) -> tuple[bool | None, list[str]]:
+) -> tuple[bool | None, list[str], tuple[Scope, set[int]] | None]:
+    """Upsert the Scope of one match.
+
+    Returns:
+        Whether the Scope was created (None when the match has no id), the
+        warnings, and when it opened/closed or its players changed, the Scope
+        with every user id that was or now is assigned to it.
+    """
     match_id = _unwrap(match.id)
     if match_id is None:
-        return None, []
+        return None, [], None
 
     warnings: list[str] = []
 
@@ -218,6 +238,12 @@ def _sync_match(
 
     users = list(CustomUser.objects.filter(cas_user_id__in=userids))
 
+    previous = Scope.objects.filter(event_game=event_game, external_id=match_id).first()
+    was_active = previous.is_active if previous else False
+    previous_user_ids = (
+        set(previous.assigned_users.values_list("id", flat=True)) if previous else set()
+    )
+
     scope, created = Scope.objects.update_or_create(
         event_game=event_game,
         external_id=match_id,
@@ -241,7 +267,12 @@ def _sync_match(
     if scope.is_active:
         warnings.extend(_push_submission_link(client, match_id, scope))
 
-    return created, warnings
+    user_ids = {user.id for user in users}
+    change = None
+    if scope.is_active != was_active or user_ids != previous_user_ids:
+        change = (scope, user_ids | previous_user_ids)
+
+    return created, warnings, change
 
 
 def _push_submission_link(

polylan_submitter/submissions/signals.py

@@ -1,15 +1,13 @@
+from django.db import transaction
 from django.db.models.signals import post_save
 from django.dispatch import receiver
 from asgiref.sync import async_to_sync
 from channels.layers import get_channel_layer
 
-from submissions.models import Submission, SubmissionFile, ImageCrop
+from submissions.models import Submission, SubmissionFile, ImageCrop, Scope
 from submissions.schemas import SubmissionOut, SubmissionFileOut, ImageCropOut
 
 
-channel_layer = get_channel_layer()
-
-
 @receiver(post_save, sender=Submission)
 def broadcast_submission_update(sender, instance, created, **kwargs):
     if not instance.user:
@@ -18,7 +16,7 @@ def broadcast_submission_update(sender, instance, created, **kwargs):
     user_group_name = f"submissions_user_{instance.user.id}"
     submission_data = SubmissionOut.from_orm(instance).dict()
 
-    async_to_sync(channel_layer.group_send)(
+    async_to_sync(get_channel_layer().group_send)(
         user_group_name,
         {
             "type": "submission.updated",
@@ -36,7 +34,7 @@ def broadcast_submission_file_update(sender, instance, created, **kwargs):
     user_group_name = f"submissions_user_{submission.user.id}"
     file_data = SubmissionFileOut.from_orm(instance).dict()
 
-    async_to_sync(channel_layer.group_send)(
+    async_to_sync(get_channel_layer().group_send)(
         user_group_name,
         {
             "type": "submission.file.updated",
@@ -54,10 +52,43 @@ def broadcast_image_crop_update(sender, instance, created, **kwargs):
     user_group_name = f"submissions_user_{submission.user.id}"
     crop_data = ImageCropOut.from_orm(instance).dict()
 
-    async_to_sync(channel_layer.group_send)(
+    async_to_sync(get_channel_layer().group_send)(
         user_group_name,
         {
             "type": "submission.crop.updated",
             "crop": crop_data,
         },
     )
+
+
+def broadcast_scope_updates(changes: dict[Scope, set[int]], /):
+    """Tell users their scope changed (closed/opened, (un)assigned), once the
+    current transaction commits, so the frontend can move them to the right
+    match without a refresh.
+
+    Args:
+        changes: Changed scopes, each with every user id that was or now is
+            assigned to it.
+    """
+
+    def send():
+        channel_layer = get_channel_layer()
+        for scope, user_ids in changes.items():
+            assigned_ids = set(scope.assigned_users.values_list("id", flat=True))
+            for user_id in user_ids:
+                async_to_sync(channel_layer.group_send)(
+                    f"submissions_user_{user_id}",
+                    {
+                        "type": "scope.updated",
+                        "scope": {
+                            "id": str(scope.id),
+                            "is_active": scope.is_active,
+                            "assigned": user_id in assigned_ids,
+                        },
+                    },
+                )
+
+    if changes:
+        # robust: a websocket hiccup only gets logged, it mustn't fail the
+        # (already committed) sync or admin save
+        transaction.on_commit(send, robust=True)

polylan_submitter/submissions/tasks.py

@@ -1,7 +1,9 @@
 import logging
-from datetime import timedelta
+from dataclasses import dataclass
+from datetime import datetime, timedelta
 from uuid import uuid4
 
+import django_rq
 from django.core.cache import cache
 from django.tasks import task
 from django.utils import timezone
@@ -9,17 +11,18 @@ from games.models import EventGame
 from submissions.models import ImageCrop, SubmissionFile, OCRStatus
 from submissions.services.aandt_sync import (
     AANDT_STATUS_IN_PROGRESS,
-    ScopeSyncError,
     sync_event_game_scopes,
 )
 
 from PIL import Image
 import pytesseract
+from rq.exceptions import InvalidJobOperation, NoSuchJobError
+from rq.job import Job, JobStatus
 
 logger = logging.getLogger(__name__)
 
 # How often to repoll an EventGame's AAndT while it's "in progress", and while
-# it isn't yet (registrations, preparation, ...) or the API call failed.
+# it isn't yet (registrations, preparation, ...) or the run failed.
 SCOPE_SYNC_REPOLL_INTERVAL = timedelta(seconds=15)
 SCOPE_SYNC_IDLE_INTERVAL = timedelta(minutes=1)
 # AAndT statuses after which there's nothing left to sync, the loop stops.
@@ -30,10 +33,59 @@ def _scope_sync_guard_key(event_game_id: int) -> str:
     return f"submissions:scope_sync:{event_game_id}"
 
 
+def _scope_sync_job_key(event_game_id: int) -> str:
+    return f"submissions:scope_sync:{event_game_id}:job"
+
+
+def _scope_sync_state_key(event_game_id: int) -> str:
+    return f"submissions:scope_sync:{event_game_id}:state"
+
+
 def _scope_sync_guard_timeout() -> int:
     return int(SCOPE_SYNC_IDLE_INTERVAL.total_seconds() * 3)
 
 
+# Last run's outcome, shown to admins in the sync modal. Kept around a while
+# after the loop stops, so they still see why it stopped.
+SCOPE_SYNC_STATE_TIMEOUT = int(timedelta(days=1).total_seconds())
+
+
+@dataclass
+class ScopeSyncState:
+    running: bool
+    aandt_status: str | None = None
+    last_run_at: datetime | None = None
+    next_run_at: datetime | None = None
+    error: str | None = None
+
+
+def get_scope_sync_state(event_game_id: int, /) -> ScopeSyncState:
+    state = cache.get(_scope_sync_state_key(event_game_id)) or {}
+    running = cache.get(_scope_sync_guard_key(event_game_id)) is not None
+    return ScopeSyncState(
+        running=running,
+        aandt_status=state.get("aandt_status"),
+        last_run_at=state.get("last_run_at"),
+        next_run_at=state.get("next_run_at") if running else None,
+        error=state.get("error"),
+    )
+
+
+def _enqueue_scope_sync(event_game_id: int, loop_id: str, run_after: datetime | None):
+    task = (
+        sync_event_game_scopes_task.using(run_after=run_after)
+        if run_after
+        else sync_event_game_scopes_task
+    )
+    result = task.enqueue(event_game_id, loop_id=loop_id)
+    # Same id as the rq job (see django_tasks_rq), so stop_scope_sync can cancel it
+    cache.set(
+        _scope_sync_job_key(event_game_id),
+        result.id,
+        timeout=_scope_sync_guard_timeout(),
+    )
+
+
 def schedule_scope_sync(event_game_id: int, /) -> bool:
     """Start the sync loop of an EventGame, unless one is already running.
     Safe to call any number of times (admin button spam, multiple workers...).
@@ -49,13 +101,32 @@ def schedule_scope_sync(event_game_id: int, /) -> bool:
     ):
         return False
 
-    sync_event_game_scopes_task.enqueue(event_game_id, loop_id=loop_id)
+    _enqueue_scope_sync(event_game_id, loop_id, run_after=None)
     return True
 
 
-def stop_scope_sync(event_game_id: int, /):
-    """Release the guard, the running loop (if any) stops at its next run."""
-    cache.delete(_scope_sync_guard_key(event_game_id))
+def stop_scope_sync(event_game_id: int, /) -> bool:
+    """Stop the sync loop of an EventGame: cancels its scheduled rq job and
+    releases the guard (a run already in progress won't reschedule itself).
+
+    Returns:
+        Whether a loop was running.
+    """
+    running = cache.get(_scope_sync_guard_key(event_game_id)) is not None
+    job_id = cache.get(_scope_sync_job_key(event_game_id))
+    cache.delete_many(
+        [_scope_sync_guard_key(event_game_id), _scope_sync_job_key(event_game_id)]
+    )
+
+    if job_id:
+        try:
+            job = Job.fetch(job_id, connection=django_rq.get_connection())
+            if job.get_status() in (JobStatus.SCHEDULED, JobStatus.QUEUED):
+                job.cancel()
+        except (NoSuchJobError, InvalidJobOperation):
+            pass
+
+    return running
 
 
 def _process_ocr_file(obj, file, crop: ImageCrop | None = None) -> bool:
@@ -142,14 +213,14 @@ def sync_event_game_scopes_task(event_game_id: int, loop_id: str | None = None):
     Started once (see schedule_scope_sync), it keeps going on its own without
     an external scheduler: every SCOPE_SYNC_REPOLL_INTERVAL while the AAndT is
     "in progress" (`7inprog`), every SCOPE_SYNC_IDLE_INTERVAL before that or
-    when the API call failed. Stops once the AAndT is done/reviewed, when the
+    when the run failed. Stops once the AAndT is done/reviewed, when the
     EventGame (or its token) is gone, or when another loop took the guard.
 
     Without loop_id it's a one-off run that leaves the loop alone.
     """
     guard_key = _scope_sync_guard_key(event_game_id)
     if loop_id is not None and cache.get(guard_key) != loop_id:
-        # Another loop owns the guard (or it expired), stop this one
+        # Another loop owns the guard (or it expired/was stopped), stop this one
         return
 
     event_game = EventGame.objects.filter(id=event_game_id).first()
@@ -160,27 +231,45 @@ def sync_event_game_scopes_task(event_game_id: int, loop_id: str | None = None):
         return
 
     status = None
+    error = None
     try:
         status = sync_event_game_scopes(event_game).status
-    except ScopeSyncError:
+    except Exception as exc:
         if loop_id is None:
             raise
-        # Probably transient (AAndT site down, ...), retry later
+        # Retry later rather than letting the loop die (AAndT site down, bug, ...),
+        # it'd otherwise only restart once the guard expires and someone restarts it
         logger.exception("Scope sync failed for EventGame %s", event_game_id)
+        error = str(exc) or exc.__class__.__name__
 
     if loop_id is None:
         return
 
-    if status in AANDT_FINISHED_STATUSES:
-        cache.delete(guard_key)
+    now = timezone.now()
+    next_run_at = None
+    # Re-checked: the loop may have been stopped while this run was syncing
+    if status not in AANDT_FINISHED_STATUSES and cache.get(guard_key) == loop_id:
+        next_run_at = now + (
+            SCOPE_SYNC_REPOLL_INTERVAL
+            if status == AANDT_STATUS_IN_PROGRESS
+            else SCOPE_SYNC_IDLE_INTERVAL
+        )
+
+    cache.set(
+        _scope_sync_state_key(event_game_id),
+        {
+            "aandt_status": status,
+            "last_run_at": now,
+            "next_run_at": next_run_at,
+            "error": error,
+        },
+        timeout=SCOPE_SYNC_STATE_TIMEOUT,
+    )
+
+    if next_run_at is None:
+        if cache.get(guard_key) == loop_id:
+            cache.delete(guard_key)
         return
 
-    interval = (
-        SCOPE_SYNC_REPOLL_INTERVAL
-        if status == AANDT_STATUS_IN_PROGRESS
-        else SCOPE_SYNC_IDLE_INTERVAL
-    )
     cache.set(guard_key, loop_id, timeout=_scope_sync_guard_timeout())
-    sync_event_game_scopes_task.using(run_after=timezone.now() + interval).enqueue(
-        event_game_id, loop_id=loop_id
-    )
+    _enqueue_scope_sync(event_game_id, loop_id, run_after=next_run_at)

polylan_submitter/submissions/tests.py

@@ -1,14 +1,23 @@
+from contextlib import ExitStack
 from unittest import mock
 
 from django.core.cache import cache
 from django.test import TestCase
 
+from rq.job import JobStatus
+
 from accounts.models import CustomUser
 from events.models import Event
 from games.models import EventGame, Game
 
 from . import tasks
-from .services.aandt_sync import ScopeSyncError, ScopeSyncResult
+from .models import Scope
+from .services.aandt_client.models import (
+    AandtApiInfosResponse200,
+    AandtApiMatchesResponse200,
+    AandtApiMembersResponse200,
+)
+from .services.aandt_sync import ScopeSyncError, ScopeSyncResult, sync_event_game_scopes
 
 
 class SyncScopesApiTests(TestCase):
@@ -59,6 +68,42 @@ class SyncScopesApiTests(TestCase):
         self.assertEqual(self.event_game.apibot_token, "abc")
         self.schedule.assert_called_once_with(self.event_game.id)
 
+    def test_status(self):
+        Scope.objects.create(event_game=self.event_game, name="M1")
+        Scope.objects.create(event_game=self.event_game, name="M2", is_active=False)
+        self.client.force_login(self.superuser)
+
+        with mock.patch(
+            "submissions.api.get_scope_sync_state",
+            return_value=tasks.ScopeSyncState(running=True, aandt_status="7inprog"),
+        ):
+            response = self.client.get(self.url)
+
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertTrue(data["running"])
+        self.assertFalse(data["has_token"])
+        self.assertEqual(data["aandt_status"], "7inprog")
+        self.assertEqual((data["active_scope_count"], data["scope_count"]), (1, 2))
+
+    def test_stop(self):
+        self.client.force_login(self.superuser)
+
+        with mock.patch("submissions.api.stop_scope_sync", return_value=True) as stop:
+            response = self.client.delete(self.url)
+
+        self.assertEqual(response.json(), {"stopped": True})
+        stop.assert_called_once_with(self.event_game.id)
+
+    def test_status_and_stop_forbidden_for_staff(self):
+        staff = CustomUser.objects.create_user(
+            username="staff", password="x", is_staff=True
+        )
+        self.client.force_login(staff)
+
+        self.assertEqual(self.client.get(self.url).status_code, 403)
+        self.assertEqual(self.client.delete(self.url).status_code, 403)
+
     def test_custom_path_game_is_404(self):
         self.game.custom_path = True
         self.game.save()
@@ -76,21 +121,38 @@ class SyncEventGameScopesTaskTests(TestCase):
         )
         self.guard_key = tasks._scope_sync_guard_key(self.event_game.id)
         self.loop_id = "loop"
+        self.job_key = tasks._scope_sync_job_key(self.event_game.id)
+        self.state_key = tasks._scope_sync_state_key(self.event_game.id)
         cache.set(self.guard_key, self.loop_id)
-        self.addCleanup(cache.delete, self.guard_key)
+        self.addCleanup(
+            cache.delete_many, [self.guard_key, self.job_key, self.state_key]
+        )
 
-    def _run(self, status=None, exc=None):
+    @staticmethod
+    def _task_mock(task):
+        # enqueue() returns a TaskResult, whose id ends up in the cache
+        task.enqueue.return_value.id = "job-1"
+        task.using.return_value.enqueue.return_value.id = "job-2"
+        return task
+
+    def _run(self, status=None, exc=None, during_sync=None):
         run = tasks.sync_event_game_scopes_task.func
+
+        def sync_side_effect(_event_game):
+            if during_sync:
+                during_sync()
+            if exc:
+                raise exc
+            return ScopeSyncResult(status=status)
+
         with (
             mock.patch.object(
-                tasks,
-                "sync_event_game_scopes",
-                side_effect=exc,
-                return_value=ScopeSyncResult(status=status),
+                tasks, "sync_event_game_scopes", side_effect=sync_side_effect
             ) as sync,
             mock.patch.object(tasks, "sync_event_game_scopes_task") as task,
             mock.patch.object(tasks, "logger"),
         ):
+            self._task_mock(task)
             run(self.event_game.id, loop_id=self.loop_id)
         return sync, task
 
@@ -98,6 +160,7 @@ class SyncEventGameScopesTaskTests(TestCase):
         cache.delete(self.guard_key)
 
         with mock.patch.object(tasks, "sync_event_game_scopes_task") as task:
+            self._task_mock(task)
             self.assertTrue(tasks.schedule_scope_sync(self.event_game.id))
             self.assertFalse(tasks.schedule_scope_sync(self.event_game.id))
 
@@ -114,6 +177,45 @@ class SyncEventGameScopesTaskTests(TestCase):
 
         task.using.assert_called_once()
 
+    def test_reschedules_after_unexpected_error(self):
+        _, task = self._run(exc=RuntimeError("boom"))
+
+        task.using.assert_called_once()
+        state = tasks.get_scope_sync_state(self.event_game.id)
+        self.assertEqual(state.error, "boom")
+        self.assertTrue(state.running)
+
+    def test_records_job_and_state(self):
+        self._run(status="7inprog")
+
+        self.assertEqual(cache.get(self.job_key), "job-2")
+        state = tasks.get_scope_sync_state(self.event_game.id)
+        self.assertEqual(state.aandt_status, "7inprog")
+        self.assertIsNotNone(state.next_run_at)
+
+    def test_stop_cancels_scheduled_job(self):
+        cache.set(self.job_key, "job-1")
+        job = mock.Mock(get_status=mock.Mock(return_value=JobStatus.SCHEDULED))
+
+        with (
+            mock.patch.object(tasks.Job, "fetch", return_value=job) as fetch,
+            mock.patch.object(tasks.django_rq, "get_connection"),
+        ):
+            self.assertTrue(tasks.stop_scope_sync(self.event_game.id))
+
+        fetch.assert_called_once_with("job-1", connection=mock.ANY)
+        job.cancel.assert_called_once()
+        self.assertIsNone(cache.get(self.guard_key))
+        self.assertFalse(tasks.stop_scope_sync(self.event_game.id))
+
+    def test_stop_during_run_does_not_reschedule(self):
+        _, task = self._run(
+            status="7inprog", during_sync=lambda: cache.delete(self.guard_key)
+        )
+
+        task.using.assert_not_called()
+        self.assertIsNone(cache.get(self.guard_key))
+
     def test_stops_once_aandt_is_done(self):
         _, task = self._run(status="9done")
 
@@ -127,3 +229,142 @@ class SyncEventGameScopesTaskTests(TestCase):
 
         sync.assert_not_called()
         task.using.assert_not_called()
+
+
+class SyncScopeBroadcastTests(TestCase):
+    def setUp(self):
+        event = Event.objects.create(name="Event", path="event")
+        game = Game.objects.create(name="Game", path="game")
+        self.event_game = EventGame.objects.create(
+            event=event, game=game, apibot_token="abc"
+        )
+        self.player = CustomUser.objects.create_user(
+            username="player", password="x", cas_user_id="42"
+        )
+
+    def _sync(self, matches):
+        api = "submissions.services.aandt_sync"
+        responses = {
+            "aandt_api_infos": AandtApiInfosResponse200.from_dict(
+                {"aandt": {"status": "7inprog"}}
+            ),
+            "aandt_api_members": AandtApiMembersResponse200.from_dict(
+                {"members": [{"id": 1, "user": {"id": 42}}]}
+            ),
+            "aandt_api_matches": AandtApiMatchesResponse200.from_dict(
+                {"matches": matches}
+            ),
+        }
+        with ExitStack() as stack:
+            for module, parsed in responses.items():
+                stack.enter_context(
+                    mock.patch(
+                        f"{api}.{module}.sync_detailed",
+                        return_value=mock.Mock(parsed=parsed),
+                    )
+                )
+            stack.enter_context(
+                mock.patch(f"{api}._push_submission_link", return_value=[])
+            )
+            stack.enter_context(mock.patch("submissions.signals.get_channel_layer"))
+            send = stack.enter_context(mock.patch("submissions.signals.async_to_sync"))
+            with self.captureOnCommitCallbacks(execute=True):
+                sync_event_game_scopes(self.event_game)
+
+        return [c.args[1]["scope"] for c in send.return_value.call_args_list]
+
+    @staticmethod
+    def _match(match_id, status):
+        return {
+            "id": match_id,
+            "title": f"M{match_id}",
+            "status": status,
+            "players": {"1": 0},
+        }
+
+    def test_match_done_and_next_one_playing_notifies_player(self):
+        self._sync([self._match(1, "playing"), self._match(2, "waitPlayer")])
+
+        sent = self._sync([self._match(1, "done"), self._match(2, "playing")])
+
+        by_external_id = {
+            Scope.objects.get(id=scope["id"]).external_id: scope for scope in sent
+        }
+        self.assertFalse(by_external_id[1]["is_active"])
+        self.assertTrue(by_external_id[2]["is_active"])
+        self.assertTrue(by_external_id[2]["assigned"])
+
+    def test_unchanged_scopes_send_nothing(self):
+        self._sync([self._match(1, "playing")])
+
+        self.assertEqual(self._sync([self._match(1, "playing")]), [])
+
+    def test_removed_match_is_closed_and_notified(self):
+        self._sync([self._match(1, "playing")])
+
+        sent = self._sync([])
+
+        self.assertEqual(len(sent), 1)
+        self.assertFalse(sent[0]["is_active"])
+
+
+class ScopeAdminBroadcastTests(TestCase):
+    def setUp(self):
+        event = Event.objects.create(name="Event", path="event")
+        game = Game.objects.create(name="Game", path="game")
+        event_game = EventGame.objects.create(event=event, game=game)
+        self.player = CustomUser.objects.create_user(username="player", password="x")
+        self.scope = Scope.objects.create(event_game=event_game, name="Match 1")
+        self.scope.assigned_users.set([self.player])
+        self.client.force_login(
+            CustomUser.objects.create_user(
+                username="admin", password="x", is_staff=True, is_superuser=True
+            )
+        )
+        self.form = {
+            "event_game": event_game.id,
+            "name": "Match 1",
+            "assigned_users": [self.player.id],
+            "is_active": "on",
+        }
+
+    def _sent(self, request):
+        with (
+            mock.patch("submissions.signals.get_channel_layer"),
+            mock.patch("submissions.signals.async_to_sync") as send,
+            self.captureOnCommitCallbacks(execute=True),
+        ):
+            response = request()
+        self.assertEqual(response.status_code, 302)
+        return [c.args[1]["scope"] for c in send.return_value.call_args_list]
+
+    def _change(self, **changes):
+        url = f"/admin/submissions/scope/{self.scope.id}/change/"
+        data = {k: v for k, v in {**self.form, **changes}.items() if v is not None}
+        return self._sent(lambda: self.client.post(url, data))
+
+    def test_closing_scope_notifies_player(self):
+        sent = self._change(is_active=None)
+
+        self.assertEqual(
+            sent, [{"id": str(self.scope.id), "is_active": False, "assigned": True}]
+        )
+
+    def test_unassigning_player_notifies_them(self):
+        sent = self._change(assigned_users=[])
+
+        self.assertEqual(
+            sent, [{"id": str(self.scope.id), "is_active": True, "assigned": False}]
+        )
+
+    def test_unchanged_save_sends_nothing(self):
+        self.assertEqual(self._change(name="Renamed"), [])
+
+    def test_delete_notifies_player(self):
+        url = f"/admin/submissions/scope/{self.scope.id}/delete/"
+
+        sent = self._sent(lambda: self.client.post(url, {"post": "yes"}))
+
+        self.assertEqual(
+            sent, [{"id": str(self.scope.id), "is_active": False, "assigned": False}]
+        )