fix(ops): db-aware health check + rate limit on paid API fan-out views

conftest.py

@@ -25,6 +25,18 @@ def pytest_configure(config) -> None:  # noqa: ARG001
     os.environ.setdefault("DJANGO_SETTINGS_MODULE", "investor_app.settings_test")
 
 
+@pytest.fixture(autouse=True)
+def _clear_django_cache():
+    """Django's cache (rate limiting, Rentometer/Walk Score/etc.) is process-
+    global — without this, state from one test leaks into the next.
+    """
+    from django.core.cache import cache
+
+    cache.clear()
+    yield
+    cache.clear()
+
+
 def pytest_collection_modifyitems(config, items) -> None:  # noqa: ARG001
     """Auto-assign test-layer markers and quarantine flaky tests.
 

core/decorators.py

@@ -0,0 +1,60 @@
+"""Shared view decorators.
+
+rate_limit / is_rate_limited guard views that fan out to paid, metered
+third-party APIs (Rentometer, HUD FMR, ATTOM via screen_property) so a user
+can't trigger an unbounded burst of outbound calls through repeated
+re-screens or previews.
+"""
+
+from __future__ import annotations
+
+import functools
+from collections.abc import Callable
+
+from django.core.cache import cache
+from django.http import HttpRequest, HttpResponse, JsonResponse
+
+
+def _increment(key: str, window_seconds: int) -> int:
+    """Atomically increment a fixed-window counter, creating it if absent."""
+    cache.add(key, 0, timeout=window_seconds)
+    return cache.incr(key)
+
+
+def is_rate_limited(
+    request: HttpRequest, key_prefix: str, limit: int, window_seconds: int
+) -> bool:
+    """True if this user (or IP, if anonymous) is over `limit` hits in the window."""
+    identity = (
+        request.user.pk
+        if request.user.is_authenticated
+        else request.META.get("REMOTE_ADDR", "anon")
+    )
+    count = _increment(f"ratelimit_{key_prefix}_{identity}", window_seconds)
+    return count > limit
+
+
+def rate_limit(
+    key_prefix: str, limit: int, window_seconds: int
+) -> Callable[[Callable[..., HttpResponse]], Callable[..., HttpResponse]]:
+    """Decorator form of is_rate_limited() for views entirely gated by one action."""
+
+    def decorator(
+        view_func: Callable[..., HttpResponse],
+    ) -> Callable[..., HttpResponse]:
+        @functools.wraps(view_func)
+        def wrapped(
+            request: HttpRequest, *args: object, **kwargs: object
+        ) -> HttpResponse:
+            if is_rate_limited(request, key_prefix, limit, window_seconds):
+                return JsonResponse(
+                    {
+                        "error": "Too many requests — please wait a few minutes and try again."
+                    },
+                    status=429,
+                )
+            return view_func(request, *args, **kwargs)
+
+        return wrapped
+
+    return decorator

core/tests/test_decorators.py

@@ -0,0 +1,60 @@
+from django.core.cache import cache
+from django.http import HttpResponse
+from django.test import RequestFactory
+
+from core.decorators import is_rate_limited, rate_limit
+
+
+def _clear_cache():
+    cache.clear()
+
+
+def test_is_rate_limited_allows_under_limit(user):
+    _clear_cache()
+    request = RequestFactory().get("/")
+    request.user = user
+
+    for _ in range(3):
+        assert is_rate_limited(request, "test_key", limit=3, window_seconds=60) is False
+
+
+def test_is_rate_limited_blocks_over_limit(user):
+    _clear_cache()
+    request = RequestFactory().get("/")
+    request.user = user
+
+    for _ in range(3):
+        is_rate_limited(request, "test_key2", limit=3, window_seconds=60)
+
+    assert is_rate_limited(request, "test_key2", limit=3, window_seconds=60) is True
+
+
+def test_is_rate_limited_keys_are_per_user(user, second_user):
+    _clear_cache()
+    request_a = RequestFactory().get("/")
+    request_a.user = user
+    request_b = RequestFactory().get("/")
+    request_b.user = second_user
+
+    for _ in range(3):
+        is_rate_limited(request_a, "test_key3", limit=3, window_seconds=60)
+
+    # A different user's requests aren't affected by user A's usage.
+    assert is_rate_limited(request_b, "test_key3", limit=3, window_seconds=60) is False
+
+
+def test_rate_limit_decorator_returns_429_over_limit(user):
+    _clear_cache()
+
+    @rate_limit("test_decorator", limit=1, window_seconds=60)
+    def view(request):
+        return HttpResponse("ok")
+
+    request = RequestFactory().get("/")
+    request.user = user
+
+    first = view(request)
+    second = view(request)
+
+    assert first.status_code == 200
+    assert second.status_code == 429

core/tests/test_views.py

@@ -1,4 +1,5 @@
 from decimal import Decimal
+from unittest.mock import patch
 
 from django.urls import reverse
 
@@ -12,13 +13,23 @@ def test_dashboard_redirects_anonymous(client):
     assert response.url.startswith("/accounts/login/")
 
 
-def test_health_check_returns_ok(client):
+def test_health_check_returns_ok(client, db):
     response = client.get(reverse("health_check"))
 
     assert response.status_code == 200
     assert response.json() == {"status": "ok"}
 
 
+def test_health_check_returns_503_when_db_unreachable(client):
+    with patch(
+        "django.db.connection.cursor", side_effect=Exception("connection refused")
+    ):
+        response = client.get(reverse("health_check"))
+
+    assert response.status_code == 503
+    assert response.json()["status"] == "error"
+
+
 def test_dashboard_returns_200_for_logged_in_user(client, user):
     client.force_login(user)
 

core/views/__init__.py

@@ -31,6 +31,7 @@
 from django.views import View
 from playwright.sync_api import sync_playwright  # type: ignore[import-not-found]
 
+from core.decorators import is_rate_limited, rate_limit
 from core.integrations.market.census import (
     discover_places_in_state,
     fetch_housing_demand_index,
@@ -158,7 +159,20 @@
 
 
 def health_check(request: HttpRequest) -> JsonResponse:
-    """Return an unauthenticated health payload for platform monitoring."""
+    """Return an unauthenticated health payload for platform monitoring.
+
+    Verifies the database is reachable — Render's healthCheckPath gates
+    deploys on this response, so a DB-down instance must not report healthy.
+    """
+    from django.db import connection
+
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT 1")
+    except Exception:
+        return JsonResponse(
+            {"status": "error", "detail": "database unreachable"}, status=503
+        )
     return JsonResponse({"status": "ok"})
 
 
@@ -2143,6 +2157,12 @@
 
     # --- Re-screen if user POSTs "rescreen" action ---
     if request.method == "POST" and request.POST.get("action") == "rescreen":
+        if is_rate_limited(request, "rescreen", limit=5, window_seconds=300):
+            messages.error(
+                request,
+                "Too many re-screens — please wait a few minutes and try again.",
+            )
+            return redirect(request.get_full_path())
         if criteria:
             # Use unfiltered queryset — rescreen ALL user properties,
             # not just the growth-area-filtered subset shown on screen
@@ -2352,23 +2372,33 @@
 
         # Re-screen all ACTIVE pipeline properties at DISCOVERED or SCREENING
         rescreen_count = 0
-        for pp in PipelineProperty.objects.filter(
-            user=request.user,
-            status=PipelineProperty.Status.ACTIVE,
-            stage__in=[
-                PipelineProperty.Stage.DISCOVERED,
-                PipelineProperty.Stage.SCREENING,
-            ],
-        ):
-            source_record = None  # Re-resolve source for accurate screening
-            from core.services.pipeline import get_source_record
+        rescreen_limited = is_rate_limited(
+            request, "rescreen_settings", limit=5, window_seconds=300
+        )
+        if not rescreen_limited:
+            for pp in PipelineProperty.objects.filter(
+                user=request.user,
+                status=PipelineProperty.Status.ACTIVE,
+                stage__in=[
+                    PipelineProperty.Stage.DISCOVERED,
+                    PipelineProperty.Stage.SCREENING,
+                ],
+            ):
+                source_record = None  # Re-resolve source for accurate screening
+                from core.services.pipeline import get_source_record
 
-            source_record = get_source_record(pp)
-            result = screen_property(pp, criteria, source_record=source_record)
-            pp.screening_passed = result.passed
-            pp.save(update_fields=["screening_passed", "updated_at"])
-            rescreen_count += 1
+                source_record = get_source_record(pp)
+                result = screen_property(pp, criteria, source_record=source_record)
+                pp.screening_passed = result.passed
+                pp.save(update_fields=["screening_passed", "updated_at"])
+                rescreen_count += 1
 
+        if rescreen_limited:
+            messages.warning(
+                request,
+                "Screening criteria saved, but re-screening was skipped "
+                "(too many re-screens recently — try again in a few minutes).",
+            )
         messages.success(
             request,
             f"Screening criteria saved. {rescreen_count} property(ies) re-screened.",
@@ -2399,6 +2429,7 @@
 
 
 @login_required
+@rate_limit("screening_preview", limit=10, window_seconds=300)
 def screening_preview(request: HttpRequest) -> HttpResponse:
     """Preview how many properties pass current criteria without saving.
 

tests/test_rentometer.py

@@ -7,7 +7,6 @@
 
 import pytest
 import requests
-from django.core.cache import cache
 
 from core.integrations.market.rentometer import (
     RentometerClient,
@@ -17,14 +16,6 @@
 )
 
 
-@pytest.fixture(autouse=True)
-def _clear_rentometer_cache():
-    """Prevent cross-test pollution — the cache is process-wide, not per-test."""
-    cache.clear()
-    yield
-    cache.clear()
-
-
 class TestRentometerClient:
     """Test suite for RentometerClient."""