1use runledger_core::jobs::WorkflowType;
2use sqlx::types::Uuid;
3
4use crate::{DbPool, DbTx, Result};
5
6use super::super::errors::validate_pagination;
7use super::super::row_decode::parse_workflow_release_mode;
8use super::super::rows::{WorkflowRunRow, WorkflowStepRow};
9use super::super::workflow_types::{
10 WorkflowRunCountFilter, WorkflowRunDbRecord, WorkflowRunListFilter, WorkflowStepDbRecord,
11 WorkflowStepDependencyDbRecord,
12};
13
14#[derive(sqlx::FromRow)]
15struct WorkflowStepDependencyLookupRow {
16 workflow_run_id: Uuid,
17 prerequisite_step_id: Uuid,
18 dependent_step_id: Uuid,
19 release_mode: String,
20 created_at: chrono::DateTime<chrono::Utc>,
21}
22
23fn workflow_step_dependency_db_record_from_lookup_row(
24 row: WorkflowStepDependencyLookupRow,
25) -> Result<WorkflowStepDependencyDbRecord> {
26 Ok(WorkflowStepDependencyDbRecord {
27 workflow_run_id: row.workflow_run_id,
28 prerequisite_step_id: row.prerequisite_step_id,
29 dependent_step_id: row.dependent_step_id,
30 release_mode: parse_workflow_release_mode(row.release_mode)?,
31 created_at: row.created_at,
32 })
33}
34
35pub async fn get_workflow_run_by_id(
36 pool: &DbPool,
37 organization_id: Option<Uuid>,
38 workflow_run_id: Uuid,
39) -> Result<Option<WorkflowRunDbRecord>> {
40 let row = sqlx::query_as!(
41 WorkflowRunRow,
42 "SELECT
43 id,
44 workflow_type,
45 organization_id,
46 status::text AS \"status!\",
47 idempotency_key,
48 result_step_key,
49 metadata,
50 started_at,
51 finished_at,
52 created_at,
53 updated_at
54 FROM workflow_runs
55 WHERE id = $1
56 AND ($2::uuid IS NULL OR organization_id = $2)
57 LIMIT 1",
58 workflow_run_id,
59 organization_id,
60 )
61 .fetch_optional(pool)
62 .await
63 .map_err(|error| crate::Error::from_query_sqlx_with_context("get workflow run by id", error))?;
64
65 row.map(WorkflowRunRow::into_record).transpose()
66}
67
68pub(in crate::jobs::workflows) async fn load_workflow_run_by_id_tx(
69 tx: &mut DbTx<'_>,
70 workflow_run_id: Uuid,
71 context: &'static str,
72) -> Result<WorkflowRunDbRecord> {
73 let run_row = sqlx::query_as!(
74 WorkflowRunRow,
75 "SELECT
76 id,
77 workflow_type,
78 organization_id,
79 status::text AS \"status!\",
80 idempotency_key,
81 result_step_key,
82 metadata,
83 started_at,
84 finished_at,
85 created_at,
86 updated_at
87 FROM workflow_runs
88 WHERE id = $1",
89 workflow_run_id,
90 )
91 .fetch_one(&mut **tx)
92 .await
93 .map_err(|error| crate::Error::from_query_sqlx_with_context(context, error))?;
94
95 run_row.into_record()
96}
97
98pub async fn list_workflow_steps(
99 pool: &DbPool,
100 organization_id: Option<Uuid>,
101 workflow_run_id: Uuid,
102) -> Result<Vec<WorkflowStepDbRecord>> {
103 let rows = sqlx::query_as::<_, WorkflowStepRow>(
104 "SELECT
105 ws.id,
106 ws.workflow_run_id,
107 ws.step_key,
108 ws.execution_kind::text AS execution_kind,
109 ws.job_type,
110 ws.organization_id,
111 ws.payload,
112 ws.priority,
113 ws.max_attempts,
114 ws.timeout_seconds,
115 ws.stage,
116 ws.status::text AS status,
117 ws.job_id,
118 ws.released_at,
119 ws.started_at,
120 ws.finished_at,
121 ws.dependency_count_total,
122 ws.dependency_count_pending,
123 ws.dependency_count_unsatisfied,
124 ws.status_reason,
125 ws.last_error_code,
126 ws.last_error_message,
127 ws.output,
128 ws.created_at,
129 ws.updated_at
130 FROM workflow_steps ws
131 JOIN workflow_runs wr ON wr.id = ws.workflow_run_id
132 WHERE ws.workflow_run_id = $1
133 AND ($2::uuid IS NULL OR wr.organization_id = $2)
134 ORDER BY ws.created_at ASC, ws.id ASC",
135 )
136 .bind(workflow_run_id)
137 .bind(organization_id)
138 .fetch_all(pool)
139 .await
140 .map_err(|error| crate::Error::from_query_sqlx_with_context("list workflow steps", error))?;
141
142 rows.into_iter().map(WorkflowStepRow::into_record).collect()
143}
144
145pub async fn list_workflow_steps_page(
146 pool: &DbPool,
147 organization_id: Option<Uuid>,
148 workflow_run_id: Uuid,
149 limit: i64,
150 offset: i64,
151) -> Result<Vec<WorkflowStepDbRecord>> {
152 validate_pagination(limit, offset)?;
153
154 let rows = sqlx::query_as::<_, WorkflowStepRow>(
155 "SELECT
156 ws.id,
157 ws.workflow_run_id,
158 ws.step_key,
159 ws.execution_kind::text AS execution_kind,
160 ws.job_type,
161 ws.organization_id,
162 ws.payload,
163 ws.priority,
164 ws.max_attempts,
165 ws.timeout_seconds,
166 ws.stage,
167 ws.status::text AS status,
168 ws.job_id,
169 ws.released_at,
170 ws.started_at,
171 ws.finished_at,
172 ws.dependency_count_total,
173 ws.dependency_count_pending,
174 ws.dependency_count_unsatisfied,
175 ws.status_reason,
176 ws.last_error_code,
177 ws.last_error_message,
178 ws.output,
179 ws.created_at,
180 ws.updated_at
181 FROM workflow_steps ws
182 JOIN workflow_runs wr ON wr.id = ws.workflow_run_id
183 WHERE ws.workflow_run_id = $1
184 AND ($2::uuid IS NULL OR wr.organization_id = $2)
185 ORDER BY ws.created_at ASC, ws.id ASC
186 LIMIT $3 OFFSET $4",
187 )
188 .bind(workflow_run_id)
189 .bind(organization_id)
190 .bind(limit)
191 .bind(offset)
192 .fetch_all(pool)
193 .await
194 .map_err(|error| {
195 crate::Error::from_query_sqlx_with_context("list workflow steps page", error)
196 })?;
197
198 rows.into_iter().map(WorkflowStepRow::into_record).collect()
199}
200
201pub async fn count_workflow_steps(
202 pool: &DbPool,
203 organization_id: Option<Uuid>,
204 workflow_run_id: Uuid,
205) -> Result<i64> {
206 sqlx::query_scalar::<_, i64>(
207 "SELECT COUNT(*)::bigint
208 FROM workflow_steps ws
209 JOIN workflow_runs wr ON wr.id = ws.workflow_run_id
210 WHERE ws.workflow_run_id = $1
211 AND ($2::uuid IS NULL OR wr.organization_id = $2)",
212 )
213 .bind(workflow_run_id)
214 .bind(organization_id)
215 .fetch_one(pool)
216 .await
217 .map_err(|error| crate::Error::from_query_sqlx_with_context("count workflow steps", error))
218}
219
220pub async fn list_workflow_runs(
221 pool: &DbPool,
222 filter: &WorkflowRunListFilter<'_>,
223) -> Result<Vec<WorkflowRunDbRecord>> {
224 validate_pagination(filter.limit, filter.offset)?;
225
226 let status_text = filter.status.map(|status| status.as_db_value());
227
228 let rows = sqlx::query_as::<_, WorkflowRunRow>(
229 "SELECT
230 id,
231 workflow_type,
232 organization_id,
233 status::text AS status,
234 idempotency_key,
235 result_step_key,
236 metadata,
237 started_at,
238 finished_at,
239 created_at,
240 updated_at
241 FROM workflow_runs
242 WHERE ($1::uuid IS NULL OR organization_id = $1)
243 AND ($2::text IS NULL OR status = $2::text::workflow_run_status)
244 AND ($3::text IS NULL OR workflow_type ILIKE '%' || $3 || '%')
245 ORDER BY created_at DESC, id DESC
246 LIMIT $4 OFFSET $5",
247 )
248 .bind(filter.organization_id)
249 .bind(status_text)
250 .bind(filter.workflow_type)
251 .bind(filter.limit)
252 .bind(filter.offset)
253 .fetch_all(pool)
254 .await
255 .map_err(|error| crate::Error::from_query_sqlx_with_context("list workflow runs", error))?;
256
257 rows.into_iter().map(WorkflowRunRow::into_record).collect()
258}
259
260pub async fn count_workflow_runs(
261 pool: &DbPool,
262 filter: &WorkflowRunCountFilter<'_>,
263) -> Result<i64> {
264 let status_text = filter.status.map(|status| status.as_db_value());
265 sqlx::query_scalar::<_, i64>(
266 "SELECT COUNT(*)::bigint
267 FROM workflow_runs
268 WHERE ($1::uuid IS NULL OR organization_id = $1)
269 AND ($2::text IS NULL OR status = $2::text::workflow_run_status)
270 AND ($3::text IS NULL OR workflow_type ILIKE '%' || $3 || '%')",
271 )
272 .bind(filter.organization_id)
273 .bind(status_text)
274 .bind(filter.workflow_type)
275 .fetch_one(pool)
276 .await
277 .map_err(|error| crate::Error::from_query_sqlx_with_context("count workflow runs", error))
278}
279
280pub async fn get_latest_workflow_run_by_type(
281 pool: &DbPool,
282 organization_id: Option<Uuid>,
283 workflow_type: WorkflowType<'_>,
284) -> Result<Option<WorkflowRunDbRecord>> {
285 let row = sqlx::query_as::<_, WorkflowRunRow>(
286 "SELECT
287 id,
288 workflow_type,
289 organization_id,
290 status::text AS status,
291 idempotency_key,
292 result_step_key,
293 metadata,
294 started_at,
295 finished_at,
296 created_at,
297 updated_at
298 FROM workflow_runs
299 WHERE ($1::uuid IS NULL OR organization_id = $1)
300 AND workflow_type = $2
301 ORDER BY created_at DESC, id DESC
302 LIMIT 1",
303 )
304 .bind(organization_id)
305 .bind(workflow_type.as_str())
306 .fetch_optional(pool)
307 .await
308 .map_err(|error| {
309 crate::Error::from_query_sqlx_with_context("get latest workflow run by type", error)
310 })?;
311
312 let Some(row) = row else {
313 return Ok(None);
314 };
315
316 Ok(Some(row.into_record()?))
317}
318
319pub async fn list_workflow_step_dependencies(
320 pool: &DbPool,
321 organization_id: Option<Uuid>,
322 workflow_run_id: Uuid,
323) -> Result<Vec<WorkflowStepDependencyDbRecord>> {
324 let rows = sqlx::query_as::<_, WorkflowStepDependencyLookupRow>(
325 "SELECT
326 wsd.workflow_run_id,
327 wsd.prerequisite_step_id,
328 wsd.dependent_step_id,
329 wsd.release_mode::text AS release_mode,
330 wsd.created_at
331 FROM workflow_step_dependencies wsd
332 JOIN workflow_runs wr ON wr.id = wsd.workflow_run_id
333 WHERE wsd.workflow_run_id = $1
334 AND ($2::uuid IS NULL OR wr.organization_id = $2)
335 ORDER BY
336 wsd.prerequisite_step_id ASC,
337 wsd.dependent_step_id ASC",
338 )
339 .bind(workflow_run_id)
340 .bind(organization_id)
341 .fetch_all(pool)
342 .await
343 .map_err(|error| {
344 crate::Error::from_query_sqlx_with_context("list workflow step dependencies", error)
345 })?;
346
347 rows.into_iter()
348 .map(workflow_step_dependency_db_record_from_lookup_row)
349 .collect()
350}
351
352pub async fn list_workflow_step_dependencies_page(
353 pool: &DbPool,
354 organization_id: Option<Uuid>,
355 workflow_run_id: Uuid,
356 limit: i64,
357 offset: i64,
358) -> Result<Vec<WorkflowStepDependencyDbRecord>> {
359 validate_pagination(limit, offset)?;
360
361 let rows = sqlx::query_as::<_, WorkflowStepDependencyLookupRow>(
362 "SELECT
363 wsd.workflow_run_id,
364 wsd.prerequisite_step_id,
365 wsd.dependent_step_id,
366 wsd.release_mode::text AS release_mode,
367 wsd.created_at
368 FROM workflow_step_dependencies wsd
369 JOIN workflow_runs wr ON wr.id = wsd.workflow_run_id
370 WHERE wsd.workflow_run_id = $1
371 AND ($2::uuid IS NULL OR wr.organization_id = $2)
372 ORDER BY
373 wsd.prerequisite_step_id ASC,
374 wsd.dependent_step_id ASC
375 LIMIT $3 OFFSET $4",
376 )
377 .bind(workflow_run_id)
378 .bind(organization_id)
379 .bind(limit)
380 .bind(offset)
381 .fetch_all(pool)
382 .await
383 .map_err(|error| {
384 crate::Error::from_query_sqlx_with_context("list workflow step dependencies page", error)
385 })?;
386
387 rows.into_iter()
388 .map(workflow_step_dependency_db_record_from_lookup_row)
389 .collect()
390}
391
392pub async fn count_workflow_step_dependencies(
393 pool: &DbPool,
394 organization_id: Option<Uuid>,
395 workflow_run_id: Uuid,
396) -> Result<i64> {
397 sqlx::query_scalar::<_, i64>(
398 "SELECT COUNT(*)::bigint
399 FROM workflow_step_dependencies wsd
400 JOIN workflow_runs wr ON wr.id = wsd.workflow_run_id
401 WHERE wsd.workflow_run_id = $1
402 AND ($2::uuid IS NULL OR wr.organization_id = $2)",
403 )
404 .bind(workflow_run_id)
405 .bind(organization_id)
406 .fetch_one(pool)
407 .await
408 .map_err(|error| {
409 crate::Error::from_query_sqlx_with_context("count workflow step dependencies", error)
410 })
411}
412
413pub async fn get_workflow_run_id_for_job(pool: &DbPool, job_id: Uuid) -> Result<Option<Uuid>> {
414 sqlx::query_scalar!(
415 "SELECT ws.workflow_run_id FROM workflow_steps ws WHERE ws.job_id = $1",
416 job_id,
417 )
418 .fetch_optional(pool)
419 .await
420 .map_err(|error| {
421 crate::Error::from_query_sqlx_with_context("get workflow run id for job", error)
422 })
423}
424
425pub async fn get_workflow_run_by_type_and_idempotency_key(
426 pool: &DbPool,
427 organization_id: Option<Uuid>,
428 workflow_type: WorkflowType<'_>,
429 idempotency_key: &str,
430) -> Result<Option<WorkflowRunDbRecord>> {
431 let mut tx = pool
432 .begin()
433 .await
434 .map_err(|error| crate::Error::ConnectionError(error.to_string()))?;
435 let run = get_workflow_run_by_type_and_idempotency_key_tx(
436 &mut tx,
437 organization_id,
438 workflow_type,
439 idempotency_key,
440 )
441 .await?;
442 tx.commit()
443 .await
444 .map_err(|error| crate::Error::ConnectionError(error.to_string()))?;
445 Ok(run)
446}
447
448pub async fn get_workflow_run_by_type_and_idempotency_key_tx(
449 tx: &mut DbTx<'_>,
450 organization_id: Option<Uuid>,
451 workflow_type: WorkflowType<'_>,
452 idempotency_key: &str,
453) -> Result<Option<WorkflowRunDbRecord>> {
454 let row = if let Some(organization_id) = organization_id {
455 sqlx::query_as!(
456 WorkflowRunRow,
457 "SELECT
458 id,
459 workflow_type,
460 organization_id,
461 status::text AS \"status!\",
462 idempotency_key,
463 result_step_key,
464 metadata,
465 started_at,
466 finished_at,
467 created_at,
468 updated_at
469 FROM workflow_runs
470 WHERE workflow_type = $1
471 AND idempotency_key = $2
472 AND organization_id = $3
473 LIMIT 1",
474 workflow_type as _,
475 idempotency_key,
476 organization_id,
477 )
478 .fetch_optional(&mut **tx)
479 .await
480 } else {
481 sqlx::query_as!(
482 WorkflowRunRow,
483 "SELECT
484 id,
485 workflow_type,
486 organization_id,
487 status::text AS \"status!\",
488 idempotency_key,
489 result_step_key,
490 metadata,
491 started_at,
492 finished_at,
493 created_at,
494 updated_at
495 FROM workflow_runs
496 WHERE workflow_type = $1
497 AND idempotency_key = $2
498 AND organization_id IS NULL
499 LIMIT 1",
500 workflow_type as _,
501 idempotency_key,
502 )
503 .fetch_optional(&mut **tx)
504 .await
505 }
506 .map_err(|error| {
507 crate::Error::from_query_sqlx_with_context(
508 "get workflow run by type and idempotency key",
509 error,
510 )
511 })?;
512
513 row.map(WorkflowRunRow::into_record).transpose()
514}