┌──────────────────────────────────────────────┐
│             Metadata Catalog                  │
│  ┌────────────────────────────────────────┐  │
│  │         Ontology Layer                 │  │
│  │  ObjectTypes, Links, Actions           │  │
│  │  (Core data model = metadata skeleton) │  │
│  └────────────────┬───────────────────────┘  │
│                   │                          │
│  ┌────────┐ ┌─────┴──────┐ ┌──────────────┐ │
│  │Technical│ │ Business   │ │ Operational  │ │
│  │Metadata │ │ Metadata   │ │ Metadata     │ │
│  │         │ │            │ │              │ │
│  │• Schema │ │• Descript. │ │• Lineage     │ │
│  │• Stats  │ │• Tags      │ │• Classific.  │ │
│  │• Partns │ │• Owners    │ │• Quality     │ │
│  │• Format │ │• Domains   │ │• Usage stats │ │
│  └────────┘ └────────────┘ └──────────────┘ │
└──────────────────────────────────────────────┘

class CatalogAsset(BaseModel):
    """Metadata catalog asset"""

    asset_id: str = Field(description="Asset unique identifier")
    asset_type: AssetType = Field(description="Asset type")

    name: str
    display_name: str
    description: str = ""
    namespace: str = "default"

    # Business metadata
    owner: AssetOwner
    domain: str = Field(description="Business domain")
    tags: list[Tag] = Field(default_factory=list)
    glossary_terms: list[str] = Field(default_factory=list)

    # Technical metadata
    technical: TechnicalMetadata

    # Operational metadata
    operational: OperationalMetadata

    # Security metadata
    classification: ClassificationLevel
    sensitivity_tags: list[str] = Field(default_factory=list)

    # Quality scores
    quality_score: float = Field(default=0.0, ge=0.0, le=1.0)
    trust_score: float = Field(default=0.0, ge=0.0, le=1.0)

    created_at: datetime
    updated_at: datetime
    last_accessed_at: datetime | None = None


class AssetType(str, Enum):
    OBJECT_TYPE = "object_type"
    LINK_TYPE = "link_type"
    ACTION_TYPE = "action_type"
    DATASET = "dataset"
    PIPELINE = "pipeline"
    DASHBOARD = "dashboard"


class TechnicalMetadata(BaseModel):
    schema_version: int
    properties: list[PropertyMetadata]
    storage_format: str
    partition_spec: dict | None = None
    row_count: int | None = None
    size_bytes: int | None = None
    last_modified: datetime | None = None


class PropertyMetadata(BaseModel):
    name: str
    display_name: str
    data_type: str
    nullable: bool = True
    description: str = ""
    classification: ClassificationLevel | None = None
    statistics: PropertyStatistics | None = None


class PropertyStatistics(BaseModel):
    distinct_count: int | None = None
    null_count: int | None = None
    min_value: str | None = None
    max_value: str | None = None
    avg_length: float | None = None
    sample_values: list[str] = Field(default_factory=list)


class OperationalMetadata(BaseModel):
    lineage_available: bool = False
    upstream_count: int = 0
    downstream_count: int = 0
    pipeline_ids: list[str] = Field(default_factory=list)
    refresh_frequency: str | None = None
    sla_target: str | None = None
    last_refresh: datetime | None = None
    access_count_30d: int = 0
    unique_users_30d: int = 0

class Tag(BaseModel):
    key: str
    value: str | None = None
    category: TagCategory = TagCategory.USER_DEFINED
    source: TagSource = TagSource.MANUAL
    confidence: float = 1.0

class TagCategory(str, Enum):
    DOMAIN = "domain"
    SENSITIVITY = "sensitivity"
    LIFECYCLE = "lifecycle"
    QUALITY = "quality"
    COMPLIANCE = "compliance"
    USER_DEFINED = "user_defined"

class TagSource(str, Enum):
    MANUAL = "manual"
    AUTO_CLASSIFIED = "auto_classified"
    INHERITED = "inherited"
    ML_SUGGESTED = "ml_suggested"

class CatalogSearchEngine:
    """Metadata catalog search engine"""

    async def search(
        self,
        query: str,
        filters: SearchFilters | None = None,
        sort: SortSpec | None = None,
        pagination: Pagination = Pagination(),
    ) -> SearchResult:
        search_query = self._build_query(query)

        if filters:
            if filters.asset_types:
                search_query = search_query.filter_by_types(filters.asset_types)
            if filters.domains:
                search_query = search_query.filter_by_domains(filters.domains)
            if filters.classification_max:
                search_query = search_query.filter_by_classification(
                    max_level=filters.classification_max,
                )
            if filters.tags:
                search_query = search_query.filter_by_tags(filters.tags)
            if filters.owner:
                search_query = search_query.filter_by_owner(filters.owner)

        results = await self._search_index.execute(
            search_query,
            offset=pagination.offset,
            limit=pagination.limit,
        )

        return SearchResult(
            total=results.total,
            items=[self._to_search_hit(r) for r in results.hits],
            facets=results.facets,
        )

    async def suggest(self, prefix: str, limit: int = 10) -> list[Suggestion]:
        return await self._search_index.suggest(prefix, limit)

class DataMap:
    """Data map - visual data asset topology"""

    async def get_domain_map(self, domain: str | None = None) -> DomainTopology:
        assets = await self._catalog.list_assets(domain=domain)

        nodes = []
        edges = []

        for asset in assets:
            nodes.append(MapNode(
                id=asset.asset_id,
                label=asset.display_name,
                type=asset.asset_type,
                classification=asset.classification,
                quality_score=asset.quality_score,
            ))

            lineage = await self._lineage_query.get_downstream(asset.asset_id, max_depth=1)
            for edge in lineage.edges:
                edges.append(MapEdge(
                    source=edge.source_id,
                    target=edge.target_id,
                    type=edge.edge_type,
                ))

        return DomainTopology(nodes=nodes, edges=edges, domain=domain)

    async def get_asset_neighborhood(
        self, asset_id: str, depth: int = 2,
    ) -> AssetNeighborhood:
        upstream = await self._lineage_query.get_upstream(asset_id, max_depth=depth)
        downstream = await self._lineage_query.get_downstream(asset_id, max_depth=depth)

        return AssetNeighborhood(
            center=asset_id,
            upstream_graph=upstream,
            downstream_graph=downstream,
        )

class DataQualityScorer:
    """Data quality scorer"""

    DIMENSIONS = {
        "completeness": 0.25,
        "accuracy": 0.20,
        "freshness": 0.20,
        "consistency": 0.15,
        "documentation": 0.10,
        "accessibility": 0.10,
    }

    async def compute_score(self, asset: CatalogAsset) -> QualityReport:
        scores = {}
        scores["completeness"] = await self._score_completeness(asset)
        scores["accuracy"] = await self._score_accuracy(asset)
        scores["freshness"] = self._score_freshness(asset)
        scores["consistency"] = await self._score_consistency(asset)
        scores["documentation"] = self._score_documentation(asset)
        scores["accessibility"] = self._score_accessibility(asset)

        overall = sum(
            scores[dim] * weight
            for dim, weight in self.DIMENSIONS.items()
        )

        return QualityReport(
            asset_id=asset.asset_id,
            overall_score=overall,
            dimension_scores=scores,
            computed_at=datetime.utcnow(),
        )

    def _score_freshness(self, asset: CatalogAsset) -> float:
        if not asset.operational.last_refresh:
            return 0.0
        age = datetime.utcnow() - asset.operational.last_refresh
        if age < timedelta(hours=1):
            return 1.0
        elif age < timedelta(hours=24):
            return 0.8
        elif age < timedelta(days=7):
            return 0.5
        elif age < timedelta(days=30):
            return 0.3
        else:
            return 0.1

    def _score_documentation(self, asset: CatalogAsset) -> float:
        score = 0.0
        if asset.description:
            score += 0.3
        if asset.owner:
            score += 0.2
        documented = sum(1 for p in asset.technical.properties if p.description)
        total = len(asset.technical.properties)
        if total > 0:
            score += 0.5 * (documented / total)
        return score

class AssetCertification:
    """Data asset certification"""

    class CertificationLevel(str, Enum):
        UNCERTIFIED = "uncertified"
        BRONZE = "bronze"      # Basic documentation complete
        SILVER = "silver"      # Quality checks passed
        GOLD = "gold"          # Fully governance compliant

    CERTIFICATION_REQUIREMENTS = {
        CertificationLevel.BRONZE: [
            "has_description", "has_owner", "has_classification",
        ],
        CertificationLevel.SILVER: [
            "has_description", "has_owner", "has_classification",
            "quality_score_above_0.7", "has_lineage",
        ],
        CertificationLevel.GOLD: [
            "has_description", "has_owner", "has_classification",
            "quality_score_above_0.9", "has_lineage", "has_sla",
            "all_properties_documented", "compliance_review_passed",
        ],
    }

    async def evaluate_certification(
        self, asset: CatalogAsset,
    ) -> CertificationResult:
        results = {}
        for level in [
            self.CertificationLevel.GOLD,
            self.CertificationLevel.SILVER,
            self.CertificationLevel.BRONZE,
        ]:
            requirements = self.CERTIFICATION_REQUIREMENTS[level]
            met = all(self._check_requirement(asset, req) for req in requirements)
            results[level] = met

        if results[self.CertificationLevel.GOLD]:
            achieved = self.CertificationLevel.GOLD
        elif results[self.CertificationLevel.SILVER]:
            achieved = self.CertificationLevel.SILVER
        elif results[self.CertificationLevel.BRONZE]:
            achieved = self.CertificationLevel.BRONZE
        else:
            achieved = self.CertificationLevel.UNCERTIFIED

        return CertificationResult(
            asset_id=asset.asset_id, level=achieved, details=results,
        )

class OwnershipManager:
    """Data asset ownership management"""

    async def assign_owner(
        self, asset_id: str, owner: AssetOwner, assigned_by: str,
    ) -> None:
        asset = await self._catalog.get_asset(asset_id)
        old_owner = asset.owner

        asset.owner = owner
        await self._catalog.update_asset(asset)

        await self._audit.emit(AuditEvent(
            event_type=AuditEventType.SCHEMA_CHANGE,
            action="assign_owner",
            changes=[AuditChange(
                field="owner",
                old_value=old_owner.model_dump_json() if old_owner else None,
                new_value=owner.model_dump_json(),
                change_type="update",
            )],
        ))

    async def find_orphan_assets(self) -> list[CatalogAsset]:
        all_assets = await self._catalog.list_assets()
        return [a for a in all_assets if not a.owner or a.owner.is_inactive]

syntax = "proto3";
package onto.catalog.v1;

service CatalogService {
    rpc SearchAssets(SearchRequest) returns (SearchResponse);
    rpc GetAsset(GetAssetRequest) returns (CatalogAsset);
    rpc ListAssets(ListAssetsRequest) returns (ListAssetsResponse);
    rpc AddTag(AddTagRequest) returns (CatalogAsset);
    rpc RemoveTag(RemoveTagRequest) returns (CatalogAsset);
    rpc GetDataMap(DataMapRequest) returns (DataMapResponse);
    rpc GetAssetNeighborhood(NeighborhoodRequest) returns (NeighborhoodResponse);
    rpc GetQualityReport(QualityRequest) returns (QualityReport);
    rpc GetCertification(CertificationRequest) returns (CertificationResult);
    rpc AssignOwner(AssignOwnerRequest) returns (CatalogAsset);
    rpc FindOrphanAssets(FindOrphansRequest) returns (ListAssetsResponse);
    rpc Suggest(SuggestRequest) returns (SuggestResponse);
}

class TestMetadataCatalog:
    async def test_search_by_keyword(self):
        results = await search_engine.search("employee salary")
        assert len(results.items) > 0
        assert any("Employee" in r.name for r in results.items)

    async def test_search_with_filters(self):
        results = await search_engine.search(
            "customer",
            filters=SearchFilters(
                asset_types=[AssetType.OBJECT_TYPE],
                classification_max=ClassificationLevel.CONFIDENTIAL,
            ),
        )
        for item in results.items:
            assert item.classification <= ClassificationLevel.CONFIDENTIAL

    async def test_quality_scoring(self):
        scorer = DataQualityScorer()
        report = await scorer.compute_score(well_documented_asset)
        assert report.overall_score > 0.7

    async def test_certification_levels(self):
        cert = AssetCertification()
        result = await cert.evaluate_certification(gold_asset)
        assert result.level == AssetCertification.CertificationLevel.GOLD

    async def test_find_orphan_assets(self):
        manager = OwnershipManager(catalog)
        orphans = await manager.find_orphan_assets()
        for orphan in orphans:
            assert not orphan.owner or orphan.owner.is_inactive