1use crate::{
2 ProviderConfig, ProviderHostEffectBatch, ProviderHostEventEffect,
3 ProviderHostRuntimeFunctionRequest, ProviderInvocation, ProviderOutcome, ProviderOutcomeStatus,
4};
5use platform_core::{
6 AppError, AppResult, CorrelationId, DbPool, ErrorCode, OutboxEvent, OutboxPublisher, TenantId,
7};
8use platform_runtime::{EnqueueFunctionRequest, FunctionTenancyMode, RuntimeClient};
9use sha2::{Digest, Sha256};
10use std::collections::HashSet;
11use std::fmt::Write as _;
12
13const MAX_PROVIDER_HOST_EFFECTS: usize = 100;
14
15#[derive(Debug, Clone)]
16pub struct ProviderHostEffectCoordinator {
17 pool: Option<DbPool>,
18}
19
20impl ProviderHostEffectCoordinator {
21 #[must_use]
22 pub fn new(pool: DbPool) -> Self {
23 Self { pool: Some(pool) }
24 }
25
26 #[must_use]
27 pub fn rejecting() -> Self {
28 Self { pool: None }
29 }
30
31 pub async fn commit(
32 &self,
33 config: &ProviderConfig,
34 invocation: &ProviderInvocation,
35 outcome: &ProviderOutcome,
36 ) -> AppResult<()> {
37 let effects = &outcome.host_effects;
38 if effects.events.is_empty() && effects.runtime_function_requests.is_empty() {
39 return Ok(());
40 }
41 let pool = self.pool.as_ref().ok_or_else(|| {
42 AppError::new(
43 ErrorCode::Internal,
44 "Provider Host effect coordinator is not configured",
45 )
46 })?;
47 if !matches!(outcome.status, ProviderOutcomeStatus::Succeeded) {
48 return Err(AppError::new(
49 ErrorCode::ExternalDependency,
50 "Provider returned Host effects for a non-succeeded outcome",
51 ));
52 }
53 validate_effects(config, invocation, effects)?;
54 let effects_digest = digest(effects)?;
55 let service_release_digest = config.service_release_digest.as_deref().ok_or_else(|| {
56 AppError::new(
57 ErrorCode::Internal,
58 "Provider Service Release is not locked",
59 )
60 })?;
61 let module_release_digest = config.module_release_digest.as_deref().ok_or_else(|| {
62 AppError::new(ErrorCode::Internal, "Provider Module Release is not locked")
63 })?;
64
65 let mut tx = pool.begin().await.map_err(map_store_error)?;
66 let inserted = sqlx::query_scalar::<_, bool>(
67 r#"
68 insert into platform.provider_host_effect_commits (
69 invocation_id, outcome_digest, effects_digest,
70 service_release_digest, module_release_digest, export_key
71 )
72 values ($1, $2, $3, $4, $5, $6)
73 on conflict (invocation_id) do nothing
74 returning true
75 "#,
76 )
77 .bind(&invocation.invocation_id)
78 .bind(&outcome.outcome_digest)
79 .bind(&effects_digest)
80 .bind(service_release_digest)
81 .bind(module_release_digest)
82 .bind(&config.export_key)
83 .fetch_optional(&mut *tx)
84 .await
85 .map_err(map_store_error)?
86 .unwrap_or(false);
87
88 if !inserted {
89 let existing = sqlx::query_as::<_, (String, String)>(
90 r#"
91 select outcome_digest, effects_digest
92 from platform.provider_host_effect_commits
93 where invocation_id = $1
94 "#,
95 )
96 .bind(&invocation.invocation_id)
97 .fetch_one(&mut *tx)
98 .await
99 .map_err(map_store_error)?;
100 if existing != (outcome.outcome_digest.clone(), effects_digest) {
101 return Err(AppError::new(
102 ErrorCode::Conflict,
103 "Provider invocation attempted to rebind committed Host effects",
104 ));
105 }
106 tx.commit().await.map_err(map_store_error)?;
107 return Ok(());
108 }
109
110 let outbox = OutboxPublisher;
111 for event in &effects.events {
112 outbox.publish_in_tx(&mut tx, &outbox_event(event)).await?;
113 }
114 let runtime = RuntimeClient::new(pool.clone());
115 for request in &effects.runtime_function_requests {
116 runtime
117 .enqueue_function_with_id_in_tx(
118 &mut tx,
119 &request.request_id,
120 runtime_request(request),
121 )
122 .await?;
123 }
124 tx.commit().await.map_err(map_store_error)
125 }
126
127 pub async fn mark_acknowledged(
128 &self,
129 invocation_id: &str,
130 outcome_digest: &str,
131 ) -> AppResult<()> {
132 let Some(pool) = self.pool.as_ref() else {
133 return Ok(());
134 };
135 sqlx::query(
136 r#"
137 update platform.provider_host_effect_commits
138 set acknowledged_at = coalesce(acknowledged_at, now())
139 where invocation_id = $1 and outcome_digest = $2
140 "#,
141 )
142 .bind(invocation_id)
143 .bind(outcome_digest)
144 .execute(pool)
145 .await
146 .map(|_| ())
147 .map_err(map_store_error)
148 }
149}
150
151fn validate_effects(
152 config: &ProviderConfig,
153 invocation: &ProviderInvocation,
154 effects: &ProviderHostEffectBatch,
155) -> AppResult<()> {
156 if effects.events.len() + effects.runtime_function_requests.len() > MAX_PROVIDER_HOST_EFFECTS {
157 return Err(AppError::new(
158 ErrorCode::Validation,
159 format!("Provider Host effects exceed the {MAX_PROVIDER_HOST_EFFECTS} effect limit"),
160 ));
161 }
162 let mut event_ids = HashSet::with_capacity(effects.events.len());
163 for event in &effects.events {
164 if event.event_id.trim().is_empty()
165 || event.event_name.trim().is_empty()
166 || event.aggregate_type.trim().is_empty()
167 || event.aggregate_id.trim().is_empty()
168 || event.source_module != config.name
169 || event.correlation_id != invocation.correlation_id
170 || !event_ids.insert(event.event_id.as_str())
171 {
172 return Err(AppError::new(
173 ErrorCode::Validation,
174 "Provider Host Event effect is not bound to the locked Module invocation",
175 ));
176 }
177 }
178 let mut request_ids = HashSet::with_capacity(effects.runtime_function_requests.len());
179 for request in &effects.runtime_function_requests {
180 if request.request_id.trim().is_empty()
181 || request.correlation_id != invocation.correlation_id
182 || !config
183 .allowed_host_function_names
184 .contains(&request.function_name)
185 || request.actor != invocation.actor
186 || request.tenant_id != invocation.tenant_id
187 || request.trace != invocation.trace
188 || !request_ids.insert(request.request_id.as_str())
189 || request
190 .max_attempts
191 .is_some_and(|value| !(1..=100).contains(&value))
192 {
193 return Err(AppError::new(
194 ErrorCode::Validation,
195 "Provider Runtime Function effect is not bound to the locked Module invocation",
196 ));
197 }
198 }
199 Ok(())
200}
201
202fn outbox_event(effect: &ProviderHostEventEffect) -> OutboxEvent {
203 OutboxEvent {
204 id: effect.event_id.clone(),
205 event_name: effect.event_name.clone(),
206 event_version: effect.event_version,
207 source_module: effect.source_module.clone(),
208 aggregate_type: effect.aggregate_type.clone(),
209 aggregate_id: effect.aggregate_id.clone(),
210 correlation_id: effect.correlation_id.clone(),
211 causation_id: effect.causation_id.clone(),
212 occurred_at: effect.occurred_at,
213 payload: effect.payload.clone(),
214 headers: effect.headers.clone(),
215 }
216}
217
218fn runtime_request(effect: &ProviderHostRuntimeFunctionRequest) -> EnqueueFunctionRequest {
219 EnqueueFunctionRequest {
220 function_name: effect.function_name.clone(),
221 input_json: effect.input.clone(),
222 correlation_id: CorrelationId::new(effect.correlation_id.clone()),
223 actor: effect.actor.clone(),
224 tenant_id: effect.tenant_id.clone().map(TenantId),
225 tenancy_mode: FunctionTenancyMode::Optional,
226 trace: effect.trace.clone(),
227 causation_id: effect.causation_id.clone(),
228 max_attempts: effect.max_attempts,
229 }
230}
231
232fn digest(value: &ProviderHostEffectBatch) -> AppResult<String> {
233 let encoded = serde_json_canonicalizer::to_vec(value).map_err(|error| {
234 AppError::new(
235 ErrorCode::Internal,
236 format!("Provider Host effects could not be encoded: {error}"),
237 )
238 })?;
239 let mut digest = String::from("sha256:");
240 for byte in Sha256::digest(encoded) {
241 write!(digest, "{byte:02x}").expect("writing to a String cannot fail");
242 }
243 Ok(digest)
244}
245
246fn map_store_error(error: sqlx::Error) -> AppError {
247 AppError::new(
248 ErrorCode::Internal,
249 format!("Provider Host effect Store operation failed: {error}"),
250 )
251}
252
253#[cfg(test)]
254mod tests {
255 use super::*;
256 use crate::{
257 PROVIDER_PROTOCOL, ProviderHostEventEffect, ProviderHostRuntimeFunctionRequest,
258 ProviderInvocationMode, ProviderOperationKind,
259 };
260 use chrono::Utc;
261 use platform_core::{ActorContext, TraceContext, apply_migrations};
262 use platform_testing::TestDatabase;
263 use serde_json::json;
264
265 #[tokio::test]
266 async fn commits_host_effects_atomically_and_replays_without_duplicates() {
267 let Some(db) = TestDatabase::create().await else {
268 return;
269 };
270 apply_migrations(&db.pool, platform_core::PLATFORM_MIGRATIONS)
271 .await
272 .unwrap();
273 apply_migrations(&db.pool, platform_runtime::RUNTIME_MIGRATIONS)
274 .await
275 .unwrap();
276
277 let config = ProviderConfig::new("lenso/support", "http://provider.test")
278 .with_export_key("support")
279 .with_locked_contract(
280 digest_value('1'),
281 digest_value('2'),
282 digest_value('3'),
283 vec![],
284 )
285 .with_allowed_host_functions(["support.follow_up.v1".to_owned()]);
286 let base_invocation = invocation("invocation-1", "correlation-1");
287 let outcome = outcome("invocation-1", "correlation-1");
288 let coordinator = ProviderHostEffectCoordinator::new(db.pool.clone());
289
290 coordinator
291 .commit(&config, &base_invocation, &outcome)
292 .await
293 .unwrap();
294 coordinator
295 .commit(&config, &base_invocation, &outcome)
296 .await
297 .unwrap();
298
299 assert_eq!(
300 count(&db.pool, "platform.provider_host_effect_commits").await,
301 1
302 );
303 assert_eq!(count(&db.pool, "platform.outbox").await, 1);
304 assert_eq!(count(&db.pool, "runtime.function_runs").await, 1);
305
306 let mut rebound = outcome;
307 rebound.outcome_digest = digest_value('9');
308 let error = coordinator
309 .commit(&config, &base_invocation, &rebound)
310 .await
311 .expect_err("committed invocation identity cannot be rebound");
312 assert_eq!(error.code, ErrorCode::Conflict);
313
314 db.cleanup().await;
315 }
316
317 #[tokio::test]
318 async fn stable_host_effects_replay_across_technical_invocation_identities() {
319 let Some(db) = TestDatabase::create().await else {
320 return;
321 };
322 apply_migrations(&db.pool, platform_core::PLATFORM_MIGRATIONS)
323 .await
324 .unwrap();
325 apply_migrations(&db.pool, platform_runtime::RUNTIME_MIGRATIONS)
326 .await
327 .unwrap();
328
329 let config = ProviderConfig::new("lenso/support", "http://provider.test")
330 .with_export_key("support")
331 .with_locked_contract(
332 digest_value('1'),
333 digest_value('2'),
334 digest_value('3'),
335 vec![],
336 )
337 .with_allowed_host_functions(["support.follow_up.v1".to_owned()]);
338 let first_invocation = invocation("invocation-attempt-1", "correlation-1");
339 let first_outcome = outcome("invocation-attempt-1", "correlation-1");
340 let second_invocation = invocation("invocation-attempt-2", "correlation-1");
341 let mut second_outcome = outcome("invocation-attempt-2", "correlation-1");
342 second_outcome.host_effects = first_outcome.host_effects.clone();
343 second_outcome.outcome_digest = digest_value('9');
344 let coordinator = ProviderHostEffectCoordinator::new(db.pool.clone());
345
346 coordinator
347 .commit(&config, &first_invocation, &first_outcome)
348 .await
349 .unwrap();
350 coordinator
351 .commit(&config, &second_invocation, &second_outcome)
352 .await
353 .expect("stable effects should survive a new technical invocation identity");
354
355 assert_eq!(
356 count(&db.pool, "platform.provider_host_effect_commits").await,
357 2
358 );
359 assert_eq!(count(&db.pool, "platform.outbox").await, 1);
360 assert_eq!(count(&db.pool, "runtime.function_runs").await, 1);
361
362 db.cleanup().await;
363 }
364
365 #[test]
366 fn rejects_unbounded_or_duplicate_host_effects_before_commit() {
367 let config = ProviderConfig::new("lenso/support", "http://provider.test")
368 .with_allowed_host_functions(["support.follow_up.v1".to_owned()]);
369 let base_invocation = invocation("invocation-1", "correlation-1");
370 let mut effects = outcome("invocation-1", "correlation-1").host_effects;
371 effects.events.push(effects.events[0].clone());
372
373 let duplicate = validate_effects(&config, &base_invocation, &effects)
374 .expect_err("duplicate Host Event identities must be rejected");
375 assert_eq!(duplicate.code, ErrorCode::Validation);
376
377 let mut effects = ProviderHostEffectBatch::default();
378 effects.events = (0..=MAX_PROVIDER_HOST_EFFECTS)
379 .map(|index| ProviderHostEventEffect {
380 event_id: format!("event-{index}"),
381 event_name: "support.updated.v1".to_owned(),
382 event_version: 1,
383 source_module: "lenso/support".to_owned(),
384 aggregate_type: "ticket".to_owned(),
385 aggregate_id: format!("ticket-{index}"),
386 correlation_id: "correlation-1".to_owned(),
387 causation_id: Some("invocation-1".to_owned()),
388 occurred_at: Utc::now(),
389 payload: json!({}),
390 headers: json!({}),
391 })
392 .collect();
393
394 let unbounded = validate_effects(&config, &base_invocation, &effects)
395 .expect_err("Host effect batches must be bounded");
396 assert_eq!(unbounded.code, ErrorCode::Validation);
397
398 let mut elevated = outcome("invocation-1", "correlation-1").host_effects;
399 elevated.runtime_function_requests[0].actor = ActorContext::System;
400 let invocation = ProviderInvocation {
401 actor: ActorContext::User {
402 user_id: "user-1".to_owned(),
403 scopes: vec!["support.read".to_owned()],
404 },
405 ..invocation("invocation-1", "correlation-1")
406 };
407 let privilege_error = validate_effects(&config, &invocation, &elevated)
408 .expect_err("a Provider must not mint Host Runtime authority");
409 assert_eq!(privilege_error.code, ErrorCode::Validation);
410 }
411
412 fn invocation(id: &str, correlation_id: &str) -> ProviderInvocation {
413 ProviderInvocation {
414 protocol: PROVIDER_PROTOCOL.to_owned(),
415 invocation_id: id.to_owned(),
416 request_id: id.to_owned(),
417 attempt: 1,
418 deadline: Utc::now().to_rfc3339(),
419 service_release_digest: digest_value('1'),
420 export_key: "support".to_owned(),
421 module_release_digest: digest_value('2'),
422 manifest_digest: digest_value('3'),
423 operation_kind: ProviderOperationKind::AdminAction,
424 operation_name: "support.act".to_owned(),
425 operation_version: "1".to_owned(),
426 mode: ProviderInvocationMode::Durable,
427 input_contract_digest: digest_value('4'),
428 output_contract_digest: digest_value('4'),
429 tenant_id: None,
430 actor: ActorContext::System,
431 delegation: None,
432 locale: None,
433 context: Default::default(),
434 correlation_id: correlation_id.to_owned(),
435 causation_id: None,
436 trace: TraceContext::default(),
437 content_type: "application/json".to_owned(),
438 payload: json!({}),
439 }
440 }
441
442 fn outcome(id: &str, correlation_id: &str) -> ProviderOutcome {
443 ProviderOutcome {
444 protocol: PROVIDER_PROTOCOL.to_owned(),
445 invocation_id: id.to_owned(),
446 status: ProviderOutcomeStatus::Succeeded,
447 result: Some(json!({ "ok": true })),
448 error: None,
449 effect_evidence: vec![],
450 host_effects: ProviderHostEffectBatch {
451 events: vec![ProviderHostEventEffect {
452 event_id: "event-1".to_owned(),
453 event_name: "support.updated.v1".to_owned(),
454 event_version: 1,
455 source_module: "lenso/support".to_owned(),
456 aggregate_type: "ticket".to_owned(),
457 aggregate_id: "ticket-1".to_owned(),
458 correlation_id: correlation_id.to_owned(),
459 causation_id: Some(id.to_owned()),
460 occurred_at: Utc::now(),
461 payload: json!({ "ticketId": "ticket-1" }),
462 headers: json!({}),
463 }],
464 runtime_function_requests: vec![ProviderHostRuntimeFunctionRequest {
465 request_id: "fnrun-provider-effect-1".to_owned(),
466 function_name: "support.follow_up.v1".to_owned(),
467 input: json!({ "ticketId": "ticket-1" }),
468 correlation_id: correlation_id.to_owned(),
469 actor: ActorContext::System,
470 tenant_id: None,
471 trace: TraceContext::default(),
472 causation_id: Some(id.to_owned()),
473 max_attempts: Some(3),
474 }],
475 },
476 outcome_digest: digest_value('8'),
477 }
478 }
479
480 fn digest_value(character: char) -> String {
481 format!("sha256:{}", character.to_string().repeat(64))
482 }
483
484 async fn count(pool: &DbPool, table: &str) -> i64 {
485 let query = format!("select count(*) from {table}");
486 sqlx::query_scalar(sqlx::AssertSqlSafe(query))
487 .fetch_one(pool)
488 .await
489 .unwrap()
490 }
491}