1use crate::TraceContext;
2use crate::db::DbPool;
3use crate::error::{AppError, AppResult, ErrorCode};
4use crate::telemetry_attrs::RuntimeSpanAttributes;
5use async_trait::async_trait;
6use chrono::{DateTime, Utc};
7use serde::{Deserialize, Serialize};
8use serde_json::{Value, json};
9use std::fmt::Debug;
10use uuid::Uuid;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
13#[serde(rename_all = "snake_case")]
14pub(crate) enum ExecutionLogSeverity {
15 Trace,
16 Debug,
17 Info,
18 Warn,
19 Error,
20}
21
22impl ExecutionLogSeverity {
23 fn as_str(self) -> &'static str {
24 match self {
25 Self::Trace => "trace",
26 Self::Debug => "debug",
27 Self::Info => "info",
28 Self::Warn => "warn",
29 Self::Error => "error",
30 }
31 }
32}
33
34#[derive(Debug, Clone)]
35pub(crate) struct ExecutionLogRecord {
36 correlation_id: String,
37 execution_id: String,
38 execution_type: String,
39 execution_name: String,
40 severity: ExecutionLogSeverity,
41 body: String,
42 attributes: Value,
43 trace: TraceContext,
44 service_name: String,
45}
46
47impl ExecutionLogRecord {
48 pub(crate) fn from_runtime_attrs(
49 attrs: RuntimeSpanAttributes,
50 severity: ExecutionLogSeverity,
51 body: impl Into<String>,
52 ) -> Self {
53 let execution_id = attrs
54 .function_run_id
55 .clone()
56 .or_else(|| attrs.outbox_event_id.clone())
57 .unwrap_or_else(|| attrs.story_id.clone());
58
59 Self {
60 correlation_id: attrs.correlation_id,
61 execution_id,
62 execution_type: attrs.execution_kind,
63 execution_name: attrs.execution_name,
64 severity,
65 body: body.into(),
66 attributes: Value::Object(Default::default()),
67 trace: TraceContext::default(),
68 service_name: "lenso".to_owned(),
69 }
70 }
71
72 pub(crate) fn with_attributes(mut self, attributes: Value) -> Self {
73 self.attributes = attributes;
74 self
75 }
76
77 pub(crate) fn with_trace(mut self, trace: TraceContext) -> Self {
78 self.trace = trace;
79 self
80 }
81}
82
83pub(crate) async fn insert_execution_log_projection(
84 pool: &DbPool,
85 record: ExecutionLogRecord,
86) -> AppResult<String> {
87 let id = next_execution_log_id();
88 sqlx::query(
89 r#"
90 insert into platform.execution_logs (
91 id,
92 correlation_id,
93 story_id,
94 execution_id,
95 execution_type,
96 execution_name,
97 occurred_at,
98 severity,
99 body,
100 attributes,
101 trace_id,
102 span_id,
103 service_name,
104 redacted_fields
105 )
106 values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
107 "#,
108 )
109 .bind(&id)
110 .bind(&record.correlation_id)
111 .bind(&record.correlation_id)
112 .bind(&record.execution_id)
113 .bind(&record.execution_type)
114 .bind(&record.execution_name)
115 .bind(Utc::now())
116 .bind(record.severity.as_str())
117 .bind(&record.body)
118 .bind(normalize_attributes(record.attributes))
119 .bind(&record.trace.trace_id)
120 .bind(&record.trace.span_id)
121 .bind(&record.service_name)
122 .bind(Vec::<String>::new())
123 .execute(pool)
124 .await
125 .map_err(map_execution_log_error)?;
126
127 Ok(id)
128}
129
130fn normalize_attributes(attributes: Value) -> Value {
131 match attributes {
132 Value::Object(_) => attributes,
133 other => json!({ "value": other }),
134 }
135}
136
137fn next_execution_log_id() -> String {
138 format!("elog_{}", Uuid::now_v7())
139}
140
141fn map_execution_log_error(source: sqlx::Error) -> AppError {
142 AppError::new(ErrorCode::Internal, "Execution log operation failed").with_source(source)
143}
144
145#[derive(Debug, Clone)]
146pub struct ExecutionLogRow {
147 pub id: String,
148 pub correlation_id: String,
149 pub story_id: String,
150 pub execution_id: String,
151 pub execution_type: String,
152 pub execution_name: String,
153 pub occurred_at: DateTime<Utc>,
154 pub severity: String,
155 pub body: String,
156 pub attributes: Value,
157 pub trace_id: Option<String>,
158 pub span_id: Option<String>,
159 pub service_name: String,
160 pub redacted_fields: Vec<String>,
161}
162
163#[derive(Debug, Clone, Default, PartialEq, Eq)]
164pub struct ExecutionLogQuery {
165 pub execution_id: String,
166 pub occurred_before: Option<DateTime<Utc>>,
167 pub limit: i64,
168}
169
170#[async_trait]
171pub trait ExecutionLogProvider: Debug + Send + Sync {
172 async fn query_execution_logs(
173 &self,
174 query: ExecutionLogQuery,
175 ) -> AppResult<Vec<ExecutionLogRow>>;
176}
177
178#[derive(Debug, Clone)]
179pub struct PostgresExecutionLogProvider {
180 pool: DbPool,
181}
182
183impl PostgresExecutionLogProvider {
184 pub fn new(pool: DbPool) -> Self {
185 Self { pool }
186 }
187}
188
189#[async_trait]
190impl ExecutionLogProvider for PostgresExecutionLogProvider {
191 async fn query_execution_logs(
192 &self,
193 query: ExecutionLogQuery,
194 ) -> AppResult<Vec<ExecutionLogRow>> {
195 let mut rows = sqlx::query_as::<_, ExecutionLogTuple>(
196 r#"
197 select *
198 from (
199 select
200 concat('elog_outbox_enqueued_', id) as id,
201 correlation_id,
202 correlation_id as story_id,
203 id as execution_id,
204 'outbox_event'::text as execution_type,
205 event_name as execution_name,
206 created_at as occurred_at,
207 'info'::text as severity,
208 'Outbox event enqueued'::text as body,
209 jsonb_build_object(
210 'event_name', event_name,
211 'event_version', event_version,
212 'aggregate_type', aggregate_type,
213 'aggregate_id', aggregate_id,
214 'source_module', source_module
215 ) as attributes,
216 headers #>> '{trace,trace_id}' as trace_id,
217 headers #>> '{trace,span_id}' as span_id,
218 source_module as service_name,
219 array[]::text[] as redacted_fields
220 from platform.outbox
221 where id = $1
222
223 union all
224
225 select
226 id,
227 correlation_id,
228 story_id,
229 execution_id,
230 execution_type,
231 execution_name,
232 occurred_at,
233 severity,
234 body,
235 attributes,
236 trace_id,
237 span_id,
238 service_name,
239 redacted_fields
240 from platform.execution_logs
241 where execution_id = $1
242 ) execution_log_rows
243 where ($2::timestamptz is null or occurred_at < $2)
244 order by occurred_at desc, id desc
245 limit $3
246 "#,
247 )
248 .bind(query.execution_id)
249 .bind(query.occurred_before)
250 .bind(query.limit)
251 .fetch_all(&self.pool)
252 .await
253 .map_err(map_execution_log_error)?
254 .into_iter()
255 .map(Into::into)
256 .collect::<Vec<_>>();
257
258 rows.reverse();
259 Ok(rows)
260 }
261}
262
263type ExecutionLogTuple = (
264 String,
265 String,
266 String,
267 String,
268 String,
269 String,
270 DateTime<Utc>,
271 String,
272 String,
273 Value,
274 Option<String>,
275 Option<String>,
276 String,
277 Vec<String>,
278);
279
280impl From<ExecutionLogTuple> for ExecutionLogRow {
281 fn from(row: ExecutionLogTuple) -> Self {
282 let (
283 id,
284 correlation_id,
285 story_id,
286 execution_id,
287 execution_type,
288 execution_name,
289 occurred_at,
290 severity,
291 body,
292 attributes,
293 trace_id,
294 span_id,
295 service_name,
296 redacted_fields,
297 ) = row;
298
299 Self {
300 id,
301 correlation_id,
302 story_id,
303 execution_id,
304 execution_type,
305 execution_name,
306 occurred_at,
307 severity,
308 body,
309 attributes,
310 trace_id,
311 span_id,
312 service_name,
313 redacted_fields,
314 }
315 }
316}