Alternate names (maiden/married), self-person link, deletion integrity

Names (the genealogy standard: maiden name primary, married/alias as typed
alternates):
- Name model already supported multiple typed names; expose full CRUD —
  NameCreate/Read/Update schemas, name_service (one-primary invariant,
  promote-on-delete), nested /persons/{id}/names routes.
- Person page gains a Names card: add/edit/delete + "make primary", with a
  curated name_type dropdown (birth/maiden, married, alias, nickname, …).

Self-person ("who am I"):
- users.self_person_id FK (use_alter for the users<->persons<->trees cycle)
  + migration; PATCH /users/me/self-person; "This is me" / "This is you"
  on the person page. Soft-deleting the linked person clears it.

Deletion integrity (fixes the broken tree view):
- delete_person now soft-deletes the relationships touching the person, so no
  dangling edges remain; family-chart also filters links to missing people.
- Optional cascade=true recursively deletes descendants (GEDCOM cleanup);
  the person page asks "only this person" vs "with all descendants".
- DELETE returns {deleted: n}.

Family view surfaces "Not connected to anyone" so dangling people aren't lost.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/api/v1/__init__.py

@@ -8,6 +8,7 @@ from app.api.v1 import (
     events,
     gedcom,
     media,
+    names,
     persons,
     relationships,
     sources,
@@ -20,6 +21,7 @@ api_router.include_router(auth.router)
 api_router.include_router(users.router)
 api_router.include_router(trees.router)
 api_router.include_router(persons.router)
+api_router.include_router(names.router)
 api_router.include_router(events.router)
 api_router.include_router(relationships.router)
 api_router.include_router(sources.router)

backend/app/api/v1/names.py

@@ -0,0 +1,90 @@
+import uuid
+
+from fastapi import APIRouter, status
+
+from app.api.deps import CurrentUser, SessionDep
+from app.schemas.name import NameCreate, NameRead, NameUpdate
+from app.services import name_service, tree_service
+
+# Names are nested under their person (which is nested under the tree tenant).
+router = APIRouter(prefix="/trees", tags=["names"])
+
+
+@router.get("/{tree_id}/persons/{person_id}/names", response_model=list[NameRead])
+async def list_names(
+    tree_id: uuid.UUID, person_id: uuid.UUID, session: SessionDep, current: CurrentUser
+) -> list[NameRead]:
+    tree = await tree_service.get_tree(session, viewer_id=current.id, tree_id=tree_id)
+    names = await name_service.list_names(
+        session, viewer_id=current.id, tree=tree, person_id=person_id
+    )
+    return [NameRead.model_validate(n) for n in names]
+
+
+@router.post(
+    "/{tree_id}/persons/{person_id}/names",
+    response_model=NameRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def create_name(
+    tree_id: uuid.UUID,
+    person_id: uuid.UUID,
+    data: NameCreate,
+    session: SessionDep,
+    current: CurrentUser,
+) -> NameRead:
+    tree = await tree_service.get_tree(session, viewer_id=current.id, tree_id=tree_id)
+    name = await name_service.create_name(
+        session,
+        actor=current,
+        tree=tree,
+        person_id=person_id,
+        name_type=data.name_type,
+        given=data.given,
+        surname=data.surname,
+        prefix=data.prefix,
+        suffix=data.suffix,
+        nickname=data.nickname,
+        is_primary=data.is_primary,
+    )
+    return NameRead.model_validate(name)
+
+
+@router.patch(
+    "/{tree_id}/persons/{person_id}/names/{name_id}", response_model=NameRead
+)
+async def update_name(
+    tree_id: uuid.UUID,
+    person_id: uuid.UUID,
+    name_id: uuid.UUID,
+    data: NameUpdate,
+    session: SessionDep,
+    current: CurrentUser,
+) -> NameRead:
+    tree = await tree_service.get_tree(session, viewer_id=current.id, tree_id=tree_id)
+    name = await name_service.update_name(
+        session,
+        actor=current,
+        tree=tree,
+        person_id=person_id,
+        name_id=name_id,
+        changes=data.model_dump(exclude_unset=True),
+    )
+    return NameRead.model_validate(name)
+
+
+@router.delete(
+    "/{tree_id}/persons/{person_id}/names/{name_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+)
+async def delete_name(
+    tree_id: uuid.UUID,
+    person_id: uuid.UUID,
+    name_id: uuid.UUID,
+    session: SessionDep,
+    current: CurrentUser,
+) -> None:
+    tree = await tree_service.get_tree(session, viewer_id=current.id, tree_id=tree_id)
+    await name_service.delete_name(
+        session, actor=current, tree=tree, person_id=person_id, name_id=name_id
+    )

backend/app/api/v1/persons.py

@@ -75,12 +75,21 @@ async def update_person(
     return PersonRead.model_validate(person)
 
 
-@router.delete("/{tree_id}/persons/{person_id}", status_code=status.HTTP_204_NO_CONTENT)
+@router.delete("/{tree_id}/persons/{person_id}")
 async def delete_person(
-    tree_id: uuid.UUID, person_id: uuid.UUID, session: SessionDep, current: CurrentUser
-) -> None:
+    tree_id: uuid.UUID,
+    person_id: uuid.UUID,
+    session: SessionDep,
+    current: CurrentUser,
+    cascade: bool = False,
+) -> dict[str, int]:
+    """Delete a person. ``cascade=true`` also deletes all descendants. Returns
+    the number of persons deleted (1 unless cascading)."""
     tree = await tree_service.get_tree(session, viewer_id=current.id, tree_id=tree_id)
-    await person_service.delete_person(session, actor=current, tree=tree, person_id=person_id)
+    deleted = await person_service.delete_person(
+        session, actor=current, tree=tree, person_id=person_id, cascade=cascade
+    )
+    return {"deleted": deleted}
 
 
 @router.post("/{tree_id}/persons/{person_id}/restore", response_model=PersonRead)

backend/app/api/v1/users.py

@@ -1,7 +1,8 @@
 from fastapi import APIRouter
 
-from app.api.deps import CurrentUser
-from app.schemas.user import UserRead
+from app.api.deps import CurrentUser, SessionDep
+from app.schemas.user import UserRead, UserSelfPersonUpdate
+from app.services import user_service
 
 router = APIRouter(prefix="/users", tags=["users"])
 
@@ -9,3 +10,14 @@ router = APIRouter(prefix="/users", tags=["users"])
 @router.get("/me", response_model=UserRead)
 async def read_me(current: CurrentUser) -> UserRead:
     return UserRead.model_validate(current)
+
+
+@router.patch("/me/self-person", response_model=UserRead)
+async def set_self_person(
+    data: UserSelfPersonUpdate, session: SessionDep, current: CurrentUser
+) -> UserRead:
+    """Link (or unlink) the Person record that represents this account."""
+    user = await user_service.set_self_person(
+        session, user=current, person_id=data.self_person_id
+    )
+    return UserRead.model_validate(user)

backend/app/models/user.py

@@ -3,9 +3,10 @@ multiple auth providers later (the provider-link table arrives with the auth
 slice). ``hashed_password`` is nullable: external/OIDC users have none.
 """
 
+import uuid
 from datetime import datetime
 
-from sqlalchemy import DateTime, String
+from sqlalchemy import DateTime, ForeignKey, String
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.models.base import Base
@@ -19,3 +20,15 @@ class User(Base, UUIDPrimaryKey, Timestamps, SoftDelete):
     email_verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     display_name: Mapped[str | None] = mapped_column(String(255))
     hashed_password: Mapped[str | None] = mapped_column(String(255))
+    # The Person record that *is* this user ("home person"). Cleared if that
+    # person is deleted, so the link can never dangle.
+    self_person_id: Mapped[uuid.UUID | None] = mapped_column(
+        # use_alter + explicit name: users<->persons<->trees form an FK cycle,
+        # so this constraint must be created/dropped via ALTER, not inline.
+        ForeignKey(
+            "persons.id",
+            ondelete="SET NULL",
+            name="fk_users_self_person_id",
+            use_alter=True,
+        )
+    )

backend/app/schemas/name.py

@@ -0,0 +1,42 @@
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class NameCreate(BaseModel):
+    # Open vocabulary: birth/maiden, married, alias, religious, nickname, ...
+    name_type: str = "birth"
+    given: str | None = None
+    surname: str | None = None
+    prefix: str | None = None
+    suffix: str | None = None
+    nickname: str | None = None
+    is_primary: bool = False
+
+
+class NameUpdate(BaseModel):
+    name_type: str | None = None
+    given: str | None = None
+    surname: str | None = None
+    prefix: str | None = None
+    suffix: str | None = None
+    nickname: str | None = None
+    is_primary: bool | None = None
+
+
+class NameRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: uuid.UUID
+    tree_id: uuid.UUID
+    person_id: uuid.UUID
+    name_type: str
+    given: str | None
+    surname: str | None
+    prefix: str | None
+    suffix: str | None
+    nickname: str | None
+    is_primary: bool
+    sort_order: int
+    created_at: datetime

backend/app/schemas/user.py

@@ -19,4 +19,10 @@ class UserRead(BaseModel):
     email: str
     display_name: str | None
     email_verified_at: datetime | None
+    self_person_id: uuid.UUID | None = None
     created_at: datetime
+
+
+class UserSelfPersonUpdate(BaseModel):
+    # null clears the link; otherwise the Person that represents this account.
+    self_person_id: uuid.UUID | None = None

backend/app/services/name_service.py

@@ -0,0 +1,208 @@
+"""Name service. A Person carries one or more Name rows — a primary (typically
+the birth/maiden name) plus typed alternates (married, alias, religious, …).
+Exactly one name is primary at a time; it drives display everywhere. Writes
+require editor rights; reads go through the tree's view check.
+"""
+
+import uuid
+from datetime import UTC, datetime
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.person import Name, Person
+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 Forbidden, NotFound
+
+
+async def _get_person(session: AsyncSession, *, tree: Tree, person_id: uuid.UUID) -> Person:
+    person = (
+        await session.execute(
+            select(Person).where(
+                Person.id == person_id, Person.tree_id == tree.id, Person.deleted_at.is_(None)
+            )
+        )
+    ).scalar_one_or_none()
+    if person is None:
+        raise NotFound("person not found")
+    return person
+
+
+async def _clear_primary(
+    session: AsyncSession, *, person_id: uuid.UUID, keep: uuid.UUID | None
+) -> None:
+    """Demote every other name so exactly one stays primary."""
+    stmt = (
+        update(Name)
+        .where(Name.person_id == person_id, Name.deleted_at.is_(None), Name.is_primary.is_(True))
+        .values(is_primary=False)
+    )
+    if keep is not None:
+        stmt = stmt.where(Name.id != keep)
+    await session.execute(stmt)
+
+
+async def list_names(
+    session: AsyncSession, *, viewer_id: uuid.UUID, tree: Tree, person_id: uuid.UUID
+) -> list[Name]:
+    if not await privacy.can_view_tree(session, user_id=viewer_id, tree=tree):
+        raise Forbidden("not permitted to view this tree")
+    await _get_person(session, tree=tree, person_id=person_id)
+    stmt = (
+        select(Name)
+        .where(Name.person_id == person_id, Name.deleted_at.is_(None))
+        .order_by(Name.is_primary.desc(), Name.sort_order, Name.created_at)
+    )
+    return list((await session.execute(stmt)).scalars().all())
+
+
+async def create_name(
+    session: AsyncSession,
+    *,
+    actor: User,
+    tree: Tree,
+    person_id: uuid.UUID,
+    name_type: str = "birth",
+    given: str | None = None,
+    surname: str | None = None,
+    prefix: str | None = None,
+    suffix: str | None = None,
+    nickname: str | None = None,
+    is_primary: bool = False,
+) -> Name:
+    if not await privacy.can_edit_tree(session, user_id=actor.id, tree=tree):
+        raise Forbidden("not an editor of this tree")
+    await _get_person(session, tree=tree, person_id=person_id)
+
+    # First name for a person is always primary; otherwise honor the flag.
+    existing = (
+        await session.execute(
+            select(Name.id).where(Name.person_id == person_id, Name.deleted_at.is_(None))
+        )
+    ).first()
+    primary = is_primary or existing is None
+    if primary:
+        await _clear_primary(session, person_id=person_id, keep=None)
+
+    name = Name(
+        tree_id=tree.id,
+        person_id=person_id,
+        name_type=name_type,
+        given=given,
+        surname=surname,
+        prefix=prefix,
+        suffix=suffix,
+        nickname=nickname,
+        is_primary=primary,
+    )
+    session.add(name)
+    await session.flush()
+    record_audit(
+        session,
+        action="create",
+        entity_type="Name",
+        entity_id=name.id,
+        tree_id=tree.id,
+        actor_user_id=actor.id,
+        after={"name_type": name_type, "given": given, "surname": surname},
+    )
+    await session.commit()
+    await session.refresh(name)
+    return name
+
+
+_NAME_FIELDS = {"name_type", "given", "surname", "prefix", "suffix", "nickname"}
+
+
+async def update_name(
+    session: AsyncSession,
+    *,
+    actor: User,
+    tree: Tree,
+    person_id: uuid.UUID,
+    name_id: uuid.UUID,
+    changes: dict,
+) -> Name:
+    if not await privacy.can_edit_tree(session, user_id=actor.id, tree=tree):
+        raise Forbidden("not an editor of this tree")
+    name = (
+        await session.execute(
+            select(Name).where(
+                Name.id == name_id,
+                Name.person_id == person_id,
+                Name.tree_id == tree.id,
+                Name.deleted_at.is_(None),
+            )
+        )
+    ).scalar_one_or_none()
+    if name is None:
+        raise NotFound("name not found")
+
+    for key in _NAME_FIELDS & changes.keys():
+        setattr(name, key, changes[key])
+    if changes.get("is_primary") is True:
+        await _clear_primary(session, person_id=person_id, keep=name.id)
+        name.is_primary = True
+
+    record_audit(
+        session,
+        action="update",
+        entity_type="Name",
+        entity_id=name.id,
+        tree_id=tree.id,
+        actor_user_id=actor.id,
+        after=changes,
+    )
+    await session.commit()
+    await session.refresh(name)
+    return name
+
+
+async def delete_name(
+    session: AsyncSession,
+    *,
+    actor: User,
+    tree: Tree,
+    person_id: uuid.UUID,
+    name_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")
+    name = (
+        await session.execute(
+            select(Name).where(
+                Name.id == name_id,
+                Name.person_id == person_id,
+                Name.tree_id == tree.id,
+                Name.deleted_at.is_(None),
+            )
+        )
+    ).scalar_one_or_none()
+    if name is None:
+        raise NotFound("name not found")
+    name.deleted_at = datetime.now(UTC)
+    was_primary = name.is_primary
+    name.is_primary = False
+    record_audit(
+        session,
+        action="delete",
+        entity_type="Name",
+        entity_id=name.id,
+        tree_id=tree.id,
+        actor_user_id=actor.id,
+    )
+    # Promote another name to primary so the person never loses their display name.
+    if was_primary:
+        nxt = (
+            await session.execute(
+                select(Name)
+                .where(Name.person_id == person_id, Name.deleted_at.is_(None))
+                .order_by(Name.sort_order, Name.created_at)
+            )
+        ).scalars().first()
+        if nxt is not None:
+            nxt.is_primary = True
+    await session.commit()

backend/app/services/person_service.py

@@ -6,11 +6,12 @@ person through the privacy engine. Each returned Person gets a transient
 import uuid
 from datetime import UTC, datetime
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.models.enums import PersonPrivacy
+from app.models.enums import PersonPrivacy, RelationshipType
 from app.models.person import Name, Person
+from app.models.relationship import Relationship
 from app.models.tree import Tree
 from app.models.user import User
 from app.services import privacy
@@ -177,9 +178,65 @@ async def get_person(
     return person
 
 
-async def delete_person(
-    session: AsyncSession, *, actor: User, tree: Tree, person_id: uuid.UUID
+async def _children_of(
+    session: AsyncSession, *, tree_id: uuid.UUID, parent_id: uuid.UUID
+) -> list[uuid.UUID]:
+    rows = (
+        await session.execute(
+            select(Relationship.person_to_id).where(
+                Relationship.tree_id == tree_id,
+                Relationship.deleted_at.is_(None),
+                Relationship.type == RelationshipType.parent_child,
+                Relationship.person_from_id == parent_id,
+            )
+        )
+    ).scalars().all()
+    return list(rows)
+
+
+async def _soft_delete_one(
+    session: AsyncSession, *, actor: User, tree: Tree, person: Person, now: datetime
 ) -> None:
+    """Soft-delete a single person and the relationships touching them, so no
+    dangling edges are left to break the tree view."""
+    person.deleted_at = now
+    rels = (
+        await session.execute(
+            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,
+                ),
+            )
+        )
+    ).scalars().all()
+    for rel in rels:
+        rel.deleted_at = now
+    record_audit(
+        session,
+        action="delete",
+        entity_type="Person",
+        entity_id=person.id,
+        tree_id=tree.id,
+        actor_user_id=actor.id,
+        after={"cascaded_relationships": len(rels)},
+    )
+
+
+async def delete_person(
+    session: AsyncSession,
+    *,
+    actor: User,
+    tree: Tree,
+    person_id: uuid.UUID,
+    cascade: bool = False,
+) -> int:
+    """Soft-delete a person. Always removes the relationships that touch them
+    (preventing dangling edges). With ``cascade=True``, recursively deletes
+    their descendants too — handy for pruning a bad GEDCOM import. Returns the
+    number of persons deleted."""
     if not await privacy.can_edit_tree(session, user_id=actor.id, tree=tree):
         raise Forbidden("not an editor of this tree")
     person = (
@@ -191,16 +248,49 @@ async def delete_person(
     ).scalar_one_or_none()
     if person is None:
         raise NotFound("person not found")
-    person.deleted_at = datetime.now(UTC)
-    record_audit(
-        session,
-        action="delete",
-        entity_type="Person",
-        entity_id=person.id,
-        tree_id=tree.id,
-        actor_user_id=actor.id,
+
+    now = datetime.now(UTC)
+
+    # Gather the set of persons to delete. For cascade, walk descendants
+    # breadth-first, guarding against cycles.
+    to_delete: list[Person] = [person]
+    if cascade:
+        seen = {person.id}
+        frontier = [person.id]
+        while frontier:
+            nxt: list[uuid.UUID] = []
+            for pid in frontier:
+                for child_id in await _children_of(session, tree_id=tree.id, parent_id=pid):
+                    if child_id not in seen:
+                        seen.add(child_id)
+                        nxt.append(child_id)
+            frontier = nxt
+        extra_ids = [pid for pid in seen if pid != person.id]
+        if extra_ids:
+            extra = (
+                await session.execute(
+                    select(Person).where(
+                        Person.id.in_(extra_ids),
+                        Person.tree_id == tree.id,
+                        Person.deleted_at.is_(None),
+                    )
+                )
+            ).scalars().all()
+            to_delete.extend(extra)
+
+    for p in to_delete:
+        await _soft_delete_one(session, actor=actor, tree=tree, person=p, now=now)
+
+    # Soft delete leaves the row in place, so the DB-level "ON DELETE SET NULL"
+    # never fires — clear any account's self-person link to a deleted person.
+    await session.execute(
+        update(User)
+        .where(User.self_person_id.in_([p.id for p in to_delete]))
+        .values(self_person_id=None)
     )
+
     await session.commit()
+    return len(to_delete)
 
 
 async def restore_person(

backend/app/services/user_service.py

@@ -8,10 +8,13 @@ import uuid
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.models.person import Person
+from app.models.tree import Tree
 from app.models.user import User
 from app.repositories.base import BaseRepository
+from app.services import privacy
 from app.services.audit import record_audit
-from app.services.exceptions import Conflict
+from app.services.exceptions import Conflict, Forbidden, NotFound
 
 
 async def create_user(
@@ -42,3 +45,39 @@ async def create_user(
 
 async def get_user(session: AsyncSession, user_id: uuid.UUID) -> User | None:
     return await BaseRepository(session, User).get(user_id)
+
+
+async def set_self_person(
+    session: AsyncSession, *, user: User, person_id: uuid.UUID | None
+) -> User:
+    """Point a user's account at the Person record that *is* them ("home
+    person"), or clear it with ``None``. The person must live in a tree the
+    user can view."""
+    if person_id is not None:
+        person = (
+            await session.execute(
+                select(Person).where(Person.id == person_id, Person.deleted_at.is_(None))
+            )
+        ).scalar_one_or_none()
+        if person is None:
+            raise NotFound("person not found")
+        tree = (
+            await session.execute(select(Tree).where(Tree.id == person.tree_id))
+        ).scalar_one_or_none()
+        if tree is None or not await privacy.can_view_tree(
+            session, user_id=user.id, tree=tree
+        ):
+            raise Forbidden("not permitted to link this person")
+
+    user.self_person_id = person_id
+    record_audit(
+        session,
+        action="update",
+        entity_type="User",
+        entity_id=user.id,
+        actor_user_id=user.id,
+        after={"self_person_id": str(person_id) if person_id else None},
+    )
+    await session.commit()
+    await session.refresh(user)
+    return user