fix(alerts): exactly-once Zyla request accounting and Firebase QA/prod isolation (#31, #32)

- Fixed Zyla request accounting to record each actual upstream HTTP attempt exactly once, eliminating double-counting when body parsing, JSON decoding, or normalization fails

backend/motorsignal_api/app/auth.py

@@ -6,6 +6,7 @@ import time
 import urllib.request
 import uuid
 from datetime import datetime, timedelta, timezone
+from typing import Optional
 
 import jwt
 from cryptography import x509
@@ -110,8 +111,14 @@ def _load_firebase_certs() -> dict[str, str]:
     return certs
 
 
-def decode_firebase_id_token(token: str) -> dict:
-    project_id = os.environ.get("FIREBASE_PROJECT_ID", "motoapp-d22e0")
+def decode_firebase_id_token(token: str, project_id: Optional[str] = None) -> dict:
+    target_project = project_id or os.environ.get("FIREBASE_PROJECT_ID")
+    if not target_project:
+        env = (os.environ.get("MOTORSIGNAL_ENV") or os.environ.get("ENVIRONMENT") or os.environ.get("APP_ENV") or "").lower()
+        if env in ("staging", "qa"):
+            target_project = "motorsignal-staging"
+        else:
+            target_project = "motoapp-d22e0"
     header = jwt.get_unverified_header(token)
     if header.get("alg") != "RS256" or not header.get("kid"):
         raise jwt.InvalidTokenError("Invalid Firebase token header")
@@ -119,7 +126,13 @@ def decode_firebase_id_token(token: str) -> dict:
     if certificate_pem is None:
         raise jwt.InvalidTokenError("Unknown Firebase signing key")
     public_key = x509.load_pem_x509_certificate(certificate_pem.encode("ascii")).public_key()
-    claims = jwt.decode(token, public_key, algorithms=["RS256"], audience=project_id, issuer=f"https://securetoken.google.com/{project_id}")
+    claims = jwt.decode(
+        token,
+        public_key,
+        algorithms=["RS256"],
+        audience=target_project,
+        issuer=f"https://securetoken.google.com/{target_project}",
+    )
     subject = claims.get("sub")
     if not isinstance(subject, str) or not subject or len(subject) > 128:
         raise jwt.InvalidTokenError("Invalid Firebase subject")

backend/motorsignal_api/app/road_alerts/cache.py

@@ -392,6 +392,10 @@ class PersistentUsageTracker:
     def requests_used(self) -> int:
         return self.get_stats()[1]
 
+    @property
+    def requests_executed(self) -> int:
+        return self.get_stats()[1]
+
     @property
     def requests_reserved(self) -> int:
         return self.get_stats()[0]

backend/motorsignal_api/app/road_alerts/zyla_adapter.py

@@ -376,6 +376,9 @@ class ZylaWazeAdapter:
         # Increment tracked requests
         self.request_count += 1
 
+        upstream_attempted = False
+        upstream_recorded = False
+
         try:
             # 5. Fixture handling or mock fallback
             if fixture_override:
@@ -417,13 +420,24 @@ class ZylaWazeAdapter:
                         "User-Agent": "MotorSignal-Backend/1.0",
                     },
                 )
+
+                upstream_attempted = True
+                try:
                     with urllib.request.urlopen(req, timeout=8.0) as response:
                         body = response.read()
-                    payload = json.loads(body)
-
-                # Record successful billable execution
-                if is_billable:
+                    # Upstream HTTP attempt succeeded; record exactly once
+                    if is_billable and not upstream_recorded:
                         self.usage_tracker.record_execution(success=True, now=current_time)
+                        upstream_recorded = True
+                except Exception as http_exc:
+                    # Upstream HTTP attempt failed (429, 5xx, timeout, network); record exactly once
+                    if is_billable and not upstream_recorded:
+                        self.usage_tracker.record_execution(success=False, now=current_time)
+                        upstream_recorded = True
+                    raise http_exc
+
+                # Parse JSON response (may raise json.JSONDecodeError if upstream returned malformed body)
+                payload = json.loads(body)
 
             alerts = self.parse_payload(payload, current_time)
             self.cache.set(min_lat, min_lng, max_lat, max_lng, alerts)
@@ -436,9 +450,14 @@ class ZylaWazeAdapter:
             self.error_count += 1
             self.last_error_at = current_time.isoformat()
             self.last_error_message = str(exc)
-            is_429 = "429" in str(exc)
+            is_429 = "429" in str(exc) or (hasattr(exc, "code") and getattr(exc, "code", None) == 429)
             self.circuit_breaker.record_failure(str(exc), is_rate_limit=is_429)
             logger.error("Zyla Waze API request failed: %s", exc)
             if is_billable:
+                if upstream_attempted and not upstream_recorded:
                     self.usage_tracker.record_execution(success=False, now=current_time)
+                    upstream_recorded = True
+                elif not upstream_attempted:
+                    # Request aborted before making any network call; release the reserved slot
+                    self.usage_tracker.release_reservation(now=current_time)
             return []

backend/motorsignal_api/tests/test_road_alerts.py

@@ -8,6 +8,7 @@ import unittest
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Optional
+from unittest.mock import MagicMock, patch
 
 # Add backend app directory and python packages to sys.path
 TEST_DIR = Path(__file__).resolve().parent
@@ -588,6 +589,171 @@ class TestAtomicUsageTrackerAndRollover(unittest.TestCase):
         self.assertEqual(reserved, 0)
         self.assertEqual(executed, 0)
 
+    def test_malformed_json_response_does_not_double_count_request(self):
+        """
+        P1 Accounting Bug Regression: Proves that when an upstream HTTP response
+        returns non-JSON / malformed data, the request is counted exactly once
+        and does NOT double-increment monthly counters or estimated spend.
+        """
+        tracker = PersistentUsageTracker(
+            provider_id="zyla_4814",
+            monthly_budget_limit=10,
+            cost_per_request_usd=0.005,
+            db_url=self.db_url,
+        )
+        self.trackers_to_clean.append(tracker)
+
+        adapter = ZylaWazeAdapter(
+            api_key="real-test-key",
+            enabled=True,
+            usage_tracker=tracker,
+            fixture_mode=False,
+        )
+
+        mock_response = MagicMock()
+        mock_response.read.return_value = b"<html><head><title>502 Bad Gateway</title></head><body>Bad Gateway</body></html>"
+        mock_response.__enter__.return_value = mock_response
+        mock_response.__exit__.return_value = False
+
+        with patch("urllib.request.urlopen", return_value=mock_response):
+            alerts = adapter.fetch_alerts(min_lat=34.2, min_lng=-85.2, max_lat=34.3, max_lng=-85.1)
+
+        self.assertEqual(len(alerts), 0)
+        # Must be counted exactly once, NOT double-counted!
+        reserved, executed, cost = tracker.get_stats()
+        self.assertEqual(reserved, 1)
+        self.assertEqual(executed, 1)
+        self.assertAlmostEqual(cost, 0.005, places=4)
+
+    def test_parsing_exception_in_payload_does_not_double_count_request(self):
+        """
+        P1 Accounting Bug Regression: Proves that if parse_payload or alert normalization
+        raises an unhandled exception on valid JSON payload, the HTTP request is counted
+        exactly once.
+        """
+        tracker = PersistentUsageTracker(
+            provider_id="zyla_4814",
+            monthly_budget_limit=10,
+            cost_per_request_usd=0.005,
+            db_url=self.db_url,
+        )
+        self.trackers_to_clean.append(tracker)
+
+        adapter = ZylaWazeAdapter(
+            api_key="real-test-key",
+            enabled=True,
+            usage_tracker=tracker,
+            fixture_mode=False,
+        )
+
+        mock_response = MagicMock()
+        mock_response.read.return_value = json.dumps({"alerts": []}).encode("utf-8")
+        mock_response.__enter__.return_value = mock_response
+        mock_response.__exit__.return_value = False
+
+        with patch("urllib.request.urlopen", return_value=mock_response), \
+             patch.object(adapter, "parse_payload", side_effect=RuntimeError("Parsing crash simulation")):
+            alerts = adapter.fetch_alerts(min_lat=34.2, min_lng=-85.2, max_lat=34.3, max_lng=-85.1)
+
+        self.assertEqual(len(alerts), 0)
+        reserved, executed, cost = tracker.get_stats()
+        self.assertEqual(reserved, 1)
+        self.assertEqual(executed, 1)
+        self.assertAlmostEqual(cost, 0.005, places=4)
+
+    def test_http_failures_and_timeouts_produce_accurate_counters(self):
+        """
+        P1: Proves that HTTP 429 rate limit, 500 server error, and network timeouts
+        record exactly one request execution per network attempt, without overcounting.
+        """
+        tracker = PersistentUsageTracker(
+            provider_id="zyla_4814",
+            monthly_budget_limit=10,
+            cost_per_request_usd=0.005,
+            db_url=self.db_url,
+        )
+        self.trackers_to_clean.append(tracker)
+
+        adapter = ZylaWazeAdapter(
+            api_key="real-test-key",
+            enabled=True,
+            usage_tracker=tracker,
+            fixture_mode=False,
+        )
+
+        import urllib.error
+
+        # 1. Simulate HTTP 429 Too Many Requests
+        http_429 = urllib.error.HTTPError(
+            url="https://zylalabs.com/api",
+            code=429,
+            msg="Too Many Requests",
+            hdrs={},
+            fp=None,
+        )
+        with patch("urllib.request.urlopen", side_effect=http_429):
+            alerts = adapter.fetch_alerts(min_lat=34.2, min_lng=-85.2, max_lat=34.3, max_lng=-85.1)
+
+        self.assertEqual(len(alerts), 0)
+        self.assertEqual(tracker.requests_reserved, 1)
+        self.assertEqual(tracker.requests_executed, 1)
+        self.assertEqual(adapter.circuit_breaker.state, "OPEN")
+
+        # Reset circuit breaker for next test
+        adapter.circuit_breaker.reset()
+
+        # 2. Simulate TimeoutError
+        timeout_err = TimeoutError("Connection timed out to Zyla gateway")
+        with patch("urllib.request.urlopen", side_effect=timeout_err):
+            alerts = adapter.fetch_alerts(min_lat=34.4, min_lng=-85.4, max_lat=34.5, max_lng=-85.3)
+
+        self.assertEqual(len(alerts), 0)
+        self.assertEqual(tracker.requests_reserved, 2)
+        self.assertEqual(tracker.requests_executed, 2)
+        self.assertAlmostEqual(tracker.estimated_cost_usd, 0.010, places=4)
+
+    def test_monthly_reservation_limit_strictly_enforced_against_upstream_calls(self):
+        """
+        P0: Proves that once the monthly reservation cap is reached, no further upstream
+        HTTP network attempts are initiated.
+        """
+        limit = 2
+        tracker = PersistentUsageTracker(
+            provider_id="zyla_4814",
+            monthly_budget_limit=limit,
+            cost_per_request_usd=0.005,
+            db_url=self.db_url,
+        )
+        self.trackers_to_clean.append(tracker)
+
+        adapter = ZylaWazeAdapter(
+            api_key="real-test-key",
+            enabled=True,
+            usage_tracker=tracker,
+            fixture_mode=False,
+        )
+
+        mock_response = MagicMock()
+        mock_response.read.return_value = json.dumps({"alerts": []}).encode("utf-8")
+        mock_response.__enter__.return_value = mock_response
+        mock_response.__exit__.return_value = False
+
+        with patch("urllib.request.urlopen", return_value=mock_response) as mock_urlopen:
+            # 1st call (within budget)
+            adapter.fetch_alerts(min_lat=34.0, min_lng=-85.0, max_lat=34.1, max_lng=-84.9)
+            # 2nd call (reaches budget)
+            adapter.fetch_alerts(min_lat=34.1, min_lng=-85.0, max_lat=34.2, max_lng=-84.9)
+            # 3rd call (exceeds budget)
+            adapter.fetch_alerts(min_lat=34.2, min_lng=-85.0, max_lat=34.3, max_lng=-84.9)
+            # 4th call (exceeds budget)
+            adapter.fetch_alerts(min_lat=34.3, min_lng=-85.0, max_lat=34.4, max_lng=-84.9)
+
+        # Upstream HTTP call was made EXACTLY 2 times (never 3 or 4)
+        self.assertEqual(mock_urlopen.call_count, 2)
+        self.assertEqual(tracker.requests_reserved, 2)
+        self.assertEqual(tracker.requests_executed, 2)
+        self.assertFalse(tracker.can_request())
+
 
 class TestFirebaseGatewayAuthenticationIntegration(unittest.TestCase):
     @classmethod
@@ -730,6 +896,115 @@ class TestFirebaseGatewayAuthenticationIntegration(unittest.TestCase):
             verify_admin_auth(authorization="Bearer test_admin_token_bypass_123", x_admin_key=None)
         self.assertEqual(ctx.exception.status_code, 403)
 
+    def test_qa_vs_production_firebase_project_id_isolation(self):
+        """
+        P0: Proves project-specific Firebase token validation correctly isolates
+        QA (motorsignal-staging) from Production (motoapp-d22e0).
+        """
+        qa_token = self._create_token(
+            sub="qa-rider-001",
+            aud="motorsignal-staging",
+            iss="https://securetoken.google.com/motorsignal-staging",
+        )
+        prod_token = self._create_token(
+            sub="prod-rider-001",
+            aud="motoapp-d22e0",
+            iss="https://securetoken.google.com/motoapp-d22e0",
+        )
+
+        # 1. Under QA configuration:
+        os.environ["FIREBASE_PROJECT_ID"] = "motorsignal-staging"
+        # QA token succeeds
+        self.assertTrue(verify_gateway_or_user_auth(authorization=f"Bearer {qa_token}", x_gateway_key=None))
+        # Production token is rejected with 401
+        with self.assertRaises(HTTPException) as ctx:
+            verify_gateway_or_user_auth(authorization=f"Bearer {prod_token}", x_gateway_key=None)
+        self.assertEqual(ctx.exception.status_code, 401)
+
+        # 2. Under Production configuration:
+        os.environ["FIREBASE_PROJECT_ID"] = "motoapp-d22e0"
+        # Production token succeeds
+        self.assertTrue(verify_gateway_or_user_auth(authorization=f"Bearer {prod_token}", x_gateway_key=None))
+        # QA token is rejected with 401
+        with self.assertRaises(HTTPException) as ctx:
+            verify_gateway_or_user_auth(authorization=f"Bearer {qa_token}", x_gateway_key=None)
+        self.assertEqual(ctx.exception.status_code, 401)
+
+    def test_environment_driven_firebase_project_resolution(self):
+        """
+        P0: Proves that when FIREBASE_PROJECT_ID is unconfigured, MOTORSIGNAL_ENV /
+        ENVIRONMENT selects the appropriate default (motorsignal-staging vs motoapp-d22e0).
+        """
+        os.environ.pop("FIREBASE_PROJECT_ID", None)
+
+        qa_token = self._create_token(
+            sub="qa-rider-002",
+            aud="motorsignal-staging",
+            iss="https://securetoken.google.com/motorsignal-staging",
+        )
+        prod_token = self._create_token(
+            sub="prod-rider-002",
+            aud="motoapp-d22e0",
+            iss="https://securetoken.google.com/motoapp-d22e0",
+        )
+
+        # In staging/QA environment:
+        os.environ["MOTORSIGNAL_ENV"] = "staging"
+        claims_qa = auth.decode_firebase_id_token(qa_token)
+        self.assertEqual(claims_qa["sub"], "qa-rider-002")
+
+        # In production environment:
+        os.environ["MOTORSIGNAL_ENV"] = "production"
+        claims_prod = auth.decode_firebase_id_token(prod_token)
+        self.assertEqual(claims_prod["sub"], "prod-rider-002")
+
+
+class TestAlembicMigrationCleanApplication(unittest.TestCase):
+    def test_alembic_migration_generates_valid_schema_sql(self):
+        """
+        Proves that Alembic migration f1a2b3c4d5eb applies cleanly and creates
+        the provider_monthly_usage table with its unique constraint and indexes.
+        """
+        import io
+        from alembic.config import Config
+        from alembic import command
+
+        ini_path = str(TEST_DIR.parent / "alembic.ini")
+        cfg = Config(ini_path)
+
+        orig_user = os.environ.get("DATABASE_USER")
+        orig_pw = os.environ.get("DATABASE_PASSWORD")
+        orig_db = os.environ.get("DATABASE_NAME")
+
+        os.environ["DATABASE_USER"] = "test_user"
+        os.environ["DATABASE_PASSWORD"] = "test_pass"
+        os.environ["DATABASE_NAME"] = "test_db"
+
+        buf = io.StringIO()
+        try:
+            with patch("sys.stdout", buf):
+                command.upgrade(cfg, "head", sql=True)
+            sql_output = buf.getvalue()
+        finally:
+            if orig_user is not None:
+                os.environ["DATABASE_USER"] = orig_user
+            else:
+                os.environ.pop("DATABASE_USER", None)
+            if orig_pw is not None:
+                os.environ["DATABASE_PASSWORD"] = orig_pw
+            else:
+                os.environ.pop("DATABASE_PASSWORD", None)
+            if orig_db is not None:
+                os.environ["DATABASE_NAME"] = orig_db
+            else:
+                os.environ.pop("DATABASE_NAME", None)
+
+        self.assertIn("CREATE TABLE provider_monthly_usage", sql_output)
+        self.assertIn("uq_provider_period", sql_output)
+        self.assertIn("ix_provider_monthly_usage_provider_id", sql_output)
+        self.assertIn("ix_provider_monthly_usage_period_month", sql_output)
+        self.assertIn("version_num='f1a2b3c4d5eb'", sql_output)
+
 
 if __name__ == "__main__":
     unittest.main()