feat(privacy): enforce avatar/presence visibility and invite restrictions

app/users/service.py

@@ -2,6 +2,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.users import repository
 from app.users.models import User
+from app.users.schemas import UserRead, UserSearchRead
 
 
 async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:
@@ -96,3 +97,51 @@ async def remove_contact(db: AsyncSession, *, user_id: int, contact_user_id: int
 
 async def list_contacts(db: AsyncSession, *, user_id: int) -> list[User]:
     return await repository.list_contacts(db, user_id=user_id)
+
+
+async def can_view_user_avatar(db: AsyncSession, *, target_user: User, viewer_user_id: int) -> bool:
+    if target_user.id == viewer_user_id:
+        return True
+    if target_user.privacy_avatar == "everyone":
+        return True
+    if target_user.privacy_avatar == "nobody":
+        return False
+    return await repository.is_user_in_contacts(db, owner_user_id=target_user.id, candidate_user_id=viewer_user_id)
+
+
+async def can_view_user_last_seen(db: AsyncSession, *, target_user: User, viewer_user_id: int) -> bool:
+    if target_user.id == viewer_user_id:
+        return True
+    if target_user.privacy_last_seen == "everyone":
+        return True
+    if target_user.privacy_last_seen == "nobody":
+        return False
+    return await repository.is_user_in_contacts(db, owner_user_id=target_user.id, candidate_user_id=viewer_user_id)
+
+
+async def can_invite_user_to_groups(db: AsyncSession, *, target_user: User, actor_user_id: int) -> bool:
+    if target_user.id == actor_user_id:
+        return False
+    if target_user.privacy_group_invites == "everyone":
+        return True
+    return await repository.is_user_in_contacts(db, owner_user_id=target_user.id, candidate_user_id=actor_user_id)
+
+
+async def serialize_user_for_viewer(db: AsyncSession, *, target_user: User, viewer_user_id: int) -> UserRead:
+    payload = UserRead.model_validate(target_user).model_dump()
+    if not await can_view_user_avatar(db, target_user=target_user, viewer_user_id=viewer_user_id):
+        payload["avatar_url"] = None
+    if target_user.id != viewer_user_id:
+        payload["allow_private_messages"] = True
+        payload["privacy_last_seen"] = "everyone"
+        payload["privacy_avatar"] = "everyone"
+        payload["privacy_group_invites"] = "everyone"
+        payload["twofa_enabled"] = False
+    return UserRead.model_validate(payload)
+
+
+async def serialize_user_search_for_viewer(db: AsyncSession, *, target_user: User, viewer_user_id: int) -> UserSearchRead:
+    payload = UserSearchRead.model_validate(target_user).model_dump()
+    if not await can_view_user_avatar(db, target_user=target_user, viewer_user_id=viewer_user_id):
+        payload["avatar_url"] = None
+    return UserSearchRead.model_validate(payload)