From 9b029c8edfd62be2a20f9539b5c20687ac34638e Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 29 Jul 2026 17:21:13 +0200 Subject: [PATCH] Refactor scheduling reconciliation into change plans --- src/govoplan_scheduling/backend/router.py | 1 - src/govoplan_scheduling/backend/service.py | 1816 ++++++++++++-------- tests/test_reconciliation_plans.py | 283 +++ tests/test_response_editing.py | 18 +- 4 files changed, 1397 insertions(+), 721 deletions(-) create mode 100644 tests/test_reconciliation_plans.py diff --git a/src/govoplan_scheduling/backend/router.py b/src/govoplan_scheduling/backend/router.py index d917351..018eac9 100644 --- a/src/govoplan_scheduling/backend/router.py +++ b/src/govoplan_scheduling/backend/router.py @@ -59,7 +59,6 @@ from govoplan_scheduling.backend.service import ( list_visible_scheduling_requests, issue_scheduling_participant_invitation, open_scheduling_request, - refresh_participant_response_state, require_visible_scheduling_results, revoke_scheduling_participant_invitation, scheduling_notification_response, diff --git a/src/govoplan_scheduling/backend/service.py b/src/govoplan_scheduling/backend/service.py index 195de17..4423fe4 100644 --- a/src/govoplan_scheduling/backend/service.py +++ b/src/govoplan_scheduling/backend/service.py @@ -1746,41 +1746,67 @@ def get_scheduling_availability_response( participant, fallback_respondent_id=respondent_id, ) - response: PollResponseRef | None = None - if not ( + response = _availability_poll_response( + session, + request=request, + participant=participant, + actor_ids=actor_ids, + canonical_respondent_id=canonical_respondent_id, + ) + answers = _availability_response_answers(request, response) + return { + "request_id": request.id, + "participant_id": participant.id, + "has_response": response is not None, + "submitted_at": response_datetime(response.submitted_at) if response is not None else None, + "answers": answers, + "comment": participant.response_comment if request.allow_comments else None, + } + + +def _availability_poll_response( + session: Session, + *, + request: SchedulingRequest, + participant: SchedulingParticipant, + actor_ids: tuple[str, ...], + canonical_respondent_id: str, +) -> PollResponseRef | None: + governed_invitation = ( participant.poll_invitation_id is not None and participant.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY - ): - response = _participant_poll_response( + ) + if not governed_invitation: + return _participant_poll_response( session, request=request, participant=participant, actor_ids=actor_ids, canonical_respondent_id=canonical_respondent_id, ) - if response is not None and response.respondent_id: - canonical_respondent_id = response.respondent_id - if ( - participant.poll_invitation_id is not None - and participant.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY - ): - provider = _poll_participation_provider() - if provider is None: - raise SchedulingError( - "Poll governed participation capability is unavailable" - ) - try: - context = provider.resolve_authenticated_participation( - session, - tenant_id=request.tenant_id, - poll_id=request.poll_id, - invitation_id=participant.poll_invitation_id, - gateway=_public_participation_gateway(request.id), - respondent_id=canonical_respondent_id, - ) - except PollCapabilityError as exc: - raise SchedulingError(str(exc)) from exc - response = context.response.response if context.response is not None else None + provider = _poll_participation_provider() + if provider is None: + raise SchedulingError("Poll governed participation capability is unavailable") + try: + context = provider.resolve_authenticated_participation( + session, + tenant_id=request.tenant_id, + poll_id=cast(str, request.poll_id), + invitation_id=cast(str, participant.poll_invitation_id), + gateway=_public_participation_gateway(request.id), + respondent_id=canonical_respondent_id, + ) + except PollCapabilityError as exc: + raise SchedulingError(str(exc)) from exc + return context.response.response if context.response is not None else None + + +def _availability_response_answers( + request: SchedulingRequest, + response: PollResponseRef | None, +) -> list[dict[str, str]]: + if response is None: + return [] slot_by_option_id = { slot.poll_option_id: slot for slot in _active_slots(request) @@ -1791,20 +1817,11 @@ def get_scheduling_availability_response( for slot in _active_slots(request) } answers: list[dict[str, str]] = [] - if response is not None: - for answer in response.answers: - slot = slot_by_option_id.get(answer.option_id) or slot_by_option_key.get(answer.option_key) - if slot is None or answer.value not in {"available", "maybe", "unavailable"}: - continue + for answer in response.answers: + slot = slot_by_option_id.get(answer.option_id) or slot_by_option_key.get(answer.option_key) + if slot is not None and answer.value in {"available", "maybe", "unavailable"}: answers.append({"slot_id": slot.id, "value": str(answer.value)}) - return { - "request_id": request.id, - "participant_id": participant.id, - "has_response": response is not None, - "submitted_at": response_datetime(response.submitted_at) if response is not None else None, - "answers": answers, - "comment": participant.response_comment if request.allow_comments else None, - } + return answers def _resolve_public_scheduling_participation( @@ -1820,6 +1837,33 @@ def _resolve_public_scheduling_participation( SchedulingParticipant, PollParticipationContextRef, ]: + request = _public_scheduling_request(session, request_id=request_id) + if lock_for_submission: + request = _lock_public_scheduling_submission(session, request) + password_dimensions = _verify_public_participation_password( + request, + token=token, + payload=payload, + client_address=client_address, + ) + context = _resolve_public_poll_context( + session, + request=request, + token=token, + participant_email=payload.participant_email, + ) + participant = _public_context_participant(request, context) + _validate_public_participant_email(participant, payload.participant_email) + if password_dimensions: + _participation_password_throttle().reset(password_dimensions[:1]) + return request, participant, context + + +def _public_scheduling_request( + session: Session, + *, + request_id: str, +) -> SchedulingRequest: request = ( session.query(SchedulingRequest) .filter( @@ -1830,65 +1874,66 @@ def _resolve_public_scheduling_participation( ) if request is None: raise SchedulingPublicParticipationError() - if request.status == "cancelled": notice_until = response_datetime(request.cancellation_notice_until) if notice_until is None or notice_until <= _now(): raise SchedulingPublicParticipationError() + return request - if lock_for_submission: - # Scheduling mutations acquire their request/participant locks before - # entering Poll. Public response submission must use the same order: - # otherwise a participant reconciliation (Scheduling -> Poll) can - # deadlock with a response (Poll -> Scheduling), and two first - # responses can both observe an unbound participant email. - try: - request = _lock_scheduling_request( - session, - tenant_id=request.tenant_id, - request_id=request.id, - lock_participants=True, - ) - except SchedulingError as exc: - raise SchedulingPublicParticipationError() from exc - try: - _require_scheduling_response_collection_open(request) - except SchedulingError as exc: - # Public participation deliberately does not disclose whether a - # token, lifecycle state, or deadline caused rejection. - raise SchedulingPublicParticipationError() from exc - password_dimensions: tuple[ThrottleDimension, ...] = () - if request.anonymous_password_protection_enabled: - password_dimensions = _participation_password_dimensions( - request, - token=token, - client_address=client_address, +def _lock_public_scheduling_submission( + session: Session, + request: SchedulingRequest, +) -> SchedulingRequest: + # Keep Scheduling -> Poll lock order consistent with participant reconciliation. + try: + locked = _lock_scheduling_request( + session, + tenant_id=request.tenant_id, + request_id=request.id, + lock_participants=True, ) - throttle = _participation_password_throttle() - decision = throttle.check(password_dimensions) - if not decision.allowed: - raise SchedulingPublicParticipationError( - retry_after_seconds=decision.retry_after_seconds, - ) - if payload.password is None: - # The UI first probes whether the link needs credentials. Missing - # credentials are not a failed password attempt. - raise SchedulingPublicParticipationError() - supplied_password = payload.password.get_secret_value() - if not verify_participant_password( - supplied_password, - request.anonymous_password_hash, - ): - decision = throttle.record(password_dimensions) - raise SchedulingPublicParticipationError( - retry_after_seconds=( - decision.retry_after_seconds - if not decision.allowed - else 0 - ), - ) + _require_scheduling_response_collection_open(locked) + except SchedulingError as exc: + raise SchedulingPublicParticipationError() from exc + return locked + +def _verify_public_participation_password( + request: SchedulingRequest, + *, + token: str, + payload: SchedulingPublicParticipationAccessRequest, + client_address: str | None, +) -> tuple[ThrottleDimension, ...]: + if not request.anonymous_password_protection_enabled: + return () + dimensions = _participation_password_dimensions( + request, + token=token, + client_address=client_address, + ) + throttle = _participation_password_throttle() + decision = throttle.check(dimensions) + if not decision.allowed: + raise SchedulingPublicParticipationError(retry_after_seconds=decision.retry_after_seconds) + if payload.password is None: + raise SchedulingPublicParticipationError() + supplied_password = payload.password.get_secret_value() + if verify_participant_password(supplied_password, request.anonymous_password_hash): + return dimensions + decision = throttle.record(dimensions) + retry_after = decision.retry_after_seconds if not decision.allowed else 0 + raise SchedulingPublicParticipationError(retry_after_seconds=retry_after) + + +def _resolve_public_poll_context( + session: Session, + *, + request: SchedulingRequest, + token: str, + participant_email: str | None, +) -> PollParticipationContextRef: provider = _poll_participation_provider() if provider is None: raise SchedulingPublicParticipationError() @@ -1898,7 +1943,7 @@ def _resolve_public_scheduling_participation( session, token=token, gateway=gateway, - participant_email=payload.participant_email, + participant_email=participant_email, participant_is_authenticated=False, verified_requirements=( frozenset({ANONYMOUS_PASSWORD_REQUIREMENT}) @@ -1914,6 +1959,13 @@ def _resolve_public_scheduling_participation( or context.gateway != gateway ): raise SchedulingPublicParticipationError() + return context + + +def _public_context_participant( + request: SchedulingRequest, + context: PollParticipationContextRef, +) -> SchedulingParticipant: participant = next( ( item @@ -1925,18 +1977,20 @@ def _resolve_public_scheduling_participation( ) if participant is None: raise SchedulingPublicParticipationError() + return participant + + +def _validate_public_participant_email( + participant: SchedulingParticipant, + supplied_email: str | None, +) -> None: if ( participant.email is not None - and payload.participant_email is not None + and supplied_email is not None and participant.email.strip().casefold() - != payload.participant_email.strip().casefold() + != supplied_email.strip().casefold() ): raise SchedulingPublicParticipationError() - if password_dimensions: - # Clear only the token-specific bucket. The broader request/client - # dimension deliberately retains failures to resist token rotation. - _participation_password_throttle().reset(password_dimensions[:1]) - return request, participant, context def _public_scheduling_participation_response( @@ -2181,6 +2235,207 @@ def require_visible_scheduling_results( raise SchedulingError("Scheduling results are not visible") +REQUEST_UPDATE_FIELDS = ( + "title", + "description", + "location", + "deadline_at", + "allow_external_participants", + "allow_participant_updates", + "result_visibility", + "participant_visibility", + "notify_on_answers", + "single_choice", + "allow_maybe", + "allow_comments", + "participant_email_required", +) +NULLABLE_REQUEST_UPDATE_FIELDS = frozenset({"description", "location", "deadline_at"}) +PUBLIC_PARTICIPATION_POLICY_FIELDS = ( + "single_choice", + "allow_maybe", + "allow_comments", + "participant_email_required", + "anonymous_password_protection_enabled", +) + + +@dataclass(frozen=True, slots=True) +class SchedulingRequestUpdatePlan: + field_updates: tuple[tuple[str, Any], ...] + deadline_changed: bool + retained_invitation_ids: tuple[tuple[str, str], ...] + update_max_participants: bool + max_participants_per_option: int | None + password_protection_update: bool | None + + +def _public_participation_policy_changed( + request: SchedulingRequest, + payload: SchedulingRequestUpdateRequest, +) -> bool: + scalar_changed = any( + field in payload.model_fields_set + and getattr(payload, field) is not None + and getattr(payload, field) != getattr(request, field) + for field in PUBLIC_PARTICIPATION_POLICY_FIELDS + ) + capacity_changed = ( + "max_participants_per_option" in payload.model_fields_set + and payload.max_participants_per_option != request.max_participants_per_option + ) + return scalar_changed or capacity_changed + + +def _plan_scheduling_request_update( + *, + request: SchedulingRequest, + payload: SchedulingRequestUpdateRequest, + participation_available: bool, +) -> SchedulingRequestUpdatePlan: + if request.status in {"decided", "handed_off", "cancelled", "archived"}: + raise SchedulingError("Decided, handed off, cancelled, or archived scheduling requests cannot be edited") + deadline_changed = ( + "deadline_at" in payload.model_fields_set + and response_datetime(payload.deadline_at) != response_datetime(request.deadline_at) + ) + retained_invitation_ids = _retained_participant_invitation_ids(request) + if deadline_changed and retained_invitation_ids and not participation_available: + raise SchedulingError("Poll participation invitation rotation capability is unavailable") + if _public_participation_policy_changed(request, payload) and retained_invitation_ids: + raise SchedulingError( + "Public participation policy cannot change after invitation links are issued" + ) + protection_update = ( + payload.anonymous_password_protection_enabled + if "anonymous_password_protection_enabled" in payload.model_fields_set + else None + ) + _validate_request_password_update( + request, + payload=payload, + protection_update=protection_update, + ) + return SchedulingRequestUpdatePlan( + field_updates=_request_field_updates(payload), + deadline_changed=deadline_changed, + retained_invitation_ids=retained_invitation_ids, + update_max_participants="max_participants_per_option" in payload.model_fields_set, + max_participants_per_option=payload.max_participants_per_option, + password_protection_update=protection_update, + ) + + +def _retained_participant_invitation_ids( + request: SchedulingRequest, +) -> tuple[tuple[str, str], ...]: + return tuple( + (participant.id, cast(str, participant.poll_invitation_id)) + for participant in _active_participants(request) + if participant.poll_invitation_id is not None + ) + + +def _request_field_updates( + payload: SchedulingRequestUpdateRequest, +) -> tuple[tuple[str, Any], ...]: + updates: list[tuple[str, Any]] = [] + for field in REQUEST_UPDATE_FIELDS: + if field not in payload.model_fields_set: + continue + value = getattr(payload, field) + if value is None and field not in NULLABLE_REQUEST_UPDATE_FIELDS: + raise SchedulingError(f"{field} cannot be null") + updates.append((field, value)) + return tuple(updates) + + +def _validate_request_password_update( + request: SchedulingRequest, + *, + payload: SchedulingRequestUpdateRequest, + protection_update: bool | None, +) -> None: + protection_enabled = request.anonymous_password_protection_enabled if protection_update is None else protection_update + if payload.anonymous_password is not None and not protection_enabled: + raise SchedulingError("Password protection must be enabled before setting an anonymous participant password") + if protection_enabled and not request.anonymous_password_hash and payload.anonymous_password is None: + raise SchedulingError("anonymous_password is required when password protection is enabled") + + +def _apply_scheduling_request_update_plan( + request: SchedulingRequest, + *, + payload: SchedulingRequestUpdateRequest, + plan: SchedulingRequestUpdatePlan, +) -> None: + for field, value in plan.field_updates: + setattr(request, field, value) + if plan.update_max_participants: + request.max_participants_per_option = plan.max_participants_per_option + if plan.password_protection_update is not None: + request.anonymous_password_protection_enabled = plan.password_protection_update + if not plan.password_protection_update: + request.anonymous_password_hash = None + if payload.anonymous_password is not None: + request.anonymous_password_hash = hash_participant_password(payload.anonymous_password.get_secret_value()) + if payload.calendar is not None: + _set_calendar_preferences(request, payload) + if payload.metadata is not None: + request.metadata_ = payload.metadata + + +def _update_retained_invitation_expiries( + session: Session, + *, + request: SchedulingRequest, + plan: SchedulingRequestUpdatePlan, +) -> None: + if not plan.deadline_changed: + return + invitations_by_participant = dict(plan.retained_invitation_ids) + for participant in _active_participants(request): + previous_invitation_id = invitations_by_participant.get(participant.id) + if previous_invitation_id is None or participant.poll_invitation_id != previous_invitation_id: + continue + _update_participant_invitation_expiry( + session, + request=request, + participant=participant, + invitation_id=previous_invitation_id, + expires_at=request.deadline_at, + ) + + +def _ensure_request_participants_allowed(request: SchedulingRequest) -> None: + if request.allow_external_participants: + return + if any(participant.participant_type == "external" for participant in _active_participants(request)): + raise SchedulingError("External participants are not allowed for this scheduling request") + + +def _update_backing_scheduling_poll(session: Session, request: SchedulingRequest) -> None: + if request.poll_id is None: + return + try: + _poll_provider().update_poll( + session, + tenant_id=request.tenant_id, + poll_id=request.poll_id, + command=PollUpdateCommand( + title=request.title, + description=request.description, + visibility="private", + result_visibility=request.result_visibility, + allow_anonymous=False, + allow_response_update=request.allow_participant_updates, + closes_at=request.deadline_at, + ), + ) + except PollCapabilityError as exc: + raise SchedulingError(str(exc)) from exc + + def _update_scheduling_request_with_invitation_tokens( session: Session, *, @@ -2197,171 +2452,30 @@ def _update_scheduling_request_with_invitation_tokens( tenant_id=tenant_id, request_id=request_id, lock_slots=payload.slots is not None, - lock_participants=( - payload.participants is not None - or "deadline_at" in payload.model_fields_set - ), + lock_participants=payload.participants is not None or "deadline_at" in payload.model_fields_set, ) - if request.status in {"decided", "handed_off", "cancelled", "archived"}: - raise SchedulingError("Decided, handed off, cancelled, or archived scheduling requests cannot be edited") - deadline_changed = ( - "deadline_at" in payload.model_fields_set - and response_datetime(payload.deadline_at) - != response_datetime(request.deadline_at) + plan = _plan_scheduling_request_update( + request=request, + payload=payload, + participation_available=_poll_participation_provider() is not None, ) - invitation_ids_before_deadline_change = { - participant.id: participant.poll_invitation_id - for participant in _active_participants(request) - if participant.poll_invitation_id is not None - } - if ( - deadline_changed - and invitation_ids_before_deadline_change - and _poll_participation_provider() is None - ): - raise SchedulingError( - "Poll participation invitation rotation capability is unavailable" - ) - policy_changed = any( - field in payload.model_fields_set - and getattr(payload, field) is not None - and getattr(payload, field) != getattr(request, field) - for field in ( - "single_choice", - "allow_maybe", - "allow_comments", - "participant_email_required", - "anonymous_password_protection_enabled", - ) - ) or ( - "max_participants_per_option" in payload.model_fields_set - and payload.max_participants_per_option - != request.max_participants_per_option - ) - if ( - policy_changed - and any(participant.poll_invitation_id for participant in _active_participants(request)) - ): - raise SchedulingError( - "Public participation policy cannot change after invitation links are issued" - ) - nullable_fields = {"description", "location", "deadline_at"} - for field in ( - "title", - "description", - "location", - "deadline_at", - "allow_external_participants", - "allow_participant_updates", - "result_visibility", - "participant_visibility", - "notify_on_answers", - "single_choice", - "allow_maybe", - "allow_comments", - "participant_email_required", - ): - if field not in payload.model_fields_set: - continue - value = getattr(payload, field) - if value is None and field not in nullable_fields: - raise SchedulingError(f"{field} cannot be null") - setattr(request, field, value) - if "max_participants_per_option" in payload.model_fields_set: - request.max_participants_per_option = payload.max_participants_per_option - if ( - "anonymous_password_protection_enabled" in payload.model_fields_set - and payload.anonymous_password_protection_enabled is not None - ): - if payload.anonymous_password_protection_enabled: - request.anonymous_password_protection_enabled = True - else: - request.anonymous_password_protection_enabled = False - request.anonymous_password_hash = None - if payload.anonymous_password is not None: - if not request.anonymous_password_protection_enabled: - raise SchedulingError( - "Password protection must be enabled before setting an anonymous participant password" - ) - request.anonymous_password_hash = hash_participant_password( - payload.anonymous_password.get_secret_value() - ) - if request.anonymous_password_protection_enabled and not request.anonymous_password_hash: - raise SchedulingError( - "anonymous_password is required when password protection is enabled" - ) - if payload.calendar is not None: - _set_calendar_preferences(request, payload) - if payload.metadata is not None: - request.metadata_ = payload.metadata + _apply_scheduling_request_update_plan(request, payload=payload, plan=plan) invitation_tokens: dict[str, str] = {} participant_mutations: tuple[SchedulingParticipantMutation, ...] = () slots_changed = False if payload.slots is not None: - slots_changed = _reconcile_scheduling_slots( - session, - request=request, - supplied_slots=payload.slots, - ) + slots_changed = _reconcile_scheduling_slots(session, request=request, supplied_slots=payload.slots) if payload.participants is not None: invitation_tokens, participant_mutations = _reconcile_scheduling_participants( session, request=request, supplied_participants=payload.participants, ) - if deadline_changed: - for participant in _active_participants(request): - previous_invitation_id = invitation_ids_before_deadline_change.get( - participant.id - ) - if ( - previous_invitation_id is None - or participant.poll_invitation_id != previous_invitation_id - ): - # Removed participants were revoked by reconciliation, while - # newly added participants already use the new deadline. - continue - _update_participant_invitation_expiry( - session, - request=request, - participant=participant, - invitation_id=previous_invitation_id, - expires_at=request.deadline_at, - ) - if ( - not request.allow_external_participants - and any( - participant.participant_type == "external" - for participant in _active_participants(request) - ) - ): - raise SchedulingError( - "External participants are not allowed for this scheduling request" - ) - if request.poll_id is not None: - try: - _poll_provider().update_poll( - session, - tenant_id=tenant_id, - poll_id=request.poll_id, - command=PollUpdateCommand( - title=request.title, - description=request.description, - visibility="private", - result_visibility=request.result_visibility, - allow_anonymous=False, - allow_response_update=request.allow_participant_updates, - closes_at=request.deadline_at, - ), - ) - except PollCapabilityError as exc: - raise SchedulingError(str(exc)) from exc + _update_retained_invitation_expiries(session, request=request, plan=plan) + _ensure_request_participants_allowed(request) + _update_backing_scheduling_poll(session, request) if slots_changed: - refresh_participant_response_state( - session, - request=request, - reset_missing=True, - ) + refresh_participant_response_state(session, request=request, reset_missing=True) session.flush() return request, invitation_tokens, participant_mutations @@ -2440,6 +2554,32 @@ class _CandidateSlotChanges: normalized_end: datetime +@dataclass(frozen=True, slots=True) +class _CandidateSlotUpdatePlan: + slot_id: str + changes: _CandidateSlotChanges + temporal_or_location_changed: bool + + +@dataclass(frozen=True, slots=True) +class _CandidateSlotOrderRef: + slot_id: str | None = None + addition_index: int | None = None + + +@dataclass(frozen=True, slots=True) +class SchedulingSlotReconciliationPlan: + updates: tuple[_CandidateSlotUpdatePlan, ...] + additions: tuple[_CandidateSlotChanges, ...] + removals: tuple[str, ...] + order: tuple[_CandidateSlotOrderRef, ...] + position_changed: bool + + @property + def changed(self) -> bool: + return bool(self.updates or self.additions or self.removals or self.position_changed) + + def _candidate_slot_changes( slot: SchedulingCandidateSlot, payload: SchedulingCandidateSlotUpdateRequest, @@ -2567,166 +2707,214 @@ def _candidate_slot_changes_from_reconcile( ) -def _reconcile_scheduling_slots( - session: Session, +def _plan_scheduling_slot_reconciliation( *, request: SchedulingRequest, supplied_slots: list[SchedulingCandidateSlotReconcileInput], -) -> bool: - """Apply a complete slot collection while keeping Poll answers coherent.""" + option_mutation_available: bool, +) -> SchedulingSlotReconciliationPlan: + current_by_id, removed_slots, new_inputs = _slot_reconciliation_members( + request, + supplied_slots=supplied_slots, + option_mutation_available=option_mutation_available, + ) + updates = _slot_reconciliation_updates( + request, + supplied_slots=supplied_slots, + current_by_id=current_by_id, + ) + additions = tuple(_candidate_slot_changes_from_reconcile(request, item) for item in new_inputs) + order = _slot_reconciliation_order(supplied_slots) + position_changed = any( + ref.slot_id is not None and current_by_id[ref.slot_id].position != position + for position, ref in enumerate(order) + ) + return SchedulingSlotReconciliationPlan( + updates=updates, + additions=additions, + removals=tuple(slot.id for slot in removed_slots), + order=order, + position_changed=position_changed, + ) + +def _slot_reconciliation_members( + request: SchedulingRequest, + *, + supplied_slots: list[SchedulingCandidateSlotReconcileInput], + option_mutation_available: bool, +) -> tuple[ + dict[str, SchedulingCandidateSlot], + list[SchedulingCandidateSlot], + list[SchedulingCandidateSlotReconcileInput], +]: if request.status not in {"draft", "collecting"}: - raise SchedulingError( - "Only draft or collecting scheduling request slots can be edited" - ) + raise SchedulingError("Only draft or collecting scheduling request slots can be edited") if request.poll_id is None: raise SchedulingError("Scheduling request has no backing poll") current_slots = _active_slots(request) current_by_id = {slot.id: slot for slot in current_slots} - unknown_ids = sorted( - item.id - for item in supplied_slots - if item.id is not None and item.id not in current_by_id - ) + unknown_ids = sorted(item.id for item in supplied_slots if item.id is not None and item.id not in current_by_id) if unknown_ids: raise SchedulingError("Unknown scheduling candidate slot") - supplied_ids = { - item.id - for item in supplied_slots - if item.id is not None - } + supplied_ids = {item.id for item in supplied_slots if item.id is not None} removed_slots = [slot for slot in current_slots if slot.id not in supplied_ids] new_inputs = [item for item in supplied_slots if item.id is None] - if (removed_slots or new_inputs) and _poll_participation_provider() is None: - raise SchedulingError( - "Poll participation option mutation capability is unavailable" - ) + if (removed_slots or new_inputs) and not option_mutation_available: + raise SchedulingError("Poll participation option mutation capability is unavailable") for slot in removed_slots: if slot.tentative_hold_event_id: - raise SchedulingError( - "A candidate slot with a tentative calendar hold cannot be removed" - ) + raise SchedulingError("A candidate slot with a tentative calendar hold cannot be removed") if slot.poll_option_id is None: raise SchedulingError("Scheduling slot has no backing poll option") + return current_by_id, removed_slots, new_inputs - planned_updates: list[ - tuple[SchedulingCandidateSlot, _CandidateSlotChanges, bool] - ] = [] + +def _slot_reconciliation_updates( + request: SchedulingRequest, + *, + supplied_slots: list[SchedulingCandidateSlotReconcileInput], + current_by_id: dict[str, SchedulingCandidateSlot], +) -> tuple[_CandidateSlotUpdatePlan, ...]: + updates: list[_CandidateSlotUpdatePlan] = [] for item in supplied_slots: if item.id is None: continue slot = current_by_id[item.id] if item.revision != scheduling_slot_revision(slot): - raise SchedulingConflictError( - "Scheduling options changed after this edit form was loaded; reload before saving" - ) + raise SchedulingConflictError("Scheduling options changed after this edit form was loaded; reload before saving") changes = _candidate_slot_changes_from_reconcile(request, item) - changed, temporal_or_location_changed = _candidate_slot_change_flags( - slot, - changes, - ) - if ( - changed - and temporal_or_location_changed - and slot.tentative_hold_event_id - ): - raise SchedulingError( - "A slot with a tentative calendar hold cannot change time, timezone, or location" - ) + changed, temporal_or_location_changed = _candidate_slot_change_flags(slot, changes) + if changed and temporal_or_location_changed and slot.tentative_hold_event_id: + raise SchedulingError("A slot with a tentative calendar hold cannot change time, timezone, or location") if changed: if slot.poll_option_id is None: raise SchedulingError("Scheduling slot has no backing poll option") - planned_updates.append((slot, changes, temporal_or_location_changed)) + updates.append(_CandidateSlotUpdatePlan(slot.id, changes, temporal_or_location_changed)) + return tuple(updates) - for slot, changes, temporal_or_location_changed in planned_updates: + +def _slot_reconciliation_order( + supplied_slots: list[SchedulingCandidateSlotReconcileInput], +) -> tuple[_CandidateSlotOrderRef, ...]: + addition_index = 0 + order: list[_CandidateSlotOrderRef] = [] + for item in supplied_slots: + if item.id is not None: + order.append(_CandidateSlotOrderRef(slot_id=item.id)) + else: + order.append(_CandidateSlotOrderRef(addition_index=addition_index)) + addition_index += 1 + return tuple(order) + + +def _add_reconciled_slot( + session: Session, + *, + request: SchedulingRequest, + changes: _CandidateSlotChanges, +) -> SchedulingCandidateSlot: + slot = SchedulingCandidateSlot( + tenant_id=request.tenant_id, + request=request, + label=changes.label, + description=changes.description, + start_at=changes.start_at, + end_at=changes.end_at, + timezone=changes.timezone, + location=changes.location, + position=max((stored.position for stored in request.slots), default=-1) + 1, + metadata_=changes.metadata, + ) + session.add(slot) + session.flush() + provider = _poll_participation_provider() + if provider is None: + raise SchedulingError("Poll participation option mutation capability is unavailable") + try: + created = provider.add_option( + session, + tenant_id=request.tenant_id, + poll_id=cast(str, request.poll_id), + command=PollOptionRequest( + key=f"slot-{slot.id}", + label=slot.label, + description=slot.description, + value={ + "slot_id": slot.id, + "start_at": changes.normalized_start.isoformat(), + "end_at": changes.normalized_end.isoformat(), + "timezone": slot.timezone, + "location": slot.location, + }, + metadata=slot.metadata_ or {}, + ), + ) + except PollCapabilityError as exc: + raise SchedulingError(str(exc)) from exc + slot.poll_option_id = created.id + slot.position = created.position + return slot + + +def _remove_reconciled_slot( + session: Session, + *, + request: SchedulingRequest, + slot: SchedulingCandidateSlot, +) -> None: + provider = _poll_participation_provider() + if provider is None: + raise SchedulingError("Poll participation option mutation capability is unavailable") + try: + provider.remove_option( + session, + tenant_id=request.tenant_id, + poll_id=cast(str, request.poll_id), + option_id=cast(str, slot.poll_option_id), + ) + except PollCapabilityError as exc: + raise SchedulingError(str(exc)) from exc + slot.deleted_at = _now() + + +def _apply_scheduling_slot_reconciliation( + session: Session, + *, + request: SchedulingRequest, + plan: SchedulingSlotReconciliationPlan, +) -> bool: + current_by_id = {slot.id: slot for slot in _active_slots(request)} + for update in plan.updates: + slot = current_by_id[update.slot_id] _update_candidate_poll_option( session, tenant_id=request.tenant_id, request=request, slot=slot, - changes=changes, + changes=update.changes, ) _apply_candidate_slot_changes( slot, - changes, - temporal_or_location_changed=temporal_or_location_changed, + update.changes, + temporal_or_location_changed=update.temporal_or_location_changed, ) - participation_provider = _poll_participation_provider() - created_slots: list[SchedulingCandidateSlot] = [] - for item in new_inputs: - changes = _candidate_slot_changes_from_reconcile(request, item) - slot = SchedulingCandidateSlot( - tenant_id=request.tenant_id, - request=request, - label=changes.label, - description=changes.description, - start_at=changes.start_at, - end_at=changes.end_at, - timezone=changes.timezone, - location=changes.location, - position=max((stored.position for stored in request.slots), default=-1) + 1, - metadata_=changes.metadata, - ) - session.add(slot) - session.flush() - if participation_provider is None: # Guarded above; keeps type narrowing explicit. - raise SchedulingError( - "Poll participation option mutation capability is unavailable" - ) - try: - created = participation_provider.add_option( - session, - tenant_id=request.tenant_id, - poll_id=request.poll_id, - command=PollOptionRequest( - key=f"slot-{slot.id}", - label=slot.label, - description=slot.description, - value={ - "slot_id": slot.id, - "start_at": changes.normalized_start.isoformat(), - "end_at": changes.normalized_end.isoformat(), - "timezone": slot.timezone, - "location": slot.location, - }, - metadata=slot.metadata_ or {}, - ), - ) - except PollCapabilityError as exc: - raise SchedulingError(str(exc)) from exc - slot.poll_option_id = created.id - slot.position = created.position - created_slots.append(slot) - - for slot in removed_slots: - if participation_provider is None: # Guarded above; keeps type narrowing explicit. - raise SchedulingError( - "Poll participation option mutation capability is unavailable" - ) - try: - participation_provider.remove_option( - session, - tenant_id=request.tenant_id, - poll_id=request.poll_id, - option_id=cast(str, slot.poll_option_id), - ) - except PollCapabilityError as exc: - raise SchedulingError(str(exc)) from exc - slot.deleted_at = _now() - - created_iter = iter(created_slots) + created_slots = [ + _add_reconciled_slot(session, request=request, changes=changes) + for changes in plan.additions + ] + for slot_id in plan.removals: + _remove_reconciled_slot(session, request=request, slot=current_by_id[slot_id]) ordered_slots = [ - current_by_id[item.id] if item.id is not None else next(created_iter) - for item in supplied_slots + current_by_id[cast(str, ref.slot_id)] + if ref.slot_id is not None + else created_slots[cast(int, ref.addition_index)] + for ref in plan.order ] if any(slot.poll_option_id is None for slot in ordered_slots): raise SchedulingError("Scheduling slot has no backing poll option") - position_changed = any( - slot.position != position - for position, slot in enumerate(ordered_slots) - ) try: _poll_provider().reorder_options( session, @@ -2743,13 +2931,23 @@ def _reconcile_scheduling_slots( raise SchedulingError(str(exc)) from exc for position, slot in enumerate(ordered_slots): slot.position = position + return plan.changed - return bool( - planned_updates - or new_inputs - or removed_slots - or position_changed + +def _reconcile_scheduling_slots( + session: Session, + *, + request: SchedulingRequest, + supplied_slots: list[SchedulingCandidateSlotReconcileInput], +) -> bool: + """Plan and transactionally apply a complete slot collection.""" + + plan = _plan_scheduling_slot_reconciliation( + request=request, + supplied_slots=supplied_slots, + option_mutation_available=_poll_participation_provider() is not None, ) + return _apply_scheduling_slot_reconciliation(session, request=request, plan=plan) def _normalized_participant_identity(value: str | None) -> str | None: @@ -3191,37 +3389,84 @@ def _update_participant_invitation_expiry( participant.participation_gateway = SCHEDULING_PARTICIPATION_GATEWAY -def _reconcile_scheduling_participants( - session: Session, +@dataclass(frozen=True, slots=True) +class _ParticipantUpdatePlan: + participant_id: str + supplied: SchedulingParticipantReconcileInput + changed_fields: tuple[str, ...] + revoke_invitation: bool + + +@dataclass(frozen=True, slots=True) +class _ParticipantCreationPlan: + supplied: SchedulingParticipantReconcileInput + replaces_participant_id: str | None + + +@dataclass(frozen=True, slots=True) +class SchedulingParticipantReconciliationPlan: + updates: tuple[_ParticipantUpdatePlan, ...] + creations: tuple[_ParticipantCreationPlan, ...] + retirements: tuple[str, ...] + + +def _plan_scheduling_participant_reconciliation( *, request: SchedulingRequest, supplied_participants: list[SchedulingParticipantReconcileInput], -) -> tuple[dict[str, str], tuple[SchedulingParticipantMutation, ...]]: - """Reconcile participants without reassigning responses to new identities.""" - + participation_available: bool, + retirement_available: bool, +) -> SchedulingParticipantReconciliationPlan: current_participants = _active_participants(request) current_by_id = {participant.id: participant for participant in current_participants} - _validate_participant_reconciliation( - request, - current_by_id, - supplied_participants, + _validate_participant_reconciliation(request, current_by_id, supplied_participants) + replacement_inputs, removed_participants, contact_link_changes = _participant_reconciliation_members( + current_participants, + current_by_id=current_by_id, + supplied_participants=supplied_participants, ) + _validate_participant_reconciliation_capabilities( + request, + removed_participants=removed_participants, + contact_link_changes=contact_link_changes, + participation_available=participation_available, + retirement_available=retirement_available, + ) + return SchedulingParticipantReconciliationPlan( + updates=_participant_reconciliation_updates( + current_by_id, + supplied_participants=supplied_participants, + replacement_ids=frozenset(replacement_inputs), + ), + creations=_participant_reconciliation_creations( + supplied_participants, + replacement_ids=frozenset(replacement_inputs), + ), + retirements=tuple(participant.id for participant in removed_participants), + ) + + +def _participant_reconciliation_members( + current_participants: list[SchedulingParticipant], + *, + current_by_id: dict[str, SchedulingParticipant], + supplied_participants: list[SchedulingParticipantReconcileInput], +) -> tuple[ + dict[str, SchedulingParticipantReconcileInput], + list[SchedulingParticipant], + list[SchedulingParticipant], +]: replacement_inputs = { cast(str, item.id): item for item in supplied_participants - if item.id is not None - and _participant_identity_changed(current_by_id[item.id], item) + if item.id is not None and _participant_identity_changed(current_by_id[item.id], item) } retained_ids = { item.id for item in supplied_participants if item.id is not None and item.id not in replacement_inputs } - removed_participants = [ - participant - for participant in current_participants - if participant.id not in retained_ids - ] + removed_participants = [participant for participant in current_participants if participant.id not in retained_ids] participants_with_contact_link_changes = [ current_by_id[cast(str, item.id)] for item in supplied_participants @@ -3231,239 +3476,332 @@ def _reconcile_scheduling_participants( and _normalized_participant_identity(current_by_id[item.id].email) != _normalized_participant_identity(item.email) ] - if (removed_participants or participants_with_contact_link_changes) and request.poll_id is None: + return replacement_inputs, removed_participants, participants_with_contact_link_changes + + +def _validate_participant_reconciliation_capabilities( + request: SchedulingRequest, + *, + removed_participants: list[SchedulingParticipant], + contact_link_changes: list[SchedulingParticipant], + participation_available: bool, + retirement_available: bool, +) -> None: + if (removed_participants or contact_link_changes) and request.poll_id is None: raise SchedulingError("Scheduling request has no backing poll") - participation_provider = _poll_participation_provider() if ( any( participant.poll_invitation_id - for participant in ( - *removed_participants, - *participants_with_contact_link_changes, - ) + for participant in (*removed_participants, *contact_link_changes) ) - and participation_provider is None + and not participation_available ): - raise SchedulingError( - "Poll participation invitation revocation capability is unavailable" - ) + raise SchedulingError("Poll participation invitation revocation capability is unavailable") retirement_candidates = [ participant for participant in removed_participants if _participant_response_identities(participant) or participant.poll_invitation_id is not None ] - retirement_provider = _poll_response_retirement_provider() - if retirement_candidates and retirement_provider is None: + if retirement_candidates and not retirement_available: raise SchedulingError( "Poll response retirement capability is unavailable; participants with possible responses cannot be removed safely" ) - mutations: list[SchedulingParticipantMutation] = [] + +def _participant_reconciliation_updates( + current_by_id: dict[str, SchedulingParticipant], + *, + supplied_participants: list[SchedulingParticipantReconcileInput], + replacement_ids: frozenset[str], +) -> tuple[_ParticipantUpdatePlan, ...]: + updates: list[_ParticipantUpdatePlan] = [] for item in supplied_participants: - if item.id is None or item.id in replacement_inputs: + if item.id is None or item.id in replacement_ids: continue participant = current_by_id[item.id] changed_fields = _participant_changed_fields(participant, item) - if not changed_fields: - continue - invitation_revoked = False - if ( - "email" in changed_fields - and participant.poll_invitation_id is not None - ): - if participation_provider is None: # Guarded above. - raise SchedulingError( - "Poll participation invitation revocation capability is unavailable" + if changed_fields: + updates.append( + _ParticipantUpdatePlan( + participant_id=participant.id, + supplied=item, + changed_fields=changed_fields, + revoke_invitation=( + "email" in changed_fields + and participant.poll_invitation_id is not None + ), ) - try: - participation_provider.revoke_invitation( - session, - tenant_id=request.tenant_id, - poll_id=cast(str, request.poll_id), - invitation_id=participant.poll_invitation_id, - ) - except PollCapabilityError as exc: - raise SchedulingError(str(exc)) from exc - participant.poll_invitation_id = None - participant.participation_gateway = None - if participant.status == "invited": - participant.status = "draft" - invitation_revoked = True - participant.respondent_id = item.respondent_id - participant.display_name = item.display_name - participant.email = item.email - participant.participant_type = item.participant_type - participant.required = item.required - participant.metadata_ = dict(item.metadata) - mutations.append( - SchedulingParticipantMutation( - action=( - "scheduling.participant_contact_updated" - if "email" in changed_fields - else "scheduling.participant_updated" - ), - participant_id=participant.id, - changed_fields=changed_fields, - invitation_revoked=invitation_revoked, ) - ) + return tuple(updates) - invitation_tokens: dict[str, str] = {} - replacement_participants: dict[str, SchedulingParticipant] = {} - for item in supplied_participants: - if item.id is not None and item.id not in replacement_inputs: - continue - replaced_participant = ( - current_by_id[item.id] - if item.id is not None - else None + +def _participant_reconciliation_creations( + supplied_participants: list[SchedulingParticipantReconcileInput], + *, + replacement_ids: frozenset[str], +) -> tuple[_ParticipantCreationPlan, ...]: + return tuple( + _ParticipantCreationPlan( + supplied=item, + replaces_participant_id=cast(str, item.id) if item.id in replacement_ids else None, ) - respondent_id = item.respondent_id - if ( - replaced_participant is not None - and (respondent_id or "").startswith("scheduling-participant:") - ): - respondent_id = None - metadata = dict(item.metadata) - if replaced_participant is not None: - metadata["identity_replacement"] = { - "replaces_participant_id": replaced_participant.id, - } - participant = SchedulingParticipant( + for item in supplied_participants + if item.id is None or item.id in replacement_ids + ) + + +def _revoke_reconciled_participant_invitation( + session: Session, + *, + request: SchedulingRequest, + participant: SchedulingParticipant, +) -> bool: + if participant.poll_invitation_id is None: + return False + provider = _poll_participation_provider() + if provider is None: + raise SchedulingError("Poll participation invitation revocation capability is unavailable") + try: + provider.revoke_invitation( + session, tenant_id=request.tenant_id, + poll_id=cast(str, request.poll_id), + invitation_id=participant.poll_invitation_id, + ) + except PollCapabilityError as exc: + raise SchedulingError(str(exc)) from exc + return True + + +def _apply_participant_update( + session: Session, + *, + request: SchedulingRequest, + participant: SchedulingParticipant, + plan: _ParticipantUpdatePlan, +) -> SchedulingParticipantMutation: + invitation_revoked = False + if plan.revoke_invitation: + invitation_revoked = _revoke_reconciled_participant_invitation( + session, request=request, - respondent_id=respondent_id, - display_name=item.display_name, - email=item.email, - participant_type=item.participant_type, - required=item.required, - status="draft", - metadata_=metadata, + participant=participant, ) - session.add(participant) - session.flush() - if replaced_participant is not None: - replacement_participants[replaced_participant.id] = participant + participant.poll_invitation_id = None + participant.participation_gateway = None + if participant.status == "invited": + participant.status = "draft" + supplied = plan.supplied + participant.respondent_id = supplied.respondent_id + participant.display_name = supplied.display_name + participant.email = supplied.email + participant.participant_type = supplied.participant_type + participant.required = supplied.required + participant.metadata_ = dict(supplied.metadata) + return SchedulingParticipantMutation( + action=( + "scheduling.participant_contact_updated" + if "email" in plan.changed_fields + else "scheduling.participant_updated" + ), + participant_id=participant.id, + changed_fields=plan.changed_fields, + invitation_revoked=invitation_revoked, + ) - for participant in removed_participants: - replacement = replacement_participants.get(participant.id) - retirement_count = 0 - response_identities = _participant_response_identities(participant) - if response_identities or participant.poll_invitation_id is not None: - if retirement_provider is None: # Guarded above. - raise SchedulingError( - "Poll response retirement capability is unavailable" - ) - try: - retirement = retirement_provider.retire_responses( - session, - tenant_id=request.tenant_id, - poll_id=cast(str, request.poll_id), - command=PollResponseRetirementCommand( - respondent_ids=response_identities, - invitation_id=participant.poll_invitation_id, - reason=( - "scheduling_participant_replaced" - if replacement is not None - else "scheduling_participant_removed" - ), - idempotency_key=( - f"scheduling:{request.id}:participant:{participant.id}:removed" - ), - metadata={ - "source_module": "scheduling", - "source_resource_type": "participant", - "source_resource_id": participant.id, - "scheduling_request_id": request.id, - }, - ), - ) - except PollCapabilityError as exc: - raise SchedulingError(str(exc)) from exc - retirement_count = retirement.newly_retired_count - invitation_revoked = False - if participant.poll_invitation_id is not None: - if participation_provider is None: # Guarded above. - raise SchedulingError( - "Poll participation invitation revocation capability is unavailable" - ) - try: - participation_provider.revoke_invitation( - session, - tenant_id=request.tenant_id, - poll_id=cast(str, request.poll_id), - invitation_id=participant.poll_invitation_id, - ) - except PollCapabilityError as exc: - raise SchedulingError(str(exc)) from exc - invitation_revoked = True - notification_id = None - if ( - replacement is not None - or participant.poll_invitation_id is not None - or participant.status == "responded" - or retirement_count > 0 - ): - notifications = create_scheduling_notification_jobs( - session, - tenant_id=request.tenant_id, - request_id=request.id, - event_kind=( - "participant_replaced" - if replacement is not None - else "participant_removed" - ), + +def _create_reconciled_participant( + session: Session, + *, + request: SchedulingRequest, + plan: _ParticipantCreationPlan, +) -> SchedulingParticipant: + supplied = plan.supplied + respondent_id = supplied.respondent_id + if plan.replaces_participant_id is not None and (respondent_id or "").startswith("scheduling-participant:"): + respondent_id = None + metadata = dict(supplied.metadata) + if plan.replaces_participant_id is not None: + metadata["identity_replacement"] = {"replaces_participant_id": plan.replaces_participant_id} + participant = SchedulingParticipant( + tenant_id=request.tenant_id, + request=request, + respondent_id=respondent_id, + display_name=supplied.display_name, + email=supplied.email, + participant_type=supplied.participant_type, + required=supplied.required, + status="draft", + metadata_=metadata, + ) + session.add(participant) + session.flush() + return participant + + +def _retire_participant_poll_responses( + session: Session, + *, + request: SchedulingRequest, + participant: SchedulingParticipant, + replacement: SchedulingParticipant | None, +) -> int: + response_identities = _participant_response_identities(participant) + if not response_identities and participant.poll_invitation_id is None: + return 0 + provider = _poll_response_retirement_provider() + if provider is None: + raise SchedulingError("Poll response retirement capability is unavailable") + try: + result = provider.retire_responses( + session, + tenant_id=request.tenant_id, + poll_id=cast(str, request.poll_id), + command=PollResponseRetirementCommand( + respondent_ids=response_identities, + invitation_id=participant.poll_invitation_id, + reason="scheduling_participant_replaced" if replacement is not None else "scheduling_participant_removed", + idempotency_key=f"scheduling:{request.id}:participant:{participant.id}:removed", metadata={ - "reason": ( - "participant_identity_replaced" - if replacement is not None - else "participant_removed" - ), - "replacement_participant_id": ( - replacement.id if replacement is not None else None - ), + "source_module": "scheduling", + "source_resource_type": "participant", + "source_resource_id": participant.id, + "scheduling_request_id": request.id, }, - participant_ids=frozenset({participant.id}), - ) - notification_id = notifications[0].id - retired_at = _now() - participant.metadata_ = { - **(participant.metadata_ or {}), - "participant_retirement": { - "reason": ( - "identity_replaced" - if replacement is not None - else "removed" - ), - "retired_at": retired_at.isoformat(), - "retired_response_count": retirement_count, - "replacement_participant_id": ( - replacement.id if replacement is not None else None - ), - }, - } - participant.status = "removed" - participant.deleted_at = retired_at - mutations.append( - SchedulingParticipantMutation( - action=( - "scheduling.participant_identity_replaced" - if replacement is not None - else "scheduling.participant_removed" - ), - participant_id=participant.id, - replacement_participant_id=( - replacement.id if replacement is not None else None - ), - changed_fields=("identity",) if replacement is not None else (), - invitation_revoked=invitation_revoked, - retired_response_count=retirement_count, - notification_id=notification_id, - ) + ), ) + except PollCapabilityError as exc: + raise SchedulingError(str(exc)) from exc + return result.newly_retired_count - return invitation_tokens, tuple(mutations) + +def _retirement_notification_id( + session: Session, + *, + request: SchedulingRequest, + participant: SchedulingParticipant, + replacement: SchedulingParticipant | None, + retirement_count: int, +) -> str | None: + should_notify = ( + replacement is not None + or participant.poll_invitation_id is not None + or participant.status == "responded" + or retirement_count > 0 + ) + if not should_notify: + return None + notifications = create_scheduling_notification_jobs( + session, + tenant_id=request.tenant_id, + request_id=request.id, + event_kind="participant_replaced" if replacement is not None else "participant_removed", + metadata={ + "reason": "participant_identity_replaced" if replacement is not None else "participant_removed", + "replacement_participant_id": replacement.id if replacement is not None else None, + }, + participant_ids=frozenset({participant.id}), + ) + return notifications[0].id + + +def _retire_reconciled_participant( + session: Session, + *, + request: SchedulingRequest, + participant: SchedulingParticipant, + replacement: SchedulingParticipant | None, +) -> SchedulingParticipantMutation: + retirement_count = _retire_participant_poll_responses( + session, + request=request, + participant=participant, + replacement=replacement, + ) + invitation_revoked = _revoke_reconciled_participant_invitation( + session, + request=request, + participant=participant, + ) + notification_id = _retirement_notification_id( + session, + request=request, + participant=participant, + replacement=replacement, + retirement_count=retirement_count, + ) + retired_at = _now() + participant.metadata_ = { + **(participant.metadata_ or {}), + "participant_retirement": { + "reason": "identity_replaced" if replacement is not None else "removed", + "retired_at": retired_at.isoformat(), + "retired_response_count": retirement_count, + "replacement_participant_id": replacement.id if replacement is not None else None, + }, + } + participant.status = "removed" + participant.deleted_at = retired_at + return SchedulingParticipantMutation( + action="scheduling.participant_identity_replaced" if replacement is not None else "scheduling.participant_removed", + participant_id=participant.id, + replacement_participant_id=replacement.id if replacement is not None else None, + changed_fields=("identity",) if replacement is not None else (), + invitation_revoked=invitation_revoked, + retired_response_count=retirement_count, + notification_id=notification_id, + ) + + +def _apply_scheduling_participant_reconciliation( + session: Session, + *, + request: SchedulingRequest, + plan: SchedulingParticipantReconciliationPlan, +) -> tuple[dict[str, str], tuple[SchedulingParticipantMutation, ...]]: + current_by_id = {participant.id: participant for participant in _active_participants(request)} + mutations = [ + _apply_participant_update( + session, + request=request, + participant=current_by_id[update.participant_id], + plan=update, + ) + for update in plan.updates + ] + replacements: dict[str, SchedulingParticipant] = {} + for creation in plan.creations: + participant = _create_reconciled_participant(session, request=request, plan=creation) + if creation.replaces_participant_id is not None: + replacements[creation.replaces_participant_id] = participant + mutations.extend( + _retire_reconciled_participant( + session, + request=request, + participant=current_by_id[participant_id], + replacement=replacements.get(participant_id), + ) + for participant_id in plan.retirements + ) + return {}, tuple(mutations) + + +def _reconcile_scheduling_participants( + session: Session, + *, + request: SchedulingRequest, + supplied_participants: list[SchedulingParticipantReconcileInput], +) -> tuple[dict[str, str], tuple[SchedulingParticipantMutation, ...]]: + """Plan and transactionally apply participant reconciliation.""" + + plan = _plan_scheduling_participant_reconciliation( + request=request, + supplied_participants=supplied_participants, + participation_available=_poll_participation_provider() is not None, + retirement_available=_poll_response_retirement_provider() is not None, + ) + return _apply_scheduling_participant_reconciliation(session, request=request, plan=plan) def update_scheduling_candidate_slot( @@ -3774,52 +4112,57 @@ def scheduling_participant_response( is_current_participant: bool = False, ) -> dict[str, Any]: expose_email = include_management_fields or is_current_participant - return { + response = { "id": participant.id, - "revision": ( - scheduling_participant_revision(participant) - if include_management_fields - else None - ), "is_current_participant": is_current_participant, - "respondent_id": ( - participant.respondent_id - if include_management_fields and not redact_sensitive - else None - ), "display_name": participant.display_name, "email": participant.email if expose_email and not redact_sensitive else None, - "participant_type": ( - participant.participant_type if include_management_fields else None - ), - "required": participant.required if include_management_fields else None, "status": participant.status, - "poll_invitation_id": ( - participant.poll_invitation_id - if include_management_fields and not redact_sensitive - else None - ), - "invitation_token": ( - invitation_token - if include_management_fields and not redact_sensitive - else None - ), - "last_invited_at": ( - response_datetime(participant.last_invited_at) - if include_management_fields and not redact_sensitive - else None - ), "responded_at": ( response_datetime(participant.responded_at) - if (include_management_fields or is_current_participant) - and not redact_sensitive + if expose_email and not redact_sensitive else None ), - "metadata": ( - (participant.metadata_ or {}) - if include_management_fields and not redact_sensitive - else {} - ), + } + response.update( + _participant_management_response_fields( + participant, + invitation_token=invitation_token, + include_management_fields=include_management_fields, + redact_sensitive=redact_sensitive, + ) + ) + return response + + +def _participant_management_response_fields( + participant: SchedulingParticipant, + *, + invitation_token: str | None, + include_management_fields: bool, + redact_sensitive: bool, +) -> dict[str, Any]: + if not include_management_fields: + return { + "revision": None, + "respondent_id": None, + "participant_type": None, + "required": None, + "poll_invitation_id": None, + "invitation_token": None, + "last_invited_at": None, + "metadata": {}, + } + expose_sensitive = not redact_sensitive + return { + "revision": scheduling_participant_revision(participant), + "respondent_id": participant.respondent_id if expose_sensitive else None, + "participant_type": participant.participant_type, + "required": participant.required, + "poll_invitation_id": participant.poll_invitation_id if expose_sensitive else None, + "invitation_token": invitation_token if expose_sensitive else None, + "last_invited_at": response_datetime(participant.last_invited_at) if expose_sensitive else None, + "metadata": (participant.metadata_ or {}) if expose_sensitive else {}, } @@ -3898,60 +4241,78 @@ def _participant_visibility_decision( return payload -def scheduling_request_response( +@dataclass(frozen=True, slots=True) +class _SchedulingRequestResponseProjection: + actor_ids: tuple[str, ...] + full_roster: bool + active_participants: tuple[SchedulingParticipant, ...] + projected_participants: tuple[SchedulingParticipant, ...] + invitation_tokens: dict[str, str] + visibility_decision: dict[str, Any] + invitation_delivery_available: bool | None + + +def _scheduling_request_response_projection( request: SchedulingRequest, *, - invitation_tokens: dict[str, str] | None = None, - actor_ids: tuple[str, ...] | None = None, - actor_user_id: str | None = None, - can_manage: bool = False, -) -> dict[str, Any]: - invitation_tokens = invitation_tokens or {} + invitation_tokens: dict[str, str], + actor_ids: tuple[str, ...] | None, + actor_user_id: str | None, + can_manage: bool, +) -> _SchedulingRequestResponseProjection: ids = _actor_ids(actor_ids or ()) full_roster = actor_ids is None or can_manage or request.organizer_user_id in ids - active_participants = _active_participants(request) - own_participants = [ + active_participants = tuple(_active_participants(request)) + own_participants = tuple( participant for participant in active_participants if _participant_matches_actor(participant, ids) - ] + ) if full_roster: - visibility_decision = { - "requested_visibility": request.participant_visibility, - "effective_visibility": "names_and_statuses", - "policy_applied": False, - "reason": "Full participant roster access is granted to organizers and scheduling managers.", - "source_path": [], - "details": {"management_access": True}, - } - projected_participants = active_participants + decision = _management_participant_visibility_decision(request) + projected = active_participants + delivery_available = notification_dispatch_provider(get_registry()) is not None else: - visibility_decision = _participant_visibility_decision( + decision = _participant_visibility_decision( request, participant=own_participants[0] if len(own_participants) == 1 else None, actor_user_id=actor_user_id, ) - projected_participants = ( - active_participants - if visibility_decision["effective_visibility"] == "names_and_statuses" - else own_participants - ) - if not full_roster: + projected = active_participants if decision["effective_visibility"] == "names_and_statuses" else own_participants + decision = {**decision, "source_path": [], "details": {}} invitation_tokens = {} - visibility_decision = { - **visibility_decision, - # Policy provenance may contain tenant, organisational-unit, or - # provider internals. Participants need the effective outcome and - # its user-facing reason, not the enforcement topology. - "source_path": [], - "details": {}, - } - public_policy_available = _poll_participation_provider() is not None - participant_invitation_delivery_available = ( - notification_dispatch_provider(get_registry()) is not None - if full_roster - else None + delivery_available = None + return _SchedulingRequestResponseProjection( + actor_ids=ids, + full_roster=full_roster, + active_participants=active_participants, + projected_participants=projected, + invitation_tokens=invitation_tokens, + visibility_decision=decision, + invitation_delivery_available=delivery_available, ) + + +def _management_participant_visibility_decision( + request: SchedulingRequest, +) -> dict[str, Any]: + return { + "requested_visibility": request.participant_visibility, + "effective_visibility": "names_and_statuses", + "policy_applied": False, + "reason": "Full participant roster access is granted to organizers and scheduling managers.", + "source_path": [], + "details": {"management_access": True}, + } + + +def _scheduling_request_scalar_response( + request: SchedulingRequest, + *, + projection: _SchedulingRequestResponseProjection, + public_policy_available: bool, +) -> dict[str, Any]: + full_roster = projection.full_roster return { "id": request.id, "tenant_id": request.tenant_id if full_roster else None, @@ -3975,68 +4336,91 @@ def scheduling_request_response( "allow_comments": request.allow_comments, "participant_email_required": request.participant_email_required, "anonymous_password_protection_enabled": request.anonymous_password_protection_enabled, - "public_participation_policy_enforcement_available": ( - public_policy_available if full_roster else None - ), + "public_participation_policy_enforcement_available": public_policy_available if full_roster else None, "public_participation_policy_enforcement_reason": ( - None - if public_policy_available or not full_roster - else PUBLIC_PARTICIPATION_POLICY_UNAVAILABLE_REASON - ), - "participant_invitation_delivery_available": ( - participant_invitation_delivery_available - ), - "effective_participant_visibility": visibility_decision["effective_visibility"], - "participant_aggregate": _participant_aggregate(active_participants), - "participant_visibility_decision": visibility_decision, - "calendar_integration_enabled": ( - request.calendar_integration_enabled if full_roster else None + None if public_policy_available or not full_roster else PUBLIC_PARTICIPATION_POLICY_UNAVAILABLE_REASON ), + "participant_invitation_delivery_available": projection.invitation_delivery_available, + "effective_participant_visibility": projection.visibility_decision["effective_visibility"], + "participant_aggregate": _participant_aggregate(list(projection.active_participants)), + "participant_visibility_decision": projection.visibility_decision, + } + + +def _scheduling_request_calendar_response( + request: SchedulingRequest, + *, + full_roster: bool, +) -> dict[str, Any]: + return { + "calendar_integration_enabled": request.calendar_integration_enabled if full_roster else None, "calendar_id": request.calendar_id if full_roster else None, - "calendar_freebusy_enabled": ( - request.calendar_freebusy_enabled if full_roster else None - ), - "calendar_hold_enabled": ( - request.calendar_hold_enabled if full_roster else None - ), - "create_calendar_event_on_decision": ( - request.create_calendar_event_on_decision if full_roster else None - ), + "calendar_freebusy_enabled": request.calendar_freebusy_enabled if full_roster else None, + "calendar_hold_enabled": request.calendar_hold_enabled if full_roster else None, + "create_calendar_event_on_decision": request.create_calendar_event_on_decision if full_roster else None, "calendar_event_id": request.calendar_event_id if full_roster else None, "handed_off_at": response_datetime(request.handed_off_at), "cancelled_at": response_datetime(request.cancelled_at), - "cancellation_notice_until": response_datetime( - request.cancellation_notice_until - ), + "cancellation_notice_until": response_datetime(request.cancellation_notice_until), "created_at": response_datetime(request.created_at), "updated_at": response_datetime(request.updated_at), "metadata": (request.metadata_ or {}) if full_roster else {}, - "slots": [ - scheduling_slot_response( - slot, - include_management_fields=full_roster, - ) - for slot in _active_slots(request) - ], - "participants": [ - scheduling_participant_response( - participant, - invitation_token=invitation_tokens.get(participant.id), - include_management_fields=full_roster, - is_current_participant=_participant_matches_actor( - participant, - ids, - ), - redact_sensitive=( - not full_roster - and not _participant_matches_actor(participant, ids) - ), - ) - for participant in projected_participants - ], } +def _scheduling_request_participant_responses( + projection: _SchedulingRequestResponseProjection, +) -> list[dict[str, Any]]: + return [ + scheduling_participant_response( + participant, + invitation_token=projection.invitation_tokens.get(participant.id), + include_management_fields=projection.full_roster, + is_current_participant=_participant_matches_actor(participant, projection.actor_ids), + redact_sensitive=( + not projection.full_roster + and not _participant_matches_actor(participant, projection.actor_ids) + ), + ) + for participant in projection.projected_participants + ] + + +def scheduling_request_response( + request: SchedulingRequest, + *, + invitation_tokens: dict[str, str] | None = None, + actor_ids: tuple[str, ...] | None = None, + actor_user_id: str | None = None, + can_manage: bool = False, +) -> dict[str, Any]: + projection = _scheduling_request_response_projection( + request, + invitation_tokens=invitation_tokens or {}, + actor_ids=actor_ids, + actor_user_id=actor_user_id, + can_manage=can_manage, + ) + public_policy_available = _poll_participation_provider() is not None + response = _scheduling_request_scalar_response( + request, + projection=projection, + public_policy_available=public_policy_available, + ) + response.update( + _scheduling_request_calendar_response( + request, + full_roster=projection.full_roster, + ) + ) + response["slots"] = [ + scheduling_slot_response(slot, include_management_fields=projection.full_roster) + for slot in _active_slots(request) + ] + response["participants"] = _scheduling_request_participant_responses(projection) + return response + + def scheduling_notification_response(notification: SchedulingNotification) -> dict[str, Any]: return { "id": notification.id, diff --git a/tests/test_reconciliation_plans.py b/tests/test_reconciliation_plans.py new file mode 100644 index 0000000..d2adc17 --- /dev/null +++ b/tests/test_reconciliation_plans.py @@ -0,0 +1,283 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest + +from govoplan_scheduling.backend.db.models import ( + SchedulingCandidateSlot, + SchedulingParticipant, + SchedulingRequest, +) +from govoplan_scheduling.backend.schemas import ( + SchedulingCandidateSlotReconcileInput, + SchedulingParticipantReconcileInput, + SchedulingRequestUpdateRequest, +) +from govoplan_scheduling.backend.service import ( + SchedulingError, + _plan_scheduling_participant_reconciliation, + _plan_scheduling_request_update, + _plan_scheduling_slot_reconciliation, + scheduling_participant_revision, + scheduling_slot_revision, +) + + +NOW = datetime(2026, 7, 29, 9, tzinfo=timezone.utc) + + +def _request() -> SchedulingRequest: + return SchedulingRequest( + id="request-1", + tenant_id="tenant-1", + title="Steering group", + timezone="Europe/Berlin", + status="collecting", + poll_id="poll-1", + allow_external_participants=True, + allow_participant_updates=True, + result_visibility="after_close", + participant_visibility="aggregates_only", + notify_on_answers=True, + single_choice=False, + max_participants_per_option=None, + allow_maybe=True, + allow_comments=False, + participant_email_required=False, + anonymous_password_protection_enabled=False, + ) + + +def _slot( + request: SchedulingRequest, + *, + slot_id: str, + position: int, + start_offset: int, +) -> SchedulingCandidateSlot: + slot = SchedulingCandidateSlot( + id=slot_id, + tenant_id=request.tenant_id, + request=request, + poll_option_id=f"option-{slot_id}", + label=f"Slot {position + 1}", + start_at=NOW + timedelta(hours=start_offset), + end_at=NOW + timedelta(hours=start_offset + 1), + timezone=request.timezone, + position=position, + freebusy_conflicts=[], + metadata_={}, + ) + return slot + + +def _slot_input( + slot: SchedulingCandidateSlot, + *, + label: str | None = None, +) -> SchedulingCandidateSlotReconcileInput: + return SchedulingCandidateSlotReconcileInput( + id=slot.id, + revision=scheduling_slot_revision(slot), + label=label or slot.label, + start_at=slot.start_at, + end_at=slot.end_at, + timezone=slot.timezone, + location=slot.location, + metadata=slot.metadata_ or {}, + ) + + +def _participant( + request: SchedulingRequest, + *, + participant_id: str, + respondent_id: str, + email: str, + required: bool, + status: str = "invited", + invitation_id: str | None = None, +) -> SchedulingParticipant: + return SchedulingParticipant( + id=participant_id, + tenant_id=request.tenant_id, + request=request, + respondent_id=respondent_id, + display_name=participant_id.title(), + email=email, + participant_type="internal", + required=required, + status=status, + poll_invitation_id=invitation_id, + participation_gateway="scheduling" if invitation_id else None, + metadata_={}, + ) + + +def _participant_input( + participant: SchedulingParticipant, + **changes: object, +) -> SchedulingParticipantReconcileInput: + values = { + "id": participant.id, + "revision": scheduling_participant_revision(participant), + "respondent_id": participant.respondent_id, + "display_name": participant.display_name, + "email": participant.email, + "participant_type": participant.participant_type, + "required": participant.required, + "metadata": participant.metadata_ or {}, + } + values.update(changes) + return SchedulingParticipantReconcileInput.model_validate(values) + + +def test_slot_plan_is_inspectable_and_does_not_mutate_models() -> None: + request = _request() + first = _slot(request, slot_id="slot-1", position=0, start_offset=1) + second = _slot(request, slot_id="slot-2", position=1, start_offset=3) + supplied = [ + _slot_input(first, label="Updated first slot"), + SchedulingCandidateSlotReconcileInput( + label="New slot", + start_at=NOW + timedelta(hours=5), + end_at=NOW + timedelta(hours=6), + timezone=request.timezone, + ), + ] + + plan = _plan_scheduling_slot_reconciliation( + request=request, + supplied_slots=supplied, + option_mutation_available=True, + ) + + assert [update.slot_id for update in plan.updates] == ["slot-1"] + assert plan.updates[0].changes.label == "Updated first slot" + assert len(plan.additions) == 1 + assert plan.removals == ("slot-2",) + assert plan.changed is True + assert first.label == "Slot 1" + assert second.deleted_at is None + assert len(request.slots) == 2 + + +def test_exact_slot_replay_produces_a_noop_plan() -> None: + request = _request() + first = _slot(request, slot_id="slot-1", position=0, start_offset=1) + + plan = _plan_scheduling_slot_reconciliation( + request=request, + supplied_slots=[_slot_input(first)], + option_mutation_available=True, + ) + + assert plan.changed is False + assert plan.updates == () + assert plan.additions == () + assert plan.removals == () + + +def test_slot_plan_rejects_removal_of_a_tentative_calendar_hold() -> None: + request = _request() + held = _slot(request, slot_id="slot-held", position=0, start_offset=1) + held.tentative_hold_event_id = "event-1" + + with pytest.raises(SchedulingError, match="tentative calendar hold"): + _plan_scheduling_slot_reconciliation( + request=request, + supplied_slots=[], + option_mutation_available=True, + ) + + +def test_participant_plan_distinguishes_updates_additions_and_retirements() -> None: + request = _request() + alice = _participant( + request, + participant_id="alice", + respondent_id="user-alice", + email="alice@example.test", + required=True, + invitation_id="invitation-alice", + ) + bob = _participant( + request, + participant_id="bob", + respondent_id="user-bob", + email="bob@example.test", + required=False, + status="responded", + ) + supplied = [ + _participant_input(alice, email="alice.new@example.test", required=False), + SchedulingParticipantReconcileInput( + respondent_id="user-charlie", + display_name="Charlie", + email="charlie@example.test", + participant_type="internal", + required=True, + ), + ] + + plan = _plan_scheduling_participant_reconciliation( + request=request, + supplied_participants=supplied, + participation_available=True, + retirement_available=True, + ) + + assert len(plan.updates) == 1 + assert plan.updates[0].participant_id == "alice" + assert plan.updates[0].revoke_invitation is True + assert set(plan.updates[0].changed_fields) == {"email", "required"} + assert plan.creations[0].replaces_participant_id is None + assert plan.creations[0].supplied.required is True + assert plan.retirements == ("bob",) + assert alice.email == "alice@example.test" + assert alice.required is True + assert bob.status == "responded" + + +def test_request_plan_records_invitation_expiry_work_without_tokens() -> None: + request = _request() + participant = _participant( + request, + participant_id="alice", + respondent_id="user-alice", + email="alice@example.test", + required=True, + invitation_id="invitation-alice", + ) + deadline = NOW + timedelta(days=2) + + plan = _plan_scheduling_request_update( + request=request, + payload=SchedulingRequestUpdateRequest(deadline_at=deadline), + participation_available=True, + ) + + assert plan.deadline_changed is True + assert plan.retained_invitation_ids == ((participant.id, "invitation-alice"),) + assert "token" not in repr(plan).casefold() + assert request.deadline_at is None + + +def test_request_plan_rejects_policy_change_after_link_issuance() -> None: + request = _request() + _participant( + request, + participant_id="alice", + respondent_id="user-alice", + email="alice@example.test", + required=True, + invitation_id="invitation-alice", + ) + + with pytest.raises(SchedulingError, match="cannot change"): + _plan_scheduling_request_update( + request=request, + payload=SchedulingRequestUpdateRequest(single_choice=True), + participation_available=True, + ) diff --git a/tests/test_response_editing.py b/tests/test_response_editing.py index 9ff7e27..f8fff9e 100644 --- a/tests/test_response_editing.py +++ b/tests/test_response_editing.py @@ -1708,10 +1708,20 @@ class SchedulingResponseEditingTests(unittest.TestCase): notice_until, ) - with patch.object( - scheduling_service, - "_now", - return_value=cancelled_at + timedelta(days=1), + with ( + patch.object( + scheduling_service, + "_now", + return_value=cancelled_at + timedelta(days=1), + ), + patch( + "govoplan_poll.backend.service._now", + return_value=cancelled_at + timedelta(days=1), + ), + patch( + "govoplan_poll.backend.participation_service._now", + return_value=cancelled_at + timedelta(days=1), + ), ): notice = get_public_scheduling_participation( self.session,