Add admin review and confirmation of cross-service user IDs

backend/app/clients/jellystat.py

@@ -46,6 +46,38 @@ class JellystatClient(ApiClient):
             raise JellystatError("Jellystat returned an unexpected library response")
         return {"connected": True}
 
+    async def check_user_ids(self, user_ids: list[str]) -> dict:
+        """Read metadata for known identities; never scan everyone's playback history."""
+        if not self.configured():
+            return {user_id: {"state": "not_configured"} for user_id in user_ids}
+        results = {user_id: {"state": "unavailable"} for user_id in user_ids}
+        semaphore = asyncio.Semaphore(6)
+        async with httpx.AsyncClient(timeout=8.0) as client:
+            async def check(user_id):
+                if not re.fullmatch(r"[a-f0-9]{32}", user_id):
+                    return
+                async with semaphore:
+                    try:
+                        response = await client.post(f"{self.base_url}/api/getUserDetails",
+                            headers={"x-api-token": self.api_key}, json={"userid": user_id})
+                        if response.status_code == 404 or (response.status_code == 200 and not response.content.strip()):
+                            results[user_id] = {"state": "missing"}
+                            return
+                        response.raise_for_status()
+                        row = response.json()
+                        if row is None:
+                            results[user_id] = {"state": "missing"}
+                        elif isinstance(row, dict) and same_user_id(row.get("Id"), user_id):
+                            results[user_id] = {"state": "matched", "id": user_id, "name": str(row.get("Name") or "")[:200]}
+                    except (httpx.HTTPError, ValueError):
+                        pass
+            try:
+                async with asyncio.timeout(25):
+                    await asyncio.gather(*(check(user_id) for user_id in user_ids))
+            except TimeoutError:
+                pass
+        return results
+
     async def get_user_history(self, user_id: str, start: datetime, end: datetime) -> tuple[list, list]:
         if not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", user_id):
             raise JellystatError("Invalid linked Jellyfin identity")

backend/app/db.py

@@ -193,6 +193,19 @@ def init_db() -> None:
                 PRIMARY KEY (source, local_user_id), UNIQUE (source, jellyfin_user_id)
             )
         """)
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS user_identity_confirmations (
+                local_user_id INTEGER PRIMARY KEY,
+                jellyfin_server_id TEXT NOT NULL,
+                jellyfin_user_id TEXT NOT NULL,
+                jellyfin_source TEXT NOT NULL,
+                seerr_source TEXT NOT NULL,
+                seerr_user_id INTEGER NOT NULL,
+                confirmed_at TEXT NOT NULL,
+                confirmed_by TEXT NOT NULL,
+                UNIQUE (jellyfin_server_id, jellyfin_user_id)
+            )
+        """)
         conn.execute("""
             CREATE TABLE IF NOT EXISTS request_repairs (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -1288,6 +1301,7 @@ def set_user_jellyseerr_id(username: str, jellyseerr_user_id: Optional[int]) ->
         conn.execute(
             """
             UPDATE users SET jellyseerr_user_id = ? WHERE username = ? COLLATE NOCASE
+            AND id NOT IN (SELECT local_user_id FROM user_identity_confirmations)
             """,
             (jellyseerr_user_id, username),
         )

backend/app/main.py

@@ -28,6 +28,7 @@ from .routers.events import router as events_router
 from .routers.portal import router as portal_router
 from .routers.operations import router as operations_router
 from .routers.insights import router as insights_router
+from .routers.identities import router as identities_router
 from .services.jellyfin_sync import run_daily_jellyfin_sync
 from .services.issue_resolution import run_issue_confirmation_loop
 from .services.operation_progress import (
@@ -282,3 +283,4 @@ app.include_router(events_router)
 app.include_router(portal_router)
 app.include_router(operations_router)
 app.include_router(insights_router)
+app.include_router(identities_router)

backend/app/routers/identities.py

@@ -0,0 +1,33 @@
+from fastapi import APIRouter, Depends, Response
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from ..auth import require_admin
+from ..services.identity_review import confirm_identities, review_identities
+
+router = APIRouter(prefix="/admin/identities", tags=["admin"], dependencies=[Depends(require_admin)])
+
+
+class Confirmation(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    revision: str = Field(pattern=r"^[a-f0-9]{64}$")
+    user_ids: list[int] = Field(min_length=1, max_length=3000)
+
+    @field_validator("user_ids")
+    @classmethod
+    def unique_positive_ids(cls, value):
+        if any(user_id <= 0 for user_id in value) or len(set(value)) != len(value):
+            raise ValueError("Choose unique positive user IDs")
+        return value
+
+
+@router.get("")
+async def review(response: Response):
+    response.headers["Cache-Control"] = "no-store"
+    report, _, _ = await review_identities()
+    return report
+
+
+@router.post("/confirm")
+async def confirm(payload: Confirmation, response: Response, admin: dict = Depends(require_admin)):
+    response.headers["Cache-Control"] = "no-store"
+    return await confirm_identities(payload.revision, payload.user_ids, admin)

backend/app/services/identity_review.py

@@ -0,0 +1,277 @@
+"""Admin-reviewed account links. Live IDs are authoritative; names only suggest candidates."""
+
+import asyncio
+import hashlib
+import json
+import re
+import sqlite3
+from collections import defaultdict
+from contextlib import closing
+from datetime import datetime, timezone
+
+from fastapi import HTTPException
+
+from .. import db
+from ..clients.jellyfin import JellyfinClient
+from ..clients.jellyseerr import JellyseerrClient
+from ..clients.jellystat import JellystatClient
+from ..runtime import get_runtime_settings
+from .jellyfin_identity import source_key
+
+MAX_USERS = 3000
+CONFIG_KEYS = ("jellyfin_base_url", "jellyfin_api_key", "jellyseerr_base_url",
+               "jellyseerr_api_key", "jellystat_base_url", "jellystat_api_key")
+
+
+def normalized_id(value):
+    value = str(value or "").lower().replace("-", "")
+    return value if re.fullmatch(r"[a-f0-9]{32}", value) else None
+
+
+def name_key(value):
+    return str(value or "").strip().casefold()
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def config_digest(runtime):
+    return digest([getattr(runtime, key, None) for key in CONFIG_KEYS])
+
+
+def snapshot(conn):
+    conn.row_factory = sqlite3.Row
+    return {
+        "users": [dict(row) for row in conn.execute(
+            "SELECT id, username, role, auth_provider, jellyseerr_user_id FROM users ORDER BY id")],
+        "links": [dict(row) for row in conn.execute(
+            "SELECT source, local_user_id, jellyfin_user_id FROM jellyfin_user_links ORDER BY source, local_user_id")],
+        "confirmations": [dict(row) for row in conn.execute(
+            "SELECT * FROM user_identity_confirmations ORDER BY local_user_id")],
+        # Detect settings changes between checking services and committing the reviewed links.
+        "config_revision": digest([tuple(row) for row in conn.execute(
+            "SELECT key, value FROM settings WHERE key IN (" + ",".join("?" for _ in CONFIG_KEYS) + ") ORDER BY key", CONFIG_KEYS)]),
+    }
+
+
+def read_snapshot():
+    with closing(db._connect()) as conn:
+        return snapshot(conn)
+
+
+async def jellyfin_directory(runtime):
+    client = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key)
+    if not client.configured():
+        return {"state": "not_configured", "users": []}
+    try:
+        users, server = await asyncio.gather(client.get_users(), client.get_system_info())
+        server_id = normalized_id(server.get("Id")) if isinstance(server, dict) else None
+        if not server_id or not isinstance(users, list) or len(users) > MAX_USERS:
+            raise ValueError()
+        clean = []
+        seen = set()
+        for user in users:
+            user_id = normalized_id(user.get("Id"))
+            if not user_id or user_id in seen or normalized_id(user.get("ServerId")) != server_id:
+                raise ValueError()
+            seen.add(user_id)
+            clean.append({"id": user_id, "name": str(user.get("Name") or "")[:200]})
+        return {"state": "available", "server_id": server_id, "users": sorted(clean, key=lambda row: row["id"])}
+    except Exception:
+        return {"state": "unavailable", "users": []}
+
+
+async def seerr_directory(runtime):
+    client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
+    if not client.base_url or not client.api_key:
+        return {"state": "not_configured", "users": []}
+    try:
+        users = []
+        seen = set()
+        expected_total = None
+        async with asyncio.timeout(20):
+            for skip in range(0, MAX_USERS, 100):
+                page = await client.get_users(take=100, skip=skip)
+                total = page["pageInfo"]["results"]
+                batch = page["results"]
+                if type(total) is not int or total < 0 or total > MAX_USERS or not isinstance(batch, list):
+                    raise ValueError()
+                if expected_total is not None and total != expected_total:
+                    raise ValueError()
+                expected_total = total
+                for user in batch:
+                    user_id = user.get("id")
+                    if type(user_id) is not int or user_id <= 0 or user_id in seen:
+                        raise ValueError()
+                    seen.add(user_id)
+                    users.append({"id": user_id, "name": str(user.get("displayName") or user.get("jellyfinUsername") or "")[:200],
+                                  "jellyfin_id": normalized_id(user.get("jellyfinUserId"))})
+                if len(users) == total:
+                    return {"state": "available", "users": sorted(users, key=lambda row: row["id"])}
+                if len(batch) != 100 or len(users) > total:
+                    raise ValueError()
+    except Exception:
+        pass
+    return {"state": "unavailable", "users": []}
+
+
+def build_report(local, jellyfin, seerr, jellystat, runtime):
+    jf_by_id = {row["id"]: row for row in jellyfin["users"]}
+    jf_by_name = defaultdict(list)
+    for row in jellyfin["users"]:
+        jf_by_name[name_key(row["name"])].append(row["id"])
+    seerr_by_id = {row["id"]: row for row in seerr["users"]}
+    seerr_by_jf = defaultdict(list)
+    for row in seerr["users"]:
+        if row["jellyfin_id"]:
+            seerr_by_jf[row["jellyfin_id"]].append(row)
+    current_source = source_key(runtime.jellyfin_base_url)
+    seerr_source = source_key(runtime.jellyseerr_base_url)
+    links = {row["local_user_id"]: normalized_id(row["jellyfin_user_id"]) for row in local["links"] if row["source"] == current_source}
+    confirmed = {row["local_user_id"]: row for row in local["confirmations"]}
+    local_by_name, local_by_seerr = defaultdict(list), defaultdict(list)
+    for user in local["users"]:
+        local_by_name[name_key(user["username"])].append(user["id"])
+        if user["jellyseerr_user_id"] is not None:
+            local_by_seerr[user["jellyseerr_user_id"]].append(user["id"])
+    rows = []
+    for user in local["users"]:
+        issues = []
+        saved = confirmed.get(user["id"])
+        linked = links.get(user["id"])
+        stored_seerr = seerr_by_id.get(user["jellyseerr_user_id"])
+        by_name = jf_by_name.get(name_key(user["username"]), [])
+        basis = "none"
+        candidate = None
+        if saved:
+            candidate = saved["jellyfin_user_id"]
+            basis = "confirmed_id"
+            if saved["jellyfin_server_id"] != jellyfin.get("server_id") or saved["seerr_source"] != seerr_source:
+                issues.append("The confirmed server or Seerr connection has changed.")
+        elif linked:
+            candidate, basis = linked, "stored_jellyfin_id"
+        elif stored_seerr and stored_seerr["jellyfin_id"]:
+            candidate, basis = stored_seerr["jellyfin_id"], "stored_seerr_id"
+        elif user["auth_provider"] == "jellyfin" and len(by_name) == 1:
+            candidate, basis = by_name[0], "suggested_username"
+        if len(local_by_name[name_key(user["username"])]) > 1:
+            issues.append("Multiple Magent rows share this username after case and whitespace normalization.")
+        if len(local_by_seerr.get(user["jellyseerr_user_id"], [])) > 1:
+            issues.append("Multiple Magent rows share the stored Seerr ID.")
+        if len(by_name) > 1:
+            issues.append("This name matches multiple distinct Jellyfin IDs.")
+        if candidate and by_name and candidate not in by_name:
+            issues.append("The stored ID and current Jellyfin username point to different accounts.")
+        if linked and candidate and linked != candidate:
+            issues.append("The stored Jellyfin link conflicts with the confirmed identity.")
+        jf = jf_by_id.get(candidate)
+        if candidate and not jf and jellyfin["state"] == "available":
+            issues.append("The linked Jellyfin ID is absent from the current server.")
+        expected_seerr = seerr_by_jf.get(candidate, [])
+        if len(expected_seerr) > 1:
+            issues.append("Multiple Seerr users reference the same Jellyfin ID.")
+        if user["jellyseerr_user_id"] is not None and seerr["state"] == "available" and (
+            len(expected_seerr) != 1 or expected_seerr[0]["id"] != user["jellyseerr_user_id"]
+        ):
+            issues.append("The stored Seerr ID does not match Seerr's Jellyfin ID mapping.")
+        if saved and user["jellyseerr_user_id"] != saved["seerr_user_id"]:
+            issues.append("The stored Seerr ID has changed since confirmation.")
+        js = jellystat.get(candidate, {"state": "not_checked"})
+        rows.append({"user": user, "jellyfin": jf, "candidate_jellyfin_id": candidate,
+                     "stored_jellyfin_id": linked, "seerr": expected_seerr, "jellystat": js,
+                     "basis": basis, "issues": issues, "confirmed_at": saved["confirmed_at"] if saved else None,
+                     "can_confirm": False, "state": "unlinked"})
+    candidates = defaultdict(list)
+    for row in rows:
+        if row["candidate_jellyfin_id"]:
+            candidates[row["candidate_jellyfin_id"]].append(row)
+    for row in rows:
+        candidate = row["candidate_jellyfin_id"]
+        if len(candidates.get(candidate, [])) > 1:
+            row["issues"].append("Multiple Magent accounts resolve to this Jellyfin ID.")
+        # Also protect IDs already reserved by a link/confirmation whose local user was deleted.
+        if any(link["local_user_id"] != row["user"]["id"] and link["source"] == current_source
+               and normalized_id(link["jellyfin_user_id"]) == candidate for link in local["links"]) or any(
+            item["local_user_id"] != row["user"]["id"] and item["jellyfin_server_id"] == jellyfin.get("server_id")
+            and item["jellyfin_user_id"] == candidate for item in local["confirmations"]):
+            row["issues"].append("This Jellyfin ID is already reserved by another Magent account.")
+        if row["issues"]:
+            row["state"] = "conflict"
+        elif jellyfin["state"] != "available" or seerr["state"] != "available" or (candidate and row["jellystat"]["state"] in {"unavailable", "not_configured"}):
+            row["state"] = "unavailable"
+        elif not row["jellyfin"] or not row["seerr"] or row["jellystat"]["state"] != "matched":
+            row["state"] = "unlinked"
+        elif row["confirmed_at"] and row["stored_jellyfin_id"] == candidate:
+            row["state"] = "confirmed"
+        else:
+            row["state"] = "ready"
+            row["can_confirm"] = True
+    upstream = [{"platform": "Seerr", "id": str(row["id"]), "name": row["name"], "jellyfin_id": row["jellyfin_id"],
+                 "detail": "No current Jellyfin account has this ID."} for row in seerr["users"]
+                if row["jellyfin_id"] not in jf_by_id and jellyfin["state"] == "available"]
+    upstream += [{"platform": "Jellyfin", "id": row["id"], "name": row["name"], "jellyfin_id": row["id"],
+                  "detail": "No Magent account resolves to this ID."} for row in jellyfin["users"] if row["id"] not in candidates]
+    services = {"jellyfin": jellyfin["state"], "seerr": seerr["state"],
+                "jellystat": "not_configured" if not runtime.jellystat_base_url or not runtime.jellystat_api_key else
+                    "not_checked" if not jellystat else
+                    "unavailable" if any(r["state"] == "unavailable" for r in jellystat.values()) else "available"}
+    report = {"server_id": jellyfin.get("server_id"), "services": services, "rows": rows, "upstream": upstream,
+              "counts": {"magent": len(rows), "jellyfin": len(jellyfin["users"]), "seerr": len(seerr["users"]),
+                         "jellystat_checked": sum(r["state"] in {"matched", "missing"} for r in jellystat.values()),
+                         **{state: sum(row["state"] == state for row in rows) for state in ("ready", "confirmed", "conflict", "unlinked", "unavailable")}}}
+    report["revision"] = digest([report, digest(local), config_digest(runtime)])
+    report["checked_at"] = datetime.now(timezone.utc).isoformat()
+    return report
+
+
+async def review_identities():
+    runtime = await asyncio.to_thread(get_runtime_settings)
+    local, jf, seerr = await asyncio.gather(asyncio.to_thread(read_snapshot), jellyfin_directory(runtime), seerr_directory(runtime))
+    if len(local["users"]) > MAX_USERS:
+        raise HTTPException(422, "The identity check supports up to 3,000 Magent accounts.")
+    ids = {row["id"] for row in jf["users"]}
+    ids.update(row["jellyfin_id"] for row in seerr["users"] if row["jellyfin_id"])
+    ids.update(normalized_id(row["jellyfin_user_id"]) for row in local["links"])
+    ids.discard(None)
+    if len(ids) > MAX_USERS:
+        raise HTTPException(422, "There are too many upstream IDs for one identity check.")
+    stats_client = JellystatClient(runtime.jellystat_base_url, runtime.jellystat_api_key)
+    js = await stats_client.check_user_ids(sorted(ids)) if stats_client.configured() else {key: {"state": "not_configured"} for key in ids}
+    return build_report(local, jf, seerr, js, runtime), local, runtime
+
+
+def save_confirmations(report, local, runtime, user_ids, admin):
+    rows = {row["user"]["id"]: row for row in report["rows"]}
+    if any(user_id not in rows or not rows[user_id]["can_confirm"] for user_id in user_ids):
+        raise HTTPException(409, "Some selected accounts cannot be confirmed. Run the check again and review the conflicts.")
+    now = datetime.now(timezone.utc).isoformat()
+    try:
+        with closing(db._connect()) as conn, conn:
+            conn.execute("BEGIN IMMEDIATE")
+            if digest(snapshot(conn)) != digest(local) or config_digest(get_runtime_settings()) != config_digest(runtime):
+                raise HTTPException(409, "Accounts or settings changed during confirmation. Run the check again.")
+            for user_id in user_ids:
+                row = rows[user_id]
+                jf_id = row["jellyfin"]["id"]
+                seerr_id = row["seerr"][0]["id"]
+                conn.execute("""INSERT INTO jellyfin_user_links (source, local_user_id, jellyfin_user_id) VALUES (?, ?, ?)
+                    ON CONFLICT(source,local_user_id) DO UPDATE SET jellyfin_user_id=excluded.jellyfin_user_id""",
+                    (source_key(runtime.jellyfin_base_url), user_id, jf_id))
+                conn.execute("UPDATE users SET jellyseerr_user_id=? WHERE id=?", (seerr_id, user_id))
+                conn.execute("""INSERT INTO user_identity_confirmations
+                    (local_user_id,jellyfin_server_id,jellyfin_user_id,jellyfin_source,seerr_source,seerr_user_id,confirmed_at,confirmed_by)
+                    VALUES (?,?,?,?,?,?,?,?) ON CONFLICT(local_user_id) DO UPDATE SET
+                    jellyfin_source=excluded.jellyfin_source,confirmed_at=excluded.confirmed_at,confirmed_by=excluded.confirmed_by""",
+                    (user_id, report["server_id"], jf_id, source_key(runtime.jellyfin_base_url), source_key(runtime.jellyseerr_base_url),
+                     seerr_id, now, admin["username"]))
+    except sqlite3.IntegrityError as exc:
+        raise HTTPException(409, "An identity is already linked to another account. Run the check again.") from exc
+    return {"confirmed": len(user_ids), "confirmed_at": now}
+
+
+async def confirm_identities(revision, user_ids, admin):
+    report, local, runtime = await review_identities()
+    if report["revision"] != revision:
+        raise HTTPException(409, "The identity check has changed. Run it again before confirming accounts.")
+    return await asyncio.to_thread(save_confirmations, report, local, runtime, user_ids, admin)

backend/app/services/jellyfin_identity.py

@@ -28,6 +28,9 @@ def link_user(username: str, jellyfin_user_id: str, base_url: str | None) -> Non
     if not user or not jellyfin_user_id or not base_url:
         return
     with closing(db._connect()) as conn, conn:
+        if conn.execute("SELECT 1 FROM user_identity_confirmations WHERE local_user_id = ?", (user["id"],)).fetchone():
+            # Reviewed identities are updated only through the admin confirmation workflow.
+            return
         # A renamed or re-created account must not silently take over an existing identity.
         conn.execute(
             "INSERT OR IGNORE INTO jellyfin_user_links (source, local_user_id, jellyfin_user_id) VALUES (?, ?, ?)",