platform_module_remote/
runtime.rs

1use crate::config::{RemoteModuleConfig, RemoteModuleTransport};
2use crate::protocol::{RemoteFunctionInvokeRequest, RemoteFunctionInvokeResponse};
3use crate::response::{ResponseBodyPolicy, decode_json_response_with_policy};
4use crate::validation::validate_path_segment;
5use platform_core::{AppError, AppResult, ErrorCode, ExecutionContext};
6use platform_runtime::{FunctionHandlerObservability, RuntimeFunction};
7use serde_json::Value;
8use std::time::Duration;
9
10const MAX_RUNTIME_FUNCTION_RESPONSE_BYTES: u64 = 4 * 1024 * 1024;
11
12#[derive(Debug, Clone)]
13pub struct RemoteRuntimeFunction {
14    client: reqwest::Client,
15    config: RemoteModuleConfig,
16    function_name: String,
17}
18
19impl RemoteRuntimeFunction {
20    pub fn new(config: RemoteModuleConfig, function_name: impl Into<String>) -> AppResult<Self> {
21        let function_name = function_name.into();
22        validate_function_name(&function_name)?;
23        let client = reqwest::Client::builder()
24            .timeout(Duration::from_millis(config.timeout_ms))
25            .build()
26            .map_err(|error| {
27                AppError::new(
28                    ErrorCode::Internal,
29                    format!("failed to build remote runtime client: {error}"),
30                )
31            })?;
32        Ok(Self {
33            client,
34            config,
35            function_name,
36        })
37    }
38
39    pub async fn invoke(&self, ctx: ExecutionContext, input: Value) -> AppResult<Value> {
40        let request_body = RemoteFunctionInvokeRequest {
41            request_id: ctx.execution_id.0.clone(),
42            function_run_id: ctx.execution_id.0,
43            function_name: self.function_name.clone(),
44            attempt: ctx.attempt,
45            correlation_id: ctx.correlation_id.0,
46            causation_id: ctx.causation_id,
47            actor: ctx.actor,
48            trace: ctx.trace,
49            input,
50        };
51        if self.config.transport == RemoteModuleTransport::Grpc {
52            return crate::grpc::invoke_function(&self.config, &request_body)
53                .await
54                .map(|response| response.output);
55        }
56
57        let mut request = self.client.post(self.invoke_url()).json(&request_body);
58        if let Some(token) = &self.config.auth_token {
59            request = request.bearer_auth(token);
60        }
61
62        let response = request.send().await.map_err(|error| {
63            AppError::new(
64                ErrorCode::ExternalDependency,
65                format!(
66                    "remote runtime function {} request failed: {error}",
67                    self.function_name
68                ),
69            )
70            .retryable()
71        })?;
72
73        let response = decode_json_response_with_policy::<RemoteFunctionInvokeResponse>(
74            response,
75            "runtime function invoke",
76            false,
77            ResponseBodyPolicy {
78                max_bytes: Some(MAX_RUNTIME_FUNCTION_RESPONSE_BYTES),
79                require_json_content_type: true,
80                allow_empty_success: false,
81            },
82        )
83        .await?
84        .ok_or_else(|| {
85            AppError::new(
86                ErrorCode::NotFound,
87                format!("remote runtime function {} not found", self.function_name),
88            )
89        })?;
90        Ok(response.output)
91    }
92
93    fn invoke_url(&self) -> String {
94        format!(
95            "{}/runtime/functions/{}/invoke",
96            self.config.base_url, self.function_name
97        )
98    }
99}
100
101#[async_trait::async_trait]
102impl RuntimeFunction for RemoteRuntimeFunction {
103    async fn call(&self, ctx: ExecutionContext, input: Value) -> AppResult<Value> {
104        self.invoke(ctx, input).await
105    }
106
107    fn observability(&self) -> Option<FunctionHandlerObservability> {
108        Some(FunctionHandlerObservability::new(
109            "remote_runtime",
110            serde_json::json!({
111                "module_name": &self.config.name,
112                "function_name": &self.function_name,
113                "remote_path": format!("/runtime/functions/{}/invoke", self.function_name),
114                "timeout_ms": self.config.timeout_ms,
115            }),
116        ))
117    }
118}
119
120pub(crate) fn validate_function_name(function_name: &str) -> AppResult<()> {
121    validate_path_segment(
122        function_name,
123        "remote runtime function name must be a stable path segment",
124    )
125}
platform_module_remote/runtime.rs

platform_module_remote/
runtime.rs
