pdf-quiz-generator/backend/app/routers/contact.py

"""Contact form — stores submissions and emails admin.
Table is created via raw SQL in setup_pgvector() to avoid race conditions with multiple workers.
"""
import logging

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, EmailStr
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.database import get_db
from app.config import settings
from app.models.user import User
from app.services import captcha
from app.utils.auth import require_admin

router = APIRouter()
log = logging.getLogger(__name__)


class ContactRequest(BaseModel):
    name: str
    email: EmailStr
    type: str   # "question" | "moderator"
    message: str
    captcha_token: str | None = None


@router.post("")
async def submit_contact(req: ContactRequest, db: Session = Depends(get_db)):
    """Public endpoint — no auth required."""
    if req.type not in ("question", "moderator"):
        raise HTTPException(status_code=400, detail="Type must be 'question' or 'moderator'")

    name = req.name.strip()[:120]
    message = req.message.strip()[:2000]
    if not name or not message:
        raise HTTPException(status_code=400, detail="Name and message are required")

    # Failing shut, unlike registration: if hCaptcha cannot be reached, the
    # cost of turning this form away is one retry, and the inbox behind it has
    # no other defence.
    await captcha.require_human(req.captcha_token, fail_open=False)

    db.execute(text(
        "INSERT INTO contact_submissions (name, email, type, message) VALUES (:name, :email, :type, :message)"
    ), {"name": name, "email": req.email.lower().strip(), "type": req.type, "message": message})
    db.commit()

    # Email admin
    await _email_admin(name, req.email, req.type, message)

    return {"success": True}


async def _email_admin(name: str, email: str, type_: str, message: str):
    try:
        from app.services.email_service import _send
        label = "Moderator Application" if type_ == "moderator" else "Contact Form Question"
        body = (
            f"<h2>New {label} — PedsHub</h2>"
            f"<p><strong>From:</strong> {name} &lt;{email}&gt;</p>"
            f"<p><strong>Message:</strong></p>"
            f"<blockquote style='border-left:4px solid #2563eb;padding-left:12px;color:#334155'>"
            f"{message.replace(chr(10), '<br>')}"
            f"</blockquote>"
        )
        admin_email = getattr(settings, "ADMIN_EMAIL", None) or getattr(settings, "SMTP_FROM", None)
        if admin_email:
            await _send(admin_email, f"[PedsHub] {label} from {name}", body)
    except Exception as e:
        log.warning(f"Failed to email admin about contact submission: {e}")


# ── Admin view of submissions ─────────────────────────────────────────────────

@router.get("/submissions")
def list_submissions(
    db: Session = Depends(get_db),
    current_user: User = Depends(require_admin),
):
    """Every message the contact form has taken. Administrators only.

    It was open to the internet: `require_admin` was imported inside the
    function body and never used as a dependency, so the import read as
    protection and was none. Anyone who guessed the path could read every
    sender's name, address and message.
    """
    rows = db.execute(text(
        "SELECT id, name, email, type, message, read, created_at FROM contact_submissions ORDER BY created_at DESC"
    )).fetchall()
    return [
        {
            "id": r.id, "name": r.name, "email": r.email,
            "type": r.type, "message": r.message, "read": r.read,
            # Some early rows have no timestamp, and a listing that raises on
            # one of them shows none of the others.
            "created_at": r.created_at.isoformat() if r.created_at else None,
        }
        for r in rows
    ]


@router.put("/submissions/{submission_id}/read")
def mark_read(
    submission_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_admin),
):
    """Mark one as read. Administrators only, for the same reason as the list."""
    result = db.execute(text("UPDATE contact_submissions SET read=1 WHERE id=:id"), {"id": submission_id})
    db.commit()
    if result.rowcount == 0:
        raise HTTPException(status_code=404, detail="Not found")
    return {"success": True}