fix: enforce community cooldown per site

apps/api/app/cli.py

@@ -12,6 +12,7 @@ from .community_importer import stage_observations
 from .retention import RetentionPolicy, apply_retention
 from .storage import delete_screenshot
 from .catalog_audit import audit_catalog
+from .community_scheduler import configured_sources, run_source
 
 
 def main() -> int:
@@ -24,6 +25,8 @@ def main() -> int:
     community = sub.add_parser("stage-community-json")
     community.add_argument("--input", default="-", help="JSON array path or - for stdin")
     community.add_argument("--limit", type=int, default=500)
+    fetch_community = sub.add_parser("fetch-community")
+    fetch_community.add_argument("source", choices=configured_sources())
     cleanup = sub.add_parser("cleanup-retention")
     cleanup.add_argument("--apply", action="store_true", help="apply changes; default is dry-run")
     sub.add_parser("audit-catalog")
@@ -45,6 +48,9 @@ def main() -> int:
                 parser.error("input must be a JSON array")
             created, updated = stage_observations(session, payload[:args.limit])
             print(f"staged: created={created} updated={updated}")
+        elif args.command == "fetch-community":
+            started = run_source(args.source)
+            print("community fetch started" if started else "community fetch skipped: disabled, locked, or cooling down")
         elif args.command == "cleanup-retention":
             policy = RetentionPolicy(
                 submission_days=settings.retention_submission_days,

apps/api/app/community_scheduler.py

@@ -7,7 +7,7 @@ from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import select, text
 
-from rf4_research.community_cli import SOURCES, fetch_html
+from rf4_research.community_cli import SOURCES, fetch_html, fetch_site_key
 from rf4_research.community_sources import parse_rf4map_point, parse_rf4posts_spot
 from .community_importer import stage_observations
 from .config import settings
@@ -35,18 +35,32 @@ def configured_sources():
         "rf4posts-spot": (settings.rf4posts_spot_url, parse_rf4posts_spot),
     }
 
+def oldest_site_source(source_system: str, latest_by_source: dict[str, datetime]) -> str:
+    sources = configured_sources()
+    site = fetch_site_key(sources[source_system][0])
+    candidates = [key for key, (url, _) in sources.items() if fetch_site_key(url) == site]
+    order = {key: index for index, key in enumerate(candidates)}
+    return min(candidates, key=lambda key: (latest_by_source.get(key, datetime.min.replace(tzinfo=timezone.utc)), order[key]))
+
 def run_source(source_system: str, *, now: datetime | None = None) -> bool:
     current = now or datetime.now(timezone.utc)
     url, parser = configured_sources()[source_system]
+    site_key = fetch_site_key(url)
+    site_sources = [key for key, (candidate_url, _) in configured_sources().items() if fetch_site_key(candidate_url) == site_key]
     with SessionLocal() as session:
         source = session.get(DataSource, source_system)
         if source is None or not source.enabled:
             return False
         # Lock before reading cooldown: committing the reservation makes it visible
         # to the next contender before releasing this transaction lock.
-        if session.bind and session.bind.dialect.name == "postgresql" and not session.scalar(text("select pg_try_advisory_xact_lock(hashtext(:key))"), {"key": f"community:{source_system}"}):
+        if session.bind and session.bind.dialect.name == "postgresql" and not session.scalar(text("select pg_try_advisory_xact_lock(hashtext(:key))"), {"key": f"community-site:{site_key}"}):
+            return False
+        recent = list(session.scalars(select(CommunityImportRun).where(CommunityImportRun.source_system.in_(site_sources)).order_by(CommunityImportRun.started_at.desc()).limit(32)))
+        latest_by_source: dict[str, datetime] = {}
+        for previous in recent:
+            latest_by_source.setdefault(previous.source_system, previous.started_at if previous.started_at.tzinfo else previous.started_at.replace(tzinfo=timezone.utc))
+        if oldest_site_source(source_system, latest_by_source) != source_system:
             return False
-        recent = list(session.scalars(select(CommunityImportRun).where(CommunityImportRun.source_system == source_system).order_by(CommunityImportRun.started_at.desc()).limit(8)))
         latest = recent[0].started_at if recent else None
         delay = retry_delay([run.status for run in recent])
         if latest and (latest if latest.tzinfo else latest.replace(tzinfo=timezone.utc)) > current - timedelta(seconds=delay):

apps/api/tests/test_community_scheduler.py

@@ -1,7 +1,9 @@
 import pytest
 from pydantic import ValidationError
 
-from app.community_scheduler import MAX_BACKOFF_SECONDS, configured_sources, retry_delay
+from datetime import datetime, timedelta, timezone
+
+from app.community_scheduler import MAX_BACKOFF_SECONDS, configured_sources, oldest_site_source, retry_delay
 from app.config import Settings
 
 
@@ -19,3 +21,10 @@ def test_failed_runs_back_off_but_success_resets_delay() -> None:
     assert retry_delay(["failed", "failed", "failed"]) == 7200
     assert retry_delay(["failed"] * 20) == MAX_BACKOFF_SECONDS
     assert retry_delay(["success", "failed"]) == 1800
+
+
+def test_same_site_endpoints_rotate_by_oldest_attempt() -> None:
+    now = datetime.now(timezone.utc)
+    assert oldest_site_source("rf4stat-fishing", {}) == "rf4stat-fishing"
+    latest = {"rf4stat-fishing": now, "rf4stat-post": now - timedelta(hours=1)}
+    assert oldest_site_source("rf4stat-fishing", latest) == "rf4stat-post"