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