from __future__ import annotations from collections.abc import Mapping, Sequence from dataclasses import asdict from datetime import UTC, datetime import hashlib import json from typing import Any import uuid from sqlalchemy import func from sqlalchemy.orm import Session from govoplan_core.core.voting import ( ExternalVotingCastRequest, ExternalVotingFinalizationRequest, ExternalVotingPreparationRequest, ExternalVotingProvider, InteractiveExternalVotingProvider, VOTING_ASSURANCE_RECORDED, VotingBallotCreateCommand, VotingBallotRef, VotingCastCommand, VotingCapabilityError, VotingReceipt, VotingResult, require_voting_provider_assurance, voting_provider_capability, ) from govoplan_voting.backend.db.models import ( VotingBallotRevision, VotingCastRecord, VotingCommandReplay, VotingLifecycleEvent, ) ALLOWED_ASSURANCE_PROFILES = frozenset( {"recorded", "confidential", "secret", "external_certified"} ) ALLOWED_METHODS = frozenset({"single_choice", "approval", "yes_no_abstain"}) TERMINAL_STATES = frozenset({"certified", "annulled"}) class VotingStoreError(ValueError): pass class SqlVotingBallots: def __init__(self, registry: object | None = None) -> None: self._registry = registry def create_ballot( self, session: object, principal: object, *, command: VotingBallotCreateCommand, idempotency_key: str, ) -> VotingBallotRef: typed_session = _session(session) tenant_id = _principal_tenant(principal) payload = _payload_from_command(command) _validate_draft(payload) request = {"command": payload} replay = _read_replay( typed_session, tenant_id=tenant_id, operation="create", idempotency_key=idempotency_key, request=request, ) if replay is not None: return _ref_from_mapping(replay) ballot_id = str(uuid.uuid4()) now = _now() row = VotingBallotRevision( tenant_id=tenant_id, ballot_id=ballot_id, revision=1, state="draft", assurance_profile=str(payload["assurance_profile"]), method=str(payload["method"]), definition_sha256=None, electorate_sha256=None, recorded_at=now, payload=payload, created_by=_principal_actor(principal), ) typed_session.add(row) typed_session.flush() _record_event( typed_session, row=row, event_type="ballot.created", principal=principal, payload={"revision": 1}, ) result = _ref(row) _write_replay( typed_session, tenant_id=tenant_id, operation="create", idempotency_key=idempotency_key, request=request, response=_ref_mapping(result), ) return result def update_draft( self, session: object, principal: object, *, ballot_id: str, command: VotingBallotCreateCommand, expected_revision: int, idempotency_key: str, ) -> VotingBallotRef: typed_session = _session(session) current = _current(typed_session, principal, ballot_id=ballot_id, lock=True) payload = _payload_from_command(command) _validate_draft(payload) request = { "ballot_id": ballot_id, "expected_revision": expected_revision, "command": payload, } operation = f"update:{ballot_id}" replay = _read_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, ) if replay is not None: return _ref_from_mapping(replay) _expect(current, revision=expected_revision, states={"draft"}) revised = _revise( typed_session, current=current, principal=principal, state="draft", payload=payload, event_type="ballot.updated", event_payload={"previous_revision": current.revision}, ) result = _ref(revised) _write_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, response=_ref_mapping(result), ) return result def get_ballot( self, session: object, principal: object, *, ballot_id: str, ) -> Mapping[str, object] | None: typed_session = _session(session) tenant_id = _principal_tenant(principal) row = ( typed_session.query(VotingBallotRevision) .filter( VotingBallotRevision.tenant_id == tenant_id, VotingBallotRevision.ballot_id == ballot_id, VotingBallotRevision.superseded_at.is_(None), ) .one_or_none() ) return _ballot_mapping(row) if row is not None else None def list_ballots( self, session: object, principal: object, *, state: str | None = None, limit: int = 100, ) -> tuple[Mapping[str, object], ...]: typed_session = _session(session) tenant_id = _principal_tenant(principal) if not 1 <= limit <= 200: raise VotingStoreError("Voting list limit must be between 1 and 200.") query = typed_session.query(VotingBallotRevision).filter( VotingBallotRevision.tenant_id == tenant_id, VotingBallotRevision.superseded_at.is_(None), ) if state: query = query.filter(VotingBallotRevision.state == state) rows = ( query.order_by(VotingBallotRevision.recorded_at.desc()).limit(limit).all() ) return tuple(_ballot_mapping(row) for row in rows) def history( self, session: object, principal: object, *, ballot_id: str, ) -> tuple[Mapping[str, object], ...]: typed_session = _session(session) tenant_id = _principal_tenant(principal) rows = ( typed_session.query(VotingLifecycleEvent) .filter( VotingLifecycleEvent.tenant_id == tenant_id, VotingLifecycleEvent.ballot_id == ballot_id, ) .order_by(VotingLifecycleEvent.sequence.asc()) .all() ) return tuple( { "sequence": row.sequence, "event_type": row.event_type, "recorded_at": _datetime_text(row.recorded_at), "actor_id": row.actor_id, "payload": dict(row.payload), } for row in rows ) def open_ballot( self, session: object, principal: object, *, ballot_id: str, expected_revision: int, idempotency_key: str, ) -> VotingBallotRef: typed_session = _session(session) current = _current(typed_session, principal, ballot_id=ballot_id, lock=True) request = {"ballot_id": ballot_id, "expected_revision": expected_revision} operation = f"open:{ballot_id}" replay = _read_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, ) if replay is not None: return _ref_from_mapping(replay) _expect(current, revision=expected_revision, states={"draft"}) payload = dict(current.payload) _validate_draft(payload) profile = str(payload["assurance_profile"]) provider: object | None = None if profile != VOTING_ASSURANCE_RECORDED: provider_id = str(payload.get("provider_id") or "").strip() provider = ( _capability(self._registry, voting_provider_capability(provider_id)) if provider_id else None ) if not isinstance(provider, ExternalVotingProvider): raise VotingStoreError( "The selected Voting assurance profile requires an available external provider." ) try: declaration = require_voting_provider_assurance( provider, provider_id=provider_id, assurance_profile=profile, at=_now(), ) except VotingCapabilityError as exc: raise VotingStoreError(str(exc)) from exc payload["provider_assurance"] = declaration.to_dict() definition_hash, electorate_hash = _frozen_hashes(payload) payload["definition_sha256"] = definition_hash payload["electorate_sha256"] = electorate_hash provider_evidence: tuple[Mapping[str, object], ...] = () if isinstance(provider, InteractiveExternalVotingProvider): provider_id = str(payload["provider_id"]) try: prepared = provider.prepare_ballot( typed_session, principal, request=ExternalVotingPreparationRequest( tenant_id=current.tenant_id, ballot_id=current.ballot_id, requested_provider_ballot_ref=_optional_text( payload.get("provider_ballot_ref") ), definition_sha256=definition_hash, electorate_sha256=electorate_hash, assurance_profile=profile, # type: ignore[arg-type] method=str(payload["method"]), options=tuple( _option_from_mapping(item) for item in payload["options"] ), electorate=tuple( _elector_from_mapping(item) for item in payload["electorate"] ), allow_replacement=bool(payload.get("allow_replacement")), quorum_weight=int(payload.get("quorum_weight") or 0), threshold_numerator=int( payload.get("threshold_numerator") or 1 ), threshold_denominator=int( payload.get("threshold_denominator") or 2 ), opens_at=_parse_datetime(payload.get("opens_at")), closes_at=_parse_datetime(payload.get("closes_at")), requested_at=_now(), idempotency_key=f"open:{_idempotency(idempotency_key)}", ), ) except ValueError as exc: raise VotingStoreError( "Voting provider preparation was rejected." ) from exc if ( prepared.provider_id != provider_id or prepared.definition_sha256 != definition_hash or prepared.electorate_sha256 != electorate_hash or prepared.state not in {"prepared", "open"} ): raise VotingStoreError( "Voting provider preparation did not match the frozen ballot." ) payload["provider_ballot_ref"] = prepared.provider_ballot_ref provider_evidence = tuple(prepared.evidence) elif ( profile != VOTING_ASSURANCE_RECORDED and not str(payload.get("provider_ballot_ref") or "").strip() ): raise VotingStoreError( "This external Voting provider requires a ballot reference before opening." ) revised = _revise( typed_session, current=current, principal=principal, state="open", payload=payload, definition_sha256=definition_hash, electorate_sha256=electorate_hash, event_type="ballot.opened", event_payload={ "revision": current.revision + 1, "definition_sha256": definition_hash, "electorate_sha256": electorate_hash, "provider_id": payload.get("provider_id"), "provider_ballot_ref": payload.get("provider_ballot_ref"), "provider_assurance": payload.get("provider_assurance"), "provider_evidence": [dict(item) for item in provider_evidence], }, ) result = _ref(revised) _write_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, response=_ref_mapping(result), ) return result def cast_ballot( self, session: object, principal: object, *, ballot_id: str, command: VotingCastCommand, ) -> VotingReceipt: typed_session = _session(session) current = _current(typed_session, principal, ballot_id=ballot_id, lock=True) if current.state != "open": raise VotingStoreError("Only an open Voting ballot accepts votes.") _validate_window(current.payload) actor_id = _principal_actor(principal) elector_id = str(command.elector_id or actor_id or "").strip() if not elector_id or not actor_id: raise VotingStoreError("Voting requires an authenticated elector identity.") if elector_id != actor_id: raise VotingStoreError( "A Voting principal cannot cast a ballot for another elector." ) electorate = { str(item["subject_id"]): item for item in list(current.payload.get("electorate") or []) } elector = electorate.get(elector_id) if elector is None: raise VotingStoreError( "The current principal is not in the frozen electorate." ) selections = tuple( str(item).strip() for item in command.selections if str(item).strip() ) _validate_selections(current.payload, selections) idempotency_key = _idempotency(command.idempotency_key) if current.assurance_profile != VOTING_ASSURANCE_RECORDED: provider_id = str(current.payload.get("provider_id") or "").strip() provider_ref = str(current.payload.get("provider_ballot_ref") or "").strip() provider = _capability( self._registry, voting_provider_capability(provider_id), ) if not isinstance(provider, InteractiveExternalVotingProvider): raise VotingStoreError( "This Voting provider does not expose an interactive cast capability." ) _require_pinned_provider_assurance(current.payload, provider) try: receipt = provider.cast_ballot( typed_session, principal, request=ExternalVotingCastRequest( tenant_id=current.tenant_id, ballot_id=current.ballot_id, provider_ballot_ref=provider_ref, definition_sha256=str(current.definition_sha256), electorate_sha256=str(current.electorate_sha256), elector_id=elector_id, selections=selections, allow_replacement=bool( current.payload.get("allow_replacement") ), requested_at=_now(), idempotency_key=idempotency_key, ), ) except ValueError as exc: raise VotingStoreError("Voting provider cast was rejected.") from exc if receipt.ballot_id != current.ballot_id: raise VotingStoreError( "Voting provider returned a receipt for another ballot." ) normalized = VotingReceipt( ballot_id=current.ballot_id, revision=current.revision, receipt_sha256=receipt.receipt_sha256, cast_at=receipt.cast_at, replaced_previous=receipt.replaced_previous, replayed=receipt.replayed, ) if not normalized.replayed: _record_event( typed_session, row=current, event_type="ballot.vote_cast", principal=principal, payload={ "receipt_sha256": normalized.receipt_sha256, "provider_id": provider_id, "replaced_previous": normalized.replaced_previous, }, ) return normalized replay = ( typed_session.query(VotingCastRecord) .filter( VotingCastRecord.tenant_id == current.tenant_id, VotingCastRecord.ballot_id == ballot_id, VotingCastRecord.idempotency_key == idempotency_key, ) .one_or_none() ) if replay is not None: if ( replay.elector_id != elector_id or tuple(replay.selections) != selections ): raise VotingStoreError( "Voting idempotency key was reused for another vote." ) return VotingReceipt( ballot_id=ballot_id, revision=current.revision, receipt_sha256=replay.receipt_sha256, cast_at=_aware(replay.cast_at), replaced_previous=replay.generation > 1, replayed=True, ) previous = ( typed_session.query(VotingCastRecord) .filter( VotingCastRecord.tenant_id == current.tenant_id, VotingCastRecord.ballot_id == ballot_id, VotingCastRecord.elector_id == elector_id, VotingCastRecord.superseded_at.is_(None), ) .with_for_update() .one_or_none() ) if previous is not None and not bool(current.payload.get("allow_replacement")): raise VotingStoreError( "This Voting ballot does not allow replacing a cast vote." ) cast_at = _now() generation = (previous.generation + 1) if previous is not None else 1 if previous is not None: previous.superseded_at = cast_at receipt_hash = _sha256( { "tenant_id": current.tenant_id, "ballot_id": ballot_id, "definition_sha256": current.definition_sha256, "elector_id": elector_id, "generation": generation, "selections": selections, "cast_at": cast_at.isoformat(), "idempotency_key": idempotency_key, } ) typed_session.add( VotingCastRecord( tenant_id=current.tenant_id, ballot_id=ballot_id, definition_sha256=str(current.definition_sha256), elector_id=elector_id, generation=generation, selections=list(selections), weight=int(elector.get("weight") or 1), cast_at=cast_at, idempotency_key=idempotency_key, receipt_sha256=receipt_hash, actor_id=actor_id, ) ) typed_session.flush() _record_event( typed_session, row=current, event_type="ballot.vote_cast", principal=principal, payload={ "receipt_sha256": receipt_hash, "elector_id": elector_id, "generation": generation, "replaced_previous": previous is not None, }, ) return VotingReceipt( ballot_id=ballot_id, revision=current.revision, receipt_sha256=receipt_hash, cast_at=cast_at, replaced_previous=previous is not None, ) def close_ballot( self, session: object, principal: object, *, ballot_id: str, expected_revision: int, idempotency_key: str, ) -> VotingResult: typed_session = _session(session) current = _current(typed_session, principal, ballot_id=ballot_id, lock=True) request = {"ballot_id": ballot_id, "expected_revision": expected_revision} operation = f"close:{ballot_id}" replay = _read_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, ) if replay is not None: return _result_from_mapping(replay) _expect(current, revision=expected_revision, states={"open"}) if current.assurance_profile == VOTING_ASSURANCE_RECORDED: result = self._tally_recorded(typed_session, current=current) else: result = self._finalize_external( typed_session, principal, current=current, idempotency_key=idempotency_key, ) normalized = VotingResult( ballot_id=ballot_id, revision=current.revision + 1, counts=result.counts, weighted_counts=result.weighted_counts, cast_count=result.cast_count, cast_weight=result.cast_weight, eligible_count=result.eligible_count, eligible_weight=result.eligible_weight, quorum_met=result.quorum_met, threshold_met=result.threshold_met, winning_options=result.winning_options, result_sha256=result.result_sha256, evidence=result.evidence, ) payload = dict(current.payload) payload["result"] = _result_mapping(normalized) revised = _revise( typed_session, current=current, principal=principal, state="closed", payload=payload, event_type="ballot.closed", event_payload={ "result_sha256": result.result_sha256, "cast_count": result.cast_count, "quorum_met": result.quorum_met, "threshold_met": result.threshold_met, }, ) if revised.revision != normalized.revision: raise VotingStoreError( "Voting result revision did not match the persisted ballot revision." ) _write_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, response=_result_mapping(normalized), ) return normalized def certify_ballot( self, session: object, principal: object, *, ballot_id: str, expected_revision: int, evidence: Sequence[Mapping[str, object]], idempotency_key: str, ) -> VotingBallotRef: typed_session = _session(session) current = _current(typed_session, principal, ballot_id=ballot_id, lock=True) evidence_items = [dict(item) for item in evidence] request = { "ballot_id": ballot_id, "expected_revision": expected_revision, "evidence": evidence_items, } operation = f"certify:{ballot_id}" replay = _read_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, ) if replay is not None: return _ref_from_mapping(replay) _expect(current, revision=expected_revision, states={"closed"}) result = dict(current.payload.get("result") or {}) if not result.get("result_sha256"): raise VotingStoreError( "A Voting ballot cannot be certified without a result hash." ) if ( current.assurance_profile != VOTING_ASSURANCE_RECORDED and not evidence_items ): raise VotingStoreError( "External Voting certification requires provider evidence." ) certification = { "certified_at": _now().isoformat(), "certified_by": _principal_actor(principal), "result_sha256": result["result_sha256"], "evidence": evidence_items, } payload = dict(current.payload) payload["certification"] = certification revised = _revise( typed_session, current=current, principal=principal, state="certified", payload=payload, event_type="ballot.certified", event_payload=certification, ) response = _ref_mapping(_ref(revised)) _write_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, response=response, ) return _ref_from_mapping(response) def challenge_ballot( self, session: object, principal: object, *, ballot_id: str, expected_revision: int, reason: str, idempotency_key: str, ) -> VotingBallotRef: return self._reasoned_transition( session, principal, ballot_id=ballot_id, expected_revision=expected_revision, reason=reason, idempotency_key=idempotency_key, target_state="challenged", allowed_states={"closed", "certified"}, ) def annul_ballot( self, session: object, principal: object, *, ballot_id: str, expected_revision: int, reason: str, idempotency_key: str, ) -> VotingBallotRef: return self._reasoned_transition( session, principal, ballot_id=ballot_id, expected_revision=expected_revision, reason=reason, idempotency_key=idempotency_key, target_state="annulled", allowed_states={"draft", "open", "closed", "certified", "challenged"}, ) def _reasoned_transition( self, session: object, principal: object, *, ballot_id: str, expected_revision: int, reason: str, idempotency_key: str, target_state: str, allowed_states: set[str], ) -> VotingBallotRef: typed_session = _session(session) current = _current(typed_session, principal, ballot_id=ballot_id, lock=True) normalized_reason = str(reason or "").strip() if not normalized_reason: raise VotingStoreError(f"Voting {target_state} requires a reason.") request = { "ballot_id": ballot_id, "expected_revision": expected_revision, "reason": normalized_reason, } operation = f"{target_state}:{ballot_id}" replay = _read_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, ) if replay is not None: return _ref_from_mapping(replay) _expect(current, revision=expected_revision, states=allowed_states) payload = dict(current.payload) payload[f"{target_state}_reason"] = normalized_reason revised = _revise( typed_session, current=current, principal=principal, state=target_state, payload=payload, event_type=f"ballot.{target_state}", event_payload={"reason": normalized_reason}, ) response = _ref_mapping(_ref(revised)) _write_replay( typed_session, tenant_id=current.tenant_id, operation=operation, idempotency_key=idempotency_key, request=request, response=response, ) return _ref_from_mapping(response) def _tally_recorded( self, session: Session, *, current: VotingBallotRevision, ) -> VotingResult: rows = ( session.query(VotingCastRecord) .filter( VotingCastRecord.tenant_id == current.tenant_id, VotingCastRecord.ballot_id == current.ballot_id, VotingCastRecord.superseded_at.is_(None), ) .all() ) option_keys = [str(item["key"]) for item in current.payload["options"]] counts = {key: 0 for key in option_keys} weighted = {key: 0 for key in option_keys} for row in rows: for selection in row.selections: counts[selection] += 1 weighted[selection] += row.weight eligible = list(current.payload["electorate"]) eligible_weight = sum(int(item.get("weight") or 1) for item in eligible) cast_weight = sum(row.weight for row in rows) quorum_met = cast_weight >= int(current.payload.get("quorum_weight") or 0) max_weight = max(weighted.values(), default=0) winners = tuple( key for key in option_keys if max_weight > 0 and weighted[key] == max_weight ) numerator = int(current.payload.get("threshold_numerator") or 1) denominator = int(current.payload.get("threshold_denominator") or 2) threshold_met = bool( quorum_met and winners and max_weight * denominator >= max(cast_weight, 1) * numerator ) body = { "ballot_id": current.ballot_id, "counts": counts, "weighted_counts": weighted, "cast_count": len(rows), "cast_weight": cast_weight, "eligible_count": len(eligible), "eligible_weight": eligible_weight, "quorum_met": quorum_met, "threshold_met": threshold_met, "winning_options": winners, "definition_sha256": current.definition_sha256, "electorate_sha256": current.electorate_sha256, } return VotingResult( ballot_id=current.ballot_id, revision=current.revision + 1, counts=counts, weighted_counts=weighted, cast_count=len(rows), cast_weight=cast_weight, eligible_count=len(eligible), eligible_weight=eligible_weight, quorum_met=quorum_met, threshold_met=threshold_met, winning_options=winners, result_sha256=_sha256(body), ) def _finalize_external( self, session: Session, principal: object, *, current: VotingBallotRevision, idempotency_key: str, ) -> VotingResult: provider_id = str(current.payload.get("provider_id") or "").strip() provider_ref = str(current.payload.get("provider_ballot_ref") or "").strip() provider = _capability(self._registry, voting_provider_capability(provider_id)) if not isinstance(provider, ExternalVotingProvider): raise VotingStoreError(f"Voting provider is unavailable: {provider_id}.") _require_pinned_provider_assurance(current.payload, provider) electorate = list(current.payload["electorate"]) try: result = provider.finalize_ballot( session, principal, request=ExternalVotingFinalizationRequest( tenant_id=current.tenant_id, ballot_id=current.ballot_id, provider_ballot_ref=provider_ref, definition_sha256=str(current.definition_sha256), electorate_sha256=str(current.electorate_sha256), options=tuple( _option_from_mapping(item) for item in current.payload["options"] ), eligible_count=len(electorate), eligible_weight=sum( int(item.get("weight") or 1) for item in electorate ), requested_at=_now(), idempotency_key=_idempotency(idempotency_key), ), ) except ValueError as exc: raise VotingStoreError( "Voting provider finalization was rejected." ) from exc option_keys = {str(item["key"]) for item in current.payload["options"]} if result.ballot_id != current.ballot_id: raise VotingStoreError( "Voting provider returned a result for another ballot." ) if ( set(result.counts) != option_keys or set(result.weighted_counts) != option_keys ): raise VotingStoreError( "Voting provider result must cover exactly the frozen options." ) if sum(result.counts.values()) < result.cast_count: raise VotingStoreError( "Voting provider counts are inconsistent with cast_count." ) if result.cast_count > len(electorate): raise VotingStoreError( "Voting provider cast_count exceeds the frozen electorate." ) if not result.evidence: raise VotingStoreError("Voting provider finalization requires evidence.") return result def _payload_from_command(command: VotingBallotCreateCommand) -> dict[str, Any]: return { "title": str(command.title or "").strip(), "description": _optional_text(command.description), "method": str(command.method or "").strip(), "assurance_profile": str(command.assurance_profile or "").strip(), "options": [asdict(item) for item in command.options], "electorate": [ { "subject_id": item.subject_id, "label": item.label, "weight": item.weight, "provenance": dict(item.provenance), } for item in command.electorate ], "context": { "module": _optional_text(command.context_module), "resource_type": _optional_text(command.context_resource_type), "resource_id": _optional_text(command.context_resource_id), }, "quorum_weight": command.quorum_weight, "threshold_numerator": command.threshold_numerator, "threshold_denominator": command.threshold_denominator, "allow_replacement": command.allow_replacement, "opens_at": _datetime_text(command.opens_at), "closes_at": _datetime_text(command.closes_at), "provider_id": _optional_text(command.provider_id), "provider_ballot_ref": _optional_text(command.provider_ballot_ref), "provider_assurance": None, "metadata": dict(command.metadata), "definition_sha256": None, "electorate_sha256": None, "result": None, "certification": None, } def _validate_draft(payload: Mapping[str, Any]) -> None: if not str(payload.get("title") or "").strip(): raise VotingStoreError("Voting ballot title is required.") method = str(payload.get("method") or "") if method not in ALLOWED_METHODS: raise VotingStoreError("Unsupported Voting method.") assurance = str(payload.get("assurance_profile") or "") if assurance not in ALLOWED_ASSURANCE_PROFILES: raise VotingStoreError("Unsupported Voting assurance profile.") options = list(payload.get("options") or []) option_keys = [str(item.get("key") or "").strip() for item in options] if ( len(options) < 2 or any(not key for key in option_keys) or len(set(option_keys)) != len(option_keys) ): raise VotingStoreError( "Voting ballots require at least two uniquely keyed options." ) electorate = list(payload.get("electorate") or []) elector_ids = [str(item.get("subject_id") or "").strip() for item in electorate] if ( not electorate or any(not item for item in elector_ids) or len(set(elector_ids)) != len(elector_ids) ): raise VotingStoreError( "Voting electorate entries must have unique subject ids." ) if any(int(item.get("weight") or 0) < 1 for item in electorate): raise VotingStoreError("Voting electorate weights must be positive integers.") eligible_weight = sum(int(item.get("weight") or 1) for item in electorate) quorum = int(payload.get("quorum_weight") or 0) if quorum < 0 or quorum > eligible_weight: raise VotingStoreError("Voting quorum cannot exceed the electorate weight.") numerator = int(payload.get("threshold_numerator") or 0) denominator = int(payload.get("threshold_denominator") or 0) if numerator < 1 or denominator < 1 or numerator > denominator: raise VotingStoreError( "Voting threshold must be a fraction between zero and one." ) opens_at = _parse_datetime(payload.get("opens_at")) closes_at = _parse_datetime(payload.get("closes_at")) if opens_at and closes_at and closes_at <= opens_at: raise VotingStoreError("Voting close time must be after its open time.") if assurance != VOTING_ASSURANCE_RECORDED: if not str(payload.get("provider_id") or "").strip(): raise VotingStoreError("External Voting assurance requires a provider id.") def _validate_selections( payload: Mapping[str, Any], selections: tuple[str, ...] ) -> None: option_keys = {str(item["key"]) for item in payload["options"]} if ( not selections or len(selections) != len(set(selections)) or not set(selections) <= option_keys ): raise VotingStoreError( "Voting selections must be unique configured option keys." ) method = str(payload["method"]) if method in {"single_choice", "yes_no_abstain"} and len(selections) != 1: raise VotingStoreError("This Voting method requires exactly one selection.") def _validate_window(payload: Mapping[str, Any]) -> None: now = _now() opens_at = _parse_datetime(payload.get("opens_at")) closes_at = _parse_datetime(payload.get("closes_at")) if opens_at and now < opens_at: raise VotingStoreError("Voting ballot has not opened yet.") if closes_at and now >= closes_at: raise VotingStoreError("Voting ballot has already reached its close time.") def _require_pinned_provider_assurance( payload: Mapping[str, Any], provider: object, ) -> None: provider_id = str(payload.get("provider_id") or "").strip() assurance_profile = str(payload.get("assurance_profile") or "").strip() try: current = require_voting_provider_assurance( provider, provider_id=provider_id, assurance_profile=assurance_profile, at=_now(), ) except VotingCapabilityError as exc: raise VotingStoreError(str(exc)) from exc pinned = payload.get("provider_assurance") if not isinstance(pinned, Mapping) or dict(pinned) != current.to_dict(): raise VotingStoreError( "Voting provider assurance changed after the ballot was frozen." ) def _frozen_hashes(payload: Mapping[str, Any]) -> tuple[str, str]: electorate = list(payload.get("electorate") or []) definition = { key: value for key, value in payload.items() if key not in { "electorate", "electorate_sha256", "definition_sha256", "result", "certification", "provider_ballot_ref", } } return _sha256(definition), _sha256(electorate) def _current( session: Session, principal: object, *, ballot_id: str, lock: bool, ) -> VotingBallotRevision: tenant_id = _principal_tenant(principal) query = session.query(VotingBallotRevision).filter( VotingBallotRevision.tenant_id == tenant_id, VotingBallotRevision.ballot_id == ballot_id, VotingBallotRevision.superseded_at.is_(None), ) if lock: query = query.with_for_update() row = query.one_or_none() if row is None: raise LookupError("Voting ballot not found.") return row def _expect(row: VotingBallotRevision, *, revision: int, states: set[str]) -> None: if row.revision != revision: raise VotingStoreError( "Voting revision conflict: the expected revision is stale." ) if row.state not in states: raise VotingStoreError( f"Voting transition is not allowed from state {row.state}." ) def _revise( session: Session, *, current: VotingBallotRevision, principal: object, state: str, payload: Mapping[str, Any], event_type: str, event_payload: Mapping[str, Any], definition_sha256: str | None = None, electorate_sha256: str | None = None, ) -> VotingBallotRevision: now = _now() current.superseded_at = now row = VotingBallotRevision( tenant_id=current.tenant_id, ballot_id=current.ballot_id, revision=current.revision + 1, previous_revision_id=current.id, state=state, assurance_profile=str(payload["assurance_profile"]), method=str(payload["method"]), definition_sha256=definition_sha256 or current.definition_sha256, electorate_sha256=electorate_sha256 or current.electorate_sha256, recorded_at=now, payload=dict(payload), created_by=_principal_actor(principal), ) session.add(row) session.flush() _record_event( session, row=row, event_type=event_type, principal=principal, payload=dict(event_payload), ) return row def _record_event( session: Session, *, row: VotingBallotRevision, event_type: str, principal: object, payload: Mapping[str, Any], ) -> None: sequence = ( session.query(func.max(VotingLifecycleEvent.sequence)) .filter( VotingLifecycleEvent.tenant_id == row.tenant_id, VotingLifecycleEvent.ballot_id == row.ballot_id, ) .scalar() or 0 ) + 1 session.add( VotingLifecycleEvent( tenant_id=row.tenant_id, ballot_id=row.ballot_id, sequence=sequence, event_type=event_type, recorded_at=_now(), actor_id=_principal_actor(principal), payload=dict(payload), ) ) session.flush() def _read_replay( session: Session, *, tenant_id: str, operation: str, idempotency_key: str, request: Mapping[str, Any], ) -> Mapping[str, Any] | None: key = _idempotency(idempotency_key) row = ( session.query(VotingCommandReplay) .filter( VotingCommandReplay.tenant_id == tenant_id, VotingCommandReplay.operation == operation, VotingCommandReplay.idempotency_key == key, ) .one_or_none() ) if row is None: return None if row.request_sha256 != _sha256(request): raise VotingStoreError("Voting idempotency key was reused for another command.") return dict(row.response) def _write_replay( session: Session, *, tenant_id: str, operation: str, idempotency_key: str, request: Mapping[str, Any], response: Mapping[str, Any], ) -> None: session.add( VotingCommandReplay( tenant_id=tenant_id, operation=operation, idempotency_key=_idempotency(idempotency_key), request_sha256=_sha256(request), response=dict(response), ) ) session.flush() def _ballot_mapping(row: VotingBallotRevision) -> dict[str, object]: return { "id": row.ballot_id, "revision": row.revision, "state": row.state, "assurance_profile": row.assurance_profile, "method": row.method, "definition_sha256": row.definition_sha256, "electorate_sha256": row.electorate_sha256, "recorded_at": _datetime_text(row.recorded_at), **dict(row.payload), } def _ref(row: VotingBallotRevision) -> VotingBallotRef: return VotingBallotRef( id=row.ballot_id, revision=row.revision, state=row.state, assurance_profile=row.assurance_profile, definition_sha256=row.definition_sha256, electorate_sha256=row.electorate_sha256, ) def _ref_mapping(value: VotingBallotRef) -> dict[str, object]: return { "id": value.id, "revision": value.revision, "state": value.state, "assurance_profile": value.assurance_profile, "definition_sha256": value.definition_sha256, "electorate_sha256": value.electorate_sha256, } def _ref_from_mapping(value: Mapping[str, Any]) -> VotingBallotRef: return VotingBallotRef( id=str(value["id"]), revision=int(value["revision"]), state=str(value["state"]), assurance_profile=str(value["assurance_profile"]), definition_sha256=_optional_text(value.get("definition_sha256")), electorate_sha256=_optional_text(value.get("electorate_sha256")), ) def _result_mapping(value: VotingResult) -> dict[str, object]: return { "ballot_id": value.ballot_id, "revision": value.revision, "counts": dict(value.counts), "weighted_counts": dict(value.weighted_counts), "cast_count": value.cast_count, "cast_weight": value.cast_weight, "eligible_count": value.eligible_count, "eligible_weight": value.eligible_weight, "quorum_met": value.quorum_met, "threshold_met": value.threshold_met, "winning_options": list(value.winning_options), "result_sha256": value.result_sha256, "evidence": [dict(item) for item in value.evidence], } def _result_from_mapping(value: Mapping[str, Any]) -> VotingResult: return VotingResult( ballot_id=str(value["ballot_id"]), revision=int(value["revision"]), counts={str(key): int(count) for key, count in dict(value["counts"]).items()}, weighted_counts={ str(key): int(count) for key, count in dict(value["weighted_counts"]).items() }, cast_count=int(value["cast_count"]), cast_weight=int(value["cast_weight"]), eligible_count=int(value["eligible_count"]), eligible_weight=int(value["eligible_weight"]), quorum_met=bool(value["quorum_met"]), threshold_met=bool(value["threshold_met"]), winning_options=tuple(str(item) for item in value.get("winning_options") or ()), result_sha256=str(value["result_sha256"]), evidence=tuple(dict(item) for item in value.get("evidence") or ()), ) def _option_from_mapping(value: Mapping[str, Any]): from govoplan_core.core.voting import VotingOption return VotingOption( key=str(value["key"]), label=str(value["label"]), description=_optional_text(value.get("description")), ) def _elector_from_mapping(value: Mapping[str, Any]): from govoplan_core.core.voting import VotingElector return VotingElector( subject_id=str(value["subject_id"]), label=_optional_text(value.get("label")), weight=int(value.get("weight") or 1), provenance=dict(value.get("provenance") or {}), ) def _sha256(value: object) -> str: encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, default=_json_default, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _json_default(value: object) -> object: if isinstance(value, datetime): return _datetime_text(value) if isinstance(value, tuple): return list(value) raise TypeError(f"Unsupported canonical Voting value: {type(value).__name__}") def _idempotency(value: str) -> str: normalized = str(value or "").strip() if not normalized or len(normalized) > 160: raise VotingStoreError( "Voting idempotency key is required and limited to 160 characters." ) return normalized def _principal_tenant(principal: object) -> str: value = str(getattr(principal, "tenant_id", "") or "").strip() if not value: raise VotingStoreError("Voting operations require a tenant-bound principal.") return value def _principal_actor(principal: object) -> str | None: for name in ("account_id", "user_id", "identity_id", "membership_id", "subject"): value = str(getattr(principal, name, "") or "").strip() if value: return value return None def _session(value: object) -> Session: if not hasattr(value, "query"): raise VotingStoreError("Voting requires a database session.") return value # type: ignore[return-value] def _capability(registry: object | None, name: str) -> object | None: if ( registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(name) ): return None return registry.capability(name) def _optional_text(value: object) -> str | None: normalized = str(value or "").strip() return normalized or None def _parse_datetime(value: object) -> datetime | None: if value in (None, ""): return None parsed = ( value if isinstance(value, datetime) else datetime.fromisoformat(str(value)) ) if parsed.tzinfo is None: raise VotingStoreError("Voting date-times must include a timezone.") return parsed def _aware(value: datetime) -> datetime: return value if value.tzinfo is not None else value.replace(tzinfo=UTC) def _datetime_text(value: datetime | None) -> str | None: if value is None: return None return _aware(value).isoformat() def _now() -> datetime: return datetime.now(UTC) __all__ = ["SqlVotingBallots", "VotingStoreError"]