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)]
8pub struct ProviderHttpCallRecord {
9 pub module_name: String,
10 pub method: String,
11 pub declared_path: String,
12 pub provider_path: String,
13 pub capability: Option<String>,
14 pub display_name: Option<String>,
15 pub story_title: Option<String>,
16 pub provider_status: Option<u16>,
17 pub duration_ms: i64,
18 pub success: bool,
19 pub error_code: Option<String>,
20 pub retryable: bool,
21 pub path_params: Value,
22 pub error_details: Value,
23}
24
25pub async fn insert_provider_http_call(
26 pool: &DbPool,
27 ids: &dyn crate::IdGenerator,
28 request_ctx: &RequestContext,
29 record: ProviderHttpCallRecord,
30) -> AppResult<String> {
31 let id = ids.new_id("rproxy");
32 let path_params = normalize_object(record.path_params.clone());
33 let error_details = normalize_array(record.error_details.clone());
34 let occurred_at = sqlx::query_scalar::<_, DateTime<Utc>>(
35 r#"
36 insert into platform.provider_http_calls (
37 id,
38 module_name,
39 method,
40 declared_path,
41 provider_path,
42 capability,
43 provider_status,
44 duration_ms,
45 success,
46 error_code,
47 retryable,
48 request_id,
49 correlation_id,
50 trace_id,
51 span_id,
52 path_params,
53 error_details
54 )
55 values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
56 returning occurred_at
57 "#,
58 )
59 .bind(&id)
60 .bind(&record.module_name)
61 .bind(&record.method)
62 .bind(&record.declared_path)
63 .bind(&record.provider_path)
64 .bind(&record.capability)
65 .bind(record.provider_status.map(i32::from))
66 .bind(record.duration_ms)
67 .bind(record.success)
68 .bind(&record.error_code)
69 .bind(record.retryable)
70 .bind(&request_ctx.request_id.0)
71 .bind(&request_ctx.correlation_id.0)
72 .bind(&request_ctx.trace.trace_id)
73 .bind(&request_ctx.trace.span_id)
74 .bind(&path_params)
75 .bind(&error_details)
76 .fetch_one(pool)
77 .await
78 .map_err(map_provider_call_error)?;
79
80 insert_provider_call_story_event(pool, &id, request_ctx, &record, &path_params, occurred_at)
81 .await?;
82
83 Ok(id)
84}
85
86fn normalize_object(value: Value) -> Value {
87 match value {
88 Value::Object(_) => value,
89 _ => Value::Object(Default::default()),
90 }
91}
92
93fn normalize_array(value: Value) -> Value {
94 match value {
95 Value::Array(_) => value,
96 _ => Value::Array(Vec::new()),
97 }
98}
99
100fn map_provider_call_error(source: sqlx::Error) -> AppError {
101 AppError::new(ErrorCode::Internal, "Remote proxy call operation failed").with_source(source)
102}
103
104async fn insert_provider_call_story_event(
105 pool: &DbPool,
106 id: &str,
107 request_ctx: &RequestContext,
108 record: &ProviderHttpCallRecord,
109 path_params: &Value,
110 occurred_at: DateTime<Utc>,
111) -> AppResult<()> {
112 let story_event_id = provider_call_story_event_id(id);
113 let completed_at = occurred_at + Duration::milliseconds(record.duration_ms.max(0));
114 let status = if record.success {
115 "completed"
116 } else {
117 "failed"
118 };
119
120 sqlx::query(
121 r#"
122 insert into platform.story_events (
123 id,
124 source_type,
125 source_id,
126 node_type,
127 name,
128 status,
129 service,
130 correlation_id,
131 causation_id,
132 started_at,
133 completed_at,
134 duration_ms,
135 error,
136 metadata,
137 trace_id,
138 span_id,
139 updated_at
140 )
141 values ($1, 'provider_call', $2, 'provider_call', $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $9)
142 on conflict (source_type, source_id) do update
143 set
144 name = excluded.name,
145 status = excluded.status,
146 service = excluded.service,
147 correlation_id = excluded.correlation_id,
148 causation_id = excluded.causation_id,
149 started_at = excluded.started_at,
150 completed_at = excluded.completed_at,
151 duration_ms = excluded.duration_ms,
152 error = excluded.error,
153 metadata = excluded.metadata,
154 trace_id = excluded.trace_id,
155 span_id = excluded.span_id,
156 updated_at = excluded.updated_at
157 "#,
158 )
159 .bind(story_event_id)
160 .bind(id)
161 .bind(provider_call_story_event_name(record))
162 .bind(status)
163 .bind(&record.module_name)
164 .bind(&request_ctx.correlation_id.0)
165 .bind(&request_ctx.causation_id)
166 .bind(occurred_at)
167 .bind(completed_at)
168 .bind(record.duration_ms)
169 .bind(provider_call_story_event_error(record))
170 .bind(provider_call_story_event_metadata(
171 id,
172 request_ctx,
173 record,
174 path_params,
175 ))
176 .bind(&request_ctx.trace.trace_id)
177 .bind(&request_ctx.trace.span_id)
178 .execute(pool)
179 .await
180 .map_err(map_provider_call_error)?;
181
182 Ok(())
183}
184
185pub fn provider_call_story_event_id(id: &str) -> String {
186 format!("remoteproxy_{id}")
187}
188
189fn provider_call_story_event_name(record: &ProviderHttpCallRecord) -> String {
190 if let Some(display_name) = record.display_name.as_deref() {
191 return display_name.to_owned();
192 }
193
194 format!(
195 "{} {} {}",
196 record.module_name, record.method, record.declared_path
197 )
198}
199
200fn provider_call_story_event_error(record: &ProviderHttpCallRecord) -> Option<String> {
201 if record.success {
202 return None;
203 }
204
205 Some(match record.error_code.as_deref() {
206 Some(error_code) => format!("remote proxy call failed with {error_code}"),
207 None => "remote proxy call failed".to_owned(),
208 })
209}
210
211fn provider_call_story_event_metadata(
212 id: &str,
213 request_ctx: &RequestContext,
214 record: &ProviderHttpCallRecord,
215 path_params: &Value,
216) -> Value {
217 serde_json::json!({
218 "provider_call_id": id,
219 "module_name": &record.module_name,
220 "method": &record.method,
221 "declared_path": &record.declared_path,
222 "provider_path": &record.provider_path,
223 "capability": &record.capability,
224 "display_name": &record.display_name,
225 "story_title": &record.story_title,
226 "provider_status": record.provider_status,
227 "duration_ms": record.duration_ms,
228 "request_id": request_ctx.request_id.0,
229 "trace_id": request_ctx.trace.trace_id,
230 "span_id": request_ctx.trace.span_id,
231 "success": record.success,
232 "error_code": &record.error_code,
233 "retryable": record.retryable,
234 "path_params": path_params,
235 "error_details": &record.error_details,
236 })
237}