Add self-service profile email management

backend/app/routers/auth.py

@@ -17,6 +17,7 @@ from ..db import (
     set_last_login,
     get_user_by_username,
     get_users_by_username_ci,
+    get_all_users,
     set_user_password,
     set_user_jellyseerr_id,
     set_user_email,
@@ -108,6 +109,18 @@ def _optional_recipient_email(value: object) -> str | None:
     return _require_recipient_email(value)
 
 
+def _optional_account_email(value: object) -> str | None:
+    if value is None or not str(value).strip():
+        return None
+    normalized = normalize_delivery_email(value)
+    if normalized:
+        return normalized
+    raise HTTPException(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        detail="Enter a valid email address.",
+    )
+
+
 def _auth_client_ip(request: Request) -> str:
     direct_host = request.client.host if request.client else None
     if request_trusts_forwarded_headers(direct_host):
@@ -1165,6 +1178,37 @@ async def profile(current_user: dict = Depends(get_current_user)) -> dict:
     }
 
 
+@router.put("/profile/email")
+async def update_profile_email(payload: dict, current_user: dict = Depends(get_current_user)) -> dict:
+    if not isinstance(payload, dict):
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid payload")
+    username = str(current_user.get("username") or "").strip()
+    if not username:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid user")
+
+    email = _optional_account_email(payload.get("email"))
+    if email:
+        duplicate = next(
+            (
+                candidate
+                for candidate in get_all_users()
+                if str(candidate.get("username") or "").casefold() != username.casefold()
+                and str(candidate.get("email") or "").strip().casefold() == email.casefold()
+            ),
+            None,
+        )
+        if duplicate:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT,
+                detail="That email address is already assigned to another account.",
+            )
+
+    if not set_user_email(username, email):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+    logger.info("User updated profile contact email: username=%s email_set=%s", username, bool(email))
+    return {"status": "ok", "email": email}
+
+
 @router.get("/profile/invites")
 async def profile_invites(current_user: dict = Depends(get_current_user)) -> dict:
     username = str(current_user.get("username") or "").strip()

backend/tests/test_backend_quality.py

@@ -1384,6 +1384,48 @@ class SnapshotHistoryTests(TempDatabaseMixin, unittest.TestCase):
 
 
 class AuthFlowTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
+    async def test_user_can_manage_own_profile_email(self) -> None:
+        db.create_user_if_missing("ProfileViewer", "password123", auth_provider="local")
+        current_user = {"username": "ProfileViewer", "role": "user"}
+
+        saved = await auth_router.update_profile_email(
+            {"email": "viewer@example.com"}, current_user
+        )
+        self.assertEqual(saved["email"], "viewer@example.com")
+        self.assertEqual(
+            db.get_user_by_username("profileviewer").get("email"),
+            "viewer@example.com",
+        )
+
+        cleared = await auth_router.update_profile_email({"email": None}, current_user)
+        self.assertIsNone(cleared["email"])
+        self.assertIsNone(db.get_user_by_username("ProfileViewer").get("email"))
+
+    async def test_user_cannot_claim_another_accounts_email(self) -> None:
+        db.create_user_if_missing(
+            "FirstViewer", "password123", email="shared@example.com", auth_provider="local"
+        )
+        db.create_user_if_missing("SecondViewer", "password123", auth_provider="local")
+
+        with self.assertRaises(HTTPException) as context:
+            await auth_router.update_profile_email(
+                {"email": "SHARED@example.com"},
+                {"username": "SecondViewer", "role": "user"},
+            )
+
+        self.assertEqual(context.exception.status_code, 409)
+
+    async def test_profile_email_requires_valid_address(self) -> None:
+        db.create_user_if_missing("ProfileViewer", "password123", auth_provider="local")
+
+        with self.assertRaises(HTTPException) as context:
+            await auth_router.update_profile_email(
+                {"email": "not-an-email"},
+                {"username": "ProfileViewer", "role": "user"},
+            )
+
+        self.assertEqual(context.exception.status_code, 400)
+
     async def test_forgot_password_is_rate_limited(self) -> None:
         request = _build_request(ip="10.1.2.3")
         payload = {"identifier": "resetuser@example.com"}