feat(persistence): ExternalUserRepository scaffold (Stage 1 PR1)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/packages/harness/deerflow/persistence/external_user/__init__.py

@@ -12,5 +12,6 @@ header parsing, and quota attribution all live in Stage 1.
 from __future__ import annotations
 
 from deerflow.persistence.external_user.model import ExternalUserRow
+from deerflow.persistence.external_user.sql import ExternalUserRepository
 
-__all__ = ["ExternalUserRow"]
+__all__ = ["ExternalUserRepository", "ExternalUserRow"]

backend/packages/harness/deerflow/persistence/external_user/sql.py

@@ -0,0 +1,98 @@
+"""SQLAlchemy-backed external user repository (Stage 1 PR1).
+
+Built but NOT yet wired to any auth path — the X-External-User-Id
+passthrough that calls ``upsert`` lands in a later track-2 PR. ``upsert``
+is idempotent on (service_account_id, external_id).
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.external_user.model import ExternalUserRow
+
+
+class ExternalUserRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    @staticmethod
+    def _row_to_dict(row: ExternalUserRow) -> dict[str, Any]:
+        return {
+            "id": row.id,
+            "workspace_id": row.workspace_id,
+            "service_account_id": row.service_account_id,
+            "external_id": row.external_id,
+            "display_name": row.display_name,
+            "metadata": dict(row.metadata_json or {}),
+            "created_at": row.created_at.isoformat() if row.created_at else None,
+            "last_seen_at": row.last_seen_at.isoformat() if row.last_seen_at else None,
+        }
+
+    async def get(self, external_user_id: str) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = await session.get(ExternalUserRow, external_user_id)
+            return self._row_to_dict(row) if row else None
+
+    async def get_by_external_id(self, *, service_account_id: str, external_id: str) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            result = await session.execute(
+                select(ExternalUserRow).where(
+                    ExternalUserRow.service_account_id == service_account_id,
+                    ExternalUserRow.external_id == external_id,
+                )
+            )
+            row = result.scalar_one_or_none()
+            return self._row_to_dict(row) if row else None
+
+    async def list_by_workspace(self, workspace_id: str) -> list[dict[str, Any]]:
+        async with self._sf() as session:
+            result = await session.execute(select(ExternalUserRow).where(ExternalUserRow.workspace_id == workspace_id).order_by(ExternalUserRow.created_at.desc()))
+            return [self._row_to_dict(r) for r in result.scalars()]
+
+    async def upsert(
+        self,
+        *,
+        workspace_id: str,
+        service_account_id: str,
+        external_id: str,
+        display_name: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Insert a new external user or refresh ``last_seen_at`` on an
+        existing (service_account_id, external_id) row."""
+        now = datetime.now(UTC)
+        async with self._sf() as session:
+            result = await session.execute(
+                select(ExternalUserRow).where(
+                    ExternalUserRow.service_account_id == service_account_id,
+                    ExternalUserRow.external_id == external_id,
+                )
+            )
+            row = result.scalar_one_or_none()
+            if row is None:
+                row = ExternalUserRow(
+                    id=str(uuid.uuid4()),
+                    workspace_id=workspace_id,
+                    service_account_id=service_account_id,
+                    external_id=external_id,
+                    display_name=display_name,
+                    metadata_json=metadata or {},
+                    created_at=now,
+                    last_seen_at=now,
+                )
+                session.add(row)
+            else:
+                row.last_seen_at = now
+                if display_name is not None:
+                    row.display_name = display_name
+                if metadata is not None:
+                    row.metadata_json = metadata
+            await session.commit()
+            await session.refresh(row)
+            return self._row_to_dict(row)

backend/tests/test_external_user_repo.py

@@ -0,0 +1,77 @@
+"""Tests for ExternalUserRepository (Stage 1 PR1).
+
+Repository is built but not yet wired to any auth path. upsert is
+idempotent on (service_account_id, external_id) per the table's
+UniqueConstraint uq_external_users_sa_external.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from deerflow.persistence.external_user import ExternalUserRepository
+from deerflow.persistence.service_account.model import ServiceAccountRow
+from deerflow.persistence.user.model import UserRow
+from deerflow.persistence.workspace.model import WorkspaceRow
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend() -> str:
+    return "asyncio"
+
+
+async def _make_repo(tmp_path):
+    from deerflow.persistence.engine import get_session_factory, init_engine
+
+    url = f"sqlite+aiosqlite:///{tmp_path / 'test.db'}"
+    await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
+    return ExternalUserRepository(get_session_factory())
+
+
+async def _cleanup():
+    from deerflow.persistence.engine import close_engine
+
+    await close_engine()
+
+
+async def _seed_sa(repo) -> None:
+    async with repo._sf() as session:
+        session.add(UserRow(id="u-alice", email="alice@example.com"))
+        await session.commit()
+    async with repo._sf() as session:
+        session.add(WorkspaceRow(id="w-1", name="WS", slug="ws", owner_id="u-alice"))
+        await session.commit()
+    async with repo._sf() as session:
+        session.add(ServiceAccountRow(id="sa-1", workspace_id="w-1", name="bot", role="member", identity_mode="external_passthrough", status="active", created_by="u-alice"))
+        await session.commit()
+
+
+async def test_upsert_inserts_then_updates_same_row(tmp_path):
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_sa(repo)
+        first = await repo.upsert(workspace_id="w-1", service_account_id="sa-1", external_id="ext-42", display_name="alice")
+        assert first["external_id"] == "ext-42"
+        assert first["last_seen_at"] is not None
+        second = await repo.upsert(workspace_id="w-1", service_account_id="sa-1", external_id="ext-42")
+        # same logical row (no duplicate)
+        assert second["id"] == first["id"]
+        rows = await repo.list_by_workspace("w-1")
+        assert len(rows) == 1
+    finally:
+        await _cleanup()
+
+
+async def test_get_by_external_id(tmp_path):
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_sa(repo)
+        await repo.upsert(workspace_id="w-1", service_account_id="sa-1", external_id="ext-7")
+        found = await repo.get_by_external_id(service_account_id="sa-1", external_id="ext-7")
+        assert found is not None
+        assert found["external_id"] == "ext-7"
+        assert await repo.get_by_external_id(service_account_id="sa-1", external_id="nope") is None
+    finally:
+        await _cleanup()