fix: add AI usage quotas and cost tracking

README.md

@@ -67,6 +67,7 @@ No account required. No API key needed. Works fully offline. Fully open source.
 | **Download Results** | Export full report as `.txt` |
 | **LLM-Ready** | Plug in OpenAI, Groq, Ollama, or any OpenAI-compatible provider via env vars |
 | **Rate Limiting** | 30 requests/minute per IP - configurable |
+| **Usage Quotas** | Track estimated AI usage costs and enforce per-user or per-team quotas |
 | **Swagger Docs** | Interactive API docs at `/docs` |
 | **Gzip Compression** | Automatic response compression |
 
@@ -252,6 +253,21 @@ Create a share link for a saved analysis, then load it back by ID for seven days
 
 ---
 
+### Usage and Quotas
+
+Authenticated users can track estimated provider usage and configure quotas:
+
+| Endpoint | Detail |
+|---|---|
+| `GET /usage/summary` | Request, token, cost, and alert summary for the authenticated user or `team_id` |
+| `GET /usage/costs` | Estimated cost breakdown grouped by provider and model |
+| `POST /quotas` | Create or update user, team, or global quota limits |
+| `GET /quotas` | Return the applicable quota configuration |
+
+When a configured quota would be exceeded, `/analyze/` returns `429` before running the analysis.
+
+---
+
 ## Project Structure
 
 ```
@@ -525,4 +541,4 @@ MIT © [Darshan G K](https://github.com/imDarshanGK)
 
 Built for the open source community  ·  GSSoC 2026
 
-</div>
+</div>

backend/app/main.py

@@ -21,10 +21,12 @@
     debugging,
     explanation,
     history,
+    quotas,
     share,
     subscribe,
     suggestions,
     upload_file,
+    usage,
     user_data,
 )
 from .routers import health as health_router
@@ -157,6 +159,8 @@ async def add_cache_header(request: Request, call_next):
 app.include_router(history.router,     prefix="/history",     tags=["History"])
 app.include_router(auth.router)
 app.include_router(chat.router)
+app.include_router(usage.router)
+app.include_router(quotas.router)
 app.include_router(share.router)
 app.include_router(user_data.router)
 app.include_router(upload_file.router, prefix="/upload",      tags=['Upload File'] )

backend/app/models.py

@@ -1,6 +1,6 @@
 from datetime import UTC, datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, Text
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .database import Base
@@ -78,3 +78,44 @@ class SharedSnippet(Base):
     created_at: Mapped[datetime] = mapped_column(
         DateTime, default=lambda: datetime.now(UTC)
     )
+
+
+class UsageLog(Base):
+    """Durable record of estimated AI provider usage for one request."""
+
+    __tablename__ = "usage_logs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), nullable=True, index=True)
+    team_id: Mapped[str | None] = mapped_column(String(120), nullable=True, index=True)
+    endpoint: Mapped[str] = mapped_column(String(80), index=True)
+    provider: Mapped[str] = mapped_column(String(80), index=True)
+    model: Mapped[str] = mapped_column(String(120), index=True)
+    prompt_tokens: Mapped[int] = mapped_column(Integer, default=0)
+    completion_tokens: Mapped[int] = mapped_column(Integer, default=0)
+    total_tokens: Mapped[int] = mapped_column(Integer, default=0)
+    estimated_cost_usd: Mapped[float] = mapped_column(Float, default=0.0)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime, default=lambda: datetime.now(UTC), index=True
+    )
+
+
+class QuotaConfig(Base):
+    """Configurable usage quota for a user, team, or global scope."""
+
+    __tablename__ = "quota_configs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), nullable=True, index=True)
+    team_id: Mapped[str | None] = mapped_column(String(120), nullable=True, index=True)
+    period: Mapped[str] = mapped_column(String(20), default="monthly")
+    max_requests: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    max_tokens: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    max_cost_usd: Mapped[float | None] = mapped_column(Float, nullable=True)
+    alert_thresholds: Mapped[str] = mapped_column(String(120), default="0.8,1.0")
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime, default=lambda: datetime.now(UTC)
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)
+    )

backend/app/routers/analyze.py

@@ -8,10 +8,24 @@
 from io import BytesIO
 from pathlib import PurePosixPath
 
-from fastapi import APIRouter, File, HTTPException, Query, Request, Response, UploadFile
+from fastapi import (
+    APIRouter,
+    Depends,
+    File,
+    Header,
+    HTTPException,
+    Query,
+    Request,
+    Response,
+    UploadFile,
+)
 from fastapi.responses import StreamingResponse
+from sqlalchemy.orm import Session
 
+from ..database import get_db
+from ..models import User
 from ..schemas import AnalyzeResponse, CodeRequest, ZipAnalyzeResponse
+from ..security import get_optional_user
 from ..services.cache import cache
 from ..services.code_assistant import (
     detect_language,
@@ -20,6 +34,7 @@
     run_explanation,
     run_suggestions,
 )
+from ..services.usage import enforce_quota, estimate_usage, log_usage
 from ..sanitize import sanitize_code_input, sanitize_language_hint
 router = APIRouter()
 
@@ -192,15 +207,28 @@ async def analyze_stream_get(
     response_model=AnalyzeResponse,
     summary="Run full analysis (explain + debug + suggest)",
 )
-async def analyze(req: CodeRequest, response: Response):
+async def analyze(
+    req: CodeRequest,
+    response: Response,
+    current_user: User | None = Depends(get_optional_user),
+    db: Session = Depends(get_db),
+    team_id: str | None = Header(default=None, alias="X-Team-Id"),
+):
+    user_id = current_user.id if current_user else None
+    preflight_estimate = estimate_usage(req.code)
+    enforce_quota(db, preflight_estimate, user_id=user_id, team_id=team_id)
+
     cache_input = f"{req.language or 'auto'}\n{req.code}"
     cached_payload = cache.get("analyze:v1", cache_input)
 
     if cached_payload is not None:
         response.headers["X-Cache"] = "HIT"
+        log_usage(db, "/analyze/", preflight_estimate, user_id=user_id, team_id=team_id)
         return cached_payload
 
     payload = full_analysis(req.code, req.language)
+    usage_estimate = estimate_usage(req.code, json.dumps(payload, sort_keys=True))
+    log_usage(db, "/analyze/", usage_estimate, user_id=user_id, team_id=team_id)
 
     cache.set("analyze:v1", cache_input, payload)
 

backend/app/routers/quotas.py

@@ -0,0 +1,67 @@
+"""Quota management endpoints for usage enforcement."""
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from ..database import get_db
+from ..models import QuotaConfig, User
+from ..schemas import QuotaResponse, QuotaUpsertRequest
+from ..security import get_current_user
+from ..services.usage import ensure_usage_tables, find_applicable_quota, quota_to_dict
+
+router = APIRouter(prefix="/quotas", tags=["Quotas"])
+
+
+@router.post("", response_model=QuotaResponse)
+def upsert_quota(
+    payload: QuotaUpsertRequest,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Create or update a user, team, or global quota configuration."""
+    ensure_usage_tables(db)
+    user_id = payload.user_id
+    if payload.team_id is None:
+        user_id = current_user.id if user_id is None else user_id
+        if user_id != current_user.id:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Cannot manage another user's quota",
+            )
+
+    query = select(QuotaConfig).where(
+        QuotaConfig.user_id.is_(None)
+        if user_id is None
+        else QuotaConfig.user_id == user_id,
+        QuotaConfig.team_id.is_(None)
+        if payload.team_id is None
+        else QuotaConfig.team_id == payload.team_id,
+    )
+    quota = db.execute(query).scalar_one_or_none()
+    if quota is None:
+        quota = QuotaConfig(user_id=user_id, team_id=payload.team_id)
+        db.add(quota)
+
+    quota.period = payload.period
+    quota.max_requests = payload.max_requests
+    quota.max_tokens = payload.max_tokens
+    quota.max_cost_usd = payload.max_cost_usd
+    quota.alert_thresholds = ",".join(str(value) for value in payload.alert_thresholds)
+    db.commit()
+    db.refresh(quota)
+    return quota_to_dict(quota)
+
+
+@router.get("", response_model=QuotaResponse | None)
+def get_quota(
+    user_id: int | None = Query(default=None),
+    team_id: str | None = Query(default=None, max_length=120),
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Return the applicable quota for a user, team, or global scope."""
+    if user_id is None and team_id is None:
+        user_id = current_user.id
+    quota = find_applicable_quota(db, user_id=user_id, team_id=team_id)
+    return quota_to_dict(quota) if quota is not None else None

backend/app/routers/usage.py

@@ -0,0 +1,53 @@
+"""Usage reporting endpoints for AI provider costs and quota alerts."""
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
+
+from ..database import get_db
+from ..models import User
+from ..schemas import UsageCostsResponse, UsageSummaryResponse
+from ..security import get_current_user
+from ..services.usage import aggregate_usage, build_alerts, find_applicable_quota, provider_costs
+
+router = APIRouter(prefix="/usage", tags=["Usage"])
+
+
+@router.get("/summary", response_model=UsageSummaryResponse)
+def usage_summary(
+    period: str = Query("monthly", pattern="^(daily|monthly)$"),
+    team_id: str | None = Query(default=None, max_length=120),
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Return current usage totals for the authenticated user or a team."""
+    user_id = None if team_id else current_user.id
+    totals = aggregate_usage(db, period=period, user_id=user_id, team_id=team_id)
+    quota = find_applicable_quota(db, user_id=user_id, team_id=team_id)
+    return {
+        "scope": "team" if team_id else "user",
+        "user_id": user_id,
+        "team_id": team_id,
+        "period": period,
+        **totals,
+        "alerts": build_alerts(totals, quota),
+    }
+
+
+@router.get("/costs", response_model=UsageCostsResponse)
+def usage_costs(
+    period: str = Query("monthly", pattern="^(daily|monthly)$"),
+    team_id: str | None = Query(default=None, max_length=120),
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Return estimated usage cost grouped by provider and model."""
+    user_id = None if team_id else current_user.id
+    providers = provider_costs(db, period=period, user_id=user_id, team_id=team_id)
+    return {
+        "scope": "team" if team_id else "user",
+        "user_id": user_id,
+        "team_id": team_id,
+        "period": period,
+        "providers": providers,
+        "total_cost_usd": round(sum(item["estimated_cost_usd"] for item in providers), 6),
+    }

backend/app/schemas.py

@@ -353,6 +353,85 @@ class ChatMessageResponse(BaseModel):
     reply: str
 
 
+class UsageAlert(BaseModel):
+    """Alert emitted when usage reaches a configured quota threshold."""
+
+    metric: str
+    threshold: float
+    percent_used: float
+    message: str
+
+
+class UsageSummaryResponse(BaseModel):
+    """Aggregated usage totals for a user, team, or global scope."""
+
+    scope: str
+    user_id: int | None = None
+    team_id: str | None = None
+    period: str
+    request_count: int
+    prompt_tokens: int
+    completion_tokens: int
+    total_tokens: int
+    estimated_cost_usd: float
+    alerts: list[UsageAlert] = Field(default_factory=list)
+
+
+class UsageCostsResponse(BaseModel):
+    """Provider-level usage and estimated cost breakdown."""
+
+    scope: str
+    user_id: int | None = None
+    team_id: str | None = None
+    period: str
+    providers: list[dict[str, Any]]
+    total_cost_usd: float
+
+
+class QuotaUpsertRequest(BaseModel):
+    """Create or update quota limits for a user, team, or global scope."""
+
+    user_id: int | None = None
+    team_id: str | None = Field(default=None, max_length=120)
+    period: str = Field(default="monthly", pattern="^(daily|monthly)$")
+    max_requests: int | None = Field(default=None, gt=0)
+    max_tokens: int | None = Field(default=None, gt=0)
+    max_cost_usd: float | None = Field(default=None, gt=0)
+    alert_thresholds: list[float] = Field(default_factory=lambda: [0.8, 1.0])
+
+    @field_validator("alert_thresholds")
+    @classmethod
+    def validate_alert_thresholds(cls, value: list[float]) -> list[float]:
+        if not value:
+            return [0.8, 1.0]
+        if any(threshold <= 0 or threshold > 1 for threshold in value):
+            raise ValueError("alert thresholds must be between 0 and 1")
+        return sorted(set(value))
+
+    @model_validator(mode="after")
+    def ensure_limit_present(self) -> "QuotaUpsertRequest":
+        if (
+            self.max_requests is None
+            and self.max_tokens is None
+            and self.max_cost_usd is None
+        ):
+            raise ValueError("at least one quota limit is required")
+        return self
+
+
+class QuotaResponse(BaseModel):
+    """Stored quota configuration."""
+
+    id: int
+    user_id: int | None = None
+    team_id: str | None = None
+    period: str
+    max_requests: int | None = None
+    max_tokens: int | None = None
+    max_cost_usd: float | None = None
+    alert_thresholds: list[float]
+
+
 # ── Explanation / Debugging / Suggestions response models ───────────────────
 class ExplanationResponse(BaseModel):
     language: str