From def45dd0c61061e20621bbea73f518c65dc5f6ca Mon Sep 17 00:00:00 2001 From: 1445043649 <> Date: Wed, 13 May 2026 09:12:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(persistence):=20backfill=20Step=203=20?= =?UTF-8?q?=E2=80=94=20orphan=20rows=20->=20legacy=5Fworkspace?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _ensure_legacy_workspace creates the nil-UUID anchor (slug=legacy) owned by the platform admin (or oldest user as fallback). Raises a clear error if the DB has no users at all so we never silently create an orphaned workspace. Step 3 UPDATEs each table's remaining workspace_id IS NULL rows to LEGACY_WORKSPACE_ID. Orchestrator wires ensure-then-loop between Step 2 and Step 3. 3 new tests: orphan fan-out, no-users error, end-to-end orchestrator with mixed owned + orphan rows. Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/scripts/backfill_workspace_id.py | 78 ++++++++++++++++-- backend/tests/test_backfill_workspace_id.py | 90 ++++++++++++++++++++- 2 files changed, 161 insertions(+), 7 deletions(-) diff --git a/backend/scripts/backfill_workspace_id.py b/backend/scripts/backfill_workspace_id.py index 7ab84d59..e8c6ea77 100644 --- a/backend/scripts/backfill_workspace_id.py +++ b/backend/scripts/backfill_workspace_id.py @@ -158,6 +158,57 @@ async def _step2_update_table_from_users( return int(rowcount) +async def _ensure_legacy_workspace( + session_factory: async_sessionmaker[AsyncSession], + *, + dry_run: bool, +) -> bool: + """Create the ``legacy_workspace`` anchor row idempotently. + + The anchor is needed before Step 3 can point orphan rows at it. We + pick the platform admin (``system_role='admin'``) as owner; if no + admin exists yet we fall back to the oldest user. If the database + has no users at all we refuse to continue — running this script on + an unbootstrapped DB would create a workspace with no owner and the + FK to ``users`` would fail anyway. + + Returns True if the workspace was just created (or would be, under + ``dry_run``). False if it already existed. + """ + from deerflow.persistence.workspace.model import WorkspaceRow + + async with session_factory() as session: + existing = await session.get(WorkspaceRow, LEGACY_WORKSPACE_ID) + if existing is not None: + return False + + async with session_factory() as session: + admin_id = (await session.execute(select(UserRow.id).where(UserRow.system_role == "admin").order_by(UserRow.created_at).limit(1))).scalar_one_or_none() + if admin_id is None: + admin_id = (await session.execute(select(UserRow.id).order_by(UserRow.created_at).limit(1))).scalar_one_or_none() + + if admin_id is None: + raise RuntimeError( + "Cannot create legacy_workspace: no users exist. Bootstrap an admin via /auth/initialize before running backfill.", + ) + + if dry_run: + logger.info("WOULD create legacy_workspace (id=%s) owned by user=%s", LEGACY_WORKSPACE_ID, admin_id) + return True + + ws_repo = WorkspaceRepository(session_factory) + await ws_repo.create( + workspace_id=LEGACY_WORKSPACE_ID, + name=LEGACY_WORKSPACE_NAME, + slug=LEGACY_WORKSPACE_SLUG, + owner_id=admin_id, + ) + m_repo = WorkspaceMembershipRepository(session_factory) + await m_repo.add(workspace_id=LEGACY_WORKSPACE_ID, user_id=admin_id, role="owner") + logger.info("Created legacy_workspace (id=%s) owned by user=%s", LEGACY_WORKSPACE_ID, admin_id) + return True + + async def _step3_assign_legacy_workspace( session_factory: async_sessionmaker[AsyncSession], table: str, @@ -166,12 +217,28 @@ async def _step3_assign_legacy_workspace( ) -> int: """Assign LEGACY_WORKSPACE_ID to *table* rows still missing workspace_id. - Filled in by T5.7 (also responsible for ensuring the legacy workspace row exists). + Callers should ensure :func:`_ensure_legacy_workspace` has run first; + the orchestrator does this between Step 2 and Step 3. Orphan rows are + rows whose ``user_id`` was already NULL (or pointed at a deleted user) + so Step 2's correlated subquery left them untouched. """ - _ = session_factory - _ = table - _ = dry_run - return 0 + model = _TABLE_MODELS[table] + workspace_col = model.workspace_id + + if dry_run: + count_stmt = select(func.count()).select_from(model).where(workspace_col.is_(None)) + async with session_factory() as session: + count = (await session.execute(count_stmt)).scalar_one() or 0 + logger.info("WOULD assign %d orphan row(s) in %s to legacy_workspace", count, table) + return int(count) + + stmt = update(model).where(workspace_col.is_(None)).values(workspace_id=LEGACY_WORKSPACE_ID) + async with session_factory() as session: + result = await session.execute(stmt) + await session.commit() + rowcount = result.rowcount or 0 + logger.info("Assigned %d orphan row(s) in %s to legacy_workspace", rowcount, table) + return int(rowcount) async def backfill( @@ -189,6 +256,7 @@ async def backfill( report["users_workspaces_created"] = await _step1_create_workspaces_for_users(session_factory, dry_run=dry_run) for table in _BUSINESS_TABLES: report[f"{table}_from_users"] = await _step2_update_table_from_users(session_factory, table, dry_run=dry_run) + report["legacy_workspace_created"] = await _ensure_legacy_workspace(session_factory, dry_run=dry_run) for table in _BUSINESS_TABLES: report[f"{table}_legacy"] = await _step3_assign_legacy_workspace(session_factory, table, dry_run=dry_run) diff --git a/backend/tests/test_backfill_workspace_id.py b/backend/tests/test_backfill_workspace_id.py index c2a49b5a..f58c263f 100644 --- a/backend/tests/test_backfill_workspace_id.py +++ b/backend/tests/test_backfill_workspace_id.py @@ -24,8 +24,12 @@ from deerflow.persistence.user.model import UserRow from deerflow.persistence.workspace.model import WorkspaceRow from deerflow.persistence.workspace_membership.model import WorkspaceMembershipRow from scripts.backfill_workspace_id import ( + LEGACY_WORKSPACE_ID, + _ensure_legacy_workspace, _step1_create_workspaces_for_users, _step2_update_table_from_users, + _step3_assign_legacy_workspace, + backfill, ) pytestmark = pytest.mark.anyio @@ -50,10 +54,10 @@ async def _close(): await close_engine() -async def _seed_user(sf, *, email: str, default_workspace_id: str | None = None) -> str: +async def _seed_user(sf, *, email: str, default_workspace_id: str | None = None, system_role: str = "user") -> str: user_id = str(uuid.uuid4()) async with sf() as session: - session.add(UserRow(id=user_id, email=email, default_workspace_id=default_workspace_id)) + session.add(UserRow(id=user_id, email=email, default_workspace_id=default_workspace_id, system_role=system_role)) await session.commit() return user_id @@ -196,3 +200,85 @@ async def test_step1_skips_blacklisted_base_slug(tmp_path): assert ws.slug == "admin-2" finally: await _close() + + +# --------------------------------------------------------------------------- +# Step 3: orphan rows -> legacy_workspace +# --------------------------------------------------------------------------- + + +async def test_backfill_orphan_rows_go_to_legacy_workspace(tmp_path): + """Rows with user_id=NULL get assigned the legacy_workspace UUID after Step 3.""" + sf = await _init_engine(tmp_path) + try: + # Seed a platform admin so the legacy workspace has an owner. + await _seed_user(sf, email="admin@example.com", system_role="admin") + # Orphan business rows (user_id=NULL): legacy data from before auth. + await _seed_business_row(sf, ThreadMetaRow, thread_id="t-orphan", user_id=None) + await _seed_business_row(sf, RunRow, run_id="r-orphan", thread_id="t-orphan", user_id=None) + await _seed_business_row(sf, FeedbackRow, feedback_id="f-orphan", thread_id="t-orphan", run_id="r-orphan", user_id=None, rating=1) + await _seed_business_row(sf, RunEventRow, thread_id="t-orphan", run_id="r-orphan", user_id=None, event_type="legacy", category="lifecycle", seq=1) + + # Ensure the anchor + reassign per table. + created = await _ensure_legacy_workspace(sf, dry_run=False) + assert created is True + for table in ("threads_meta", "runs", "feedback", "run_events"): + count = await _step3_assign_legacy_workspace(sf, table, dry_run=False) + assert count == 1, table + + # Re-running the anchor helper is a no-op. + assert await _ensure_legacy_workspace(sf, dry_run=False) is False + + async with sf() as session: + tm = (await session.execute(select(ThreadMetaRow))).scalar_one() + run = (await session.execute(select(RunRow))).scalar_one() + fb = (await session.execute(select(FeedbackRow))).scalar_one() + ev = (await session.execute(select(RunEventRow))).scalar_one() + legacy = (await session.execute(select(WorkspaceRow).where(WorkspaceRow.id == LEGACY_WORKSPACE_ID))).scalar_one() + legacy_mem = (await session.execute(select(WorkspaceMembershipRow).where(WorkspaceMembershipRow.workspace_id == LEGACY_WORKSPACE_ID))).scalar_one() + + assert tm.workspace_id == LEGACY_WORKSPACE_ID + assert run.workspace_id == LEGACY_WORKSPACE_ID + assert fb.workspace_id == LEGACY_WORKSPACE_ID + assert ev.workspace_id == LEGACY_WORKSPACE_ID + assert legacy.slug == "legacy" + assert legacy_mem.role == "owner" + finally: + await _close() + + +async def test_ensure_legacy_workspace_refuses_when_no_users(tmp_path): + """ensure_legacy_workspace raises a clear error if the DB has no users.""" + sf = await _init_engine(tmp_path) + try: + with pytest.raises(RuntimeError, match="no users exist"): + await _ensure_legacy_workspace(sf, dry_run=False) + finally: + await _close() + + +async def test_full_backfill_orchestrator(tmp_path): + """End-to-end: backfill() runs all three steps and reports per-step counts.""" + sf = await _init_engine(tmp_path) + try: + await _seed_user(sf, email="admin@example.com", system_role="admin") + user_id = await _seed_user(sf, email="gina@example.com") + await _seed_business_row(sf, ThreadMetaRow, thread_id="t-owned", user_id=user_id) + await _seed_business_row(sf, ThreadMetaRow, thread_id="t-orphan", user_id=None) + + report = await backfill(sf, dry_run=False) + assert report["dry_run"] is False + # Two users were missing a default workspace (admin too — we + # didn't pre-populate admin's default_workspace_id). + assert report["users_workspaces_created"] == 2 + assert report["threads_meta_from_users"] == 1 + assert report["legacy_workspace_created"] is True + assert report["threads_meta_legacy"] == 1 + + async with sf() as session: + rows = {r.thread_id: r.workspace_id for r in (await session.execute(select(ThreadMetaRow))).scalars().all()} + assert rows["t-orphan"] == LEGACY_WORKSPACE_ID + assert rows["t-owned"] != LEGACY_WORKSPACE_ID + assert rows["t-owned"] is not None + finally: + await _close()