alien_bindings/providers/worker/
gcp_cloudrun.rs

1use crate::error::{binding_env_var, ErrorData, Result};
2use crate::traits::{Binding, Worker, WorkerInvokeRequest, WorkerInvokeResponse};
3use alien_core::bindings::CloudRunWorkerBinding;
4use alien_error::{AlienError, Context, IntoAlienError};
5use alien_gcp_clients::cloudrun::{CloudRunApi, CloudRunClient};
6use alien_gcp_clients::GcpClientConfig;
7use async_trait::async_trait;
8use reqwest::Client;
9use std::collections::BTreeMap;
10
11/// GCP Cloud Run worker binding implementation
12#[derive(Debug)]
13pub struct CloudRunWorker {
14    client: Client,
15    cloudrun_client: CloudRunClient,
16    binding: CloudRunWorkerBinding,
17}
18
19impl CloudRunWorker {
20    pub fn new(client: Client, config: GcpClientConfig, binding: CloudRunWorkerBinding) -> Self {
21        let cloudrun_client = CloudRunClient::new(client.clone(), config);
22        Self {
23            client,
24            cloudrun_client,
25            binding,
26        }
27    }
28
29    /// Get the private URL from the binding, resolving template expressions if needed
30    fn get_private_url(&self) -> Result<String> {
31        self.binding
32            .private_url
33            .clone()
34            .into_value("worker", "private_url")
35            .context(ErrorData::BindingConfigInvalid {
36                env_var: binding_env_var("worker"),
37                binding_name: "worker".to_string(),
38                reason: "Failed to resolve private_url from binding".to_string(),
39            })
40    }
41
42    /// Resolve the target URL for invocation
43    async fn resolve_target_url(&self, target_worker: &str) -> Result<String> {
44        if !target_worker.is_empty() {
45            // Check if target_worker looks like a URL (starts with http)
46            if target_worker.starts_with("http://") || target_worker.starts_with("https://") {
47                // Use the provided target worker as URL
48                Ok(target_worker.to_string())
49            } else {
50                // target_worker is likely a path/identifier, use binding URL
51                self.get_private_url()
52            }
53        } else {
54            // Use the private URL from binding
55            self.get_private_url()
56        }
57    }
58}
59
60impl Binding for CloudRunWorker {}
61
62#[async_trait]
63impl Worker for CloudRunWorker {
64    async fn invoke(&self, request: WorkerInvokeRequest) -> Result<WorkerInvokeResponse> {
65        let target_url = self.resolve_target_url(&request.target_worker).await?;
66
67        // Construct the full URL with path
68        let url = if request.path.starts_with('/') {
69            format!("{}{}", target_url.trim_end_matches('/'), request.path)
70        } else {
71            format!("{}/{}", target_url.trim_end_matches('/'), request.path)
72        };
73
74        // Build the HTTP request
75        let method = match request.method.to_uppercase().as_str() {
76            "GET" => reqwest::Method::GET,
77            "POST" => reqwest::Method::POST,
78            "PUT" => reqwest::Method::PUT,
79            "DELETE" => reqwest::Method::DELETE,
80            "PATCH" => reqwest::Method::PATCH,
81            "HEAD" => reqwest::Method::HEAD,
82            "OPTIONS" => reqwest::Method::OPTIONS,
83            _ => {
84                return Err(AlienError::new(ErrorData::InvalidInput {
85                    operation_context: "Worker invocation".to_string(),
86                    details: format!("Unsupported HTTP method: {}", request.method),
87                    field_name: Some("method".to_string()),
88                }));
89            }
90        };
91
92        let mut req_builder = self.client.request(method, &url);
93
94        // Add headers
95        for (key, value) in &request.headers {
96            req_builder = req_builder.header(key, value);
97        }
98
99        // Add body if present
100        if !request.body.is_empty() {
101            req_builder = req_builder.body(request.body.clone());
102        }
103
104        // Set timeout if specified
105        if let Some(timeout) = request.timeout {
106            req_builder = req_builder.timeout(timeout);
107        }
108
109        // Send the request
110        let response =
111            req_builder
112                .send()
113                .await
114                .into_alien_error()
115                .context(ErrorData::HttpRequestFailed {
116                    url: url.clone(),
117                    method: request.method.clone(),
118                })?;
119
120        // Extract response components
121        let status = response.status().as_u16();
122
123        let headers = response
124            .headers()
125            .iter()
126            .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
127            .collect::<BTreeMap<String, String>>();
128
129        let body = response
130            .bytes()
131            .await
132            .into_alien_error()
133            .context(ErrorData::HttpRequestFailed {
134                url: url.clone(),
135                method: "READ_BODY".to_string(),
136            })?
137            .to_vec();
138
139        Ok(WorkerInvokeResponse {
140            status,
141            headers,
142            body,
143        })
144    }
145
146    async fn get_worker_url(&self) -> Result<Option<String>> {
147        // First check if we have it in the binding
148        if let Some(url_binding) = &self.binding.public_url {
149            let url = url_binding
150                .clone()
151                .into_value("worker", "public_url")
152                .context(ErrorData::BindingConfigInvalid {
153                    env_var: binding_env_var("worker"),
154                    binding_name: "worker".to_string(),
155                    reason: "Failed to resolve public_url from binding".to_string(),
156                })?;
157            return Ok(Some(url));
158        }
159
160        // If not in binding, try to fetch it from GCP
161        let service_name = self
162            .binding
163            .service_name
164            .clone()
165            .into_value("worker", "service_name")
166            .context(ErrorData::BindingConfigInvalid {
167                env_var: binding_env_var("worker"),
168                binding_name: "worker".to_string(),
169                reason: "Failed to resolve service_name from binding".to_string(),
170            })?;
171
172        let location = self
173            .binding
174            .location
175            .clone()
176            .into_value("worker", "location")
177            .context(ErrorData::BindingConfigInvalid {
178                env_var: binding_env_var("worker"),
179                binding_name: "worker".to_string(),
180                reason: "Failed to resolve location from binding".to_string(),
181            })?;
182
183        match self
184            .cloudrun_client
185            .get_service(location, service_name)
186            .await
187        {
188            Ok(service) => {
189                // Return the first URL if available
190                Ok(service.urls.first().cloned())
191            }
192            Err(_) => Ok(None), // Service doesn't exist or no public URL
193        }
194    }
195
196    fn as_any(&self) -> &dyn std::any::Any {
197        self
198    }
199}
alien_bindings/providers/worker/gcp_cloudrun.rs

alien_bindings/providers/worker/
gcp_cloudrun.rs
