group_fqcd_jr/app/service/relationship_service.py

from collections.abc import Sequence
from typing import Any, Protocol


class GraphDriver(Protocol):
    async def execute_query(self, query: str, **parameters: Any) -> Sequence[Any]: ...


class GraphDegradedResult:
    def __init__(self, reason: str) -> None:
        self.degraded = True
        self.reason = reason


class RelationshipService:
    """Controlled Neo4j boundary; callers cannot submit arbitrary labels or Cypher."""

    ALLOWED_RELATIONSHIPS = frozenset({
        "PREFERS", "HAS_GOAL", "INTERESTED_IN", "TRADED", "HOLDS",
        "TRIGGERED_RISK", "BELONGS_TO_CATEGORY", "EXPOSED_TO_INDUSTRY",
    })

    def __init__(self, driver: GraphDriver) -> None:
        self.driver = driver

    async def neighbors(
        self, customer_id: int, relationship: str, *, limit: int = 50
    ) -> Sequence[Any]:
        if relationship not in self.ALLOWED_RELATIONSHIPS:
            raise ValueError("relationship is not allowed")
        safe_limit = max(1, min(limit, 100))
        query = (
            "MATCH (c:Customer {customer_id: $customer_id})"
            f"-[r:{relationship}]->(n) RETURN n, r LIMIT $limit"
        )
        try:
            return await self.driver.execute_query(query, customer_id=customer_id, limit=safe_limit)
        except Exception as exc:
            return [GraphDegradedResult(f"neo4j_unavailable:{type(exc).__name__}")]

    async def paths(self, customer_id: int, *, max_hops: int = 2, limit: int = 50) -> Sequence[Any]:
        hops = max(1, min(max_hops, 2))
        safe_limit = max(1, min(limit, 100))
        query = (
            "MATCH p=(c:Customer {customer_id: $customer_id})-[*1.."
            f"{hops}]->(n) RETURN p LIMIT $limit"
        )
        try:
            return await self.driver.execute_query(query, customer_id=customer_id, limit=safe_limit)
        except Exception as exc:
            return [GraphDegradedResult(f"neo4j_unavailable:{type(exc).__name__}")]

    async def product_context(self, product_id: int) -> dict[str, object]:
        """Read only approved Product-to-Industry/Category context, never arbitrary Cypher."""
        query = (
            "MATCH (p:Product {product_id: $product_id}) "
            "OPTIONAL MATCH (p)-[:EXPOSED_TO_INDUSTRY]->(i:Industry) "
            "OPTIONAL MATCH (p)-[:BELONGS_TO_CATEGORY]->(c:Category) "
            "RETURN collect(DISTINCT i.name) AS industries, "
            "collect(DISTINCT c.name) AS categories"
        )
        try:
            rows = await self.driver.execute_query(query, product_id=product_id)
        except Exception as exc:
            return {"degraded": True, "reason": f"neo4j_unavailable:{type(exc).__name__}"}
        if not rows:
            return {"degraded": False, "industries": [], "categories": []}
        row = rows[0]
        if isinstance(row, GraphDegradedResult):
            return {"degraded": True, "reason": row.reason}
        if hasattr(row, "data"):
            row = row.data()
        if not isinstance(row, dict):
            return {"degraded": True, "reason": "neo4j_invalid_result"}
        return {
            "degraded": False,
            "industries": self._names(row.get("industries")),
            "categories": self._names(row.get("categories")),
        }

    async def portfolio_industry_context(
        self, customer_id: int, *, limit: int = 5
    ) -> dict[str, object]:
        """Returns only shared product-to-industry relationship evidence for explanation."""
        safe_limit = max(1, min(limit, 20))
        query = (
            "MATCH (c:Customer {customer_id: $customer_id})-[:HOLDS]->(p:Product)"
            "-[:EXPOSED_TO_INDUSTRY]->(i:Industry) "
            "WITH i.name AS industry_name, count(DISTINCT p) AS product_count "
            "WHERE product_count >= 2 "
            "RETURN industry_name, product_count "
            "ORDER BY product_count DESC, industry_name ASC LIMIT $limit"
        )
        try:
            rows = await self.driver.execute_query(
                query, customer_id=customer_id, limit=safe_limit
            )
        except Exception as exc:
            return {"degraded": True, "reason": f"neo4j_unavailable:{type(exc).__name__}"}
        overlaps: list[dict[str, object]] = []
        for row in rows:
            if hasattr(row, "data"):
                row = row.data()
            if not isinstance(row, dict):
                return {"degraded": True, "reason": "neo4j_invalid_result"}
            name, count = row.get("industry_name"), row.get("product_count")
            if isinstance(name, str) and isinstance(count, int) and count >= 2:
                overlaps.append({"industry_name": name, "product_count": count})
        return {"degraded": False, "overlaps": overlaps}

    async def replace_portfolio_projection(
        self, customer_id: int, holdings: list[dict[str, object]]
    ) -> None:
        """Worker-only snapshot replacement of Customer-HOLDS-Product relationships."""
        if len(holdings) > 500:
            raise ValueError("portfolio projection exceeds holding limit")
        query = (
            "MERGE (c:Customer {customer_id: $customer_id}) "
            "WITH c OPTIONAL MATCH (c)-[old:HOLDS]->(:Product) DELETE old "
            "WITH c UNWIND $holdings AS holding "
            "MERGE (p:Product {product_id: holding.product_id}) "
            "SET p.product_code = holding.product_code, p.risk_level = holding.risk_level, "
            "p.product_category = holding.product_category "
            "MERGE (c)-[r:HOLDS]->(p) "
            "SET r.source_holding_id = holding.source_holding_id, r.updated_at = holding.updated_at"
        )
        await self.driver.execute_query(query, customer_id=customer_id, holdings=holdings)

    async def replace_product_industry_projection(
        self, products: list[dict[str, object]]
    ) -> None:
        """Worker-only snapshot replacement of Product-EXPOSED_TO_INDUSTRY reference edges."""
        if len(products) > 500:
            raise ValueError("product projection exceeds product limit")
        query = (
            "UNWIND $products AS product "
            "MERGE (p:Product {product_id: product.product_id}) "
            "WITH p, product OPTIONAL MATCH (p)-[old:EXPOSED_TO_INDUSTRY]->(:Industry) DELETE old "
            "WITH p, product UNWIND product.industries AS industry "
            "MERGE (i:Industry {code: industry.industry_code}) "
            "SET i.name = industry.industry_name "
            "MERGE (p)-[r:EXPOSED_TO_INDUSTRY]->(i) "
            "SET r.exposure_weight_pct = industry.exposure_weight_pct, "
            "r.as_of_date = industry.as_of_date, r.source = industry.source"
        )
        await self.driver.execute_query(query, products=products)

    @staticmethod
    def _names(value: object) -> list[str]:
        if not isinstance(value, list):
            return []
        return [item for item in value if isinstance(item, str) and item][:20]