feat: resolve upstream auth at runtime

crates/crank-runtime/src/executor.rs

@@ -9,7 +9,8 @@ use crank_core::{ExecutionMode, Target, TransportBehavior};
 use serde_json::{Map, Value, json};
 
 use crate::{
-    AdapterResponse, PreparedRequest, RuntimeError, RuntimeOperation, WindowExecutionResult,
+    AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeOperation,
+    WindowExecutionResult,
 };
 
 #[derive(Clone, Debug)]
@@ -42,8 +43,18 @@ impl RuntimeExecutor {
         &self,
         operation: &RuntimeOperation,
         input: &Value,
+    ) -> Result<Value, RuntimeError> {
+        self.execute_with_auth(operation, input, None).await
+    }
+
+    pub async fn execute_with_auth(
+        &self,
+        operation: &RuntimeOperation,
+        input: &Value,
+        resolved_auth: Option<&ResolvedAuth>,
     ) -> Result<Value, RuntimeError> {
         let prepared_request = self.prepare_request(operation, input)?;
+        let prepared_request = apply_resolved_auth(prepared_request, resolved_auth);
         self.execute_prepared(operation, prepared_request).await
     }
 
@@ -51,6 +62,15 @@ impl RuntimeExecutor {
         &self,
         operation: &RuntimeOperation,
         input: &Value,
+    ) -> Result<WindowExecutionResult, RuntimeError> {
+        self.execute_window_with_auth(operation, input, None).await
+    }
+
+    pub async fn execute_window_with_auth(
+        &self,
+        operation: &RuntimeOperation,
+        input: &Value,
+        resolved_auth: Option<&ResolvedAuth>,
     ) -> Result<WindowExecutionResult, RuntimeError> {
         let Some(streaming) = operation.execution_config.streaming.as_ref() else {
             return Err(RuntimeError::MissingStreamingConfig {
@@ -66,6 +86,7 @@ impl RuntimeExecutor {
         }
 
         let prepared_request = self.prepare_request(operation, input)?;
+        let prepared_request = apply_resolved_auth(prepared_request, resolved_auth);
         let adapter_response = if matches!(
             streaming.transport_behavior,
             TransportBehavior::ServerStream
@@ -86,6 +107,16 @@ impl RuntimeExecutor {
         &self,
         operation: &RuntimeOperation,
         input: &Value,
+    ) -> Result<WindowExecutionResult, RuntimeError> {
+        self.execute_session_seed_with_auth(operation, input, None)
+            .await
+    }
+
+    pub async fn execute_session_seed_with_auth(
+        &self,
+        operation: &RuntimeOperation,
+        input: &Value,
+        resolved_auth: Option<&ResolvedAuth>,
     ) -> Result<WindowExecutionResult, RuntimeError> {
         let Some(streaming) = operation.execution_config.streaming.as_ref() else {
             return Err(RuntimeError::MissingStreamingConfig {
@@ -113,7 +144,8 @@ impl RuntimeExecutor {
             config.max_items = Some(seed_limit);
         }
 
-        self.execute_window(&seeded_operation, input).await
+        self.execute_window_with_auth(&seeded_operation, input, resolved_auth)
+            .await
     }
 
     pub fn prepare_request(
@@ -396,6 +428,17 @@ fn finalize_output(
         .unwrap_or_else(|| Value::Object(Map::new())))
 }
 
+fn apply_resolved_auth(
+    mut prepared_request: PreparedRequest,
+    resolved_auth: Option<&ResolvedAuth>,
+) -> PreparedRequest {
+    if let Some(resolved_auth) = resolved_auth {
+        resolved_auth.apply(&mut prepared_request);
+    }
+
+    prepared_request
+}
+
 fn merge_headers(
     static_headers: &BTreeMap<String, String>,
     execution_headers: &BTreeMap<String, String>,