feat(config): rule http

16 files changed, 204 insertions, 20 deletions

diff --git a/crates/configuration/src/server/http_svc.rs b/crates/configuration/src/server/http_svc.rs
index 45cdcfa..b07aece 100644
--- a/crates/configuration/src/server/http_svc.rs
+++ b/crates/configuration/src/server/http_svc.rs
@@ -11,6 +11,7 @@ use utoipa_scalar::Servable as _;
 use crate::state::AppHandle;
 
 const TAG_ROUTING: &str = "Routing";
+const TAG_RULES: &str = "Rules";
 
 #[derive(OpenApi)]
 #[openapi(
diff --git a/crates/configuration/src/server/http_svc/routes.rs b/crates/configuration/src/server/http_svc/routes.rs
index bec7c77..281b231 100644
--- a/crates/configuration/src/server/http_svc/routes.rs
+++ b/crates/configuration/src/server/http_svc/routes.rs
@@ -1,4 +1,5 @@
 mod routing;
+mod rule;
 
 use utoipa_axum::{router::OpenApiRouter, routes};
 
@@ -7,8 +8,16 @@ use crate::state::AppHandle;
 pub fn router(store: AppHandle) -> OpenApiRouter {
     OpenApiRouter::new()
         .routes(routes!(
+            /* routing */
             routing::get_active::active_routing,
-            routing::post_routing::post_routing
+            routing::post_routing::post_routing,
+            routing::delete_routing::delete,
+            routing::replace_routing::replace,
+            /* rule */
+            rule::create::create_rule,
+            rule::update::update_rule_config,
+            rule::delete::delete_rule_config,
+            rule::get::get_rule,
         ))
         .with_state(store)
 }
diff --git a/crates/configuration/src/server/http_svc/routes/routing/delete_routing.rs b/crates/configuration/src/server/http_svc/routes/routing/delete_routing.rs
index 07148cc..23733d3 100644
--- a/crates/configuration/src/server/http_svc/routes/routing/delete_routing.rs
+++ b/crates/configuration/src/server/http_svc/routes/routing/delete_routing.rs
@@ -28,6 +28,7 @@ use crate::{
 #[axum::debug_handler]
 #[tracing::instrument(skip(state))]
 pub async fn delete(
+    version: Version,
     State(state): State<AppHandle>,
     Path(id): Path<String>,
     axum::Json(body): axum::Json<RoutingConfiguration>,
diff --git a/crates/configuration/src/server/http_svc/routes/routing/replace_routing.rs b/crates/configuration/src/server/http_svc/routes/routing/replace_routing.rs
index ccf184f..5c7cd02 100644
--- a/crates/configuration/src/server/http_svc/routes/routing/replace_routing.rs
+++ b/crates/configuration/src/server/http_svc/routes/routing/replace_routing.rs
@@ -27,6 +27,7 @@ use crate::{
 #[axum::debug_handler]
 #[tracing::instrument(skip(state))]
 pub async fn replace(
+    version: Version,
     State(state): State<AppHandle>,
     Path(id): Path<String>,
     axum::Json(body): axum::Json<RoutingConfiguration>,
diff --git a/crates/configuration/src/server/http_svc/routes/rule.rs b/crates/configuration/src/server/http_svc/routes/rule.rs
new file mode 100644
index 0000000..f4b0d33
--- /dev/null
+++ b/crates/configuration/src/server/http_svc/routes/rule.rs
@@ -0,0 +1,4 @@
+pub mod create;
+pub mod delete;
+pub mod get;
+pub mod update;
diff --git a/crates/configuration/src/server/http_svc/routes/rule/create.rs b/crates/configuration/src/server/http_svc/routes/rule/create.rs
new file mode 100644
index 0000000..809c00b
--- /dev/null
+++ b/crates/configuration/src/server/http_svc/routes/rule/create.rs
@@ -0,0 +1,38 @@
+use axum::{extract::State, response::IntoResponse};
+use warden_core::configuration::rule::{
+    RuleConfiguration, mutate_rule_configuration_server::MutateRuleConfiguration,
+};
+
+use crate::{
+    server::{error::AppError, http_svc::TAG_RULES, version::Version},
+    state::AppHandle,
+};
+
+/// Create rule configuration
+#[utoipa::path(
+    post,
+    path = "/{version}/rule",
+    params(
+        ("version" = Version, Path, description = "API version, e.g., v1, v2, v3"),
+    ),
+    responses((
+        status = CREATED,
+        body = RuleConfiguration,
+    )),
+    operation_id = "create_rule_configuration", // https://github.com/juhaku/utoipa/issues/1170
+    tag = TAG_RULES,
+    )
+]
+#[axum::debug_handler]
+#[tracing::instrument(skip(state))]
+pub async fn create_rule(
+    version: Version,
+    State(state): State<AppHandle>,
+    axum::Json(body): axum::Json<RuleConfiguration>,
+) -> Result<impl IntoResponse, AppError> {
+    let response = state
+        .create_rule_configuration(tonic::Request::new(body))
+        .await?
+        .into_inner();
+    Ok((axum::http::StatusCode::CREATED, axum::Json(response)))
+}
diff --git a/crates/configuration/src/server/http_svc/routes/rule/delete.rs b/crates/configuration/src/server/http_svc/routes/rule/delete.rs
new file mode 100644
index 0000000..2352fba
--- /dev/null
+++ b/crates/configuration/src/server/http_svc/routes/rule/delete.rs
@@ -0,0 +1,42 @@
+use axum::extract::{Query, State};
+use tonic::IntoRequest;
+use warden_core::configuration::rule::{
+    DeleteRuleConfigurationRequest, RuleConfiguration,
+    mutate_rule_configuration_server::MutateRuleConfiguration,
+};
+
+use crate::{
+    server::{error::AppError, http_svc::TAG_RULES, version::Version},
+    state::AppHandle,
+};
+
+/// Delete rule configuration
+#[utoipa::path(
+    delete,
+    path = "/{version}/rule",
+    responses((
+        status = OK,
+        body = RuleConfiguration
+    )),
+    params(
+        ("version" = Version, Path, description = "API version, e.g., v1, v2, v3"),
+        DeleteRuleConfigurationRequest,
+    ),
+    operation_id = "delete_rule_configuration", // https://github.com/juhaku/utoipa/issues/1170
+    tag = TAG_RULES,
+    )
+]
+#[axum::debug_handler]
+#[tracing::instrument(skip(state))]
+pub async fn delete_rule_config(
+    version: Version,
+    State(state): State<AppHandle>,
+    Query(body): Query<DeleteRuleConfigurationRequest>,
+) -> Result<axum::Json<RuleConfiguration>, AppError> {
+    let body = state
+        .delete_rule_configuration(body.into_request())
+        .await?
+        .into_inner();
+
+    Ok(axum::Json(body))
+}
diff --git a/crates/configuration/src/server/http_svc/routes/rule/get.rs b/crates/configuration/src/server/http_svc/routes/rule/get.rs
new file mode 100644
index 0000000..935eefb
--- /dev/null
+++ b/crates/configuration/src/server/http_svc/routes/rule/get.rs
@@ -0,0 +1,42 @@
+use axum::extract::{Query, State};
+use warden_core::configuration::rule::{
+    RuleConfiguration, RuleConfigurationRequest,
+    query_rule_configuration_server::QueryRuleConfiguration,
+};
+
+use crate::{
+    server::{error::AppError, http_svc::TAG_RULES, version::Version},
+    state::AppHandle,
+};
+
+/// Get rule configuration
+#[utoipa::path(
+    get,
+    path = "/{version}/rule",
+    responses((
+        status = OK,
+        body = RuleConfiguration
+    )),
+    params(
+        ("version" = Version, Path, description = "API version, e.g., v1, v2, v3"),
+        RuleConfigurationRequest
+    ),
+    operation_id = "get_rule_configuration", // https://github.com/juhaku/utoipa/issues/1170
+    tag = TAG_RULES,
+    )
+]
+#[axum::debug_handler]
+#[tracing::instrument(skip(state))]
+pub async fn get_rule(
+    version: Version,
+    State(state): State<AppHandle>,
+    Query(body): Query<RuleConfigurationRequest>,
+) -> Result<axum::Json<Option<RuleConfiguration>>, AppError> {
+    let response = state
+        .get_rule_configuration(tonic::Request::new(body))
+        .await?
+        .into_inner()
+        .configuration;
+
+    Ok(axum::Json(response))
+}
diff --git a/crates/configuration/src/server/http_svc/routes/rule/update.rs b/crates/configuration/src/server/http_svc/routes/rule/update.rs
new file mode 100644
index 0000000..7bf3fe0
--- /dev/null
+++ b/crates/configuration/src/server/http_svc/routes/rule/update.rs
@@ -0,0 +1,45 @@
+use axum::extract::State;
+use tonic::IntoRequest;
+use warden_core::configuration::rule::{
+    RuleConfiguration, UpdateRuleRequest, mutate_rule_configuration_server::MutateRuleConfiguration,
+};
+
+use crate::{
+    server::{error::AppError, http_svc::TAG_RULES, version::Version},
+    state::AppHandle,
+};
+
+/// Update the routing configuration
+#[utoipa::path(
+    put,
+    path = "/{version}/rule",
+    params(
+        ("version" = Version, Path, description = "API version, e.g., v1, v2, v3"),
+    ),
+    responses((
+        status = OK,
+        body = RuleConfiguration
+    )),
+    operation_id = "update rule configuration", // https://github.com/juhaku/utoipa/issues/1170
+    tag = TAG_RULES,
+    )
+]
+#[axum::debug_handler]
+#[tracing::instrument(skip(state))]
+pub async fn update_rule_config(
+    version: Version,
+    State(state): State<AppHandle>,
+    axum::Json(body): axum::Json<RuleConfiguration>,
+) -> Result<axum::Json<RuleConfiguration>, AppError> {
+    let config = state
+        .update_rule_configuration(
+            UpdateRuleRequest {
+                configuration: Some(body),
+            }
+            .into_request(),
+        )
+        .await?
+        .into_inner();
+
+    Ok(axum::Json(config))
+}
diff --git a/crates/router/src/main.rs b/crates/router/src/main.rs
index cc4c927..ad81700 100644
--- a/crates/router/src/main.rs
+++ b/crates/router/src/main.rs
@@ -52,9 +52,7 @@ async fn main() -> Result<()> {
         .take()
         .ok_or_else(|| anyhow::anyhow!("jetstream is not ready"))?;
 
-    let services = state::Services {
-        jetstream,
-    };
+    let services = state::Services { jetstream };
 
     processor::serve(services, config, provider)
         .await
diff --git a/crates/router/src/processor.rs b/crates/router/src/processor.rs
index b8c69f3..9afe726 100644
--- a/crates/router/src/processor.rs
+++ b/crates/router/src/processor.rs
@@ -1,20 +1,25 @@
 pub mod grpc;
+mod load;
+mod publish;
 mod reload;
 mod route;
-mod publish;
-mod load;
 
 use std::sync::Arc;
 
 use anyhow::Result;
-use async_nats::jetstream::{consumer::{pull, Consumer}, Context};
+use async_nats::jetstream::{
+    Context,
+    consumer::{Consumer, pull},
+};
+use futures_util::StreamExt;
 use tokio::signal;
 use tracing::{error, trace};
 use warden_stack::{Configuration, tracing::SdkTracerProvider};
-use futures_util::StreamExt;
-
-use crate::{cnfg::Nats, state::{AppHandle, AppState, Services}};
 
+use crate::{
+    cnfg::Nats,
+    state::{AppHandle, AppState, Services},
+};
 
 pub async fn serve(
     services: Services,
@@ -31,7 +36,6 @@ pub async fn serve(
     Ok(())
 }
 
-
 async fn run(state: AppHandle) -> anyhow::Result<()> {
     let config = Arc::clone(&state);
     let (consumer, _) = tokio::join!(
@@ -77,7 +81,6 @@ async fn get_or_create_stream(
         .await?)
 }
 
-
 async fn shutdown_signal(provider: SdkTracerProvider) -> Result<()> {
     let ctrl_c = async {
         signal::ctrl_c()
diff --git a/crates/router/src/processor/load.rs b/crates/router/src/processor/load.rs
index 9d3fd0d..b3fdf1c 100644
--- a/crates/router/src/processor/load.rs
+++ b/crates/router/src/processor/load.rs
@@ -1,7 +1,7 @@
 use opentelemetry_semantic_conventions::attribute;
-use tracing_opentelemetry::OpenTelemetrySpanExt;
 use tracing::{Instrument, debug, info, info_span, instrument, warn};
-use warden_core::{configuration::routing::RoutingConfiguration, google };
+use tracing_opentelemetry::OpenTelemetrySpanExt;
+use warden_core::{configuration::routing::RoutingConfiguration, google};
 
 use crate::{cnfg::CACHE_KEY, state::AppHandle};
 
diff --git a/crates/router/src/processor/publish.rs b/crates/router/src/processor/publish.rs
index 16dcec8..277b674 100644
--- a/crates/router/src/processor/publish.rs
+++ b/crates/router/src/processor/publish.rs
@@ -1,7 +1,7 @@
-use opentelemetry::global; 
+use opentelemetry::global;
 use opentelemetry_semantic_conventions::attribute;
-use tracing_opentelemetry::OpenTelemetrySpanExt;
 use tracing::{Instrument, Span, debug, info, info_span, warn};
+use tracing_opentelemetry::OpenTelemetrySpanExt;
 use warden_core::{configuration::routing::RoutingConfiguration, message::Payload};
 use warden_stack::tracing::telemetry::nats::injector;
 
diff --git a/crates/router/src/processor/reload.rs b/crates/router/src/processor/reload.rs
index c75465c..900b7ce 100644
--- a/crates/router/src/processor/reload.rs
+++ b/crates/router/src/processor/reload.rs
@@ -1,6 +1,6 @@
-use futures_util::StreamExt;
 use async_nats::jetstream::consumer;
-use tracing::{trace, debug, error, info};
+use futures_util::StreamExt;
+use tracing::{debug, error, info, trace};
 use uuid::Uuid;
 use warden_core::configuration::ReloadEvent;
 
diff --git a/crates/router/src/processor/route.rs b/crates/router/src/processor/route.rs
index 404c2ca..d51a86a 100644
--- a/crates/router/src/processor/route.rs
+++ b/crates/router/src/processor/route.rs
@@ -3,7 +3,7 @@ use std::{collections::HashSet, sync::Arc};
 
 use opentelemetry::global;
 use prost::Message;
-use tracing::{info_span, instrument, trace, trace_span, warn, Instrument, Span};
+use tracing::{Instrument, Span, info_span, instrument, trace, trace_span, warn};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 use warden_core::{google, message::Payload};
 use warden_stack::tracing::telemetry::nats;
diff --git a/crates/router/src/state.rs b/crates/router/src/state.rs
index e01629e..4ede2de 100644
--- a/crates/router/src/state.rs
+++ b/crates/router/src/state.rs
@@ -8,7 +8,7 @@ use tracing::error;
 use warden_core::configuration::routing::{
     RoutingConfiguration, query_routing_client::QueryRoutingClient,
 };
-use warden_stack::{Configuration};
+use warden_stack::Configuration;
 
 use crate::{
     cnfg::LocalConfig,