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
    (Icebox, Triage Queue, 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, and the Duplicate state is reserved. So this script never
archives. It renames Linear's defaults into the six Agile columns -- keeping
Backlog / In Progress / Done / Canceled, and repurposing the unstarted Todo as
Triage Queue -- and creates only the genuinely-new Icebox. Linear's reserved
Duplicate state is left alone (it's hidden in the board UI); the script reports
any other leftover column for you to tidy.

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

Two notes on the result:
  - Triage Queue is Linear's "unstarted" type, so pearl's grouped view (which
    orders by state type, then position) lists it after the backlog columns,
    even though the Linear board itself keeps your column order.
  - CustomViewCreateInput has no grouping field, so Pearl Open Issues is created
    with its filter only -- group it by category in pearl (pearl-set-grouping)
    or the Linear UI.

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

import argparse
import json
import os
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. `sources` names the Linear default state a column is
# reconciled from when it isn't present by its own name yet (Triage Queue takes
# over the unstarted Todo). Positions are distinct and nonzero.
TARGET_STATES = [
    {"name": "Icebox", "type": "backlog", "color": "#bec2c8", "position": 1.0, "sources": []},
    {"name": "Triage Queue", "type": "unstarted", "color": "#e2e2e2", "position": 2.0,
     "sources": ["Todo"]},
    {"name": "Backlog", "type": "backlog", "color": "#95a2b3", "position": 3.0, "sources": []},
    {"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"]},
]

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 Queue"}}}},
    ]


def _target_aliases(target):
    return {target["name"].lower(), *(s.lower() for s in target.get("sources", []))}


def reconcile_state_plan(existing, targets=None):
    """Match each target column to an existing state, or mark it for creation.

    EXISTING is a list of state dicts (id, name, type). Returns a list of
    {target, match_id}: match_id is the id of the existing state a column
    reconciles onto (its own name or a `sources` default, case-insensitive), or
    None when the column has to be created. Pure: no I/O. Each existing state is
    claimed by at most one target.
    """
    targets = targets or TARGET_STATES
    plan, used = [], set()
    for target in targets:
        aliases = _target_aliases(target)
        match = next((s for s in existing
                      if s["id"] not in used and s["name"].lower() in aliases), None)
        if match:
            used.add(match["id"])
        plan.append({"target": target, "match_id": match["id"] if match else None})
    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, 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 "
            "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, position):
        self.execute(
            "mutation StateUpdate($id: String!, $input: WorkflowStateUpdateInput!) "
            "{ workflowStateUpdate(id: $id, input: $input) { success } }",
            {"id": state_id, "input": {"name": name, "position": position}})

    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, views_created, leftover_states}.
    """
    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
    else:
        states = team.get("states", {}).get("nodes", [])
    team_id = team["id"]

    states_created = []
    for entry in reconcile_state_plan(states):
        target, match_id = entry["target"], entry["match_id"]
        if match_id is None:
            match_id = client.create_state(team_id, target)["id"]
            states_created.append(target["name"])
        # Set name + position in one update; a freshly-created state needs the
        # position pass because Linear appends it at a high position otherwise.
        client.update_state(match_id, target["name"], target["position"])
    leftover = leftover_state_names(states)

    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": states_created,
            "views_created": views_created, "leftover_states": leftover}


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)

    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 = "update" if entry["match_id"] else "create"
            print(f"    {verb}: {entry['target']['name']}")
        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"  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())