1use crate::context::RequestContext;
2use crate::db::DbPool;
3use crate::error::{AppError, AppResult, ErrorCode};
4use chrono::{DateTime, Duration, Utc};
5use serde_json::Value;
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum ProviderHttpBodyCaptureStatus {
9 Captured,
10 NotApplicable,
11 NotCaptured,
12}
13
14impl ProviderHttpBodyCaptureStatus {
15 #[must_use]
16 pub const fn as_str(self) -> &'static str {
17 match self {
18 Self::Captured => "captured",
19 Self::NotApplicable => "not_applicable",
20 Self::NotCaptured => "not_captured",
21 }
22 }
23}
24
25#[derive(Debug, Clone)]
26pub struct ProviderHttpBodyEvidence {
27 body: Option<Value>,
28 capture_status: ProviderHttpBodyCaptureStatus,
29 capture_reason: Option<String>,
30 observed_bytes: Option<i64>,
31}
32
33impl ProviderHttpBodyEvidence {
34 #[must_use]
35 pub fn captured(body: Value, observed_bytes: usize) -> Self {
36 Self {
37 body: Some(body),
38 capture_status: ProviderHttpBodyCaptureStatus::Captured,
39 capture_reason: None,
40 observed_bytes: Some(i64::try_from(observed_bytes).unwrap_or(i64::MAX)),
41 }
42 }
43
44 #[must_use]
45 pub fn not_applicable(reason: impl Into<String>) -> Self {
46 Self {
47 body: None,
48 capture_status: ProviderHttpBodyCaptureStatus::NotApplicable,
49 capture_reason: Some(reason.into()),
50 observed_bytes: None,
51 }
52 }
53
54 #[must_use]
55 pub fn not_captured(reason: impl Into<String>, observed_bytes: Option<usize>) -> Self {
56 Self {
57 body: None,
58 capture_status: ProviderHttpBodyCaptureStatus::NotCaptured,
59 capture_reason: Some(reason.into()),
60 observed_bytes: observed_bytes.map(|bytes| i64::try_from(bytes).unwrap_or(i64::MAX)),
61 }
62 }
63
64 #[must_use]
65 pub const fn body(&self) -> Option<&Value> {
66 self.body.as_ref()
67 }
68
69 #[must_use]
70 pub const fn capture_status(&self) -> ProviderHttpBodyCaptureStatus {
71 self.capture_status
72 }
73
74 #[must_use]
75 pub fn capture_reason(&self) -> Option<&str> {
76 self.capture_reason.as_deref()
77 }
78
79 #[must_use]
80 pub const fn observed_bytes(&self) -> Option<i64> {
81 self.observed_bytes
82 }
83}
84
85#[derive(Debug, Clone)]
86pub struct ProviderHttpCallBodyEvidence {
87 pub request: ProviderHttpBodyEvidence,
88 pub response: ProviderHttpBodyEvidence,
89}
90
91impl ProviderHttpCallBodyEvidence {
92 #[must_use]
93 pub fn not_captured(reason: impl Into<String>) -> Self {
94 let reason = reason.into();
95 Self {
96 request: ProviderHttpBodyEvidence::not_captured(reason.clone(), None),
97 response: ProviderHttpBodyEvidence::not_captured(reason, None),
98 }
99 }
100}
101
102#[derive(Debug, Clone)]
103pub struct ProviderHttpCallRecord {
104 pub module_name: String,
105 pub method: String,
106 pub declared_path: String,
107 pub provider_path: String,
108 pub capability: Option<String>,
109 pub display_name: Option<String>,
110 pub story_title: Option<String>,
111 pub provider_status: Option<u16>,
112 pub duration_ms: i64,
113 pub success: bool,
114 pub error_code: Option<String>,
115 pub retryable: bool,
116 pub path_params: Value,
117 pub error_details: Value,
118}
119
120pub async fn insert_provider_http_call(
121 pool: &DbPool,
122 ids: &dyn crate::IdGenerator,
123 request_ctx: &RequestContext,
124 record: ProviderHttpCallRecord,
125) -> AppResult<String> {
126 insert_provider_http_call_with_body_evidence(
127 pool,
128 ids,
129 request_ctx,
130 record,
131 ProviderHttpCallBodyEvidence::not_captured("caller_did_not_supply_evidence"),
132 )
133 .await
134}
135
136pub async fn insert_provider_http_call_with_body_evidence(
137 pool: &DbPool,
138 ids: &dyn crate::IdGenerator,
139 request_ctx: &RequestContext,
140 record: ProviderHttpCallRecord,
141 body_evidence: ProviderHttpCallBodyEvidence,
142) -> AppResult<String> {
143 let id = ids.new_id("rproxy");
144 let path_params = normalize_object(record.path_params.clone());
145 let error_details = normalize_array(record.error_details.clone());
146 let occurred_at = sqlx::query_scalar::<_, DateTime<Utc>>(
147 r#"
148 insert into platform.provider_http_calls (
149 id,
150 module_name,
151 method,
152 declared_path,
153 provider_path,
154 capability,
155 provider_status,
156 duration_ms,
157 success,
158 error_code,
159 retryable,
160 request_id,
161 correlation_id,
162 trace_id,
163 span_id,
164 path_params,
165 error_details,
166 request_body,
167 request_body_capture_status,
168 request_body_capture_reason,
169 request_body_observed_bytes,
170 response_body,
171 response_body_capture_status,
172 response_body_capture_reason,
173 response_body_observed_bytes
174 )
175 values (
176 $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,
177 $16, $17, $18, $19, $20, $21, $22, $23, $24, $25
178 )
179 returning occurred_at
180 "#,
181 )
182 .bind(&id)
183 .bind(&record.module_name)
184 .bind(&record.method)
185 .bind(&record.declared_path)
186 .bind(&record.provider_path)
187 .bind(&record.capability)
188 .bind(record.provider_status.map(i32::from))
189 .bind(record.duration_ms)
190 .bind(record.success)
191 .bind(&record.error_code)
192 .bind(record.retryable)
193 .bind(&request_ctx.request_id.0)
194 .bind(&request_ctx.correlation_id.0)
195 .bind(&request_ctx.trace.trace_id)
196 .bind(&request_ctx.trace.span_id)
197 .bind(&path_params)
198 .bind(&error_details)
199 .bind(&body_evidence.request.body)
200 .bind(body_evidence.request.capture_status.as_str())
201 .bind(&body_evidence.request.capture_reason)
202 .bind(body_evidence.request.observed_bytes)
203 .bind(&body_evidence.response.body)
204 .bind(body_evidence.response.capture_status.as_str())
205 .bind(&body_evidence.response.capture_reason)
206 .bind(body_evidence.response.observed_bytes)
207 .fetch_one(pool)
208 .await
209 .map_err(map_provider_call_error)?;
210
211 insert_provider_call_story_event(
212 pool,
213 &id,
214 request_ctx,
215 &record,
216 &body_evidence,
217 &path_params,
218 occurred_at,
219 )
220 .await?;
221
222 Ok(id)
223}
224
225fn normalize_object(value: Value) -> Value {
226 match value {
227 Value::Object(_) => value,
228 _ => Value::Object(Default::default()),
229 }
230}
231
232fn normalize_array(value: Value) -> Value {
233 match value {
234 Value::Array(_) => value,
235 _ => Value::Array(Vec::new()),
236 }
237}
238
239fn map_provider_call_error(source: sqlx::Error) -> AppError {
240 AppError::new(ErrorCode::Internal, "Remote proxy call operation failed").with_source(source)
241}
242
243async fn insert_provider_call_story_event(
244 pool: &DbPool,
245 id: &str,
246 request_ctx: &RequestContext,
247 record: &ProviderHttpCallRecord,
248 body_evidence: &ProviderHttpCallBodyEvidence,
249 path_params: &Value,
250 occurred_at: DateTime<Utc>,
251) -> AppResult<()> {
252 let story_event_id = provider_call_story_event_id(id);
253 let completed_at = occurred_at + Duration::milliseconds(record.duration_ms.max(0));
254 let status = if record.success {
255 "completed"
256 } else {
257 "failed"
258 };
259
260 sqlx::query(
261 r#"
262 insert into platform.story_events (
263 id,
264 source_type,
265 source_id,
266 node_type,
267 name,
268 status,
269 service,
270 correlation_id,
271 causation_id,
272 started_at,
273 completed_at,
274 duration_ms,
275 error,
276 metadata,
277 trace_id,
278 span_id,
279 updated_at
280 )
281 values ($1, 'provider_call', $2, 'provider_call', $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $9)
282 on conflict (source_type, source_id) do update
283 set
284 name = excluded.name,
285 status = excluded.status,
286 service = excluded.service,
287 correlation_id = excluded.correlation_id,
288 causation_id = excluded.causation_id,
289 started_at = excluded.started_at,
290 completed_at = excluded.completed_at,
291 duration_ms = excluded.duration_ms,
292 error = excluded.error,
293 metadata = excluded.metadata,
294 trace_id = excluded.trace_id,
295 span_id = excluded.span_id,
296 updated_at = excluded.updated_at
297 "#,
298 )
299 .bind(story_event_id)
300 .bind(id)
301 .bind(provider_call_story_event_name(record))
302 .bind(status)
303 .bind(&record.module_name)
304 .bind(&request_ctx.correlation_id.0)
305 .bind(&request_ctx.causation_id)
306 .bind(occurred_at)
307 .bind(completed_at)
308 .bind(record.duration_ms)
309 .bind(provider_call_story_event_error(record))
310 .bind(provider_call_story_event_metadata(
311 id,
312 request_ctx,
313 record,
314 body_evidence,
315 path_params,
316 ))
317 .bind(&request_ctx.trace.trace_id)
318 .bind(&request_ctx.trace.span_id)
319 .execute(pool)
320 .await
321 .map_err(map_provider_call_error)?;
322
323 Ok(())
324}
325
326pub fn provider_call_story_event_id(id: &str) -> String {
327 format!("remoteproxy_{id}")
328}
329
330fn provider_call_story_event_name(record: &ProviderHttpCallRecord) -> String {
331 if let Some(display_name) = record.display_name.as_deref() {
332 return display_name.to_owned();
333 }
334
335 format!(
336 "{} {} {}",
337 record.module_name, record.method, record.declared_path
338 )
339}
340
341fn provider_call_story_event_error(record: &ProviderHttpCallRecord) -> Option<String> {
342 if record.success {
343 return None;
344 }
345
346 Some(match record.error_code.as_deref() {
347 Some(error_code) => format!("remote proxy call failed with {error_code}"),
348 None => "remote proxy call failed".to_owned(),
349 })
350}
351
352fn provider_call_story_event_metadata(
353 id: &str,
354 request_ctx: &RequestContext,
355 record: &ProviderHttpCallRecord,
356 body_evidence: &ProviderHttpCallBodyEvidence,
357 path_params: &Value,
358) -> Value {
359 serde_json::json!({
360 "provider_call_id": id,
361 "module_name": &record.module_name,
362 "method": &record.method,
363 "declared_path": &record.declared_path,
364 "provider_path": &record.provider_path,
365 "capability": &record.capability,
366 "display_name": &record.display_name,
367 "story_title": &record.story_title,
368 "provider_status": record.provider_status,
369 "duration_ms": record.duration_ms,
370 "request_id": request_ctx.request_id.0,
371 "trace_id": request_ctx.trace.trace_id,
372 "span_id": request_ctx.trace.span_id,
373 "success": record.success,
374 "error_code": &record.error_code,
375 "retryable": record.retryable,
376 "path_params": path_params,
377 "error_details": &record.error_details,
378 "request_body_capture_status": body_evidence.request.capture_status.as_str(),
379 "request_body_capture_reason": &body_evidence.request.capture_reason,
380 "request_body_observed_bytes": body_evidence.request.observed_bytes,
381 "response_body_capture_status": body_evidence.response.capture_status.as_str(),
382 "response_body_capture_reason": &body_evidence.response.capture_reason,
383 "response_body_observed_bytes": body_evidence.response.observed_bytes,
384 })
385}