path: root/scripts/seed_pearl_workspace.py

#!/usr/bin/env python3
"""Seed a Linear workspace with Pearl's dogfooding conventions.

Brings a workspace to:

  - a "Pearl" team whose columns are the six Agile states, in board order
    (Icebox, Triage, Backlog, In Progress, Done, Canceled),
  - a "Pearl" project inside that team,
  - three Custom Views: Pearl Open Issues, Pearl Icebox, Pearl Inbox.

Re-running is safe: it reconciles against what's already there by name.

How the columns are built is shaped by what Linear actually allows (verified
live, not assumed): teamCreate seeds its own default columns, a team must keep
an unstarted state, the Duplicate state is reserved, and a state's `type` is
immutable -- workflowStateUpdate takes name/color/description/position only.

That last constraint drives the column types. Linear's board orders columns by
type category first (backlog before unstarted), then by position within a
category, so position alone can't move a column across a type boundary. To get
Icebox, Triage, Backlog all left of In Progress in that order, two of them have
to be the backlog type and the third the unstarted type rendered last:

  - Icebox  -> backlog
  - Triage  -> backlog
  - Backlog -> unstarted   (this is the team's required unstarted state)

Because type can't be changed in place, a column found with the wrong type is
*recreated*: the script creates a fresh state with the right type, moves the old
state's issues onto it, and archives the old one. The phases are ordered so a
new unstarted state exists before the old unstarted one is archived, and so a
name is freed before another column reuses it.

Positions are set in a final pass with distinct nonzero values, because Linear
ignores a 0.0 position and appends a freshly-created state at a high one.

Usage:
  LINEAR_API_KEY=lin_api_... python3 seed_pearl_workspace.py [--dry-run]
"""

import argparse
import json
import os
import stat
import sys
import urllib.request

LINEAR_GRAPHQL_URL = "https://api.linear.app/graphql"

TEAM_NAME = "Pearl"
TEAM_KEY = "PEARL"
PROJECT_NAME = "Pearl"

# The six Agile columns in board order. `type` is chosen so Linear's
# type-grouped board renders them in this order (see the module docstring):
# Icebox and Triage are backlog-type, Backlog is the unstarted state. `sources`
# names the Linear default a column reconciles from when no column already
# carries its own name -- on a fresh team Icebox takes over the default Backlog
# and Backlog takes over the unstarted Todo. Positions are distinct and nonzero.
TARGET_STATES = [
    {"name": "Icebox", "type": "backlog", "color": "#bec2c8", "position": 1.0,
     "sources": ["Backlog"]},
    {"name": "Triage", "type": "backlog", "color": "#e2e2e2", "position": 2.0,
     "sources": []},
    {"name": "Backlog", "type": "unstarted", "color": "#95a2b3", "position": 3.0,
     "sources": ["Todo"]},
    {"name": "In Progress", "type": "started", "color": "#f2c94c", "position": 4.0,
     "sources": []},
    {"name": "Done", "type": "completed", "color": "#0f9d58", "position": 5.0,
     "sources": []},
    {"name": "Canceled", "type": "canceled", "color": "#eb5757", "position": 6.0,
     "sources": ["Cancelled"]},
]

# New issues should land in the intake column, not Backlog. The team's
# defaultIssueState is pointed here after the columns reconcile.
INTAKE_STATE = "Triage"

VIEW_NAMES = ["Pearl Open Issues", "Pearl Icebox", "Pearl Inbox"]


def desired_views(project_id):
    """Return the three view specs (name + IssueFilter filterData) for PROJECT_ID."""
    proj = {"project": {"id": {"eq": project_id}}}
    return [
        {"name": "Pearl Open Issues",
         "filterData": {**proj, "state": {"type": {"nin": ["completed", "canceled"]}}}},
        {"name": "Pearl Icebox",
         "filterData": {**proj, "state": {"name": {"eq": "Icebox"}}}},
        {"name": "Pearl Inbox",
         "filterData": {**proj, "state": {"name": {"eq": "Triage"}}}},
    ]


def reconcile_state_plan(existing, targets=None):
    """Decide, per target column, whether to update, recreate, or create it.

    EXISTING is a list of state dicts (id, name, type). Returns a list of
    {target, action, match_id}:

      - "update"   -- a state already carries this column's identity with the
                      right type; rename/reposition it in place. match_id set.
      - "recreate" -- a state matches but has the wrong type (type is immutable),
                      so it must be replaced. match_id is the old state's id.
      - "create"   -- nothing matches; make a new state. match_id is None.

    Matching prefers a type-correct candidate, so a column is only recreated when
    no usable state of the right type exists. The passes:

      1. exact name + correct type  -> update
      2. source alias + correct type -> update (reuse a default, e.g. Todo -> Backlog)
      3. exact name, wrong type      -> recreate (type is immutable)

    This makes the same target definitions fit both a fresh team and the live
    workspace. On a fresh team Icebox reuses the backlog default "Backlog" and
    Backlog reuses the unstarted "Todo" (pass 2). On the live workspace, where a
    correctly-named Icebox already exists and there's no Todo, the wrong-type
    Triage and Backlog fall to pass 3 and get recreated. Pure: no I/O. Each
    existing state is claimed by at most one target.
    """
    targets = targets or TARGET_STATES
    used, result = set(), {}

    def claim(target, predicate, action):
        if target["name"] in result:
            return
        m = next((s for s in existing if s["id"] not in used and predicate(s)), None)
        if m:
            used.add(m["id"])
            result[target["name"]] = (action, m["id"])

    for target in targets:                      # pass 1: exact name + correct type
        claim(target, lambda s, t=target:
              s["name"].lower() == t["name"].lower() and s["type"] == t["type"], "update")
    for target in targets:                      # pass 2: source alias + correct type
        aliases = {s.lower() for s in target.get("sources", [])}
        claim(target, lambda s, t=target, a=aliases:
              s["name"].lower() in a and s["type"] == t["type"], "update")
    for target in targets:                      # pass 3: exact name, wrong type
        claim(target, lambda s, t=target: s["name"].lower() == t["name"].lower(), "recreate")

    plan = []
    for target in targets:
        action, match_id = result.get(target["name"], ("create", None))
        plan.append({"target": target, "action": action, "match_id": match_id})
    return plan


def leftover_state_names(existing, targets=None):
    """Return names of existing states no target column claims (e.g. Duplicate)."""
    claimed = {p["match_id"] for p in reconcile_state_plan(existing, targets) if p["match_id"]}
    return [s["name"] for s in existing if s["id"] not in claimed]


class LinearError(RuntimeError):
    """A GraphQL request returned an errors array."""


def _http_transport(api_key):
    def transport(query, variables):
        body = json.dumps({"query": query, "variables": variables}).encode()
        req = urllib.request.Request(
            LINEAR_GRAPHQL_URL, data=body,
            headers={"Authorization": api_key, "Content-Type": "application/json"})
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read().decode())
    return transport


class LinearClient:
    """Thin Linear GraphQL client; the only network boundary in this module.

    Each method names its GraphQL operation distinctly (Teams, TeamStates,
    TeamCreate, StateCreate, StateUpdate, StateArchive, StateIssues, IssueMove,
    TeamUpdate, Projects, ProjectCreate, CustomViews, CustomViewCreate) so a fake
    transport can route on the operation name.
    """

    def __init__(self, api_key, transport=None):
        self._transport = transport or _http_transport(api_key)

    def execute(self, query, variables):
        resp = self._transport(query, variables)
        if resp.get("errors"):
            raise LinearError("; ".join(e.get("message", str(e)) for e in resp["errors"]))
        return resp["data"]

    def find_team(self):
        data = self.execute(
            "query Teams { teams { nodes { id key name defaultIssueState { id } "
            "states { nodes { id name type position } } } } }", {})
        return next((t for t in data["teams"]["nodes"] if t["name"] == TEAM_NAME), None)

    def team_states(self, team_id):
        data = self.execute(
            "query TeamStates($id: String!) { team(id: $id) "
            "{ states { nodes { id name type position } } } }", {"id": team_id})
        return data["team"]["states"]["nodes"]

    def create_team(self):
        data = self.execute(
            "mutation TeamCreate($input: TeamCreateInput!) { teamCreate(input: $input) "
            "{ success team { id key name } } }",
            {"input": {"name": TEAM_NAME, "key": TEAM_KEY}})
        return data["teamCreate"]["team"]

    def create_state(self, team_id, spec):
        data = self.execute(
            "mutation StateCreate($input: WorkflowStateCreateInput!) "
            "{ workflowStateCreate(input: $input) { success workflowState { id name } } }",
            {"input": {"teamId": team_id, "name": spec["name"], "type": spec["type"],
                       "color": spec["color"], "position": spec["position"]}})
        return data["workflowStateCreate"]["workflowState"]

    def update_state(self, state_id, name=None, position=None):
        fields = {}
        if name is not None:
            fields["name"] = name
        if position is not None:
            fields["position"] = position
        self.execute(
            "mutation StateUpdate($id: String!, $input: WorkflowStateUpdateInput!) "
            "{ workflowStateUpdate(id: $id, input: $input) { success } }",
            {"id": state_id, "input": fields})

    def archive_state(self, state_id):
        self.execute(
            "mutation StateArchive($id: String!) "
            "{ workflowStateArchive(id: $id) { success } }", {"id": state_id})

    def issues_in_state(self, state_id):
        """Return every issue id in a workflow state, paging through results."""
        ids, after = [], None
        while True:
            data = self.execute(
                "query StateIssues($id: String!, $after: String) { workflowState(id: $id) "
                "{ issues(first: 250, after: $after) "
                "{ nodes { id } pageInfo { hasNextPage endCursor } } } }",
                {"id": state_id, "after": after})
            conn = data["workflowState"]["issues"]
            ids.extend(n["id"] for n in conn["nodes"])
            if not conn["pageInfo"]["hasNextPage"]:
                return ids
            after = conn["pageInfo"]["endCursor"]

    def move_issue(self, issue_id, state_id):
        self.execute(
            "mutation IssueMove($id: String!, $state: String!) "
            "{ issueUpdate(id: $id, input: { stateId: $state }) { success } }",
            {"id": issue_id, "state": state_id})

    def set_default_issue_state(self, team_id, state_id):
        self.execute(
            "mutation TeamUpdate($id: String!, $input: TeamUpdateInput!) "
            "{ teamUpdate(id: $id, input: $input) { success } }",
            {"id": team_id, "input": {"defaultIssueStateId": state_id}})

    def find_project(self):
        data = self.execute("query Projects { projects { nodes { id name } } }", {})
        return next((p for p in data["projects"]["nodes"] if p["name"] == PROJECT_NAME), None)

    def create_project(self, team_id):
        data = self.execute(
            "mutation ProjectCreate($input: ProjectCreateInput!) "
            "{ projectCreate(input: $input) { success project { id name } } }",
            {"input": {"name": PROJECT_NAME, "teamIds": [team_id]}})
        return data["projectCreate"]["project"]

    def existing_view_names(self):
        data = self.execute("query CustomViews { customViews { nodes { id name } } }", {})
        return {v["name"] for v in data["customViews"]["nodes"]}

    def create_view(self, team_id, spec):
        self.execute(
            "mutation CustomViewCreate($input: CustomViewCreateInput!) "
            "{ customViewCreate(input: $input) { success customView { id name } } }",
            {"input": {"name": spec["name"], "teamId": team_id, "shared": True,
                       "filterData": spec["filterData"]}})


def seed(client):
    """Reconcile the Pearl team, columns, project, and views; idempotent.

    Returns {team, project, states_created, states_recreated, issues_moved,
    views_created, leftover_states, default_issue_state}.
    """
    team = client.find_team()
    if team is None:
        team = client.create_team()
        states = client.team_states(team["id"])   # re-fetch: defaults aren't in the create reply
        current_default = None
    else:
        states = team.get("states", {}).get("nodes", [])
        current_default = (team.get("defaultIssueState") or {}).get("id")
    team_id = team["id"]

    plan = reconcile_state_plan(states)
    leftover = leftover_state_names(states)

    # Phase 1: free up the canonical names held by states we're about to replace,
    # so the new state can be created under the real name.
    for entry in plan:
        if entry["action"] == "recreate":
            client.update_state(entry["match_id"], name=f"{entry['target']['name']} (migrating)")

    # Phase 2: create new states -- genuinely new columns and type-healed
    # replacements. A replacement's new type is set here (the only place type can
    # be set), so the new unstarted Backlog exists before any archive in phase 5.
    resolved, created, recreated = {}, [], []
    for entry in plan:
        target = entry["target"]
        if entry["action"] == "create":
            resolved[target["name"]] = client.create_state(team_id, target)["id"]
            created.append(target["name"])
        elif entry["action"] == "recreate":
            resolved[target["name"]] = client.create_state(team_id, target)["id"]
            recreated.append(target["name"])
        else:
            resolved[target["name"]] = entry["match_id"]

    # Phase 3: move issues off each replaced state onto its replacement.
    issues_moved = 0
    for entry in plan:
        if entry["action"] == "recreate":
            new_id = resolved[entry["target"]["name"]]
            for issue_id in client.issues_in_state(entry["match_id"]):
                client.move_issue(issue_id, new_id)
                issues_moved += 1

    # Phase 4: point the new-issue default at the intake column (before archiving
    # the old default's state in phase 5).
    intake_id = resolved.get(INTAKE_STATE)
    default_set = bool(intake_id) and intake_id != current_default
    if default_set:
        client.set_default_issue_state(team_id, intake_id)

    # Phase 5: archive the replaced states. A new unstarted state now exists, so
    # archiving the old unstarted one is allowed.
    for entry in plan:
        if entry["action"] == "recreate":
            client.archive_state(entry["match_id"])

    # Phase 6: set each column's final name + position, in target order so a
    # rename chain (a fresh team's Backlog -> Icebox, then Todo -> Backlog) never
    # collides on a name an unprocessed column still holds.
    for entry in plan:
        target = entry["target"]
        client.update_state(resolved[target["name"]], name=target["name"],
                            position=target["position"])

    project = client.find_project() or client.create_project(team_id)
    project_id = project["id"]

    have_views = client.existing_view_names()
    views_created = []
    for spec in desired_views(project_id):
        if spec["name"] not in have_views:
            client.create_view(team_id, spec)
            views_created.append(spec["name"])

    return {"team": team_id, "project": project_id, "states_created": created,
            "states_recreated": recreated, "issues_moved": issues_moved,
            "views_created": views_created, "leftover_states": leftover,
            "default_issue_state": INTAKE_STATE if default_set else "unchanged"}


def warn_if_key_file_world_readable(path, stream=None):
    """Warn when PATH exists and is readable by group or others.

    The key file holds a Linear API key, so group/other access is a leak risk.
    Returns True when a warning was emitted, False otherwise (absent or 600)."""
    if stream is None:
        stream = sys.stderr
    try:
        mode = os.stat(path).st_mode
    except OSError:
        return False
    if mode & (stat.S_IRWXG | stat.S_IRWXO):
        print(f"warning: {path} is group/other-accessible; chmod 600 it "
              "(it holds your Linear API key)", file=stream)
        return True
    return False


def main(argv=None):
    parser = argparse.ArgumentParser(description="Seed a Pearl Linear workspace.")
    parser.add_argument("--dry-run", action="store_true",
                        help="print the reconcile plan without changing anything")
    args = parser.parse_args(argv)

    warn_if_key_file_world_readable("apikey.txt")
    api_key = os.environ.get("LINEAR_API_KEY")
    if not api_key:
        print("LINEAR_API_KEY is not set", file=sys.stderr)
        return 2

    client = LinearClient(api_key)
    if args.dry_run:
        team = client.find_team()
        states = team.get("states", {}).get("nodes", []) if team else []
        print("Plan:")
        print(f"  team: {'exists' if team else 'create'}")
        for entry in reconcile_state_plan(states):
            verb = entry["action"]
            note = ""
            if verb == "recreate":
                moving = len(client.issues_in_state(entry["match_id"]))
                note = f" (type change; moving {moving} issue(s))"
            print(f"    {verb:8} {entry['target']['name']}{note}")
        if states:
            print(f"  leftover columns: {leftover_state_names(states) or 'none'}")
        return 0

    summary = seed(client)
    print(f"Seeded Pearl workspace: team {summary['team']}, project {summary['project']}")
    print(f"  states created:   {summary['states_created'] or 'none'}")
    print(f"  states recreated: {summary['states_recreated'] or 'none'} "
          f"(moved {summary['issues_moved']} issue(s))")
    print(f"  views created:    {summary['views_created'] or 'none'}")
    if summary["leftover_states"]:
        print(f"  leftover columns (left as-is; Duplicate is reserved/hidden): "
              f"{summary['leftover_states']}")
    return 0


if __name__ == "__main__":
    sys.exit(main())