fix: accept version in query param for all get config apis

crates/context_aware_config/migrations/2024-04-22-122806_config_verions/up.sql

@@ -2,8 +2,9 @@
 -- Name: functions; Type: TABLE; Schema: public; Owner: -
 --
 CREATE TABLE public.config_versions (
-    id text PRIMARY KEY,
-    config json,
+    id bigint PRIMARY KEY,
+    config json NOT NULL,
+    version_type TEXT CHECK (version_type IN ('STABLE', 'NOISY')),
     created_at timestamp without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL
 );
 --

crates/context_aware_config/src/api/config/handlers.rs

@@ -19,7 +19,7 @@ use diesel::{
     r2d2::{ConnectionManager, PooledConnection},
     ExpressionMethods, PgConnection, QueryDsl, RunQueryDsl,
 };
-use serde_json::{json, Map, Value};
+use serde_json::{from_value, json, Map, Value};
 use service_utils::service::types::DbConnection;
 use service_utils::{bad_argument, db_error, unexpected_error};
 
@@ -99,6 +99,40 @@ fn is_not_modified(max_created_at: Option<NaiveDateTime>, req: &HttpRequest) ->
     max_created_at.is_some() && parsed_max <= last_modified
 }
 
+pub fn generate_config_from_version(
+    version: Option<Value>,
+    conn: &mut PooledConnection<ConnectionManager<PgConnection>>,
+) -> superposition::Result<Config> {
+    let config_version = match version {
+        None => config_versions::config_versions
+            .select(config_versions::config)
+            .order_by(config_versions::created_at.desc())
+            .first::<Value>(conn)
+            .map_err(|err| {
+                log::error!("failed to fetch config with error: {}", err);
+                db_error!(err)
+            }),
+        Some(version_val) => {
+            let version_id = from_value::<i64>(version_val).map_err(|e| {
+                log::error!("failed to decode version_id as integer: {}", e);
+                bad_argument!("version_id is not of type integer")
+            })?;
+            config_versions::config_versions
+                .select(config_versions::config)
+                .filter(config_versions::id.eq(version_id))
+                .get_result::<Value>(conn)
+                .map_err(|err| {
+                    log::error!("failed to fetch config with error: {}", err);
+                    db_error!(err)
+                })
+        }
+    }?;
+
+    serde_json::from_value::<Config>(config_version).map_err(|err| {
+        log::error!("failed to decode config: {}", err);
+        unexpected_error!("failed to decode config")
+    })
+}
 pub fn generate_cac(
     conn: &mut PooledConnection<ConnectionManager<PgConnection>>,
 ) -> superposition::Result<Config> {
@@ -161,8 +195,6 @@ async fn get(
     req: HttpRequest,
     db_conn: DbConnection,
 ) -> superposition::Result<HttpResponse> {
-    // let DbConnection(mut conn) = db_conn;
-    // let mut conn: PooledConnection<ConnectionManager<PgConnection>> = db_conn;
     let DbConnection(mut conn) = db_conn;
 
     let max_created_at = get_max_created_at(&mut conn)
@@ -188,36 +220,13 @@ async fn get(
         query_params_map.insert(
             key,
             value
-                .parse::<i32>()
+                .parse::<i64>()
                 .map_or_else(|_| json!(value), |int_val| json!(int_val)),
         );
     }
 
-    let mut config_version = json!({});
-    if let Some(Value::String(version_id)) = query_params_map.get("version") {
-        config_version = config_versions::config_versions
-            .select(config_versions::config)
-            .filter(config_versions::id.eq(version_id))
-            .get_result::<Value>(&mut conn)
-            .map_err(|err| {
-                log::error!("failed to fetch config with error: {}", err);
-                db_error!(err)
-            })?;
-    } else {
-        config_version = config_versions::config_versions
-            .select(config_versions::config)
-            .order_by(config_versions::created_at.desc())
-            .first::<Value>(&mut conn)
-            .map_err(|err| {
-                log::error!("failed to fetch config with error: {}", err);
-                db_error!(err)
-            })?;
-    };
-    let mut config = serde_json::from_value::<Config>(config_version).map_err(|err| {
-        log::error!("failed to decode config: {}", err);
-        unexpected_error!("failed to decode config")
-    })?;
-    query_params_map.remove("version");
+    let mut config =
+        generate_config_from_version(query_params_map.remove("version"), &mut conn)?;
     if let Some(prefix) = query_params_map.get("prefix") {
         let prefix_list: HashSet<&str> = prefix
             .as_str()
@@ -260,7 +269,7 @@ async fn get_resolved_config(
         query_params_map.insert(
             item.0,
             item.1
-                .parse::<i32>()
+                .parse::<i64>()
                 .map_or_else(|_| json!(item.1), |int_val| json!(int_val)),
         );
     }
@@ -275,7 +284,8 @@ async fn get_resolved_config(
         return Ok(HttpResponse::NotModified().finish());
     }
 
-    let res = generate_cac(&mut conn)?;
+    let res =
+        generate_config_from_version(query_params_map.remove("version"), &mut conn)?;
 
     let cac_client_contexts = res
         .contexts
@@ -344,11 +354,12 @@ async fn get_filtered_config(
         query_params_map.insert(
             key,
             value
-                .parse::<i32>()
+                .parse::<i64>()
                 .map_or_else(|_| json!(value), |int_val| json!(int_val)),
         );
     }
-    let config = generate_cac(&mut conn)?;
+    let config =
+        generate_config_from_version(query_params_map.remove("version"), &mut conn)?;
     let contexts = config.contexts;
 
     let filtered_context = filter_context(&contexts, &query_params_map)?;

crates/context_aware_config/src/api/context/handlers.rs

@@ -7,8 +7,9 @@ use crate::helpers::{
 use crate::{
     api::{
         context::types::{
-            ContextAction, ContextBulkResponse, DimensionCondition, MoveReq,
-            PaginationParams, PutReq, PutResp,
+            BulkOperationQParams, ContextAction, ContextBulkResponse, DeleteQParams,
+            DimensionCondition, MoveQParams, MoveReq, PaginationParams, PutQParams,
+            PutReq, PutResp,
         },
         dimension::get_all_dimension_schema_map,
     },
@@ -21,7 +22,10 @@ use crate::{
     },
 };
 use actix_web::web::Data;
-use service_utils::service::types::AppState;
+use service_utils::{
+    helpers::generate_snowflake_id,
+    service::types::{AppState, ConfigVersionType},
+};
 
 use actix_web::{
     delete, get, put,
@@ -281,19 +285,23 @@ fn put(
 async fn put_handler(
     state: Data<AppState>,
     req: Json<PutReq>,
+    qparams: Query<PutQParams>,
     mut db_conn: DbConnection,
     user: User,
 ) -> superposition::Result<Json<PutResp>> {
-    let mut snowflake_generator = state.snowflake_generator.lock().unwrap();
-    let version_id = snowflake_generator.real_time_generate();
+    let version_id = generate_snowflake_id(&state)?;
+    let version_type = qparams
+        .update_type
+        .map_or(ConfigVersionType::STABLE, |v| v)
+        .to_string();
     db_conn.transaction::<_, superposition::AppError, _>(|transaction_conn| {
         let put_response = put(req, transaction_conn, &user)
             .map(|resp| Json(resp))
             .map_err(|err: superposition::AppError| {
                 log::info!("context put failed with error: {:?}", err);
                 err
             })?;
-        add_config_version(version_id, transaction_conn)?;
+        add_config_version(version_id, version_type, transaction_conn)?;
         Ok(put_response)
     })
 }
@@ -368,19 +376,23 @@ async fn move_handler(
     state: Data<AppState>,
     path: Path<String>,
     req: Json<MoveReq>,
+    qparams: Query<MoveQParams>,
     mut db_conn: DbConnection,
     user: User,
 ) -> superposition::Result<Json<PutResp>> {
-    let mut snowflake_generator = state.snowflake_generator.lock().unwrap();
-    let version_id = snowflake_generator.real_time_generate();
+    let version_id = generate_snowflake_id(&state)?;
+    let version_type = qparams
+        .update_type
+        .map_or(ConfigVersionType::STABLE, |v| v)
+        .to_string();
     db_conn.transaction::<_, superposition::AppError, _>(|transaction_conn| {
         let move_reponse = r#move(path.into_inner(), req, transaction_conn, &user)
             .map(|resp| Json(resp))
             .map_err(|err| {
                 log::info!("move api failed with error: {:?}", err);
                 err
             })?;
-        add_config_version(version_id, transaction_conn)?;
+        add_config_version(version_id, version_type, transaction_conn)?;
         Ok(move_reponse)
     })
 }
@@ -438,22 +450,26 @@ async fn list_contexts(
 async fn delete_context(
     state: Data<AppState>,
     path: Path<String>,
+    qparams: Query<DeleteQParams>,
     db_conn: DbConnection,
     user: User,
 ) -> superposition::Result<HttpResponse> {
     use contexts::dsl;
     let DbConnection(mut conn) = db_conn;
 
     let ctx_id = path.into_inner();
-    let mut snowflake_generator = state.snowflake_generator.lock().unwrap();
-    let version_id = snowflake_generator.real_time_generate();
+    let version_id = generate_snowflake_id(&state)?;
+    let version_type = qparams
+        .update_type
+        .map_or(ConfigVersionType::STABLE, |v| v)
+        .to_string();
     conn.transaction::<_, superposition::AppError, _>(|transaction_conn| {
         let deleted_row =
             delete(dsl::contexts.filter(dsl::id.eq(&ctx_id))).execute(transaction_conn);
         match deleted_row {
             Ok(0) => Err(not_found!("Context Id `{}` doesn't exists", ctx_id)),
             Ok(_) => {
-                add_config_version(version_id, transaction_conn)?;
+                add_config_version(version_id, version_type, transaction_conn)?;
                 log::info!("{ctx_id} context deleted by {}", user.get_email());
                 Ok(HttpResponse::NoContent().finish())
             }
@@ -469,13 +485,17 @@ async fn delete_context(
 async fn bulk_operations(
     state: Data<AppState>,
     reqs: Json<Vec<ContextAction>>,
+    qparams: Query<BulkOperationQParams>,
     db_conn: DbConnection,
     user: User,
 ) -> superposition::Result<Json<Vec<ContextBulkResponse>>> {
     use contexts::dsl::contexts;
     let DbConnection(mut conn) = db_conn;
-    let mut snowflake_generator = state.snowflake_generator.lock().unwrap();
-    let version_id = snowflake_generator.real_time_generate();
+    let version_id = generate_snowflake_id(&state)?;
+    let version_type = qparams
+        .update_type
+        .map_or(ConfigVersionType::STABLE, |v| v)
+        .to_string();
 
     let mut response = Vec::<ContextBulkResponse>::new();
     conn.transaction::<_, superposition::AppError, _>(|transaction_conn| {
@@ -530,7 +550,7 @@ async fn bulk_operations(
                 }
             }
         }
-        add_config_version(version_id, transaction_conn)?;
+        add_config_version(version_id, version_type, transaction_conn)?;
         Ok(()) // Commit the transaction
     })?;
     Ok(Json(response))

crates/context_aware_config/src/api/context/types.rs

@@ -1,5 +1,6 @@
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
+use service_utils::service::types::ConfigVersionType;
 
 #[derive(Deserialize, Clone)]
 pub struct PutReq {
@@ -49,3 +50,23 @@ pub struct FunctionsInfo {
     pub name: String,
     pub code: Option<String>,
 }
+
+#[derive(Deserialize, Clone)]
+pub struct MoveQParams {
+    pub update_type: Option<ConfigVersionType>,
+}
+
+#[derive(Deserialize, Clone)]
+pub struct BulkOperationQParams {
+    pub update_type: Option<ConfigVersionType>,
+}
+
+#[derive(Deserialize, Clone)]
+pub struct PutQParams {
+    pub update_type: Option<ConfigVersionType>,
+}
+
+#[derive(Deserialize, Clone)]
+pub struct DeleteQParams {
+    pub update_type: Option<ConfigVersionType>,
+}

crates/context_aware_config/src/api/default_config/handlers.rs

@@ -1,6 +1,12 @@
 extern crate base64;
-use super::types::CreateReq;
-use service_utils::{bad_argument, unexpected_error, validation_error};
+use super::types::{CreateQParams, CreateReq};
+use service_utils::{
+    bad_argument,
+    helpers::generate_snowflake_id,
+    result as superposition,
+    service::types::{AppState, ConfigVersionType, DbConnection},
+    unexpected_error, validation_error,
+};
 
 use superposition_types::{SuperpositionUser, User};
 
@@ -23,10 +29,6 @@ use diesel::{
 };
 use jsonschema::{Draft, JSONSchema, ValidationError};
 use serde_json::{json, Value};
-use service_utils::{
-    result as superposition,
-    service::types::{AppState, DbConnection},
-};
 
 pub fn endpoints() -> Scope {
     Scope::new("").service(create).service(get)
@@ -37,12 +39,17 @@ async fn create(
     state: Data<AppState>,
     key: web::Path<String>,
     request: web::Json<CreateReq>,
+    qparams: web::Query<CreateQParams>,
     db_conn: DbConnection,
     user: User,
 ) -> superposition::Result<HttpResponse> {
     let DbConnection(mut conn) = db_conn;
     let req = request.into_inner();
     let key = key.into_inner();
+    let version_type = qparams
+        .update_type
+        .map_or(ConfigVersionType::STABLE, |v| v)
+        .to_string();
 
     if req.value.is_none() && req.schema.is_none() && req.function_name.is_none() {
         log::error!("No data provided in the request body for {key}");
@@ -140,8 +147,7 @@ async fn create(
             )?;
         }
     }
-    let mut snowflake_generator = state.snowflake_generator.lock().unwrap();
-    let version_id = snowflake_generator.real_time_generate();
+    let version_id = generate_snowflake_id(&state)?;
     conn.transaction::<_, superposition::AppError, _>(|transaction_conn| {
         let upsert = diesel::insert_into(default_configs)
             .values(&default_config)
@@ -161,7 +167,7 @@ async fn create(
                 ))
             }
         }?;
-        add_config_version(version_id, transaction_conn)?;
+        add_config_version(version_id, version_type, transaction_conn)?;
         Ok(ok_resp)
     })
 }

crates/context_aware_config/src/api/default_config/types.rs

@@ -1,5 +1,6 @@
 use serde::{Deserialize, Deserializer};
 use serde_json::{Map, Value};
+use service_utils::service::types::ConfigVersionType;
 
 #[derive(Debug, Deserialize)]
 pub struct CreateReq {
@@ -17,3 +18,8 @@ where
     let value: Value = Deserialize::deserialize(deserializer)?;
     Ok(Some(value))
 }
+
+#[derive(Deserialize, Clone)]
+pub struct CreateQParams {
+    pub update_type: Option<ConfigVersionType>,
+}