"""GitHub 组织登记表的读写,以及成员台账的共享原语。 这一层**不碰 GitHub API**。它只维护 Helix 这一侧的事实:我们纳管哪些 组织、台账里有谁、每条成员资格经历过什么。真正调 GitHub 的动作在驱动 层(PR2),由申请链路和离职回收调用;把两者分开,是为了让「凭证还没到位」 这件事只影响一层,而不是让整条链路无法上线。 台账写入的唯一入口是 :func:`upsert_membership` / :func:`mark_removed` 这几个函数。它们负责两件容易被漏掉的事: * **状态迁移必须落一条事件**。散在各处的 ``m.state = "active"`` 会让 事件流出现空洞,而事件流正是访问复核唯一的证据来源。 * **``github_login_lower`` 必须与 ``github_login`` 同步写**。唯一约束挂在 小写列上,只改大小写列等于绕过约束。 """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ConflictError, NotFoundError, ValidationError from app.identity.person_models import IdentityPerson from . import constants as const from .models import ( GithubOrg, GithubOrgMembership, GithubOrgMembershipEvent, ) from .validators import ( bounded_text, normalize_github_login, normalize_org_slug, validate_enum, ) def _now() -> datetime: return datetime.now(UTC) # ════════════════════════════════════════════════════════════════════ # 组织登记表 # ════════════════════════════════════════════════════════════════════ async def list_orgs( db: AsyncSession, *, include_inactive: bool = False, request_enabled_only: bool = False, ) -> list[GithubOrg]: q = select(GithubOrg).where(GithubOrg.deleted_at.is_(None)) if not include_inactive: q = q.where(GithubOrg.is_active.is_(True)) if request_enabled_only: q = q.where(GithubOrg.request_enabled.is_(True)) q = q.order_by(GithubOrg.sort_order, GithubOrg.slug) return list((await db.execute(q)).scalars().all()) async def get_org(db: AsyncSession, org_id: str) -> GithubOrg: org = await db.get(GithubOrg, str(org_id)) if org is None or org.deleted_at is not None: raise NotFoundError(f"GitHub organization {org_id} not found") return org async def get_org_by_slug(db: AsyncSession, slug: str) -> GithubOrg | None: q = select(GithubOrg).where( GithubOrg.slug_lower == slug.strip().lower(), GithubOrg.deleted_at.is_(None), ) return (await db.execute(q)).scalars().first() async def org_member_counts(db: AsyncSession) -> dict[str, int]: """一次查询算出所有组织的成员计数,喂给 ``serialize_org``。 返回的键是 ``"{org_id}:{bucket}"``,bucket ∈ {active, pending_invite, unlinked}。扁平字典而不是嵌套,是为了让序列化那边一次 ``.get`` 拿到, 不用做存在性判断 —— 一个没有任何成员的组织应该显示 0,不是显示空。 """ counts: dict[str, int] = {} rows = ( await db.execute( select( GithubOrgMembership.org_id, GithubOrgMembership.state, func.count().label("n"), ).group_by(GithubOrgMembership.org_id, GithubOrgMembership.state) ) ).all() for org_id, state, n in rows: counts[f"{org_id}:{state}"] = int(n) # 「在 GitHub 上有、但 Helix 认不出是谁」的条数 —— 这是治理面板上 # 最该被看见的一个数,所以它单独算,而不是让前端从别的数里减。 unlinked = ( await db.execute( select(GithubOrgMembership.org_id, func.count().label("n")) .where( GithubOrgMembership.person_id.is_(None), GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES), ) .group_by(GithubOrgMembership.org_id) ) ).all() for org_id, n in unlinked: counts[f"{org_id}:unlinked"] = int(n) return counts async def create_org( db: AsyncSession, *, slug: str, display_name: str, description: str | None = None, is_community: bool = False, offboarding_policy: str = const.OFFBOARDING_POLICY_REMOVE, request_enabled: bool = True, auto_join_enabled: bool = False, default_role: str = const.MEMBERSHIP_ROLE_MEMBER, feishu_option_value: str | None = None, integration_id: str | None = None, sort_order: int = 0, notes: str | None = None, actor_person_id: str | None = None, ) -> GithubOrg: normalized = normalize_org_slug(slug) lower = normalized.lower() existing = ( await db.execute( select(GithubOrg).where(GithubOrg.slug_lower == lower) ) ).scalars().first() if existing is not None: # 软删过的同名组织:复活它而不是新建一行。新建会撞唯一约束 # (约束不认 deleted_at),而且会把旧台账留在一个不可见的孤儿 # 组织下面 —— 那些成员资格是真的,不该因为一次误删就消失。 if existing.deleted_at is not None: existing.deleted_at = None existing.is_active = True existing.display_name = bounded_text( display_name, 128, "display_name", ) or existing.display_name return existing raise ConflictError( f"GitHub organization '{normalized}' is already registered", ) name = bounded_text(display_name, 128, "display_name") if not name: raise ValidationError("display_name is required") org = GithubOrg( id=str(uuid.uuid4()), slug=normalized, slug_lower=lower, display_name=name, description=bounded_text(description, 2000, "description"), is_community=bool(is_community), offboarding_policy=validate_enum( offboarding_policy, const.OFFBOARDING_POLICIES, "offboarding_policy", ), request_enabled=bool(request_enabled), auto_join_enabled=bool(auto_join_enabled), default_role=validate_enum( default_role, const.MEMBERSHIP_ROLES, "default_role", ), feishu_option_value=bounded_text( feishu_option_value, 128, "feishu_option_value", ), integration_id=integration_id or None, sort_order=int(sort_order or 0), notes=bounded_text(notes, 2000, "notes"), is_active=True, created_by=str(actor_person_id) if actor_person_id else None, ) db.add(org) await db.flush() return org #: ``update_org`` 认得的字段与它们的校验方式。表驱动而不是一长串 if, #: 是为了让「加一个可编辑字段」只需要加一行,而不是在函数中间插一段。 _ORG_TEXT_FIELDS = { "display_name": 128, "description": 2000, "feishu_option_value": 128, "notes": 2000, } _ORG_BOOL_FIELDS = ( "is_community", "request_enabled", "auto_join_enabled", "is_active", ) _ORG_ENUM_FIELDS = { "offboarding_policy": const.OFFBOARDING_POLICIES, "default_role": const.MEMBERSHIP_ROLES, } async def update_org( db: AsyncSession, org: GithubOrg, *, changes: dict[str, Any], ) -> dict[str, dict[str, Any]]: """按 PATCH 语义就地更新,返回 ``{field: {"from":…, "to":…}}`` 差异。 ``changes`` 只包含调用方**显式给出**的字段(路由层用 ``model_dump(exclude_unset=True)`` 取)。这一点是必须的:``None`` 在这里意思是「清空」,不是「没给」—— 两者混淆会让一次只改排序的 请求把备注洗掉。 返回差异而不是 True/False,是因为审计要按**值**记。只记「改了几个 字段」会让一次扩权(例如把离职策略从 remove 改成 exempt)在审计里 和一次改备注长得一模一样。 """ diff: dict[str, dict[str, Any]] = {} def _apply(field: str, new_value: Any) -> None: """就地写入并记差异 —— 三类字段共用一个出口。 差异按**值**记(见调用方 router 的说明):只记字段名会让 「离职策略 remove → exempt」和「改了个备注」在审计里长得一样。 """ old_value = getattr(org, field) if old_value != new_value: diff[field] = {"from": old_value, "to": new_value} setattr(org, field, new_value) for field, limit in _ORG_TEXT_FIELDS.items(): if field not in changes: continue text_value = bounded_text(changes[field], limit, field) if field == "display_name" and not text_value: raise ValidationError("display_name cannot be empty") _apply(field, text_value) for field in _ORG_BOOL_FIELDS: if field not in changes: continue _apply(field, bool(changes[field])) for field, allowed in _ORG_ENUM_FIELDS.items(): if field not in changes: continue _apply(field, validate_enum(str(changes[field]), allowed, field)) if "sort_order" in changes: _apply("sort_order", int(changes["sort_order"] or 0)) if "integration_id" in changes: _apply("integration_id", changes["integration_id"] or None) return diff async def soft_delete_org(db: AsyncSession, org: GithubOrg) -> None: """停用并软删一个组织。 **不**级联删台账。理由:一条成员资格记录的是「这个人在 GitHub 上 确实有过访问权」,这个事实不因为我们不再纳管这个组织而改变;而且 离职回收如果因为组织被摘掉就跳过它,等于给了一条静默的漏网路径。 真要清理,先把成员移除干净,再摘组织。 """ live = ( await db.execute( select(func.count()) .select_from(GithubOrgMembership) .where( GithubOrgMembership.org_id == org.id, GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES), ) ) ).scalar_one() if live: raise ConflictError( f"organization still has {live} live membership(s) — " "remove them before retiring it", ) org.is_active = False org.deleted_at = _now() # ════════════════════════════════════════════════════════════════════ # 成员台账 # ════════════════════════════════════════════════════════════════════ async def find_membership( db: AsyncSession, *, org_id: str, github_login: str, ) -> GithubOrgMembership | None: q = select(GithubOrgMembership).where( GithubOrgMembership.org_id == str(org_id), GithubOrgMembership.github_login_lower == github_login.strip().lower(), ) return (await db.execute(q)).scalars().first() async def find_person_membership_in_org( db: AsyncSession, *, org_id: str, person_id: str, ) -> GithubOrgMembership | None: """这个**人**在这个组织里有没有活着的成员资格 —— 不问是哪个 handle。 与 :func:`find_membership`(按 handle 查)是两个不同的问题,而混用 它们会造成一个具体的错误:一个人换了 GitHub 账号之后,按新 handle 查不到旧那条记录,于是自动加入会**再邀请一次**,同一个组织里留下 这个人的两条成员资格。 「他已经有访问权了吗」这个问题的答案永远按人算;「这个 handle 是 什么状态」才按 handle 算。 """ q = select(GithubOrgMembership).where( GithubOrgMembership.org_id == str(org_id), GithubOrgMembership.person_id == str(person_id), GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES), ) return (await db.execute(q)).scalars().first() async def list_memberships_for_person( db: AsyncSession, person_id: str, *, live_only: bool = True, ) -> list[GithubOrgMembership]: q = select(GithubOrgMembership).where( GithubOrgMembership.person_id == str(person_id), ) if live_only: q = q.where(GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES)) q = q.order_by(GithubOrgMembership.created_at.desc()) return list((await db.execute(q)).scalars().all()) async def add_event( db: AsyncSession, membership: GithubOrgMembership, *, event_type: str, from_state: str | None = None, to_state: str | None = None, actor_person_id: str | None = None, actor_kind: str = "system", detail: dict[str, Any] | None = None, trace_id: str | None = None, ) -> GithubOrgMembershipEvent: event = GithubOrgMembershipEvent( id=str(uuid.uuid4()), membership_id=membership.id, event_type=event_type, from_state=from_state, to_state=to_state, actor_person_id=str(actor_person_id) if actor_person_id else None, actor_kind=actor_kind, detail=detail, trace_id=trace_id, ) db.add(event) return event async def list_events( db: AsyncSession, membership_id: str, *, limit: int = 100, ) -> list[GithubOrgMembershipEvent]: q = ( select(GithubOrgMembershipEvent) .where(GithubOrgMembershipEvent.membership_id == str(membership_id)) .order_by(GithubOrgMembershipEvent.created_at.desc()) .limit(min(max(int(limit), 1), 500)) ) return list((await db.execute(q)).scalars().all()) async def upsert_membership( db: AsyncSession, *, org: GithubOrg, github_login: str, person_id: str | None, state: str, source: str, role: str | None = None, join_request_id: str | None = None, auto_join_rule_id: str | None = None, external_invitation_id: str | None = None, actor_person_id: str | None = None, actor_kind: str = "system", trace_id: str | None = None, event_type: str | None = None, detail: dict[str, Any] | None = None, ) -> GithubOrgMembership: """建立或推进一条 ``(org, handle)`` 的台账记录,并落一条事件。 幂等:同一个 ``(org, handle)`` 只有一行,重复调用推进状态而不是 插新行 —— 这正是「飞书重投一次回调」「对账重复跑一遍」时需要的性质。 ``person_id`` 只在**从空补上**时写入,不做覆盖:一条已经认到人的 记录被另一次调用改指到别人身上,是台账最坏的一种损坏。真要改指, 走 :func:`relink_membership`,它会留下一条明确的事件。 """ login = normalize_github_login(github_login) validate_enum(state, const.MEMBERSHIP_STATES, "state") validate_enum(source, const.MEMBERSHIP_SOURCES, "source") resolved_role = validate_enum( role or org.default_role, const.MEMBERSHIP_ROLES, "role", ) membership = await find_membership( db, org_id=org.id, github_login=login, ) now = _now() if membership is None: membership = GithubOrgMembership( id=str(uuid.uuid4()), org_id=org.id, person_id=str(person_id) if person_id else None, github_login=login, github_login_lower=login.lower(), state=state, role=resolved_role, source=source, join_request_id=join_request_id, auto_join_rule_id=auto_join_rule_id, external_invitation_id=external_invitation_id, ) _stamp_state_timestamps(membership, state, now) db.add(membership) await db.flush() await add_event( db, membership, event_type=event_type or _default_event_for_state(state), from_state=None, to_state=state, actor_person_id=actor_person_id, actor_kind=actor_kind, detail=detail, trace_id=trace_id, ) return membership previous = membership.state if person_id and not membership.person_id: membership.person_id = str(person_id) await add_event( db, membership, event_type=const.EVENT_LINKED_PERSON, actor_person_id=actor_person_id, actor_kind=actor_kind, detail={"person_id": str(person_id)}, trace_id=trace_id, ) if join_request_id: membership.join_request_id = join_request_id if auto_join_rule_id: membership.auto_join_rule_id = auto_join_rule_id if external_invitation_id is not None: membership.external_invitation_id = external_invitation_id if role and role != membership.role: await add_event( db, membership, event_type=const.EVENT_ROLE_CHANGED, actor_person_id=actor_person_id, actor_kind=actor_kind, detail={"from": membership.role, "to": resolved_role}, trace_id=trace_id, ) membership.role = resolved_role if previous != state: membership.state = state membership.source = source _stamp_state_timestamps(membership, state, now) await add_event( db, membership, event_type=event_type or _default_event_for_state(state), from_state=previous, to_state=state, actor_person_id=actor_person_id, actor_kind=actor_kind, detail=detail, trace_id=trace_id, ) return membership def _default_event_for_state(state: str) -> str: return { const.MEMBERSHIP_STATE_PENDING_INVITE: const.EVENT_INVITED, const.MEMBERSHIP_STATE_AWAITING_MANUAL: const.EVENT_AWAITING_MANUAL, const.MEMBERSHIP_STATE_ACTIVE: const.EVENT_JOINED, const.MEMBERSHIP_STATE_REMOVED: const.EVENT_REMOVED, const.MEMBERSHIP_STATE_DECLINED: const.EVENT_DECLINED, const.MEMBERSHIP_STATE_EXPIRED: const.EVENT_EXPIRED, }.get(state, const.EVENT_RECONCILED) def _stamp_state_timestamps( m: GithubOrgMembership, state: str, now: datetime, ) -> None: """状态与时间戳一起写。 分散着写是这类表最常见的腐烂方式:一条 active 却没有 ``joined_at`` 的记录,在「这个人什么时候进的组」这个问题上只能耸肩。 """ naive = now.replace(tzinfo=None) if state == const.MEMBERSHIP_STATE_AWAITING_MANUAL: # **不设 invitation_expires_at** —— 没有邀请,就没有到期这回事。 # 设了它,邀请对账会按时间去判一条根本不存在的邀请。 m.invitation_expires_at = None m.external_invitation_id = None m.removed_at = None m.removed_reason = None m.removal_source = None m.removed_by = None elif state == const.MEMBERSHIP_STATE_PENDING_INVITE: m.invited_at = naive m.invitation_expires_at = naive + timedelta( days=const.GITHUB_INVITATION_TTL_DAYS, ) m.removed_at = None m.removed_reason = None m.removal_source = None m.removed_by = None elif state == const.MEMBERSHIP_STATE_ACTIVE: m.joined_at = m.joined_at or naive m.invitation_expires_at = None m.external_invitation_id = None m.removed_at = None m.removed_reason = None m.removal_source = None m.removed_by = None elif state in ( const.MEMBERSHIP_STATE_DECLINED, const.MEMBERSHIP_STATE_EXPIRED, ): m.invitation_expires_at = None m.external_invitation_id = None elif state == const.MEMBERSHIP_STATE_REMOVED: m.removed_at = naive m.invitation_expires_at = None m.external_invitation_id = None async def mark_removed( db: AsyncSession, membership: GithubOrgMembership, *, reason: str, removal_source: str, actor_person_id: str | None = None, actor_kind: str = "system", trace_id: str | None = None, detail: dict[str, Any] | None = None, ) -> GithubOrgMembership: """把一条成员资格标成已移除。 幂等:已经是 removed 的记录直接返回,不重复落事件 —— 离职流程每个 tick 都会重扫,重复落事件会把事件流淹掉。 """ if membership.state == const.MEMBERSHIP_STATE_REMOVED: return membership previous = membership.state membership.state = const.MEMBERSHIP_STATE_REMOVED membership.removed_reason = (reason or "")[:255] or None membership.removal_source = removal_source membership.removed_by = str(actor_person_id) if actor_person_id else None _stamp_state_timestamps(membership, const.MEMBERSHIP_STATE_REMOVED, _now()) await add_event( db, membership, event_type=const.EVENT_REMOVED, from_state=previous, to_state=const.MEMBERSHIP_STATE_REMOVED, actor_person_id=actor_person_id, actor_kind=actor_kind, detail={**(detail or {}), "reason": reason, "source": removal_source}, trace_id=trace_id, ) return membership async def relink_membership( db: AsyncSession, membership: GithubOrgMembership, *, person_id: str | None, actor_person_id: str | None, trace_id: str | None = None, ) -> None: """把一条台账记录改指到另一个人(或解绑)。 与 ``upsert_membership`` 里「只从空补上」的规则相对:这里是一次 **明确的**改指,所以留一条自己的事件,把改前改后的人都写进去。 """ old = membership.person_id new = str(person_id) if person_id else None if old == new: return membership.person_id = new was_external = membership.external_confirmed_at is not None if new and was_external: # 认出是谁了,那之前那个「确认是外部人员」就是错的。清掉而不是 # 并存:两个标记同时为真时,离职回收的行为取决于哪个查询先跑到, # 而那是一个没人能从界面上看出来的分叉。 membership.external_confirmed_at = None membership.external_confirmed_by = None membership.external_note = None membership.external_kind = None await add_event( db, membership, event_type=const.EVENT_LINKED_PERSON, actor_person_id=actor_person_id, actor_kind="person", detail={ "from_person_id": old, "to_person_id": new, "cleared_external": bool(new and was_external), }, trace_id=trace_id, ) async def resolve_person_by_login( db: AsyncSession, github_login: str, ) -> IdentityPerson | None: """按 handle 反查员工。 只认 ``identity_persons.github_id``(来自 HRIS 入职表单),**不**回退到 「邮箱前缀 == handle」那种猜测。审计域里有一处正是这么猜的,而它猜错 的代价在那边只是一条对不上的 MFA 记录,在这里却是「把某个人踢出 GitHub 组织」—— 不能用同一个精度做两件严重性差这么远的事。 handle 大小写不敏感;``github_id`` 列里存的是 HRIS 原样值,所以两边 都 casefold 再比。 """ target = (github_login or "").strip().lower() if not target: return None q = select(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), func.lower(func.trim(IdentityPerson.github_id)) == target, ) return (await db.execute(q)).scalars().first() async def count_leavers_still_in_orgs(db: AsyncSession) -> int: """已离职 / 待离职、却仍在某个**不豁免**组织里的成员资格条数。 这是整套东西承诺的反面验证:自动化的全部意义就是这个数应该是 0。 豁免的社区组织不算 —— 那里的成员本来就允许离职后保留,把它算进来 只会训练人忽略这个数。 """ from app.identity.person_models import IdentityPerson as _P return int( ( await db.execute( select(func.count()) .select_from(GithubOrgMembership) .join(GithubOrg, GithubOrg.id == GithubOrgMembership.org_id) .join(_P, _P.id == GithubOrgMembership.person_id) .where( GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES), GithubOrg.deleted_at.is_(None), GithubOrg.offboarding_policy != const.OFFBOARDING_POLICY_EXEMPT, _P.employment_status.in_(("left", "leaver_pending")), ) ) ).scalar_one() ) async def count_unlinked(db: AsyncSession) -> int: """还没认出是谁的条数。**不含**已被人确认为外部人员的行 —— 见 :func:`confirm_external` 关于「一个永远不归零的数等于没有数」。""" return int( ( await db.execute( select(func.count()) .select_from(GithubOrgMembership) .where( GithubOrgMembership.person_id.is_(None), GithubOrgMembership.external_confirmed_at.is_(None), GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES), ) ) ).scalar_one() ) async def count_confirmed_external(db: AsyncSession) -> int: """已被人确认为外部人员的条数。单独报因为它是**进展**指标:它涨说明 有人在清这个队列。""" return int( ( await db.execute( select(func.count()) .select_from(GithubOrgMembership) .where( GithubOrgMembership.person_id.is_(None), GithubOrgMembership.external_confirmed_at.is_not(None), GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES), ) ) ).scalar_one() ) async def confirm_external( db: AsyncSession, membership: GithubOrgMembership, *, actor_person_id: str | None, note: str | None = None, kind: str | None = None, trace_id: str | None = None, ) -> None: """记下「人看过了,这个 handle 不是我们的员工」。 这一步**不动 GitHub**,只动台账。外部协作者留在开源组织里是正常的, 要处理的不是他们的访问权,是「这条记录该不该继续出现在待办里」。 为什么需要这个动作:上线导入后有 331 条认不出是谁,而抽样拉 GitHub 个人资料发现相当一部分 company 写着别家公司、基金会、高校 —— 他们 永远也「认」不出来。没有这个终态,那个数永远不归零,而一个永远不归零 的待办数等于没有待办数。 冲突处理:已经关联到人的记录**拒绝**被标成外部 —— 那两件事互相 矛盾,而静默接受会得到一条既有 person_id 又被当成外部的行,它在 离职回收里的行为取决于哪个查询先跑到。 """ if membership.person_id: raise ConflictError( "This membership is already linked to a person; unlink it first " "if it really belongs to an external contributor", ) if membership.external_confirmed_at is not None: return membership.external_confirmed_at = _now().replace(tzinfo=None) membership.external_confirmed_by = actor_person_id membership.external_note = (note or None) and note.strip()[:255] # 「是哪一种外部」。不认的值一律落 None 而不是原样存下来 —— 一个拼错的 # 'Bot' 会让按 kind 的分组统计凭空多出一类,而那类在界面上没有任何解释。 membership.external_kind = ( kind if kind in const.EXTERNAL_KINDS else None ) await add_event( db, membership, event_type=const.EVENT_CONFIRMED_EXTERNAL, actor_person_id=actor_person_id, actor_kind="person", detail={"note": membership.external_note, "kind": membership.external_kind}, trace_id=trace_id, ) async def clear_external( db: AsyncSession, membership: GithubOrgMembership, *, actor_person_id: str | None, trace_id: str | None = None, ) -> None: """撤销「确认是外部人员」—— 人认错了,把它放回待办队列。 必须能撤销,而且必须留痕:一条被误标成外部的成员资格,后果是这个人 离职时不会被回收。那是这套东西最不想发生的事,所以它至少要查得出来 是谁在什么时候标的。 """ if membership.external_confirmed_at is None: return membership.external_confirmed_at = None membership.external_confirmed_by = None membership.external_note = None membership.external_kind = None await add_event( db, membership, event_type=const.EVENT_EXTERNAL_CLEARED, actor_person_id=actor_person_id, actor_kind="person", detail=None, trace_id=trace_id, )