app/api/routes/billing.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 
 from typing import Any
 
-from fastapi import APIRouter, Depends, Header, Request, status
+from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
 from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -96,3 +96,37 @@ async def list_invoices(
     """
     invoices = await stripe_service.list_invoices(current_user.id, db)
     return invoices
+
+
+# ── Quota check ────────────────────────────────────────────────────────
+
+from app.billing.quota import check_folder_quota, QuotaExceeded  # noqa: E402
+
+
+class QuotaCheckRequest(BaseModel):
+    feature: str
+    estimated_files: int
+
+
+@router.post("/quota/check")
+async def quota_check(
+    payload: QuotaCheckRequest,
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> dict:
+    """Pre-flight folder quota check. 402 if tier limits would be exceeded."""
+    if payload.feature != "folder_index":
+        raise HTTPException(status_code=400, detail="Unknown feature")
+    try:
+        await check_folder_quota(
+            user_id=current_user.id,
+            tier=current_user.tier,
+            estimated_files=payload.estimated_files,
+            db=db,
+        )
+    except QuotaExceeded as exc:
+        raise HTTPException(
+            status_code=402,
+            detail={"reason": exc.reason, "message": str(exc)},
+        )
+    return {"ok": True}

tests/conftest.py

@@ -162,6 +162,12 @@ async def test_user_power(db_session: AsyncSession):
     return result.scalar_one()
 
 
+@pytest.fixture
+def auth_headers_free() -> dict[str, str]:
+    """Authorization header for the seeded free-tier user."""
+    return auth_header("free")
+
+
 # ── CLI options ───────────────────────────────────────────────────────
 
 def pytest_addoption(parser):

tests/test_folder_quota.py

@@ -71,3 +71,24 @@ async def test_add_token_usage_returns_exhausted_when_over_cap(db, test_user_fre
     )
     assert result.exhausted is True
     assert result.tokens_used == 150_000
+
+
+def test_quota_check_endpoint_rejects(client, auth_headers_free):
+    res = client.post(
+        "/api/v1/billing/quota/check",
+        json={"feature": "folder_index", "estimated_files": 500},
+        headers=auth_headers_free,
+    )
+    assert res.status_code == 402
+    body = res.json()
+    assert body["detail"]["reason"] == "max_files"
+
+
+def test_quota_check_endpoint_passes(client, auth_headers_free):
+    res = client.post(
+        "/api/v1/billing/quota/check",
+        json={"feature": "folder_index", "estimated_files": 50},
+        headers=auth_headers_free,
+    )
+    assert res.status_code == 200
+    assert res.json() == {"ok": True}