alien_bindings/providers/worker/
aws_lambda.rs

1use crate::error::{binding_env_var, ErrorData, Result};
2use crate::traits::{Binding, Worker, WorkerInvokeRequest, WorkerInvokeResponse};
3use alien_aws_clients::lambda::{InvocationType, InvokeRequest, LambdaApi, LambdaClient};
4use alien_aws_clients::AwsCredentialProvider;
5use alien_core::bindings::LambdaWorkerBinding;
6use alien_error::{AlienError, Context, IntoAlienError};
7use async_trait::async_trait;
8use base64::Engine;
9use reqwest::Client;
10use std::collections::BTreeMap;
11
12/// AWS Lambda worker binding implementation
13#[derive(Debug)]
14pub struct LambdaWorker {
15    client: LambdaClient,
16    binding: LambdaWorkerBinding,
17}
18
19impl LambdaWorker {
20    pub fn new(
21        client: Client,
22        credentials: AwsCredentialProvider,
23        binding: LambdaWorkerBinding,
24    ) -> Self {
25        let lambda_client = LambdaClient::new(client, credentials);
26        Self {
27            client: lambda_client,
28            binding,
29        }
30    }
31
32    /// Get the worker name from the binding, resolving template expressions if needed
33    fn get_worker_name(&self) -> Result<String> {
34        self.binding
35            .worker_name
36            .clone()
37            .into_value("worker", "worker_name")
38            .context(ErrorData::BindingConfigInvalid {
39                env_var: binding_env_var("worker"),
40                binding_name: "worker".to_string(),
41                reason: "Failed to resolve worker_name from binding".to_string(),
42            })
43    }
44}
45
46impl Binding for LambdaWorker {}
47
48#[async_trait]
49impl Worker for LambdaWorker {
50    async fn invoke(&self, request: WorkerInvokeRequest) -> Result<WorkerInvokeResponse> {
51        let worker_name = self.get_worker_name()?;
52
53        // Create the invoke request payload
54        // For Lambda, we need to construct an HTTP-like payload that the runtime can understand
55        let payload = serde_json::json!({
56            "httpMethod": request.method.to_uppercase(),
57            "path": request.path,
58            "headers": request.headers,
59            "body": base64::engine::general_purpose::STANDARD.encode(&request.body),
60            "isBase64Encoded": true
61        });
62
63        let payload_bytes =
64            serde_json::to_vec(&payload)
65                .into_alien_error()
66                .context(ErrorData::Other {
67                    message: "Failed to serialize Lambda invoke payload".to_string(),
68                })?;
69
70        // Use the target_worker if provided, otherwise use the bound worker
71        let target_worker = if !request.target_worker.is_empty() {
72            request.target_worker.clone()
73        } else {
74            worker_name
75        };
76
77        let invoke_request = InvokeRequest::builder()
78            .function_name(target_worker.clone())
79            .invocation_type(InvocationType::RequestResponse)
80            .payload(payload_bytes)
81            .build();
82
83        let response = self
84            .client
85            .invoke(invoke_request)
86            .await
87            .context(ErrorData::Other {
88                message: format!("Failed to invoke Lambda worker '{}'", target_worker),
89            })?;
90
91        // Check for worker error
92        if let Some(function_error) = response.function_error {
93            return Err(AlienError::new(ErrorData::Other {
94                message: format!(
95                    "Lambda worker '{}' returned error: {}",
96                    target_worker, function_error
97                ),
98            }));
99        }
100
101        // Parse the response payload
102        let lambda_response: serde_json::Value = serde_json::from_slice(&response.payload)
103            .into_alien_error()
104            .context(ErrorData::Other {
105                message: "Failed to parse Lambda response payload".to_string(),
106            })?;
107
108        // Extract HTTP response components
109        let status = lambda_response
110            .get("statusCode")
111            .and_then(|s| s.as_u64())
112            .unwrap_or(200) as u16;
113
114        let headers = lambda_response
115            .get("headers")
116            .and_then(|h| h.as_object())
117            .map(|obj| {
118                obj.iter()
119                    .map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string()))
120                    .collect::<BTreeMap<String, String>>()
121            })
122            .unwrap_or_default();
123
124        let body = if let Some(body_str) = lambda_response.get("body").and_then(|b| b.as_str()) {
125            // Check if body is base64 encoded
126            let is_base64 = lambda_response
127                .get("isBase64Encoded")
128                .and_then(|b| b.as_bool())
129                .unwrap_or(false);
130
131            if is_base64 {
132                base64::engine::general_purpose::STANDARD
133                    .decode(body_str)
134                    .into_alien_error()
135                    .context(ErrorData::Other {
136                        message: "Failed to decode base64 response body".to_string(),
137                    })?
138            } else {
139                body_str.as_bytes().to_vec()
140            }
141        } else {
142            Vec::new()
143        };
144
145        Ok(WorkerInvokeResponse {
146            status,
147            headers,
148            body,
149        })
150    }
151
152    async fn get_worker_url(&self) -> Result<Option<String>> {
153        // First check if we have it in the binding
154        if let Some(url_binding) = &self.binding.url {
155            let url = url_binding.clone().into_value("worker", "url").context(
156                ErrorData::BindingConfigInvalid {
157                    env_var: binding_env_var("worker"),
158                    binding_name: "worker".to_string(),
159                    reason: "Failed to resolve url from binding".to_string(),
160                },
161            )?;
162            return Ok(Some(url));
163        }
164
165        // If not in binding, try to fetch it from AWS
166        let worker_name = self.get_worker_name()?;
167        match self
168            .client
169            .get_function_url_config(&worker_name, None)
170            .await
171        {
172            Ok(url_config) => Ok(Some(url_config.function_url)),
173            Err(_) => Ok(None), // Worker URL doesn't exist
174        }
175    }
176
177    fn as_any(&self) -> &dyn std::any::Any {
178        self
179    }
180}
alien_bindings/providers/worker/aws_lambda.rs

alien_bindings/providers/worker/
aws_lambda.rs
