update based on state rework

swanandx · swanandx · commit 030ef3aca7df · 2025-09-23T17:31:44.000+05:30
Signed-off-by: Swanand Mulay &lt;73115739+swanandx@users.noreply.github.com&gt;
diff --git a/crates/pipeline-manager/migrations/V28__add_pipeline_lifecycle_history.sql b/crates/pipeline-manager/migrations/V28__add_pipeline_lifecycle_history.sql
@@ -2,7 +2,9 @@ CREATE TABLE IF NOT EXISTS pipeline_lifecycle_events (
     event_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
     pipeline_id uuid NOT NULL,
     tenant_id uuid NOT NULL,
-    deployment_status varchar NOT NULL,
+    deployment_resources_status varchar NOT NULL,
+    deployment_runtime_status varchar,
+    deployment_runtime_desired_status varchar,
     info text,
     recorded_at TIMESTAMP NOT NULL DEFAULT now(),
     FOREIGN KEY (pipeline_id) REFERENCES pipeline(id) ON DELETE CASCADE,
diff --git a/crates/pipeline-manager/src/api/lifecycle_events.rs b/crates/pipeline-manager/src/api/lifecycle_events.rs
@@ -3,15 +3,13 @@ use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 use uuid::Uuid;
 
-use crate::db::types::pipeline::PipelineStatus;
-
 // Pipeline Lifecycle Events
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, ToSchema)]
 pub struct PipelineLifecycleEvent {
     pub event_id: Uuid,
-    // pub tenant_id: TenantId,
-    // pub pipeline_id: PipelineId,
-    pub deployment_status: PipelineStatus,
+    pub deployment_resources_status: String,
+    pub deployment_runtime_status: Option<String>,
+    pub deployment_runtime_desired_status: Option<String>,
     pub info: Option<String>,
     pub recorded_at: NaiveDateTime,
 }
diff --git a/crates/pipeline-manager/src/db/operations/pipeline.rs b/crates/pipeline-manager/src/db/operations/pipeline.rs
@@ -1326,6 +1326,10 @@ pub(crate) async fn set_deployment_resources_status(
         Some(ref v) => Some(serialize_error_response(v)?),
     };
 
+    let deployment_resources_status = new_deployment_resources_status.to_string(); // $7: deployment_resources_status,
+    let deployment_runtime_status =  final_deployment_runtime_status.map(runtime_status_to_string); // $8: deployment_runtime_status,
+    let deployment_runtime_desired_status =  final_deployment_runtime_desired_status.map(runtime_desired_status_to_string); // $9: deployment_runtime_desired_status,
+
     // Execute query
     let stmt = txn
         .prepare_cached(
@@ -1349,18 +1353,15 @@ pub(crate) async fn set_deployment_resources_status(
         .execute(
             &stmt,
             &[
-                &match final_deployment_error {
-                    None => None,
-                    Some(v) => Some(serialize_error_response(&v)?),
-                }, // $1: deployment_error
+                &final_deployment_error, // $1: deployment_error
                 &final_deployment_config.map(|v| v.to_string()), // $2: deployment_config
                 &final_deployment_location, // $3: deployment_location
                 &final_suspend_info.map(|v| v.to_string()), // $4: suspend_info
                 &final_deployment_id, // $5: deployment_id
                 &final_deployment_initial.map(runtime_desired_status_to_string), // $6: deployment_initial
-                &new_deployment_resources_status.to_string(), // $7: deployment_resources_status,
-                &final_deployment_runtime_status.map(runtime_status_to_string), // $8: deployment_runtime_status,
-                &final_deployment_runtime_desired_status.map(runtime_desired_status_to_string), // $9: deployment_runtime_desired_status,
+                &deployment_resources_status, // $7: deployment_resources_status,
+                &deployment_runtime_status, // $8: deployment_runtime_status,
+                &deployment_runtime_desired_status, // $9: deployment_runtime_desired_status,
                 &tenant_id.0, // $10: tenant_id
                 &pipeline_id.0, // $11: id
             ],
@@ -1373,7 +1374,9 @@ pub(crate) async fn set_deployment_resources_status(
             txn,
             tenant_id,
             pipeline_id,
-            new_deployment_status,
+                &deployment_resources_status, // $7: deployment_resources_status,
+                &deployment_runtime_status, // $8: deployment_runtime_status,
+                &deployment_runtime_desired_status, // $9: deployment_runtime_desired_status,
             &final_deployment_error,
         )
         .await?;
@@ -1692,13 +1695,15 @@ pub(crate) async fn store_pipeline_lifecycle_event(
     transaction: &Transaction<'_>,
     tenant_id: TenantId,
     pipeline_id: PipelineId,
-    deployment_status: PipelineStatus,
+    deployment_resources_status: &str,
+    deployment_runtime_status: &Option<String>,
+    deployment_runtime_desired_status: &Option<String>,
     info: &Option<impl AsRef<str>>,
 ) -> Result<(), DBError> {
     let query = r#"
         INSERT INTO pipeline_lifecycle_events
-        (tenant_id, pipeline_id, deployment_status, info)
-        VALUES ($1, $2, $3, $4)
+        (tenant_id, pipeline_id, deployment_resources_status, deployment_runtime_status, deployment_runtime_desired_status, info)
+        VALUES ($1, $2, $3, $4, $5, $6)
     "#;
 
     transaction
@@ -1707,7 +1712,9 @@ pub(crate) async fn store_pipeline_lifecycle_event(
             &[
                 &tenant_id.0,
                 &pipeline_id.0,
-                &deployment_status.to_string(),
+                &deployment_resources_status,
+                &deployment_runtime_status,
+                &deployment_runtime_desired_status,
                 &info.as_ref().map(|v| v.as_ref()),
             ],
         )
@@ -1724,7 +1731,7 @@ pub(crate) async fn get_pipeline_lifecycle_events(
     limit: u32,
 ) -> Result<Vec<PipelineLifecycleEvent>, DBError> {
     let stmt = r#"
-        SELECT p.event_id, p.deployment_status, p.info, p.recorded_at
+        SELECT p.event_id, p.deployment_resources_status, p.deployment_runtime_status, p.deployment_runtime_desired_status, p.info, p.recorded_at
         FROM pipeline_lifecycle_events as p
         WHERE p.tenant_id = $1 AND p.pipeline_id = $2
         ORDER BY p.recorded_at ASC
@@ -1739,18 +1746,13 @@ pub(crate) async fn get_pipeline_lifecycle_events(
 
     let ret: Vec<PipelineLifecycleEvent> = rows
         .into_iter()
-        .filter_map(|row| {
-            if let Ok(status) = row.get::<_, String>(1).try_into() {
-                let event = PipelineLifecycleEvent {
-                    event_id: row.get(0),
-                    deployment_status: status,
-                    info: row.get(2),
-                    recorded_at: row.get(3),
-                };
-                Some(event)
-            } else {
-                None
-            }
+        .map(|row| PipelineLifecycleEvent {
+            event_id: row.get(0),
+            deployment_resources_status: row.get(1),
+            deployment_runtime_status: row.get(2),
+            deployment_runtime_desired_status: row.get(3),
+            info: row.get(4),
+            recorded_at: row.get(5),
         })
         .collect();
     Ok(ret)
diff --git a/crates/pipeline-manager/src/db/storage_postgres.rs b/crates/pipeline-manager/src/db/storage_postgres.rs
@@ -26,6 +26,7 @@ use async_trait::async_trait;
 use deadpool_postgres::{Manager, Pool, RecyclingMethod};
 use feldera_types::config::{PipelineConfig, RuntimeConfig};
 use feldera_types::error::ErrorResponse;
+use feldera_types::runtime_status::{ExtendedRuntimeStatus, RuntimeDesiredStatus};
 use log::{debug, info, log, warn, Level};
 use tokio_postgres::Row;
 use uuid::Uuid;
diff --git a/crates/pipeline-manager/src/db/test.rs b/crates/pipeline-manager/src/db/test.rs
