feat(db): scheduled DB maintenance — daily targeted VACUUM (ANALYZE)

Adds a daily off-hours VACUUM (ANALYZE) over the high-churn tables the
retention/purge sweeps churn (app_logs, notifications, token tables, notes,
note_versions), on top of Postgres autovacuum, to reclaim bloat left by the
nightly bulk DELETEs and keep planner stats fresh.

- services/db_maintenance.py: run_maintenance() over a closed table allowlist
  via an AUTOCOMMIT connection (VACUUM can't run in a txn); per-table summary
  persisted as the db_maintenance_last_run admin setting.
- services/db_maintenance_scheduler.py: BackgroundScheduler cron (default
  04:00 UTC, after the 03:30 trash purge); enabled-gate checked at fire time;
  live reschedule on hour change. Wired into app.py start/stop.
- routes/admin.py: admin-only GET/PUT /api/admin/db-maintenance + POST /run.
- settings.py: set_admin_setting() (write-side of get_admin_setting) for
  out-of-request writes.
- SettingsView.vue: admin 'Database maintenance' card — enable toggle, run-hour
  (UTC), Run-now, last-run summary.
- Tests: allowlist is closed, VACUUM issued per table, one failure doesn't
  abort the rest, summary persisted; route/scheduler/service surface.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/test_routes_db_maintenance.py

@@ -0,0 +1,43 @@
+"""Structural tests for the DB-maintenance admin routes + scheduler surface."""
+
+
+def test_admin_handlers_callable():
+    from scribe.routes import admin as admin_routes
+    for name in ("get_db_maintenance", "update_db_maintenance", "run_db_maintenance_now"):
+        assert callable(getattr(admin_routes, name))
+
+
+def test_db_maintenance_routes_registered():
+    from scribe.app import create_app
+    app = create_app()
+    rules = {r.rule for r in app.url_map.iter_rules()}
+    assert "/api/admin/db-maintenance" in rules
+    assert "/api/admin/db-maintenance/run" in rules
+
+
+def test_scheduler_surface():
+    from scribe.services import db_maintenance_scheduler as sched
+    for fn in (
+        "start_db_maintenance_scheduler",
+        "stop_db_maintenance_scheduler",
+        "reschedule_db_maintenance",
+        "get_maintenance_hour",
+        "is_maintenance_enabled",
+    ):
+        assert callable(getattr(sched, fn)), f"scheduler missing {fn}"
+
+
+def test_service_surface_and_allowlist():
+    from scribe.services.db_maintenance import MAINTENANCE_TABLES, get_last_run, run_maintenance
+    assert callable(run_maintenance) and callable(get_last_run)
+    # The allowlist is a closed tuple covering exactly the high-churn tables.
+    assert isinstance(MAINTENANCE_TABLES, tuple)
+    assert set(MAINTENANCE_TABLES) == {
+        "app_logs", "notifications", "password_reset_tokens",
+        "invitation_tokens", "notes", "note_versions",
+    }
+
+
+def test_set_admin_setting_exists():
+    from scribe.services.settings import set_admin_setting
+    assert callable(set_admin_setting)

tests/test_services_db_maintenance.py

@@ -0,0 +1,87 @@
+"""Tests for services/db_maintenance.py — mocks the engine (no real DB).
+
+Verifies that VACUUM (ANALYZE) is issued once per allowlisted table, that the
+allowlist is closed (an arbitrary name can't be vacuumed), and that a single
+table failure doesn't abort the rest of the sweep.
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services.db_maintenance import MAINTENANCE_TABLES, run_maintenance
+
+
+def _mock_engine(exec_side_effect=None):
+    """An engine whose connect() yields a conn with a recording exec_driver_sql."""
+    conn = MagicMock()
+    autocommit = MagicMock()
+    autocommit.exec_driver_sql = AsyncMock(side_effect=exec_side_effect)
+    conn.execution_options.return_value = autocommit  # sync on AsyncConnection
+    cm = AsyncMock()
+    cm.__aenter__ = AsyncMock(return_value=conn)
+    cm.__aexit__ = AsyncMock(return_value=False)
+    engine = MagicMock()
+    engine.connect.return_value = cm
+    return engine, autocommit
+
+
+@pytest.mark.asyncio
+async def test_vacuums_each_allowlisted_table_once():
+    engine, autocommit = _mock_engine()
+    with patch("scribe.services.db_maintenance.engine", engine), \
+         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
+        summary = await run_maintenance()
+
+    issued = [c.args[0] for c in autocommit.exec_driver_sql.await_args_list]
+    assert issued == [f"VACUUM (ANALYZE) {t}" for t in MAINTENANCE_TABLES]
+    assert all(r["ok"] for r in summary["tables"])
+    assert len(summary["tables"]) == len(MAINTENANCE_TABLES)
+    # AUTOCOMMIT was requested — VACUUM cannot run inside a transaction.
+    conn = engine.connect.return_value.__aenter__.return_value
+    conn.execution_options.assert_called_once_with(isolation_level="AUTOCOMMIT")
+
+
+@pytest.mark.asyncio
+async def test_allowlist_is_closed():
+    """A caller-supplied name not on the allowlist is silently ignored."""
+    engine, autocommit = _mock_engine()
+    with patch("scribe.services.db_maintenance.engine", engine), \
+         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
+        summary = await run_maintenance(tables=["app_logs", "users; DROP TABLE notes"])
+
+    issued = [c.args[0] for c in autocommit.exec_driver_sql.await_args_list]
+    assert issued == ["VACUUM (ANALYZE) app_logs"]
+    assert [r["table"] for r in summary["tables"]] == ["app_logs"]
+
+
+@pytest.mark.asyncio
+async def test_one_table_failure_does_not_abort_the_rest():
+    # First table raises, the remaining tables still get vacuumed.
+    calls = {"n": 0}
+
+    def flaky(_sql):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("boom")
+        return MagicMock()
+
+    engine, autocommit = _mock_engine(exec_side_effect=flaky)
+    with patch("scribe.services.db_maintenance.engine", engine), \
+         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
+        summary = await run_maintenance()
+
+    assert autocommit.exec_driver_sql.await_count == len(MAINTENANCE_TABLES)
+    assert summary["tables"][0]["ok"] is False
+    assert summary["tables"][0]["error"] == "boom"
+    assert all(r["ok"] for r in summary["tables"][1:])
+
+
+@pytest.mark.asyncio
+async def test_summary_is_persisted_as_admin_setting():
+    engine, _ = _mock_engine()
+    setter = AsyncMock()
+    with patch("scribe.services.db_maintenance.engine", engine), \
+         patch("scribe.services.db_maintenance.set_admin_setting", setter):
+        await run_maintenance()
+    setter.assert_awaited_once()
+    assert setter.await_args.args[0] == "db_maintenance_last_run"