Unify user management and add reviewed identity repairs

backend/app/db.py

@@ -206,6 +206,16 @@ def init_db() -> None:
                 UNIQUE (jellyfin_server_id, jellyfin_user_id)
             )
         """)
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS user_identity_repairs (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                local_user_id INTEGER NOT NULL,
+                before_json TEXT NOT NULL,
+                after_json TEXT NOT NULL,
+                repaired_at TEXT NOT NULL,
+                repaired_by TEXT NOT NULL
+            )
+        """)
         conn.execute("""
             CREATE TABLE IF NOT EXISTS request_repairs (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,

backend/app/routers/identities.py

@@ -2,7 +2,7 @@ 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, resolve_identity
+from ..services.identity_review import confirm_identities, review_identities, resolve_identity, repair_identity
 
 router = APIRouter(prefix="/admin/identities", tags=["admin"], dependencies=[Depends(require_admin)])
 
@@ -53,3 +53,23 @@ async def check_resolution(payload: Resolution, response: Response):
 async def confirm_resolution(payload: ResolutionConfirmation, response: Response, admin: dict = Depends(require_admin)):
     response.headers["Cache-Control"] = "no-store"
     return await resolve_identity(payload.user_id, payload.jellyfin_user_id, payload.revision, admin)
+
+
+class RepairResolution(Resolution):
+    create_seerr: bool = Field(default=False, strict=True)
+
+
+class RepairConfirmation(RepairResolution):
+    revision: str = Field(pattern=r'^[a-f0-9]{64}$')
+
+
+@router.post('/repair/check')
+async def check_repair(payload: RepairResolution, response: Response):
+    response.headers['Cache-Control'] = 'no-store'
+    return await repair_identity(payload.user_id, payload.jellyfin_user_id, create_seerr=payload.create_seerr)
+
+
+@router.post('/repair/confirm')
+async def confirm_repair(payload: RepairConfirmation, response: Response, admin: dict = Depends(require_admin)):
+    response.headers['Cache-Control'] = 'no-store'
+    return await repair_identity(payload.user_id, payload.jellyfin_user_id, payload.revision, admin, payload.create_seerr)

backend/app/services/identity_review.py

@@ -1,10 +1,13 @@
 """Admin-reviewed account links. Live IDs are authoritative; names only suggest candidates."""
 
 import asyncio
+import copy
 import hashlib
 import json
 import re
 import sqlite3
+
+import httpx
 from collections import defaultdict
 from contextlib import closing
 from datetime import datetime, timezone
@@ -116,8 +119,17 @@ async def seerr_directory(runtime):
     return {"state": "unavailable", "users": []}
 
 
-def build_report(local, jellyfin, seerr, jellystat, runtime, selections=None):
+def build_report(local, jellyfin, seerr, jellystat, runtime, selections=None, repair=False):
+    original = local
     selections = selections or {}
+    if repair:
+        local = copy.deepcopy(local)
+        for user in local['users']:
+            if user['id'] in selections:
+                user['jellyseerr_user_id'] = None
+        local['links'] = [link for link in local['links'] if not (
+            link['local_user_id'] in selections and link['source'] == source_key(runtime.jellyfin_base_url))]
+        local['confirmations'] = [item for item in local['confirmations'] if item['local_user_id'] not in selections]
     if any(user_id not in {user['id'] for user in local['users']} for user_id in selections):
         raise HTTPException(404, "This Magent account no longer exists. Run the check again.")
     jf_by_id = {row["id"]: row for row in jellyfin["users"]}
@@ -158,6 +170,10 @@ def build_report(local, jellyfin, seerr, jellystat, runtime, selections=None):
             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 repair and user['id'] in selections and any(
+            item['local_user_id'] == user['id'] and item['jellyfin_server_id'] != jellyfin.get('server_id')
+            for item in original['confirmations']):
+            issues.append('The Jellyfin server changed. A server migration requires separate review.')
         if user["id"] in selections:
             chosen = selections[user["id"]]
             if saved and chosen != saved["jellyfin_user_id"]:
@@ -225,16 +241,16 @@ def build_report(local, jellyfin, seerr, jellystat, runtime, selections=None):
                     "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,
-              "jellyfin_users": jellyfin["users"],
+              "jellyfin_users": jellyfin["users"], "seerr_users": seerr["users"],
               "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["revision"] = digest([report, digest(original), config_digest(runtime), repair])
     report["checked_at"] = datetime.now(timezone.utc).isoformat()
     return report
 
 
-async def review_identities(selections=None):
+async def review_identities(selections=None, repair=False):
     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:
@@ -247,10 +263,10 @@ async def review_identities(selections=None):
         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, selections), local, runtime
+    return build_report(local, jf, seerr, js, runtime, selections, repair), local, runtime
 
 
-def save_confirmations(report, local, runtime, user_ids, admin):
+def save_confirmations(report, local, runtime, user_ids, admin, repair=False):
     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.")
@@ -274,6 +290,20 @@ def save_confirmations(report, local, runtime, user_ids, admin):
                     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"]))
+                if repair:
+                    before_user = next(user for user in local['users'] if user['id'] == user_id)
+                    before = {'seerr_user_id': before_user['jellyseerr_user_id'],
+                              'links': [link for link in local['links'] if link['local_user_id'] == user_id],
+                              'confirmation': next((item for item in local['confirmations'] if item['local_user_id'] == user_id), None)}
+                    conn.execute("""UPDATE user_identity_confirmations SET jellyfin_server_id=?,jellyfin_user_id=?,
+                        jellyfin_source=?,seerr_source=?,seerr_user_id=? WHERE local_user_id=?""",
+                        (report['server_id'], jf_id, source_key(runtime.jellyfin_base_url),
+                         source_key(runtime.jellyseerr_base_url), seerr_id, user_id))
+                    conn.execute("""INSERT INTO user_identity_repairs
+                        (local_user_id,before_json,after_json,repaired_at,repaired_by) VALUES (?,?,?,?,?)""",
+                        (user_id, json.dumps(before, sort_keys=True), json.dumps({
+                            'jellyfin_server_id': report['server_id'], 'jellyfin_user_id': jf_id,
+                            'seerr_user_id': seerr_id}, sort_keys=True), 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}
@@ -294,3 +324,46 @@ async def resolve_identity(user_id, jellyfin_user_id, revision=None, admin=None)
         return await asyncio.to_thread(save_confirmations, report, local, runtime, [user_id], admin)
     return {"revision": report["revision"], "server_id": report["server_id"],
             "row": next(row for row in report["rows"] if row["user"]["id"] == user_id)}
+
+
+async def repair_identity(user_id, jellyfin_user_id, revision=None, admin=None, create_seerr=False):
+    report, local, runtime = await review_identities({user_id: jellyfin_user_id}, repair=True)
+    row = next(row for row in report['rows'] if row['user']['id'] == user_id)
+    importing = bool(create_seerr and row['state'] == 'unlinked' and row['jellyfin']
+                     and not row['seerr'] and row['jellystat']['state'] == 'matched'
+                     and report['services']['seerr'] == 'available')
+    if importing and any(name_key(account['name']) == name_key(row['jellyfin']['name'])
+                         for account in report['seerr_users']):
+        importing = False
+        row['issues'].append('A Seerr account already has this name. Review its existing link before importing.')
+    report['revision'] = digest([report['revision'], create_seerr])
+    if revision is not None:
+        if report['revision'] != revision:
+            raise HTTPException(409, 'The repair preview changed. Check the selected account again.')
+        if importing:
+            if digest(await asyncio.to_thread(read_snapshot)) != digest(local) or config_digest(get_runtime_settings()) != config_digest(runtime):
+                raise HTTPException(409, 'Accounts or settings changed. Preview the repair again.')
+            client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
+            try:
+                await client.post('/api/v1/user/import-from-jellyfin', payload={'jellyfinUserIds': [jellyfin_user_id]})
+            except (httpx.HTTPError, ValueError) as exc:
+                raise HTTPException(502, 'The Seerr import could not be verified. Run a fresh check before trying again; an account may already have been imported.') from exc
+            # Upstream and SQLite cannot share a transaction. Reconcile using live IDs;
+            # never delete an imported account if the local save is blocked or interrupted.
+            refreshed, _, fresh_runtime = await review_identities({user_id: jellyfin_user_id}, repair=True)
+            if config_digest(fresh_runtime) != config_digest(runtime):
+                raise HTTPException(409, 'Seerr import completed but settings changed. Check accounts again before saving Magent links.')
+            try:
+                return await asyncio.to_thread(save_confirmations, refreshed, local, runtime, [user_id], admin, True)
+            except HTTPException as exc:
+                raise HTTPException(409, 'Seerr import completed, but Magent links could not be saved. Run another check to review the imported account. No account was deleted.') from exc
+        return await asyncio.to_thread(save_confirmations, report, local, runtime, [user_id], admin, True)
+    before = next(user for user in local['users'] if user['id'] == user_id)
+    linked = next((link['jellyfin_user_id'] for link in local['links'] if link['local_user_id'] == user_id
+                   and link['source'] == source_key(runtime.jellyfin_base_url)), None)
+    row['can_confirm'] = row['can_confirm'] or importing
+    return {'revision': report['revision'], 'server_id': report['server_id'], 'row': row,
+            'action': 'import_seerr' if importing else 'repair_magent',
+            'before': {'jellyfin_user_id': linked, 'seerr_user_id': before['jellyseerr_user_id']},
+            'seerr_users': report['seerr_users'],
+            'scope': ('Import this single Jellyfin account into Seerr, then verify and save Magent links. Existing Seerr accounts stay unchanged.' if importing else 'Repair Magent links only. Jellyfin and Jellystat IDs and Seerr accounts stay unchanged.')}

backend/tests/test_identity_review.py

@@ -99,6 +99,139 @@ class IdentityReviewTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
             result = await review.resolve_identity(self.user_id, JF, preview['revision'], ADMIN)
             self.assertEqual(result['confirmed'], 1)
 
+    async def test_repair_replaces_wrong_local_link_and_records_before_after(self):
+        link_user('Georgia', OTHER, self.runtime.jellyfin_base_url)
+        db.set_user_jellyseerr_id('Georgia', 999)
+        before = review.read_snapshot()
+        report = review.build_report(before, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
+        self.assertTrue(self.row(report)['can_confirm'])
+        self.assertEqual(review.read_snapshot(), before)
+        review.save_confirmations(report, before, self.runtime, [self.user_id], ADMIN, repair=True)
+        self.assertEqual(linked_user_id('Georgia', self.runtime.jellyfin_base_url), JF)
+        self.assertEqual(db.get_user_by_username('Georgia')['jellyseerr_user_id'], 20)
+        with closing(db._connect()) as conn:
+            audit = conn.execute('SELECT before_json,after_json,repaired_by FROM user_identity_repairs').fetchone()
+        self.assertEqual(json.loads(audit[0])['seerr_user_id'], 999)
+        self.assertEqual(json.loads(audit[1])['jellyfin_user_id'], JF)
+        self.assertEqual(audit[2], 'admin')
+
+    async def test_repair_preserves_duplicate_ownership_and_server_guards(self):
+        db.create_user('Owner', 'password', auth_provider='local', jellyseerr_user_id=20)
+        local = review.read_snapshot()
+        report = review.build_report(local, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
+        self.assertFalse(self.row(report)['can_confirm'])
+        with self.assertRaises(HTTPException):
+            review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN, repair=True)
+        with closing(db._connect()) as conn, conn:
+            conn.execute('DELETE FROM users WHERE username=?', ('Owner',))
+        report, local = self.build()
+        review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
+        self.jf['server_id'] = OTHER
+        report = review.build_report(review.read_snapshot(), self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
+        self.assertFalse(self.row(report)['can_confirm'])
+
+    async def test_repair_does_not_invent_missing_seerr_identity(self):
+        self.seerr['users'][0]['jellyfin_id'] = OTHER
+        local = review.read_snapshot()
+        report = review.build_report(local, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
+        self.assertEqual(self.row(report)['state'], 'unlinked')
+        with self.assertRaises(HTTPException):
+            review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN, repair=True)
+        self.assertEqual(local, review.read_snapshot())
+
+    async def test_repair_audit_failure_rolls_back_links(self):
+        link_user('Georgia', OTHER, self.runtime.jellyfin_base_url)
+        with closing(db._connect()) as conn, conn:
+            conn.execute("CREATE TRIGGER fail_identity_audit BEFORE INSERT ON user_identity_repairs BEGIN SELECT RAISE(ABORT, 'fixture'); END")
+        local = review.read_snapshot()
+        report = review.build_report(local, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
+        with self.assertRaises(HTTPException):
+            review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN, repair=True)
+        self.assertEqual(local, review.read_snapshot())
+
+    async def test_repair_rechecks_revision_and_updates_confirmed_ids(self):
+        report, local = self.build()
+        review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
+        self.jf['users'][0]['id'] = OTHER
+        self.seerr['users'][0]['jellyfin_id'] = OTHER
+        self.js = {OTHER: {'state': 'matched', 'id': OTHER}}
+        with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
+             patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
+             patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js):
+            preview = await review.repair_identity(self.user_id, OTHER)
+            with self.assertRaises(HTTPException):
+                await review.repair_identity(self.user_id, OTHER, 'f' * 64, ADMIN)
+            await review.repair_identity(self.user_id, OTHER, preview['revision'], ADMIN)
+        self.assertEqual(review.read_snapshot()['confirmations'][0]['jellyfin_user_id'], OTHER)
+
+    async def test_single_account_import_is_explicit_and_rechecked_before_local_save(self):
+        self.seerr['users'] = []
+        async def imported(*args, **kwargs):
+            self.seerr['users'] = [{'id': 25, 'name': 'Georgia', 'jellyfin_id': JF}]
+            return []
+        with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
+             patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
+             patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js), \
+             patch.object(review.JellyseerrClient, 'post', new_callable=AsyncMock, side_effect=imported) as post:
+            before = review.read_snapshot()
+            blocked = await review.repair_identity(self.user_id, JF)
+            self.assertFalse(blocked['row']['can_confirm'])
+            preview = await review.repair_identity(self.user_id, JF, create_seerr=True)
+            self.assertEqual(preview['action'], 'import_seerr')
+            self.assertTrue(preview['row']['can_confirm'])
+            post.assert_not_called()
+            self.assertEqual(review.read_snapshot(), before)
+            with self.assertRaises(HTTPException):
+                await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, False)
+            post.assert_not_called()
+            await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, True)
+            post.assert_awaited_once_with('/api/v1/user/import-from-jellyfin', payload={'jellyfinUserIds': [JF]})
+        self.assertEqual(db.get_user_by_username('Georgia')['jellyseerr_user_id'], 25)
+
+    async def test_failed_import_never_writes_local_links_or_retries(self):
+        self.seerr['users'] = []
+        with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
+             patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
+             patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js), \
+             patch.object(review.JellyseerrClient, 'post', new_callable=AsyncMock, side_effect=httpx.ReadTimeout('fixture')) as post:
+            before = review.read_snapshot()
+            preview = await review.repair_identity(self.user_id, JF, create_seerr=True)
+            with self.assertRaises(HTTPException) as error:
+                await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, True)
+            self.assertEqual(error.exception.status_code, 502)
+            self.assertEqual(post.await_count, 1)
+            self.assertEqual(review.read_snapshot(), before)
+
+    async def test_import_preserves_upstream_account_when_local_save_is_blocked(self):
+        self.seerr['users'] = []
+        async def imported(*args, **kwargs):
+            self.seerr['users'] = [{'id': 25, 'name': 'Georgia', 'jellyfin_id': JF}]
+            db.set_user_jellyseerr_id('Georgia', 999)
+        with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
+             patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
+             patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js), \
+             patch.object(review.JellyseerrClient, 'post', new_callable=AsyncMock, side_effect=imported), \
+             patch.object(review.JellyseerrClient, 'delete_user', new_callable=AsyncMock) as delete:
+            preview = await review.repair_identity(self.user_id, JF, create_seerr=True)
+            with self.assertRaises(HTTPException) as error:
+                await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, True)
+            self.assertIn('Seerr import completed', error.exception.detail)
+            self.assertEqual(db.get_user_by_username('Georgia')['jellyseerr_user_id'], 999)
+            self.assertEqual(review.read_snapshot()['confirmations'], [])
+            delete.assert_not_called()
+
+    async def test_import_blocks_existing_name_with_different_jellyfin_id(self):
+        self.seerr['users'] = [{'id': 25, 'name': 'Georgia', 'jellyfin_id': OTHER}]
+        with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
+             patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
+             patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js), \
+             patch.object(review.JellyseerrClient, 'post', new_callable=AsyncMock) as post:
+            preview = await review.repair_identity(self.user_id, JF, create_seerr=True)
+            self.assertFalse(preview['row']['can_confirm'])
+            with self.assertRaises(HTTPException):
+                await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, True)
+            post.assert_not_called()
+
     async def test_georgia_preview_is_read_only_and_uses_seerr_jellyfin_id(self):
         before = review.read_snapshot()
         report, _ = self.build()
@@ -319,8 +452,10 @@ class IdentityRouteTests(unittest.TestCase):
             if endpoint == 'confirm': body['revision'] = 'a' * 64
             for role, status in [(None, 401), ('user', 403)]:
                 self.assertEqual(self.client(role).post('/admin/identities/resolve/' + endpoint, json=body).status_code, status)
+                self.assertEqual(self.client(role).post('/admin/identities/repair/' + endpoint, json=body).status_code, status)
             for invalid in [{'user_id': True}, {'jellyfin_user_id': 'invalid'}, {'seerr_user_id': 22}]:
                 self.assertEqual(self.client('admin').post('/admin/identities/resolve/' + endpoint, json={**body, **invalid}).status_code, 422)
+                self.assertEqual(self.client('admin').post('/admin/identities/repair/' + endpoint, json={**body, **invalid}).status_code, 422)
         with patch.object(identities, 'resolve_identity', new_callable=AsyncMock, return_value={'row': {}}):
             result = self.client('admin').post('/admin/identities/resolve/check', json={'user_id': 1, 'jellyfin_user_id': JF})
             self.assertEqual(result.headers['cache-control'], 'no-store')