"""Sync service that materialises edges into ``rel_relationships``. Two entry points: * ``run_full_sync()`` — full rebuild of every source's edges. Runs daily 04:30 via ``relationships.full_sync`` scheduled job. Handles soft-delete of edges no longer present. * ``run_incremental_sync()`` — runs every 30 min. Re-emits edges for resources whose ``updated_at`` is recent. Soft-delete logic is conservative — only deletes edges whose updated parent is no longer carrying the FK to its current target. Each "source" in the constants module owns a fixed set of (rel_type, from_type, to_type) tuples. The sync function for a source SCANS its authoritative table, EMITS edges, and at the end of the run SOFT-DELETES any same-source edges that weren't touched. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select, text from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.core.safe_logging import get_safe_logger from app.core.txn import best_effort from app.domains.apikey.models import ApiKeyAssignment from app.domains.asset.models import ( AssetItem, AssetSoftwareCatalog, SoftwareSeatAllocation, ) from app.domains.cmdb.models import ( CmdbCloudAccessGrant, CmdbCloudScope, CmdbDomain, CmdbServer, ) from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.domains.permission.models import UserPermission from app.domains.relationships.constants import ( NODE_TYPE_API_KEY, NODE_TYPE_CLOUD_SCOPE, NODE_TYPE_CLOUD_SERVER, NODE_TYPE_DEPARTMENT, NODE_TYPE_DOMAIN, NODE_TYPE_EMAIL_GROUP, NODE_TYPE_HARDWARE_ASSET, NODE_TYPE_PERMISSION, NODE_TYPE_PERSON, NODE_TYPE_SOFTWARE, NODE_TYPE_SOFTWARE_SEAT, REL_ASSIGNED_TO, REL_GRANTS_ACCESS_TO, REL_HAS_API_KEY, REL_HOLDS_SEAT, REL_IT_CONTACT_FOR, REL_MANAGES, REL_MEMBER_OF, REL_OWNS, REL_REPORTS_TO, REL_RESPONSIBLE_FOR, SOURCE_API_KEY, SOURCE_ASSET, SOURCE_CMDB, SOURCE_EMAIL_GROUP, SOURCE_IDENTITY, SOURCE_PERMISSION, SOURCE_SEAT, SOURCE_SOFTWARE, ) from app.domains.relationships.models import RelRelationship from app.identity.person_models import IdentityPerson logger = get_safe_logger(__name__) # Single MySQL/TiDB advisory-lock name shared by full_sync, # incremental_sync, and the admin manual trigger. ``GET_LOCK`` is # session-scoped — it auto-releases when the session closes — but we # also call ``RELEASE_LOCK`` explicitly in a ``finally`` so a # long-lived session (or the connection pool reusing this session) # doesn't end up holding the lock past the run boundary. _REL_SYNC_LOCK_NAME = "helix.relationships.sync" async def _acquire_sync_lock(db: AsyncSession) -> bool: """Try to acquire the relationships-sync advisory lock without blocking. Returns ``True`` on acquisition, ``False`` when another session holds it.""" result = await db.execute( text("SELECT GET_LOCK(:name, 0)"), {"name": _REL_SYNC_LOCK_NAME}, ) val = result.scalar() if val is None: # NULL = the lock subsystem itself errored. Treating it as 0 # ("another run holds it") would silently skip the sweep with # status='ok' and mask a real failure — raise instead so the # handler marks the run failed (handlers-raise-on-failure rule). raise RuntimeError( "relationships sync: GET_LOCK returned NULL (lock subsystem error)" ) return val == 1 async def _release_sync_lock(db: AsyncSession) -> None: """Release the advisory lock; tolerates "not held" results.""" async with best_effort( db, label="sync_service best-effort step", log=logger, ) as be_111_step: await db.execute( text("SELECT RELEASE_LOCK(:name)"), {"name": _REL_SYNC_LOCK_NAME}, ) if not be_111_step.ok: logger.info("relationships sync: RELEASE_LOCK failed (non-fatal)") class _EdgeBatch: """Accumulates edges for one source then flushes via UPSERT.""" def __init__(self, source: str, db: AsyncSession, sync_started_at: datetime): self.source = source self.db = db self.now = sync_started_at self._touched_ids: set[str] = set() self._buffer: list[dict[str, Any]] = [] def emit( self, *, from_type: str, from_id: str, from_label: str | None, to_type: str, to_id: str, to_label: str | None, rel_type: str, attributes: dict[str, Any] | None = None, weight: float = 1.0, ) -> None: self._buffer.append({ "from_type": from_type, "from_id": from_id, "from_label": from_label, "to_type": to_type, "to_id": to_id, "to_label": to_label, "rel_type": rel_type, "attributes": attributes, "weight": weight, }) async def flush(self) -> tuple[int, int]: """UPSERT every buffered edge. Returns (inserted, updated). Uses MySQL/TiDB ``INSERT ... ON DUPLICATE KEY UPDATE`` against the ``uq_rel_edge`` 5-tuple unique index. Single statement per edge, no SELECT-then-update — keeps the row-lock window short and avoids the autoflush-during-Query trap that was tripping the row lock-wait timeout on ``rel_relationships`` when two sync runs (full + every-5-min incremental) overlapped. ``rowcount`` is 1 on fresh insert, 2 on update-on-duplicate (MySQL convention); we count anything != 1 as an update. """ inserted = 0 updated = 0 for edge in self._buffer: stmt = mysql_insert(RelRelationship).values( id=str(uuid.uuid4()), from_type=edge["from_type"], from_id=edge["from_id"], from_label=edge["from_label"], to_type=edge["to_type"], to_id=edge["to_id"], to_label=edge["to_label"], rel_type=edge["rel_type"], attributes=edge["attributes"], source=self.source, weight=edge["weight"], first_seen_at=self.now, last_seen_at=self.now, ).on_duplicate_key_update( from_label=edge["from_label"], to_label=edge["to_label"], attributes=edge["attributes"], weight=edge["weight"], last_seen_at=self.now, # Re-activate any soft-deleted edge that re-appeared # (e.g. ownership transferred away then back). deleted_at=None, # Source can rotate if a different sync has stronger # authority on the same edge — keep the most recent # writer. source=self.source, ) result = await self.db.execute(stmt) if result.rowcount == 1: inserted += 1 else: updated += 1 self._touched_ids.add(self._key(edge)) return inserted, updated async def soft_delete_untouched(self) -> int: """Soft-delete every edge of this source that wasn't touched in the current run (i.e. it disappeared from the source data).""" live = list((await self.db.execute( select(RelRelationship).where( RelRelationship.source == self.source, RelRelationship.deleted_at.is_(None), ) )).scalars().all()) deleted = 0 for row in live: key = self._key({ "from_type": row.from_type, "from_id": row.from_id, "to_type": row.to_type, "to_id": row.to_id, "rel_type": row.rel_type, }) if key not in self._touched_ids: row.deleted_at = self.now deleted += 1 await self.db.flush() return deleted @staticmethod def _key(edge: dict[str, Any]) -> str: return ( f"{edge['from_type']}::{edge['from_id']}|" f"{edge['rel_type']}|" f"{edge['to_type']}::{edge['to_id']}" ) class RelationshipSyncService: """Top-level orchestrator. Each ``_sync_`` method scans one authoritative table and emits edges into a batch; the batch UPSERTs and then soft-deletes anything not touched.""" def __init__(self, db: AsyncSession) -> None: self.db = db async def run_full_sync(self) -> dict[str, Any]: ts_now = datetime.now(UTC) summary: dict[str, Any] = {"started_at": ts_now.isoformat()} total_inserted = 0 total_updated = 0 total_deleted = 0 # Serialize concurrent sync runs (daily full + every-5-min # incremental + admin manual trigger). Without this two runs # can race on the same ``rel_relationships`` rows and one ends # up waiting on the row lock until ``innodb_lock_wait_timeout`` # expires. Using ``GET_LOCK(_, 0)`` so we fail fast if another # run holds it — the next scheduled tick will pick up. if not await _acquire_sync_lock(self.db): logger.info( "relationships.full_sync skipped — another sync run holds the lock", ) summary["skipped"] = "another sync run in progress" summary["completed_at"] = datetime.now(UTC).isoformat() return summary try: for source, fn in ( (SOURCE_IDENTITY, self._sync_identity), (SOURCE_ASSET, self._sync_assets), (SOURCE_SEAT, self._sync_seats), (SOURCE_API_KEY, self._sync_api_keys), (SOURCE_PERMISSION, self._sync_permissions), (SOURCE_EMAIL_GROUP, self._sync_email_groups), (SOURCE_SOFTWARE, self._sync_software), (SOURCE_CMDB, self._sync_cmdb), ): batch = _EdgeBatch(source=source, db=self.db, sync_started_at=ts_now) try: await fn(batch) ins, upd = await batch.flush() deleted = await batch.soft_delete_untouched() # Commit per source so row locks held during this # source's UPSERTs release before the next source # starts — keeps the worst-case lock window down to # one source's edges, not the whole graph. await self.db.commit() except Exception as exc: # noqa: BLE001 logger.exception( "relationships.full_sync source %s failed", source, ) await self.db.rollback() summary[source] = {"error": f"{type(exc).__name__}: {exc}"} continue summary[source] = { "inserted": ins, "updated": upd, "soft_deleted": deleted, } total_inserted += ins total_updated += upd total_deleted += deleted finally: await _release_sync_lock(self.db) summary["totals"] = { "inserted": total_inserted, "updated": total_updated, "soft_deleted": total_deleted, } summary["completed_at"] = datetime.now(UTC).isoformat() # raise-on-source-failure (feedback_handler_status_must_raise). # Per-source try/except keeps one source's failure from # aborting the others (good), but unconditionally returning # the summary left scheduled_job_runs.status='ok' even when # every source failed — failure-notifier stayed silent. failed_sources = [ s for s, v in summary.items() if isinstance(v, dict) and "error" in v ] if failed_sources: raise RuntimeError( f"relationships.full_sync: {len(failed_sources)} source(s) " f"failed: {', '.join(failed_sources)}. See per-source " "'error' in run summary." ) return summary async def run_incremental_sync(self, *, lookback_minutes: int = 60) -> dict[str, Any]: """Lightweight sync: re-emit edges for objects whose ``updated_at`` is within the last ``lookback_minutes``. Does NOT do a soft-delete sweep (the daily full sync handles that) — incremental's job is fast freshness, not consistency.""" ts_now = datetime.now(UTC) cutoff = ts_now - timedelta(minutes=lookback_minutes) summary: dict[str, Any] = { "started_at": ts_now.isoformat(), "cutoff": cutoff.isoformat(), } # Same advisory lock as full_sync — every-5-min incremental can # back up against an in-flight full_sync (or a slow prior # incremental) and produce row lock-wait timeouts on # ``rel_relationships``. Skip the run if locked; next tick # picks up. if not await _acquire_sync_lock(self.db): logger.info( "relationships.incremental_sync skipped — another sync run holds the lock", ) summary["skipped"] = "another sync run in progress" summary["completed_at"] = datetime.now(UTC).isoformat() return summary try: # Per-source commit identical to full_sync — release row # locks before the next source starts. for source, fn in ( (SOURCE_IDENTITY, self._sync_identity), (SOURCE_ASSET, self._sync_assets), (SOURCE_SEAT, self._sync_seats), (SOURCE_API_KEY, self._sync_api_keys), (SOURCE_CMDB, self._sync_cmdb), ): batch = _EdgeBatch(source, self.db, ts_now) try: await fn(batch, since=cutoff) ins, upd = await batch.flush() await self.db.commit() except Exception as exc: # noqa: BLE001 logger.exception( "relationships.incremental_sync source %s failed", source, ) await self.db.rollback() summary[source] = {"error": f"{type(exc).__name__}: {exc}"} continue summary[source] = {"inserted": ins, "updated": upd} finally: await _release_sync_lock(self.db) summary["completed_at"] = datetime.now(UTC).isoformat() # Mirror full_sync — raise if any source erred. failed_sources = [ s for s, v in summary.items() if isinstance(v, dict) and "error" in v ] if failed_sources: raise RuntimeError( f"relationships.incremental_sync: {len(failed_sources)} " f"source(s) failed: {', '.join(failed_sources)}. See " "per-source 'error' in run summary." ) return summary # ── Per-source sync implementations ───────────────────────── async def _load_person_map( self, raw_ids: Any, ) -> dict[str, IdentityPerson]: """Batch-load ``IdentityPerson`` rows into a ``str(id) -> person`` map in ONE query, replacing the per-row ``self.db.get`` that made the nightly full sync issue one round-trip per edge (thousands of serial round-trips over assets / seats / permissions / servers / scopes / domains / grants). Does NOT filter ``deleted_at`` — callers keep their own liveness checks, matching the prior ``db.get`` semantics. ids are cast to ``str`` (a raw ``uuid.UUID`` in an IN clause binds to 0 rows on asyncmy — see ``reference_asyncmy_uuid_bind``). """ str_ids = {str(i) for i in raw_ids if i is not None} if not str_ids: return {} rows = (await self.db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(str_ids)) )).scalars().all() return {str(p.id): p for p in rows} async def _sync_identity( self, batch: _EdgeBatch, *, since: datetime | None = None, ) -> None: """Manager chain + department membership.""" query = select(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), ) if since is not None: query = query.where(IdentityPerson.updated_at >= since) rows = list((await self.db.execute(query)).scalars().all()) # Manager lookups via a batched map instead of one db.get per # person. Full sync loads every person so the map covers all # managers; incremental sync may reference a manager outside the # ``since`` window, so batch-fetch the misses once. person_by_id: dict[str, IdentityPerson] = {str(p.id): p for p in rows} missing_mgr = { str(p.manager_person_id) for p in rows if p.manager_person_id is not None and str(p.manager_person_id) not in person_by_id } if missing_mgr: person_by_id.update(await self._load_person_map(missing_mgr)) for person in rows: if person.manager_person_id is not None: manager = person_by_id.get(str(person.manager_person_id)) if manager is not None and manager.deleted_at is None: batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(manager.id), from_label=manager.display_name or manager.work_email, to_type=NODE_TYPE_PERSON, to_id=str(person.id), to_label=person.display_name or person.work_email, rel_type=REL_MANAGES, ) batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_PERSON, to_id=str(manager.id), to_label=manager.display_name or manager.work_email, rel_type=REL_REPORTS_TO, ) # Department membership — pull from primary org_membership. primary = next( (m for m in person.org_memberships if m.is_primary and m.deleted_at is None), None, ) if primary is not None and primary.org_node_source_id: batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_DEPARTMENT, to_id=primary.org_node_source_id, to_label=primary.org_node_name, rel_type=REL_MEMBER_OF, attributes={"path": primary.org_path}, ) async def _sync_assets( self, batch: _EdgeBatch, *, since: datetime | None = None, ) -> None: query = select(AssetItem).where(AssetItem.assigned_to.isnot(None)) if hasattr(AssetItem, "deleted_at"): query = query.where(AssetItem.deleted_at.is_(None)) if since is not None and hasattr(AssetItem, "updated_at"): query = query.where(AssetItem.updated_at >= since) rows = list((await self.db.execute(query)).scalars().all()) person_map = await self._load_person_map(r.assigned_to for r in rows) for asset in rows: holder = person_map.get(str(asset.assigned_to)) if holder is None: continue # Direct attribute access — a column rename surfaces as # AttributeError instead of silently leaving a UUID-only # label in the relationship graph (memory: getattr-on-ORM # antipattern). asset_label = ( asset.asset_name or asset.asset_tag or str(asset.id) ) batch.emit( from_type=NODE_TYPE_HARDWARE_ASSET, from_id=str(asset.id), from_label=asset_label, to_type=NODE_TYPE_PERSON, to_id=str(holder.id), to_label=holder.display_name or holder.work_email, rel_type=REL_ASSIGNED_TO, ) async def _sync_seats( self, batch: _EdgeBatch, *, since: datetime | None = None, ) -> None: query = select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.user_id.isnot(None), SoftwareSeatAllocation.status == "active", ) if since is not None and hasattr(SoftwareSeatAllocation, "updated_at"): query = query.where(SoftwareSeatAllocation.updated_at >= since) rows = list((await self.db.execute(query)).scalars().all()) person_map = await self._load_person_map(r.user_id for r in rows) for seat in rows: person = person_map.get(str(seat.user_id)) if person is None: continue # SoftwareSeatAllocation has no direct software_id; the # software is reached via plan → SoftwareLicensePlan. # Both relationships are lazy="joined", so this does not # add a round-trip. plan = seat.plan sw = plan.software if plan is not None else None seat_label = f"seat:{seat.id}" batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_SOFTWARE_SEAT, to_id=str(seat.id), to_label=seat_label, rel_type=REL_HOLDS_SEAT, attributes={ "software_id": str(sw.id) if sw is not None else None, "plan_id": str(seat.plan_id), }, ) if sw is not None: batch.emit( from_type=NODE_TYPE_SOFTWARE_SEAT, from_id=str(seat.id), from_label=seat_label, to_type=NODE_TYPE_SOFTWARE, to_id=str(sw.id), to_label=sw.name, rel_type=REL_OWNS, ) async def _sync_api_keys( self, batch: _EdgeBatch, *, since: datetime | None = None, ) -> None: # #1773: only LIVE assignments own an API key. ``ApiKeyAssignment`` # starts ``pending_approval`` and revoke/release updates the row in # place (never deletes it), so without a status filter the nightly # sync emitted (and every night re-touched, defeating the soft-delete # sweep) ``REL_HAS_API_KEY`` edges for pending / rejected / revoked # keys — the governance graph then永久 showed a departed / rejected # holder as holding a key. Mirror ``_sync_seats`` (status=='active') # / cloud grants; the live set for keys is {active, approved} (see # ``apikey.analytics.active_statuses``). query = select(ApiKeyAssignment).where( ApiKeyAssignment.assigned_to.isnot(None), ApiKeyAssignment.status.in_(("active", "approved")), ) if hasattr(ApiKeyAssignment, "deleted_at"): query = query.where(ApiKeyAssignment.deleted_at.is_(None)) if since is not None and hasattr(ApiKeyAssignment, "updated_at"): query = query.where(ApiKeyAssignment.updated_at >= since) rows = list((await self.db.execute(query)).scalars().all()) person_map = await self._load_person_map(r.assigned_to for r in rows) for assignment in rows: person = person_map.get(str(assignment.assigned_to)) if person is None: continue label = ( getattr(assignment, "key_label", None) or getattr(assignment, "purpose", None) or f"api_key:{assignment.id}" ) batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_API_KEY, to_id=str(assignment.id), to_label=label, rel_type=REL_HAS_API_KEY, ) async def _sync_permissions(self, batch: _EdgeBatch) -> None: from app.domains.permission.models import PERM_ACTIVE, SoftwarePermission # #1773: ``UserPermission.status`` spans pending / active / revoked / # rejected / expired / cancelled, and revoke updates the row in place # (never deletes it). Emitting ``REL_OWNS`` for every row painted #离职者 / 被拒申请 as permanently holding the permission (re-touched # nightly, so the soft-delete sweep never reaped the stale edge). # Only ``active`` rows own the permission — mirror ``_sync_seats`` / # cloud grants (status=='active'). rows = list((await self.db.execute( select(UserPermission).where(UserPermission.status == PERM_ACTIVE) )).scalars().all()) if not rows: return # UserPermission has no direct permission_key/software_id; resolve # via permission_id → SoftwarePermission, batched. permission_ids = {p.permission_id for p in rows if p.permission_id} catalog_by_id: dict[Any, SoftwarePermission] = {} if permission_ids: cat_rows = (await self.db.execute( select(SoftwarePermission).where( SoftwarePermission.id.in_(permission_ids), ) )).scalars().all() catalog_by_id = {c.id: c for c in cat_rows} person_map = await self._load_person_map(p.user_id for p in rows) for perm in rows: person = person_map.get(str(perm.user_id)) if person is None: continue cat = catalog_by_id.get(perm.permission_id) label = ( cat.permission_key if cat is not None else f"perm:{perm.id}" ) attributes: dict[str, Any] = {} if cat is not None: attributes["permission_key"] = cat.permission_key attributes["software_id"] = str(cat.software_id) batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_PERMISSION, to_id=str(perm.id), to_label=label, rel_type=REL_OWNS, attributes=attributes or None, ) async def _sync_email_groups(self, batch: _EdgeBatch) -> None: owners = list((await self.db.execute( select(EmailGroupOwner) )).scalars().all()) if not owners: return # FK on EmailGroupOwner is `email_group_id`, not `group_id`. group_ids = {o.email_group_id for o in owners if o.email_group_id} group_by_id: dict[Any, EmailGroup] = {} if group_ids: grp_rows = (await self.db.execute( select(EmailGroup).where(EmailGroup.id.in_(group_ids)) )).scalars().all() group_by_id = {g.id: g for g in grp_rows} person_map = await self._load_person_map(o.person_id for o in owners) for owner in owners: person = person_map.get(str(owner.person_id)) if person is None: continue group = group_by_id.get(owner.email_group_id) if group is None or group.status != "active": continue batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_EMAIL_GROUP, to_id=str(group.id), to_label=group.email, rel_type=REL_OWNS, ) async def _sync_software(self, batch: _EdgeBatch) -> None: rows = list((await self.db.execute( select(AssetSoftwareCatalog) )).scalars().all()) person_map = await self._load_person_map( pid for sw in rows for pid in ( getattr(sw, "owner_person_id", None), getattr(sw, "it_contact_person_id", None), ) ) for sw in rows: if getattr(sw, "owner_person_id", None): person = person_map.get(str(sw.owner_person_id)) if person is not None: batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_SOFTWARE, to_id=str(sw.id), to_label=sw.name, rel_type=REL_RESPONSIBLE_FOR, ) if getattr(sw, "it_contact_person_id", None): person = person_map.get(str(sw.it_contact_person_id)) if person is not None: batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_SOFTWARE, to_id=str(sw.id), to_label=sw.name, rel_type=REL_IT_CONTACT_FOR, ) async def _sync_cmdb( self, batch: _EdgeBatch, *, since: datetime | None = None, ) -> None: from app.integrations.models import Integration # CmdbServer / CmdbCloudScope / CmdbDomain do NOT carry a # ``provider`` column — that string lives on the Integration row # they reference via ``integration_id``. Pre-fetch the entire # integrations table once so each relationship-sync run does not # hammer the DB with one ``Integration.get`` per row. integ_rows = list((await self.db.execute( select(Integration) )).scalars().all()) provider_by_integ: dict[str, str] = { str(i.id): i.provider for i in integ_rows } def _person_uuid(raw: Any) -> uuid.UUID | None: try: return uuid.UUID(str(raw)) except (ValueError, TypeError): return None # Servers srv_query = select(CmdbServer).where( CmdbServer.owner_person_id.isnot(None), CmdbServer.deleted_at.is_(None), ) if since is not None and hasattr(CmdbServer, "updated_at"): srv_query = srv_query.where(CmdbServer.updated_at >= since) srv_rows = list((await self.db.execute(srv_query)).scalars().all()) srv_person_map = await self._load_person_map( _person_uuid(r.owner_person_id) for r in srv_rows ) for srv in srv_rows: pid = _person_uuid(srv.owner_person_id) if pid is None: continue person = srv_person_map.get(str(pid)) if person is None: continue batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_CLOUD_SERVER, to_id=str(srv.id), to_label=srv.name or str(srv.id), rel_type=REL_OWNS, attributes={ "provider": provider_by_integ.get(str(srv.integration_id)), "region": srv.region, }, ) # Cloud scopes scope_query = select(CmdbCloudScope).where( CmdbCloudScope.owner_person_id.isnot(None), ) if hasattr(CmdbCloudScope, "deleted_at"): scope_query = scope_query.where(CmdbCloudScope.deleted_at.is_(None)) scope_rows = list((await self.db.execute(scope_query)).scalars().all()) scope_person_map = await self._load_person_map( _person_uuid(r.owner_person_id) for r in scope_rows ) for scope in scope_rows: pid = _person_uuid(scope.owner_person_id) if pid is None: continue person = scope_person_map.get(str(pid)) if person is None: continue batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_CLOUD_SCOPE, to_id=str(scope.id), to_label=scope.display_name or str(scope.id), rel_type=REL_OWNS, attributes={ "provider": provider_by_integ.get(str(scope.integration_id)), "scope_type": scope.scope_type, }, ) # Domains dom_query = select(CmdbDomain).where( CmdbDomain.owner_person_id.isnot(None), CmdbDomain.deleted_at.is_(None), ) dom_rows = list((await self.db.execute(dom_query)).scalars().all()) dom_person_map = await self._load_person_map( _person_uuid(r.owner_person_id) for r in dom_rows ) for dom in dom_rows: pid = _person_uuid(dom.owner_person_id) if pid is None: continue person = dom_person_map.get(str(pid)) if person is None: continue batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_DOMAIN, to_id=str(dom.id), to_label=dom.domain_name or str(dom.id), rel_type=REL_OWNS, attributes={ "provider": provider_by_integ.get(str(dom.integration_id)), }, ) # Cloud access grants — emits one edge per (person, scope, role). # ``scope_id`` is a non-nullable FK on CmdbCloudAccessGrant, so a # missing scope is a data-integrity issue (skip rather than emit # a self-loop). grant_query = select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.status == "active", ) grant_rows = list((await self.db.execute(grant_query)).scalars().all()) grant_person_map = await self._load_person_map( _person_uuid(g.person_id) for g in grant_rows ) # Batch the scope lookups too (was one db.get per grant). grant_scope_map: dict[str, CmdbCloudScope] = {} scope_ids = {str(g.scope_id) for g in grant_rows if g.scope_id} if scope_ids: grant_scope_map = { str(s.id): s for s in (await self.db.execute( select(CmdbCloudScope).where(CmdbCloudScope.id.in_(scope_ids)) )).scalars().all() } for grant in grant_rows: pid = _person_uuid(grant.person_id) if pid is None: continue person = grant_person_map.get(str(pid)) if person is None: continue scope = grant_scope_map.get(str(grant.scope_id)) if scope is None: logger.warning( "relationships sync: grant %s references missing " "scope %s — skipping edge", grant.id, grant.scope_id, ) continue batch.emit( from_type=NODE_TYPE_PERSON, from_id=str(person.id), from_label=person.display_name or person.work_email, to_type=NODE_TYPE_CLOUD_SCOPE, to_id=str(grant.scope_id), to_label=scope.display_name or str(scope.id), rel_type=REL_GRANTS_ACCESS_TO, attributes={ "role_key": grant.role_key, "expires_at": ( grant.expires_at.isoformat() if grant.expires_at else None ), "provider": provider_by_integ.get(str(scope.integration_id)), }, ) __all__ = ["RelationshipSyncService"]