OC member queries: cursor pagination via fetch_all_member_nodes() — members(limit:100) silently hid everyone past member #100 (same class as the Cloudron 25-per-page trap). Helper RAISES on any page failure so callers' fail-safes apply (never a partial list — the sweep would deactivate everyone missing from one); fetch_members returns [] on failure (sweep skips), fetch_org_backers propagates (sweep keeps linked accounts active), webhook payer lookup and /admin/org-link wrapped.
This commit is contained in:
1 parent
fed4a7005e
commit
42c664c076
1 file changed
+59
-43
+59
-43
@@ -645,6 +645,44 @@ def compute_max_budget(allowance: float, credit_balance: float, team_spend: floa
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
async def fetch_all_member_nodes() -> list[dict]:
|
||||||
|
"""Every member node of the collective, following OC's cursor pagination.
|
||||||
|
|
||||||
|
The members connection caps at 100 per page: an unpaginated query silently
|
||||||
|
hides everyone past the first page — the same failure class as Cloudron's
|
||||||
|
25-per-page user list. Raises on ANY page failure so callers can apply
|
||||||
|
their own fail-safes; never returns a partial list (a partial member list
|
||||||
|
would make the sweep deactivate everyone missing from it).
|
||||||
|
"""
|
||||||
|
if not OC_PERSONAL_TOKEN:
|
||||||
|
raise RuntimeError("OC_PERSONAL_TOKEN not set; cannot fetch members")
|
||||||
|
nodes: list[dict] = []
|
||||||
|
after: str | None = None
|
||||||
|
async with httpx.AsyncClient() as client:
|
||||||
|
for _ in range(20): # hard cap: 20 pages × 100 = 1000 members
|
||||||
|
after_q = f', after: "{after}"' if after else ""
|
||||||
|
query = (
|
||||||
|
'{ collective(slug: "%s") { members(limit: 100%s) { nodes '
|
||||||
|
'{ role tier { name } account { name slug type ... on Individual { email } } } '
|
||||||
|
'pageInfo { hasNextPage endCursor } } } }'
|
||||||
|
% (OC_COLLECTIVE_SLUG, after_q)
|
||||||
|
)
|
||||||
|
r = await client.post(
|
||||||
|
OC_GRAPHQL_URL,
|
||||||
|
headers={"Personal-Token": OC_PERSONAL_TOKEN},
|
||||||
|
json={"query": query},
|
||||||
|
)
|
||||||
|
if r.status_code != 200:
|
||||||
|
raise RuntimeError(f"OC member fetch failed: HTTP {r.status_code}")
|
||||||
|
data = ((r.json().get("data") or {}).get("collective") or {}).get("members") or {}
|
||||||
|
nodes.extend(data.get("nodes") or [])
|
||||||
|
info = data.get("pageInfo") or {}
|
||||||
|
if not info.get("hasNextPage") or not info.get("endCursor"):
|
||||||
|
return nodes
|
||||||
|
after = info["endCursor"]
|
||||||
|
raise RuntimeError("OC member pagination exceeded 20 pages")
|
||||||
|
|
||||||
|
|
||||||
async def fetch_members() -> list[dict]:
|
async def fetch_members() -> list[dict]:
|
||||||
"""Return the full active-member list: email, name, slug, role, tier.
|
"""Return the full active-member list: email, name, slug, role, tier.
|
||||||
|
|
||||||
@@ -660,20 +698,9 @@ async def fetch_members() -> list[dict]:
|
|||||||
if not OC_PERSONAL_TOKEN:
|
if not OC_PERSONAL_TOKEN:
|
||||||
logger.warning("OC_PERSONAL_TOKEN not set; cannot fetch members")
|
logger.warning("OC_PERSONAL_TOKEN not set; cannot fetch members")
|
||||||
return []
|
return []
|
||||||
query = (
|
# Paginated fetch (raises on failure → caller fail-safes apply: the sweep
|
||||||
'{ collective(slug: "%s") { members(limit: 100) { nodes { role tier { name } account { name slug ... on Individual { email } } } } } }'
|
# skips deactivation entirely when we can't get the full member list).
|
||||||
% OC_COLLECTIVE_SLUG
|
nodes = await fetch_all_member_nodes()
|
||||||
)
|
|
||||||
async with httpx.AsyncClient() as client:
|
|
||||||
r = await client.post(
|
|
||||||
OC_GRAPHQL_URL,
|
|
||||||
headers={"Personal-Token": OC_PERSONAL_TOKEN},
|
|
||||||
json={"query": query},
|
|
||||||
)
|
|
||||||
if r.status_code != 200:
|
|
||||||
logger.warning("OC member fetch failed: %s", r.status_code)
|
|
||||||
return []
|
|
||||||
nodes = r.json().get("data", {}).get("collective", {}).get("members", {}).get("nodes", [])
|
|
||||||
result: list[dict] = []
|
result: list[dict] = []
|
||||||
for n in nodes:
|
for n in nodes:
|
||||||
role = n.get("role", "")
|
role = n.get("role", "")
|
||||||
@@ -706,21 +733,11 @@ async def fetch_org_backers() -> dict[str, dict]:
|
|||||||
org slug: {slug: {name, role, tier}}.
|
org slug: {slug: {name, role, tier}}.
|
||||||
"""
|
"""
|
||||||
if not OC_PERSONAL_TOKEN:
|
if not OC_PERSONAL_TOKEN:
|
||||||
return {}
|
logger.warning("OC_PERSONAL_TOKEN not set; cannot fetch org members")
|
||||||
query = (
|
raise RuntimeError("OC_PERSONAL_TOKEN not set")
|
||||||
'{ collective(slug: "%s") { members(limit: 100) { nodes { role tier { name } account { name slug type } } } } }'
|
# Paginated fetch; RAISES on failure so the sweep's fail-safe keeps linked
|
||||||
% OC_COLLECTIVE_SLUG
|
# accounts active rather than deactivating them on a fetch error.
|
||||||
)
|
nodes = await fetch_all_member_nodes()
|
||||||
async with httpx.AsyncClient() as client:
|
|
||||||
r = await client.post(
|
|
||||||
OC_GRAPHQL_URL,
|
|
||||||
headers={"Personal-Token": OC_PERSONAL_TOKEN},
|
|
||||||
json={"query": query},
|
|
||||||
)
|
|
||||||
if r.status_code != 200:
|
|
||||||
logger.warning("OC org-member fetch failed: %s", r.status_code)
|
|
||||||
return {}
|
|
||||||
nodes = r.json().get("data", {}).get("collective", {}).get("members", {}).get("nodes", [])
|
|
||||||
result: dict[str, dict] = {}
|
result: dict[str, dict] = {}
|
||||||
for n in nodes:
|
for n in nodes:
|
||||||
role = n.get("role", "")
|
role = n.get("role", "")
|
||||||
@@ -1912,19 +1929,12 @@ async def opencollective_webhook(request: Request, token: str):
|
|||||||
# The webhook payload strips emails; look up the payer via the
|
# The webhook payload strips emails; look up the payer via the
|
||||||
# admin token (bot account sees contributor emails).
|
# admin token (bot account sees contributor emails).
|
||||||
try:
|
try:
|
||||||
q = ('{ collective(slug: "%s") { members(limit: 100) { nodes '
|
# Paginated member fetch (raises on failure → caught below)
|
||||||
'{ account { name slug ... on Individual { email } } } } } }' % OC_COLLECTIVE_SLUG)
|
for n in await fetch_all_member_nodes():
|
||||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
if (n.get("account") or {}).get("slug") == slug:
|
||||||
r = await client.post(OC_GRAPHQL_URL,
|
buyer_email = ((n.get("account") or {}).get("email") or "").strip()
|
||||||
headers={"Personal-Token": OC_PERSONAL_TOKEN,
|
buyer_name = (n.get("account") or {}).get("name") or buyer_name
|
||||||
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"},
|
break
|
||||||
json={"query": q})
|
|
||||||
if r.status_code == 200:
|
|
||||||
for n in r.json().get("data", {}).get("collective", {}).get("members", {}).get("nodes", []):
|
|
||||||
if (n.get("account") or {}).get("slug") == slug:
|
|
||||||
buyer_email = (n["account"].get("email") or "").strip()
|
|
||||||
buyer_name = n["account"].get("name") or buyer_name
|
|
||||||
break
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning("Credit-pack hold: payer lookup failed for %s: %s", slug, e)
|
logger.warning("Credit-pack hold: payer lookup failed for %s: %s", slug, e)
|
||||||
|
|
||||||
@@ -2368,7 +2378,13 @@ async def admin_org_link(request: Request):
|
|||||||
if not org_slug or not email or "@" not in email:
|
if not org_slug or not email or "@" not in email:
|
||||||
raise HTTPException(400, "org_slug and a valid email are required")
|
raise HTTPException(400, "org_slug and a valid email are required")
|
||||||
|
|
||||||
org = (await fetch_org_backers()).get(org_slug)
|
try:
|
||||||
|
org = (await fetch_org_backers()).get(org_slug)
|
||||||
|
except Exception as e:
|
||||||
|
# OC unreachable: link without provisioning — the sweep provisions and
|
||||||
|
# activates once OC is reachable again.
|
||||||
|
logger.warning("org-link: OC status fetch failed for %s (linking anyway): %s", org_slug, e)
|
||||||
|
org = None
|
||||||
|
|
||||||
conn = get_db()
|
conn = get_db()
|
||||||
existing = conn.execute("SELECT email FROM members WHERE org_slug = ?", (org_slug,)).fetchone()
|
existing = conn.execute("SELECT email FROM members WHERE org_slug = ?", (org_slug,)).fetchone()
|
||||||
|
|||||||
Reference in new issue
Block a user