zed/script/github-community-pr-board.py

#!/usr/bin/env python3
"""
Route community PRs to the correct review track on a GitHub Project board,
and surface signal fields (Size, Issue Linked, Contributor, Upvotes) that the
board's views can filter and sort on.

Reads the event payload dispatched by the GitHub Actions workflow and:
- On `labeled` or `reopened`: adds the PR to the board (idempotent), sets the
  Track field to the most specific matching track, and recomputes signal fields.
- On `unlabeled`: re-resolves Track from remaining labels, or clears it
  if no area/platform labels remain (PR stays on the board for visibility).
  Signals are recomputed if the PR is on the board.
- On `edited`: recomputes signals if the PR is on the board (a body edit
  can change which issues the PR closes).
- On `assigned`: if the assignee is a staff team member, sets Status to
  "In Progress (us)".
- On `review_requested`: if current Status is "In Progress (author)",
  flips it to "In Progress (us)" — the author is explicitly asking for
  re-review.
- On `issue_comment.created`: if the commenter is the PR author and
  current Status is "In Progress (author)", flips it to
  "In Progress (us)" — the author is likely signaling they're done.
- On `workflow_dispatch` with a PR number: re-resolves Track and
  recomputes signals for a manually specified PR.
- When run with REFRESH_ALL=1: walks every open PR on the board and
  recomputes its signals. Used by the scheduled refresh workflow as a
  belt-and-suspenders pass against missed webhook events (notably
  intentionally-unsubscribed `synchronize` events that change Size, and
  sidebar edits to linked issues that don't fire a PR webhook).

Review-based status changes (approved → "In Progress (us)", changes
requested → "In Progress (author)") are handled by built-in board
automations, not this script.

Signal fields are written best-effort: if a field or option is missing
from the project, the script logs and skips it rather than failing, so
new fields can be rolled out in the project UI independently of deploys.

Requires:
    requests (pip install requests)

Usage (called by the workflow, not directly):
    python github-community-pr-board.py
"""

import json
import os
import sys
from pathlib import Path

import requests
from github_helpers import (
    add_github_project_item,
    fetch_github_project,
    github_graphql,
    github_rest_api,
    set_github_project_field,
)

REPO_OWNER = "zed-industries"
REPO_NAME = "zed"
STAFF_TEAM_SLUG = "staff"

SKIP_LABELS = {"staff", "bot"}


STATUS_IN_PROGRESS_US = "In Progress (us)"
STATUS_IN_PROGRESS_AUTHOR = (
    "In Progress (author)"  # set by built-in board automation, read by this script
)

MAPPING_PATH = Path(__file__).parent / "community-pr-track-mapping.json"


def sync_track_from_labels(pr, project_number):
    """Sync the PR's Track field on the board with its current labels.

    Returns (project, project_item) if the PR ends up on (or stays on)
    the board, so callers can do further per-item work without a second
    project/item lookup. Returns (None, None) otherwise.
    """
    pr_labels = {label["name"] for label in pr.get("labels", [])}
    track_name = resolve_track(pr_labels, load_mapping())

    project = fetch_github_project(project_number)
    project_item = github_find_project_item(project["id"], pr["node_id"])

    if not track_name:
        if project_item:
            github_clear_field(project, project_item, "Track")
            print(f"No track matched, cleared Track on PR #{pr['number']}")
            return project, project_item
        print(
            f"No track matched for labels on PR #{pr['number']}, not on board, nothing to do"
        )
        return None, None

    print(f"Resolved track: {track_name}")

    if not project_item:
        project_item = add_github_project_item(project["id"], pr["node_id"])
    set_github_project_field(project, project_item, "Track", track_name)
    return project, project_item


def set_progress_status_on_assignment(pr, assignee_login, project_number):
    """Set Status to 'In Progress (us)' when a staff member self-assigns."""
    if not github_is_staff_member(assignee_login):
        print(f"Assignee '{assignee_login}' is not a staff member, skipping")
        return

    project = fetch_github_project(project_number)
    item_id = github_find_project_item(project["id"], pr["node_id"])
    if not item_id:
        print(f"PR #{pr['number']} not on board, skipping assignment status update")
        return

    set_github_project_field(project, item_id, "Status", STATUS_IN_PROGRESS_US)


def return_to_reviewer(pr, project_number, reason):
    """Flip Status from 'In Progress (author)' to 'In Progress (us)'.

    Called when the author signals they're ready for re-review, either
    by re-requesting review or by commenting on the PR.
    """
    project = fetch_github_project(project_number)
    item_id = github_find_project_item(project["id"], pr["node_id"])
    if not item_id:
        print(f"PR #{pr['number']} not on board, skipping")
        return

    current_status = github_get_field_value(item_id, "Status")
    if current_status == STATUS_IN_PROGRESS_AUTHOR:
        print(
            f"{reason}, flipping status from '{current_status}' to '{STATUS_IN_PROGRESS_US}'"
        )
        set_github_project_field(project, item_id, "Status", STATUS_IN_PROGRESS_US)
    else:
        print(f"Current status is '{current_status}', not flipping ({reason})")


def compute_size_bucket(total_changes):
    """Return the size bucket label for a PR's total (additions + deletions).

    Size is computed from raw additions + deletions; we don't try to strip
    generated files.
    """
    buckets = [(30, "XS"), (150, "S"), (400, "M"), (1000, "L")]
    for threshold, name in buckets:
        if total_changes < threshold:
            return name
    return "XL"


def compute_contributor(pr_labels):
    """Return the Contributor value derived from the PR's labels.

    The auto-labeler is responsible for applying `guild`, `community champion`
    and `first contribution` based on the author's membership/history; anything
    else on the board is treated as a returning contributor (PRs from staff/bots
    don't reach this function because they're filtered upstream).

    Guild takes precedence over Champion because current-cohort guild members
    are frequently also community champions, and their guild status is the more
    relevant signal while the cohort is active.
    """
    if "guild" in pr_labels:
        return "Guild"
    if "community champion" in pr_labels:
        return "Champion"
    if "first contribution" in pr_labels:
        return "First PR"
    return "Returning"


def recompute_signals(pr, project, project_item):
    """Recompute Size, Contributor, Issue Linked, and Upvotes on a board item.

    Best-effort: missing fields/options are logged and skipped so the
    project's field set can be rolled out independently.
    """
    pr_labels = {label["name"] for label in pr.get("labels", [])}
    total_changes = pr.get("additions", 0) + pr.get("deletions", 0)
    author_login = (pr.get("user") or {}).get("login")

    issue_type, upvotes = github_pr_issue_signals(pr["node_id"], author_login)

    set_field_optional(project, project_item, "Size", compute_size_bucket(total_changes))
    set_field_optional(project, project_item, "Contributor", compute_contributor(pr_labels))
    set_field_optional(project, project_item, "Issue Linked", issue_type)
    set_number_field_optional(project, project_item, "Upvotes", upvotes)


def refresh_signals_if_on_board(pr, project_number):
    """Recompute signals for a PR only if it's already on the board."""
    project = fetch_github_project(project_number)
    project_item = github_find_project_item(project["id"], pr["node_id"])
    if not project_item:
        print(f"PR #{pr['number']} not on board, skipping signal refresh")
        return
    recompute_signals(pr, project, project_item)


def refresh_all_board_items(project_number):
    """Walk every open PR on the board and recompute its signals.

    Backstop for the daily refresh workflow. Individual PR failures are
    logged and don't abort the rest of the run.
    """
    project = fetch_github_project(project_number)
    processed = skipped = errors = 0
    for item in github_list_project_items(project["id"]):
        if item.get("isArchived"):
            skipped += 1
            continue
        content = item.get("content") or {}
        if content.get("__typename") != "PullRequest":
            skipped += 1
            continue
        if content.get("state") != "OPEN" or content.get("isDraft"):
            skipped += 1
            continue
        label_names = [n["name"] for n in content["labels"]["nodes"]]
        if set(label_names) & SKIP_LABELS:
            skipped += 1
            continue
        pr = {
            "number": content["number"],
            "node_id": content["id"],
            "labels": [{"name": n} for n in label_names],
            "additions": content["additions"],
            "deletions": content["deletions"],
            "user": {"login": (content.get("author") or {}).get("login")},
        }
        print(f"--- Refreshing PR #{pr['number']} ---")
        try:
            recompute_signals(pr, project, item["id"])
            processed += 1
        except Exception as exc:
            print(f"Failed to refresh PR #{pr['number']}: {exc}")
            errors += 1
    print(
        f"Refresh complete: {processed} processed, {skipped} skipped, {errors} errors"
    )


def load_mapping(path=MAPPING_PATH):
    """Load the Track-to-labels mapping from the JSON file."""
    with open(path) as f:
        data = json.load(f)
    return data["tracks"]


def resolve_track(pr_labels, tracks):
    """Return the name of the most specific track matching the PR's labels.

    Tracks are checked in order; the first match wins (most specific first).
    """
    for track in tracks:
        if pr_labels & set(track["labels"]):
            return track["name"]
    return None


def github_is_staff_member(username):
    """Check if a user is a member of the staff team."""
    try:
        github_rest_api(
            "GET", f"orgs/{REPO_OWNER}/teams/{STAFF_TEAM_SLUG}/members/{username}"
        )
        return True
    except requests.HTTPError as error:
        if error.response.status_code == 404:
            return False
        print(f"Warning: failed to check staff membership for '{username}': {error}")
        return False
    except requests.RequestException as error:
        print(f"Warning: failed to check staff membership for '{username}': {error}")
        return False


def github_fetch_pr(pr_number):
    """Fetch a PR by number via the REST API."""
    return github_rest_api("GET", f"repos/{REPO_OWNER}/{REPO_NAME}/pulls/{pr_number}")



def github_list_project_items(project_id):
    """Yield every item on a project board, paginating as needed.

    Each item includes `isArchived` so callers can cheaply filter archived
    items out without an extra round-trip per item.
    """
    cursor = None
    while True:
        data = github_graphql(
            """
            query($projectId: ID!, $cursor: String) {
              node(id: $projectId) {
                ... on ProjectV2 {
                  items(first: 100, after: $cursor) {
                    pageInfo { hasNextPage endCursor }
                    nodes {
                      id
                      isArchived
                      content {
                        __typename
                        ... on PullRequest {
                          id
                          number
                          state
                          isDraft
                          additions
                          deletions
                          author { login }
                          labels(first: 50) { nodes { name } }
                        }
                      }
                    }
                  }
                }
              }
            }
            """,
            {"projectId": project_id, "cursor": cursor},
        )
        page = data["node"]["items"]
        for item in page["nodes"]:
            yield item
        if not page["pageInfo"]["hasNextPage"]:
            return
        cursor = page["pageInfo"]["endCursor"]


POSITIVE_REACTIONS = (
    ("thumbsUp", "THUMBS_UP"),
    ("heart", "HEART"),
    ("hooray", "HOORAY"),
    ("rocket", "ROCKET"),
)


def github_reaction_page(subject_node_id, content, cursor):
    data = github_graphql(
        """
        query($subjectId: ID!, $content: ReactionContent!, $cursor: String!) {
          node(id: $subjectId) {
            ... on Reactable {
              reactions(first: 100, content: $content, after: $cursor) {
                nodes { user { login } }
                pageInfo { hasNextPage endCursor }
              }
            }
          }
        }
        """,
        {"subjectId": subject_node_id, "content": content, "cursor": cursor},
    )
    return data["node"]["reactions"]


def add_positive_reactors(subject, author_login, reactors):
    for connection_name, content in POSITIVE_REACTIONS:
        reactions = subject[connection_name]
        while True:
            for reaction in reactions["nodes"]:
                user = reaction.get("user")
                login = user.get("login") if user else None
                if login and login != author_login:
                    reactors.add(login)

            page_info = reactions["pageInfo"]
            if not page_info["hasNextPage"]:
                break
            reactions = github_reaction_page(
                subject["id"], content, page_info["endCursor"]
            )


def github_pr_issue_signals(pr_node_id, author_login):
    """Return linked issue type and deduplicated upvotes for a PR.

    Includes reactions on closing issues and excludes the PR author.
    """
    data = github_graphql(
        """
        fragment PositiveReactions on Reactable {
          id
          thumbsUp: reactions(first: 100, content: THUMBS_UP) {
            nodes { user { login } }
            pageInfo { hasNextPage endCursor }
          }
          heart: reactions(first: 100, content: HEART) {
            nodes { user { login } }
            pageInfo { hasNextPage endCursor }
          }
          hooray: reactions(first: 100, content: HOORAY) {
            nodes { user { login } }
            pageInfo { hasNextPage endCursor }
          }
          rocket: reactions(first: 100, content: ROCKET) {
            nodes { user { login } }
            pageInfo { hasNextPage endCursor }
          }
        }

        query($prId: ID!) {
          node(id: $prId) {
            ... on PullRequest {
              ...PositiveReactions
              closingIssuesReferences(first: 20) {
                totalCount
                nodes {
                  issueType { name }
                  ...PositiveReactions
                }
              }
            }
          }
        }
        """,
        {"prId": pr_node_id},
    )
    pr = data["node"]
    refs = pr["closingIssuesReferences"]

    reactors = set()
    add_positive_reactors(pr, author_login, reactors)
    for issue in refs["nodes"]:
        add_positive_reactors(issue, author_login, reactors)

    if refs["totalCount"] == 0:
        return "No issue", len(reactors)

    type_names = {
        issue["issueType"]["name"]
        for issue in refs["nodes"]
        if issue.get("issueType")
    }
    for priority in ("Crash", "Bug", "Feature", "Docs"):
        if priority in type_names:
            return priority, len(reactors)

    print(
        f"Warning: PR has {refs['totalCount']} linked issue(s) but none with "
        f"a recognized type (saw: {type_names or 'no type set'})"
    )
    return "No issue", len(reactors)


def github_find_project_item(project_id, content_node_id):
    """Find a PR's item ID on the project board, or None if not present.

    Uses a read-only query so it won't add the PR as a side effect.
    """
    data = github_graphql(
        """
        query($contentId: ID!) {
          node(id: $contentId) {
            ... on PullRequest {
              projectItems(first: 50) {
                nodes {
                  id
                  project { id }
                }
              }
            }
          }
        }
        """,
        {"contentId": content_node_id},
    )
    for item in data["node"]["projectItems"]["nodes"]:
        if item["project"]["id"] == project_id:
            return item["id"]
    return None



def set_field_optional(project, item_id, field_name, option_name):
    """Set a single-select field, logging and skipping if the field or
    option doesn't exist on the project yet.

    Used for signal fields so they can be added in the project UI
    independently of script deploys without crashing the workflow.
    """
    field = None
    for f in project["fields"]["nodes"]:
        if f.get("name") == field_name:
            field = f
            break
    if not field:
        print(f"Field '{field_name}' not on project, skipping")
        return
    option_id = None
    for opt in field.get("options", []):
        if opt["name"] == option_name:
            option_id = opt["id"]
            break
    if not option_id:
        available = [opt["name"] for opt in field.get("options", [])]
        print(
            f"Option '{option_name}' not in field '{field_name}' "
            f"(available: {available}), skipping"
        )
        return
    github_graphql(
        """
        mutation($projectId: ID!, $itemId: ID!, $fieldId: ID!, $optionId: String!) {
          updateProjectV2ItemFieldValue(input: {
            projectId: $projectId
            itemId: $itemId
            fieldId: $fieldId
            value: { singleSelectOptionId: $optionId }
          }) {
            projectV2Item { id }
          }
        }
        """,
        {
            "projectId": project["id"],
            "itemId": item_id,
            "fieldId": field["id"],
            "optionId": option_id,
        },
    )
    print(f"Set '{field_name}' to '{option_name}'")


def set_number_field_optional(project, item_id, field_name, value):
    """Set a number field, logging and skipping if the field doesn't exist
    on the project yet. Mirror of `set_field_optional` for non-select fields.
    """
    field = None
    for f in project["fields"]["nodes"]:
        if f.get("name") == field_name:
            field = f
            break
    if not field:
        print(f"Field '{field_name}' not on project, skipping")
        return
    github_graphql(
        """
        mutation($projectId: ID!, $itemId: ID!, $fieldId: ID!, $value: Float!) {
          updateProjectV2ItemFieldValue(input: {
            projectId: $projectId
            itemId: $itemId
            fieldId: $fieldId
            value: { number: $value }
          }) {
            projectV2Item { id }
          }
        }
        """,
        {
            "projectId": project["id"],
            "itemId": item_id,
            "fieldId": field["id"],
            "value": value,
        },
    )
    print(f"Set '{field_name}' to {value}")


def github_clear_field(project, item_id, field_name):
    """Clear a single-select field on a project item."""
    field_id = None
    for field in project["fields"]["nodes"]:
        if field.get("name") == field_name:
            field_id = field["id"]
            break

    if not field_id:
        available = [f["name"] for f in project["fields"]["nodes"] if "name" in f]
        raise RuntimeError(
            f"Field '{field_name}' not found on project. Available: {available}"
        )

    github_graphql(
        """
        mutation($projectId: ID!, $itemId: ID!, $fieldId: ID!) {
          clearProjectV2ItemFieldValue(input: {
            projectId: $projectId
            itemId: $itemId
            fieldId: $fieldId
          }) {
            projectV2Item { id }
          }
        }
        """,
        {
            "projectId": project["id"],
            "itemId": item_id,
            "fieldId": field_id,
        },
    )
    print(f"Cleared '{field_name}'")


def github_get_field_value(item_id, field_name):
    """Read the current value of a single-select field on a project item."""
    data = github_graphql(
        """
        query($itemId: ID!) {
          node(id: $itemId) {
            ... on ProjectV2Item {
              fieldValues(first: 20) {
                nodes {
                  ... on ProjectV2ItemFieldSingleSelectValue {
                    field { ... on ProjectV2SingleSelectField { name } }
                    name
                  }
                }
              }
            }
          }
        }
        """,
        {"itemId": item_id},
    )
    for field_value in data["node"]["fieldValues"]["nodes"]:
        if field_value.get("field", {}).get("name") == field_name:
            return field_value.get("name")
    return None


if __name__ == "__main__":

    project_number = int(os.environ["PROJECT_NUMBER"])

    if os.environ.get("REFRESH_ALL") == "1":
        refresh_all_board_items(project_number)
        sys.exit(0)

    manual_pr_number = os.environ.get("MANUAL_PR_NUMBER")

    if manual_pr_number:
        pr = github_fetch_pr(manual_pr_number)
        action = "labeled"
        event = {}
        print(f"Manual dispatch for PR #{manual_pr_number}")
    else:
        event_name = os.environ["GITHUB_EVENT_NAME"]
        with open(os.environ["GITHUB_EVENT_PATH"]) as f:
            event = json.load(f)

        if event_name in ("pull_request", "pull_request_target"):
            pr = event["pull_request"]
            action = event["action"]
        elif event_name == "issue_comment":
            issue = event["issue"]
            if "pull_request" not in issue:
                print("Comment is on an issue, not a PR, skipping")
                sys.exit(0)
            commenter = event["comment"]["user"]["login"]
            pr_author = issue["user"]["login"]
            if commenter != pr_author:
                print(
                    f"Commenter '{commenter}' is not PR author '{pr_author}', skipping"
                )
                sys.exit(0)
            pr = github_fetch_pr(issue["number"])
            action = "author_commented"
        else:
            print(f"Unexpected event: {event_name}")
            sys.exit(0)

    pr_labels = {label["name"] for label in pr.get("labels", [])}
    if pr_labels & SKIP_LABELS:
        print(f"Skipping PR #{pr['number']} (has {pr_labels & SKIP_LABELS} label)")
        sys.exit(0)

    if pr.get("draft"):
        print(f"Skipping draft PR #{pr['number']}")
        sys.exit(0)

    if pr.get("state") != "open":
        print(f"Skipping PR #{pr['number']} (state: {pr.get('state')})")
        sys.exit(0)

    print(f"Processing PR #{pr['number']}: action={action}")

    if action in ("labeled", "unlabeled", "reopened"):
        project, project_item = sync_track_from_labels(pr, project_number)
        if project_item:
            recompute_signals(pr, project, project_item)
    elif action == "assigned":
        assignee_login = event.get("assignee", {}).get("login")
        if not assignee_login:
            print("No assignee login in event payload, skipping")
        else:
            set_progress_status_on_assignment(pr, assignee_login, project_number)
    elif action == "review_requested":
        return_to_reviewer(pr, project_number, "Author re-requested review")
    elif action == "author_commented":
        return_to_reviewer(pr, project_number, "Author commented on PR")
    elif action == "edited":
        refresh_signals_if_on_board(pr, project_number)
    else:
        print(f"Ignoring action: {action}")