diff --git a/app/graphql/mutations/namespaces/projects/flows/update.rb b/app/graphql/mutations/namespaces/projects/flows/update.rb
new file mode 100644
index 00000000..8e60a24f
--- /dev/null
+++ b/app/graphql/mutations/namespaces/projects/flows/update.rb
@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+
+module Mutations
+  module Namespaces
+    module Projects
+      module Flows
+        class Update < BaseMutation
+          description 'Update an existing flow.'
+
+          field :flow, Types::FlowType, null: true, description: 'The updated flow.'
+
+          argument :flow_id, Types::GlobalIdType[Flow],
+                   required: true, description: 'The ID of the flow to update'
+
+          argument :flow_input, Types::Input::FlowInputType, description: 'The updated flow', required: true
+
+          def resolve(flow_id:, flow_input:, **_params)
+            flow = SagittariusSchema.object_from_id(flow_id)
+
+            return { errors: [create_error(:flow_not_found, 'Flow does not exist')] } if flow.nil?
+
+            flow_type = SagittariusSchema.object_from_id(flow_input.type)
+            return { errors: [create_error(:flow_type_not_found, 'Invalid flow type id')] } if flow_type.nil?
+
+            ::Namespaces::Projects::Flows::UpdateService.new(
+              current_authentication,
+              flow,
+              flow_input
+            ).execute.to_mutation_response(success_key: :flow)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/app/graphql/types/mutation_type.rb b/app/graphql/types/mutation_type.rb
index bdb10cf3..b3833d21 100644
--- a/app/graphql/types/mutation_type.rb
+++ b/app/graphql/types/mutation_type.rb
@@ -12,10 +12,11 @@ class MutationType < Types::BaseObject
     mount_mutation Mutations::Namespaces::Members::Invite
     mount_mutation Mutations::Namespaces::Projects::AssignRuntimes
     mount_mutation Mutations::Namespaces::Projects::Create
-    mount_mutation Mutations::Namespaces::Projects::Update
     mount_mutation Mutations::Namespaces::Projects::Delete
+    mount_mutation Mutations::Namespaces::Projects::Update
     mount_mutation Mutations::Namespaces::Projects::Flows::Create
     mount_mutation Mutations::Namespaces::Projects::Flows::Delete
+    mount_mutation Mutations::Namespaces::Projects::Flows::Update
     mount_mutation Mutations::Namespaces::Roles::AssignAbilities
     mount_mutation Mutations::Namespaces::Roles::AssignProjects
     mount_mutation Mutations::Namespaces::Roles::Create
diff --git a/app/services/error_code.rb b/app/services/error_code.rb
index 1e205b87..e51c5abd 100644
--- a/app/services/error_code.rb
+++ b/app/services/error_code.rb
@@ -83,6 +83,10 @@ def self.error_codes
       data_type_not_found: { description: 'The data type with the given identifier was not found' },
       invalid_flow_type: { description: 'The flow type is invalid because of active model errors' },
       no_data_type_for_identifier: { description: 'No data type could be found for the given identifier' },
+      node_not_found: { description: 'The node with this id does not exist' },
+      function_value_not_found: { description: 'The id for the function value node does not exist' },
+      invalid_node_parameter: { description: 'The node parameter is invalid' },
+      invalid_node_function: { description: 'The node function is invalid' },
 
       primary_level_not_found: { description: '', deprecation_reason: 'Outdated concept' },
       secondary_level_not_found: { description: '', deprecation_reason: 'Outdated concept' },
diff --git a/app/services/namespaces/projects/flows/create_service.rb b/app/services/namespaces/projects/flows/create_service.rb
index b4526a41..19b08c9b 100644
--- a/app/services/namespaces/projects/flows/create_service.rb
+++ b/app/services/namespaces/projects/flows/create_service.rb
@@ -5,6 +5,7 @@ module Projects
     module Flows
       class CreateService
         include Sagittarius::Database::Transactional
+        include FlowServiceHelper
 
         attr_reader :current_authentication, :namespace_project, :params
 
@@ -122,7 +123,7 @@ def create_node_function(node_function_id, input_nodes, t)
             if parameter.value.function_value.present?
               params << NodeParameter.create(
                 runtime_parameter: runtime_parameter,
-                function_value: create_node_function(parameter.value.function_value, t)
+                function_value: create_node_function(parameter.value.function_value, input_nodes, t)
               )
               next
             end
@@ -143,7 +144,11 @@ def create_node_function(node_function_id, input_nodes, t)
               runtime_parameter: runtime_parameter,
               reference_value: ReferenceValue.create(
                 node_function: referenced_node,
-                data_type_identifier: get_data_type_identifier(parameter.value.reference_value.data_type_identifier, t),
+                data_type_identifier: get_data_type_identifier(
+                  namespace_project.primary_runtime,
+                  parameter.value.reference_value.data_type_identifier,
+                  t
+                ),
                 depth: parameter.value.reference_value.depth,
                 node: parameter.value.reference_value.node,
                 scope: parameter.value.reference_value.scope,
@@ -167,47 +172,6 @@ def create_node_function(node_function_id, input_nodes, t)
             node_parameters: params
           )
         end
-
-        private
-
-        def get_data_type_identifier(identifier, t)
-          return DataTypeIdentifier.create(generic_key: identifier.generic_key) if identifier.generic_key.present?
-
-          if identifier.generic_type.present?
-            data_type = namespace_project.primary_runtime.data_types.find_by(
-              id: identifier.generic_type.data_type_id.model_id
-            )
-
-            if data_type.nil?
-              t.rollback_and_return! ServiceResponse.error(
-                message: 'Data type not found',
-                error_code: :data_type_not_found
-              )
-            end
-
-            mappers = identifier.generic_type.mappers.map do |mapper|
-              GenericMapper.create(
-                generic_mapper_id: mapper.generic_mapper_id,
-                source: mapper.source,
-                target: mapper.target
-              )
-            end
-            return DataTypeIdentifier.create(generic_type: GenericType.create(data_type: data_type, mappers: mappers))
-          end
-
-          return if identifier.data_type_id.blank?
-
-          data_type = namespace_project.primary_runtime.data_types.find_by(id: identifier.data_type_id.model_id)
-
-          if data_type.nil?
-            t.rollback_and_return! ServiceResponse.error(
-              message: 'Data type not found',
-              error_code: :data_type_not_found
-            )
-          end
-
-          DataTypeIdentifier.create(data_type: data_type)
-        end
       end
     end
   end
diff --git a/app/services/namespaces/projects/flows/flow_service_helper.rb b/app/services/namespaces/projects/flows/flow_service_helper.rb
new file mode 100644
index 00000000..8efc0d03
--- /dev/null
+++ b/app/services/namespaces/projects/flows/flow_service_helper.rb
@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+module Namespaces
+  module Projects
+    module Flows
+      module FlowServiceHelper
+        def get_data_type_identifier(runtime, identifier, t)
+          if identifier.generic_key.present?
+            return DataTypeIdentifier.find_or_create_by(runtime: runtime,
+                                                        generic_key: identifier.generic_key)
+          end
+
+          if identifier.generic_type.present?
+            data_type = namespace_project.primary_runtime.data_types.find_by(
+              id: identifier.generic_type.data_type_id.model_id
+            )
+
+            if data_type.nil?
+              t.rollback_and_return! ServiceResponse.error(
+                message: 'Data type not found',
+                error_code: :data_type_not_found
+              )
+            end
+
+            mappers = identifier.generic_type.mappers.map do |mapper|
+              GenericMapper.find_or_create_by(
+                runtime: runtime,
+                generic_mapper_id: mapper.generic_mapper_id,
+                source: mapper.source,
+                target: mapper.target
+              )
+            end
+            generic_type = GenericType.joins(:generic_mappers).find_or_create_by(data_type: data_type,
+                                                                                 generic_mappers: mappers)
+            return DataTypeIdentifier.find_or_create_by(runtime: runtime, generic_type: generic_type)
+          end
+
+          data_type = namespace_project.primary_runtime.data_types.find_by(id: identifier.data_type_id.model_id)
+
+          if data_type.nil?
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Data type not found',
+              error_code: :data_type_not_found
+            )
+          end
+
+          DataTypeIdentifier.find_or_create_by(runtime: runtime, data_type: data_type)
+        end
+      end
+    end
+  end
+end
diff --git a/app/services/namespaces/projects/flows/update_service.rb b/app/services/namespaces/projects/flows/update_service.rb
new file mode 100644
index 00000000..82e60da2
--- /dev/null
+++ b/app/services/namespaces/projects/flows/update_service.rb
@@ -0,0 +1,267 @@
+# frozen_string_literal: true
+
+module Namespaces
+  module Projects
+    module Flows
+      class UpdateService
+        include Sagittarius::Database::Transactional
+        include FlowServiceHelper
+
+        attr_reader :current_authentication, :flow, :flow_input
+
+        def initialize(current_authentication, flow, flow_input)
+          @current_authentication = current_authentication
+          @flow = flow
+          @flow_input = flow_input
+        end
+
+        def execute
+          unless Ability.allowed?(current_authentication, :update_flow, flow)
+            return ServiceResponse.error(message: 'Missing permission', error_code: :missing_permission)
+          end
+
+          transactional do |t|
+            update_settings(t)
+            update_nodes(t)
+
+            unless flow.save
+              t.rollback_and_return! ServiceResponse.error(
+                message: 'Flow is invalid',
+                error_code: :invalid_flow,
+                details: flow.errors
+              )
+            end
+
+            validate_flow(t)
+
+            UpdateRuntimesForProjectJob.perform_later(flow.project.id)
+
+            create_audit_event
+
+            ServiceResponse.success(message: 'Flow updated', payload: flow)
+          end
+        end
+
+        private
+
+        def update_settings(t)
+          flow_input.settings.each do |setting|
+            flow_setting = flow.flow_settings.find_or_initialize_by(flow_setting_id: setting.flow_setting_identifier)
+            flow_setting.object = setting.value
+
+            next if flow_setting.valid?
+
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Invalid flow settings',
+              error_code: :invalid_flow_setting,
+              details: flow_setting.errors
+            )
+          end
+
+          flow.flow_settings.where.not(flow_setting_id: flow_input.settings.map(&:flow_setting_identifier)).destroy_all
+        end
+
+        def update_nodes(t)
+          all_nodes = flow.collect_node_functions
+
+          current_node_input_id = flow_input.starting_node_id
+          node_index = 0
+
+          updated_nodes = []
+
+          until current_node_input_id.nil?
+            current_node = all_nodes[node_index] || NodeFunction.new
+            current_node_input = flow_input.nodes.find { |n| n.id == current_node_input_id }
+
+            update_node(t, current_node, current_node_input)
+            updated_nodes << { node: current_node, input: current_node_input }
+
+            current_node_input_id = current_node_input.next_node_id
+            node_index += 1
+          end
+
+          updated_nodes.each do |node|
+            update_node_parameters(t, node[:node], node[:input], updated_nodes)
+            update_next_node(t, node[:node], node[:input], updated_nodes)
+
+            next if node[:node].save
+
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Invalid node',
+              error_code: :invalid_node_function,
+              details: node[:node].errors
+            )
+          end
+
+          update_starting_node(t, updated_nodes)
+
+          delete_old_nodes(t, all_nodes.reject { |node| updated_nodes.pluck(:node).pluck(:id).include?(node.id) })
+        end
+
+        def update_starting_node(t, all_nodes)
+          starting_node = all_nodes.find { |n| n[:input].id == flow_input.starting_node_id }
+
+          if starting_node.nil?
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Starting node not found',
+              error_code: :node_not_found
+            )
+          end
+
+          flow.starting_node = starting_node[:node]
+        end
+
+        def delete_old_nodes(t, remaining_nodes)
+          remaining_nodes.each do |node|
+            node.destroy
+            next unless node.persisted?
+
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Failed to delete node',
+              error_code: :invalid_node_function,
+              details: node.errors
+            )
+          end
+        end
+
+        def update_node(t, current_node, current_node_input)
+          runtime_function_definition = flow.project.primary_runtime.runtime_function_definitions.find_by(
+            id: current_node_input.runtime_function_id.model_id
+          )
+          if runtime_function_definition.nil?
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Invalid runtime function id',
+              error_code: :invalid_runtime_function_id
+            )
+          end
+
+          current_node.runtime_function = runtime_function_definition
+        end
+
+        def update_next_node(t, current_node, current_node_input, all_nodes)
+          next_node = all_nodes.find { |n| n[:input].id == current_node_input.next_node_id }
+
+          if next_node.nil? && current_node_input.next_node_id.present?
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Next node not found',
+              error_code: :node_not_found
+            )
+          end
+
+          current_node.next_node = next_node&.[](:node)
+        end
+
+        def update_node_parameters(t, current_node, current_node_input, all_nodes)
+          db_parameters = current_node.node_parameters.first(current_node_input.parameters.count)
+          current_node_input.parameters.each_with_index do |parameter, index|
+            db_parameters[index] ||= current_node.node_parameters.build
+
+            runtime_parameter = current_node.runtime_function.parameters.find_by(
+              id: parameter.runtime_parameter_definition_id.model_id
+            )
+            if runtime_parameter.nil?
+              t.rollback_and_return! ServiceResponse.error(
+                message: 'Invalid runtime parameter id',
+                error_code: :invalid_runtime_parameter_id
+              )
+            end
+
+            db_parameters[index].runtime_parameter = runtime_parameter
+
+            db_parameters[index].literal_value = parameter.value.literal_value.presence
+
+            if parameter.value.function_value.present?
+              node = all_nodes.find { |n| n[:input].id == parameter.value.function_value.runtime_function_id }
+
+              if node.nil?
+                t.rollback_and_return! ServiceResponse.error(
+                  message: 'Invalid function value for parameter',
+                  error_code: :function_value_not_found
+                )
+              end
+
+              db_parameters[index].function_value = node[:node]
+            else
+              db_parameters[index].function_value = nil
+            end
+
+            if parameter.value.reference_value.present?
+              data_type_identifier = get_data_type_identifier(
+                flow.project.primary_runtime,
+                parameter.value.reference_value.data_type_identifier,
+                t
+              )
+
+              referenced_node = NodeFunction.joins(:runtime_function).find_by(
+                id: parameter.value.reference_value.node_function_id.model_id,
+                runtime_function_definitions: { runtime_id: flow.project.primary_runtime.id }
+              )
+
+              if referenced_node.nil?
+                t.rollback_and_return! ServiceResponse.error(
+                  message: 'Referenced node function not found',
+                  error_code: :referenced_value_not_found
+                )
+              end
+
+              db_parameters[index].reference_value ||= ReferenceValue.new
+              reference_value = db_parameters[index].reference_value
+
+              reference_paths_input = parameter.value.reference_value.reference_paths
+              reference_paths = reference_value.reference_paths.first(reference_paths_input.length)
+              reference_paths_input.each_with_index do |path, i|
+                reference_paths[i] ||= reference_value.reference_paths.build
+                reference_paths[i].assign_attributes(path: path.path, array_index: path.array_index)
+              end
+
+              reference_value.assign_attributes(
+                data_type_identifier: data_type_identifier,
+                node_function: reference_value,
+                depth: parameter.value.reference_value.depth,
+                node: parameter.value.reference_value.node,
+                scope: parameter.value.reference_value.scope,
+                reference_paths: reference_paths
+              )
+            else
+              db_parameters[index].reference_value = nil
+            end
+
+            next if db_parameters[index].valid?
+
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Invalid node parameter',
+              error_code: :invalid_node_parameter,
+              details: db_parameters[index].errors
+            )
+          end
+
+          current_node.node_parameters = db_parameters
+        end
+
+        def validate_flow(t)
+          res = Validation::ValidationService.new(current_authentication, flow).execute
+
+          return unless res.error?
+
+          t.rollback_and_return! ServiceResponse.error(
+            message: 'Flow validation failed',
+            error_code: res.payload[:error_code],
+            details: res.payload[:details]
+          )
+        end
+
+        def create_audit_event
+          AuditService.audit(
+            :flow_updated,
+            author_id: current_authentication.user.id,
+            entity: flow,
+            target: flow.project,
+            details: {
+              **flow.attributes.except('created_at', 'updated_at'),
+            }
+          )
+        end
+      end
+    end
+  end
+end
diff --git a/docs/graphql/enum/errorcodeenum.md b/docs/graphql/enum/errorcodeenum.md
index 5c525357..e8a0e6a5 100644
--- a/docs/graphql/enum/errorcodeenum.md
+++ b/docs/graphql/enum/errorcodeenum.md
@@ -21,6 +21,7 @@ Represents the available error responses
 | `FLOW_NOT_FOUND` | The flow with the given identifier was not found |
 | `FLOW_TYPE_NOT_FOUND` | The flow type with the given identifier was not found |
 | `FLOW_VALIDATION_FAILED` | The flow validation has failed |
+| `FUNCTION_VALUE_NOT_FOUND` | The id for the function value node does not exist |
 | `GENERIC_KEY_NOT_FOUND` | The given key was not found in the data type |
 | `IDENTITY_NOT_FOUND` | The external identity with the given identifier was not found |
 | `IDENTITY_VALIDATION_FAILED` | Failed to validate the external identity |
@@ -37,6 +38,8 @@ Represents the available error responses
 | `INVALID_NAMESPACE_MEMBER` | The namespace member is invalid because of active model errors |
 | `INVALID_NAMESPACE_PROJECT` | The namespace project is invalid because of active model errors |
 | `INVALID_NAMESPACE_ROLE` | The namespace role is invalid because of active model errors |
+| `INVALID_NODE_FUNCTION` | The node function is invalid |
+| `INVALID_NODE_PARAMETER` | The node parameter is invalid |
 | `INVALID_ORGANIZATION` | The organization is invalid because of active model errors |
 | `INVALID_PASSWORD_REPEAT` | The provided password repeat does not match the password |
 | `INVALID_RUNTIME` | The runtime is invalid because of active model errors |
@@ -63,6 +66,7 @@ Represents the available error responses
 | `NAMESPACE_NOT_FOUND` | The namespace with the given identifier was not found |
 | `NAMESPACE_PROJECT_NOT_FOUND` | The namespace project with the given identifier was not found |
 | `NAMESPACE_ROLE_NOT_FOUND` | The namespace role with the given identifier was not found |
+| `NODE_NOT_FOUND` | The node with this id does not exist |
 | `NO_DATATYPE_IDENTIFIER_FOR_GENERIC_KEY` | No data type identifier could be found for the given generic key |
 | `NO_DATA_TYPE_FOR_IDENTIFIER` | No data type could be found for the given identifier |
 | `NO_FREE_LICENSE_SEATS` | There are no free license seats to complete this operation |
diff --git a/docs/graphql/mutation/namespacesprojectsflowsupdate.md b/docs/graphql/mutation/namespacesprojectsflowsupdate.md
new file mode 100644
index 00000000..08c8e8a0
--- /dev/null
+++ b/docs/graphql/mutation/namespacesprojectsflowsupdate.md
@@ -0,0 +1,21 @@
+---
+title: namespacesProjectsFlowsUpdate
+---
+
+Update an existing flow.
+
+## Arguments
+
+| Name | Type | Description |
+|------|------|-------------|
+| `clientMutationId` | [`String`](../scalar/string.md) | A unique identifier for the client performing the mutation. |
+| `flowId` | [`FlowID!`](../scalar/flowid.md) | The ID of the flow to update |
+| `flowInput` | [`FlowInput!`](../input_object/flowinput.md) | The updated flow |
+
+## Fields
+
+| Name | Type | Description |
+|------|------|-------------|
+| `clientMutationId` | [`String`](../scalar/string.md) | A unique identifier for the client performing the mutation. |
+| `errors` | [`[Error!]!`](../object/error.md) | Errors encountered during execution of the mutation. |
+| `flow` | [`Flow`](../object/flow.md) | The updated flow. |
diff --git a/spec/graphql/mutations/namespaces/projects/flows/update_spec.rb b/spec/graphql/mutations/namespaces/projects/flows/update_spec.rb
new file mode 100644
index 00000000..b4b0c372
--- /dev/null
+++ b/spec/graphql/mutations/namespaces/projects/flows/update_spec.rb
@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe Mutations::Namespaces::Projects::Flows::Update do
+  it { expect(described_class.graphql_name).to eq('NamespacesProjectsFlowsUpdate') }
+end
diff --git a/spec/requests/graphql/mutation/namespace/projects/flows/update_mutation_spec.rb b/spec/requests/graphql/mutation/namespace/projects/flows/update_mutation_spec.rb
new file mode 100644
index 00000000..5e7dd9a6
--- /dev/null
+++ b/spec/requests/graphql/mutation/namespace/projects/flows/update_mutation_spec.rb
@@ -0,0 +1,148 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe 'namespacesProjectsFlowsUpdate Mutation' do
+  include GraphqlHelpers
+
+  subject(:mutate!) { post_graphql mutation, variables: variables, current_user: current_user }
+
+  let(:mutation) do
+    <<~QUERY
+      mutation($input: NamespacesProjectsFlowsUpdateInput!) {
+        namespacesProjectsFlowsUpdate(input: $input) {
+          #{error_query}
+          flow {
+            id
+            startingNodeId
+            nodes {
+              count
+              nodes {
+                id
+                parameters {
+                  count
+                  nodes {
+                    id
+                  }
+                }
+              }
+            }
+            settings {
+              nodes {
+                flowSettingIdentifier
+                id
+                value
+              }
+            }
+          }
+        }
+      }
+    QUERY
+  end
+
+  let(:runtime) { create(:runtime) }
+  let(:project) { create(:namespace_project, primary_runtime: runtime) }
+  let(:flow_type) { create(:flow_type, runtime: runtime) }
+  let(:flow) { create(:flow, project: project, flow_type: flow_type) }
+  let(:runtime_function) do
+    create(:runtime_function_definition, runtime: runtime,
+                                         parameters: [
+
+                                           create(:runtime_parameter_definition,
+                                                  data_type: create(:data_type_identifier,
+                                                                    data_type: create(:data_type)))
+
+                                         ])
+  end
+  let(:input) do
+    {
+      flowId: flow.to_global_id.to_s,
+      flowInput: {
+        name: generate(:flow_name),
+        type: flow_type.to_global_id.to_s,
+        startingNodeId: 'gid://sagittarius/NodeFunction/999',
+        settings: {
+          flowSettingIdentifier: 'key',
+          value: {
+            'key' => 'value',
+          },
+        },
+        nodes: [
+          {
+            id: 'gid://sagittarius/NodeFunction/999',
+            runtimeFunctionId: runtime_function.to_global_id.to_s,
+            parameters: [
+              runtimeParameterDefinitionId: runtime_function.parameters.first.to_global_id.to_s,
+              value: {
+                literalValue: 'test_value',
+              }
+            ],
+            nextNodeId: 'gid://sagittarius/NodeFunction/991',
+          },
+          {
+            id: 'gid://sagittarius/NodeFunction/991',
+            runtimeFunctionId: runtime_function.to_global_id.to_s,
+            parameters: [
+              runtimeParameterDefinitionId: runtime_function.parameters.first.to_global_id.to_s,
+              value: {
+                literalValue: 'test_value2',
+              }
+            ],
+          }
+        ],
+      },
+    }
+  end
+
+  let(:variables) { { input: input } }
+  let(:current_user) { create(:user) }
+
+  context 'when user has the permission' do
+    before do
+      stub_allowed_ability(NamespaceProjectPolicy, :update_flow, user: current_user, subject: project)
+      stub_allowed_ability(NamespaceProjectPolicy, :read_namespace_project, user: current_user, subject: project)
+    end
+
+    it 'updates flow' do
+      mutate!
+
+      updated_flow_id = graphql_data_at(:namespaces_projects_flows_update, :flow, :id)
+      expect(updated_flow_id).to be_present
+      flow = SagittariusSchema.object_from_id(updated_flow_id)
+
+      expect(graphql_data_at(:namespaces_projects_flows_update, :flow, :settings).size).to eq(1)
+
+      nodes = graphql_data_at(:namespaces_projects_flows_update, :flow, :nodes, :nodes)
+      starting_node = nodes.find do |n|
+        n['id'] == graphql_data_at(:namespaces_projects_flows_update, :flow, :starting_node_id)
+      end
+      expect(starting_node['parameters']['count']).to eq(1)
+
+      expect(flow).to be_present
+      expect(project.flows).to include(flow)
+      expect(flow.collect_node_functions.count).to eq(2)
+
+      is_expected.to create_audit_event(
+        :flow_updated,
+        author_id: current_user.id,
+        entity_id: flow.id,
+        entity_type: 'Flow',
+        details: {
+          **flow.attributes.except('created_at', 'updated_at'),
+        },
+        target_id: project.id,
+        target_type: 'NamespaceProject'
+      )
+    end
+  end
+
+  context 'when user does not have the permission' do
+    it 'returns an error' do
+      mutate!
+
+      expect(graphql_data_at(:namespaces_projects_flows_update, :errors)).to be_present
+      expect(graphql_data_at(:namespaces_projects_flows_update, :flow)).to be_nil
+      expect(graphql_data_at(:namespaces_projects_flows_update, :errors).first['errorCode']).to eq('MISSING_PERMISSION')
+    end
+  end
+end
diff --git a/spec/services/namespaces/projects/flows/update_service_spec.rb b/spec/services/namespaces/projects/flows/update_service_spec.rb
new file mode 100644
index 00000000..8a8bcba5
--- /dev/null
+++ b/spec/services/namespaces/projects/flows/update_service_spec.rb
@@ -0,0 +1,89 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe Namespaces::Projects::Flows::UpdateService do
+  subject(:service_response) do
+    described_class.new(create_authentication(current_user), flow, flow_input).execute
+  end
+
+  let(:runtime) { create(:runtime) }
+  let(:namespace_project) { create(:namespace_project, primary_runtime: runtime) }
+  let(:starting_node) do
+    create(:node_function, runtime_function: create(:runtime_function_definition, runtime: runtime))
+  end
+  let(:flow) { create(:flow, project: namespace_project, flow_type: create(:flow_type, runtime: runtime)) }
+  let(:flow_input) do
+    Struct.new(:settings, :starting_node_id, :nodes).new(
+      [],
+      starting_node.to_global_id,
+      [
+        Struct.new(:id, :runtime_function_id, :next_node_id, :parameters).new(
+          starting_node.to_global_id,
+          starting_node.runtime_function.to_global_id,
+          nil,
+          []
+        )
+      ]
+    )
+  end
+
+  shared_examples 'does not update' do
+    it { is_expected.to be_error }
+
+    it 'does not update flow' do
+      expect { service_response }.not_to change { flow.reload.updated_at }
+    end
+
+    it { expect { service_response }.not_to create_audit_event }
+  end
+
+  context 'when user does not exist' do
+    let(:current_user) { nil }
+
+    it_behaves_like 'does not update'
+  end
+
+  context 'when params are invalid' do
+    let(:current_user) { create(:user) }
+
+    context 'when starting node is nil' do
+      let(:params) { { project: namespace_project, flow_type: create(:flow_type), starting_node: nil } }
+
+      it_behaves_like 'does not update'
+    end
+  end
+
+  context 'when user and params are valid' do
+    let(:current_user) { create(:user) }
+
+    before do
+      stub_allowed_ability(NamespaceProjectPolicy, :update_flow, user: current_user, subject: namespace_project)
+    end
+
+    it { is_expected.to be_success }
+    it { expect(service_response.payload.reload).to be_valid }
+
+    it do
+      is_expected.to create_audit_event(
+        :flow_updated,
+        author_id: current_user.id,
+        entity_type: 'Flow',
+        entity_id: service_response.payload.id,
+        details: {
+          **service_response.payload.attributes.except('created_at', 'updated_at'),
+        },
+        target_id: namespace_project.id,
+        target_type: 'NamespaceProject'
+      )
+    end
+
+    it 'queues job to update runtimes' do
+      allow(UpdateRuntimesForProjectJob).to receive(:perform_later)
+
+      service_response
+
+      expect(UpdateRuntimesForProjectJob).to have_received(:perform_later).with(namespace_project.id)
+    end
+  end
+end