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