feat(v1alpha api): Allow passing parameters in Run Workflow API call (#323)

Changes the Run workflow API implementation to allow passing parameters
and improve performance.

Previously, the request was synchronously waiting for repo-proxy to
create a hook by retrieving data from the git provider and then
schedule the workflow on the plumber.
This had performance issues, and it did not support passing parameters to
be used in the pipeline.

The new implementation goes directly to the plumber, which only saves the
request in the database before responding, and all hook processing is
done asynchronously.
This should improve performance, and also the plumber supports passing
parameters out-of-the-box.

Author: DamjanBecirovic

docs/docs/reference/api.md

@@ -100,6 +100,7 @@ Parameters:
 - `reference` (**required**) - git reference for the desired branch, tag, or pull request--e.g. *refs/heads/master*, *refs/tags/v1.0*, or *refs/pull/123*.
 - `commit_sha` (*optional*) - Commit sha of the desired commit.
 - `pipeline_file` (*optional*) - The path within the repository to the YAML file that contains the pipeline definition. The default value is *.semaphore/semaphore.yml*.
+- `parameters`: (key-values) specify parameter values that will be available in all jobs of the initial pipeline and can be used in the same way as the parameters from the [parameterized promotions](../using-semaphore/promotions#parameters). 
 
 Response:
 
@@ -108,17 +109,25 @@ HTTP status: 200
 
 {
   "workflow_id": "32a689e0-9082-4c5b-a648-bb3dc645452d",
-  "pipeline_id": "2abeb1a9-eb4a-4834-84b8-cb7806aec063",
-  "hook_id": "ff7d57ef-92c5-4fcd-9c0c-6ae9e24bfcec"
+  "pipeline_id": "2abeb1a9-eb4a-4834-84b8-cb7806aec063"
 }
 ```
 
 Example:
 
 ```shell
-curl -i -H "Authorization: Token {api_token}" \
-     -d "project_id={project_id}&reference={reference}" \
-     -X POST  "https://<organization-url>.semaphoreci.com/api/v1alpha/plumber-workflows"
+curl -X POST --location "https://<organization-url>.semaphoreci.com/api/v1alpha/plumber-workflows" \
+    -H "Authorization: Token {api_token}" \
+    -H "Content-Type: application/json" \
+    -d $'{
+    "project_id": "my_project_id",
+    "reference": "refs/heads/master",
+    "pipeline_file": "/.semaphore/deploy.yml",
+    "parameters": {
+        "PARAM_NAME": "PARAM_VALUE",
+        "PARAM_NAME_2": "PARAM_VALUE_2"
+    }
+}'
 ```
 
 ### Describe a workflow

github_hooks/lib/internal_api/repo_proxy/repo_proxy_server.rb

@@ -51,6 +51,64 @@ class RepoProxyServer < RepoProxyService::Service
         end
       end
 
+      define_rpc :create_blank do |req, logger|
+        project = ::Project.find(req.project_id)
+        user    = ::User.find(req.requester_id)
+
+        payload_builder = InternalApi::RepoProxy::PayloadFactory.create(req.git.reference, req.git.commit_sha)
+        payload = payload_builder.call(project, user)
+
+        params = ActionController::Parameters.new
+        params["hash_id"] = project.id
+        params["payload"] = payload.to_json
+
+        workflow = ::Semaphore::RepoHost::Hooks::Recorder.record_hook(params, project)
+        workflow.update(:result => ::Workflow::RESULT_OK)
+
+        branch = ::Branch.find_or_create_for_workflow(workflow)
+        branch.unarchive
+        workflow.update(:branch_id => branch.id)
+
+        if workflow.payload.pull_request?
+          branch.update(:pull_request_mergeable => true)
+          workflow.update(
+            :commit_author => payload["commit_author"],
+            :commit_sha => payload["merge_commit_sha"],
+            :git_ref => payload["semaphore_ref"]
+          )
+        end
+
+        workflow.update(:ppl_id => req.pipeline_id)
+        workflow.update(:wf_id => req.wf_id)
+        workflow.update(:state => Workflow::STATE_LAUNCHING)
+
+        InternalApi::RepoProxy::CreateBlankResponse.new(
+          :hook_id => workflow.id,
+          :wf_id => req.wf_id,
+          :pipeline_id => req.pipeline_id,
+          :branch_id => branch.id,
+          :repo => InternalApi::RepoProxy::CreateBlankResponse::Repo.new(
+            :owner => branch.project.repository.owner,
+            :repo_name => branch.project.repository.name,
+            :branch_name => branch.name,
+            :commit_sha => workflow.commit_sha,
+            :repository_id => branch.project.repository.id
+          )
+        )
+
+      rescue ::InternalApi::RepoProxy::PrPayload::PrNotMergeableError => e
+        raise GRPC::Aborted, e.message
+      rescue ::InternalApi::RepoProxy::PayloadFactory::InvalidReferenceError => e
+        raise GRPC::InvalidArgument, e.message
+      rescue ::RepoHost::RemoteException::NotFound
+        raise GRPC::NotFound, "Reference not found on GitHub #{req.git.reference} #{req.git.commit_sha}"
+      rescue ::RepoHost::RemoteException::Unknown => e
+        logger.error("Unknown error", error: e.message)
+        raise GRPC::Internal, "Unknown error"
+      rescue ::ActiveRecord::RecordNotFound => e
+        raise GRPC::NotFound, e.message
+      end
+
       define_rpc :create do |req, logger|
         project = ::Project.find(req.project_id)
 
@@ -102,8 +160,6 @@ def create_for_github_project(req, logger)
             workflow.branch_name
           end
 
-        integration_token, = ::Semaphore::ProjectIntegrationToken.new.project_token(branch.project)
-
         client  = InternalApi::PlumberWF::WorkflowService::Stub.new(App.plumber_internal_url, :this_channel_is_insecure)
         request = InternalApi::PlumberWF::ScheduleRequest.new(
           :service => InternalApi::PlumberWF::ScheduleRequest::ServiceType::GIT_HUB,

github_hooks/spec/lib/internal_api/repo_proxy/repo_proxy_server_spec.rb

@@ -217,6 +217,170 @@
     end
   end
 
+  describe "#create_blank" do
+    let(:user) { FactoryBot.create(:user, :github_connection) }
+    let(:repository) do
+      FactoryBot.create(
+        :repository,
+        name: "sandbox",
+        owner: "renderedtext",
+        integration_type: "github_app"
+      )
+    end
+    let(:project) { FactoryBot.create(:project, repository: repository) }
+    let(:workflow) { FactoryBot.create(:workflow_with_branch, project: project) }
+    let(:branch) { workflow.branch }
+
+    let(:git) do
+      InternalApi::RepoProxy::CreateBlankRequest::Git.new(
+        reference: "refs/heads/main",
+        commit_sha: "abc123"
+      )
+    end
+
+    let(:req) do
+      InternalApi::RepoProxy::CreateBlankRequest.new(
+        project_id: project.id,
+        requester_id: user.id,
+        pipeline_id: "pipeline-id",
+        wf_id: "workflow-id",
+        git: git
+      )
+    end
+
+    let(:payload_hash) do
+      {
+        "commit_author" => "[email protected]",
+        "merge_commit_sha" => "abc123",
+        "semaphore_ref" => "refs/merge"
+      }
+    end
+
+    before do
+      payload = instance_double(InternalApi::RepoProxy::PrPayload, call: payload_hash)
+      allow(InternalApi::RepoProxy::PayloadFactory).to receive(:create)
+        .with(req.git.reference, req.git.commit_sha)
+        .and_return(payload)
+
+      allow(Semaphore::RepoHost::Hooks::Recorder).to receive(:record_hook)
+        .and_return(workflow)
+
+      allow(Branch).to receive(:find_or_create_for_workflow).with(workflow).and_return(branch)
+      allow(branch).to receive(:unarchive)
+      allow(branch).to receive(:update)
+      allow(workflow).to receive(:update)
+    end
+
+    it "creates a blank hook and returns the expected response" do
+      allow(payload_hash).to receive(:pull_request?).and_return(false)
+      allow(workflow).to receive(:payload).and_return(payload_hash)
+      expect(workflow).to receive(:update).with(state: Workflow::STATE_LAUNCHING)
+      result = server.create_blank(req, call)
+
+      expect(result).to be_a(InternalApi::RepoProxy::CreateBlankResponse)
+      expect(result.hook_id).to eq(workflow.id)
+      expect(result.wf_id).to eq(req.wf_id)
+      expect(result.pipeline_id).to eq(req.pipeline_id)
+      expect(result.branch_id).to eq(branch.id)
+
+      repo = result.repo
+      expect(repo.owner).to eq(repository.owner)
+      expect(repo.repo_name).to eq(repository.name)
+      expect(repo.branch_name).to eq(branch.name)
+      expect(repo.commit_sha).to eq(workflow.commit_sha)
+      expect(repo.repository_id).to eq(repository.id)
+    end
+
+    context "when pull request is not mergeable" do
+      before do
+        allow(InternalApi::RepoProxy::PayloadFactory).to receive(:create).and_raise(
+          InternalApi::RepoProxy::PrPayload::PrNotMergeableError.new("PR not mergeable")
+        )
+      end
+
+      it "raises GRPC::Aborted" do
+        expect do
+          server.create_blank(req, call)
+        end.to raise_error(GRPC::Aborted, /PR not mergeable/)
+      end
+    end
+
+    context "when reference is invalid" do
+      before do
+        allow(InternalApi::RepoProxy::PayloadFactory).to receive(:create)
+          .and_raise(InternalApi::RepoProxy::PayloadFactory::InvalidReferenceError.new("Invalid ref"))
+      end
+
+      it "raises GRPC::InvalidArgument" do
+        expect do
+          server.create_blank(req, call)
+        end.to raise_error(GRPC::InvalidArgument, /Invalid ref/)
+      end
+    end
+
+    context "when reference is not found on GitHub" do
+      before do
+        allow(InternalApi::RepoProxy::PayloadFactory).to receive(:create)
+          .and_raise(RepoHost::RemoteException::NotFound)
+      end
+
+      it "raises GRPC::NotFound" do
+        expect do
+          server.create_blank(req, call)
+        end.to raise_error(GRPC::NotFound, /Reference not found/)
+      end
+    end
+
+    context "when unknown error occurs" do
+      before do
+        allow(Semaphore::RepoHost::Hooks::Recorder).to receive(:record_hook)
+          .and_raise(RepoHost::RemoteException::Unknown.new("Boom"))
+      end
+
+      it "raises GRPC::Internal" do
+        expect do
+          server.create_blank(req, call)
+        end.to raise_error(GRPC::Internal, /Unknown error/)
+      end
+    end
+
+    context "when the user is not found" do
+      before do
+        @invalid_req = InternalApi::RepoProxy::CreateBlankRequest.new(
+          project_id: project.id,
+          requester_id: "invalid-user-id",
+          pipeline_id: "pipeline-id",
+          wf_id: "workflow-id",
+          git: git
+        )
+      end
+
+      it "raises GRPC::NotFound for missing user" do
+        expect do
+          server.create_blank(@invalid_req, call)
+        end.to raise_error(GRPC::NotFound, /Couldn't find User/)
+      end
+    end
+
+    context "when the project is not found" do
+      before do
+        @invalid_req = InternalApi::RepoProxy::CreateBlankRequest.new(
+          project_id: "invalid-project-id",
+          requester_id: user.id,
+          pipeline_id: "pipeline-id",
+          wf_id: "workflow-id",
+          git: git
+        )
+      end
+
+      it "raises GRPC::NotFound for missing project" do
+        expect do
+          server.create_blank(@invalid_req, call)
+        end.to raise_error(GRPC::NotFound, /Couldn't find Project/)
+      end
+    end
+  end
+
   describe "#create" do
     before "when unknown remote error is raised" do
       allow(InternalApi::RepoProxy::PayloadFactory).to receive(

plumber/ppl/lib/ppl/actions/schedule_impl.ex

@@ -79,11 +79,11 @@ defmodule Ppl.Actions.ScheduleImpl do
 
   # Schedule
 
-  def schedule(ctx, top_level?, initial_request?, task_workflow?) do
+  def schedule(ctx, top_level?, initial_request?, start_in_conceived?) do
     log_run_request(ctx)
 
     ctx
-    |> prepare_request_multi(top_level?, initial_request?, task_workflow?)
+    |> prepare_request_multi(top_level?, initial_request?, start_in_conceived?)
     |> persist_request
     |> case do
       {:ok, %{ppl_req: ppl_req}} ->
@@ -92,7 +92,7 @@ defmodule Ppl.Actions.ScheduleImpl do
                retry_count: publish_retry_count(), timeout_ms: publish_timeout()),
 
              predicate   <- fn query -> query |> where(ppl_id: ^ppl_req.id) end,
-             :ok         <- execute_first_state_with_predicate(predicate, task_workflow?),
+             :ok         <- execute_first_state_with_predicate(predicate, start_in_conceived?),
         do: response(ppl_req)
       # Idempotency -> return {:ok, ...}
       {:error, :ppl_req, {:request_token_exists, request_token}, _} ->
@@ -136,32 +136,32 @@ defmodule Ppl.Actions.ScheduleImpl do
     |> Map.put("suppressed_attributes", attribute_list)
   end
 
-  def prepare_request_multi(ctx, top_level?, initial_request?, task_workflow?) do
+  def prepare_request_multi(ctx, top_level?, initial_request?, start_in_conceived?) do
     ctx = RequestReviser.revise(ctx)
 
     Multi.new()
     # insert pipeline request
     |> Multi.run(:ppl_req, fn _, _ ->
       Metrics.benchmark("Ppl.schedule_break_down", ["insert_request"], fn ->
-        PplRequestsQueries.insert_request(ctx, top_level?, initial_request?, task_workflow?)
+        PplRequestsQueries.insert_request(ctx, top_level?, initial_request?, start_in_conceived?)
       end)
     end)
     # insert pipeline based on that request
     |> Multi.run(:ppl, fn _, %{ppl_req: ppl_req} ->
       Metrics.benchmark("Ppl.schedule_break_down", ["insert_pipeline"], fn ->
-        PplsQueries.insert(ppl_req, "", task_workflow?)
+        PplsQueries.insert(ppl_req, "", start_in_conceived?)
       end)
     end)
     # update pipeline to include wf_number
     |> Multi.run(:wf_num, fn _, %{ppl_req: ppl_req, ppl: ppl} ->
       Metrics.benchmark("Ppl.schedule_break_down", ["set_wf_num"], fn ->
-        set_workflow_number(ppl, ppl_req, task_workflow?)
+        set_workflow_number(ppl, ppl_req, start_in_conceived?)
       end)
     end)
     # insert pipeline sub init for this pipeline
     |> Multi.run(:ppl_sub_init, fn _, %{ppl_req: ppl_req} ->
       Metrics.benchmark("Ppl.schedule_break_down", ["insert_subinit"], fn ->
-        PplSubInitsQueries.insert(ppl_req, "regular", task_workflow?)
+        PplSubInitsQueries.insert(ppl_req, "regular", start_in_conceived?)
       end)
     end)
     # save inital_request separately for easier debug
@@ -179,25 +179,25 @@ defmodule Ppl.Actions.ScheduleImpl do
   end
 
   # promotions
-  def set_workflow_number(ppl, req = %{request_args: %{"wf_number" => num}}, task_workflow?)
+  def set_workflow_number(ppl, req = %{request_args: %{"wf_number" => num}}, start_in_conceived?)
     when is_integer(num) and num > 0 do
       with service     <- Map.get(req.request_args, "service"),
-           {:ok, _ppl} <- update_ppl(ppl, service, num, task_workflow?),
+           {:ok, _ppl} <- update_ppl(ppl, service, num, start_in_conceived?),
       do: {:ok, num}
   end
   # partial rebuilds
-  def set_workflow_number(ppl = %{partial_rebuild_of: val}, ppl_req, task_workflow?)
+  def set_workflow_number(ppl = %{partial_rebuild_of: val}, ppl_req, start_in_conceived?)
     when is_binary(val) and val != "" do
       with {:ok, l_wf} <- calculate_wf_num(ppl, ppl_req),
            service     <- Map.get(ppl_req.request_args, "service"),
-           {:ok, _ppl} <- update_ppl(ppl, service, l_wf.wf_number + 1, task_workflow?),
+           {:ok, _ppl} <- update_ppl(ppl, service, l_wf.wf_number + 1, start_in_conceived?),
       do: {:ok, l_wf.wf_number + 1}
   end
   # regular schedule and wf_rebuild
-  def set_workflow_number(ppl, ppl_req, task_workflow?) do
+  def set_workflow_number(ppl, ppl_req, start_in_conceived?) do
     with {:ok, l_wf} <- read_from_latest_wf_table(ppl, ppl_req),
          service     <- Map.get(ppl_req.request_args, "service"),
-         {:ok, _ppl} <- update_ppl(ppl, service, l_wf.wf_number + 1, task_workflow?),
+         {:ok, _ppl} <- update_ppl(ppl, service, l_wf.wf_number + 1, start_in_conceived?),
          {:ok, _}    <- LatestWfsQueries.insert_or_update(l_wf, ppl_req, l_wf.wf_number + 1),
     do: {:ok, l_wf.wf_number + 1}
   end
@@ -224,8 +224,8 @@ defmodule Ppl.Actions.ScheduleImpl do
   defp get_initial_wf_ppl(%{wf_id: wf_id}, _ppl),
     do: PplsQueries.get_initial_wf_ppl(wf_id)
 
-  defp update_ppl(ppl, service, wf_num, task_workflow?) do
-    with_repo_data? = !task_workflow?
+  defp update_ppl(ppl, service, wf_num, start_in_conceived?) do
+    with_repo_data? = !start_in_conceived?
 
     ppl
     |> Ppls.changeset(%{wf_number: wf_num}, service == "listener_proxy", with_repo_data?)