"""Scheduled-job handler for Zoom seat sync. ``zoom.seat_sync`` pulls the current state of the Zoom account once a day and reconciles it with Helix's ``software_seat_allocations``: * **Base Licensed plan** (Enterprise Essentials) — every Zoom user with ``type=2`` (Licensed) is mirrored into the ``evict_lru`` plan on the Zoom software. New users → new seat row; Zoom-side disappearances → ``status=revoked``. * **Add-on plans** (Large Meeting, Webinars Plus) — for each licensed user we fetch ``/users/{id}/settings`` and inspect the ``feature`` object. Users with ``feature.large_meeting=true`` are mirrored into the Large Meeting plan; ``feature.webinar=true`` → Webinars Plus. Requires the ``user:read:settings:admin`` scope on the S2S OAuth app; if missing, add-on sync is silently skipped. * **Seat cap auto-tune** — ``/accounts/me/plans`` and ``/accounts/me/plans/usage`` feed ``SoftwareLicensePlan.max_seats`` for each plan so the UI reflects the real purchased quota. * **Alias-aware matching** — external users whose primary email is a Google Workspace alias (e.g. ``reece.r@pingcap.com`` aliased to ``reece.reynolds@pingcap.com``) are resolved via ``identity_persons.email_aliases``, populated by the ``email_group`` domain's Google-group mirror sync (the legacy ``google.dept_sync`` was retired in PR #identity-sync-center when Google dept-field projection moved into the Identity Sync Center; alias harvesting was retained on the email_group path). Before that field existed the sync silently skipped ~5 % of licensed Zoom users as "external". Idempotent: re-running yields zero diffs in steady state. """ from __future__ import annotations import asyncio import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import format_exception_for_log from app.core.txn import best_effort from app.scheduled_jobs._seat_sync_helpers import ( ExternalHolderObservation, import_dedup_filter, is_safe_max_seats_change, raise_if_total_failure, record_seat_reconciliation, safe_to_drift_revoke, write_drift_event_audit, write_sync_audit, ) from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) # Zoom user.type enum (kept in sync with ZoomSoftwareDriver._TYPE_BY_PLAN) _ZOOM_TYPE_LICENSED = 2 # Per-user settings fetch concurrency. Zoom's Light-tier limit is # ~80rps; 20 concurrent async calls sits well under that and keeps # total latency under ~10s for 200 users even on a slow link. _SETTINGS_CONCURRENCY = 20 @register_scheduled_job( key="zoom.seat_sync", display_name="Zoom · 坐席与订阅同步", description=( "每天拉取 Zoom 全部活跃用户 + 订阅计划,把 Licensed 坐席镜像到 " "software_seat_allocations;同时读每个用户的 feature 设置," "把 Large Meeting / Webinars Plus 等 add-on 坐席镜像到对应 plan。" "Zoom 上已消失的坐席标为 revoked;max_seats 按订阅数自动回写。" "通过 identity_persons.email_aliases 支持 Google Workspace 别名匹配。" ), ) async def run_zoom_seat_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # Deadlock-victim restart on fresh sessions — same incident class # as jira.seat_sync 2026-06-12 (worker restart double-run racing # INSERTs). The reconcile is idempotent: import has the 25h-grace # dedup filter, revoke has safe_to_drift_revoke. from app.scheduled_jobs.handlers import run_with_lock_retry return await run_with_lock_retry( db, lambda session: _run_zoom_seat_sync_once(session, config), label="zoom.seat_sync", ) async def _run_zoom_seat_sync_once( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.coordinators.zoom_seat_coordinator import ( get_zoom_plans, load_zoom_driver, ) from app.domains.asset.models import ( SoftwareExternalHolder, SoftwareSeatSyncStat, ) from app.identity.email_lookup import build_person_by_email_index from app.integrations.drivers.zoom_software import ZoomSoftwareDriver empty_result: dict[str, Any] = { "zoom_users_listed": 0, "licensed_users_on_zoom": 0, "base": {"imported": 0, "revoked": 0, "active": 0}, "large_meeting": {"imported": 0, "revoked": 0, "active": 0}, "webinar": {"imported": 0, "revoked": 0, "active": 0}, "max_seats_updated": [], "external_licensed_skipped": [], "errors": [], } driver = await load_zoom_driver(db) if driver is None: empty_result["errors"].append( "zoom integration not configured or not enabled" ) return empty_result plans = await get_zoom_plans(db) base_plan = plans.get("base") large_plan = plans.get("large_meeting") webinar_plan = plans.get("webinar") if base_plan is None: empty_result["errors"].append( "Zoom base Licensed plan (over_quota_policy='evict_lru') not " "found; run migration 051 or create a plan manually." ) return empty_result errors: list[str] = [] # 1. List Zoom users — include both active and pending. A user # invited but who hasn't yet accepted (Zoom status="pending") is # already billed at the Licensed seat price; if we only fetched # ``active``, their email would be missing from # ``licensed_by_email`` and the daily drift sweep would revoke # their seat as "no longer on Licensed on Zoom" — until they # accept the invite, when the next sync re-imports them. # (Round-17 audit.) try: zoom_users = await driver.list_users(status="active") except Exception as e: # noqa: BLE001 # Primary list_users(active) is the source of truth for daily # seat reconcile. Failing it silently lands ``status='ok'`` and # nobody gets paged — meanwhile import/revoke logic runs against # empty data. Must raise so failure_notifier fires (matches the # invariant in feedback_handler_status_must_raise memory). raise RuntimeError( f"zoom.list_users(active) failed: {e}" ) from e try: pending_users = await driver.list_users(status="pending") except Exception as e: # noqa: BLE001 # Pending users are already billed as Licensed seats. If we # continue with pending_users=[] their emails are absent from # licensed_by_email, so the drift-revoke loop removes their # seats the same tick — a silent data-loss bug. Raise so the # failure_notifier fires and the run is retried clean. raise RuntimeError( f"zoom.list_users(pending) failed: {e}" ) from e if pending_users: zoom_users = list(zoom_users) + list(pending_users) # Filter to Licensed (type=2) with usable email keys. licensed_by_email: dict[str, dict[str, Any]] = {} for u in zoom_users: if int(u.get("type") or 0) != _ZOOM_TYPE_LICENSED: continue email = str(u.get("email") or "").strip().lower() if email: licensed_by_email[email] = u # 2. Alias-aware lookup — resolves both primary work_email and # any Google Workspace alias mirrored by google.dept_sync. # Leavers (``left`` / ``leaver_pending``) are excluded so the # daily reconcile doesn't re-import an active seat for a person # who is already gone on the Helix side but still lingering on # the vendor side (the offboarding flow may not have finished # the vendor-side revoke yet). Their existing seats fall through # to drift cleanup naturally — they'll appear in ``existing`` # but not in ``want_user_ids``, so they get revoked. # Notice-period exception (2026-08-11): a ``leaver_pending`` person # whose last working day has not passed is still working and stays # in the index — see ``leaver_notice_grace`` in the helper. Without # it their 3 Zoom seat rows were revoked the night the termination # was confirmed and re-imported as new rows days later, which # defeated offboarding's per-seat dedup (duplicate cancel events). person_by_email = await build_person_by_email_index( db, licensed_by_email.keys(), exclude_employment_statuses=("left", "leaver_pending"), ) # 3. max_seats auto-tune from Zoom subscription. max_seats_updated: list[str] = [] try: plan_payload = await driver.get_account_plan() except Exception as e: # noqa: BLE001 errors.append( f"get_account_plan failed: {format_exception_for_log(e)}" ) plan_payload = None base_hosts = ZoomSoftwareDriver.extract_licensed_seats(plan_payload) if base_hosts is not None and base_hosts != base_plan.max_seats: ok, reason = is_safe_max_seats_change( current=base_plan.max_seats, new=base_hosts, ) if ok: logger.info( "zoom.seat_sync: updating base plan %s max_seats %s → %s", base_plan.id, base_plan.max_seats, base_hosts, ) previous = base_plan.max_seats base_plan.max_seats = base_hosts max_seats_updated.append(f"base={base_hosts}") # Per-change audit row — the plan's max_seats drives # eviction and quota decisions, so a silent change is # exactly the kind of "why did everyone get evicted on # April 17" question the run-level summary won't answer # cleanly six months later. async with best_effort( db, label="zoom_handlers best-effort step", log=logger, ) as be_227_step: from app.audit.service import AuditService await AuditService(db).log( trace_id=f"zoom.seat_sync:max_seats:{base_plan.id}", actor_type="system", actor_user_id=None, action="software_plan.max_seats_auto_tuned", status="success", resource_type="software_license_plan", resource_id=str(base_plan.id), detail={ "vendor": "zoom", "scope": "base", "previous": previous, "new": base_hosts, }, ) if not be_227_step.ok: logger.error("max_seats audit failed") else: errors.append(f"base plan: {reason}") # Add-on max_seats from /accounts/me/plans/usage. try: usage_payload = await driver.get_account_plans_usage() except Exception as e: # noqa: BLE001 errors.append( f"get_account_plans_usage failed: {format_exception_for_log(e)}" ) usage_payload = None if usage_payload: await _apply_addon_max_seats( db, usage_payload, large_plan, webinar_plan, max_seats_updated, logger, errors, ) # 4. Per-user /users/{id}/settings — feeds add-on membership. # Only needed if at least one add-on plan is configured. user_settings: dict[str, dict[str, Any]] = {} if large_plan is not None or webinar_plan is not None: try: user_settings = await _fetch_user_settings( driver, licensed_by_email, ) except Exception as e: # noqa: BLE001 errors.append( f"fetch_user_settings failed: {format_exception_for_log(e)}" ) now = datetime.now(UTC) # 5. Reconcile each plan independently. base_stats = await _reconcile_plan( db, base_plan, licensed_by_email, person_by_email, lambda email: True, # every licensed user belongs to base now, "auto-imported from Zoom (licensed)", errors, ) if large_plan is not None: large_stats = await _reconcile_plan( db, large_plan, licensed_by_email, person_by_email, lambda email: _has_feature(user_settings, email, "large_meeting"), now, "auto-imported from Zoom (feature.large_meeting)", errors, ) else: large_stats = {"imported": 0, "revoked": 0, "active": 0} if webinar_plan is not None: webinar_stats = await _reconcile_plan( db, webinar_plan, licensed_by_email, person_by_email, lambda email: _has_feature(user_settings, email, "webinar"), now, "auto-imported from Zoom (feature.webinar)", errors, ) else: webinar_stats = {"imported": 0, "revoked": 0, "active": 0} # Licensed Zoom users we couldn't resolve to any Helix person — # shared mailboxes, contractors, or HRIS-absent accounts. Surface # a short preview so the admin can clean them up. external_skipped = sorted( e for e in licensed_by_email if e not in person_by_email ) # Persist the full external set + base-plan reconciliation snapshot. # External Licensed accounts belong to the base (Licensed) plan — the # add-on tiers are feature-gated subsets, not a separate billing # roster. ``vendor_license_holders`` counts with-email Licensed # accounts, so matched + external reconciles exactly (no_email=0). external_observations = [ ExternalHolderObservation( email=email, vendor_account_id=(licensed_by_email[email].get("id")), account_status=(licensed_by_email[email].get("status")), display_name=( licensed_by_email[email].get("display_name") or " ".join( p for p in ( licensed_by_email[email].get("first_name"), licensed_by_email[email].get("last_name"), ) if p ) or None ), ) for email in external_skipped ] await record_seat_reconciliation( db, holder_model=SoftwareExternalHolder, stat_model=SoftwareSeatSyncStat, integration_id=uuid.UUID(driver.integration_id), software_id=base_plan.software_id, plan_id=base_plan.id, provider="zoom", product_key="zoom-licensed", site=None, vendor_license_holders=len(licensed_by_email), holders_matched=len(licensed_by_email) - len(external_skipped), holders_no_email=0, externals=external_observations, now=now, ) # Synthetic flat progress counter used by ``raise_if_total_failure`` # — Zoom's per-plan stats are nested, so we sum imported+revoked # across all three plans for the all-failed gate. # # Round-42 audit: only mutation counters count as progress. # ``zoom_users_listed`` (a fetch-counter) is intentionally NOT in the # progress set: ``raise_if_total_failure`` only fires when # ``summary['errors']`` is non-empty, so a true steady-state run (no # work needed, no errors) returns early and never reaches the gate. # The bug class was: errors AND no real work AND fetch=non-zero kept # the run green even though every per-plan operation failed. seat_changes_total = ( int(base_stats.get("imported") or 0) + int(base_stats.get("revoked") or 0) + int(large_stats.get("imported") or 0) + int(large_stats.get("revoked") or 0) + int(webinar_stats.get("imported") or 0) + int(webinar_stats.get("revoked") or 0) ) summary = { "zoom_users_listed": len(zoom_users), "licensed_users_on_zoom": len(licensed_by_email), "base": base_stats, "large_meeting": large_stats, "webinar": webinar_stats, "max_seats_updated": max_seats_updated, "external_licensed_skipped": external_skipped[:30], "external_licensed_total": len(external_skipped), "seat_changes_total": seat_changes_total, "errors": errors, } await write_sync_audit(db, job_key="zoom.seat_sync", summary=summary) await db.commit() raise_if_total_failure( job_key="zoom.seat_sync", summary=summary, progress_keys=("seat_changes_total",), ) return summary async def _apply_addon_max_seats( db, usage_payload: dict[str, Any], large_plan, webinar_plan, updates: list[str], log: logging.Logger, errors: list[str], ) -> None: """Pick the ``hosts`` count for Large Meeting / Webinar entries and push it into the matching plan's ``max_seats``. Zoom returns each add-on as a list of sub-SKUs (e.g. large500 and large1000 separately). We sum the ``hosts`` across sub-SKUs of the same family — the Helix plan row tracks the family as a whole. Drops larger than ``MAX_SEATS_DROP_FLOOR`` are refused and surfaced as an error rather than silently shrinking the plan. Round-49: emits ``software_plan.max_seats_auto_tuned`` audit rows on every mutation, mirroring the base-plan branch in the caller — add-on capacity (webinar quota, large-meeting host count) is the same class of eviction-driving decision and deserves a per-change forensic trail. """ def _sum_hosts(entries: Any) -> int | None: if not isinstance(entries, list): return None total = 0 found = False for e in entries: if isinstance(e, dict) and isinstance(e.get("hosts"), int): total += e["hosts"] found = True return total if found else None async def _apply(label: str, plan, hosts: int | None) -> None: if hosts is None or hosts == plan.max_seats: return ok, reason = is_safe_max_seats_change( current=plan.max_seats, new=hosts, ) if not ok: errors.append(f"{label}: {reason}") return log.info( "zoom.seat_sync: %s max_seats %s → %s", label, plan.max_seats, hosts, ) previous = plan.max_seats plan.max_seats = hosts updates.append(f"{label}={hosts}") async with best_effort( db, label="zoom_handlers best-effort step", log=logger, ) as be_445_step: from app.audit.service import AuditService # noqa: PLC0415 await AuditService(db).log( trace_id=f"zoom.seat_sync:max_seats:{plan.id}", actor_type="system", actor_user_id=None, action="software_plan.max_seats_auto_tuned", status="success", resource_type="software_license_plan", resource_id=str(plan.id), detail={ "vendor": "zoom", "scope": label, "previous": previous, "new": hosts, }, ) if not be_445_step.ok: log.error( "zoom.seat_sync: add-on max_seats audit failed " "(scope=%s plan=%s): %s", label, plan.id, be_445_step.reason(), ) if large_plan is not None: await _apply( "large_meeting", large_plan, _sum_hosts(usage_payload.get("plan_large_meeting")), ) if webinar_plan is not None: await _apply( "webinar", webinar_plan, _sum_hosts(usage_payload.get("plan_webinar")), ) async def _fetch_user_settings( driver, licensed_by_email: dict[str, dict[str, Any]], ) -> dict[str, dict[str, Any]]: """Parallel fetch of /users/{id}/settings for every licensed user. Keyed by the lowercased email so the caller doesn't have to hold onto Zoom user IDs. ``None`` results (missing scope, 404) are skipped — the email just won't appear in the output dict, which the feature predicate treats as "user does not have this feature". """ sem = asyncio.Semaphore(_SETTINGS_CONCURRENCY) async def one(email: str, uid: str) -> tuple[str, dict[str, Any] | None]: async with sem: try: s = await driver.get_user_settings(uid) except Exception as e: # noqa: BLE001 logger.warning( "zoom.seat_sync: user_settings(%s) failed: %s", email, e, ) s = None return email, s tasks = [ one(email, u.get("id") or email) for email, u in licensed_by_email.items() ] results = await asyncio.gather(*tasks) return {email: settings for email, settings in results if settings is not None} def _has_feature( user_settings: dict[str, dict[str, Any]], email: str, feature: str, ) -> bool: s = user_settings.get(email) if not s: return False return bool((s.get("feature") or {}).get(feature)) async def _reconcile_plan( db: AsyncSession, plan, licensed_by_email: dict[str, dict[str, Any]], person_by_email: dict, predicate, now: datetime, reason_on_create: str, errors: list[str], ) -> dict[str, int]: """Mirror Zoom membership for one plan into ``software_seat_allocations``. ``predicate(email) -> bool`` decides whether the licensed user belongs to this plan (always True for the base plan; feature-flag check for add-on plans). """ from app.domains.asset.models import SoftwareSeatAllocation # Seats currently in the (plan, user) unique-index scope — # active OR pending. The dedup map below uses the wider scope so # a stuck-pending row from a parallel approval flow doesn't make # the INSERT below fail with IntegrityError under migration 119's # ``uq_software_seat_active_pending`` partial unique index. # Drift cleanup iterates only active rows separately. Round-3 # logic-scan P1. # # Round 12 hardening 2026-05-08: ``import_dedup_filter`` extends # the scope to include rows revoked recently. Without it, sync # races admin manual-revoke / leaver_blocked flips: vendor side # still has the user → import branch creates a duplicate active # row → admin sees "I revoked it, why is it back?" Window is # ``SEAT_SYNC_REVOKE_GRACE_HOURS`` (25h as of 2026-05-28, bumped # from 2h to cover the cross-day sync schedule — the # shengbo.ma / sunxiaoguang case). existing_all = list((await db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.plan_id == plan.id, import_dedup_filter(SoftwareSeatAllocation, now=now), ) )).scalars().all()) existing_by_user_id = {s.user_id: s for s in existing_all} existing = [s for s in existing_all if s.status == "active"] # Zoom-side set of Helix persons who should be on this plan want_user_ids: set = set() for email, zuser in licensed_by_email.items(): if not predicate(email): continue person = person_by_email.get(email) if person is None: continue # unmatched — counted separately by the handler want_user_ids.add(person.id) imported = 0 for email, zuser in licensed_by_email.items(): if not predicate(email): continue person = person_by_email.get(email) if person is None or person.id in existing_by_user_id: continue try: # Per-row SAVEPOINT — one bad INSERT (FK / unique-index # collision) rolls back just this row, not the session. async with db.begin_nested(): db.add(SoftwareSeatAllocation( plan_id=plan.id, user_id=person.id, status="active", allocated_at=now, reason=( f"{reason_on_create} " f"(last_login={zuser.get('last_login_time') or 'never'})" ), )) await db.flush() imported += 1 except OperationalError: # Deadlock (1213) killed the whole transaction — propagate # so the fresh-session lock-retry wrapper restarts the run. raise except Exception as e: # noqa: BLE001 errors.append( f"{plan.name}:{email}: import failed: " f"{format_exception_for_log(e)}" ) if len(errors) > 30: errors.append("...truncated") break revoked = 0 # SAFETY GUARD via shared helper — three independent nets: # # 1. ``matched_count`` (vendor health) — ``len(licensed_by_email)`` # is the BASE roster Zoom returned. A 0 here means Zoom returned # nothing (genuine outage / credentials revoked). Same shape as # slack/jira/google handlers. # # 2. Ratio guard — matched / existing must be ≥ 50% so a paginator # truncation can't sneak past. # # 3. ``want_count`` (predicate health, round 7 hardening # 2026-05-05) — for add-on plans whose predicate is a # feature-flag check sourced from ``/users/{id}/settings``, # a silent 401 / scope-downgrade on that endpoint produces # an empty want set while the BASE roster still looks healthy. # Without this third net, every paid Large Meeting / Webinar # seat would flip to revoked on a transient scope-issue. # Floor is the hard-coded ``_DEFAULT_EMPTY_WANT_FLOOR`` (5) # inside the helper — predicate emptying the want set while # >5 active seats exist refuses cleanup. safe, reason = safe_to_drift_revoke( matched_count=len(licensed_by_email), existing_count=len(existing), label=f"{plan.name}", want_count=len(want_user_ids), ) if not safe: errors.append(reason or "drift cleanup refused") else: for seat in existing: if seat.user_id in want_user_ids: continue # Fresh seats (<2h old) might be in-flight grants the sync # is racing — leave them alone. allocated = seat.allocated_at if allocated is not None: if allocated.tzinfo is None: allocated = allocated.replace(tzinfo=UTC) if (now - allocated).total_seconds() < 7200: continue # rollback expires modified instances; capture first seat_id, seat_user_id, seat_plan_id = ( seat.id, seat.user_id, seat.plan_id, ) try: # Per-seat SAVEPOINT — one bad flush must not poison # the session and take the rest of the run with it. async with db.begin_nested(): seat.status = "revoked" seat.revoked_at = now seat.reason = ( (seat.reason or "") + "\n[auto-revoked by zoom.seat_sync: " f"user no longer on {plan.name} on Zoom]" ).strip() await write_drift_event_audit( db, job_key="zoom.seat_sync", seat_id=seat_id, user_id=seat_user_id, plan_id=seat_plan_id, event="revoked", detail={ "reason": ( f"user no longer on {plan.name} on Zoom" ), }, ) await db.flush() revoked += 1 except OperationalError: # Whole-transaction rollback (deadlock) — see import loop. raise except Exception as e: # noqa: BLE001 errors.append( f"{plan.name}:seat {seat_id}: drift revoke failed: " f"{format_exception_for_log(e)}" ) active_after = len(existing) - revoked + imported return {"imported": imported, "revoked": revoked, "active": active_after}