feat(api): harden app_user.settings updates against clobbering and races

echo/server/dembrane/api/v2/me.py

@@ -254,17 +254,53 @@ async def update_me(
         cleaned = body.display_name.replace("\r", " ").replace("\n", " ").strip()
         payload["display_name"] = cleaned
 
+    db_updated = False
     if body.settings is not None:
-        existing_settings = app_user.get("settings") or {}
-        if not isinstance(existing_settings, dict):
-            existing_settings = {}
-        merged_settings = {**existing_settings, **body.settings}
-        payload["settings"] = merged_settings
+        # Server-side flatness enforcement: reject any nested dictionaries
+        for k, v in body.settings.items():
+            if isinstance(v, dict):
+                raise HTTPException(
+                    status_code=400,
+                    detail="Nested settings objects are not supported. Settings must be flat."
+                )
+
+        # Try atomic database-level concatenation using psycopg
+        try:
+            import psycopg
+            import json
+            from dembrane.settings import get_settings
+
+            settings_config = get_settings()
+            db_url = settings_config.database.database_url
+            if db_url.startswith("postgresql+psycopg://"):
+                db_url = db_url.replace("postgresql+psycopg://", "postgresql://", 1)
+
+            async with await psycopg.AsyncConnection.connect(db_url, connect_timeout=5) as conn:
+                async with conn.cursor() as cur:
+                    await cur.execute(
+                        "UPDATE app_user SET settings = COALESCE(settings, '{}'::jsonb) || %s::jsonb WHERE id = %s",
+                        (json.dumps(body.settings), app_user["id"])
+                    )
+            db_updated = True
+        except Exception as exc:
+            logger.warning(
+                "Atomic DB settings update failed, falling back to read-merge-write via Directus: %s",
+                exc,
+                exc_info=True
+            )
+
+        if not db_updated:
+            existing_settings = app_user.get("settings") or {}
+            if not isinstance(existing_settings, dict):
+                existing_settings = {}
+            merged_settings = {**existing_settings, **body.settings}
+            payload["settings"] = merged_settings
 
-    if not payload:
+    if not payload and not db_updated:
         raise HTTPException(status_code=400, detail="Nothing to update")
 
-    await async_directus.update_item("app_user", app_user["id"], payload)
+    if payload:
+        await async_directus.update_item("app_user", app_user["id"], payload)
     return {"status": "success"}
 
 

echo/server/pyproject.toml

@@ -36,6 +36,7 @@ dependencies = [
     "types-python-jose>=3.3.4.20240106",
     "litellm==1.84.0",
     # Additional Dependencies
+    "psycopg[binary]>=3.1.18",
     "nest-asyncio==1.6.0",
     "pydantic==2.12.5",
     "pydantic-settings==2.6.1",

echo/server/tests/test_user_settings_flags.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, patch, MagicMock
 
 import pytest
 from httpx import AsyncClient, ASGITransport
@@ -48,7 +48,7 @@ async def test_get_me_returns_settings(
         "display_name": "Test User",
         "settings": {"enable_collapsible_sidebar": True},
     }
-    mock_directus.get_items.return_value = []  # No memberships, etc.
+    mock_directus.get_items = AsyncMock(return_value=[])  # No memberships, etc.
 
     app = _build_app()
     async with AsyncClient(
@@ -102,3 +102,84 @@ async def test_patch_me_updates_and_merges_settings(
             }
         },
     )
+
+
+@pytest.mark.asyncio
+@patch("dembrane.api.v2.me.async_directus")
+@patch("dembrane.api.v2.me.get_app_user_or_raise")
+async def test_patch_me_rejects_nested_settings(
+    mock_get_raise: AsyncMock,
+    mock_directus: AsyncMock,
+):
+    """PATCH /v2/me rejects nested dictionaries under settings with 400."""
+    mock_get_raise.return_value = {
+        "id": _APP_USER_ID,
+        "settings": {},
+    }
+
+    app = _build_app()
+    async with AsyncClient(
+        transport=ASGITransport(app=app), base_url="http://test"
+    ) as ac:
+        response = await ac.patch(
+            "/v2/me",
+            json={
+                "settings": {"nested_key": {"some": "value"}}
+            },
+        )
+
+    assert response.status_code == 400
+    assert "Nested settings objects are not supported" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+@patch("psycopg.AsyncConnection.connect")
+@patch("dembrane.api.v2.me.async_directus")
+@patch("dembrane.api.v2.me.get_app_user_or_raise")
+async def test_patch_me_uses_psycopg_atomically(
+    mock_get_raise: AsyncMock,
+    mock_directus: AsyncMock,
+    mock_psycopg_connect: AsyncMock,
+):
+    """PATCH /v2/me uses psycopg.AsyncConnection to atomically merge settings."""
+    mock_get_raise.return_value = {
+        "id": _APP_USER_ID,
+        "settings": {"existing_flag": True},
+    }
+
+    # Setup the mock connection and cursor context managers
+    mock_cursor = AsyncMock()
+    mock_cursor.__aenter__.return_value = mock_cursor
+    mock_cursor.__aexit__.return_value = None
+
+    mock_conn = AsyncMock()
+    mock_conn.__aenter__.return_value = mock_conn
+    mock_conn.__aexit__.return_value = None
+    mock_conn.cursor = MagicMock(return_value=mock_cursor)
+
+    mock_psycopg_connect.return_value = mock_conn
+
+    app = _build_app()
+    async with AsyncClient(
+        transport=ASGITransport(app=app), base_url="http://test"
+    ) as ac:
+        response = await ac.patch(
+            "/v2/me",
+            json={
+                "settings": {"new_flag": "yes"}
+            },
+        )
+
+    assert response.status_code == 200
+    assert response.json() == {"status": "success"}
+
+    # Verify that the directus update_item is NOT called for settings
+    # because the atomic psycopg update succeeded
+    mock_directus.update_item.assert_not_called()
+
+    # Verify psycopg execute was called with correct arguments
+    mock_cursor.execute.assert_called_once()
+    sql_args = mock_cursor.execute.call_args[0]
+    assert "UPDATE app_user SET settings = COALESCE(settings, '{}'::jsonb) ||" in sql_args[0]
+    assert sql_args[1][0] == '{"new_flag": "yes"}'
+    assert sql_args[1][1] == _APP_USER_ID