feat: add optimistic concurrency guards for draft editor, pricing and publish flows

add optimistic concurrency guards via expected scheme version id

protect draft editor, pricing snapshot, remap and publish flows from stale mutations
protect version creation from stale current version state

keep backward compatibility with optional query guards

verify 409 conflict behavior for stale clients and 200 for valid flows

backend/app/api/routes/pricing.py

@@ -1,6 +1,6 @@
 from decimal import Decimal
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 
 from app.core.config import settings
 from app.repositories.audit import create_audit_event
@@ -16,7 +16,6 @@ from app.repositories.pricing import (
 )
 from app.repositories.scheme_version_pricing import replace_scheme_version_pricing_snapshot
 from app.repositories.schemes import get_scheme_record_by_scheme_id
-from app.repositories.scheme_versions import get_current_scheme_version
 from app.schemas.pricing import (
     PriceRuleCreateRequest,
     PriceRuleItem,
@@ -27,18 +26,20 @@ from app.schemas.pricing import (
     PricingCategoryUpdateRequest,
 )
 from app.security.auth import require_api_key
+from app.services.draft_guard import get_current_draft_context
 
 router = APIRouter()
 
 
-async def _refresh_current_draft_snapshot_if_possible(scheme_id: str) -> dict | None:
-    scheme = await get_scheme_record_by_scheme_id(scheme_id)
-    version = await get_current_scheme_version(
-        scheme_id=scheme.scheme_id,
-        current_version_number=scheme.current_version_number,
+async def _refresh_current_draft_snapshot_if_possible(
+    *,
+    scheme_id: str,
+    expected_scheme_version_id: str | None = None,
+) -> dict | None:
+    scheme, version = await get_current_draft_context(
+        scheme_id=scheme_id,
+        expected_scheme_version_id=expected_scheme_version_id,
     )
-    if scheme.status != "draft" or version.status != "draft":
-        return None
 
     return await replace_scheme_version_pricing_snapshot(
         scheme_id=scheme.scheme_id,
@@ -82,26 +83,41 @@ async def get_pricing_bundle(scheme_id: str, role: str = Depends(require_api_key
 async def create_pricing_category_endpoint(
     scheme_id: str,
     payload: PricingCategoryCreateRequest,
+    expected_scheme_version_id: str | None = Query(default=None),
     role: str = Depends(require_api_key),
 ):
-    pricing_category_id = await create_pricing_category(
+    scheme, version = await get_current_draft_context(
         scheme_id=scheme_id,
+        expected_scheme_version_id=expected_scheme_version_id,
+    )
+
+    pricing_category_id = await create_pricing_category(
+        scheme_id=scheme.scheme_id,
         name=payload.name,
         code=payload.code,
     )
-    snapshot = await _refresh_current_draft_snapshot_if_possible(scheme_id)
+    snapshot = await _refresh_current_draft_snapshot_if_possible(
+        scheme_id=scheme.scheme_id,
+        expected_scheme_version_id=version.scheme_version_id,
+    )
 
     await create_audit_event(
-        scheme_id=scheme_id,
+        scheme_id=scheme.scheme_id,
         event_type="pricing.category.created",
         object_type="pricing_category",
         object_ref=pricing_category_id,
-        details={"name": payload.name, "code": payload.code, "snapshot": snapshot},
+        details={
+            "name": payload.name,
+            "code": payload.code,
+            "scheme_version_id": version.scheme_version_id,
+            "snapshot": snapshot,
+        },
     )
 
     return {
         "pricing_category_id": pricing_category_id,
-        "scheme_id": scheme_id,
+        "scheme_id": scheme.scheme_id,
+        "scheme_version_id": version.scheme_version_id,
         "name": payload.name,
         "code": payload.code,
     }
@@ -112,27 +128,42 @@ async def update_pricing_category_endpoint(
     scheme_id: str,
     pricing_category_id: str,
     payload: PricingCategoryUpdateRequest,
+    expected_scheme_version_id: str | None = Query(default=None),
     role: str = Depends(require_api_key),
 ):
-    row = await update_pricing_category(
+    scheme, version = await get_current_draft_context(
         scheme_id=scheme_id,
+        expected_scheme_version_id=expected_scheme_version_id,
+    )
+
+    row = await update_pricing_category(
+        scheme_id=scheme.scheme_id,
         pricing_category_id=pricing_category_id,
         name=payload.name,
         code=payload.code,
     )
-    snapshot = await _refresh_current_draft_snapshot_if_possible(scheme_id)
+    snapshot = await _refresh_current_draft_snapshot_if_possible(
+        scheme_id=scheme.scheme_id,
+        expected_scheme_version_id=version.scheme_version_id,
+    )
 
     await create_audit_event(
-        scheme_id=scheme_id,
+        scheme_id=scheme.scheme_id,
         event_type="pricing.category.updated",
         object_type="pricing_category",
         object_ref=pricing_category_id,
-        details={"name": payload.name, "code": payload.code, "snapshot": snapshot},
+        details={
+            "name": payload.name,
+            "code": payload.code,
+            "scheme_version_id": version.scheme_version_id,
+            "snapshot": snapshot,
+        },
     )
 
     return {
         "pricing_category_id": row.pricing_category_id,
         "scheme_id": row.scheme_id,
+        "scheme_version_id": version.scheme_version_id,
         "name": row.name,
         "code": row.code,
     }
@@ -142,31 +173,53 @@ async def update_pricing_category_endpoint(
 async def delete_pricing_category_endpoint(
     scheme_id: str,
     pricing_category_id: str,
+    expected_scheme_version_id: str | None = Query(default=None),
     role: str = Depends(require_api_key),
 ):
-    await delete_pricing_category(
+    scheme, version = await get_current_draft_context(
         scheme_id=scheme_id,
+        expected_scheme_version_id=expected_scheme_version_id,
+    )
+
+    await delete_pricing_category(
+        scheme_id=scheme.scheme_id,
         pricing_category_id=pricing_category_id,
     )
-    snapshot = await _refresh_current_draft_snapshot_if_possible(scheme_id)
+    snapshot = await _refresh_current_draft_snapshot_if_possible(
+        scheme_id=scheme.scheme_id,
+        expected_scheme_version_id=version.scheme_version_id,
+    )
 
     await create_audit_event(
-        scheme_id=scheme_id,
+        scheme_id=scheme.scheme_id,
         event_type="pricing.category.deleted",
         object_type="pricing_category",
         object_ref=pricing_category_id,
-        details={"snapshot": snapshot},
+        details={
+            "scheme_version_id": version.scheme_version_id,
+            "snapshot": snapshot,
+        },
     )
 
-    return {"deleted": True, "pricing_category_id": pricing_category_id}
+    return {
+        "deleted": True,
+        "pricing_category_id": pricing_category_id,
+        "scheme_version_id": version.scheme_version_id,
+    }
 
 
 @router.post(f"{settings.api_v1_prefix}/schemes/{{scheme_id}}/pricing/rules")
 async def create_price_rule_endpoint(
     scheme_id: str,
     payload: PriceRuleCreateRequest,
+    expected_scheme_version_id: str | None = Query(default=None),
     role: str = Depends(require_api_key),
 ):
+    scheme, version = await get_current_draft_context(
+        scheme_id=scheme_id,
+        expected_scheme_version_id=expected_scheme_version_id,
+    )
+
     try:
         amount = Decimal(payload.amount)
     except Exception:
@@ -176,17 +229,20 @@ async def create_price_rule_endpoint(
         )
 
     price_rule_id = await create_price_rule(
-        scheme_id=scheme_id,
+        scheme_id=scheme.scheme_id,
         pricing_category_id=payload.pricing_category_id,
         target_type=payload.target_type,
         target_ref=payload.target_ref,
         amount=amount,
         currency=payload.currency,
     )
-    snapshot = await _refresh_current_draft_snapshot_if_possible(scheme_id)
+    snapshot = await _refresh_current_draft_snapshot_if_possible(
+        scheme_id=scheme.scheme_id,
+        expected_scheme_version_id=version.scheme_version_id,
+    )
 
     await create_audit_event(
-        scheme_id=scheme_id,
+        scheme_id=scheme.scheme_id,
         event_type="pricing.rule.created",
         object_type="price_rule",
         object_ref=price_rule_id,
@@ -196,13 +252,15 @@ async def create_price_rule_endpoint(
             "target_ref": payload.target_ref,
             "amount": payload.amount,
             "currency": payload.currency,
+            "scheme_version_id": version.scheme_version_id,
             "snapshot": snapshot,
         },
     )
 
     return {
         "price_rule_id": price_rule_id,
-        "scheme_id": scheme_id,
+        "scheme_id": scheme.scheme_id,
+        "scheme_version_id": version.scheme_version_id,
         "pricing_category_id": payload.pricing_category_id,
         "target_type": payload.target_type,
         "target_ref": payload.target_ref,
@@ -216,8 +274,14 @@ async def update_price_rule_endpoint(
     scheme_id: str,
     price_rule_id: str,
     payload: PriceRuleUpdateRequest,
+    expected_scheme_version_id: str | None = Query(default=None),
     role: str = Depends(require_api_key),
 ):
+    scheme, version = await get_current_draft_context(
+        scheme_id=scheme_id,
+        expected_scheme_version_id=expected_scheme_version_id,
+    )
+
     try:
         amount = Decimal(payload.amount)
     except Exception:
@@ -227,7 +291,7 @@ async def update_price_rule_endpoint(
         )
 
     row = await update_price_rule(
-        scheme_id=scheme_id,
+        scheme_id=scheme.scheme_id,
         price_rule_id=price_rule_id,
         pricing_category_id=payload.pricing_category_id,
         target_type=payload.target_type,
@@ -235,10 +299,13 @@ async def update_price_rule_endpoint(
         amount=amount,
         currency=payload.currency,
     )
-    snapshot = await _refresh_current_draft_snapshot_if_possible(scheme_id)
+    snapshot = await _refresh_current_draft_snapshot_if_possible(
+        scheme_id=scheme.scheme_id,
+        expected_scheme_version_id=version.scheme_version_id,
+    )
 
     await create_audit_event(
-        scheme_id=scheme_id,
+        scheme_id=scheme.scheme_id,
         event_type="pricing.rule.updated",
         object_type="price_rule",
         object_ref=price_rule_id,
@@ -248,6 +315,7 @@ async def update_price_rule_endpoint(
             "target_ref": payload.target_ref,
             "amount": payload.amount,
             "currency": payload.currency,
+            "scheme_version_id": version.scheme_version_id,
             "snapshot": snapshot,
         },
     )
@@ -255,6 +323,7 @@ async def update_price_rule_endpoint(
     return {
         "price_rule_id": row.price_rule_id,
         "scheme_id": row.scheme_id,
+        "scheme_version_id": version.scheme_version_id,
         "pricing_category_id": row.pricing_category_id,
         "target_type": row.target_type,
         "target_ref": row.target_ref,
@@ -267,20 +336,36 @@ async def update_price_rule_endpoint(
 async def delete_price_rule_endpoint(
     scheme_id: str,
     price_rule_id: str,
+    expected_scheme_version_id: str | None = Query(default=None),
     role: str = Depends(require_api_key),
 ):
-    await delete_price_rule(
+    scheme, version = await get_current_draft_context(
         scheme_id=scheme_id,
+        expected_scheme_version_id=expected_scheme_version_id,
+    )
+
+    await delete_price_rule(
+        scheme_id=scheme.scheme_id,
         price_rule_id=price_rule_id,
     )
-    snapshot = await _refresh_current_draft_snapshot_if_possible(scheme_id)
+    snapshot = await _refresh_current_draft_snapshot_if_possible(
+        scheme_id=scheme.scheme_id,
+        expected_scheme_version_id=version.scheme_version_id,
+    )
 
     await create_audit_event(
-        scheme_id=scheme_id,
+        scheme_id=scheme.scheme_id,
         event_type="pricing.rule.deleted",
         object_type="price_rule",
         object_ref=price_rule_id,
-        details={"snapshot": snapshot},
+        details={
+            "scheme_version_id": version.scheme_version_id,
+            "snapshot": snapshot,
+        },
     )
 
-    return {"deleted": True, "price_rule_id": price_rule_id}
+    return {
+        "deleted": True,
+        "price_rule_id": price_rule_id,
+        "scheme_version_id": version.scheme_version_id,
+    }