platform_module_remote/
event.rs

1use crate::config::{RemoteModuleConfig, RemoteModuleTransport};
2use crate::protocol::{
3    RemoteEventHandleRequest, RemoteEventHandleResponse, RemoteEventResultAction,
4};
5use crate::response::{ResponseBodyPolicy, decode_json_response_with_policy};
6use crate::validation::validate_path_segment;
7use platform_core::{
8    ActorContext, AppError, AppResult, ClaimedOutboxEvent, CorrelationId, ErrorCode, EventHandler,
9    trace_context_from_headers,
10};
11use platform_runtime::{EnqueueFunctionRequest, FunctionRegistry, RuntimeClient};
12use std::collections::BTreeSet;
13use std::sync::Arc;
14use std::time::Duration;
15
16const MAX_EVENT_HANDLER_RESPONSE_BYTES: u64 = 1024 * 1024;
17const MAX_EVENT_HANDLER_RESULT_ACTIONS: usize = 1;
18
19#[derive(Debug, Clone)]
20pub struct RemoteEventHandler {
21    client: reqwest::Client,
22    config: RemoteModuleConfig,
23    handler_name: String,
24    event_name: String,
25    action_runner: Arc<dyn RemoteEventActionRunner>,
26}
27
28impl RemoteEventHandler {
29    pub fn new(
30        config: RemoteModuleConfig,
31        handler_name: impl Into<String>,
32        event_name: impl Into<String>,
33    ) -> AppResult<Self> {
34        let handler_name = handler_name.into();
35        let event_name = event_name.into();
36        validate_event_handler_name(&handler_name)?;
37        validate_event_name(&event_name)?;
38        let client = reqwest::Client::builder()
39            .timeout(Duration::from_millis(config.timeout_ms))
40            .build()
41            .map_err(|error| {
42                AppError::new(
43                    ErrorCode::Internal,
44                    format!("failed to build remote event handler client: {error}"),
45                )
46            })?;
47        Ok(Self {
48            client,
49            config,
50            handler_name,
51            event_name,
52            action_runner: Arc::new(RejectingRemoteEventActionRunner),
53        })
54    }
55
56    #[must_use]
57    pub fn with_host_action_runner(mut self, action_runner: RemoteEventHostActionRunner) -> Self {
58        self.action_runner = Arc::new(action_runner);
59        self
60    }
61
62    pub async fn invoke(&self, event: &ClaimedOutboxEvent) -> AppResult<()> {
63        let request_body = RemoteEventHandleRequest {
64            request_id: format!("{}:{}", event.id, self.handler_name),
65            outbox_event_id: event.id.clone(),
66            handler_name: self.handler_name.clone(),
67            event_name: event.event_name.clone(),
68            event_version: event.event_version,
69            source_module: event.source_module.clone(),
70            aggregate_type: event.aggregate_type.clone(),
71            aggregate_id: event.aggregate_id.clone(),
72            correlation_id: event.correlation_id.clone(),
73            causation_id: event.causation_id.clone(),
74            occurred_at: event.occurred_at.to_rfc3339(),
75            actor: actor_from_event(event),
76            trace: trace_context_from_headers(&event.headers),
77            payload: event.payload.clone(),
78            headers: event.headers.clone(),
79        };
80        if self.config.transport == RemoteModuleTransport::Grpc {
81            let response = crate::grpc::handle_event(&self.config, &request_body).await?;
82            self.action_runner
83                .run_actions(event, &self.handler_name, response.actions)
84                .await?;
85            return Ok(());
86        }
87
88        let mut request = self.client.post(self.invoke_url()).json(&request_body);
89        if let Some(token) = &self.config.auth_token {
90            request = request.bearer_auth(token);
91        }
92
93        let response = request.send().await.map_err(|error| {
94            AppError::new(
95                ErrorCode::ExternalDependency,
96                format!(
97                    "remote event handler {} request failed: {error}",
98                    self.handler_name
99                ),
100            )
101            .retryable()
102        })?;
103
104        let response = decode_json_response_with_policy::<RemoteEventHandleResponse>(
105            response,
106            "event handler invoke",
107            false,
108            ResponseBodyPolicy {
109                max_bytes: Some(MAX_EVENT_HANDLER_RESPONSE_BYTES),
110                require_json_content_type: true,
111                allow_empty_success: true,
112            },
113        )
114        .await?;
115        if let Some(response) = response {
116            self.action_runner
117                .run_actions(event, &self.handler_name, response.actions)
118                .await?;
119        }
120        Ok(())
121    }
122
123    fn invoke_url(&self) -> String {
124        format!(
125            "{}/events/handlers/{}/invoke",
126            self.config.base_url, self.handler_name
127        )
128    }
129}
130
131#[async_trait::async_trait]
132impl EventHandler for RemoteEventHandler {
133    fn event_name(&self) -> &str {
134        &self.event_name
135    }
136
137    async fn handle(&self, event: &ClaimedOutboxEvent) -> AppResult<()> {
138        self.invoke(event).await
139    }
140}
141
142#[async_trait::async_trait]
143trait RemoteEventActionRunner: std::fmt::Debug + Send + Sync {
144    async fn run_actions(
145        &self,
146        event: &ClaimedOutboxEvent,
147        handler_name: &str,
148        actions: Vec<RemoteEventResultAction>,
149    ) -> AppResult<()>;
150}
151
152#[derive(Debug, Clone)]
153pub struct RemoteEventHostActionRunner {
154    runtime: RuntimeClient,
155    function_registry: Arc<FunctionRegistry>,
156    allowed_function_names: BTreeSet<String>,
157}
158
159impl RemoteEventHostActionRunner {
160    #[must_use]
161    pub fn new(
162        runtime: RuntimeClient,
163        function_registry: Arc<FunctionRegistry>,
164        allowed_function_names: impl IntoIterator<Item = String>,
165    ) -> Self {
166        Self {
167            runtime,
168            function_registry,
169            allowed_function_names: allowed_function_names.into_iter().collect(),
170        }
171    }
172}
173
174#[async_trait::async_trait]
175impl RemoteEventActionRunner for RemoteEventHostActionRunner {
176    async fn run_actions(
177        &self,
178        event: &ClaimedOutboxEvent,
179        handler_name: &str,
180        actions: Vec<RemoteEventResultAction>,
181    ) -> AppResult<()> {
182        if actions.len() > MAX_EVENT_HANDLER_RESULT_ACTIONS {
183            return Err(AppError::new(
184                ErrorCode::Validation,
185                format!(
186                    "remote event handler {handler_name} returned too many result actions: {}",
187                    actions.len()
188                ),
189            ));
190        }
191
192        for (index, action) in actions.into_iter().enumerate() {
193            match action {
194                RemoteEventResultAction::EnqueueFunction {
195                    function_name,
196                    input,
197                } => {
198                    self.enqueue_function(event, handler_name, index, function_name, input)
199                        .await?;
200                }
201            }
202        }
203
204        Ok(())
205    }
206}
207
208impl RemoteEventHostActionRunner {
209    async fn enqueue_function(
210        &self,
211        event: &ClaimedOutboxEvent,
212        handler_name: &str,
213        action_index: usize,
214        function_name: String,
215        input: serde_json::Value,
216    ) -> AppResult<()> {
217        if !self.allowed_function_names.contains(&function_name) {
218            return Err(AppError::new(
219                ErrorCode::Validation,
220                format!(
221                    "remote event handler {handler_name} requested runtime function {function_name} that is not declared by its module"
222                ),
223            ));
224        }
225
226        let definition = self.function_registry.get(&function_name).ok_or_else(|| {
227            AppError::new(
228                ErrorCode::Internal,
229                format!("remote event handler {handler_name} requested unregistered runtime function {function_name}"),
230            )
231        })?;
232        let run_id = self
233            .runtime
234            .enqueue_function(EnqueueFunctionRequest {
235                function_name: function_name.clone(),
236                input_json: input,
237                correlation_id: CorrelationId::new(event.correlation_id.clone()),
238                actor: actor_from_event(event),
239                trace: trace_context_from_headers(&event.headers),
240                causation_id: Some(format!(
241                    "remote_event_handler:{}:{handler_name}:{action_index}",
242                    event.id
243                )),
244                max_attempts: Some(runtime_max_attempts_for_enqueue(
245                    definition.retry_policy.max_attempts,
246                )),
247            })
248            .await?;
249
250        tracing::info!(
251            outbox_event_id = %event.id,
252            handler_name = %handler_name,
253            function_name = %function_name,
254            function_run_id = %run_id,
255            "remote event handler enqueued runtime function"
256        );
257
258        Ok(())
259    }
260}
261
262#[derive(Debug)]
263struct RejectingRemoteEventActionRunner;
264
265#[async_trait::async_trait]
266impl RemoteEventActionRunner for RejectingRemoteEventActionRunner {
267    async fn run_actions(
268        &self,
269        _event: &ClaimedOutboxEvent,
270        handler_name: &str,
271        actions: Vec<RemoteEventResultAction>,
272    ) -> AppResult<()> {
273        if actions.is_empty() {
274            return Ok(());
275        }
276
277        Err(AppError::new(
278            ErrorCode::Validation,
279            format!(
280                "remote event handler {handler_name} returned result actions but host actions are not configured"
281            ),
282        ))
283    }
284}
285
286fn actor_from_event(event: &ClaimedOutboxEvent) -> ActorContext {
287    event
288        .headers
289        .get("actor")
290        .cloned()
291        .and_then(|actor| serde_json::from_value(actor).ok())
292        .unwrap_or_default()
293}
294
295pub(crate) fn validate_event_handler_name(value: &str) -> AppResult<()> {
296    validate_path_segment(
297        value,
298        "remote event handler name must be a stable path segment",
299    )
300}
301
302pub(crate) fn validate_event_name(value: &str) -> AppResult<()> {
303    validate_path_segment(value, "remote event name must be a stable path segment")
304}
305
306fn runtime_max_attempts_for_enqueue(max_attempts: u32) -> i32 {
307    i32::try_from(max_attempts).unwrap_or(i32::MAX)
308}
platform_module_remote/event.rs

platform_module_remote/
event.rs
