fix: harden catch report idempotency

apps/api/app/routers/submissions.py

@@ -39,25 +39,14 @@ def create_catch_report(payload: CatchReportCreate, request: Request, db: Db, id
         db.expire_all()
         existing = db.scalar(select(SubmissionAttempt).where(SubmissionAttempt.idempotency_key == key_hash, SubmissionAttempt.created_at >= cutoff))
         if existing is not None:
-            logger.info("idempotent hit", extra={"idempotency_key": idempotency_key[:8]})
-            if existing.payload_hash and not hmac.compare_digest(existing.payload_hash, payload_hash):
-                raise HTTPException(status_code=409, detail="Idempotency-Key was already used with different payload")
-            if existing.catch_report is None:
-                raise HTTPException(status_code=409, detail="idempotency record is incomplete; retry with a new key")
-            replay_token = _replay_token(key_hash)
-            if not hmac.compare_digest(hashlib.sha256(replay_token.encode()).hexdigest(), existing.catch_report.screenshot_upload_token_hash or ""):
-                raise HTTPException(status_code=409, detail="idempotency record token mismatch; retry with a new key")
-            return JSONResponse(status_code=200, content=_accepted(existing.catch_report, replay_token, True))
+            return _replay_existing(existing, payload_hash, key_hash, cutoff)
         logger.info("idempotency check miss", extra={"idempotency_key": idempotency_key[:8]})
     check_rate_limit(request, db, settings)
     fish = db.scalar(select(Fish).where(Fish.slug == payload.fish_slug))
     waterbody = db.scalar(select(Waterbody).where(Waterbody.slug == payload.waterbody_slug))
     if fish is None or waterbody is None:
         raise HTTPException(status_code=422, detail="unknown fish or waterbody")
-    spot = db.scalar(select(Spot).where(Spot.waterbody_id == waterbody.id, Spot.x == payload.x, Spot.y == payload.y))
-    if spot is None:
-        spot = Spot(waterbody=waterbody, x=payload.x, y=payload.y)
-        db.add(spot)
+    spot = _spot(db, waterbody, payload.x, payload.y)
     bait = _bait(db, payload.bait_name)
     upload_token = _replay_token(key_hash) if key_hash else secrets.token_urlsafe(32)
     report = CatchReport(fish=fish, spot=spot, waterbody=waterbody, bait=bait, weight_g=payload.weight_g, fishing_method=payload.fishing_method, rig_type=payload.rig_type, retrieve_method=payload.retrieve_method, retrieve_speed=payload.retrieve_speed, caught_at=payload.caught_at, reported_at=datetime.now(timezone.utc), player_name=payload.player_name, source_type=SourceType.user, source_url=payload.source_url, source_confidence=60, moderation_status=ModerationStatus.pending, raw_payload={"comment": payload.comment} if payload.comment else None, screenshot_upload_token_hash=hashlib.sha256(upload_token.encode()).hexdigest())
@@ -77,8 +66,8 @@ def create_catch_report(payload: CatchReportCreate, request: Request, db: Db, id
         except IntegrityError:
             db.rollback()
             winner = db.scalar(select(SubmissionAttempt).where(SubmissionAttempt.idempotency_key == key_hash))
-            if winner and winner.catch_report:
-                return JSONResponse(status_code=200, content=_accepted(winner.catch_report, _replay_token(key_hash), True))
+            if winner is not None:
+                return _replay_existing(winner, payload_hash, key_hash, cutoff)
             raise
         logger.info("idempotency key stored", extra={"idempotency_key": idempotency_key[:8]})
     else:
@@ -114,12 +103,54 @@ def _accepted(report: CatchReport, token: str, idempotent: bool) -> dict[str, ob
     return {"id": str(report.id), "moderation_status": report.moderation_status.value, "screenshot_upload_token": token, "idempotent": idempotent}
 
 
+def _replay_existing(existing: SubmissionAttempt, payload_hash: str, key_hash: str, cutoff: datetime) -> JSONResponse:
+    logger.info("idempotent hit", extra={"idempotency_key": key_hash[:8]})
+    created_at = existing.created_at if existing.created_at.tzinfo else existing.created_at.replace(tzinfo=timezone.utc)
+    if created_at < cutoff:
+        raise HTTPException(status_code=409, detail="Idempotency-Key has expired; retry with a new key")
+    if existing.payload_hash and not hmac.compare_digest(existing.payload_hash, payload_hash):
+        raise HTTPException(status_code=409, detail="Idempotency-Key was already used with different payload")
+    if existing.catch_report is None:
+        raise HTTPException(status_code=409, detail="idempotency record is incomplete; retry with a new key")
+    replay_token = _replay_token(key_hash)
+    if existing.catch_report.screenshot_key is None:
+        if not hmac.compare_digest(hashlib.sha256(replay_token.encode()).hexdigest(), existing.catch_report.screenshot_upload_token_hash or ""):
+            raise HTTPException(status_code=409, detail="idempotency record token mismatch; retry with a new key")
+    return JSONResponse(status_code=200, content=_accepted(existing.catch_report, replay_token, True))
+
+
+def _spot(db: Db, waterbody: Waterbody, x: int, y: int) -> Spot:
+    spot = db.scalar(select(Spot).where(Spot.waterbody_id == waterbody.id, Spot.x == x, Spot.y == y))
+    if spot is not None:
+        return spot
+    candidate = Spot(waterbody=waterbody, x=x, y=y)
+    try:
+        with db.begin_nested():
+            db.add(candidate)
+            db.flush()
+        return candidate
+    except IntegrityError:
+        winner = db.scalar(select(Spot).where(Spot.waterbody_id == waterbody.id, Spot.x == x, Spot.y == y))
+        if winner is None:
+            raise
+        return winner
+
+
 def _bait(db: Db, value: str | None) -> Bait | None:
     if not value or not value.strip():
         return None
     key = normalize(value)
     bait = db.scalar(select(Bait).where(Bait.normalized_name == key))
-    if bait is None:
-        bait = Bait(name=value.strip(), normalized_name=key, kind=BaitKind.unknown)
-        db.add(bait)
-    return bait
+    if bait is not None:
+        return bait
+    candidate = Bait(name=value.strip(), normalized_name=key, kind=BaitKind.unknown)
+    try:
+        with db.begin_nested():
+            db.add(candidate)
+            db.flush()
+        return candidate
+    except IntegrityError:
+        winner = db.scalar(select(Bait).where(Bait.normalized_name == key))
+        if winner is None:
+            raise
+        return winner

apps/api/tests/test_api.py

@@ -538,6 +538,47 @@ def test_pending_report_accepts_one_validated_screenshot(monkeypatch) -> None:
     assert reused.status_code == 401
 
 
+def test_idempotency_replay_survives_completed_screenshot(monkeypatch) -> None:
+    import uuid
+
+    monkeypatch.setattr("app.routers.submissions.check_rate_limit", lambda *args, **kwargs: None)
+    key = f"idem-upload-{uuid.uuid4().hex}"
+    payload = {"fish_slug": "pike", "waterbody_slug": "test-lake", "x": 94, "y": 95, "weight_g": 4300}
+    created = client.post("/api/v1/catch-reports", json=payload, headers={"Idempotency-Key": key}).json()
+    monkeypatch.setattr("app.routers.submissions.upload_screenshot", lambda raw, **metadata: "reports/idempotent.jpg")
+    upload = client.post(
+        f"/api/v1/catch-reports/{created['id']}/screenshot",
+        headers={"X-Upload-Token": created["screenshot_upload_token"]},
+        files={"screenshot": ("catch.jpg", b"image-bytes", "image/jpeg")},
+    )
+    assert upload.status_code == 204
+
+    replay = client.post("/api/v1/catch-reports", json=payload, headers={"Idempotency-Key": key})
+    assert replay.status_code == 200
+    assert replay.json()["id"] == created["id"]
+    assert replay.json()["screenshot_upload_token"] == created["screenshot_upload_token"]
+    assert client.post("/api/v1/catch-reports", json=dict(payload, weight_g=4301), headers={"Idempotency-Key": key}).status_code == 409
+
+
+def test_expired_idempotency_key_integrity_winner_is_not_replayed(monkeypatch) -> None:
+    import uuid
+
+    monkeypatch.setattr("app.routers.submissions.check_rate_limit", lambda *args, **kwargs: None)
+    key = f"idem-expired-{uuid.uuid4().hex}"
+    payload = {"fish_slug": "pike", "waterbody_slug": "test-lake", "x": 96, "y": 97, "weight_g": 4400}
+    created = client.post("/api/v1/catch-reports", json=payload, headers={"Idempotency-Key": key})
+    assert created.status_code == 201
+    with Session(engine) as db:
+        attempt = db.scalar(select(SubmissionAttempt).where(SubmissionAttempt.catch_report_id == UUID(created.json()["id"])))
+        assert attempt is not None
+        attempt.created_at = datetime.now(timezone.utc) - timedelta(minutes=6)
+        db.commit()
+
+    expired = client.post("/api/v1/catch-reports", json=payload, headers={"Idempotency-Key": key})
+    assert expired.status_code == 409
+    assert "expired" in expired.json()["detail"]
+
+
 def test_admin_delete_anonymizes_report_removes_screenshot_and_keeps_audit(monkeypatch) -> None:
     created = client.post("/api/v1/catch-reports", json={"fish_slug": "pike", "waterbody_slug": "test-lake", "x": 93, "y": 94, "weight_g": 4300, "player_name": "Private Player", "source_url": "https://example.test/private", "comment": "private comment"}).json()
     with Session(engine) as db:

docs/ROADMAP.md

@@ -25,7 +25,7 @@ R-пункты уточняют критерии существующих B/G/U/
 
 - [x] **R01 · P1 · Долговременное хранение media-решений (A04/B23/A07).** Production Compose получил named `media_data` volume и отдельный root-only `media-init`: baseline из Git атомарно merge-ится только для новых assets, существующие решения/provenance не заменяются, volume передаётся API-пользователю. `backup.sh` сохраняет persistent manifest/assets в `media.tar.gz`, `restore.sh` восстанавливает их вместе с PostgreSQL и MinIO. Изолированный bootstrap подтвердил сохранение manifest после `--force-recreate` API; backup/restore drill подтвердил checksum и восстановление контрольного media-решения. Production secrets, внешнее backup-хранилище и серверный A07 acceptance-run остаются открытыми.
 - [x] **R02 · P1 · Публикация выбранных media и защита от гонок (A04/B23).** Admin API получил preview и явный набор SHA-256 asset IDs; publish больше не выбирает скрытые `upgrade_stored`, проверяет целостность выбранных файлов/вариантов, сверяет monotonic manifest version и возвращает `409` для устаревшей вкладки. Publish/rollback и CLI-совместимый путь сериализованы общим lock-файлом; решения пишутся атомарно и увеличивают версию. Regression покрывает выбор одного кандидата из двух и stale-version conflict; web admin передаёт ID и версию из заголовка manifest. Browser/production acceptance A04/A06/A07 остаётся отдельным gate.
-- [ ] **R03 · P1 · Идемпотентная отправка улова.** Сверять payload/срок ключа во всех ветках, включая IntegrityError и завершённый upload; обработать конкурентное создание Spot/Bait. Критерий: одинаковый повтор возвращает тот же результат, другой payload — 409, ожидаемые гонки не дают 500.
+- [x] **R03 · P1 · Идемпотентная отправка улова.** Единый replay-контракт теперь сверяет payload hash и TTL также после `IntegrityError`; старый уникальный ключ получает `409`, а завершённый screenshot больше не ломает повтор исходной отправки. Spot/Bait создаются через savepoint и повторный select после unique-конфликта, поэтому ожидаемые гонки не дают необработанный `500`. Regression покрывает одинаковый повтор, изменённый payload, завершённый upload и просроченный ключ.
 - [ ] **R04 · P1 · Восстановление загрузки скриншота.** Атомарное одноразовое сохранение, повтор после потерянного ответа, очистка idempotency-cookie после восстановления. Критерий: нет лишних объектов, retry завершает форму и следующий новый улов отправляется.
 - [ ] **R05 · P1 · Полноценный каталог снастей (G06/G09).** Отдельная пагинация сборок/предметов, независимые ошибки, валидация URL-фильтров, empty search отдельно от незаполненного каталога. Критерий: 49+ сборок достижимы при 0 предметов, 422 не маскируется под 503. Убрать вложенные ссылки: карточка оборачивает DataPassport с source-link; проверить итоговый DOM и keyboard наполненных карточек.
 - [ ] **R06 · P1 · Надёжный перенос плана (U05).** Preview, объединение/замена с восстановлением, однократный импорт, storage errors, вычисляемая свежесть и точное описание передачи share-данных. Критерий: ссылка не стирает план без выбора, reload не возвращает удалённое, старые данные отмечены.