feat: allow multiple concurrent status updates for issues (#180)

- **feat: allows multiple issues to get status updates concurrently**
- **fix: ensure mutations on issues happen concurrently**

Author: bougyman

app/lib/linear_cli/cli.ex

@@ -246,7 +246,7 @@ defmodule LinearCli.CLI do
     end
   end
 
-  # `issue list`/`take`/`update` all set `allow_unknown_args: true` so bare
+  # `issue list`/`take`/`status`/`update` all set `allow_unknown_args: true` so bare
   # tokens (e.g. `CRY-1`) can be captured as issue ids via `result.unknown`
   # rather than a declared positional arg (Optimus has no `type: :array`
   # equivalent - see their subcommand specs below). That same bucket also
@@ -256,7 +256,7 @@ defmodule LinearCli.CLI do
   # clearly. Every other subcommand has `allow_unknown_args: false` (the
   # default), where Optimus itself already rejects unknown args before we
   # ever see a parse_result - so `result.unknown` is only ever non-empty here
-  # for those three subcommands, and only ever contains genuine bare ids
+  # for those four subcommands, and only ever contains genuine bare ids
   # once this filters out anything flag-shaped.
   defp reject_unknown_flags(unknown_tokens) do
     case Enum.filter(unknown_tokens, &String.starts_with?(&1, "-")) do
@@ -737,10 +737,8 @@ defmodule LinearCli.CLI do
             ],
             status: [
               name: "status",
-              about: "Change the workflow state of an issue",
-              args: [
-                issue_id: [value_name: "ISSUE_ID", help: "The Issue (i.e. CRY-1)", required: true]
-              ],
+              about: "Change workflow state (ISSUE_ID...)",
+              allow_unknown_args: true,
               options: [
                 status: [
                   short: "-s",

app/lib/linear_cli/cli/commands.ex

@@ -7,6 +7,8 @@ defmodule LinearCli.CLI.Commands do
   alias LinearCli.CLI.{Display, IssueHelpers, Projects, Prompt, WhatFor}
   alias LinearCli.{Favorites, Git, Linear, Profiles}
 
+  @max_concurrent_issue_updates 20
+
   @doc "Ported from commands/whoami.rb."
   def whoami(%{flags: flags, options: options}) do
     with {:ok, user} <- Linear.me() do
@@ -518,32 +520,91 @@ defmodule LinearCli.CLI.Commands do
   defp validate_issue_ids(_issue_ids), do: :ok
 
   @doc """
-  Changes the workflow state of an issue.
+  Changes the workflow state of one or more issues. Optimus captures the IDs in
+  `unknown`, since it has no variadic positional-argument type.
 
   With `--status`/`-s`, matches the given name against the issue's team's
   workflow states (case-insensitive exact, then unique prefix). Without it,
   prompts interactively via `LinearCli.CLI.Prompt.select/2`.
 
-  With `--comment`/`-m`, adds a comment to the issue before transitioning.
+  With `--comment`/`-m`, adds a comment to each issue before transitioning it.
+  Mutations for separate issues run concurrently with a limit of 20 in flight.
   """
   @spec issue_status(Optimus.ParseResult.t()) :: :ok | {:error, term()}
-  def issue_status(%{args: %{issue_id: issue_id}, options: options}) do
-    expanded_id = IssueHelpers.expand_issue_id(issue_id)
+  def issue_status(%{unknown: issue_ids, options: options}) do
+    with :ok <- validate_issue_ids(issue_ids),
+         {:ok, issues} <-
+           Linear.issues(%{ids: Enum.map(issue_ids, &IssueHelpers.expand_issue_id/1)}),
+         {:ok, planned_updates} <- plan_status_updates(issues, options.status),
+         {:ok, completed_updates} <- apply_status_updates(planned_updates, options.comment) do
+      show_status_updates(completed_updates, options.output)
+    end
+  end
 
-    with {:ok, [issue]} <- Linear.issues(%{ids: [expanded_id]}),
-         {:ok, states} <- Linear.workflow_states_by_team(issue.team.id),
-         {:ok, target_state} <- resolve_target_state(states, options.status),
-         :ok <- maybe_add_status_comment(issue, options.comment),
+  defp plan_status_updates(issues, status) do
+    issues
+    |> Enum.reduce_while({:ok, []}, fn issue, {:ok, updates} ->
+      with {:ok, states} <- Linear.workflow_states_by_team(issue.team.id),
+           {:ok, target_state} <- resolve_target_state(states, status) do
+        {:cont, {:ok, [{issue, target_state} | updates]}}
+      else
+        {:error, reason} -> {:halt, {:error, reason}}
+      end
+    end)
+    |> reverse_status_updates()
+  end
+
+  defp apply_status_updates([], _comment), do: {:ok, []}
+
+  defp apply_status_updates(planned_updates, comment) do
+    planned_updates
+    |> Task.async_stream(
+      fn {issue, target_state} ->
+        apply_status_update(issue, target_state, comment)
+      end,
+      max_concurrency: min(length(planned_updates), @max_concurrent_issue_updates),
+      ordered: true,
+      timeout: 30_000
+    )
+    |> Enum.reduce_while({:ok, []}, fn
+      {:ok, {:ok, update}}, {:ok, updates} ->
+        {:cont, {:ok, [update | updates]}}
+
+      {:ok, {:error, reason}}, {:ok, _updates} ->
+        {:halt, {:error, reason}}
+
+      {:exit, reason}, {:ok, _updates} ->
+        {:halt, {:error, {:task_exit, reason}}}
+    end)
+    |> reverse_status_updates()
+  end
+
+  defp apply_status_update(issue, target_state, comment) do
+    with :ok <- maybe_add_status_comment(issue, comment),
          {:ok, updated} <- Linear.set_issue_status(issue, target_state.id) do
-      Display.show(updated, %{output: options.output})
+      {:ok, {updated, target_state}}
+    end
+  end
 
-      if options.output != "json",
-        do: Prompt.ok("#{updated.identifier} status set to #{target_state.name}")
+  defp reverse_status_updates({:ok, updates}), do: {:ok, Enum.reverse(updates)}
+  defp reverse_status_updates(error), do: error
 
-      :ok
+  defp show_status_updates(completed_updates, output) do
+    updated_issues = Enum.map(completed_updates, &elem(&1, 0))
+    Display.show(one_or_many(updated_issues), %{output: output})
+
+    if output != "json" do
+      Enum.each(completed_updates, fn {updated, target_state} ->
+        Prompt.ok("#{updated.identifier} status set to #{target_state.name}")
+      end)
     end
+
+    :ok
   end
 
+  defp one_or_many([one]), do: one
+  defp one_or_many(many), do: many
+
   defp resolve_target_state(states, nil) do
     choices = Enum.sort_by(states, & &1.position) |> Enum.map(&{&1.name, &1})
     {:ok, Prompt.select("Choose a status", choices)}

app/test/linear_cli/cli/issue_commands_test.exs

@@ -775,6 +775,121 @@ defmodule LinearCli.CLI.IssueCommandsTest do
       assert output =~ "status set to Done"
     end
 
+    test "--status updates multiple issue IDs concurrently and emits a JSON array" do
+      test_pid = self()
+
+      issue_details = fn
+        "CRY-1" -> {"i1", "t1", "ENG", "Engineering", "s-eng-done"}
+        "CRY-2" -> {"i2", "t2", "OPS", "Operations", "s-ops-done"}
+      end
+
+      Req.Test.stub(LinearCli.Api, fn conn ->
+        {:ok, body, conn} = Plug.Conn.read_body(conn)
+        decoded = Jason.decode!(body)
+        %{"query" => query} = decoded
+        variables = decoded["variables"] || %{}
+
+        cond do
+          String.contains?(query, "issue(id: $id)") ->
+            identifier = variables["id"]
+            {id, team_id, team_key, team_name, _state_id} = issue_details.(identifier)
+
+            Req.Test.json(conn, %{
+              "data" => %{
+                "issue" =>
+                  issue_map(%{
+                    "id" => id,
+                    "identifier" => identifier,
+                    "team" => %{"id" => team_id, "key" => team_key, "name" => team_name}
+                  })
+              }
+            })
+
+          String.contains?(query, "states {") ->
+            team_id = variables["teamId"]
+            state_id = if team_id == "t1", do: "s-eng-done", else: "s-ops-done"
+            send(test_pid, {:states_queried, team_id})
+            Req.Test.json(conn, workflow_states([state_map(state_id, "Done", 1.0, "completed")]))
+
+          String.contains?(query, "issueUpdate") ->
+            identifier = variables["id"]
+            state_id = variables["input"]["stateId"]
+            {id, team_id, team_key, team_name, ^state_id} = issue_details.(identifier)
+            update_pid = self()
+            send(test_pid, {:status_update_started, identifier, state_id, update_pid})
+
+            receive do
+              :finish_status_update -> :ok
+            after
+              2_000 -> raise "status update was not released by the concurrency assertion"
+            end
+
+            Req.Test.json(conn, %{
+              "data" => %{
+                "issueUpdate" => %{
+                  "issue" =>
+                    issue_map(%{
+                      "id" => id,
+                      "identifier" => identifier,
+                      "team" => %{"id" => team_id, "key" => team_key, "name" => team_name},
+                      "state" => %{"id" => state_id, "name" => "Done", "type" => "completed"}
+                    })
+                }
+              }
+            })
+
+          true ->
+            raise "no stub matched query: #{query}"
+        end
+      end)
+
+      command =
+        Task.async(fn ->
+          capture_io(fn ->
+            assert :ok =
+                     LinearCli.CLI.main([
+                       "issue",
+                       "status",
+                       "--status",
+                       "Done",
+                       "--output",
+                       "json",
+                       "CRY-1",
+                       "CRY-2"
+                     ])
+          end)
+        end)
+
+      assert_receive {:status_update_started, "CRY-1", "s-eng-done", first_update}, 1_000
+      assert_receive {:status_update_started, "CRY-2", "s-ops-done", second_update}, 1_000
+      send(first_update, :finish_status_update)
+      send(second_update, :finish_status_update)
+
+      output = Task.await(command)
+
+      assert_received {:states_queried, "t1"}
+      assert_received {:states_queried, "t2"}
+
+      assert {:ok, decoded} = Jason.decode(output)
+      assert Enum.map(decoded, & &1["identifier"]) == ["CRY-1", "CRY-2"]
+    end
+
+    test "variadic issue IDs do not swallow unrecognized options" do
+      test_pid = self()
+      halt = fn code -> send(test_pid, {:halted, code}) end
+
+      stderr =
+        capture_io(:stderr, fn ->
+          LinearCli.CLI.main(
+            ["issue", "status", "--statuz", "Done", "CRY-1", "CRY-2"],
+            halt
+          )
+        end)
+
+      assert_received {:halted, 22}
+      assert stderr =~ "unrecognized option(s): --statuz"
+    end
+
     test "-s short flag also sets the workflow state" do
       test_pid = self()