"""EmailGroupAdminService — the write half of the 邮件组管理 desk. Static (hand-maintained) mail groups previously had exactly one admin write path: delete. Everything else — adding a member, granting or revoking ownership, fixing a group's name, managing its aliases — was either owner-only through the all-employee directory, or simply not possible in Helix at all. HR ran those in the Google admin console, which means they never appeared in Helix's audit trail: the operation log could show that a group was deleted but not that its entire membership was rebuilt the week before. Design rules this service follows, all of them consequences of Google being the authority for membership: * **Google first, then the local mirror.** Every method calls the driver and only then touches ``email_groups`` / ``email_group_owners``. A Google failure raises before any local write, so Helix never claims a change Google refused. The next ``google.group_sync`` tick is the backstop that repairs any drift. * **Live role reads, never the mirror.** ``email_group_owners`` only tracks owners whose address resolves to a known Helix person, so an external or departed owner is invisible in it. Every ownership guard therefore reads roles live from ``members.list``. Trusting the mirror is how "remove the last owner" slips through. * **Idempotent by construction.** Adding an existing member, removing an absent one, and re-granting an existing owner all succeed and report what was already true. Retries and double-clicks are safe. * **Dynamic groups are refused.** Their membership is rule-computed; a manual edit is silently reverted by Google, so accepting one would make Helix report a change that quietly un-happens. ``kind='unknown'`` is refused too — an undetermined group might be dynamic. Every method is gated by ``can_manage_email_groups`` (hr + global_admin), audits with a reason, and notifies the affected person where a person is affected. """ from __future__ import annotations import logging from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from uuid import UUID from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.audit.service import AuditService from app.audit.service import _esc as _like_esc from app.core.exceptions import AppException, NotFoundError from app.core.rbac import can_manage_email_groups from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.domains.email_group.service import EmailGroupService from app.identity.person_models import IdentityPerson from app.notifications.service import NotificationEnvelope, NotificationService logger = logging.getLogger(__name__) # Google group member roles, most privileged last. ROLE_MEMBER = "MEMBER" ROLE_MANAGER = "MANAGER" ROLE_OWNER = "OWNER" VALID_ROLES = (ROLE_MEMBER, ROLE_MANAGER, ROLE_OWNER) # Cap on one batch-add call. Each entry is a separate Google # ``members.insert``, so the bound is about keeping a single request # inside the platform's HTTP timeout rather than about Google quota. MAX_BATCH_MEMBERS = 50 # Mirrors ``EmailGroup.name`` (String(255)) and the router schema's # description bound. Enforced in the service because the MCP surface has # no Pydantic layer — see ``update_display``. _NAME_MAX = 255 _DESCRIPTION_MAX = 2000 @dataclass class MemberOpResult: """Outcome of one member/owner mutation, per address.""" email: str ok: bool #: ``added`` | ``already_member`` | ``removed`` | ``already_absent`` #: | ``role_changed`` | ``failed`` outcome: str role: str | None = None error: str | None = None def to_dict(self) -> dict[str, Any]: return { "email": self.email, "ok": self.ok, "outcome": self.outcome, "role": self.role, "error": self.error, } class EmailGroupAdminService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self.notify = NotificationService(db) self._eg = EmailGroupService(db) async def _get_driver(self): return await self._eg._get_driver() # ── gate + loading ──────────────────────────────────────────── def _require_manage(self, actor_roles: list[str]) -> None: if not can_manage_email_groups(actor_roles): raise AppException( code="NOT_AUTHORIZED", message="只有 HR / IT Admin 才能管理邮件组成员与负责人。", status_code=403, ) async def _load_static_group( self, group_id: UUID, *, actor_roles: list[str], ) -> EmailGroup: """Load an active group and refuse the rule-driven ones. ``is_dynamic is True`` → rule-computed, a manual edit is reverted by Google. ``is_dynamic is None`` → Cloud Identity detection never ran determinately, so the group *might* be dynamic; refusing is the only answer that cannot silently no-op. Both refusals name the reason so the operator knows which of the two they hit. """ self._require_manage(actor_roles) group = (await self.db.execute( select(EmailGroup) .where(EmailGroup.id == group_id) .options(selectinload(EmailGroup.owners)) )).scalar_one_or_none() if group is None or group.status != "active": raise NotFoundError("EmailGroup", str(group_id)) if group.is_dynamic is True: raise AppException( code="DYNAMIC_GROUP_CANNOT_EDIT", message=( f"邮件组 {group.email} 是动态组,成员由规则自动计算," "手工增删会被 Google 覆盖。请改规则。" ), status_code=409, ) if group.is_dynamic is None: raise AppException( code="GROUP_KIND_UNKNOWN", message=( f"邮件组 {group.email} 的静态/动态类型尚未判定" "(Cloud Identity 检测未成功运行),无法确认手工改成员是否" "会被规则覆盖。请先运行一次邮件组同步。" ), status_code=409, ) return group async def _live_member_entries(self, group_email: str) -> list[dict]: """The RAW ``members.list`` result — every entry, email or not. Google returns members of ``type: "CUSTOMER"`` (whole-domain membership) with an ``id`` and NO ``email``. Any count derived from an email-keyed view silently drops those, and ``direct_members_count`` feeds the cleanup pool's predicate — so an email-keyed count can write a **false zero** into the input of an irreversible hard delete. Role lookups still go through :meth:`_live_roles` (they are keyed BY address, correctly); this is for anything that counts. """ from app.integrations.drivers.google_workspace import GoogleApiError driver = await self._get_driver() try: return list(await driver.list_group_members(group_email)) except GoogleApiError as exc: raise AppException( code="GOOGLE_LIST_MEMBERS_FAILED", message=( f"无法从 Google 读取邮件组 {group_email} 的当前成员," f"操作已中止(不在未确认状态下改成员):{exc}" ), status_code=502, ) from exc async def _live_roles(self, group_email: str) -> dict[str, str]: """``{lowercased email: role}`` read live from Google. NOT from ``email_group_owners``: that mirror only holds owners who resolve to a known Helix person, so an external or departed owner is absent from it. Ownership guards that trust the mirror are how the last owner gets removed. """ # ``list_group_members`` is fail-stop by design and builds an # actionable hint ("SA likely missing # admin.directory.group.member.readonly"); ``_live_member_entries`` # converts that into a 502 that keeps the hint. raw = await self._live_member_entries(group_email) return { (m.get("email") or "").strip().lower(): (m.get("role") or ROLE_MEMBER) for m in raw if m.get("email") } @staticmethod def _norm(email: str) -> str: return (email or "").strip().lower() def _validate_role(self, role: str) -> str: upper = (role or "").strip().upper() if upper not in VALID_ROLES: raise AppException( code="INVALID_MEMBER_ROLE", message=f"role 必须是 {', '.join(VALID_ROLES)} 之一。", status_code=400, ) return upper # ── owner mirror maintenance ───────────────────────────────── # # ``google.group_sync`` is the authoritative writer of # ``email_group_owners``, but it runs every 6 hours. Without a local # update here, an owner granted through this service would not appear # as an owner in Helix — and would not block their own offboarding — # until the next tick. So we mirror immediately AND let the sync # correct us later; the two agree because both derive from Google. async def _mirror_owner_added( self, group: EmailGroup, person: IdentityPerson | None, ) -> None: if person is None: return try: async with self.db.begin_nested(): self.db.add(EmailGroupOwner( email_group_id=group.id, person_id=person.id, )) except IntegrityError: # Unique (group, person) — already mirrored. Nothing to do. pass async def _mirror_has_owner( self, group: EmailGroup, person: IdentityPerson | None, ) -> bool: """Does ``email_group_owners`` already hold this pair? Read BEFORE a repair so the audit row is written only when the repair actually inserted something — an unconditional row would put an ``owner_mirror_repaired`` entry in the operation log every time somebody re-clicks 设为负责人 on an owner, which is how a new action becomes noise the next reader filters out. """ if person is None: return False found = (await self.db.execute( select(EmailGroupOwner.id).where( EmailGroupOwner.email_group_id == group.id, EmailGroupOwner.person_id == person.id, ) )).scalar_one_or_none() return found is not None async def _mirror_owner_removed( self, group: EmailGroup, person: IdentityPerson | None, ) -> None: """Drop the local owner link, ORM-style. A Core ``DELETE`` would bypass the identity map and leave the already-``selectinload``-ed ``group.owners`` collection holding the removed row — a trap armed for the next reader of that collection in the same request. ``session.delete`` is what the domain's own ``transfer_group_owner`` uses. """ if person is None: return row = (await self.db.execute( select(EmailGroupOwner).where( EmailGroupOwner.email_group_id == group.id, EmailGroupOwner.person_id == person.id, ) )).scalars().first() if row is not None: await self.db.delete(row) async def _person_by_email(self, email: str) -> IdentityPerson | None: return (await self.db.execute( select(IdentityPerson).where( func.lower(IdentityPerson.work_email) == self._norm(email), IdentityPerson.deleted_at.is_(None), ) )).scalars().first() async def _refresh_member_count(self, group: EmailGroup) -> None: """Re-read the live member count into the local mirror. ``google.group_sync`` owns this column, but it runs every 6 hours — and the cleanup pool's entire predicate is ``direct_members_count == 0``. Without this, a group HR just populated keeps advertising 0 members and is served to the cleanup workbench as ``risk: "low"``; the executor's live re-verify then refuses it, which is safe but destroys the operator's trust in the signals the workbench exists to provide. Conversely a group whose last member was just removed stays out of the pool for up to 6 hours. Best-effort: the membership change already succeeded on Google, so a failure to refresh the mirror must not fail the request — the next sync tick corrects it either way. """ try: entries = await self._live_member_entries(group.email) except Exception: logger.warning( "member-count refresh failed for %s (non-fatal; next " "group_sync tick corrects it)", group.email, ) return # RAW entry count, not the email-keyed role map: an entry without # an address is still a member, and under-counting here writes a # false zero into the cleanup pool's predicate. group.direct_members_count = len(entries) # Deliberately NOT touching ``last_synced_at``. That column means # "google.group_sync reconciled this row", and the UI renders it # as 最后同步 — stamping it here would claim a full reconcile # (name, description, aliases, dynamic flag, owner set) on the # strength of one members.list call. # ── members ────────────────────────────────────────────────── async def add_members( self, *, group_id: UUID, emails: list[str], role: str, actor: IdentityPerson, actor_roles: list[str], reason: str | None, trace_id: str, ) -> dict[str, Any]: """Add one or more addresses to a static group. Per-address isolation: one rejected address (typo, external domain Google refuses, quota blip) must not discard the ones that worked — the caller gets a row per address and can retry just the failures. A single all-or-nothing call would make a 50-name paste unusable. External addresses are allowed. Google groups routinely carry vendor / contractor / alias addresses, and refusing anything that does not resolve to an ``identity_persons`` row would make this tool unable to reproduce the groups it is meant to manage. """ group = await self._load_static_group(group_id, actor_roles=actor_roles) target_role = self._validate_role(role) cleaned: list[str] = [] seen: set[str] = set() for raw in emails: e = self._norm(raw) if not e or e in seen: continue if "@" not in e: raise AppException( code="INVALID_EMAIL", message=f"{raw!r} 不是一个邮箱地址。", status_code=400, ) seen.add(e) cleaned.append(e) if not cleaned: raise AppException( code="NO_MEMBERS_GIVEN", message="请至少提供一个成员邮箱。", status_code=400, ) if len(cleaned) > MAX_BATCH_MEMBERS: raise AppException( code="TOO_MANY_MEMBERS", message=( f"一次最多添加 {MAX_BATCH_MEMBERS} 个成员" f"(本次 {len(cleaned)} 个)。请分批提交。" ), status_code=400, ) driver = await self._get_driver() existing = await self._live_roles(group.email) results: list[MemberOpResult] = [] # Addresses whose local owner row this call reconciled # (Google already had them as OWNER). Audited once after the # loop so a 50-address batch writes one row, not fifty. mirror_repairs: list[str] = [] for email in cleaned: current = existing.get(email) if current == target_role: # Same repair as ``grant_owner``'s ``already_owner`` # branch, and this is the path that actually gets used: # the drawer renders 设为负责人 only for a non-owner row, # so an operator whose mirror is missing a real Google # owner reaches for 添加负责人 — which posts HERE with # ``role=OWNER``. Returning ``already_member`` without # reconciling left the only repair affordance in the UI # pointing at a no-op, and the group still rendering # 无负责人. if target_role == ROLE_OWNER: person = await self._person_by_email(email) # ``person is not None`` is load-bearing, not # defensive. Both ``_mirror_has_owner`` and # ``_mirror_owner_added`` return early for an # unresolvable address — and this method explicitly # supports external owners (vendor addresses), with # departed employees unresolvable too. Without it, # such an address books an ``owner_mirror_repaired`` # audit row asserting a repair that did not happen, # again on every retry, while the group still reads # 无负责人. ``grant_owner`` got this right; this copy # did not. if ( person is not None and not await self._mirror_has_owner(group, person) ): await self._mirror_owner_added(group, person) # Recorded for the audit row after the loop. # ``grant_owner`` audits its twin repair, and # leaving THIS one unaudited was worse than # either: this is the path the UI reaches (the # drawer renders 设为负责人 only for a non-owner # row, so an operator fixing a missing mirror # row uses 添加负责人 → here), so the brand-new # ``owner_mirror_repaired`` action matched # nothing in the ordinary case while the # ownerless-group KPI and the offboarding # blocker set silently gained an owner. mirror_repairs.append(email) results.append(MemberOpResult( email=email, ok=True, outcome="already_member", role=current, )) continue if current == ROLE_OWNER and target_role != ROLE_OWNER: # REFUSE. Re-roling an owner downward here would be a # demotion with no owner-count check, so pasting a # re-sync list that happens to include the sole owner # would leave the group ownerless — the exact state # ``revoke_owner`` refuses to create, reached through a # door that asks for no reason and shows no warning. # It would also be recorded as ``member_added``, so an # auditor filtering ``owner_removed`` would never see # the group lose its owner. results.append(MemberOpResult( email=email, ok=False, outcome="failed", role=current, error=( "该地址是这个组的负责人。把负责人「加为普通成员」" "等于降权,请用「取消负责人」—— 那条路径会检查不会" "让组失去最后一个负责人。" ), )) continue if current is not None: # Present with a different, non-owner role → re-role # rather than insert (an insert would 409). resp = await driver.set_group_member_role( group_email=group.email, member_email=email, role=target_role, ) outcome = "role_changed" elif target_role == ROLE_OWNER: resp = await driver.set_group_owner( group_email=group.email, member_email=email, ) outcome = "added" else: resp = await driver.add_group_member( group_email=group.email, member_email=email, ) outcome = "added" if resp.ok and (resp.data or {}).get("already_member"): # Google says they were already in the group even # though our live read did not show them — a # concurrent edit. Report the truth, not our guess. outcome = "already_member" if not resp.ok: results.append(MemberOpResult( email=email, ok=False, outcome="failed", error=resp.error_message, )) continue # ``set_group_member_role`` maps Google's 404 to ``ok=True`` # with ``already_absent`` so a caller can say "they are not # in this group". Treating that as success is how a re-role # Google REFUSED gets reported as done — and, for # ``role=OWNER``, how an ``email_group_owners`` row is # written for somebody who is not in the group at all, which # then blocks their offboarding until the next sync tick. if (resp.data or {}).get("already_absent"): results.append(MemberOpResult( email=email, ok=False, outcome="failed", role=current, error=( "Google 说这个地址已经不在组里了(候选状态是本次请求" "开始时读到的,期间被别人移除)。请刷新成员列表后重试。" ), )) continue # ``add_group_member`` always inserts as MEMBER; promote # when a higher role was asked for and the insert path ran. if ( target_role == ROLE_MANAGER and current is None and outcome == "added" ): promote = await driver.set_group_member_role( group_email=group.email, member_email=email, role=ROLE_MANAGER, ) # ``already_absent`` here is Google's post-insert settle # window: the insert succeeded but the patch cannot see # the member yet. Reporting ``role=MANAGER`` for someone # left as a plain MEMBER is the failure this checks for. if not promote.ok or (promote.data or {}).get("already_absent"): results.append(MemberOpResult( email=email, ok=False, outcome="failed", role=ROLE_MEMBER, error=( "已加入为普通成员,但提升为管理员失败:" f"{promote.error_message or 'Google 暂时看不到该成员'}" ), )) continue results.append(MemberOpResult( email=email, ok=True, outcome=outcome, role=target_role, )) if target_role == ROLE_OWNER: await self._mirror_owner_added( group, await self._person_by_email(email), ) if any(r.ok and r.outcome != "already_member" for r in results): await self._refresh_member_count(group) await self.db.flush() succeeded = [r for r in results if r.ok] # The action names WHAT happened, not which endpoint ran. This # method serves both 添加成员 and 添加负责人 (the UI's owner # button posts here with ``role=OWNER`` so a batch of owners is # one call), and filing an owner grant under # ``member_added`` makes the owner history incomplete on the one # surface built to answer "who made this person an owner" — # ``grant_owner`` writes ``owner_added`` for the identical state # change. ``detail.role`` carries the role either way. action = ( "email_group.owner_added" if target_role == ROLE_OWNER else "email_group.member_added" ) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action=action, status="success" if succeeded else "failure", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "role": target_role, "requested": cleaned, "results": [r.to_dict() for r in results], "reason": reason, }, ) if mirror_repairs: await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="email_group.owner_mirror_repaired", status="success", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "owner_emails": mirror_repairs, "note": ( "这些地址在 Google 上已经是负责人,但本地 " "email_group_owners 缺行;本次已补齐" "(Google 侧未做任何改动)。" ), "reason": reason, }, ) for r in succeeded: if r.outcome == "already_member": continue await self._notify_membership( group=group, email=r.email, actor=actor, role=target_role, reason=reason, trace_id=trace_id, ) return { "group_email": group.email, "role": target_role, "results": [r.to_dict() for r in results], "added": sum(1 for r in results if r.ok and r.outcome == "added"), "already_member": sum( 1 for r in results if r.ok and r.outcome == "already_member" ), "role_changed": sum( 1 for r in results if r.ok and r.outcome == "role_changed" ), "failed": sum(1 for r in results if not r.ok), } async def _group_holding_alias(self, alias: str) -> EmailGroup | None: """The group whose mirrored ``aliases`` array contains ``alias``. A JSON-cast LIKE, matching how ``manage_service`` and ``service`` already search the same column. Includes soft-deleted rows: Google keeps an alias reserved until the group is really gone, so a tombstone still explains the 409. The pattern is ``%""%`` — quoted, so it matches a whole JSON ARRAY ELEMENT. An unquoted ``%%`` is the same unanchored-substring bug this domain already fixed once in ``_internal_reference_index``, inverted: looking up ``it@pingcap.com`` would match every array holding ``audit@pingcap.com`` / ``submit@pingcap.com`` / ``recruit@pingcap.com``, and with 760 groups on prod the real holder could fall outside the row window entirely — so the parsed-array confirm below would find nothing and the operator would get back the bare "Google 添加别名失败" this guard exists to replace. """ from sqlalchemy import String as SaString aliases_text = func.lower( func.coalesce(func.cast(EmailGroup.aliases, SaString), "[]"), ) rows = (await self.db.execute( select(EmailGroup) .where(aliases_text.like( f'%"{_like_esc(alias)}"%', escape="\\", )) .order_by(EmailGroup.email) .limit(20) )).scalars().all() for row in rows: # The LIKE is a pre-filter; confirm on the parsed array so a # substring hit ("ops@x.com" inside "devops@x.com") cannot # produce a wrong holder in the error message. if alias in (row.aliases or []): return row return None async def _notify_owner_revoked( self, *, group: EmailGroup, email: str, actor: IdentityPerson, demoted: bool, reason: str | None, trace_id: str, ) -> None: """Tell someone they are no longer an owner. Best-effort. Losing owner status is a change in RESPONSIBILITY, not just in what mail you receive: until they are told, the group's former owner still believes they are handling its requests and nobody is. (An earlier version of this docstring claimed member removals are deliberately silent, to justify this one. That was simply false — ``remove_members`` dispatches ``mail_group_member_removed`` for every resolvable address — and the same wrong rationale reached CHANGELOG and PLATFORM, where it would have given the next reader a wrong baseline for employee push volume. The whole owner-facing family notifies: this, ``mail_group_owner_transfer`` to the incoming owner, and ``mail_group_owner_transferred_away`` to the outgoing one.) """ person = await self._person_by_email(email) if person is None: return tail = "你仍然是该组的普通成员。" if demoted else "你已不在该组中。" # Two template keys, not one with a yes/no variable. The only # thing that differs between the demote and remove cases is a # boolean, and filling it with ``是``/``否`` put a Chinese # character into the middle of the English and Japanese bodies # ("Still a plain member: 是") — for a notification whose # recipient is an ordinary employee, i.e. the surface that MUST # render per-person language. template_key = ( "mail_group_owner_revoked_demoted" if demoted else "mail_group_owner_revoked_removed" ) try: env = NotificationEnvelope( recipient=person, title=f"你已不再是邮件组 {group.email} 的负责人", body_text=( f"你已不再是邮件组「{group.name}」({group.email})的负责人。" f"{tail}" + (f"\n\n操作原因:{reason}" if reason else "") ), template_key=template_key, context={ "person_name": person.display_name or person.work_email or "", "group_email": group.email, "group_name": group.name, "actor_name": actor.display_name or actor.work_email or "", "reason": reason or "—", }, related_resource_type="email_group", related_resource_id=group.id, ) # ``enqueue``, NOT ``dispatch``. This runs inside # ``revoke_owner``'s ``SELECT … FOR UPDATE`` window, and # ``dispatch`` = enqueue + a synchronous first delivery # attempt against the real channels — a Feishu call on a # 30s timeout would hold the ``email_groups`` row lock for # that whole time (and, because ``email_group_owners`` has an # FK to it, block the 6-hourly owner-join rewrite for this # group too). ``enqueue`` writes the durable outbox row and # nothing else; ``notification.outbox_retry`` delivers it, # which is the established pattern for exactly this (see the # ~10 ``notify.enqueue(env)`` call sites under # ``domains/offboarding`` and ``domains/asset``). # # The trade-off, stated so it stays a decision: delivery can # now lag by up to one retry tick (hourly) instead of going # out inside the request. That is the right price here — # "you are no longer an owner" is a responsibility FYI, not # something anyone acts on in the next sixty minutes, and # ``enqueue`` writes the durable row so the notice cannot be # lost. Holding an ``email_groups`` row lock across a Feishu # call to save that lag would be trading a real availability # problem for a cosmetic one. The worker picks up # ``status=pending`` with ``next_attempt_at IS NULL`` first # (``retry_pending_outbox``), so this row is at the head of # the queue rather than behind the backoff of failed ones. await self.notify.enqueue(env) except Exception: logger.exception( "mail-group owner-revoked notify failed (non-fatal) for %s", email, ) async def _notify_membership( self, *, group: EmailGroup, email: str, actor: IdentityPerson, role: str, reason: str | None, trace_id: str, ) -> None: """Tell the person they were added. Best-effort. Only resolvable people are notified — an external address has no Helix person row and therefore no notification channel. The add itself already succeeded, so a notification failure must never surface as a failed operation. """ person = await self._person_by_email(email) if person is None: return role_label = { ROLE_OWNER: "负责人", ROLE_MANAGER: "管理员", ROLE_MEMBER: "成员", }.get(role, role) try: env = NotificationEnvelope( recipient=person, title=f"你已被加入邮件组 {group.email}", body_text=( f"你已被加入邮件组「{group.name}」({group.email})," f"身份:{role_label}。" + (f"\n\n操作原因:{reason}" if reason else "") ), template_key="mail_group_member_added", context={ "person_name": person.display_name or person.work_email or "", "group_email": group.email, "group_name": group.name, "role": role_label, "actor_name": actor.display_name or actor.work_email or "", "reason": reason or "—", }, related_resource_type="email_group", related_resource_id=group.id, ) await self.notify.dispatch( env, trace_id=trace_id, actor_user_id=actor.id, ) except Exception: logger.exception( "mail-group member-added notify failed (non-fatal) for %s", email, ) async def remove_members( self, *, group_id: UUID, emails: list[str], actor: IdentityPerson, actor_roles: list[str], reason: str | None, trace_id: str, ) -> dict[str, Any]: """Remove addresses from a static group. Owners are ALWAYS refused here. Removing an owner is a different operation with a different guard (never leave the group ownerless), so it lives behind :meth:`revoke_owner`. There is deliberately no override parameter: an unused flag whose only effect is to disable an owner-safety check is a loaded gun for the next author. """ group = await self._load_static_group(group_id, actor_roles=actor_roles) cleaned = [self._norm(e) for e in emails if self._norm(e)] if not cleaned: raise AppException( code="NO_MEMBERS_GIVEN", message="请至少提供一个成员邮箱。", status_code=400, ) if len(cleaned) > MAX_BATCH_MEMBERS: raise AppException( code="TOO_MANY_MEMBERS", message=( f"一次最多移除 {MAX_BATCH_MEMBERS} 个成员" f"(本次 {len(cleaned)} 个)。请分批提交。" ), status_code=400, ) driver = await self._get_driver() roles = await self._live_roles(group.email) results: list[MemberOpResult] = [] for email in cleaned: current = roles.get(email) if current is None: results.append(MemberOpResult( email=email, ok=True, outcome="already_absent", )) continue if current == ROLE_OWNER: results.append(MemberOpResult( email=email, ok=False, outcome="failed", role=current, error=( "该地址是这个组的负责人。请用「移除负责人」或" "「转交负责人」,那条路径会检查不会让组失去最后一个负责人。" ), )) continue resp = await driver.remove_group_member( group_email=group.email, member_email=email, ) if not resp.ok: results.append(MemberOpResult( email=email, ok=False, outcome="failed", role=current, error=resp.error_message, )) continue already = bool((resp.data or {}).get("already_absent")) results.append(MemberOpResult( email=email, ok=True, outcome="already_absent" if already else "removed", role=current, )) # No owner-mirror handling here on purpose: an OWNER never # reaches this point (refused above), so a mirror update # would be dead code that implies owners ARE removable on # this path. if any(r.ok and r.outcome == "removed" for r in results): await self._refresh_member_count(group) await self.db.flush() succeeded = [r for r in results if r.ok] await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="email_group.member_removed", status="success" if succeeded else "failure", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "requested": cleaned, "results": [r.to_dict() for r in results], "reason": reason, "via": "admin_batch", }, ) for r in results: if r.ok and r.outcome == "removed": await self._notify_removed( group=group, email=r.email, actor=actor, reason=reason, trace_id=trace_id, ) return { "group_email": group.email, "results": [r.to_dict() for r in results], "removed": sum(1 for r in results if r.ok and r.outcome == "removed"), "already_absent": sum( 1 for r in results if r.ok and r.outcome == "already_absent" ), "failed": sum(1 for r in results if not r.ok), } async def _notify_removed( self, *, group: EmailGroup, email: str, actor: IdentityPerson, reason: str | None, trace_id: str, ) -> None: person = await self._person_by_email(email) if person is None: return try: body = f"你已被移出邮件组 {group.email}。" if reason: body += f"\n\n操作原因:{reason}" env = NotificationEnvelope( recipient=person, title=f"你已被移出邮件组 {group.email}", body_text=body, template_key="mail_group_member_removed", context={ "person_name": person.display_name or person.work_email or "", "group_email": group.email, "group_name": group.name, "actor_name": actor.display_name or actor.work_email or "", "reason": reason or "—", }, related_resource_type="email_group", related_resource_id=group.id, ) await self.notify.dispatch( env, trace_id=trace_id, actor_user_id=actor.id, ) except Exception: logger.exception( "mail-group member-removed notify failed (non-fatal) for %s", email, ) # ── owners ─────────────────────────────────────────────────── async def grant_owner( self, *, group_id: UUID, email: str, actor: IdentityPerson, actor_roles: list[str], reason: str | None, trace_id: str, ) -> dict[str, Any]: """Make one address an OWNER (insert or promote). This is the fix for the 459 ownerless static groups: it does not require an existing owner to transfer from, which ``transfer_group_owner`` does — an ownerless group cannot be rescued by a transfer. """ group = await self._load_static_group(group_id, actor_roles=actor_roles) target = self._norm(email) if not target or "@" not in target: raise AppException( code="INVALID_EMAIL", message=f"{email!r} 不是一个邮箱地址。", status_code=400, ) roles = await self._live_roles(group.email) if roles.get(target) == ROLE_OWNER: # Already an owner on Google — but the LOCAL mirror may be # missing the row (an owner Google gained while # ``list_group_owners`` was returning 403, say). This is the # one action an operator would reach for to repair that, so # it must not be a pure no-op: reconcile the mirror, then # report that Google needed no change. person = await self._person_by_email(target) before = await self._mirror_has_owner(group, person) await self._mirror_owner_added(group, person) await self.db.flush() if not before and person is not None: # An ``email_group_owners`` INSERT is a real governance # state change: it feeds the ownerless-group KPI, the # ``owners=none`` filter and the offboarding blocker # set. Committing it with no audit row would make the # one surface built to answer "who made this person an # owner" unable to — the same completeness hole that # filing an owner grant under ``member_added`` created. await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="email_group.owner_mirror_repaired", status="success", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "owner_email": target, "resolved_person_id": str(person.id), "note": ( "Google 上已经是负责人,但本地 email_group_owners " "缺这一行;本次已补齐(Google 侧未做任何改动)。" ), "reason": reason, }, ) return { "group_email": group.email, "email": target, "outcome": "already_owner", "resolved_person_id": str(person.id) if person else None, "mirror_repaired": bool(not before and person is not None), } driver = await self._get_driver() resp = await driver.set_group_owner( group_email=group.email, member_email=target, ) if not resp.ok: raise AppException( code="GOOGLE_SET_OWNER_FAILED", message=f"Google 设置负责人失败: {resp.error_message}", status_code=502, ) person = await self._person_by_email(target) await self._mirror_owner_added(group, person) # A grant can INSERT a member (owners are members), so the count # moves too. await self._refresh_member_count(group) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="email_group.owner_added", status="success", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "owner_email": target, "previous_role": roles.get(target), "resolved_person_id": str(person.id) if person else None, "reason": reason, }, ) if person is not None: await self._notify_membership( group=group, email=target, actor=actor, role=ROLE_OWNER, reason=reason, trace_id=trace_id, ) return { "group_email": group.email, "email": target, "outcome": "granted", "resolved_person_id": str(person.id) if person else None, } async def revoke_owner( self, *, group_id: UUID, email: str, demote_to_member: bool, actor: IdentityPerson, actor_roles: list[str], reason: str | None, trace_id: str, ) -> dict[str, Any]: """Drop one owner, refusing to leave the group ownerless. Google requires ≥1 owner per group and Helix treats ownership as an offboarding blocker, so an ownerless group is a governance hole, not a neutral state. The count comes from the LIVE roster, so an external co-owner (invisible in ``email_group_owners``) correctly counts as cover. ``demote_to_member=True`` keeps them in the group as a plain member — the usual intent for "they should not be responsible for this any more" — while ``False`` removes them outright. """ group = await self._load_static_group(group_id, actor_roles=actor_roles) # Serialize the last-owner decision. The check below is # check-then-act across an EXTERNAL system, so without a lock two # operators revoking the two owners of a 2-owner group both read # ``owner_count == 2``, both pass, and the group ends up # ownerless — the exact state the guard exists to prevent. # ``@idempotent`` does not help: it dedupes one key, not two # distinct requests. Same primitive ``create_mail_group`` already # uses for the (less consequential) create rate limit. # # The lock IS held across the Google calls, because the # decision and the mutation have to be atomic together — that is # the whole point. Be honest about the width: it spans TWO to # THREE Directory API round trips on ONE ``email_groups`` row # (``members.list`` for the decision, the patch/delete, and on an # outright removal one more ``members.list`` to re-count) — not # the "one call" an earlier version of this comment claimed. # That is still not the shape that has burned this codebase # before (a lock held across a 700-call fan-out): a second revoke # on the SAME group waits for those round trips, and every other # group is unaffected. What must NOT be in here is anything # slower or less predictable than a Directory call — see the # ``enqueue`` (not ``dispatch``) note on the notification below. await self.db.execute( select(EmailGroup.id) .where(EmailGroup.id == group.id) .with_for_update() ) target = self._norm(email) roles = await self._live_roles(group.email) current = roles.get(target) if current is None: raise AppException( code="NOT_A_MEMBER", message=f"{target} 不是邮件组 {group.email} 的成员。", status_code=404, ) if current != ROLE_OWNER: raise AppException( code="NOT_AN_OWNER", message=f"{target} 不是邮件组 {group.email} 的负责人。", status_code=409, ) owner_count = sum(1 for r in roles.values() if r == ROLE_OWNER) if owner_count <= 1: raise AppException( code="LAST_OWNER", message=( f"{target} 是邮件组 {group.email} 的最后一个负责人。" "Google 要求每个组至少有一个负责人,而无负责人的组也是" "离职卡点。请先指定新的负责人,再移除这一个。" ), status_code=409, ) driver = await self._get_driver() if demote_to_member: resp = await driver.set_group_member_role( group_email=group.email, member_email=target, role=ROLE_MEMBER, ) outcome = "demoted" else: resp = await driver.remove_group_member( group_email=group.email, member_email=target, ) outcome = "removed" if not resp.ok: raise AppException( code="GOOGLE_REVOKE_OWNER_FAILED", message=f"Google 移除负责人失败: {resp.error_message}", status_code=502, ) already_absent = bool((resp.data or {}).get("already_absent")) if already_absent: # Google answered "that address is not in this group" — the # roster this request was decided on went stale in the # milliseconds between the read and the write. # # This RESOLVES the operator's intent rather than failing it: # the person is not an owner any more, which is what they # asked for, just not by the route they picked. So it is a # 200 with a DISTINCT outcome, not "demoted" (which would # claim they are still a plain member of a group they are not # in at all) and not an error. # # An earlier attempt DID raise a 409 here, after clearing the # mirror row and flushing. That never worked: ``get_db`` # (``core/database.py``) does ``except Exception: # rollback()``, so the DELETE was discarded by the very # exception it preceded — leaving the phantom # ``email_group_owners`` row (an offboarding blocker no UI # action could clear, since every retry took the same branch) # behind an error message that said it had been cleaned up. # **A flush is not a commit: on any path that raises, the # router never commits and every write in the request is # rolled back.** The test that "proved" the fix only passed # because ``conftest``'s ``override_get_db`` yields the # session without that rollback. outcome = "already_absent" person = await self._person_by_email(target) await self._mirror_owner_removed(group, person) # Only an outright removal changes the member count — a demote # keeps them in the group. Skipping it on the demote branch also # keeps one Google round trip out of the ``FOR UPDATE`` window. # ``already_absent`` needs it too, on either branch: the roster # moved under us, so the mirrored count is known-stale. if not demote_to_member or already_absent: await self._refresh_member_count(group) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="email_group.owner_removed", status="success", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "owner_email": target, "outcome": outcome, # ``None`` when the roster moved under us: the count was # derived from the pre-write read, so reporting it as the # remaining number would be a fabrication in the one # case we KNOW the read was stale. "remaining_owner_count": ( None if already_absent else owner_count - 1 ), "resolved_person_id": str(person.id) if person else None, "reason": reason, }, ) # Only when Helix actually took the role away. On # ``already_absent`` somebody else already did, in the Google # console, and this request's own audit row is the record of # what Helix observed — a "you are no longer an owner" push whose # stated actor is an HR admin who did not do it is worse than # silence. if not already_absent: await self._notify_owner_revoked( group=group, email=target, actor=actor, demoted=demote_to_member, reason=reason, trace_id=trace_id, ) return { "group_email": group.email, "email": target, "outcome": outcome, "remaining_owner_count": ( None if already_absent else owner_count - 1 ), } # ── display metadata ───────────────────────────────────────── async def update_display( self, *, group_id: UUID, name: str | None, description: str | None, actor: IdentityPerson, actor_roles: list[str], trace_id: str, ) -> dict[str, Any]: """Rename a static group / edit its description. Uses the Directory API (``admin.directory.group``, already delegated) rather than Cloud Identity, so this carries no new service-account scope prerequisite — see ``patch_group_directory_display``. """ group = await self._load_static_group(group_id, actor_roles=actor_roles) new_name = name.strip() if name is not None else None if new_name is not None and not new_name: raise AppException( code="INVALID_GROUP_NAME", message="邮件组名称不能为空。", status_code=400, ) # Bound HERE, not only in the router's Pydantic model: the MCP # tool passes ``name``/``description`` straight through. Google # is patched before the local write, so an over-long value would # rename the group on Google and then blow up on the local flush # (MySQL 1406 under strict mode) → rollback → Google renamed, # Helix unchanged, no audit row. Silent drift on the mirror the # estate list and the cleanup pool read from. if new_name is not None and len(new_name) > _NAME_MAX: raise AppException( code="GROUP_NAME_TOO_LONG", message=f"邮件组名称最多 {_NAME_MAX} 个字符。", status_code=400, ) if description is not None and len(description) > _DESCRIPTION_MAX: raise AppException( code="GROUP_DESCRIPTION_TOO_LONG", message=f"邮件组描述最多 {_DESCRIPTION_MAX} 个字符。", status_code=400, ) # ``description`` may legitimately be set to "" (clear it), so # only ``None`` means "leave alone". new_desc = description if new_name is None and new_desc is None: raise AppException( code="NOTHING_TO_UPDATE", message="请至少提供 name 或 description。", status_code=400, ) # Per-field comparison. ``None`` means "leave alone", and # ``None == ""`` is False — so a conjunction over both fields can # NEVER short-circuit a single-field payload, which is the normal # shape for an edit dialog that sends only what the user touched. # Without this the route (which deliberately carries no # idempotency key BECAUSE unchanged edits are free) writes a # phantom "renamed X → X" row on every save and every redelivery. name_same = new_name is None or new_name == group.name desc_same = new_desc is None or new_desc == (group.description or "") if name_same and desc_same: return { "group_email": group.email, "outcome": "unchanged", "name": group.name, "description": group.description, } driver = await self._get_driver() resp = await driver.patch_group_directory_display( group_email=group.email, name=new_name, description=new_desc, ) if not resp.ok: raise AppException( code="GOOGLE_PATCH_GROUP_FAILED", message=f"Google 更新邮件组信息失败: {resp.error_message}", status_code=502, ) old_name, old_desc = group.name, group.description if new_name is not None: group.name = new_name if new_desc is not None: group.description = new_desc or None group.updated_at = datetime.now(UTC) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="email_group.display_updated", status="success", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "old_name": old_name, "new_name": group.name, "old_description": old_desc, "new_description": group.description, }, ) return { "group_email": group.email, "outcome": "updated", "name": group.name, "description": group.description, } # ── aliases ────────────────────────────────────────────────── async def add_alias( self, *, group_id: UUID, alias_email: str, actor: IdentityPerson, actor_roles: list[str], trace_id: str, reason: str | None = None, ) -> dict[str, Any]: """Add an alias address to a static group.""" group = await self._load_static_group(group_id, actor_roles=actor_roles) alias = self._norm(alias_email) if "@" not in alias: raise AppException( code="INVALID_EMAIL", message=f"{alias_email!r} 不是一个邮箱地址。", status_code=400, ) suffix = alias.rsplit("@", 1)[1] whitelist = await self._eg.get_suffix_whitelist() if suffix not in whitelist: raise AppException( code="SUFFIX_NOT_ALLOWED", message=( f"后缀 {suffix!r} 不在白名单中。允许的后缀:" f" {', '.join(whitelist)}" ), status_code=400, ) if alias == group.email: raise AppException( code="ALIAS_IS_PRIMARY", message="别名不能和邮件组主地址相同。", status_code=400, ) if alias in (group.aliases or []): return { "group_email": group.email, "alias": alias, "outcome": "already_present", } # An address already used as another group's primary or alias # would be rejected by Google with a bare 409; check locally so # the operator is told WHICH group holds it. holder = await self._eg.find_group_by_email(alias, include_deleted=True) if holder is not None: raise AppException( code="ALIAS_TAKEN", message=( f"{alias} 已经是邮件组 {holder.email} 的主地址" f"(状态:{holder.status})。" ), status_code=409, ) # ...and an address already held as ANOTHER group's alias. Google # rejects that with the same bare 409 as a primary collision, so # without this the operator gets "Google 添加别名失败" and no # idea which group to look at. One indexed-free but tiny scan # (760 rows on prod) over the mirrored alias arrays. alias_holder = await self._group_holding_alias(alias) if alias_holder is not None: raise AppException( code="ALIAS_TAKEN", message=( f"{alias} 已经是邮件组 {alias_holder.email} 的别名" f"(状态:{alias_holder.status})。" ), status_code=409, ) driver = await self._get_driver() resp = await driver.add_alias( target_email=group.email, alias_email=alias, target_kind="group", ) if not resp.ok: raise AppException( code="GOOGLE_ADD_ALIAS_FAILED", message=f"Google 添加别名失败: {resp.error_message}", status_code=502, ) # Mirror immediately so the list reflects the change before the # next sync tick — but only when we HAVE a mirror. On a # never-alias-synced row (``aliases IS NULL``) writing # ``[alias]`` would convert "we have never read this group's # aliases" into the positive claim "this is the complete set", # while Google may hold several non-editable ones. The tri-state # is documented on the column and the UI renders it # (``alias_never_synced``); leave NULL for the sync to fill. if group.aliases is not None: group.aliases = sorted({*group.aliases, alias}) group.updated_at = datetime.now(UTC) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="email_group.alias_added", status="success", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "alias": alias, "reason": reason, }, ) return { "group_email": group.email, "alias": alias, "outcome": "added", } async def remove_alias( self, *, group_id: UUID, alias_email: str, actor: IdentityPerson, actor_roles: list[str], trace_id: str, reason: str | None = None, ) -> dict[str, Any]: """Remove an alias from a static group. Google exposes two alias arrays and only the editable one can be deleted; ``email_groups.aliases`` merges both, so an attempt to delete a non-editable alias comes back from Google as an error rather than silently succeeding. """ group = await self._load_static_group(group_id, actor_roles=actor_roles) alias = self._norm(alias_email) if "@" not in alias: raise AppException( code="INVALID_EMAIL", message=f"{alias_email!r} 不是一个邮箱地址。", status_code=400, ) if alias == group.email: raise AppException( code="ALIAS_IS_PRIMARY", message="这是邮件组的主地址,不是别名,不能在这里移除。", status_code=400, ) # Refuse an address this group does not carry. Without it a # typo produces a green "removed" toast and an # ``alias_removed`` audit row against a group that never had # that alias — pollution of the very log this desk exists to # make trustworthy. ``aliases is None`` means never synced, so # we cannot tell: allow it through and let Google decide. if group.aliases is not None and alias not in group.aliases: raise AppException( code="ALIAS_NOT_ON_GROUP", message=( f"{alias} 不是邮件组 {group.email} 的别名。" ), status_code=404, ) driver = await self._get_driver() resp = await driver.remove_alias( target_email=group.email, alias_email=alias, target_kind="group", ) if not resp.ok: raise AppException( code="GOOGLE_REMOVE_ALIAS_FAILED", message=f"Google 移除别名失败: {resp.error_message}", status_code=502, ) already = bool((resp.data or {}).get("already_absent")) # Only touch the mirror when something was actually removed, and # only when we HAVE a mirror. Two traps here: # * ``(group.aliases or [])`` on a NULL column would write # ``[]``, converting "never alias-synced" into the positive # claim "Google has no aliases for this group" — on the # authority of one delete call. The cleanup workbench reads # that claim: an empty list suppresses the ``has_real_alias`` # BLOCKING signal, making a group with a live second address # sweepable in the batch delete. # * A Google 404 (``already_absent``) removed nothing, so # stripping the address from the mirror would hide an alias # that may still be routing mail (Google's non-editable # aliases cannot be deleted but ARE merged into this column). if not already and group.aliases is not None: group.aliases = sorted(a for a in group.aliases if a != alias) group.updated_at = datetime.now(UTC) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="email_group.alias_removed", # Nothing changed on Google, so this is not a success. The # row still exists — the attempt is the audit-worthy fact. status="success" if not already else "noop", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "alias": alias, "already_absent_on_google": already, # Recorded because THIS is the write that can unlock an # irreversible delete: a real alias raises the cleanup # workbench's ``has_real_alias`` BLOCKING signal, so # dropping the last one moves the row toward # ``risk: low``. "reason": reason, "remaining_aliases": ( None if group.aliases is None else len(group.aliases) ), }, ) return { "group_email": group.email, "alias": alias, "outcome": "already_absent" if already else "removed", }