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 RemoteHttpProxyCallRecord {
9 pub module_name: String,
10 pub method: String,
11 pub declared_path: String,
12 pub remote_path: String,
13 pub capability: Option<String>,
14 pub display_name: Option<String>,
15 pub story_title: Option<String>,
16 pub remote_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_remote_http_proxy_call(
26 pool: &DbPool,
27 ids: &dyn crate::IdGenerator,
28 request_ctx: &RequestContext,
29 record: RemoteHttpProxyCallRecord,
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.remote_http_proxy_calls (
37 id,
38 module_name,
39 method,
40 declared_path,
41 remote_path,
42 capability,
43 remote_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.remote_path)
64 .bind(&record.capability)
65 .bind(record.remote_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_remote_proxy_call_error)?;
79
80 insert_remote_proxy_call_story_event(
81 pool,
82 &id,
83 request_ctx,
84 &record,
85 &path_params,
86 occurred_at,
87 )
88 .await?;
89
90 Ok(id)
91}
92
93fn normalize_object(value: Value) -> Value {
94 match value {
95 Value::Object(_) => value,
96 _ => Value::Object(Default::default()),
97 }
98}
99
100fn normalize_array(value: Value) -> Value {
101 match value {
102 Value::Array(_) => value,
103 _ => Value::Array(Vec::new()),
104 }
105}
106
107fn map_remote_proxy_call_error(source: sqlx::Error) -> AppError {
108 AppError::new(ErrorCode::Internal, "Remote proxy call operation failed").with_source(source)
109}
110
111async fn insert_remote_proxy_call_story_event(
112 pool: &DbPool,
113 id: &str,
114 request_ctx: &RequestContext,
115 record: &RemoteHttpProxyCallRecord,
116 path_params: &Value,
117 occurred_at: DateTime<Utc>,
118) -> AppResult<()> {
119 let story_event_id = remote_proxy_call_story_event_id(id);
120 let completed_at = occurred_at + Duration::milliseconds(record.duration_ms.max(0));
121 let status = if record.success {
122 "completed"
123 } else {
124 "failed"
125 };
126
127 sqlx::query(
128 r#"
129 insert into platform.story_events (
130 id,
131 source_type,
132 source_id,
133 node_type,
134 name,
135 status,
136 service,
137 correlation_id,
138 causation_id,
139 started_at,
140 completed_at,
141 duration_ms,
142 error,
143 metadata,
144 trace_id,
145 span_id,
146 updated_at
147 )
148 values ($1, 'remote_proxy_call', $2, 'remote_proxy_call', $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $9)
149 on conflict (source_type, source_id) do update
150 set
151 name = excluded.name,
152 status = excluded.status,
153 service = excluded.service,
154 correlation_id = excluded.correlation_id,
155 causation_id = excluded.causation_id,
156 started_at = excluded.started_at,
157 completed_at = excluded.completed_at,
158 duration_ms = excluded.duration_ms,
159 error = excluded.error,
160 metadata = excluded.metadata,
161 trace_id = excluded.trace_id,
162 span_id = excluded.span_id,
163 updated_at = excluded.updated_at
164 "#,
165 )
166 .bind(story_event_id)
167 .bind(id)
168 .bind(remote_proxy_call_story_event_name(record))
169 .bind(status)
170 .bind(&record.module_name)
171 .bind(&request_ctx.correlation_id.0)
172 .bind(&request_ctx.causation_id)
173 .bind(occurred_at)
174 .bind(completed_at)
175 .bind(record.duration_ms)
176 .bind(remote_proxy_call_story_event_error(record))
177 .bind(remote_proxy_call_story_event_metadata(
178 id,
179 request_ctx,
180 record,
181 path_params,
182 ))
183 .bind(&request_ctx.trace.trace_id)
184 .bind(&request_ctx.trace.span_id)
185 .execute(pool)
186 .await
187 .map_err(map_remote_proxy_call_error)?;
188
189 Ok(())
190}
191
192pub fn remote_proxy_call_story_event_id(id: &str) -> String {
193 format!("remoteproxy_{id}")
194}
195
196fn remote_proxy_call_story_event_name(record: &RemoteHttpProxyCallRecord) -> String {
197 if let Some(display_name) = record.display_name.as_deref() {
198 return display_name.to_owned();
199 }
200
201 format!(
202 "{} {} {}",
203 record.module_name, record.method, record.declared_path
204 )
205}
206
207fn remote_proxy_call_story_event_error(record: &RemoteHttpProxyCallRecord) -> Option<String> {
208 if record.success {
209 return None;
210 }
211
212 Some(match record.error_code.as_deref() {
213 Some(error_code) => format!("remote proxy call failed with {error_code}"),
214 None => "remote proxy call failed".to_owned(),
215 })
216}
217
218fn remote_proxy_call_story_event_metadata(
219 id: &str,
220 request_ctx: &RequestContext,
221 record: &RemoteHttpProxyCallRecord,
222 path_params: &Value,
223) -> Value {
224 serde_json::json!({
225 "remote_proxy_call_id": id,
226 "module_name": &record.module_name,
227 "method": &record.method,
228 "declared_path": &record.declared_path,
229 "remote_path": &record.remote_path,
230 "capability": &record.capability,
231 "display_name": &record.display_name,
232 "story_title": &record.story_title,
233 "remote_status": record.remote_status,
234 "duration_ms": record.duration_ms,
235 "request_id": request_ctx.request_id.0,
236 "trace_id": request_ctx.trace.trace_id,
237 "span_id": request_ctx.trace.span_id,
238 "success": record.success,
239 "error_code": &record.error_code,
240 "retryable": record.retryable,
241 "path_params": path_params,
242 "error_details": &record.error_details,
243 })
244}