provenance/backend/app/services/relationship_service.py

"""Relationship service. Typed, qualified edges between two Persons in a tree.
Writes require editor rights; reads go through the privacy engine."""

import uuid
from datetime import UTC, datetime

from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.enums import ParentChildQualifier, RelationshipType
from app.models.person import Person
from app.models.relationship import Relationship
from app.models.tree import Tree
from app.models.user import User
from app.services import privacy
from app.services.audit import record_audit
from app.services.exceptions import Conflict, Forbidden, NotFound


async def _person_in_tree(session: AsyncSession, person_id: uuid.UUID, tree_id: uuid.UUID) -> bool:
    row = (
        await session.execute(
            select(Person.id).where(
                Person.id == person_id, Person.tree_id == tree_id, Person.deleted_at.is_(None)
            )
        )
    ).scalar_one_or_none()
    return row is not None


async def create_relationship(
    session: AsyncSession,
    *,
    actor: User,
    tree: Tree,
    type: RelationshipType,
    person_from_id: uuid.UUID,
    person_to_id: uuid.UUID,
    qualifier: ParentChildQualifier | None = None,
    notes: str | None = None,
) -> Relationship:
    if not await privacy.can_edit_tree(session, user_id=actor.id, tree=tree):
        raise Forbidden("not an editor of this tree")
    if person_from_id == person_to_id:
        raise Conflict("a relationship needs two different people")
    if qualifier is not None and type is not RelationshipType.parent_child:
        raise Conflict("qualifier only applies to parent_child relationships")
    for pid in (person_from_id, person_to_id):
        if not await _person_in_tree(session, pid, tree.id):
            raise NotFound("person not found in this tree")

    # Reject an equivalent existing edge so the same two people can't be linked
    # the same way twice. parent_child is directional (parent -> child);
    # partnership/sibling are symmetric, so match the pair in either order.
    if type is RelationshipType.parent_child:
        pair = and_(
            Relationship.person_from_id == person_from_id,
            Relationship.person_to_id == person_to_id,
        )
    else:
        pair = or_(
            and_(
                Relationship.person_from_id == person_from_id,
                Relationship.person_to_id == person_to_id,
            ),
            and_(
                Relationship.person_from_id == person_to_id,
                Relationship.person_to_id == person_from_id,
            ),
        )
    existing = (
        await session.execute(
            select(Relationship.id).where(
                Relationship.tree_id == tree.id,
                Relationship.type == type,
                Relationship.deleted_at.is_(None),
                pair,
            )
        )
    ).scalar_one_or_none()
    if existing is not None:
        raise Conflict("these two people are already linked that way")

    relationship = Relationship(
        tree_id=tree.id,
        type=type,
        person_from_id=person_from_id,
        person_to_id=person_to_id,
        qualifier=qualifier,
        notes=notes,
    )
    session.add(relationship)
    await session.flush()
    record_audit(
        session,
        action="create",
        entity_type="Relationship",
        entity_id=relationship.id,
        tree_id=tree.id,
        actor_user_id=actor.id,
        after={"type": type.value, "from": str(person_from_id), "to": str(person_to_id)},
    )
    await session.commit()
    await session.refresh(relationship)
    return relationship


async def list_relationships(
    session: AsyncSession, *, viewer_id: uuid.UUID, tree: Tree
) -> list[Relationship]:
    """All relationships in the tree — powers the family/pedigree view in one call."""
    if not await privacy.can_view_tree(session, user_id=viewer_id, tree=tree):
        raise Forbidden("not permitted to view this tree")
    # Non-members: drop relationships touching a hidden person.
    if await privacy.get_membership_role(session, viewer_id, tree.id) is None:
        from app.services import public_view_service

        return await public_view_service.list_public_relationships(
            session, viewer_id=viewer_id, tree=tree
        )
    stmt = (
        select(Relationship)
        .where(Relationship.tree_id == tree.id, Relationship.deleted_at.is_(None))
        .order_by(Relationship.created_at)
    )
    return list((await session.execute(stmt)).scalars().all())


async def list_relationships_for_person(
    session: AsyncSession, *, viewer_id: uuid.UUID, tree: Tree, person_id: uuid.UUID
) -> list[Relationship]:
    if not await privacy.can_view_tree(session, user_id=viewer_id, tree=tree):
        raise Forbidden("not permitted to view this tree")
    if await privacy.get_membership_role(session, viewer_id, tree.id) is None:
        from app.services import public_view_service

        return await public_view_service.list_public_relationships_for_person(
            session, viewer_id=viewer_id, tree=tree, person_id=person_id
        )
    stmt = (
        select(Relationship)
        .where(
            Relationship.tree_id == tree.id,
            Relationship.deleted_at.is_(None),
            or_(
                Relationship.person_from_id == person_id,
                Relationship.person_to_id == person_id,
            ),
        )
        .order_by(Relationship.created_at)
    )
    return list((await session.execute(stmt)).scalars().all())


async def update_relationship(
    session: AsyncSession, *, actor: User, tree: Tree, relationship_id: uuid.UUID, changes: dict
) -> Relationship:
    if not await privacy.can_edit_tree(session, user_id=actor.id, tree=tree):
        raise Forbidden("not an editor of this tree")
    relationship = (
        await session.execute(
            select(Relationship).where(
                Relationship.id == relationship_id,
                Relationship.tree_id == tree.id,
                Relationship.deleted_at.is_(None),
            )
        )
    ).scalar_one_or_none()
    if relationship is None:
        raise NotFound("relationship not found")
    if (
        "qualifier" in changes
        and changes["qualifier"] is not None
        and relationship.type is not RelationshipType.parent_child
    ):
        raise Conflict("qualifier only applies to parent_child relationships")
    for key in {"qualifier", "notes"} & changes.keys():
        setattr(relationship, key, changes[key])
    record_audit(
        session,
        action="update",
        entity_type="Relationship",
        entity_id=relationship.id,
        tree_id=tree.id,
        actor_user_id=actor.id,
        after=changes,
    )
    await session.commit()
    await session.refresh(relationship)
    return relationship


async def delete_relationship(
    session: AsyncSession, *, actor: User, tree: Tree, relationship_id: uuid.UUID
) -> None:
    if not await privacy.can_edit_tree(session, user_id=actor.id, tree=tree):
        raise Forbidden("not an editor of this tree")
    relationship = (
        await session.execute(
            select(Relationship).where(
                Relationship.id == relationship_id,
                Relationship.tree_id == tree.id,
                Relationship.deleted_at.is_(None),
            )
        )
    ).scalar_one_or_none()
    if relationship is None:
        raise NotFound("relationship not found")
    relationship.deleted_at = datetime.now(UTC)
    record_audit(
        session,
        action="delete",
        entity_type="Relationship",
        entity_id=relationship.id,
        tree_id=tree.id,
        actor_user_id=actor.id,
    )
    await session.commit()