feat: lock concurrent moderation decisions

apps/api/alembic/versions/0015_moderation_version.py

@@ -0,0 +1,22 @@
+"""add optimistic moderation versions
+
+Revision ID: 0015
+Revises: 0014
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0015"
+down_revision = "0014"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("catch_report", sa.Column("moderation_version", sa.Integer(), nullable=False, server_default="0"))
+    op.add_column("external_observation", sa.Column("moderation_version", sa.Integer(), nullable=False, server_default="0"))
+
+
+def downgrade() -> None:
+    op.drop_column("external_observation", "moderation_version")
+    op.drop_column("catch_report", "moderation_version")

apps/api/app/main.py

@@ -33,7 +33,7 @@ from .routers.public_data import router as public_data_router
 from .routers.submissions import router as submissions_router
 from .time_utils import aware
 from .public_cache import public_cache
-from .schemas import ActivityOut, AdminCatchReportOut, AdminModerationHistoryOut, CatchReportAccepted, CatchReportCreate, CatchReportCreated, ExternalAliasSuggestionOut, ExternalObservationDecision, ExternalObservationMapping, ExternalObservationOut, ExternalObservationPublished, ImportRunOut, ModerationUpdate
+from .schemas import ActivityOut, AdminCatchReportOut, AdminModerationHistoryOut, CatchReportAccepted, CatchReportCreate, CatchReportCreated, ExternalAliasSuggestionOut, ExternalObservationAction, ExternalObservationDecision, ExternalObservationMapping, ExternalObservationOut, ExternalObservationPublished, ImportRunOut, ModerationUpdate
 from .storage import ScreenshotError, client as storage_client, delete_screenshot, signed_screenshot_url, upload_screenshot
 from .submission_security import check_rate_limit
 from .submission_security import is_trusted_proxy as _is_trusted_proxy
@@ -237,6 +237,7 @@ def _external_out(item: ExternalObservation) -> ExternalObservationOut:
         waterbody_slug=item.waterbody.slug if item.waterbody else None,
         catch_report_id=item.catch_report_id, review_note=item.review_note,
         missing_fields=missing_fields, source_payload=allowed_payload,
+        moderation_version=item.moderation_version,
     )
 
 
@@ -312,13 +313,16 @@ def admin_map_external_observation(
     observation_id: UUID, payload: ExternalObservationMapping, db: Db,
     _: Annotated[str, Depends(_admin)],
 ) -> ExternalObservationOut:
-    observation = db.get(ExternalObservation, observation_id)
+    observation = db.scalar(select(ExternalObservation).where(ExternalObservation.id == observation_id).with_for_update())
     fish = db.scalar(select(Fish).where(Fish.slug == payload.fish_slug))
     waterbody = db.scalar(select(Waterbody).where(Waterbody.slug == payload.waterbody_slug))
     if observation is None:
         raise HTTPException(status_code=404, detail="external observation not found")
     if fish is None or waterbody is None:
         raise HTTPException(status_code=422, detail="unknown fish or waterbody")
+    if observation.moderation_version != payload.expected_version:
+        raise HTTPException(status_code=409, detail="observation changed; reload the queue")
+    observation.moderation_version += 1
     try:
         return _external_out(map_observation(db, observation, fish, waterbody, note=payload.note))
     except ExternalReviewError as exc:
@@ -327,11 +331,14 @@ def admin_map_external_observation(
 
 @app.post("/api/v1/admin/external-observations/{observation_id}/publish", response_model=ExternalObservationPublished)
 def admin_publish_external_observation(
-    observation_id: UUID, db: Db, _: Annotated[str, Depends(_admin)],
+    observation_id: UUID, payload: ExternalObservationAction, db: Db, _: Annotated[str, Depends(_admin)],
 ) -> ExternalObservationPublished:
-    observation = db.get(ExternalObservation, observation_id)
+    observation = db.scalar(select(ExternalObservation).where(ExternalObservation.id == observation_id).with_for_update())
     if observation is None:
         raise HTTPException(status_code=404, detail="external observation not found")
+    if observation.moderation_version != payload.expected_version:
+        raise HTTPException(status_code=409, detail="observation changed; reload the queue")
+    observation.moderation_version += 1
     try:
         report = publish_observation(db, observation)
     except ExternalReviewError as exc:
@@ -345,9 +352,12 @@ def admin_reject_external_observation(
     observation_id: UUID, payload: ExternalObservationDecision, db: Db,
     _: Annotated[str, Depends(_admin)],
 ) -> ExternalObservationOut:
-    observation = db.get(ExternalObservation, observation_id)
+    observation = db.scalar(select(ExternalObservation).where(ExternalObservation.id == observation_id).with_for_update())
     if observation is None:
         raise HTTPException(status_code=404, detail="external observation not found")
+    if observation.moderation_version != payload.expected_version:
+        raise HTTPException(status_code=409, detail="observation changed; reload the queue")
+    observation.moderation_version += 1
     try:
         return _external_out(reject_observation(db, observation, reason=payload.reason))
     except ExternalReviewError as exc:
@@ -459,16 +469,19 @@ app.include_router(submissions_router)
 @app.get("/api/v1/admin/catch-reports", response_model=list[AdminCatchReportOut])
 def admin_reports(db: Db, _: Annotated[str, Depends(_admin)], status: ModerationStatus = ModerationStatus.pending, limit: int = Query(50, ge=1, le=100), offset: int = Query(0, ge=0)) -> list[AdminCatchReportOut]:
     reports = list(db.scalars(select(CatchReport).options(joinedload(CatchReport.fish), joinedload(CatchReport.waterbody), joinedload(CatchReport.spot), joinedload(CatchReport.bait)).where(CatchReport.source_type == SourceType.user, CatchReport.moderation_status == status, CatchReport.deleted_at.is_(None)).order_by(CatchReport.reported_at, CatchReport.id).offset(offset).limit(limit)))
-    return [AdminCatchReportOut(id=r.id, fish=r.fish.name_ru, waterbody=r.waterbody.name_ru, coordinates=f"{r.spot.x}:{r.spot.y}" if r.spot else "—", weight_g=r.weight_g, bait=r.bait.name if r.bait else None, player_name=r.player_name, reported_at=r.reported_at, moderation_status=r.moderation_status.value, comment=(r.raw_payload or {}).get("comment"), screenshot_url=signed_screenshot_url(r.screenshot_key) if r.screenshot_key else None) for r in reports]
+    return [AdminCatchReportOut(id=r.id, fish=r.fish.name_ru, waterbody=r.waterbody.name_ru, coordinates=f"{r.spot.x}:{r.spot.y}" if r.spot else "—", weight_g=r.weight_g, bait=r.bait.name if r.bait else None, player_name=r.player_name, reported_at=r.reported_at, moderation_status=r.moderation_status.value, comment=(r.raw_payload or {}).get("comment"), screenshot_url=signed_screenshot_url(r.screenshot_key) if r.screenshot_key else None, moderation_version=r.moderation_version) for r in reports]
 
 
 @app.patch("/api/v1/admin/catch-reports/{report_id}", response_model=CatchReportCreated)
 def moderate_report(report_id: UUID, payload: ModerationUpdate, db: Db, moderator: Annotated[str, Depends(_admin)]) -> CatchReportCreated:
-    report = db.get(CatchReport, report_id)
+    report = db.scalar(select(CatchReport).where(CatchReport.id == report_id).with_for_update())
     if report is None or report.source_type != SourceType.user or report.deleted_at is not None:
         raise HTTPException(status_code=404, detail="catch report not found")
+    if report.moderation_version != payload.expected_version:
+        raise HTTPException(status_code=409, detail="report changed; reload the queue")
     previous = report.moderation_status
     report.moderation_status = ModerationStatus(payload.status)
+    report.moderation_version += 1
     db.add(ModerationEvent(catch_report=report, created_at=datetime.now(timezone.utc), previous_status=previous, new_status=report.moderation_status, moderator=moderator, reason=payload.reason))
     db.commit()
     public_cache.invalidate()
@@ -476,10 +489,12 @@ def moderate_report(report_id: UUID, payload: ModerationUpdate, db: Db, moderato
 
 
 @app.delete("/api/v1/admin/catch-reports/{report_id}", status_code=204, response_class=Response)
-def delete_report(report_id: UUID, db: Db, moderator: Annotated[str, Depends(_admin)]) -> Response:
-    report = db.get(CatchReport, report_id)
+def delete_report(report_id: UUID, db: Db, moderator: Annotated[str, Depends(_admin)], expected_version: int = Query(ge=0)) -> Response:
+    report = db.scalar(select(CatchReport).where(CatchReport.id == report_id).with_for_update())
     if report is None or report.source_type != SourceType.user or report.deleted_at is not None:
         raise HTTPException(status_code=404, detail="catch report not found")
+    if report.moderation_version != expected_version:
+        raise HTTPException(status_code=409, detail="report changed; reload the queue")
     previous = report.moderation_status
     if report.screenshot_key:
         try:
@@ -487,6 +502,7 @@ def delete_report(report_id: UUID, db: Db, moderator: Annotated[str, Depends(_ad
         except Exception as exc:
             raise HTTPException(status_code=502, detail="screenshot deletion failed") from exc
     report.moderation_status = ModerationStatus.rejected
+    report.moderation_version += 1
     report.deleted_at = datetime.now(timezone.utc)
     report.player_name = None
     report.source_url = None

apps/api/app/models.py

@@ -95,6 +95,7 @@ class CatchReport(Base):
     screenshot_upload_token_hash: Mapped[str | None] = mapped_column(String(64))
     deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     raw_payload: Mapped[dict | None] = mapped_column(JSON)
+    moderation_version: Mapped[int] = mapped_column(default=0)
     fish: Mapped[Fish] = relationship()
     spot: Mapped[Spot | None] = relationship()
     waterbody: Mapped[Waterbody] = relationship()
@@ -197,6 +198,7 @@ class ExternalObservation(Base):
     catch_report_id: Mapped[uuid.UUID | None] = mapped_column(ForeignKey("catch_report.id"), unique=True)
     review_note: Mapped[str | None] = mapped_column(Text)
     reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    moderation_version: Mapped[int] = mapped_column(default=0)
     source: Mapped[DataSource] = relationship()
     fish: Mapped[Fish | None] = relationship()
     waterbody: Mapped[Waterbody | None] = relationship()

apps/api/app/schemas.py

@@ -199,11 +199,13 @@ class AdminCatchReportOut(BaseModel):
     moderation_status: str
     comment: str | None
     screenshot_url: str | None
+    moderation_version: int
 
 
 class ModerationUpdate(BaseModel):
     status: str
     reason: str | None = Field(default=None, max_length=1000)
+    expected_version: int = Field(ge=0)
 
     @field_validator("status")
     @classmethod
@@ -236,12 +238,14 @@ class ExternalObservationOut(BaseModel):
     review_note: str | None
     missing_fields: list[str]
     source_payload: dict[str, str | int | float | bool | None]
+    moderation_version: int
 
 
 class ExternalObservationMapping(BaseModel):
     fish_slug: str
     waterbody_slug: str
     note: str | None = Field(default=None, max_length=1000)
+    expected_version: int = Field(ge=0)
 
 
 class ExternalAliasSuggestionOut(BaseModel):
@@ -261,6 +265,11 @@ class AdminModerationHistoryOut(BaseModel):
 
 class ExternalObservationDecision(BaseModel):
     reason: str = Field(min_length=1, max_length=1000)
+    expected_version: int = Field(ge=0)
+
+
+class ExternalObservationAction(BaseModel):
+    expected_version: int = Field(ge=0)
 
 
 class ExternalObservationPublished(BaseModel):

apps/api/openapi.json

@@ -147,6 +147,10 @@
             "title": "Moderation Status",
             "type": "string"
           },
+          "moderation_version": {
+            "title": "Moderation Version",
+            "type": "integer"
+          },
           "player_name": {
             "anyOf": [
               {
@@ -194,7 +198,8 @@
           "reported_at",
           "moderation_status",
           "comment",
-          "screenshot_url"
+          "screenshot_url",
+          "moderation_version"
         ],
         "title": "AdminCatchReportOut",
         "type": "object"
@@ -643,8 +648,27 @@
         "title": "ExternalAliasSuggestionOut",
         "type": "object"
       },
+      "ExternalObservationAction": {
+        "properties": {
+          "expected_version": {
+            "minimum": 0.0,
+            "title": "Expected Version",
+            "type": "integer"
+          }
+        },
+        "required": [
+          "expected_version"
+        ],
+        "title": "ExternalObservationAction",
+        "type": "object"
+      },
       "ExternalObservationDecision": {
         "properties": {
+          "expected_version": {
+            "minimum": 0.0,
+            "title": "Expected Version",
+            "type": "integer"
+          },
           "reason": {
             "maxLength": 1000,
             "minLength": 1,
@@ -653,13 +677,19 @@
           }
         },
         "required": [
-          "reason"
+          "reason",
+          "expected_version"
         ],
         "title": "ExternalObservationDecision",
         "type": "object"
       },
       "ExternalObservationMapping": {
         "properties": {
+          "expected_version": {
+            "minimum": 0.0,
+            "title": "Expected Version",
+            "type": "integer"
+          },
           "fish_slug": {
             "title": "Fish Slug",
             "type": "string"
@@ -683,7 +713,8 @@
         },
         "required": [
           "fish_slug",
-          "waterbody_slug"
+          "waterbody_slug",
+          "expected_version"
         ],
         "title": "ExternalObservationMapping",
         "type": "object"
@@ -750,6 +781,10 @@
             "title": "Missing Fields",
             "type": "array"
           },
+          "moderation_version": {
+            "title": "Moderation Version",
+            "type": "integer"
+          },
           "published_at": {
             "anyOf": [
               {
@@ -906,7 +941,8 @@
           "catch_report_id",
           "review_note",
           "missing_fields",
-          "source_payload"
+          "source_payload",
+          "moderation_version"
         ],
         "title": "ExternalObservationOut",
         "type": "object"
@@ -1189,6 +1225,11 @@
       },
       "ModerationUpdate": {
         "properties": {
+          "expected_version": {
+            "minimum": 0.0,
+            "title": "Expected Version",
+            "type": "integer"
+          },
           "reason": {
             "anyOf": [
               {
@@ -1207,7 +1248,8 @@
           }
         },
         "required": [
-          "status"
+          "status",
+          "expected_version"
         ],
         "title": "ModerationUpdate",
         "type": "object"
@@ -1897,6 +1939,16 @@
               "type": "string"
             }
           },
+          {
+            "in": "query",
+            "name": "expected_version",
+            "required": true,
+            "schema": {
+              "minimum": 0,
+              "title": "Expected Version",
+              "type": "integer"
+            }
+          },
           {
             "in": "header",
             "name": "authorization",
@@ -2353,6 +2405,16 @@
             }
           }
         ],
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/ExternalObservationAction"
+              }
+            }
+          },
+          "required": true
+        },
         "responses": {
           "200": {
             "content": {

apps/api/tests/test_api.py

@@ -286,8 +286,11 @@ def test_user_report_requires_moderation_before_activity() -> None:
     assert pending.status_code == 200
     assert pending.headers["Cache-Control"] == "no-store"
     assert any(item["id"] == report_id for item in pending.json())
-    approved = client.patch(f"/api/v1/admin/catch-reports/{report_id}", headers=headers, json={"status": "approved", "reason": "fixture verified"})
+    approved = client.patch(f"/api/v1/admin/catch-reports/{report_id}", headers=headers, json={"status": "approved", "reason": "fixture verified", "expected_version": 0})
     assert approved.status_code == 200
+    stale = client.patch(f"/api/v1/admin/catch-reports/{report_id}", headers=headers, json={"status": "rejected", "reason": "stale tab", "expected_version": 0})
+    assert stale.status_code == 409
+    assert "reload" in stale.json()["detail"]
     activity = client.get("/api/v1/activity?waterbody=test-lake&fish=pike&hours=24").json()
     assert any(item["x"] == 77 and item["catches"] == 1 for item in activity["items"])
 
@@ -312,19 +315,19 @@ def test_external_observation_requires_mapping_and_complete_data_before_publicat
             ExternalObservation.source_external_id == "review-complete"
         ))
     headers = {"Authorization": "Bearer change-me-in-production"}
-    premature = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers)
+    premature = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers, json={"expected_version": 0})
     assert premature.status_code == 409
     mapped = client.patch(
         f"/api/v1/admin/external-observations/{observation_id}/mapping", headers=headers,
-        json={"fish_slug": "pike", "waterbody_slug": "test-lake", "note": "verified fixture"},
+        json={"fish_slug": "pike", "waterbody_slug": "test-lake", "note": "verified fixture", "expected_version": 0},
     )
     assert mapped.status_code == 200
     assert mapped.json()["status"] == "ready"
-    published = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers)
+    published = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers, json={"expected_version": 1})
     assert published.status_code == 200
     assert published.json()["status"] == "published"
-    repeated = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers)
-    assert repeated.json()["catch_report_id"] == published.json()["catch_report_id"]
+    repeated = client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers, json={"expected_version": 1})
+    assert repeated.status_code == 409
     with Session(engine) as db:
         observation = db.get(ExternalObservation, observation_id)
         report = db.get(CatchReport, observation.catch_report_id)
@@ -365,13 +368,13 @@ def test_incomplete_external_observation_is_publicly_labelled_but_not_counted()
     assert all(item["x"] != 32 or item["y"] != 42 for item in client.get("/api/v1/activity").json()["items"])
     mapped = client.patch(
         f"/api/v1/admin/external-observations/{observation_id}/mapping", headers=headers,
-        json={"fish_slug": "pike", "waterbody_slug": "test-lake"},
+        json={"fish_slug": "pike", "waterbody_slug": "test-lake", "expected_version": 0},
     )
     assert mapped.json()["status"] == "mapped"
-    assert client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers).status_code == 409
+    assert client.post(f"/api/v1/admin/external-observations/{observation_id}/publish", headers=headers, json={"expected_version": 1}).status_code == 409
     rejected = client.patch(
         f"/api/v1/admin/external-observations/{observation_id}/reject", headers=headers,
-        json={"reason": "weight is absent"},
+        json={"reason": "weight is absent", "expected_version": 1},
     )
     assert rejected.json()["status"] == "rejected"
     assert all(item["id"] != str(observation_id) for item in client.get("/api/v1/community-observations").json())
@@ -430,7 +433,7 @@ def test_admin_delete_anonymizes_report_removes_screenshot_and_keeps_audit(monke
     deleted_keys: list[str] = []
     monkeypatch.setattr("app.main.delete_screenshot", deleted_keys.append)
     headers = {"Authorization": "Bearer change-me-in-production"}
-    response = client.delete(f"/api/v1/admin/catch-reports/{created['id']}", headers=headers)
+    response = client.delete(f"/api/v1/admin/catch-reports/{created['id']}?expected_version=0", headers=headers)
     assert response.status_code == 204
     assert deleted_keys == ["reports/private.jpg"]
     with Session(engine) as db:
@@ -442,7 +445,7 @@ def test_admin_delete_anonymizes_report_removes_screenshot_and_keeps_audit(monke
         event = db.query(ModerationEvent).filter_by(catch_report_id=report.id).order_by(ModerationEvent.created_at.desc()).first()
         assert event is not None
         assert event.reason == "user report deleted and anonymized"
-    assert client.delete(f"/api/v1/admin/catch-reports/{created['id']}", headers=headers).status_code == 404
+    assert client.delete(f"/api/v1/admin/catch-reports/{created['id']}?expected_version=0", headers=headers).status_code == 404
 
 
 def test_catch_report_idempotency_key_prevents_duplicates(monkeypatch) -> None: