from __future__ import annotations import hashlib import json from collections.abc import Mapping from datetime import datetime from typing import Any from pydantic import ValidationError from sqlalchemy.orm import Session from govoplan_poll.backend.db.models import ( Poll, PollInvitation, PollParticipationSubmission, PollResponse, ) from govoplan_poll.backend.participation import ( ANONYMOUS_PASSWORD_REQUIREMENT, PollGovernedResponseCommand, PollParticipationPolicy, PollResponseGatewayRef, ) from govoplan_poll.backend.schemas import ( PollAnswerInput, PollParticipationPolicyInput, PollResponseGatewayInput, PollSubmitResponseRequest, ) from govoplan_poll.backend.service import ( PollError, _assert_governed_invitation_gateway_allowed, _assert_poll_accepts_responses, _existing_response, _insert_or_reconcile_identified_response, _lock_poll_for_response, _now, _update_existing_response, assert_no_sensitive_participation_metadata, get_poll_invitation_by_token, normalize_response_answers, response_datetime, ) GENERIC_PARTICIPATION_ERROR = "Poll invitation not found" MAX_COMMENT_LENGTH = 4_000 def response_gateway_payload(gateway: PollResponseGatewayRef) -> dict[str, str]: try: return PollResponseGatewayInput( module_id=gateway.module_id, resource_type=gateway.resource_type, resource_id=gateway.resource_id, ).model_dump(mode="json") except ValidationError as exc: raise PollError("Invalid participation response gateway") from exc def participation_policy_payload(policy: PollParticipationPolicy) -> dict[str, Any]: try: return PollParticipationPolicyInput( version=policy.version, single_choice=policy.single_choice, allow_maybe=policy.allow_maybe, max_participants_per_option=policy.max_participants_per_option, allow_comments=policy.allow_comments, participant_email_required=policy.participant_email_required, anonymous_password_required=policy.anonymous_password_required, ).model_dump(mode="json") except ValidationError as exc: raise PollError("Invalid participation policy") from exc def response_gateway_ref(value: Mapping[str, Any] | None) -> PollResponseGatewayRef: try: parsed = PollResponseGatewayInput.model_validate(value) except ValidationError as exc: raise PollError(GENERIC_PARTICIPATION_ERROR) from exc return PollResponseGatewayRef( module_id=parsed.module_id, resource_type=parsed.resource_type, resource_id=parsed.resource_id, ) def participation_policy_ref(value: Mapping[str, Any] | None) -> PollParticipationPolicy: try: parsed = PollParticipationPolicyInput.model_validate(value) except ValidationError as exc: raise PollError(GENERIC_PARTICIPATION_ERROR) from exc return PollParticipationPolicy( version=parsed.version, single_choice=parsed.single_choice, allow_maybe=parsed.allow_maybe, max_participants_per_option=parsed.max_participants_per_option, allow_comments=parsed.allow_comments, participant_email_required=parsed.participant_email_required, anonymous_password_required=parsed.anonymous_password_required, ) def governed_invitation( session: Session, *, token: str, gateway: PollResponseGatewayRef, lock: bool = False, ) -> PollInvitation: """Resolve a gateway-bound token without revealing which check failed.""" try: invitation = get_poll_invitation_by_token(session, token=token) _assert_governed_invitation(invitation, gateway=gateway) if lock: invitation = ( session.query(PollInvitation) .filter(PollInvitation.id == invitation.id) .populate_existing() .with_for_update() .one_or_none() ) _assert_governed_invitation(invitation, gateway=gateway) except (PollError, ValidationError) as exc: raise PollError(GENERIC_PARTICIPATION_ERROR) from exc if invitation is None: raise PollError(GENERIC_PARTICIPATION_ERROR) return invitation def _assert_governed_invitation( invitation: PollInvitation | None, *, gateway: PollResponseGatewayRef, authenticated_respondent_id: str | None = None, ) -> None: if invitation is None or invitation.poll.deleted_at is not None: raise PollError(GENERIC_PARTICIPATION_ERROR) if invitation.revoked_at is not None: raise PollError(GENERIC_PARTICIPATION_ERROR) if ( invitation.expires_at is not None and response_datetime(invitation.expires_at) <= _now() ): raise PollError(GENERIC_PARTICIPATION_ERROR) _assert_governed_invitation_gateway_allowed( poll=invitation.poll, gateway=response_gateway_payload(gateway), ) if response_gateway_ref(invitation.response_gateway_) != gateway: raise PollError(GENERIC_PARTICIPATION_ERROR) participation_policy_ref(invitation.participation_policy_) if authenticated_respondent_id is not None and ( invitation.respondent_id is not None and invitation.respondent_id != authenticated_respondent_id ): raise PollError(GENERIC_PARTICIPATION_ERROR) def governed_invitation_by_id( session: Session, *, tenant_id: str, poll_id: str, invitation_id: str, gateway: PollResponseGatewayRef, respondent_id: str, lock: bool = False, ) -> PollInvitation: """Resolve an active governed invitation for one authenticated identity.""" if not respondent_id: raise PollError(GENERIC_PARTICIPATION_ERROR) try: query = session.query(PollInvitation).filter( PollInvitation.tenant_id == tenant_id, PollInvitation.poll_id == poll_id, PollInvitation.id == invitation_id, ) if lock: query = query.populate_existing().with_for_update() invitation = query.one_or_none() _assert_governed_invitation( invitation, gateway=gateway, authenticated_respondent_id=respondent_id, ) except PollError as exc: raise PollError(GENERIC_PARTICIPATION_ERROR) from exc if invitation is None: raise PollError(GENERIC_PARTICIPATION_ERROR) return invitation def update_governed_invitation_expiry( session: Session, *, tenant_id: str, poll_id: str, invitation_id: str, gateway: PollResponseGatewayRef, expires_at: datetime | None, ) -> tuple[PollInvitation, bool]: """Update expiry without rotating the bearer token or reviving revocation.""" poll = _lock_poll_for_response( session, tenant_id=tenant_id, poll_id=poll_id, ) invitation = ( session.query(PollInvitation) .filter( PollInvitation.tenant_id == tenant_id, PollInvitation.poll_id == poll.id, PollInvitation.id == invitation_id, ) .populate_existing() .with_for_update() .one_or_none() ) try: if invitation is None or invitation.revoked_at is not None: raise PollError(GENERIC_PARTICIPATION_ERROR) _assert_governed_invitation_gateway_allowed( poll=poll, gateway=response_gateway_payload(gateway), ) if response_gateway_ref(invitation.response_gateway_) != gateway: raise PollError(GENERIC_PARTICIPATION_ERROR) participation_policy_ref(invitation.participation_policy_) except (PollError, ValidationError) as exc: raise PollError(GENERIC_PARTICIPATION_ERROR) from exc normalized_expiry = response_datetime(expires_at) replayed = response_datetime(invitation.expires_at) == normalized_expiry if not replayed: invitation.expires_at = normalized_expiry session.flush() return invitation, replayed def _normalize_email(value: str | None) -> str | None: if value is None: return None normalized = value.strip().casefold() if not normalized: return None if ( normalized.count("@") != 1 or any(character.isspace() for character in normalized) or not all(normalized.split("@", 1)) ): raise PollError("Participant email is invalid") return normalized def _canonical_respondent_id( invitation: PollInvitation, *, respondent_id: str | None, participant_email: str | None, participant_is_authenticated: bool, ) -> str: if invitation.respondent_id: return invitation.respondent_id if participant_is_authenticated and respondent_id: return respondent_id if participant_email and invitation.email is None: email_fingerprint = hashlib.sha256(participant_email.encode("utf-8")).hexdigest() return f"invitation:{invitation.id}:email:{email_fingerprint}" return f"invitation:{invitation.id}" def _normalize_comment(value: str | None) -> str | None: if value is None: return None normalized = value.strip() if len(normalized) > MAX_COMMENT_LENGTH: raise PollError(f"Participation comments cannot exceed {MAX_COMMENT_LENGTH} characters") return normalized or None def _answer_reserves_capacity(poll: Poll, answer: Mapping[str, Any]) -> bool: if poll.kind == "availability": return answer.get("value") == "available" return True def _validate_gateway_rules( *, invitation: PollInvitation, policy: PollParticipationPolicy, command: PollGovernedResponseCommand, ) -> tuple[str | None, str | None, list[PollAnswerInput]]: normalized_email = _normalize_email(command.participant_email or invitation.email) invitation_email = _normalize_email(invitation.email) if invitation_email is not None and normalized_email != invitation_email: raise PollError("Participant email does not match this invitation") if ( not command.participant_is_authenticated and policy.participant_email_required and normalized_email is None ): raise PollError("Participant email is required") if ( not command.participant_is_authenticated and policy.anonymous_password_required and ANONYMOUS_PASSWORD_REQUIREMENT not in command.verified_requirements ): raise PollError("Anonymous password verification is required") comment = _normalize_comment(command.comment) if comment is not None and not policy.allow_comments: raise PollError("Comments are not enabled for this participation link") answers = [ PollAnswerInput( option_id=answer.option_id, option_key=answer.option_key, value=answer.value, rank=answer.rank, ) for answer in command.answers ] return normalized_email, comment, answers def _validate_normalized_gateway_answers( *, poll: Poll, policy: PollParticipationPolicy, normalized_answers: list[dict[str, Any]], ) -> None: """Enforce policy against Poll-resolved option identities and values.""" if not policy.allow_maybe and any( answer.get("value") == "maybe" or answer.get("option_key") == "maybe" for answer in normalized_answers ): raise PollError("Maybe responses are not enabled for this participation link") selected_count = sum( ( answer.get("value") != "unavailable" if poll.kind == "availability" else True ) for answer in normalized_answers ) if policy.single_choice and selected_count > 1: raise PollError("Only one poll option may be selected") def _selected_capacity_option_ids( poll: Poll, normalized_answers: list[dict[str, Any]], ) -> set[str]: return { str(answer["option_id"]) for answer in normalized_answers if answer.get("option_id") and _answer_reserves_capacity(poll, answer) } def _enforce_capacity( session: Session, *, poll: Poll, existing: PollResponse | None, normalized_answers: list[dict[str, Any]], limit: int | None, ) -> None: if limit is None: return selected = _selected_capacity_option_ids(poll, normalized_answers) if not selected: return counts = dict.fromkeys(selected, 0) responses = ( session.query(PollResponse) .filter( PollResponse.tenant_id == poll.tenant_id, PollResponse.poll_id == poll.id, PollResponse.deleted_at.is_(None), ) .order_by(PollResponse.id.asc()) .all() ) for response in responses: if existing is not None and response.id == existing.id: continue for answer in response.answers or []: if ( isinstance(answer, Mapping) and answer.get("option_id") in counts and _answer_reserves_capacity(poll, answer) ): counts[str(answer["option_id"])] += 1 full = {option_id for option_id, count in counts.items() if count >= limit} if full: raise PollError("Participant limit reached for one or more poll options") def _normalize_idempotency_key(value: str | None) -> str | None: if value is None: return None normalized = value.strip() if not normalized: raise PollError("Idempotency key cannot be empty") if len(normalized) > 255: raise PollError("Idempotency key cannot be longer than 255 characters") return normalized def _submission_fingerprint( *, gateway: PollResponseGatewayRef, command: PollGovernedResponseCommand, normalized_email: str | None, comment: str | None, ) -> str: value = { "gateway": response_gateway_payload(gateway), "respondent_id": command.respondent_id, "respondent_label": command.respondent_label, "participant_email": normalized_email, "participant_is_authenticated": command.participant_is_authenticated, "answers": [ { "option_id": answer.option_id, "option_key": answer.option_key, "value": answer.value, "rank": answer.rank, } for answer in command.answers ], "comment": comment, "verified_requirements": sorted(command.verified_requirements), "metadata": dict(command.metadata), } encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _idempotent_submission( session: Session, *, invitation_id: str, idempotency_key: str | None, request_fingerprint: str, ) -> PollResponse | None: if idempotency_key is None: return None submission = ( session.query(PollParticipationSubmission) .filter( PollParticipationSubmission.invitation_id == invitation_id, PollParticipationSubmission.idempotency_key == idempotency_key, ) .one_or_none() ) if submission is None: return None if submission.request_fingerprint != request_fingerprint: raise PollError("Idempotency key was already used for a different response") response = ( session.query(PollResponse) .filter(PollResponse.id == submission.response_id) .one_or_none() ) if response is None: raise PollError("Idempotent response record is no longer available") return response def submit_governed_poll_response( session: Session, *, token: str, gateway: PollResponseGatewayRef, command: PollGovernedResponseCommand, ) -> tuple[PollInvitation, PollResponse, bool]: """Submit through a bound gateway and enforce its snapshot atomically.""" initial = governed_invitation(session, token=token, gateway=gateway) poll = _lock_poll_for_response( session, tenant_id=initial.tenant_id, poll_id=initial.poll_id, ) invitation = governed_invitation(session, token=token, gateway=gateway, lock=True) if invitation.poll_id != poll.id: raise PollError(GENERIC_PARTICIPATION_ERROR) return _submit_locked_governed_response( session, poll=poll, invitation=invitation, gateway=gateway, command=command, ) def submit_authenticated_poll_response( session: Session, *, tenant_id: str, poll_id: str, invitation_id: str, gateway: PollResponseGatewayRef, respondent_id: str, command: PollGovernedResponseCommand, ) -> tuple[PollInvitation, PollResponse, bool]: """Submit by governed invitation id for one authenticated respondent.""" if ( not command.participant_is_authenticated or not respondent_id or command.respondent_id != respondent_id ): raise PollError(GENERIC_PARTICIPATION_ERROR) initial = governed_invitation_by_id( session, tenant_id=tenant_id, poll_id=poll_id, invitation_id=invitation_id, gateway=gateway, respondent_id=respondent_id, ) poll = _lock_poll_for_response( session, tenant_id=tenant_id, poll_id=poll_id, ) invitation = governed_invitation_by_id( session, tenant_id=tenant_id, poll_id=poll_id, invitation_id=invitation_id, gateway=gateway, respondent_id=respondent_id, lock=True, ) if initial.id != invitation.id or invitation.poll_id != poll.id: raise PollError(GENERIC_PARTICIPATION_ERROR) return _submit_locked_governed_response( session, poll=poll, invitation=invitation, gateway=gateway, command=command, ) def _submit_locked_governed_response( session: Session, *, poll: Poll, invitation: PollInvitation, gateway: PollResponseGatewayRef, command: PollGovernedResponseCommand, ) -> tuple[PollInvitation, PollResponse, bool]: policy = participation_policy_ref(invitation.participation_policy_) assert_no_sensitive_participation_metadata(command.metadata) normalized_email, comment, answer_inputs = _validate_gateway_rules( invitation=invitation, policy=policy, command=command, ) respondent_id = _canonical_respondent_id( invitation, respondent_id=command.respondent_id, participant_email=normalized_email, participant_is_authenticated=command.participant_is_authenticated, ) idempotency_key = _normalize_idempotency_key(command.idempotency_key) request_fingerprint = _submission_fingerprint( gateway=gateway, command=command, normalized_email=normalized_email, comment=comment, ) replay = _idempotent_submission( session, invitation_id=invitation.id, idempotency_key=idempotency_key, request_fingerprint=request_fingerprint, ) if replay is not None: return invitation, replay, True _assert_poll_accepts_responses(poll) payload = PollSubmitResponseRequest( respondent_id=respondent_id, respondent_label=( command.respondent_label or invitation.respondent_label or invitation.email or normalized_email ), answers=answer_inputs, metadata={ key: value for key, value in command.metadata.items() if key not in {"invitation_id", "participant_email", "comment", "response_gateway"} }, ) normalized_answers = normalize_response_answers(poll, payload) _validate_normalized_gateway_answers( poll=poll, policy=policy, normalized_answers=normalized_answers, ) existing = _existing_response(session, poll=poll, respondent_id=respondent_id) _enforce_capacity( session, poll=poll, existing=existing, normalized_answers=normalized_answers, limit=policy.max_participants_per_option, ) trusted_metadata = dict(payload.metadata) trusted_metadata["invitation_id"] = invitation.id trusted_metadata["response_gateway"] = response_gateway_payload(gateway) if normalized_email is not None: trusted_metadata["participant_email"] = normalized_email if comment is not None: trusted_metadata["comment"] = comment now = _now() if existing is not None: response = _update_existing_response( session, poll=poll, response=existing, answers=normalized_answers, respondent_label=payload.respondent_label, submitted_at=now, metadata=trusted_metadata, ) else: response, _reconciled = _insert_or_reconcile_identified_response( session, poll=poll, respondent_id=respondent_id, respondent_label=payload.respondent_label, answers=normalized_answers, submitted_at=now, metadata=trusted_metadata, conflict_validator=lambda winner: _enforce_capacity( session, poll=poll, existing=winner, normalized_answers=normalized_answers, limit=policy.max_participants_per_option, ), ) if idempotency_key is not None: session.add( PollParticipationSubmission( tenant_id=poll.tenant_id, poll_id=poll.id, invitation_id=invitation.id, response_id=response.id, idempotency_key=idempotency_key, request_fingerprint=request_fingerprint, ) ) invitation.last_used_at = response.submitted_at session.flush() return invitation, response, False def response_metadata(response: PollResponse) -> tuple[str | None, str | None]: metadata = response.metadata_ or {} email = metadata.get("participant_email") comment = metadata.get("comment") return ( email if isinstance(email, str) else None, comment if isinstance(comment, str) else None, ) def response_for_invitation( session: Session, *, invitation: PollInvitation, respondent_id: str | None = None, participant_email: str | None = None, ) -> PollResponse | None: normalized_email = _normalize_email(participant_email) if invitation.respondent_id: response_respondent_id = invitation.respondent_id elif respondent_id: response_respondent_id = respondent_id elif invitation.email: response_respondent_id = f"invitation:{invitation.id}" elif normalized_email: response_respondent_id = _canonical_respondent_id( invitation, respondent_id=None, participant_email=normalized_email, participant_is_authenticated=False, ) else: return None return ( session.query(PollResponse) .filter( PollResponse.tenant_id == invitation.tenant_id, PollResponse.poll_id == invitation.poll_id, PollResponse.respondent_id == response_respondent_id, PollResponse.deleted_at.is_(None), ) .order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc()) .first() ) __all__ = [ "GENERIC_PARTICIPATION_ERROR", "MAX_COMMENT_LENGTH", "governed_invitation", "governed_invitation_by_id", "participation_policy_payload", "participation_policy_ref", "response_for_invitation", "response_gateway_payload", "response_gateway_ref", "response_metadata", "submit_authenticated_poll_response", "submit_governed_poll_response", "update_governed_invitation_expiry", ]