DataOps-platform/app/core/governance/domain_templates.py

"""Generic governance object contracts and versioned domain templates."""

from __future__ import annotations

import hashlib
import json
import re
import uuid
from collections.abc import Callable
from copy import deepcopy
from dataclasses import dataclass
from typing import Any

from app.core.common.identifiers import new_governance_uid

IDENTIFIER_RE = re.compile(r"^[a-z][a-z0-9_]{1,63}$")
CODE_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{1,79}$")
LIFECYCLE_STATUSES = frozenset({"draft", "active", "retired"})
FIELD_TYPES = frozenset(
    {"string", "integer", "number", "boolean", "date", "datetime", "object", "array"}
)
SECRET_MARKERS = ("password", "secret", "token", "credential", "api_key", "private_key")
MAX_OBJECT_TYPES = 100
MAX_COLLECTION_ITEMS = 500


class DomainTemplateValidationError(ValueError):
    pass


class DomainTemplateNotFound(LookupError):
    pass


@dataclass(frozen=True)
class GovernanceObjectType:
    """Runtime contract for one template-defined governance object type."""

    template_code: str
    type_code: str
    name: str
    stable_uid_prefix: str
    source_identity_fields: tuple[str, ...]
    fields: tuple[dict[str, Any], ...]
    lifecycle_status: str = "active"
    current_version: int = 1

    @classmethod
    def from_definition(
        cls,
        template_code: str,
        definition: dict[str, Any],
        *,
        current_version: int = 1,
    ):
        normalized = _normalize_object_types([definition])[0]
        return cls(
            template_code=_require_identifier(template_code, "template_code"),
            type_code=normalized["type_code"],
            name=normalized["name"],
            stable_uid_prefix=normalized["stable_uid_prefix"],
            source_identity_fields=tuple(normalized["source_identity_fields"]),
            fields=tuple(normalized["fields"]),
            lifecycle_status=normalized["lifecycle_status"],
            current_version=int(current_version),
        )

    def stable_uid(self, source_identity: dict[str, Any]) -> str:
        missing = [
            field
            for field in self.source_identity_fields
            if source_identity.get(field) is None
            or str(source_identity.get(field)).strip() == ""
        ]
        if missing:
            raise DomainTemplateValidationError(
                "source_identity is missing required fields: " + ", ".join(missing)
            )
        canonical_identity = {
            field: source_identity[field] for field in self.source_identity_fields
        }
        return stable_governance_object_uid(
            self.template_code,
            self.type_code,
            canonical_identity,
        )


def _canonical_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def _content_hash(value: Any) -> str:
    return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()


def _require_identifier(value: Any, field: str) -> str:
    normalized = str(value or "").strip()
    if not IDENTIFIER_RE.fullmatch(normalized):
        raise DomainTemplateValidationError(
            f"{field} must be a lowercase snake_case identifier"
        )
    return normalized


def _require_code(value: Any, field: str) -> str:
    normalized = str(value or "").strip()
    if not CODE_RE.fullmatch(normalized):
        raise DomainTemplateValidationError(f"{field} must be a stable code")
    return normalized


def _require_text(value: Any, field: str, max_length: int = 200) -> str:
    normalized = str(value or "").strip()
    if not normalized or len(normalized) > max_length:
        raise DomainTemplateValidationError(
            f"{field} must be between 1 and {max_length} characters"
        )
    return normalized


def _reject_secrets(value: Any, path: str = "template") -> None:
    if isinstance(value, dict):
        for key, nested in value.items():
            normalized_key = str(key).lower().replace("-", "_")
            if any(marker in normalized_key for marker in SECRET_MARKERS):
                raise DomainTemplateValidationError(
                    f"secret-bearing field is not allowed at {path}.{key}"
                )
            _reject_secrets(nested, f"{path}.{key}")
    elif isinstance(value, list):
        for index, nested in enumerate(value):
            _reject_secrets(nested, f"{path}[{index}]")


def _normalize_fields(fields: Any, object_code: str) -> list[dict[str, Any]]:
    if fields is None:
        fields = []
    if not isinstance(fields, list) or len(fields) > MAX_COLLECTION_ITEMS:
        raise DomainTemplateValidationError(f"{object_code}.fields must be a bounded list")
    normalized = []
    seen = set()
    for item in fields:
        if not isinstance(item, dict):
            raise DomainTemplateValidationError(f"{object_code}.fields items must be objects")
        code = _require_identifier(item.get("code"), f"{object_code}.field.code")
        if any(marker in code for marker in SECRET_MARKERS):
            raise DomainTemplateValidationError(
                f"secret-bearing field is not allowed at {object_code}.{code}"
            )
        if code in seen:
            raise DomainTemplateValidationError(f"duplicate field {object_code}.{code}")
        seen.add(code)
        field_type = str(item.get("type") or "string").strip().lower()
        if field_type not in FIELD_TYPES:
            raise DomainTemplateValidationError(
                f"{object_code}.{code}.type is unsupported"
            )
        normalized.append(
            {
                "code": code,
                "name": _require_text(
                    item.get("name") or code, f"{object_code}.{code}.name"
                ),
                "type": field_type,
                "required": bool(item.get("required", False)),
                "description": str(item.get("description") or "").strip(),
            }
        )
    return sorted(normalized, key=lambda item: item["code"])


def _normalize_object_types(items: Any) -> list[dict[str, Any]]:
    if not isinstance(items, list) or not items or len(items) > MAX_OBJECT_TYPES:
        raise DomainTemplateValidationError("object_types must be a non-empty bounded list")
    normalized = []
    seen = set()
    for item in items:
        if not isinstance(item, dict):
            raise DomainTemplateValidationError("object type must be an object")
        code = _require_identifier(item.get("type_code"), "object_types.type_code")
        if code in seen:
            raise DomainTemplateValidationError(f"duplicate object type: {code}")
        seen.add(code)
        source_identity_fields = item.get("source_identity_fields")
        if not isinstance(source_identity_fields, list) or not source_identity_fields:
            raise DomainTemplateValidationError(
                f"{code}.source_identity_fields must be a non-empty list"
            )
        source_identity_fields = [
            _require_identifier(value, f"{code}.source_identity_fields")
            for value in source_identity_fields
        ]
        if any(
            marker in field
            for field in source_identity_fields
            for marker in SECRET_MARKERS
        ):
            raise DomainTemplateValidationError(
                f"{code}.source_identity_fields cannot contain secret-bearing fields"
            )
        if len(set(source_identity_fields)) != len(source_identity_fields):
            raise DomainTemplateValidationError(
                f"{code}.source_identity_fields contains duplicates"
            )
        normalized.append(
            {
                "type_code": code,
                "name": _require_text(item.get("name"), f"{code}.name"),
                "description": str(item.get("description") or "").strip(),
                "stable_uid_prefix": _require_text(
                    item.get("stable_uid_prefix"), f"{code}.stable_uid_prefix", 16
                ).upper(),
                "source_identity_fields": source_identity_fields,
                "fields": _normalize_fields(item.get("fields"), code),
                "lifecycle_status": str(
                    item.get("lifecycle_status") or "active"
                ).strip().lower(),
                "extension": deepcopy(item.get("extension") or {}),
            }
        )
        if normalized[-1]["lifecycle_status"] not in LIFECYCLE_STATUSES:
            raise DomainTemplateValidationError(
                f"{code}.lifecycle_status is unsupported"
            )
    return sorted(normalized, key=lambda item: item["type_code"])


def _normalize_named_collection(value: Any, name: str) -> list[dict[str, Any]]:
    if value is None:
        return []
    if not isinstance(value, list) or len(value) > MAX_COLLECTION_ITEMS:
        raise DomainTemplateValidationError(f"{name} must be a bounded list")
    normalized = []
    seen = set()
    for item in value:
        if not isinstance(item, dict):
            raise DomainTemplateValidationError(f"{name} items must be objects")
        code = _require_code(item.get("code"), f"{name}.code")
        if code in seen:
            raise DomainTemplateValidationError(f"duplicate {name} code: {code}")
        seen.add(code)
        normalized.append({**deepcopy(item), "code": code})
    return sorted(normalized, key=lambda item: item["code"])


def normalize_domain_template(definition: Any) -> dict[str, Any]:
    if not isinstance(definition, dict):
        raise DomainTemplateValidationError("template definition must be an object")
    _reject_secrets(definition)
    lifecycle_status = str(definition.get("lifecycle_status") or "draft").strip().lower()
    if lifecycle_status not in LIFECYCLE_STATUSES:
        raise DomainTemplateValidationError("lifecycle_status is unsupported")
    seed_data = deepcopy(definition.get("seed_data") or [])
    if not isinstance(seed_data, list) or len(seed_data) > MAX_COLLECTION_ITEMS:
        raise DomainTemplateValidationError("seed_data must be a bounded list")
    normalized = {
        "template_code": _require_identifier(
            definition.get("template_code"), "template_code"
        ),
        "name": _require_text(definition.get("name"), "name"),
        "description": str(definition.get("description") or "").strip(),
        "lifecycle_status": lifecycle_status,
        "object_types": _normalize_object_types(definition.get("object_types")),
        "responsibility_roles": _normalize_named_collection(
            definition.get("responsibility_roles"), "responsibility_roles"
        ),
        "rules": _normalize_named_collection(definition.get("rules"), "rules"),
        "metrics": _normalize_named_collection(definition.get("metrics"), "metrics"),
        "seed_data": seed_data,
        "source_identity_contract": {
            "strategy": "uuid5",
            "canonicalization": "sorted-json",
            "required": True,
        },
        "version_contract": {
            "strategy": "append-only",
            "current_version_field": "current_version",
            "content_hash": "sha256",
        },
        "lifecycle_contract": {
            "statuses": sorted(LIFECYCLE_STATUSES),
            "transitions": {
                "draft": ["active", "retired"],
                "active": ["retired"],
                "retired": [],
            },
        },
    }
    normalized["content_hash"] = _content_hash(
        {key: value for key, value in normalized.items() if key != "content_hash"}
    )
    return normalized


def stable_governance_object_uid(
    template_code: str,
    object_type: str,
    source_identity: dict[str, Any],
) -> str:
    template_code = _require_identifier(template_code, "template_code")
    object_type = _require_identifier(object_type, "object_type")
    if not isinstance(source_identity, dict) or not source_identity:
        raise DomainTemplateValidationError("source_identity must be a non-empty object")
    if any(value is None or str(value).strip() == "" for value in source_identity.values()):
        raise DomainTemplateValidationError("source_identity values must be non-empty")
    identity = f"{template_code}/{object_type}/{_canonical_json(source_identity)}"
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"dataops-governance:{identity}"))


def diff_domain_templates(
    before: dict[str, Any] | None,
    after: dict[str, Any],
) -> dict[str, Any]:
    before = before or {}
    before_types = {
        item["type_code"]: item for item in before.get("object_types", [])
    }
    after_types = {item["type_code"]: item for item in after.get("object_types", [])}
    shared = set(before_types) & set(after_types)
    sections = ("responsibility_roles", "rules", "metrics", "seed_data")
    return {
        "added_object_types": sorted(set(after_types) - set(before_types)),
        "removed_object_types": sorted(set(before_types) - set(after_types)),
        "changed_object_types": sorted(
            code for code in shared if before_types[code] != after_types[code]
        ),
        "changed_sections": sorted(
            section
            for section in sections
            if before.get(section, []) != after.get(section, [])
        ),
        "lifecycle_changed": (
            bool(before)
            and before.get("lifecycle_status") != after.get("lifecycle_status")
        ),
    }


class DomainTemplateService:
    def __init__(
        self,
        repository,
        *,
        uid_factory: Callable[[], str] = new_governance_uid,
    ):
        self.repository = repository
        self.uid_factory = uid_factory

    def list_templates(self):
        return self.repository.list_templates()

    def get_template(self, template_code: str):
        template_code = _require_identifier(template_code, "template_code")
        result = self.repository.get_template(template_code)
        if result is None:
            raise DomainTemplateNotFound(f"domain template {template_code} was not found")
        return result

    def list_imports(self, template_code: str):
        template_code = _require_identifier(template_code, "template_code")
        return self.repository.list_imports(template_code)

    def dry_run(self, definition: Any):
        normalized = normalize_domain_template(definition)
        before = self.repository.get_template(normalized["template_code"])
        return {
            "valid": True,
            "template": normalized,
            "diff": diff_domain_templates(before, normalized),
        }

    def import_template(
        self,
        definition: Any,
        *,
        actor_uid: str,
        operation: str = "import",
        target_version: int | None = None,
    ):
        normalized = normalize_domain_template(definition)
        before = self.repository.get_template(normalized["template_code"])
        version = int(before.get("current_version", 0)) + 1 if before else 1
        persisted = {**normalized, "current_version": version}
        audit = {
            "uid": self.uid_factory(),
            "template_code": normalized["template_code"],
            "operation": operation,
            "status": "applied",
            "version": version,
            "target_version": target_version,
            "before_state": deepcopy(before),
            "after_state": deepcopy(persisted),
            "diff": diff_domain_templates(before, normalized),
            "actor_uid": str(actor_uid),
        }
        return self.repository.apply_import(
            {
                "uid": self.uid_factory(),
                "template_code": normalized["template_code"],
                "version": version,
                "template": persisted,
                "audit": audit,
            }
        )

    def rollback(
        self,
        template_code: str,
        *,
        target_version: int,
        actor_uid: str,
    ):
        template_code = _require_identifier(template_code, "template_code")
        if not isinstance(target_version, int) or target_version < 1:
            raise DomainTemplateValidationError("target_version must be a positive integer")
        target = self.repository.get_version(template_code, target_version)
        if target is None:
            raise DomainTemplateNotFound(
                f"domain template {template_code} version {target_version} was not found"
            )
        definition = {
            key: deepcopy(value)
            for key, value in target.items()
            if key not in {"current_version", "content_hash"}
        }
        return self.import_template(
            definition,
            actor_uid=actor_uid,
            operation="rollback",
            target_version=target_version,
        )