1use crate::{
2 ProviderConfig, ProviderHostEffectCoordinator, ProviderHttpProxyRegistry, ProviderSource,
3 ProviderTransport,
4};
5use async_trait::async_trait;
6use lenso_module_management::{
7 EndpointResolverSource, PROVIDER_RUNTIME_PLAN_PROTOCOL, ProviderRuntimeModule,
8 ProviderRuntimePlan, ProviderRuntimeService, ServiceIdentityPolicy, ServiceReference,
9 ServiceTransportBinding, StaticEndpointDeclaration,
10};
11use platform_core::{AppError, AppResult, ErrorCode};
12use platform_module::Module;
13use std::collections::BTreeMap;
14use std::sync::Arc;
15
16pub const BEARER_ENV_TRUST_PROFILE: &str = "bearer_env";
17
18#[derive(Debug, Clone)]
19pub struct ProviderEndpointResolutionRequest {
20 pub service_ref: ServiceReference,
21 pub source_id: String,
22 pub public_config: BTreeMap<String, String>,
23 pub secret_references: Vec<String>,
24}
25
26#[async_trait]
27pub trait ProviderEndpointResolver: std::fmt::Debug + Send + Sync {
28 async fn resolve(
29 &self,
30 request: &ProviderEndpointResolutionRequest,
31 ) -> AppResult<Vec<StaticEndpointDeclaration>>;
32}
33
34#[async_trait]
35pub trait ProviderCredentialResolver: std::fmt::Debug + Send + Sync {
36 async fn resolve_bearer(&self, policy: &ServiceIdentityPolicy) -> AppResult<String>;
37}
38
39#[derive(Debug, Clone, Default)]
40pub struct ProviderRuntimeAdapters {
41 endpoint_resolvers: BTreeMap<String, Arc<dyn ProviderEndpointResolver>>,
42 credential_resolvers: BTreeMap<String, Arc<dyn ProviderCredentialResolver>>,
43}
44
45impl ProviderRuntimeAdapters {
46 #[must_use]
47 pub fn production_defaults() -> Self {
48 Self::default().with_credential_resolver(
49 BEARER_ENV_TRUST_PROFILE,
50 Arc::new(EnvironmentBearerCredentialResolver),
51 )
52 }
53
54 #[must_use]
55 pub fn with_endpoint_resolver(
56 mut self,
57 source_id: impl Into<String>,
58 resolver: Arc<dyn ProviderEndpointResolver>,
59 ) -> Self {
60 self.endpoint_resolvers.insert(source_id.into(), resolver);
61 self
62 }
63
64 #[must_use]
65 pub fn with_credential_resolver(
66 mut self,
67 trust_profile: impl Into<String>,
68 resolver: Arc<dyn ProviderCredentialResolver>,
69 ) -> Self {
70 self.credential_resolvers
71 .insert(trust_profile.into(), resolver);
72 self
73 }
74}
75
76#[derive(Debug, Clone, Copy)]
77pub struct EnvironmentBearerCredentialResolver;
78
79#[async_trait]
80impl ProviderCredentialResolver for EnvironmentBearerCredentialResolver {
81 async fn resolve_bearer(&self, policy: &ServiceIdentityPolicy) -> AppResult<String> {
82 if policy.credential_references.len() != 1 {
83 return Err(AppError::new(
84 ErrorCode::Validation,
85 "bearer_env requires exactly one credential reference",
86 ));
87 }
88 let reference = &policy.credential_references[0];
89 let Some(name) = reference.strip_prefix("env://") else {
90 return Err(AppError::new(
91 ErrorCode::Validation,
92 "bearer_env accepts only opaque env:// credential references",
93 ));
94 };
95 if name.is_empty()
96 || !name.chars().all(|character| {
97 character == '_' || character.is_ascii_uppercase() || character.is_ascii_digit()
98 })
99 {
100 return Err(AppError::new(
101 ErrorCode::Validation,
102 "bearer_env credential reference contains an unsafe environment name",
103 ));
104 }
105 std::env::var(name)
106 .ok()
107 .filter(|value| !value.trim().is_empty())
108 .ok_or_else(|| {
109 AppError::new(
110 ErrorCode::Validation,
111 format!("Provider credential reference '{reference}' is unresolved"),
112 )
113 })
114 }
115}
116
117#[derive(Debug, Clone)]
123pub struct ProviderRuntimeAdapter {
124 plan: ProviderRuntimePlan,
125 adapters: ProviderRuntimeAdapters,
126 effects: ProviderHostEffectCoordinator,
127}
128
129#[derive(Debug)]
130pub struct LoadedProviderRuntime {
131 modules: Vec<Module>,
132 configs: Vec<ProviderConfig>,
133}
134
135impl ProviderRuntimeAdapter {
136 pub fn new(plan: ProviderRuntimePlan) -> AppResult<Self> {
137 if plan.protocol != PROVIDER_RUNTIME_PLAN_PROTOCOL {
138 return Err(AppError::new(
139 ErrorCode::Validation,
140 "Provider Runtime Plan protocol is unsupported",
141 ));
142 }
143 Ok(Self {
144 plan,
145 adapters: ProviderRuntimeAdapters::production_defaults(),
146 effects: ProviderHostEffectCoordinator::rejecting(),
147 })
148 }
149
150 pub fn with_adapters(
151 plan: ProviderRuntimePlan,
152 adapters: ProviderRuntimeAdapters,
153 ) -> AppResult<Self> {
154 let mut runtime = Self::new(plan)?;
155 runtime.adapters = adapters;
156 Ok(runtime)
157 }
158
159 #[must_use]
160 pub fn with_effect_coordinator(mut self, effects: ProviderHostEffectCoordinator) -> Self {
161 self.effects = effects;
162 self
163 }
164
165 pub async fn load_verified(self) -> AppResult<LoadedProviderRuntime> {
166 let mut modules = Vec::new();
167 let mut configs = Vec::new();
168 for provider in &self.plan.providers {
169 let bearer = resolve_identity(&self.adapters, provider).await?;
170 let endpoints = resolve_endpoints(&self.adapters, provider).await?;
171 let endpoint = select_endpoint(provider, &endpoints)?;
172 for locked in &provider.modules {
173 let config = module_config(
174 locked,
175 &provider.service_release.digest,
176 endpoint,
177 bearer.as_deref(),
178 );
179 let loaded = ProviderSource::new(config)?
180 .with_effect_coordinator(self.effects.clone())
181 .load_locked(
182 &provider.service_ref.service_id,
183 &provider.service_release.version,
184 &provider.service_release.digest,
185 &locked.export_key,
186 &locked.module_release_digest,
187 &locked.manifest_digest,
188 &locked.contract_digests,
189 &locked.manifest,
190 )
191 .await?;
192 configs.push(loaded.config);
193 modules.push(loaded.module);
194 }
195 }
196 Ok(LoadedProviderRuntime { modules, configs })
197 }
198}
199
200async fn resolve_identity(
201 adapters: &ProviderRuntimeAdapters,
202 provider: &ProviderRuntimeService,
203) -> AppResult<Option<String>> {
204 if provider
205 .endpoint_binding
206 .identity_policy
207 .credential_references
208 .is_empty()
209 {
210 return Ok(None);
211 }
212 let profile = &provider.endpoint_binding.identity_policy.trust_profile;
213 let resolver = adapters.credential_resolvers.get(profile).ok_or_else(|| {
214 AppError::new(
215 ErrorCode::Validation,
216 format!(
217 "Provider Service '{}' requires credential adapter '{}'",
218 provider.service_ref.service_id, profile
219 ),
220 )
221 })?;
222 resolver
223 .resolve_bearer(&provider.endpoint_binding.identity_policy)
224 .await
225 .map(Some)
226}
227
228impl LoadedProviderRuntime {
229 #[must_use]
230 pub fn proxy_registry(&self) -> ProviderHttpProxyRegistry {
231 ProviderHttpProxyRegistry::from_modules(&self.modules, &self.configs)
232 }
233
234 #[must_use]
235 pub fn into_modules(self) -> Vec<Module> {
236 self.modules
237 }
238
239 #[must_use]
240 pub fn into_parts(self) -> (Vec<Module>, Vec<ProviderConfig>) {
241 (self.modules, self.configs)
242 }
243}
244
245async fn resolve_endpoints(
246 adapters: &ProviderRuntimeAdapters,
247 provider: &ProviderRuntimeService,
248) -> AppResult<Vec<StaticEndpointDeclaration>> {
249 let (source_id, public_config, secret_references) =
250 match &provider.endpoint_binding.resolver_source {
251 EndpointResolverSource::Static { endpoints } => return Ok(endpoints.clone()),
252 EndpointResolverSource::LocalProcess { source_id } => {
253 (source_id.clone(), BTreeMap::new(), Vec::new())
254 }
255 EndpointResolverSource::Adapter {
256 adapter_id,
257 public_config,
258 secret_references,
259 } => (
260 adapter_id.clone(),
261 public_config.clone(),
262 secret_references.clone(),
263 ),
264 };
265 let resolver = adapters.endpoint_resolvers.get(&source_id).ok_or_else(|| {
266 AppError::new(
267 ErrorCode::Validation,
268 format!(
269 "Provider Service '{}' requires endpoint resolver adapter '{}'",
270 provider.service_ref.service_id, source_id
271 ),
272 )
273 })?;
274 resolver
275 .resolve(&ProviderEndpointResolutionRequest {
276 service_ref: provider.service_ref.clone(),
277 source_id,
278 public_config,
279 secret_references,
280 })
281 .await
282}
283
284fn select_endpoint<'a>(
285 provider: &ProviderRuntimeService,
286 endpoints: &'a [StaticEndpointDeclaration],
287) -> AppResult<&'a StaticEndpointDeclaration> {
288 let allowed = &provider.endpoint_binding.allowed_bindings;
289 let preferred_regions = &provider.endpoint_binding.selection_policy.preferred_regions;
290 endpoints
291 .iter()
292 .filter(|endpoint| {
293 allowed.contains(&endpoint.binding)
294 && matches!(
295 endpoint.binding,
296 ServiceTransportBinding::ProviderHttpJson
297 | ServiceTransportBinding::ProviderGrpc
298 )
299 && !endpoint.address.trim().is_empty()
300 })
301 .min_by_key(|endpoint| {
302 let region_rank = endpoint
303 .region
304 .as_ref()
305 .and_then(|region| preferred_regions.iter().position(|item| item == region))
306 .unwrap_or(usize::MAX);
307 (region_rank, endpoint.priority, endpoint.address.as_str())
308 })
309 .ok_or_else(|| {
310 AppError::new(
311 ErrorCode::Validation,
312 format!(
313 "Provider Service '{}' has no eligible resolved endpoint",
314 provider.service_ref.service_id
315 ),
316 )
317 })
318}
319
320#[derive(Debug, Clone, Default)]
321pub struct FixedProviderEndpointResolver {
322 endpoints: BTreeMap<ServiceReference, Vec<StaticEndpointDeclaration>>,
323}
324
325impl FixedProviderEndpointResolver {
326 #[must_use]
327 pub fn new(
328 endpoints: impl IntoIterator<Item = (ServiceReference, Vec<StaticEndpointDeclaration>)>,
329 ) -> Self {
330 Self {
331 endpoints: endpoints.into_iter().collect(),
332 }
333 }
334}
335
336#[async_trait]
337impl ProviderEndpointResolver for FixedProviderEndpointResolver {
338 async fn resolve(
339 &self,
340 request: &ProviderEndpointResolutionRequest,
341 ) -> AppResult<Vec<StaticEndpointDeclaration>> {
342 self.endpoints
343 .get(&request.service_ref)
344 .cloned()
345 .ok_or_else(|| {
346 AppError::new(
347 ErrorCode::Validation,
348 format!(
349 "endpoint resolver '{}' has no state for Provider Service '{}'",
350 request.source_id, request.service_ref.service_id
351 ),
352 )
353 })
354 }
355}
356
357#[derive(Debug, Clone)]
358pub struct FixedBearerCredentialResolver {
359 bearer: String,
360}
361
362impl FixedBearerCredentialResolver {
363 #[must_use]
364 pub fn new(bearer: impl Into<String>) -> Self {
365 Self {
366 bearer: bearer.into(),
367 }
368 }
369}
370
371#[async_trait]
372impl ProviderCredentialResolver for FixedBearerCredentialResolver {
373 async fn resolve_bearer(&self, _policy: &ServiceIdentityPolicy) -> AppResult<String> {
374 if self.bearer.trim().is_empty() {
375 return Err(AppError::new(
376 ErrorCode::Validation,
377 "fixed Provider bearer credential is empty",
378 ));
379 }
380 Ok(self.bearer.clone())
381 }
382}
383
384fn module_config(
385 locked: &ProviderRuntimeModule,
386 service_release_digest: &str,
387 endpoint: &StaticEndpointDeclaration,
388 bearer: Option<&str>,
389) -> ProviderConfig {
390 let transport = match endpoint.binding {
391 ServiceTransportBinding::ProviderHttpJson => ProviderTransport::HttpJson,
392 ServiceTransportBinding::ProviderGrpc => ProviderTransport::Grpc,
393 _ => unreachable!("endpoint selection accepts only Provider transports"),
394 };
395 let config = ProviderConfig::new(&locked.module_id, &endpoint.address)
396 .with_transport(transport, &endpoint.address)
397 .with_export_key(&locked.export_key)
398 .with_locked_contract(
399 service_release_digest,
400 &locked.module_release_digest,
401 &locked.manifest_digest,
402 locked.contract_digests.clone(),
403 )
404 .with_allowed_host_functions(
405 locked
406 .manifest
407 .runtime
408 .iter()
409 .flat_map(|runtime| runtime.functions.iter())
410 .map(|function| function.name.clone()),
411 );
412 match bearer {
413 Some(bearer) => config.with_auth_token(bearer),
414 None => config,
415 }
416}