""" Per-voter approval tokens. Stateless signed tokens (itsdangerous) keyed on settings.SECRET_KEY with a versioned salt. Single-use is enforced by the presence of a MealPlanVote row for (item, voter) — the table already has UniqueConstraint on that pair, so the DB is the source of truth, not a token-status column. """ from __future__ import annotations from typing import TYPE_CHECKING from uuid import UUID from fastapi import HTTPException from itsdangerous import ( BadSignature, SignatureExpired, URLSafeTimedSerializer, ) from app.config import settings from app.models import FamilyMember, MealPlanVote if TYPE_CHECKING: # pragma: no cover from sqlalchemy.orm import Session SALT = "meal-approval-v1" DEFAULT_MAX_AGE_SECONDS = 7 * 24 * 3600 def _serializer() -> URLSafeTimedSerializer: return URLSafeTimedSerializer(secret_key=settings.SECRET_KEY, salt=SALT) def issue_token(meal_plan_item_id: UUID, family_member_id: UUID) -> str: payload = { "item": str(meal_plan_item_id), "voter": str(family_member_id), } return _serializer().dumps(payload) def verify_token(token: str, max_age_seconds: int = DEFAULT_MAX_AGE_SECONDS) -> dict: try: payload = _serializer().loads(token, max_age=max_age_seconds) except SignatureExpired: raise HTTPException(status_code=401, detail="Token expired") except BadSignature: raise HTTPException(status_code=401, detail="Invalid token") if not isinstance(payload, dict) or "item" not in payload or "voter" not in payload: raise HTTPException(status_code=401, detail="Invalid token payload") return payload def consume_token( db: "Session", token: str, meal_plan_item_id: UUID, max_age_seconds: int = DEFAULT_MAX_AGE_SECONDS, ) -> FamilyMember: """Verify + match URL + enforce single-use. Returns the voter on success. Single-use is checked by looking for an existing MealPlanVote row for (item, voter). If one exists, raise 409. """ payload = verify_token(token, max_age_seconds=max_age_seconds) if str(payload["item"]) != str(meal_plan_item_id): raise HTTPException(status_code=400, detail="Token not valid for this meal") voter_id = UUID(str(payload["voter"])) voter = db.query(FamilyMember).filter(FamilyMember.id == voter_id).first() if not voter: raise HTTPException(status_code=404, detail="Voter not found") existing = ( db.query(MealPlanVote) .filter( MealPlanVote.meal_plan_item_id == meal_plan_item_id, MealPlanVote.family_member_id == voter_id, ) .first() ) if existing: raise HTTPException(status_code=409, detail="Already voted") return voter