"""Keeping the approval-flow register in step with Feishu. Feishu has **no endpoint that lists approval definitions** — verified against the docs and live against this tenant (2026-08-26). The only definition endpoint takes a code you already have. So the register is built from the one thing the API will enumerate: approval *instances*, queried per person. Three sweeps, run in that order by :mod:`app.scheduled_jobs.approval_registry_handlers`: 1. **Discovery** — ``POST /approval/v4/instances/query`` for every active employee, collecting the distinct definitions behind their instances. This is also the only place group names come from; the same field on ``tasks/search`` is unreliable (measured: 13 of 25 definitions returned more than one group name, 12 of them disagreeing with the instance API outright). 2. **Definition sync** — ``GET /approval/v4/approvals/{code} ?with_admin_id=true`` per known flow for the upstream name, status **and 流程管理员**, so a rename, a stop or a change of manager shows up on the all-staff page without anyone re-running discovery. Subscription is explicitly disabled: these are other teams' flows and we have no business receiving their instance events. The ``with_admin_id`` param is load-bearing — see :func:`_fetch_definitions`; without it the admins are simply absent from the response, which this sweep once mistook for their not existing. 3. **Owner inference** — ``POST /approval/v4/tasks/search`` per active employee, counting who actually decides each flow. A dominant, well-evidenced approver becomes a *candidate* owner marked 待确认; anything ambiguous stays 待认领 rather than naming somebody the flow does not belong to. It runs last and yields to a Feishu-declared admin: a statistic must not overwrite a declaration. Every sweep is three-phase — read from the database, materialise, then do the network work with **no transaction open**, then write. A sweep that held its pooled connection across a multi-minute vendor fan-out would come back to a socket TiDB had reclaimed and lose the whole run (the ``cmdb.server_metrics`` 2013 incident, 2026-08-10). """ from __future__ import annotations import asyncio import logging import uuid from collections import defaultdict from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.approval.feishu_client import FeishuApprovalClient from app.core.exceptions import AppException, ValidationError from app.domains.approval_registry import constants as const from app.domains.approval_registry import service as svc from app.domains.approval_registry.models import ApprovalFlow from app.identity.email_lookup import build_person_by_email_index from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config logger = logging.getLogger(__name__) #: Feishu rejects an instance window wider than this. MAX_WINDOW_DAYS = 30 def _now() -> datetime: return datetime.now(UTC) def _naive(dt: datetime) -> datetime: return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt def _ms_to_datetime(raw: Any) -> datetime | None: """Feishu hands epoch milliseconds back as a *string*.""" try: value = int(str(raw)) except (TypeError, ValueError): return None if value <= 0: return None try: return datetime.fromtimestamp(value / 1000, tz=UTC) except (OverflowError, OSError, ValueError): return None async def _client(db: AsyncSession) -> FeishuApprovalClient: cfg = await get_feishu_config(session=db) return FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) # ════════════════════════════════════════════════════════════════════ # Phase 1 — the roster # ════════════════════════════════════════════════════════════════════ async def _active_roster( db: AsyncSession, *, limit: int | None = None, ) -> list[tuple[uuid.UUID, str]]: """``(person_id, work_email)`` for every active employee, as plain tuples — never ORM rows. The caller commits straight after and then spends minutes in Feishu; a lazily-loaded attribute touched during that window would silently check a connection back out of the pool and re-open the transaction this phase exists to close. """ stmt = ( select(IdentityPerson.id, IdentityPerson.work_email) .where( IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), IdentityPerson.work_email.is_not(None), ) .order_by(IdentityPerson.work_email.asc()) ) if limit: stmt = stmt.limit(limit) rows = (await db.execute(stmt)).all() return [(r[0], r[1]) for r in rows if r[1]] # ════════════════════════════════════════════════════════════════════ # Phase 2 — the network sweeps (no transaction open) # ════════════════════════════════════════════════════════════════════ async def _bounded_gather( tasks: list[Any], *, concurrency: int, ) -> list[Any]: sem = asyncio.Semaphore(max(1, concurrency)) async def _run(coro_factory): async with sem: return await coro_factory() return await asyncio.gather( *[_run(t) for t in tasks], return_exceptions=True, ) async def _sweep_instances( client: FeishuApprovalClient, user_ids: list[str], *, window_days: int | None, max_pages: int, concurrency: int, ) -> tuple[dict[str, dict[str, Any]], dict[str, Any]]: """Collect ``{approval_code: {...}}`` across the roster. Per-person failures are isolated and counted, never fatal: one person whose Feishu account was deleted mid-sweep must not cost the other 499 people's results (the same rule that keeps a single out-of-scope department from failing the whole contact sync). """ start_ms = end_ms = None if window_days: span = min(int(window_days), MAX_WINDOW_DAYS) now = _now() start_ms = str(int((now - timedelta(days=span)).timestamp() * 1000)) end_ms = str(int(now.timestamp() * 1000)) found: dict[str, dict[str, Any]] = {} stats = {"queried": 0, "failed": 0, "truncated": 0, "instances": 0} def _factory(uid: str): async def _call(): return uid, await client.query_instances_for_user( uid, max_pages=max_pages, start_time_ms=start_ms, end_time_ms=end_ms, page_size=const.DISCOVERY_PAGE_SIZE, ) return _call results = await _bounded_gather( [_factory(uid) for uid in user_ids], concurrency=concurrency, ) for outcome in results: if isinstance(outcome, BaseException): stats["failed"] += 1 logger.warning( "approval-registry discovery: instance query failed: %s", str(outcome)[:200], ) continue _uid, (items, truncated) = outcome stats["queried"] += 1 if truncated: stats["truncated"] += 1 for item in items: approval = item.get("approval") or {} code = (approval.get("code") or "").strip() name = (approval.get("name") or "").strip() if not code or not name: continue stats["instances"] += 1 entry = found.setdefault( code, { "name": name, "group": None, "instances": 0, "last_instance_at": None, "instance_codes": set(), }, ) entry["name"] = name # Group comes from THIS endpoint only — see the module # docstring for the measurement behind that. group = ((item.get("group") or {}).get("name") or "").strip() if group and not entry["group"]: entry["group"] = group instance = item.get("instance") or {} # The same instance surfaces once per participant, so count # distinct instance codes rather than rows — otherwise a # flow with many approvers looks busier than one with many # applicants purely because more people can see it. inst_code = (instance.get("code") or "").strip() if inst_code: entry["instance_codes"].add(inst_code) started = _ms_to_datetime(instance.get("start_time")) if started and ( entry["last_instance_at"] is None or started > entry["last_instance_at"] ): entry["last_instance_at"] = started for entry in found.values(): entry["instances"] = len(entry.pop("instance_codes")) return found, stats async def _sweep_tasks( client: FeishuApprovalClient, roster: list[tuple[uuid.UUID, str]], email_to_uid: dict[str, str], *, max_pages: int, concurrency: int, ) -> tuple[dict[str, dict[uuid.UUID, int]], dict[str, Any]]: """Count, per flow, how many approval tasks each person decided. Returns ``{approval_code: {person_id: task_count}}``. Only tasks inside :data:`constants.INFERENCE_WINDOW_DAYS` count: somebody who ran a flow two years ago and moved on should not keep it. """ uid_to_person: dict[str, uuid.UUID] = {} for person_id, email in roster: uid = email_to_uid.get((email or "").lower()) if uid: uid_to_person.setdefault(uid, person_id) cutoff = _now() - timedelta(days=const.INFERENCE_WINDOW_DAYS) counts: dict[str, dict[uuid.UUID, int]] = defaultdict(lambda: defaultdict(int)) stats = {"queried": 0, "failed": 0, "truncated": 0, "tasks": 0} def _factory(uid: str): async def _call(): return uid, await client.search_tasks_for_user( uid, max_pages=max_pages, page_size=const.DISCOVERY_PAGE_SIZE, ) return _call results = await _bounded_gather( [_factory(uid) for uid in uid_to_person], concurrency=concurrency, ) for outcome in results: if isinstance(outcome, BaseException): stats["failed"] += 1 logger.warning( "approval-registry inference: task search failed: %s", str(outcome)[:200], ) continue uid, (items, truncated) = outcome stats["queried"] += 1 if truncated: stats["truncated"] += 1 approver_id = uid_to_person.get(uid) if approver_id is None: continue for item in items: code = ((item.get("approval") or {}).get("code") or "").strip() if not code: continue task = item.get("task") or {} # ``status`` on a task is PENDING / APPROVED / REJECTED / # TRANSFERRED / DONE. A pending task says the flow was # routed to them, not that they decided anything — and a # single stale pending task would otherwise be enough to # hand somebody a flow. if str(task.get("status") or "").upper() == "PENDING": continue decided = _ms_to_datetime(task.get("end_time")) or _ms_to_datetime( task.get("start_time"), ) if decided is not None and decided < cutoff: continue stats["tasks"] += 1 counts[code][approver_id] += 1 return {k: dict(v) for k, v in counts.items()}, stats async def _fetch_definitions( client: FeishuApprovalClient, codes: list[str], *, concurrency: int, ) -> dict[str, dict[str, Any]]: """``{code: {"name":…, "status":…, "admin_ids": [...]}}`` or ``{"error": "..."}``. ``subscribe=False``: these are other teams' approval definitions. Subscribing would redirect their instance events at Helix's webhook — thousands of payloads a day that no handler consumes. ``with_admin_id=True`` is not optional here and the reason is worth keeping in the code: without it the response has no ``approval_admin_ids`` key at all, and this sweep read that absence as 「飞书没有流程管理员」 for a whole release while the console showed a name on 42 of 49 definitions. """ out: dict[str, dict[str, Any]] = {} def _factory(code: str): async def _call(): try: data = await client.get_approval_definition( code, subscribe=False, with_admin_id=True, user_id_type="user_id", ) raw_admins = data.get("approval_admin_ids") return code, { "name": (data.get("approval_name") or "").strip() or None, "status": (data.get("status") or "").strip() or None, "admin_ids": [ a.strip() for a in (raw_admins or []) if isinstance(a, str) and a.strip() ], } except AppException as exc: return code, {"error": exc.message} except Exception as exc: # noqa: BLE001 — one flow must not fail the sweep return code, {"error": str(exc)[:200]} return _call results = await _bounded_gather( [_factory(c) for c in codes], concurrency=concurrency, ) for outcome in results: if isinstance(outcome, BaseException): continue code, payload = outcome out[code] = payload return out def _email_variants(email: str) -> list[str]: """``x@pingcap.com`` and ``x@pingcap.com.cn`` are the same person. Feishu accounts in this tenant are registered under either domain while ``identity_persons.work_email`` holds one of them — an exact-match-only lookup drops roughly a quarter of the approval admins on the floor (measured: 8 of 37, including the sole admin of two flows). """ lowered = email.strip().lower() if not lowered: return [] local, _, domain = lowered.partition("@") out = [lowered] sibling = { "pingcap.com": "pingcap.com.cn", "pingcap.com.cn": "pingcap.com", }.get(domain) if sibling and local: out.append(f"{local}@{sibling}") return out def _pick_admin_owner( flow: ApprovalFlow, admin_ids: list[str], people: dict[str, IdentityPerson], ) -> IdentityPerson | None: """Which 流程管理员 answers for this flow. ``None`` when none can. Every flow Feishu names an administrator for gets ONE owner, including the 25 here whose admin list is a whole team (user decision, 2026-08-27). A team is not an owner in the strict sense — but an empty owner column means nobody gets the half-year card and nobody is blocked at offboarding, and one named person who can answer 「不是我」 beats nobody to ask. **The pick must not come from Feishu's array order.** ``approval_admin_ids`` is rotated on every call: five reads of 名片申请 (7 admins) returned four different orders and four different heads. Taking ``admin_ids[0]`` would hand the flow to a different colleague most nights, log a transfer each time, and send December's confirmation card to whoever won that morning's roll. So: candidates are the admins who both resolve to a Helix person and still work here, ordered by ``user_id`` — arbitrary, but the same arbitrary answer every night, which is the only property that matters. And the incumbent wins: once a ``feishu_admin`` owner is on the flow, they keep it while they remain on the list and employed, so a later change to the Feishu list cannot silently re-point a flow somebody has already been answering for. """ candidates = [ (uid, people[uid]) for uid in sorted(set(admin_ids)) if uid in people and people[uid].employment_status == "active" and people[uid].deleted_at is None ] if not candidates: return None if flow.owner_source == const.OWNER_SOURCE_FEISHU_ADMIN: incumbent = next( (p for _uid, p in candidates if p.id == flow.owner_person_id), None, ) if incumbent is not None: return incumbent return candidates[0][1] async def _resolve_admins( db: AsyncSession, profiles: dict[str, dict[str, Any]], ) -> tuple[dict[str, dict[str, Any]], dict[str, IdentityPerson]]: """``({user_id: snapshot}, {user_id: IdentityPerson})``. Every admin Feishu named is kept in the snapshot map, matched or not. An admin with no ``identity_persons`` row still answers 「找谁」 by name, and dropping them would silently shrink the published answer back toward 「还没人认领」 — the exact failure this whole change exists to undo. The second map holds only the ones that matched, for the sole-admin adoption. """ candidates: list[str] = [] for profile in profiles.values(): candidates.extend(_email_variants(profile.get("email") or "")) index = ( await build_person_by_email_index(db, candidates) if candidates else {} ) resolved: dict[str, dict[str, Any]] = {} matched: dict[str, IdentityPerson] = {} for user_id, profile in profiles.items(): person = None for variant in _email_variants(profile.get("email") or ""): person = index.get(variant) if person is not None: break if person is not None: matched[user_id] = person resolved[user_id] = { "user_id": user_id, # Prefer Helix's display name when we matched a person: the # register renders alongside Helix people everywhere else, # and two spellings of one colleague reads as two colleagues. "name": ( (person.display_name or person.work_email) if person is not None else profile.get("name") ), "email": ( person.work_email if person is not None else profile.get("email") ), "person_id": str(person.id) if person is not None else None, # Employment truth comes from Helix when we have it, and # from Feishu's resignation flag when we do not. A departed # 流程管理员 is a governance hole the page must be able to # show as one rather than publishing them as the answer. "active": ( ( person.employment_status == "active" and person.deleted_at is None ) if person is not None else not profile.get("resigned") ), } return resolved, matched # ════════════════════════════════════════════════════════════════════ # The orchestrated run # ════════════════════════════════════════════════════════════════════ async def run_sync( db: AsyncSession, *, window_days: int | None = None, person_limit: int | None = None, max_pages: int = const.DISCOVERY_MAX_PAGES, concurrency: int = const.DISCOVERY_CONCURRENCY, discover: bool = True, sync_definitions: bool = True, infer_owners: bool = True, ) -> dict[str, Any]: """Discovery → definition sync → owner inference, in one run. ``window_days=None`` sweeps a person's whole visible history (the instance API accepts a query with no window and returns everything it can see); pass a number ≤30 for the incremental nightly pass. Feishu rejects anything wider than 30 days, and a half-open window outright. """ summary: dict[str, Any] = { "roster": 0, "feishu_resolved": 0, "feishu_unresolved": 0, "flows_discovered": 0, "flows_created": 0, "flows_seen": 0, "definitions_synced": 0, "definitions_failed": 0, # 流程管理员, read from the definition with ``with_admin_id=true``. # ``admin_ids_unresolved`` is a real signal, not noise: a Feishu # user_id Feishu itself rejects is an admin whose account is # gone, and the flow they were answering for now has nobody. # ``admin_ids_lookup_failed`` is the opposite — our call broke, # the person is fine — and the flows it touches keep the admin # list they already had (``admins_kept_stale``). "admin_ids_seen": 0, "admin_ids_unresolved": 0, "admin_ids_lookup_failed": 0, "admins_kept_stale": 0, "flows_with_admins": 0, "admins_changed": 0, "owners_from_feishu_admin": 0, "owners_inferred": 0, "inference_candidates": 0, "instances_sampled": 0, "tasks_sampled": 0, "people_failed": 0, "people_truncated": 0, # Definitions the sweep saw but could not record — reported # rather than silently absent, because "we found 40 flows" and # "we found 41 and dropped one" are different statements. "flows_rejected": 0, } # ── Phase A: read the roster, then let the connection go ───────── roster = await _active_roster(db, limit=person_limit) summary["roster"] = len(roster) if not roster: return summary client = await _client(db) await db.commit() # ── Phase B: network only ──────────────────────────────────────── email_to_uid = await client.batch_resolve_user_ids_by_email( [email for _pid, email in roster], ) summary["feishu_resolved"] = len(email_to_uid) summary["feishu_unresolved"] = len(roster) - len(email_to_uid) user_ids = sorted(set(email_to_uid.values())) discovered: dict[str, dict[str, Any]] = {} if discover and user_ids: discovered, inst_stats = await _sweep_instances( client, user_ids, window_days=window_days, max_pages=max_pages, concurrency=concurrency, ) summary["flows_discovered"] = len(discovered) summary["instances_sampled"] = inst_stats["instances"] summary["people_failed"] += inst_stats["failed"] summary["people_truncated"] += inst_stats["truncated"] # ── Phase C: write what discovery found ────────────────────────── if discovered: created = 0 for code, entry in discovered.items(): # Per-flow isolation, the same rule the per-person fan-out # follows: one definition whose code or name the platform # will not accept must not cost the other few hundred their # nightly refresh. Before this, a single malformed code took # the whole sweep red and nothing was recorded at all. try: _flow, was_created = await svc.upsert_discovered( db, approval_code=code, name=entry["name"], source_group=entry["group"], observed_instances=entry["instances"], last_instance_at=( _naive(entry["last_instance_at"]) if entry["last_instance_at"] else None ), ) except ValidationError: summary["flows_rejected"] += 1 logger.warning( "approval-registry discovery: skipping unusable " "definition (code rejected by validation)", ) continue created += 1 if was_created else 0 summary["flows_created"] = created await db.commit() # ── Phase D: definition sync (read codes → network → write) ────── if sync_definitions: codes = [ row[0] for row in ( await db.execute( select(ApprovalFlow.approval_code).where( ApprovalFlow.deleted_at.is_(None), ), ) ).all() ] summary["flows_seen"] = len(codes) await db.commit() if codes: definitions = await _fetch_definitions( client, codes, concurrency=concurrency, ) # Still on the network, still with no transaction open: one # contact lookup for every DISTINCT admin across all # definitions, not one per definition. This tenant's 49 # flows name 37 distinct people between them, and several # admins sit on five or six flows each. admin_ids = sorted({ a for payload in definitions.values() for a in (payload.get("admin_ids") or []) }) admin_profiles, admin_lookup_failed = ( await client.batch_get_user_profiles(admin_ids) if admin_ids else ({}, set()) ) summary["admin_ids_seen"] = len(admin_ids) summary["admin_ids_lookup_failed"] = len(admin_lookup_failed) # Genuinely gone, NOT merely un-looked-up. The two were the # same number until a live measurement caught one id # resolving on one pass and erroring on the next, which # flipped a flow from 「有唯一管理员」 to 「一个人都没有」 # between two runs minutes apart. summary["admin_ids_unresolved"] = len( [a for a in admin_ids if a not in admin_profiles and a not in admin_lookup_failed], ) # One read for the whole batch rather than one per # definition: the sweep already fans out over hundreds of # rows, and a per-row SELECT here doubles its DB round-trips # for no benefit. known = { f.approval_code: f for f in await svc.flows_by_codes(db, list(definitions)) } resolved_admins, admin_people = await _resolve_admins( db, admin_profiles, ) for code, payload in definitions.items(): # Normalised on the way in, because ``flows_by_codes`` # keys its result that way. These particular codes come # straight off the column and so are already normal — # but a lookup that only works because two functions # happen to agree about casing is one refactor away from # silently skipping every definition. flow = known.get(svc.normalize_code(code)) if flow is None: continue if payload.get("error"): summary["definitions_failed"] += 1 await svc.apply_definition( db, flow, name=None, feishu_status=None, error=str(payload["error"]), ) # Deliberately NOT clearing ``feishu_admins``: a # definition that failed to read this time still had # the admins it had yesterday, and blanking them # would put 「还没人认领」 back on the page every time # Feishu rate-limited one call. continue summary["definitions_synced"] += 1 await svc.apply_definition( db, flow, name=payload.get("name"), feishu_status=payload.get("status"), ) admin_ids_here = payload.get("admin_ids") or [] # A transient contact-API failure on ANY of this flow's # admins makes the whole list untrustworthy: writing the # survivors would silently delete a live administrator's # name from the all-staff page and log a false # 「管理员变更」 on the flow's trail. Same rule as the # failed definition read above — keep yesterday's answer # and try again tomorrow. if any(a in admin_lookup_failed for a in admin_ids_here): summary["admins_kept_stale"] += 1 continue admins = [ resolved_admins[a] for a in admin_ids_here if a in resolved_admins ] if await svc.apply_feishu_admins(db, flow, admins=admins): summary["admins_changed"] += 1 if admins: summary["flows_with_admins"] += 1 pick = _pick_admin_owner(flow, admin_ids_here, admin_people) if pick is not None and await svc.apply_feishu_admin_owner( db, flow, person=pick, ): summary["owners_from_feishu_admin"] += 1 await db.commit() # ── Phase E: owner inference (network → write) ─────────────────── if infer_owners and user_ids: task_counts, task_stats = await _sweep_tasks( client, roster, email_to_uid, max_pages=max_pages, concurrency=concurrency, ) summary["tasks_sampled"] = task_stats["tasks"] summary["people_failed"] += task_stats["failed"] summary["people_truncated"] += task_stats["truncated"] inferred = await _apply_inference(db, task_counts, summary) summary["owners_inferred"] = inferred await db.commit() return summary async def _apply_inference( db: AsyncSession, task_counts: dict[str, dict[uuid.UUID, int]], summary: dict[str, Any], ) -> int: # Both lookups are batched: one read for every flow the sweep saw, # one for every candidate person, instead of two per code. known = { f.approval_code: f for f in await svc.flows_by_codes(db, list(task_counts)) } candidates = { p.id: p for p in ( await db.execute( select(IdentityPerson) .where( IdentityPerson.id.in_( {pid for counts in task_counts.values() for pid in counts}, ), ) .options(selectinload(IdentityPerson.org_memberships)), ) ).scalars().all() } if task_counts else {} applied = 0 for code, per_person in task_counts.items(): total = sum(per_person.values()) if total < const.MIN_INFERENCE_TASKS: continue top_person_id, top_count = max(per_person.items(), key=lambda kv: kv[1]) share = top_count / total # A flow whose approver is "the applicant's line manager" spreads # its decisions over dozens of people; the busiest of them has no # say over the flow itself. Both gates must hold, or the row # stays 待认领 — which is the honest answer and the one the # confirmation round can fix. if share < const.MIN_INFERENCE_SHARE or top_count < const.MIN_INFERENCE_TASKS: continue summary["inference_candidates"] += 1 try: flow = known.get(svc.normalize_code(code)) except ValidationError: continue if flow is None: continue person = candidates.get(top_person_id) if person is None: continue changed = await svc.apply_inferred_owner( db, flow, person=person, evidence={ "task_count": top_count, "total_tasks": total, "share_pct": round(share * 100, 1), "window_days": const.INFERENCE_WINDOW_DAYS, "distinct_approvers": len(per_person), "inferred_at": _now().isoformat(), }, ) applied += 1 if changed else 0 return applied