Keep a MySQL deployment on MySQL, and stop the last admin disappearing

A legacy copy of DatabaseManager.__init__ was defined at the bottom of
database.py and patched over the real constructor at import time.  It predates
connection pooling and never sets self._mysql_pool, so the first query on a
MySQL-backed manager raises AttributeError.  main.py swallows that into
"Failed to initialize database", which leaves the registry empty, so the next
(config-less) get_config_database() call silently registers the DEFAULT SQLITE
database instead.  A MySQL deployment therefore migrates itself to an empty
~/.aisbf/aisbf.db without saying so.  On aisbf.cloud that happened on
2026-07-22 and split the user table across two databases for three months.

The real constructor already handles both backends, so the patch is deleted
rather than repaired; the only behaviour it carried that the real one lacked -
a config-less cache manager getting its own file and schema - is folded in.

Separately, nothing prevented an installation from losing its last admin.
/dashboard/delete-account checks the password but not the role, admin user
management will delete anyone, and no dashboard page can mint a new admin, so
the first two are one click away from locking everybody out of user
management, settings and the market admin pages.  The same installation sat
with 782 users and zero admins until a row was written by hand.  The guard
goes in the database layer, not the route, so every caller is covered:
delete_user() and update_user() now raise LastAdminError instead of removing,
demoting or deactivating the only active admin.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

aisbf/database.py

@@ -65,6 +65,17 @@ except ImportError:
 
 logger = logging.getLogger(__name__)
 
+
+class LastAdminError(RuntimeError):
+    """Raised when an operation would leave the installation with no admin.
+
+    Nothing recreates an admin from the dashboard, so losing the last one locks
+    everybody out of user management, settings and the market admin pages until
+    somebody writes a row by hand.  aisbf.cloud spent months in exactly that
+    state, so the database refuses the operation instead of trusting callers.
+    """
+
+
 # Global thread pool executor for database operations
 _db_executor = None
 
@@ -134,17 +145,20 @@ class DatabaseManager:
             db_config: Database configuration dictionary. If None, uses default SQLite config.
         """
         if db_config is None:
-            # Default SQLite configuration
+            # Default SQLite configuration.  The cache database gets its own file
+            # and schema name so a config-less fallback can never land cache rows
+            # in the permanent database.
+            is_cache = database_type == 'cache'
             aisbf_dir = Path.home() / '.aisbf'
             aisbf_dir.mkdir(exist_ok=True)
             self.db_config = {
                 'type': 'sqlite',
-                'sqlite_path': str(aisbf_dir / 'aisbf.db'),
+                'sqlite_path': str(aisbf_dir / ('cache.db' if is_cache else 'aisbf.db')),
                 'mysql_host': 'localhost',
                 'mysql_port': 3306,
                 'mysql_user': 'aisbf',
                 'mysql_password': '',
-                'mysql_database': 'aisbf'
+                'mysql_database': 'aisbf_cache' if is_cache else 'aisbf'
             }
         else:
             self.db_config = db_config
@@ -1770,13 +1784,48 @@ class DatabaseManager:
         msg = str(exc).lower()
         return 'no such table' in msg or "doesn't exist" in msg
 
+    def is_last_active_admin(self, user_id: int) -> bool:
+        """Whether this user is the only active admin left.
+
+        Args:
+            user_id: User ID to test
+
+        Returns:
+            True when the user is an active admin and no other active admin exists.
+        """
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            placeholder = '?' if self.db_type == 'sqlite' else '%s'
+            cursor.execute(
+                f'SELECT role, is_active FROM users WHERE id = {placeholder}',
+                (user_id,),
+            )
+            row = cursor.fetchone()
+            if not row or row[0] != 'admin' or not row[1]:
+                return False
+            cursor.execute(
+                f"SELECT COUNT(*) FROM users WHERE role = 'admin' AND is_active = 1 "
+                f'AND id <> {placeholder}',
+                (user_id,),
+            )
+            return cursor.fetchone()[0] == 0
+
     def delete_user(self, user_id: int):
         """
         Delete a user and all their related rows across every child table.
 
         Args:
             user_id: User ID to delete
+
+        Raises:
+            LastAdminError: if this would remove the last active admin.
         """
+        if self.is_last_active_admin(user_id):
+            raise LastAdminError(
+                f'user {user_id} is the last active admin; promote another '
+                'account to admin before deleting this one'
+            )
+
         with self._get_connection() as conn:
             cursor = conn.cursor()
             placeholder = '?' if self.db_type == 'sqlite' else '%s'
@@ -1806,7 +1855,18 @@ class DatabaseManager:
             role: New role (optional)
             is_active: New active status (optional)
             display_name: New display name (optional)
-        """
+
+        Raises:
+            LastAdminError: if this would demote or deactivate the last active admin.
+        """
+        demoting = role is not None and role != 'admin'
+        deactivating = is_active is not None and not is_active
+        if (demoting or deactivating) and self.is_last_active_admin(user_id):
+            raise LastAdminError(
+                f'user {user_id} is the last active admin; promote another '
+                'account to admin before changing this one'
+            )
+
         with self._get_connection() as conn:
             cursor = conn.cursor()
             placeholder = '?' if self.db_type == 'sqlite' else '%s'
@@ -5699,52 +5759,18 @@ def initialize_database(db_config: Optional[Dict[str, Any]] = None):
     logger.info("Database initialized and old records cleaned up")
 
 
-# Now modify DatabaseManager constructor to accept type parameter
-def DatabaseManager__init__(self, db_config: Optional[Dict[str, Any]] = None, database_type: str = DatabaseRegistry.TYPE_CONFIG):
-    """
-    Initialize the database manager.
-
-    Args:
-        db_config: Database configuration dictionary. If None, uses default SQLite config.
-        database_type: TYPE_CONFIG for permanent user data, TYPE_CACHE for temporary cache
-    """
-    self.database_type = database_type
-    
-    if db_config is None:
-        # Default SQLite configuration
-        aisbf_dir = Path.home() / '.aisbf'
-        aisbf_dir.mkdir(exist_ok=True)
-        if database_type == DatabaseRegistry.TYPE_CONFIG:
-            db_path = str(aisbf_dir / 'aisbf.db')
-        else:
-            db_path = str(aisbf_dir / 'cache.db')
-            
-        self.db_config = {
-            'type': 'sqlite',
-            'sqlite_path': db_path,
-            'mysql_host': 'localhost',
-            'mysql_port': 3306,
-            'mysql_user': 'aisbf',
-            'mysql_password': '',
-            'mysql_database': 'aisbf' if database_type == DatabaseRegistry.TYPE_CONFIG else 'aisbf_cache'
-        }
-    else:
-        self.db_config = db_config
-
-    self.db_type = self.db_config.get('type', 'sqlite').lower()
-    self.executor = get_db_executor()
-
-    if self.db_type == 'mysql':
-        # Import the module-level MYSQL_AVAILABLE flag
-        import aisbf.database as db_module
-        if not db_module.MYSQL_AVAILABLE:
-            logger.error(f"❌ DEBUG: MySQL connector not available!")
-            raise ImportError("MySQL connector not available. Install mysql-connector-python.")
-        logger.info(f"✅ DEBUG: MySQL connector available, proceeding...")
-
-    self._initialize_database()
-    logger.info(f"Database initialized: {self.db_type} [TYPE: {self.database_type}]")
-DatabaseManager.__init__ = DatabaseManager__init__
+# NOTE: a copy of DatabaseManager.__init__ used to be defined here and patched
+# over the real constructor.  It predated connection pooling and never set
+# self._mysql_pool, so every MySQL-backed manager raised
+#   AttributeError: 'DatabaseManager' object has no attribute '_mysql_pool'
+# on its first query.  In main.py that exception is swallowed by the
+# "Failed to initialize database" handler, which leaves the registry empty, so
+# the next (config-less) get_config_database() call silently registered the
+# DEFAULT SQLITE database instead.  A MySQL deployment therefore moved itself
+# to an empty ~/.aisbf/aisbf.db without saying so: on aisbf.cloud that happened
+# on 2026-07-22 and stranded three months of users in the wrong database.
+# The real constructor above already handles both backends and the cache path,
+# so the patch is gone rather than repaired.
 
 
 # ============================================================

aisbf/routes/auth.py

@@ -4,7 +4,7 @@ from typing import Optional
 import time, logging, secrets, hashlib, os, re, hmac
 from pathlib import Path
 from datetime import datetime, timedelta
-from aisbf.database import DatabaseRegistry
+from aisbf.database import DatabaseRegistry, LastAdminError
 from aisbf.database import _hash_password as _db_hash_password, _verify_password as _db_verify_password
 from aisbf.app.startup import _login_rate_limit_check, _login_record_failure, _login_clear_failures, _send_admin_notification_email
 from aisbf.app.templates import url_for, get_base_url
@@ -906,7 +906,14 @@ async def dashboard_delete_account_confirm(request: Request, password: str = For
             return RedirectResponse(url=url_for(request, "/dashboard/delete-account?error=Incorrect password"), status_code=303)
 
         username = request.session.get('username', f'user #{user_id}')
-        db.delete_user(user_id)
+        try:
+            db.delete_user(user_id)
+        except LastAdminError:
+            return RedirectResponse(
+                url=url_for(request, "/dashboard/delete-account")
+                + "?error=You+are+the+last+admin.+Promote+another+account+to+admin+before+deleting+this+one.",
+                status_code=303
+            )
 
         _send_admin_notification_email(
             _config,

aisbf/routes/dashboard/settings.py

@@ -4,7 +4,7 @@ from typing import Optional
 import json, logging, os, time, re
 from pathlib import Path
 from datetime import datetime, timedelta
-from aisbf.database import DatabaseRegistry
+from aisbf.database import DatabaseRegistry, LastAdminError
 from aisbf.database import _hash_password as _db_hash_password
 from aisbf import __version__
 from aisbf.app.templates import url_for, get_base_url
@@ -1031,6 +1031,8 @@ async def dashboard_users_delete(request: Request, user_id: int):
     try:
         db.delete_user(user_id)
         return JSONResponse({"success": True})
+    except LastAdminError as e:
+        return JSONResponse({"success": False, "error": str(e)}, status_code=409)
     except Exception as e:
         return JSONResponse({"success": False, "error": str(e)}, status_code=500)
 

tests/test_database_backend_selection.py

+"""A configured backend must be the backend that gets used.
+
+A legacy copy of DatabaseManager.__init__ was patched over the real constructor
+at import time.  It never set self._mysql_pool, so the first query on a
+MySQL-backed manager raised AttributeError; main.py swallows that into
+"Failed to initialize database", leaves the registry empty, and the next
+config-less get_config_database() call registers the DEFAULT SQLite database.
+A MySQL deployment therefore migrated itself to an empty local file in silence.
+On aisbf.cloud that happened on 2026-07-22 and was only noticed three months
+later, with the user table split across two databases.
+"""
+import aisbf.database as database_module
+from aisbf.database import DatabaseManager, DatabaseRegistry
+
+
+def test_no_constructor_patch_shadows_the_real_init():
+    assert not hasattr(database_module, 'DatabaseManager__init__')
+
+
+def test_manager_always_has_a_pool_attribute(tmp_path):
+    """_get_connection() reads self._mysql_pool; it must always exist."""
+    db = DatabaseManager({'type': 'sqlite', 'sqlite_path': str(tmp_path / 'a.db')})
+    assert db._mysql_pool is None
+
+
+def test_mysql_config_is_not_silently_downgraded(tmp_path, monkeypatch):
+    """A MySQL config must reach the manager as MySQL, or raise - never become SQLite."""
+    captured = {}
+
+    def fake_init_pool(self):
+        captured['pooled'] = True
+        self._mysql_pool = None
+
+    def fake_initialize_database(self):
+        captured['initialized'] = True
+
+    monkeypatch.setattr(DatabaseManager, '_init_mysql_pool', fake_init_pool)
+    monkeypatch.setattr(DatabaseManager, '_initialize_database', fake_initialize_database)
+    monkeypatch.setattr(database_module, 'MYSQL_AVAILABLE', True)
+
+    db = DatabaseManager({
+        'type': 'mysql',
+        'mysql_host': '127.0.0.1',
+        'mysql_port': 3306,
+        'mysql_user': 'aisbf',
+        'mysql_password': 'x',
+        'mysql_database': 'aisbf',
+    })
+
+    assert db.db_type == 'mysql'
+    assert captured == {'pooled': True, 'initialized': True}
+
+
+def test_config_less_cache_manager_keeps_its_own_file(tmp_path, monkeypatch):
+    """A config-less cache manager must not fall back onto the permanent database."""
+    monkeypatch.setattr(database_module.Path, 'home', staticmethod(lambda: tmp_path))
+
+    config_db = DatabaseManager(None, database_type=DatabaseRegistry.TYPE_CONFIG)
+    cache_db = DatabaseManager(None, database_type=DatabaseRegistry.TYPE_CACHE)
+
+    assert config_db.db_config['sqlite_path'].endswith('aisbf.db')
+    assert cache_db.db_config['sqlite_path'].endswith('cache.db')
+    assert cache_db.db_config['mysql_database'] == 'aisbf_cache'

tests/test_last_admin_guard.py

+"""The database must never let the last admin disappear.
+
+aisbf.cloud lost its only admin this way: /dashboard/delete-account checks the
+password but not the role, nothing in the dashboard can mint a new admin, and
+the installation sat with 782 users and zero admins until a row was written by
+hand.  These tests pin the guard to the database layer so every caller -
+self-service deletion, admin user management, role edits - is covered.
+"""
+import pytest
+
+from aisbf.database import DatabaseManager, LastAdminError
+
+
+@pytest.fixture
+def db(tmp_path):
+    return DatabaseManager({
+        'type': 'sqlite',
+        'sqlite_path': str(tmp_path / "users.db"),
+    })
+
+
+@pytest.fixture
+def sole_admin(db):
+    return db.create_user('admin', 'hash', role='admin', email='admin@example.com')
+
+
+def test_last_admin_cannot_be_deleted(db, sole_admin):
+    with pytest.raises(LastAdminError):
+        db.delete_user(sole_admin)
+    assert db.get_user_by_id(sole_admin) is not None
+
+
+def test_last_admin_cannot_be_demoted(db, sole_admin):
+    with pytest.raises(LastAdminError):
+        db.update_user(sole_admin, 'admin', role='user')
+    assert db.get_user_by_id(sole_admin)['role'] == 'admin'
+
+
+def test_last_admin_cannot_be_deactivated(db, sole_admin):
+    with pytest.raises(LastAdminError):
+        db.update_user(sole_admin, 'admin', is_active=False)
+    assert db.get_user_by_id(sole_admin)['is_active']
+
+
+def test_second_admin_makes_the_first_deletable(db, sole_admin):
+    other = db.create_user('admin2', 'hash', role='admin', email='admin2@example.com')
+    db.delete_user(sole_admin)
+    assert db.get_user_by_id(sole_admin) is None
+    assert db.get_user_by_id(other)['role'] == 'admin'
+
+
+def test_an_inactive_second_admin_does_not_count(db, sole_admin):
+    disabled = db.create_user('admin2', 'hash', role='admin', email='admin2@example.com')
+    db.update_user(disabled, 'admin2', is_active=False)
+    with pytest.raises(LastAdminError):
+        db.delete_user(sole_admin)
+
+
+def test_plain_users_are_unaffected(db, sole_admin):
+    user = db.create_user('bob', 'hash', role='user', email='bob@example.com')
+    db.delete_user(user)
+    assert db.get_user_by_id(user) is None
+
+
+def test_renaming_the_last_admin_is_allowed(db, sole_admin):
+    """The guard must only fire on demotion/deactivation, not on any edit."""
+    db.update_user(sole_admin, 'admin', display_name='Stefy')
+    assert db.get_user_by_id(sole_admin)['display_name'] == 'Stefy'
+
+
+def test_promoting_the_last_admin_again_is_allowed(db, sole_admin):
+    db.update_user(sole_admin, 'admin', role='admin')
+    assert db.get_user_by_id(sole_admin)['role'] == 'admin'
+
+
+def test_stale_signup_cleanup_still_works(db, sole_admin):
+    """The sweep deletes plain users; the admin guard must not block it."""
+    db.create_user('ghost', 'hash', role='user', email='ghost@example.com')
+    with db._get_connection() as conn:
+        conn.cursor().execute(
+            "UPDATE users SET created_at = datetime('now', '-30 days') "
+            "WHERE username = 'ghost'"
+        )
+        conn.commit()
+    assert db.delete_stale_unverified_signup_users(inactivity_days=14) == 1
+    assert db.get_user_by_username('ghost') is None
+    assert db.get_user_by_id(sole_admin) is not None