fix(processing_engine): Use the configured request path for Request plugins. (#25945)

influxdb3/tests/cli/mod.rs

@@ -1635,7 +1635,7 @@ def process_request(influxdb3_local, query_parameters, request_headers, request_
         "--plugin-filename",
         plugin_filename,
         "--trigger-spec",
-        "request:foo",
+        "request:bar",
         "--trigger-arguments",
         "test_arg=hello",
         trigger_path,
@@ -1646,7 +1646,7 @@ def process_request(influxdb3_local, query_parameters, request_headers, request_
     // send an HTTP request to the server
     let client = reqwest::Client::new();
     let response = client
-        .post(format!("{}/api/v3/engine/foo", server_addr))
+        .post(format!("{}/api/v3/engine/bar", server_addr))
         .header("Content-Type", "application/json")
         .query(&[("q1", "whatevs")])
         .body(r#"{"hello": "world"}"#)
@@ -1694,7 +1694,7 @@ def process_request(influxdb3_local, query_parameters, request_headers, request_
 
     // send an HTTP request to the server
     let response = client
-        .post(format!("{}/api/v3/engine/foo", server_addr))
+        .post(format!("{}/api/v3/engine/bar", server_addr))
         .header("Content-Type", "application/json")
         .query(&[("q1", "whatevs")])
         .body(r#"{"hello": "world"}"#)

influxdb3_processing_engine/src/lib.rs

@@ -477,11 +477,15 @@ impl ProcessingEngineManager for ProcessingEngineManagerImpl {
                     )?
                 }
                 PluginType::Request => {
+                    let TriggerSpecificationDefinition::RequestPath { path } = &trigger.trigger
+                    else {
+                        unreachable!()
+                    };
                     let rec = self
                         .plugin_event_tx
                         .write()
                         .await
-                        .add_request_trigger(trigger_name.to_string());
+                        .add_request_trigger(path.to_string());
 
                     plugins::run_request_plugin(
                         db_name.to_string(),

influxdb3_processing_engine/src/plugins.rs

@@ -71,6 +71,12 @@ pub enum PluginError {
     #[error("non-schedule plugin with schedule trigger: {0}")]
     NonSchedulePluginWithScheduleTrigger(String),
 
+    #[error("Trigger schedule type {schedule_type} invalid for trigger type {trigger_type} and type mismatch")]
+    TriggerScheduleTypeMismatch {
+        schedule_type: String,
+        trigger_type: String,
+    },
+
     #[error("error reading file from Github: {0} {1}")]
     FetchingFromGithub(reqwest::StatusCode, String),
 }

influxdb3_wal/src/lib.rs

@@ -728,7 +728,7 @@ impl TriggerSpecificationDefinition {
             }
             _ => Err(Error::TriggerSpecificationParseError {
                 trigger_spec: spec_str.to_string(),
-                context: Some("expect one of the following prefixes: 'table:', 'all_tables:', 'cron:', or 'every:'".to_string()),
+                context: Some("expect one of the following prefixes: 'table:', 'all_tables:', 'cron:', 'every:', or 'request:'".to_string()),
             }),
         }
     }