refactor: simplify poll transition orchestration

src/govoplan_poll/backend/service.py

@@ -423,6 +423,93 @@ def _transition_for_idempotency_key(
     )
 
 
+def _transition_decision_option(
+    poll: Poll,
+    *,
+    action: str,
+    option_id: str | None,
+    option_key: str | None,
+) -> PollOption | None:
+    if action == "decide":
+        return _option_by_id_or_key(poll, option_id=option_id, option_key=option_key)
+    if option_id is not None or option_key is not None:
+        raise PollError("Only a decide transition accepts a poll option")
+    return None
+
+
+def _replayed_transition(
+    poll: Poll,
+    *,
+    existing: PollLifecycleTransition | None,
+    action: str,
+    decision_option: PollOption | None,
+) -> PollTransitionResult | None:
+    if existing is None:
+        return None
+    requested_option_id = decision_option.id if decision_option is not None else None
+    if existing.action != action or existing.decision_option_id != requested_option_id:
+        raise PollError("Idempotency key was already used for a different poll transition")
+    return PollTransitionResult(poll=poll, transition=existing, replayed=True)
+
+
+def _clear_expired_close_for_open(poll: Poll, *, action: str, now: datetime) -> datetime | None:
+    if action != "open" or poll.closes_at is None:
+        return None
+    normalized_closes_at = response_datetime(poll.closes_at)
+    if normalized_closes_at is None or normalized_closes_at > now:
+        return None
+    poll.closes_at = None
+    return normalized_closes_at
+
+
+def _apply_transition_state(
+    poll: Poll,
+    *,
+    action: str,
+    from_status: str,
+    to_status: str,
+    decision_option: PollOption | None,
+    now: datetime,
+) -> None:
+    if action == "archive":
+        poll.archived_from_status = from_status
+    elif action == "unarchive":
+        poll.archived_from_status = None
+    elif action == "close":
+        poll.closed_at = now
+    elif action == "decide":
+        if decision_option is None:
+            raise PollError("A decide transition requires a poll option")
+        poll.decided_option_id = decision_option.id
+        poll.decided_at = now
+        if poll.closed_at is None:
+            poll.closed_at = now
+    poll.status = to_status
+
+
+def _transition_audit_metadata(
+    *,
+    action: str,
+    from_status: str,
+    to_status: str,
+    used_legacy_unarchive_fallback: bool,
+    cleared_expired_closes_at: datetime | None,
+) -> dict[str, Any]:
+    metadata: dict[str, Any] = {}
+    if action == "archive":
+        metadata["archived_from_status"] = from_status
+    elif action == "unarchive":
+        metadata["restored_status"] = to_status
+        if used_legacy_unarchive_fallback:
+            metadata["legacy_restore_fallback"] = {
+                "reason": "missing_archived_from_status",
+                "restored_status": to_status,
+            }
+    if cleared_expired_closes_at is not None:
+        metadata["cleared_expired_closes_at"] = cleared_expired_closes_at.isoformat()
+    return metadata
+
+
 def transition_poll(
     session: Session,
     *,
@@ -440,22 +527,26 @@ def transition_poll(
 
     poll = _lock_poll_for_transition(session, tenant_id=tenant_id, poll_id=poll_id)
     normalized_key = _normalize_idempotency_key(idempotency_key)
-    decision_option = None
+    decision_option = _transition_decision_option(
-    if action == "decide":
+        poll,
-        decision_option = _option_by_id_or_key(poll, option_id=option_id, option_key=option_key)
+        action=action,
-    elif option_id is not None or option_key is not None:
+        option_id=option_id,
-        raise PollError("Only a decide transition accepts a poll option")
+        option_key=option_key,
+    )
 
     existing = _transition_for_idempotency_key(
         session,
         poll_id=poll.id,
         idempotency_key=normalized_key,
     )
-    if existing is not None:
+    replayed = _replayed_transition(
-        requested_option_id = decision_option.id if decision_option is not None else None
+        poll,
-        if existing.action != action or existing.decision_option_id != requested_option_id:
+        existing=existing,
-            raise PollError("Idempotency key was already used for a different poll transition")
+        action=action,
-        return PollTransitionResult(poll=poll, transition=existing, replayed=True)
+        decision_option=decision_option,
+    )
+    if replayed is not None:
+        return replayed
 
     try:
         plan = transition_engine.plan(
@@ -469,40 +560,22 @@ def transition_poll(
     previous_decision_option_id = poll.decided_option_id
     now = _now()
     used_legacy_unarchive_fallback = action == "unarchive" and poll.archived_from_status is None
-    cleared_expired_closes_at: datetime | None = None
+    cleared_expired_closes_at = _clear_expired_close_for_open(poll, action=action, now=now)
-    if action == "open" and poll.closes_at is not None:
+    _apply_transition_state(
-        normalized_closes_at = response_datetime(poll.closes_at)
+        poll,
-        if normalized_closes_at is not None and normalized_closes_at <= now:
+        action=action,
-            cleared_expired_closes_at = normalized_closes_at
+        from_status=plan.from_status,
-            poll.closes_at = None
+        to_status=plan.to_status,
-
+        decision_option=decision_option,
-    if action == "archive":
+        now=now,
-        poll.archived_from_status = plan.from_status
+    )
-    elif action == "unarchive":
+    transition_metadata = _transition_audit_metadata(
-        poll.archived_from_status = None
+        action=action,
-    elif action == "close":
+        from_status=plan.from_status,
-        poll.closed_at = now
+        to_status=plan.to_status,
-    elif action == "decide":
+        used_legacy_unarchive_fallback=used_legacy_unarchive_fallback,
-        if decision_option is None:
+        cleared_expired_closes_at=cleared_expired_closes_at,
-            raise PollError("A decide transition requires a poll option")
+    )
-        poll.decided_option_id = decision_option.id
-        poll.decided_at = now
-        if poll.closed_at is None:
-            poll.closed_at = now
-
-    poll.status = plan.to_status
-    transition_metadata: dict[str, Any] = {}
-    if action == "archive":
-        transition_metadata["archived_from_status"] = plan.from_status
-    elif action == "unarchive":
-        transition_metadata["restored_status"] = plan.to_status
-        if used_legacy_unarchive_fallback:
-            transition_metadata["legacy_restore_fallback"] = {
-                "reason": "missing_archived_from_status",
-                "restored_status": plan.to_status,
-            }
-    if cleared_expired_closes_at is not None:
-        transition_metadata["cleared_expired_closes_at"] = cleared_expired_closes_at.isoformat()
     transition = PollLifecycleTransition(
         tenant_id=tenant_id,
         poll_id=poll.id,