Add pipeline params handling for deeploy (#295)

* feat: add pipeline params in deeploy

* fix: pipeline sanitization

* fix: pipeline params in deeploy_specs

* fix: rm legacy support

* chore: inc version

Author: toderian

extensions/business/deeploy/deeploy_const.py

@@ -39,6 +39,8 @@ class DEEPLOY_KEYS:
   CHAINSTORE_RESPONSE = "chainstore_response"
   PIPELINE_INPUT_TYPE = "pipeline_input_type"
   PIPELINE_INPUT_URI = "pipeline_input_uri"
+  PIPELINE_PARAMS = "pipeline_params"
+  JOB_CONFIG = "job_config"
   # App params keys
   APP_PARAMS = "app_params"
   APP_PARAMS_IMAGE = "IMAGE"
@@ -543,6 +545,7 @@ class JOB_APP_TYPES:
   # Pipeline data source
   "pipeline_input_type"  : "void",  # Pipeline TYPE
   "pipeline_input_uri" : None,
+  "pipeline_params": {},
   "chainstore_response" : True,
 
   # Plugins array (NEW FORMAT - each object is a plugin instance)

extensions/business/deeploy/deeploy_job_mixin.py

@@ -67,6 +67,11 @@ def extract_invariable_data_from_pipeline(self, pipeline: dict):
     }
     """
 
+    if pipeline is None:
+      return None
+    if not isinstance(pipeline, dict):
+      return None
+
     # Create a copy of the pipeline and remove the TIME field
     extracted_data = pipeline.copy()
     extracted_data.pop("TIME", None)
@@ -88,9 +93,20 @@ def save_job_pipeline_in_cstore(self, pipeline: dict, job_id: int):
     """
     result = False
     try: 
+      if pipeline is None:
+        self.P(f"Skipping CSTORE save for job {job_id}: pipeline is None", color='y')
+        return False
+      if not isinstance(pipeline, dict):
+        self.P(f"Skipping CSTORE save for job {job_id}: pipeline is of type {type(pipeline).__name__}", color='y')
+        return False
+
       self.P("Saving pipeline to CSTORE...")
 
       sanitized_pipeline = self.extract_invariable_data_from_pipeline(pipeline)
+      if sanitized_pipeline is None:
+        self.P(f"Skipping CSTORE save for job {job_id}: unable to sanitize pipeline payload", color='y')
+        return False
+
       sorted_pipeline = self._recursively_sort_pipeline_data(sanitized_pipeline)
       cid = self._save_pipeline_to_r1fs(sorted_pipeline)
 

extensions/business/deeploy/deeploy_manager_api.py

@@ -169,6 +169,9 @@ def _process_pipeline_request(
       job_id = inputs.get(DEEPLOY_KEYS.JOB_ID, None)
       is_confirmable_job = inputs.chainstore_response
 
+      pipeline_params = self._extract_pipeline_params(inputs)
+      inputs[DEEPLOY_KEYS.PIPELINE_PARAMS] = pipeline_params
+
       # Validate plugins array structure and required fields for each plugin
       plugins_array = inputs.get(DEEPLOY_KEYS.PLUGINS)
       if plugins_array:
@@ -276,6 +279,22 @@ def _process_pipeline_request(
       inputs[DEEPLOY_KEYS.TARGET_NODES_COUNT] = len(deployment_nodes)
       inputs.target_nodes_count = len(deployment_nodes)
 
+      if deeploy_specs_for_update is not None and not isinstance(deeploy_specs_for_update, dict):
+        msg = (
+          f"{DEEPLOY_ERRORS.REQUEST3}. Unexpected 'deeploy_specs' payload type "
+          f"{type(deeploy_specs_for_update).__name__}."
+        )
+        raise ValueError(msg)
+      deeploy_specs_payload = (
+        self.deepcopy(deeploy_specs_for_update)
+        if isinstance(deeploy_specs_for_update, dict)
+        else {}
+      )
+      deeploy_specs_payload = self._ensure_deeploy_specs_job_config(
+        deeploy_specs_payload,
+        pipeline_params=pipeline_params,
+      )
+
       dct_status, str_status = self.check_and_deploy_pipelines(
         sender=sender,
         inputs=inputs,
@@ -285,7 +304,7 @@ def _process_pipeline_request(
         new_nodes=deployment_nodes,
         update_nodes=[],
         discovered_plugin_instances=discovered_plugin_instances,
-        dct_deeploy_specs_create=deeploy_specs_for_update,
+        dct_deeploy_specs_create=deeploy_specs_payload,
         job_app_type=job_app_type,
       )
 
@@ -317,6 +336,8 @@ def _process_pipeline_request(
         plugins_array = inputs.get(DEEPLOY_KEYS.PLUGINS)
         if plugins_array:
           dct_request['plugins_count'] = len(plugins_array)
+        # if pipeline_params:
+        #   dct_request[DEEPLOY_KEYS.PIPELINE_PARAMS] = pipeline_params
 
       result = {
         DEEPLOY_KEYS.STATUS: str_status,
@@ -370,6 +391,9 @@ def create_pipeline(
       job_tags : list
           Tags for filtering target nodes
           Example: ["KYB", "DC:HOSTINGER", "CT:FR|IT|RO", "REG:EU"]
+      pipeline_params : dict, optional
+          Additional pipeline-level parameters forwarded to the data capture thread. `null` falls back to `{}`.
+          The provided keys are merged into the pipeline configuration at the top level.
 
       nonce : str
           The nonce used for signing the request
@@ -486,6 +510,10 @@ def update_pipeline(
       job_id : int
           The job ID from blockchain
 
+      pipeline_params : dict, optional
+          Additional pipeline-level parameters forwarded to the data capture thread. `null` falls back to `{}`.
+          The provided keys are merged into the pipeline configuration at the top level.
+
       nonce : str
           The nonce used for signing the request