from __future__ import annotations from datetime import UTC, datetime import hashlib import hmac import json import re from typing import Any, Mapping, Sequence import uuid from sqlalchemy import func from sqlalchemy.orm import Session from govoplan_core.core.events import ( EventActorRef, EventObjectRef, EventTenantRef, PlatformEvent, emit_platform_event, ) from govoplan_core.core.institutional import ( FormDefinition, FormConditionExpression, FormFieldDefinition, InstitutionalContextError, InstitutionalReference, ) from govoplan_forms.backend.db.models import FormDefinitionRevision _PUBLICATION_TRANSITIONS: dict[str, frozenset[str]] = { "draft": frozenset({"draft", "published", "retired"}), "published": frozenset({"published", "retired"}), "retired": frozenset(), } class FormDefinitionStoreError(ValueError): pass def definition_from_mapping(value: Mapping[str, object]) -> FormDefinition: try: return FormDefinition.from_mapping(value) except InstitutionalContextError as exc: raise FormDefinitionStoreError(str(exc)) from exc def record_form_definition( session: Session, principal: object, *, definition: FormDefinition, expected_revision: str | None = None, ) -> FormDefinition: tenant_id = _principal_tenant(principal) _validate_definition(definition, tenant_id=tenant_id) payload = definition.to_dict() replay = ( session.query(FormDefinitionRevision) .filter( FormDefinitionRevision.tenant_id == tenant_id, FormDefinitionRevision.form_id == definition.reference.object_id, FormDefinitionRevision.revision == definition.temporal.revision, ) .one_or_none() ) if replay is not None: if replay.payload != payload: raise FormDefinitionStoreError( "A different Form definition already uses this revision." ) return _definition_from_row(replay) current = _current_row( session, tenant_id=tenant_id, form_id=definition.reference.object_id, lock=True, ) if current is None: if expected_revision is not None: raise FormDefinitionStoreError( "Form definition revision conflict: no current revision exists." ) key_collision = ( session.query(FormDefinitionRevision.id) .filter( FormDefinitionRevision.tenant_id == tenant_id, FormDefinitionRevision.form_key == definition.key, FormDefinitionRevision.superseded_at.is_(None), ) .first() ) if key_collision is not None: raise FormDefinitionStoreError( "Form definition key is already in use in this tenant." ) else: if expected_revision != current.revision: raise FormDefinitionStoreError( "Form definition revision conflict: the expected revision is stale." ) if definition.key != current.form_key: raise FormDefinitionStoreError( "A Form definition key cannot change across revisions." ) if ( definition.publication_state not in _PUBLICATION_TRANSITIONS[current.publication_state] ): raise FormDefinitionStoreError( f"Form publication transition {current.publication_state!r} to " f"{definition.publication_state!r} is not allowed." ) current.superseded_at = _recorded_at(definition) row = FormDefinitionRevision( tenant_id=tenant_id, form_id=definition.reference.object_id, form_key=definition.key, revision=definition.temporal.revision, previous_revision_id=current.id if current is not None else None, publication_state=definition.publication_state, title=definition.title, recorded_at=_recorded_at(definition), search_text=f"{definition.key} {definition.title} {definition.description or ''}".casefold(), payload=payload, changed_by=_principal_actor(principal), ) session.add(row) session.flush() event_id = str(uuid.uuid4()) emit_platform_event( session, PlatformEvent( event_id=event_id, type="forms.definition.recorded", module_id="forms", payload={ "form_id": row.form_id, "form_key": row.form_key, "revision": row.revision, "publication_state": row.publication_state, "field_count": len(definition.fields), }, occurred_at=row.recorded_at, actor=EventActorRef(type="account", id=_principal_actor(principal)), tenant=EventTenantRef(id=tenant_id), resource=EventObjectRef( type="form_definition", id=row.form_id, label=row.title, ), classification="internal", ), ) return _definition_from_row(row) def form_definition_diagnostics( definition: FormDefinition, ) -> tuple[Mapping[str, object], ...]: """Return deterministic, non-blocking authoring diagnostics. Structural errors are rejected by ``_validate_definition``. Diagnostics are reserved for useful publication quality feedback such as untranslated text. """ diagnostics: list[Mapping[str, object]] = [] field_by_key = {item.key: item for item in definition.fields} page_keys = {item.key for item in definition.pages} section_keys = {item.key for page in definition.pages for item in page.sections} for localization in definition.localizations: locale = localization.locale if not localization.title: diagnostics.append( _definition_diagnostic( "warning", "translation.title_missing", f"{locale} does not translate the Form title.", locale=locale, ) ) for field in definition.fields: if field.key not in localization.field_labels: diagnostics.append( _definition_diagnostic( "warning", "translation.field_label_missing", f"{locale} does not translate field {field.key!r}.", locale=locale, subject=field.key, ) ) if field.help_text and field.key not in localization.field_help_texts: diagnostics.append( _definition_diagnostic( "warning", "translation.field_help_missing", f"{locale} does not translate help for field {field.key!r}.", locale=locale, subject=field.key, ) ) translated_options = localization.option_labels.get(field.key, {}) for option in field.options: if option not in translated_options: diagnostics.append( _definition_diagnostic( "warning", "translation.option_missing", f"{locale} does not translate option {option!r} of field {field.key!r}.", locale=locale, subject=f"{field.key}:{option}", ) ) for key in sorted(page_keys - set(localization.page_titles)): diagnostics.append( _definition_diagnostic( "warning", "translation.page_title_missing", f"{locale} does not translate page {key!r}.", locale=locale, subject=key, ) ) for key in sorted(section_keys - set(localization.section_titles)): diagnostics.append( _definition_diagnostic( "warning", "translation.section_title_missing", f"{locale} does not translate section {key!r}.", locale=locale, subject=key, ) ) # These sets are validated structurally. Keeping the lookup here makes # diagnostics stable if future compatible readers retain unknown keys. _ = field_by_key return tuple(diagnostics) def export_form_definition_fragment( definition: FormDefinition, *, exported_at: datetime, exported_by: str | None, ) -> dict[str, object]: if exported_at.tzinfo is None or exported_at.utcoffset() is None: raise FormDefinitionStoreError("Package exported_at must include a timezone.") definition_payload = definition.to_dict() digest = _payload_sha256(definition_payload) return { "kind": "govoplan.forms.definition", "contract_version": "0.1.0", "definition": definition_payload, "definition_sha256": digest, "provenance": { "owner_module": "forms", "tenant_id": definition.reference.tenant_id, "form_id": definition.reference.object_id, "revision": definition.reference.version, "exported_at": exported_at.isoformat(), "exported_by": exported_by, }, } def assess_form_definition_fragment( session: Session, principal: object, *, fragment: Mapping[str, object], ) -> dict[str, object]: source = _definition_from_fragment(fragment) tenant_id = _principal_tenant(principal) current = get_form_definition( session, principal, form_id=source.reference.object_id, ) key_collision = ( session.query(FormDefinitionRevision.form_id) .filter( FormDefinitionRevision.tenant_id == tenant_id, FormDefinitionRevision.form_key == source.key, FormDefinitionRevision.superseded_at.is_(None), ) .first() ) same_tenant = source.reference.tenant_id == tenant_id if ( same_tenant and current is not None and current.reference.version == source.reference.version ): outcome = ( "replay" if current.to_dict() == source.to_dict() else "revision_conflict" ) elif current is not None: outcome = "new_revision_required" elif key_collision is not None: outcome = "key_conflict" else: outcome = "create" return { "outcome": outcome, "portable": True, "same_tenant": same_tenant, "source": { "tenant_id": source.reference.tenant_id, "form_id": source.reference.object_id, "key": source.key, "revision": source.reference.version, }, "current_revision": current.reference.version if current else None, "requires_remap": not same_tenant, } def import_form_definition_fragment( session: Session, principal: object, *, fragment: Mapping[str, object], target_form_id: str | None, target_key: str | None, expected_revision: str | None, change_reason: str, recorded_at: datetime, ) -> FormDefinition: source = _definition_from_fragment(fragment) tenant_id = _principal_tenant(principal) if recorded_at.tzinfo is None or recorded_at.utcoffset() is None: raise FormDefinitionStoreError( "Package import recorded_at must include a timezone." ) clean_reason = str(change_reason or "").strip() if not clean_reason or len(clean_reason) > 1000: raise FormDefinitionStoreError( "Package import requires a change reason of at most 1000 characters." ) resolved_id = str(target_form_id or source.reference.object_id).strip() resolved_key = str(target_key or source.key).strip() current = get_form_definition(session, principal, form_id=resolved_id) revision = str(uuid.uuid4()) payload = source.to_dict() payload["reference"] = { **dict(payload["reference"]), "object_id": resolved_id, "tenant_id": tenant_id, "version": revision, } payload["key"] = current.key if current is not None else resolved_key payload["temporal"] = { "revision": revision, "valid_from": recorded_at.isoformat(), "valid_to": None, "recorded_at": recorded_at.isoformat(), "superseded_at": None, "change_reason": clean_reason, } payload["publication_state"] = "draft" metadata = dict(source.metadata) metadata["package_import"] = { "source_tenant_id": source.reference.tenant_id, "source_form_id": source.reference.object_id, "source_revision": source.reference.version, "source_sha256": str(fragment.get("definition_sha256") or ""), } payload["metadata"] = metadata imported = definition_from_mapping(payload) return record_form_definition( session, principal, definition=imported, expected_revision=expected_revision, ) def get_form_definition( session: Session, principal: object, *, form_id: str, revision: str | None = None, ) -> FormDefinition | None: tenant_id = _principal_tenant(principal) query = session.query(FormDefinitionRevision).filter( FormDefinitionRevision.tenant_id == tenant_id, FormDefinitionRevision.form_id == form_id, ) if revision is None: query = query.filter(FormDefinitionRevision.superseded_at.is_(None)) else: query = query.filter(FormDefinitionRevision.revision == revision) row = query.order_by(FormDefinitionRevision.recorded_at.desc()).first() return _definition_from_row(row) if row is not None else None def list_form_definitions( session: Session, principal: object, *, query: str = "", publication_states: Sequence[str] | None = None, offset: int = 0, limit: int = 100, ) -> tuple[tuple[FormDefinition, ...], int]: tenant_id = _principal_tenant(principal) if offset < 0 or not 1 <= limit <= 200: raise FormDefinitionStoreError( "Form definition offset must be non-negative and limit between 1 and 200." ) statement = session.query(FormDefinitionRevision).filter( FormDefinitionRevision.tenant_id == tenant_id, FormDefinitionRevision.superseded_at.is_(None), ) if publication_states: statement = statement.filter( FormDefinitionRevision.publication_state.in_(tuple(publication_states)) ) clean_query = query.strip().casefold() if clean_query: statement = statement.filter( FormDefinitionRevision.search_text.contains(clean_query) ) total = int(statement.with_entities(func.count()).scalar() or 0) rows = ( statement.order_by( FormDefinitionRevision.form_key.asc(), FormDefinitionRevision.recorded_at.desc(), ) .offset(offset) .limit(limit) .all() ) return tuple(_definition_from_row(row) for row in rows), total def form_definition_history( session: Session, principal: object, *, form_id: str, limit: int = 100, ) -> tuple[FormDefinition, ...]: tenant_id = _principal_tenant(principal) if not 1 <= limit <= 200: raise FormDefinitionStoreError("Form history limit must be between 1 and 200.") rows = ( session.query(FormDefinitionRevision) .filter( FormDefinitionRevision.tenant_id == tenant_id, FormDefinitionRevision.form_id == form_id, ) .order_by(FormDefinitionRevision.recorded_at.desc()) .limit(limit) .all() ) return tuple(_definition_from_row(row) for row in rows) class SqlFormDefinitionProvider: def get_form_definition( self, session: object, principal: object, *, reference: InstitutionalReference, effective_at: datetime | None = None, ) -> FormDefinition | None: tenant_id = _principal_tenant(principal) if ( reference.kind != "form" or reference.owner_module != "forms" or reference.tenant_id != tenant_id or not reference.version ): raise InstitutionalContextError( "Form definition lookup requires an exact same-tenant Forms reference." ) definition = get_form_definition( _session(session), principal, form_id=reference.object_id, revision=reference.version, ) if definition is None or ( effective_at is not None and not definition.temporal.effective_at(effective_at) ): return None return definition def list_form_definitions( self, session: object, principal: object, *, tenant_id: str, query: str = "", limit: int = 100, ) -> Sequence[FormDefinition]: if tenant_id != _principal_tenant(principal): raise InstitutionalContextError( "Form definition catalogue lookup cannot cross tenants." ) items, _ = list_form_definitions( _session(session), principal, query=query, publication_states=("published",), limit=limit, ) return items def _current_row( session: Session, *, tenant_id: str, form_id: str, lock: bool, ) -> FormDefinitionRevision | None: query = session.query(FormDefinitionRevision).filter( FormDefinitionRevision.tenant_id == tenant_id, FormDefinitionRevision.form_id == form_id, FormDefinitionRevision.superseded_at.is_(None), ) if lock: query = query.with_for_update() return query.one_or_none() def _definition_from_row(row: FormDefinitionRevision) -> FormDefinition: payload: dict[str, Any] = dict(row.payload) temporal = dict(payload.get("temporal") or {}) temporal["superseded_at"] = _datetime_text(row.superseded_at) payload["temporal"] = temporal return FormDefinition.from_mapping(payload) def _validate_definition(definition: FormDefinition, *, tenant_id: str) -> None: if definition.reference.owner_module != "forms": raise FormDefinitionStoreError("Form definitions must be owned by Forms.") if definition.reference.tenant_id != tenant_id: raise FormDefinitionStoreError("Form definitions cannot cross tenants.") if definition.temporal.superseded_at is not None: raise FormDefinitionStoreError("Clients cannot set Form superseded_at.") _recorded_at(definition) if not str(definition.temporal.change_reason or "").strip(): raise FormDefinitionStoreError( "A Form definition revision requires a change reason." ) _validate_form_composition(definition) def _validate_form_composition(definition: FormDefinition) -> None: fields = {item.key: item for item in definition.fields} if definition.pages: field_occurrences = [ field_key for page in definition.pages for section in page.sections for field_key in section.field_keys ] unknown = sorted(set(field_occurrences) - set(fields)) if unknown: raise FormDefinitionStoreError( f"Form pages reference unknown fields: {', '.join(unknown)}." ) duplicates = sorted( key for key in set(field_occurrences) if field_occurrences.count(key) > 1 ) if duplicates: raise FormDefinitionStoreError( f"Form pages place fields more than once: {', '.join(duplicates)}." ) missing = sorted(set(fields) - set(field_occurrences)) if missing: raise FormDefinitionStoreError( f"Form pages do not place fields: {', '.join(missing)}." ) conditions: list[tuple[str, FormConditionExpression]] = [] for field in definition.fields: if field.visibility_condition is not None: conditions.append((f"field:{field.key}", field.visibility_condition)) for page in definition.pages: if page.visibility_condition is not None: conditions.append((f"page:{page.key}", page.visibility_condition)) for section in page.sections: if section.visibility_condition is not None: conditions.append( (f"section:{section.key}", section.visibility_condition) ) for subject, condition in conditions: _validate_condition(condition, fields=fields, subject=subject) graph = { field.key: set(field.visibility_condition.referenced_fields) if field.visibility_condition is not None else set() for field in definition.fields } _reject_condition_cycles(graph) page_keys = {item.key for item in definition.pages} section_keys = {item.key for page in definition.pages for item in page.sections} for localization in definition.localizations: unknown_fields = ( set(localization.field_labels) | set(localization.field_help_texts) | set(localization.option_labels) ) - set(fields) if unknown_fields: raise FormDefinitionStoreError( f"Form localization {localization.locale!r} references unknown fields: " f"{', '.join(sorted(unknown_fields))}." ) if set(localization.page_titles) - page_keys: raise FormDefinitionStoreError( f"Form localization {localization.locale!r} references unknown pages." ) if set(localization.section_titles) - section_keys: raise FormDefinitionStoreError( f"Form localization {localization.locale!r} references unknown sections." ) for field_key, labels in localization.option_labels.items(): field = fields[field_key] if set(labels) - set(field.options): raise FormDefinitionStoreError( f"Form localization {localization.locale!r} translates unknown " f"options for field {field_key!r}." ) def _validate_condition( condition: FormConditionExpression, *, fields: Mapping[str, FormFieldDefinition], subject: str, ) -> None: if condition.kind != "predicate": for child in condition.conditions: _validate_condition(child, fields=fields, subject=subject) return field_key = str(condition.field_key) field = fields.get(field_key) if field is None: raise FormDefinitionStoreError( f"Form condition on {subject} references unknown field {field_key!r}." ) operator = str(condition.operator) if operator in {"lt", "lte", "gt", "gte"} and field.value_type not in { "integer", "number", "date", "datetime", }: raise FormDefinitionStoreError( f"Form condition operator {operator!r} is incompatible with " f"field {field_key!r} ({field.value_type})." ) if operator == "contains" and field.value_type not in { "text", "multiline_text", "email", "multi_choice", "list", }: raise FormDefinitionStoreError( f"Form condition operator 'contains' is incompatible with field {field_key!r}." ) if operator in {"in", "not_in"} and ( not isinstance(condition.value, Sequence) or isinstance(condition.value, (str, bytes)) ): raise FormDefinitionStoreError( f"Form condition operator {operator!r} requires a list value." ) if operator not in {"is_empty", "is_not_empty", "in", "not_in"}: if not _condition_value_matches(field, condition.value): raise FormDefinitionStoreError( f"Form condition value is incompatible with field {field_key!r} " f"({field.value_type})." ) def _condition_value_matches(field: FormFieldDefinition, value: object) -> bool: if value is None: return True if field.value_type == "boolean": return isinstance(value, bool) if field.value_type == "integer": return isinstance(value, int) and not isinstance(value, bool) if field.value_type == "number": return isinstance(value, (int, float)) and not isinstance(value, bool) if field.value_type in {"object"}: return isinstance(value, Mapping) if field.value_type in {"list", "multi_choice"}: return isinstance(value, Sequence) and not isinstance(value, (str, bytes)) return isinstance(value, str) def _reject_condition_cycles(graph: Mapping[str, set[str]]) -> None: visiting: set[str] = set() visited: set[str] = set() def visit(key: str, path: tuple[str, ...]) -> None: if key in visiting: cycle = " -> ".join((*path, key)) raise FormDefinitionStoreError( f"Form visibility conditions contain a dependency cycle: {cycle}." ) if key in visited: return visiting.add(key) for dependency in sorted(graph.get(key, set())): if dependency in graph: visit(dependency, (*path, key)) visiting.remove(key) visited.add(key) for key in sorted(graph): visit(key, ()) def _definition_from_fragment(fragment: Mapping[str, object]) -> FormDefinition: if fragment.get("kind") != "govoplan.forms.definition": raise FormDefinitionStoreError("Unsupported Forms package fragment kind.") if fragment.get("contract_version") != "0.1.0": raise FormDefinitionStoreError("Unsupported Forms package contract version.") payload = fragment.get("definition") if not isinstance(payload, Mapping): raise FormDefinitionStoreError("Forms package definition must be an object.") expected = str(fragment.get("definition_sha256") or "") if not re.fullmatch(r"[0-9a-f]{64}", expected) or not _constant_time_equal( expected, _payload_sha256(payload), ): raise FormDefinitionStoreError("Forms package definition digest is invalid.") return definition_from_mapping(payload) def _payload_sha256(value: Mapping[str, object]) -> str: encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _constant_time_equal(left: str, right: str) -> bool: return hmac.compare_digest(left, right) def _definition_diagnostic( severity: str, code: str, message: str, *, locale: str | None = None, subject: str | None = None, ) -> Mapping[str, object]: return { "severity": severity, "code": code, "message": message, "locale": locale, "subject": subject, } def _recorded_at(definition: FormDefinition) -> datetime: if definition.temporal.recorded_at is None: raise FormDefinitionStoreError( "A Form definition revision requires recorded_at." ) return definition.temporal.recorded_at def _principal_tenant(principal: object) -> str: tenant_id = str(getattr(principal, "tenant_id", "") or "").strip() if not tenant_id: raise InstitutionalContextError( "Form definition operations require a tenant-bound principal." ) return tenant_id def _principal_actor(principal: object) -> str | None: for name in ("account_id", "identity_id", "membership_id"): value = str(getattr(principal, name, "") or "").strip() if value: return value return None def _session(value: object) -> Session: if not hasattr(value, "query"): raise InstitutionalContextError( "Form definition provider requires a database session." ) return value # type: ignore[return-value] def _datetime_text(value: datetime | None) -> str | None: if value is None: return None if value.tzinfo is None: value = value.replace(tzinfo=UTC) return value.isoformat() __all__ = [ "FormDefinitionStoreError", "SqlFormDefinitionProvider", "definition_from_mapping", "form_definition_history", "get_form_definition", "list_form_definitions", "record_form_definition", ]