From 75b0af8b7d27e3c00b18eae44d1c21ea26a75559 Mon Sep 17 00:00:00 2001 From: Venu Madhav Date: Wed, 16 Sep 2026 23:10:43 +0530 Subject: [PATCH] feat(observability): add merchant thresholds APIs --- crates/api_models/src/observability.rs | 1 + .../src/observability/alert_manager.rs | 1 + .../alert_manager/merchant_thresholds.rs | 185 +++++ crates/diesel_models/src/observability.rs | 1 + .../src/observability/alert_manager.rs | 1 + .../alert_manager/merchant_thresholds.rs | 107 +++ .../diesel_models/src/observability/query.rs | 1 + .../src/observability/query/alert_manager.rs | 1 + .../alert_manager/merchant_thresholds.rs | 255 ++++++ .../diesel_models/src/observability/schema.rs | 3 +- .../up.sql | 2 +- crates/observability/src/core.rs | 1 + .../observability/src/core/alert_manager.rs | 1 + .../core/alert_manager/merchant_thresholds.rs | 103 +++ crates/observability/src/db.rs | 9 +- crates/observability/src/db/alert_manager.rs | 1 + .../db/alert_manager/merchant_thresholds.rs | 124 +++ crates/observability/src/domain_models.rs | 1 + .../src/domain_models/alert_manager.rs | 1 + .../alert_manager/merchant_thresholds.rs | 768 ++++++++++++++++++ crates/observability/src/routes.rs | 1 + .../observability/src/routes/alert_manager.rs | 1 + .../alert_manager/merchant_thresholds.rs | 93 +++ crates/observability/src/routes/app.rs | 23 +- .../tests/merchant_thresholds.rs | 182 +++++ 25 files changed, 1862 insertions(+), 5 deletions(-) create mode 100644 crates/api_models/src/observability/alert_manager.rs create mode 100644 crates/api_models/src/observability/alert_manager/merchant_thresholds.rs create mode 100644 crates/diesel_models/src/observability/alert_manager.rs create mode 100644 crates/diesel_models/src/observability/alert_manager/merchant_thresholds.rs create mode 100644 crates/diesel_models/src/observability/query/alert_manager.rs create mode 100644 crates/diesel_models/src/observability/query/alert_manager/merchant_thresholds.rs create mode 100644 crates/observability/src/core/alert_manager.rs create mode 100644 crates/observability/src/core/alert_manager/merchant_thresholds.rs create mode 100644 crates/observability/src/db/alert_manager.rs create mode 100644 crates/observability/src/db/alert_manager/merchant_thresholds.rs create mode 100644 crates/observability/src/domain_models/alert_manager.rs create mode 100644 crates/observability/src/domain_models/alert_manager/merchant_thresholds.rs create mode 100644 crates/observability/src/routes/alert_manager.rs create mode 100644 crates/observability/src/routes/alert_manager/merchant_thresholds.rs create mode 100644 crates/observability/tests/merchant_thresholds.rs diff --git a/crates/api_models/src/observability.rs b/crates/api_models/src/observability.rs index 0b1b00a4af4..3411a82d2d7 100644 --- a/crates/api_models/src/observability.rs +++ b/crates/api_models/src/observability.rs @@ -1,3 +1,4 @@ //! Request and response types for the standalone `observability` service. +pub mod alert_manager; pub mod alerts_info; diff --git a/crates/api_models/src/observability/alert_manager.rs b/crates/api_models/src/observability/alert_manager.rs new file mode 100644 index 00000000000..bb0dfb1d610 --- /dev/null +++ b/crates/api_models/src/observability/alert_manager.rs @@ -0,0 +1 @@ +pub mod merchant_thresholds; diff --git a/crates/api_models/src/observability/alert_manager/merchant_thresholds.rs b/crates/api_models/src/observability/alert_manager/merchant_thresholds.rs new file mode 100644 index 00000000000..67908a9c714 --- /dev/null +++ b/crates/api_models/src/observability/alert_manager/merchant_thresholds.rs @@ -0,0 +1,185 @@ +//! Per-merchant threshold overrides, as the observability service's +//! `/alerts/alerts_manager/merchant_thresholds` routes accept and return them. + +use std::collections::BTreeMap; + +use serde::{Deserialize, Deserializer, Serialize}; +use serde_json::Value; +use time::PrimitiveDateTime; + +/// One value, or a list of values to match any of. An empty list means no filter on this field. +#[derive(Clone, Debug, Deserialize)] +#[serde(untagged)] +pub enum MerchantThresholdsTextFilter { + One(String), + Many(Vec), +} + +/// A range of `last_updated_at`, or a bare timestamp meaning "at or after". +#[derive(Clone, Debug, Deserialize)] +#[serde(untagged)] +pub enum MerchantThresholdsTimeFilter { + Range { + #[serde(default, with = "common_utils::custom_serde::iso8601::option")] + start: Option, + #[serde(default, with = "common_utils::custom_serde::iso8601::option")] + end: Option, + }, + From(#[serde(with = "common_utils::custom_serde::iso8601")] PrimitiveDateTime), +} + +/// The body of `POST /alerts/alerts_manager/merchant_thresholds/list`. +/// +/// Every field is optional; `{}` matches every row. +#[derive(Clone, Debug, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MerchantThresholdsListRequest { + pub id: Option, + pub name: Option, + pub product: Option, + pub merchant_id: Option, + pub profile_id: Option, + pub author: Option, + pub is_enabled: Option, + pub metadata: Option>, + pub last_updated_at: Option, +} + +/// The body of `POST /alerts/alerts_manager/merchant_thresholds`. +/// +/// `name`, `product`, `merchant_id` and `profile_id` are required strings, empty allowed. On a +/// conflict of `(name, product, merchant_id, profile_id, is_enabled, author)`, only the non-key +/// columns actually sent here are overwritten. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MerchantThresholdsUpsertRequest { + pub name: String, + pub product: String, + pub merchant_id: String, + pub profile_id: String, + pub thresholds_min_volume: Option, + pub thresholds_min_impacted_volume: Option, + pub thresholds_tolerance: Option, + pub thresholds_diff_threshold: Option, + pub thresholds_merchant_impact: Option, + pub thresholds_alert_period: Option, + pub thresholds_min_observations: Option, + pub thresholds_min_history_volume: Option, + pub thresholds_filter_percentile: Option, + pub thresholds_current_min_volume: Option, + pub metadata: Option, + pub author: Option, + pub is_enabled: Option, + #[serde(default, with = "common_utils::custom_serde::iso8601::option")] + pub last_updated_at: Option, +} + +/// Deserialize a field that distinguishes "absent" from "sent as `null`": absent leaves the +/// `#[serde(default)]` outer `None`, `null` becomes `Some(None)`, and a value becomes `Some(Some(v))`. +fn present<'de, D, T>(deserializer: D) -> Result>, D::Error> +where + D: Deserializer<'de>, + T: Deserialize<'de>, +{ + Option::::deserialize(deserializer).map(Some) +} + +/// The [`present`] of an ISO 8601 timestamp. +fn present_iso8601<'de, D>(deserializer: D) -> Result>, D::Error> +where + D: Deserializer<'de>, +{ + common_utils::custom_serde::iso8601::option::deserialize(deserializer).map(Some) +} + +/// The body of `POST /alerts/alerts_manager/merchant_thresholds/update`. +/// +/// `name`, `product`, `merchant_id` and `profile_id` are each optional keys, but at least one is +/// required; only the keys sent filter which rows are touched. Every other field is absent to keep +/// a column, `null` to clear it, or a value to set it. +#[derive(Clone, Debug, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MerchantThresholdsUpdateRequest { + pub name: Option, + pub product: Option, + pub merchant_id: Option, + pub profile_id: Option, + #[serde(default, deserialize_with = "present")] + pub thresholds_min_volume: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_min_impacted_volume: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_tolerance: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_diff_threshold: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_merchant_impact: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_alert_period: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_min_observations: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_min_history_volume: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_filter_percentile: Option>, + #[serde(default, deserialize_with = "present")] + pub thresholds_current_min_volume: Option>, + #[serde(default, deserialize_with = "present")] + pub metadata: Option>, + #[serde(default, deserialize_with = "present")] + pub author: Option>, + #[serde(default, deserialize_with = "present")] + pub is_enabled: Option>, + #[serde(default, deserialize_with = "present_iso8601")] + pub last_updated_at: Option>, +} + +/// The body of `POST /alerts/alerts_manager/merchant_thresholds/delete`. +/// +/// `name` and `product` are required; `merchant_id` and `profile_id` narrow further when sent. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MerchantThresholdsDeleteByFilterRequest { + pub name: String, + pub product: String, + pub merchant_id: Option, + pub profile_id: Option, +} + +/// Built from the `{id}` path of `DELETE /alerts/alerts_manager/merchant_thresholds/{id}`. +#[derive(Clone, Debug)] +pub struct MerchantThresholdsDeleteRequest { + pub id: String, +} + +/// One stored merchant threshold override. +#[derive(Clone, Debug, Serialize)] +pub struct MerchantThresholdsResponse { + pub id: String, + pub name: Option, + pub product: Option, + pub merchant_id: Option, + pub profile_id: String, + pub thresholds_min_volume: Option, + pub thresholds_min_impacted_volume: Option, + pub thresholds_tolerance: Option, + pub thresholds_diff_threshold: Option, + pub thresholds_merchant_impact: Option, + pub thresholds_alert_period: Option, + pub thresholds_min_observations: Option, + pub thresholds_min_history_volume: Option, + pub thresholds_filter_percentile: Option, + pub thresholds_current_min_volume: Option, + pub metadata: Option, + pub author: Option, + pub is_enabled: Option, + #[serde(with = "common_utils::custom_serde::iso8601::option")] + pub last_updated_at: Option, +} + +/// The body of every merchant threshold route that returns more than one row. +#[derive(Clone, Debug, Serialize)] +pub struct MerchantThresholdsListResponse { + pub count: usize, + pub data: Vec, +} diff --git a/crates/diesel_models/src/observability.rs b/crates/diesel_models/src/observability.rs index c834167fce2..d799f38b533 100644 --- a/crates/diesel_models/src/observability.rs +++ b/crates/diesel_models/src/observability.rs @@ -4,6 +4,7 @@ //! owns, with its own migration lineage under `crates/observability/migrations` and its own diesel //! configuration in `diesel_observability.toml`. +pub mod alert_manager; pub mod alerts_info; pub mod query; pub mod schema; diff --git a/crates/diesel_models/src/observability/alert_manager.rs b/crates/diesel_models/src/observability/alert_manager.rs new file mode 100644 index 00000000000..bb0dfb1d610 --- /dev/null +++ b/crates/diesel_models/src/observability/alert_manager.rs @@ -0,0 +1 @@ +pub mod merchant_thresholds; diff --git a/crates/diesel_models/src/observability/alert_manager/merchant_thresholds.rs b/crates/diesel_models/src/observability/alert_manager/merchant_thresholds.rs new file mode 100644 index 00000000000..2f3f1207815 --- /dev/null +++ b/crates/diesel_models/src/observability/alert_manager/merchant_thresholds.rs @@ -0,0 +1,107 @@ +//! Per-merchant threshold overrides: a row overrides one alert's thresholds for one merchant +//! (and, optionally, one profile), and a job coalesces it with the alert's own definition. + +use diesel::{AsChangeset, Identifiable, Insertable, Queryable, Selectable}; +use time::PrimitiveDateTime; + +use crate::observability::schema::merchant_thresholds; + +/// A new merchant threshold override. +/// +/// The application supplies the ID. A `None` `author` or `is_enabled` falls back to its column +/// default rather than being stored as `null`. +#[derive(Clone, Debug, Insertable, serde::Serialize, serde::Deserialize)] +#[diesel(table_name = merchant_thresholds)] +pub struct MerchantThresholdsNew { + pub id: String, + pub name: String, + pub product: String, + pub merchant_id: String, + pub profile_id: String, + pub thresholds_min_volume: Option, + pub thresholds_min_impacted_volume: Option, + pub thresholds_tolerance: Option, + pub thresholds_diff_threshold: Option, + pub thresholds_merchant_impact: Option, + pub thresholds_alert_period: Option, + pub thresholds_min_observations: Option, + pub thresholds_min_history_volume: Option, + pub thresholds_filter_percentile: Option, + pub thresholds_current_min_volume: Option, + pub metadata: Option, + pub author: Option, + pub is_enabled: Option, + pub last_updated_at: Option, +} + +/// A stored merchant threshold override. +#[derive( + Clone, Debug, Identifiable, Queryable, Selectable, serde::Serialize, serde::Deserialize, +)] +#[diesel(table_name = merchant_thresholds, primary_key(id), check_for_backend(diesel::pg::Pg))] +pub struct MerchantThresholds { + pub id: String, + pub name: Option, + pub product: Option, + pub merchant_id: Option, + pub profile_id: String, + pub thresholds_min_volume: Option, + pub thresholds_min_impacted_volume: Option, + pub thresholds_tolerance: Option, + pub thresholds_diff_threshold: Option, + pub thresholds_merchant_impact: Option, + pub thresholds_alert_period: Option, + pub thresholds_min_observations: Option, + pub thresholds_min_history_volume: Option, + pub thresholds_filter_percentile: Option, + pub thresholds_current_min_volume: Option, + pub metadata: Option, + pub author: Option, + pub is_enabled: Option, + pub last_updated_at: Option, +} + +/// What `update_by_key_filter` may set or clear. `None` skips a column; `Some(None)` writes NULL. +#[derive(Clone, Debug, Default, AsChangeset)] +#[diesel(table_name = merchant_thresholds)] +pub struct MerchantThresholdsUpdate { + pub thresholds_min_volume: Option>, + pub thresholds_min_impacted_volume: Option>, + pub thresholds_tolerance: Option>, + pub thresholds_diff_threshold: Option>, + pub thresholds_merchant_impact: Option>, + pub thresholds_alert_period: Option>, + pub thresholds_min_observations: Option>, + pub thresholds_min_history_volume: Option>, + pub thresholds_filter_percentile: Option>, + pub thresholds_current_min_volume: Option>, + pub metadata: Option>, + pub author: Option>, + pub is_enabled: Option>, + pub last_updated_at: Option>, +} + +/// The filters `list_by_filter` accepts. `metadata` pairs match `metadata ->> key = ANY(values)`. +#[derive(Clone, Debug, Default)] +pub struct MerchantThresholdsFilter { + pub ids: Option>, + pub names: Option>, + pub products: Option>, + pub merchant_ids: Option>, + pub profile_ids: Option>, + pub authors: Option>, + pub is_enabled: Option, + pub metadata: Vec<(String, Vec)>, + pub updated_from: Option, + pub updated_to: Option, +} + +/// The keys `update_by_key_filter` and `delete_by_key_filter` select rows by. Only the keys sent +/// filter; at least one must be sent. +#[derive(Clone, Debug, Default)] +pub struct MerchantThresholdsKeyFilter { + pub name: Option, + pub product: Option, + pub merchant_id: Option, + pub profile_id: Option, +} diff --git a/crates/diesel_models/src/observability/query.rs b/crates/diesel_models/src/observability/query.rs index 845cfa628e7..d2975523aef 100644 --- a/crates/diesel_models/src/observability/query.rs +++ b/crates/diesel_models/src/observability/query.rs @@ -1,3 +1,4 @@ //! Queries against the observability database's tables. +pub mod alert_manager; pub mod alerts_info; diff --git a/crates/diesel_models/src/observability/query/alert_manager.rs b/crates/diesel_models/src/observability/query/alert_manager.rs new file mode 100644 index 00000000000..bb0dfb1d610 --- /dev/null +++ b/crates/diesel_models/src/observability/query/alert_manager.rs @@ -0,0 +1 @@ +pub mod merchant_thresholds; diff --git a/crates/diesel_models/src/observability/query/alert_manager/merchant_thresholds.rs b/crates/diesel_models/src/observability/query/alert_manager/merchant_thresholds.rs new file mode 100644 index 00000000000..4edce69b32f --- /dev/null +++ b/crates/diesel_models/src/observability/query/alert_manager/merchant_thresholds.rs @@ -0,0 +1,255 @@ +use async_bb8_diesel::AsyncRunQueryDsl; +use diesel::{ + associations::HasTable, + debug_query, + expression_methods::{PgAnyJsonExpressionMethods, PgJsonbExpressionMethods}, + pg::Pg, + sql_types::{Bool, Nullable}, + BoolExpressionMethods, BoxableExpression, ExpressionMethods, NullableExpressionMethods, + QueryDsl, +}; +use error_stack::report; +use router_env::logger; + +use crate::{ + errors::DatabaseError, + observability::{ + alert_manager::merchant_thresholds::{ + MerchantThresholds, MerchantThresholdsFilter, MerchantThresholdsKeyFilter, + MerchantThresholdsNew, MerchantThresholdsUpdate, + }, + schema::merchant_thresholds::{self, dsl}, + }, + query::generics::{ + self, + db_metrics::{track_database_call, DatabaseOperation}, + }, + DatabaseConnectionWithContext, StorageResult, +}; + +impl MerchantThresholdsNew { + pub async fn upsert( + self, + conn: &DatabaseConnectionWithContext<'_>, + on_conflict: Option, + ) -> StorageResult> { + let conflict_target = ( + dsl::name, + dsl::product, + dsl::merchant_id, + dsl::profile_id, + dsl::is_enabled, + dsl::author, + ); + + match on_conflict { + Some(update) => { + let query = diesel::insert_into(::table()) + .values(self) + .on_conflict(conflict_target) + .do_update() + .set(update); + + logger::debug!(query = %debug_query::(&query).to_string()); + + track_database_call::( + conn.request_id(), + conn.event_emitter(), + DatabaseOperation::Insert, + query.get_results_async(conn.raw_connection()), + ) + .await + .map_err(storage_error) + } + None => { + let query = diesel::insert_into(::table()) + .values(self) + .on_conflict(conflict_target) + .do_nothing(); + + logger::debug!(query = %debug_query::(&query).to_string()); + + track_database_call::( + conn.request_id(), + conn.event_emitter(), + DatabaseOperation::Insert, + query.get_results_async(conn.raw_connection()), + ) + .await + .map_err(storage_error) + } + } + } +} + +impl MerchantThresholds { + pub async fn list_by_filter( + conn: &DatabaseConnectionWithContext<'_>, + filter: MerchantThresholdsFilter, + ) -> StorageResult> { + let mut query = crate::list::into_boxed_list(Self::table()); + + if let Some(ids) = filter.ids { + query = query.filter(dsl::id.eq_any(ids)); + } + if let Some(names) = filter.names { + query = query.filter(dsl::name.eq_any(names)); + } + if let Some(products) = filter.products { + query = query.filter(dsl::product.eq_any(products)); + } + if let Some(merchant_ids) = filter.merchant_ids { + query = query.filter(dsl::merchant_id.eq_any(merchant_ids)); + } + if let Some(profile_ids) = filter.profile_ids { + query = query.filter(dsl::profile_id.eq_any(profile_ids)); + } + if let Some(authors) = filter.authors { + query = query.filter(dsl::author.eq_any(authors)); + } + if let Some(is_enabled) = filter.is_enabled { + query = query.filter(dsl::is_enabled.eq(is_enabled)); + } + for (key, values) in filter.metadata { + query = query.filter(dsl::metadata.retrieve_as_text(key).eq_any(values)); + } + if let Some(from) = filter.updated_from { + query = query.filter(dsl::last_updated_at.ge(from)); + } + if let Some(to) = filter.updated_to { + query = query.filter(dsl::last_updated_at.le(to)); + } + + logger::debug!(query = %debug_query::(&query).to_string()); + + track_database_call::( + conn.request_id(), + conn.event_emitter(), + DatabaseOperation::Filter, + query.get_results_async(conn.raw_connection()), + ) + .await + .map_err(storage_error) + } + + pub async fn update_by_key_filter( + conn: &DatabaseConnectionWithContext<'_>, + filter: MerchantThresholdsKeyFilter, + update: MerchantThresholdsUpdate, + metadata_merge: Option, + ) -> StorageResult> { + let predicate = key_predicate(filter)?; + let merge = metadata_merge + .map(|value| dsl::metadata.eq(PgJsonbExpressionMethods::concat(dsl::metadata, value))); + + let query = diesel::update(Self::table().filter(predicate)).set((update, merge)); + + logger::debug!(query = %debug_query::(&query).to_string()); + + track_database_call::( + conn.request_id(), + conn.event_emitter(), + DatabaseOperation::UpdateWithResults, + query.get_results_async(conn.raw_connection()), + ) + .await + .map_err(storage_error) + } + + pub async fn delete_by_id( + conn: &DatabaseConnectionWithContext<'_>, + id: String, + ) -> StorageResult { + generics::generic_delete_one_with_result::<::Table, _, _>( + conn, + dsl::id.eq(id), + ) + .await + } + + pub async fn delete_by_key_filter( + conn: &DatabaseConnectionWithContext<'_>, + filter: MerchantThresholdsKeyFilter, + ) -> StorageResult> { + let predicate = key_predicate(filter)?; + let query = diesel::delete(Self::table().filter(predicate)); + + logger::debug!(query = %debug_query::(&query).to_string()); + + track_database_call::( + conn.request_id(), + conn.event_emitter(), + DatabaseOperation::DeleteWithResult, + query.get_results_async(conn.raw_connection()), + ) + .await + .map_err(storage_error) + } +} + +/// The `WHERE` clause `update_by_key_filter` and `delete_by_key_filter` share: only the keys sent +/// filter, folded with `AND`. Refuses an empty filter, so neither statement ever runs without one — +/// an unkeyed update or delete would touch every row. +#[allow(clippy::type_complexity)] +fn key_predicate( + filter: MerchantThresholdsKeyFilter, +) -> StorageResult< + Box>>, +> { + let mut predicates: Vec< + Box>>, + > = Vec::new(); + + if let Some(name) = filter.name { + predicates.push(Box::new(dsl::name.eq(name).nullable())); + } + if let Some(product) = filter.product { + predicates.push(Box::new(dsl::product.eq(product).nullable())); + } + if let Some(merchant_id) = filter.merchant_id { + predicates.push(Box::new(dsl::merchant_id.eq(merchant_id).nullable())); + } + if let Some(profile_id) = filter.profile_id { + predicates.push(Box::new(dsl::profile_id.eq(profile_id).nullable())); + } + + predicates + .into_iter() + .reduce(|left, right| Box::new(left.and(right))) + .ok_or_else(|| { + report!(DatabaseError::QueryGenerationFailed).attach_printable( + "merchant_thresholds update or delete needs at least one of name, product, \ + merchant_id or profile_id", + ) + }) +} + +fn storage_error(error: diesel::result::Error) -> error_stack::Report { + let context = match &error { + diesel::result::Error::DatabaseError( + diesel::result::DatabaseErrorKind::UniqueViolation, + _, + ) => DatabaseError::UniqueViolation, + diesel::result::Error::NotFound => DatabaseError::NotFound, + diesel::result::Error::QueryBuilderError(_) => DatabaseError::NoFieldsToUpdate, + _ => DatabaseError::Others, + }; + + report!(error).change_context(context) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + + #[test] + fn key_predicate_refuses_an_empty_filter() { + assert!(key_predicate(MerchantThresholdsKeyFilter::default()).is_err()); + assert!(key_predicate(MerchantThresholdsKeyFilter { + name: Some("Volume Drop".to_owned()), + ..Default::default() + }) + .is_ok()); + } +} diff --git a/crates/diesel_models/src/observability/schema.rs b/crates/diesel_models/src/observability/schema.rs index 83a14cc4db3..42539e01667 100644 --- a/crates/diesel_models/src/observability/schema.rs +++ b/crates/diesel_models/src/observability/schema.rs @@ -143,7 +143,8 @@ diesel::table! { diesel::table! { merchant_thresholds (id) { - id -> Uuid, + #[max_length = 64] + id -> Varchar, #[max_length = 64] name -> Nullable, #[max_length = 64] diff --git a/crates/observability/migrations/2026-09-16-000001_create_merchant_thresholds/up.sql b/crates/observability/migrations/2026-09-16-000001_create_merchant_thresholds/up.sql index bab27197ede..d6ecfeeebc7 100644 --- a/crates/observability/migrations/2026-09-16-000001_create_merchant_thresholds/up.sql +++ b/crates/observability/migrations/2026-09-16-000001_create_merchant_thresholds/up.sql @@ -1,5 +1,5 @@ CREATE TABLE IF NOT EXISTS merchant_thresholds ( - id UUID DEFAULT gen_random_uuid() PRIMARY KEY, + id VARCHAR(64) NOT NULL PRIMARY KEY, name VARCHAR(64), product VARCHAR(64), merchant_id VARCHAR(64), diff --git a/crates/observability/src/core.rs b/crates/observability/src/core.rs index 8cdf87d1aa2..e5164af9f95 100644 --- a/crates/observability/src/core.rs +++ b/crates/observability/src/core.rs @@ -6,6 +6,7 @@ //! Distinct from [`crate::domain`], which holds the traits and the types they exchange: `domain` //! says what delivering an alert *is*, `core` says what one HTTP request does about it. +pub mod alert_manager; pub mod alerts_info; pub mod cloudwatch; pub mod notifier; diff --git a/crates/observability/src/core/alert_manager.rs b/crates/observability/src/core/alert_manager.rs new file mode 100644 index 00000000000..bb0dfb1d610 --- /dev/null +++ b/crates/observability/src/core/alert_manager.rs @@ -0,0 +1 @@ +pub mod merchant_thresholds; diff --git a/crates/observability/src/core/alert_manager/merchant_thresholds.rs b/crates/observability/src/core/alert_manager/merchant_thresholds.rs new file mode 100644 index 00000000000..2e3ed6af27b --- /dev/null +++ b/crates/observability/src/core/alert_manager/merchant_thresholds.rs @@ -0,0 +1,103 @@ +//! Per-request logic for per-merchant threshold overrides. + +use api_models::observability::alert_manager::merchant_thresholds::{ + MerchantThresholdsDeleteByFilterRequest, MerchantThresholdsDeleteRequest, + MerchantThresholdsListRequest, MerchantThresholdsListResponse, MerchantThresholdsResponse, + MerchantThresholdsUpdateRequest, MerchantThresholdsUpsertRequest, +}; +use error_stack::ResultExt; + +use crate::{ + domain_models::alert_manager::merchant_thresholds::{ + list_response, parse_id, MerchantThresholdsBulkUpdate, MerchantThresholdsFilter, + MerchantThresholdsKeyFilter, MerchantThresholdsNew, + }, + errors::{ObservabilityApiResult, ObservabilityError, StorageErrorExt}, + state::AppState, +}; + +/// List rows matching the given filter. +pub async fn list_merchant_thresholds( + state: AppState, + request: MerchantThresholdsListRequest, +) -> ObservabilityApiResult { + let filter = MerchantThresholdsFilter::try_from(request)?; + + let rows = state + .store + .list_merchant_thresholds_by_filter(filter) + .await + .change_context(ObservabilityError::InternalServerError) + .attach_printable("Failed to list merchant_thresholds")?; + + Ok(list_response(rows)) +} + +/// Insert a new override, or, on a conflict of `(name, product, merchant_id, profile_id, +/// is_enabled, author)`, set only the non-key columns actually sent. +pub async fn upsert_merchant_threshold( + state: AppState, + request: MerchantThresholdsUpsertRequest, +) -> ObservabilityApiResult { + let new = MerchantThresholdsNew::try_from(request)?; + + let rows = state + .store + .upsert_merchant_threshold(new) + .await + .change_context(ObservabilityError::InternalServerError) + .attach_printable("Failed to upsert into merchant_thresholds")?; + + Ok(list_response(rows)) +} + +/// Set or clear columns on every row matching the keys sent. +pub async fn update_merchant_thresholds( + state: AppState, + request: MerchantThresholdsUpdateRequest, +) -> ObservabilityApiResult { + let MerchantThresholdsBulkUpdate { filter, update } = + MerchantThresholdsBulkUpdate::try_from(request)?; + + let rows = state + .store + .update_merchant_thresholds_by_filter(filter, update) + .await + .to_duplicate_response(ObservabilityError::DuplicateResource) + .attach_printable("Failed to update merchant_thresholds")?; + + Ok(list_response(rows)) +} + +/// Delete one override by id. +pub async fn delete_merchant_threshold( + state: AppState, + request: MerchantThresholdsDeleteRequest, +) -> ObservabilityApiResult { + let id = parse_id(&request.id)?; + + let row = state + .store + .delete_merchant_threshold_by_id(id) + .await + .to_not_found_response(ObservabilityError::ResourceNotFound)?; + + Ok(MerchantThresholdsResponse::from(row)) +} + +/// Delete every row matching `name`, `product` and, when sent, `merchant_id` and `profile_id`. +pub async fn delete_merchant_thresholds_by_filter( + state: AppState, + request: MerchantThresholdsDeleteByFilterRequest, +) -> ObservabilityApiResult { + let filter = MerchantThresholdsKeyFilter::from(request); + + let rows = state + .store + .delete_merchant_thresholds_by_filter(filter) + .await + .change_context(ObservabilityError::InternalServerError) + .attach_printable("Failed to delete from merchant_thresholds")?; + + Ok(list_response(rows)) +} diff --git a/crates/observability/src/db.rs b/crates/observability/src/db.rs index f967c65c1bb..0182498a371 100644 --- a/crates/observability/src/db.rs +++ b/crates/observability/src/db.rs @@ -14,6 +14,7 @@ //! applies here: there is one database, no tenants, no replica, no Redis-backed storage scheme and //! no encrypted columns. +pub mod alert_manager; pub mod alerts_info; use std::{sync::Arc, time::Duration}; @@ -31,7 +32,13 @@ use crate::{errors::ConfigurationError, settings::Database}; /// /// Held by [`crate::state::AppState`] as `Arc`: one store shared by every /// worker, so a store never needs to be cloneable itself. -pub trait StorageInterface: Send + Sync + alerts_info::AlertsInfoInterface {} +pub trait StorageInterface: + Send + + Sync + + alerts_info::AlertsInfoInterface + + alert_manager::merchant_thresholds::MerchantThresholdsInterface +{ +} impl StorageInterface for Store {} diff --git a/crates/observability/src/db/alert_manager.rs b/crates/observability/src/db/alert_manager.rs new file mode 100644 index 00000000000..bb0dfb1d610 --- /dev/null +++ b/crates/observability/src/db/alert_manager.rs @@ -0,0 +1 @@ +pub mod merchant_thresholds; diff --git a/crates/observability/src/db/alert_manager/merchant_thresholds.rs b/crates/observability/src/db/alert_manager/merchant_thresholds.rs new file mode 100644 index 00000000000..057b49701ff --- /dev/null +++ b/crates/observability/src/db/alert_manager/merchant_thresholds.rs @@ -0,0 +1,124 @@ +//! Storage operations on `merchant_thresholds`. + +use diesel_models::{observability::alert_manager::merchant_thresholds as storage, StorageResult}; + +use crate::{db::Store, domain_models::alert_manager::merchant_thresholds as domain_models}; + +/// Storage operations on per-merchant threshold overrides. +#[async_trait::async_trait] +pub trait MerchantThresholdsInterface { + /// List rows matching the given filter. + async fn list_merchant_thresholds_by_filter( + &self, + filter: domain_models::MerchantThresholdsFilter, + ) -> StorageResult>; + + /// Insert a new override, or, on a conflict of `(name, product, merchant_id, profile_id, + /// is_enabled, author)`, set only the non-key columns the caller actually sent. + async fn upsert_merchant_threshold( + &self, + new: domain_models::MerchantThresholdsNew, + ) -> StorageResult>; + + /// Set or clear columns on every row matching the given keys. + async fn update_merchant_thresholds_by_filter( + &self, + filter: domain_models::MerchantThresholdsKeyFilter, + update: domain_models::MerchantThresholdsUpdate, + ) -> StorageResult>; + + /// Delete one override by id. + async fn delete_merchant_threshold_by_id( + &self, + id: String, + ) -> StorageResult; + + /// Delete every row matching the given keys. + async fn delete_merchant_thresholds_by_filter( + &self, + filter: domain_models::MerchantThresholdsKeyFilter, + ) -> StorageResult>; +} + +#[async_trait::async_trait] +impl MerchantThresholdsInterface for Store { + async fn list_merchant_thresholds_by_filter( + &self, + filter: domain_models::MerchantThresholdsFilter, + ) -> StorageResult> { + let connection = self.connection().await?; + + storage::MerchantThresholds::list_by_filter(&connection, filter.into()) + .await + .map(|rows| { + rows.into_iter() + .map(domain_models::MerchantThresholds::from) + .collect() + }) + } + + async fn upsert_merchant_threshold( + &self, + new: domain_models::MerchantThresholdsNew, + ) -> StorageResult> { + let connection = self.connection().await?; + let on_conflict = new.conflict_update().map(|update| update.into_storage().0); + + storage::MerchantThresholdsNew::from(new) + .upsert(&connection, on_conflict) + .await + .map(|rows| { + rows.into_iter() + .map(domain_models::MerchantThresholds::from) + .collect() + }) + } + + async fn update_merchant_thresholds_by_filter( + &self, + filter: domain_models::MerchantThresholdsKeyFilter, + update: domain_models::MerchantThresholdsUpdate, + ) -> StorageResult> { + let connection = self.connection().await?; + let (changeset, merge) = update.into_storage(); + + storage::MerchantThresholds::update_by_key_filter( + &connection, + filter.into(), + changeset, + merge, + ) + .await + .map(|rows| { + rows.into_iter() + .map(domain_models::MerchantThresholds::from) + .collect() + }) + } + + async fn delete_merchant_threshold_by_id( + &self, + id: String, + ) -> StorageResult { + let connection = self.connection().await?; + + storage::MerchantThresholds::delete_by_id(&connection, id) + .await + .map(domain_models::MerchantThresholds::from) + } + + async fn delete_merchant_thresholds_by_filter( + &self, + filter: domain_models::MerchantThresholdsKeyFilter, + ) -> StorageResult> { + let connection = self.connection().await?; + + storage::MerchantThresholds::delete_by_key_filter(&connection, filter.into()) + .await + .map(|rows| { + rows.into_iter() + .map(domain_models::MerchantThresholds::from) + .collect() + }) + } +} diff --git a/crates/observability/src/domain_models.rs b/crates/observability/src/domain_models.rs index 0600917f655..7bfab8adfef 100644 --- a/crates/observability/src/domain_models.rs +++ b/crates/observability/src/domain_models.rs @@ -7,6 +7,7 @@ //! //! Distinct from [`crate::domain`], which holds the traits that say what delivering an alert *is*. +pub mod alert_manager; pub mod alerts_info; use error_stack::{report, ResultExt}; diff --git a/crates/observability/src/domain_models/alert_manager.rs b/crates/observability/src/domain_models/alert_manager.rs new file mode 100644 index 00000000000..bb0dfb1d610 --- /dev/null +++ b/crates/observability/src/domain_models/alert_manager.rs @@ -0,0 +1 @@ +pub mod merchant_thresholds; diff --git a/crates/observability/src/domain_models/alert_manager/merchant_thresholds.rs b/crates/observability/src/domain_models/alert_manager/merchant_thresholds.rs new file mode 100644 index 00000000000..353cad33f81 --- /dev/null +++ b/crates/observability/src/domain_models/alert_manager/merchant_thresholds.rs @@ -0,0 +1,768 @@ +//! Per-merchant threshold overrides: what an alert's thresholds should be for one merchant (and, +//! optionally, one profile) rather than the alert's own definition. A job coalesces a row's value +//! with the definition's, favouring the row where it is not `NULL`. + +use api_models::observability::alert_manager::merchant_thresholds as api; +use common_utils::generate_time_ordered_id; +use diesel_models::observability::alert_manager::merchant_thresholds as storage; +use error_stack::{report, ResultExt}; +use serde_json::Value; +use time::PrimitiveDateTime; + +use crate::domain_models::{optional_text, SHORT_TEXT_MAX_CHARS}; +use crate::errors::{ObservabilityApiResult, ObservabilityError}; + +/// An override that has not been stored yet. +#[derive(Clone, Debug)] +pub struct MerchantThresholdsNew { + pub id: String, + pub name: String, + pub product: String, + pub merchant_id: String, + pub profile_id: String, + pub thresholds_min_volume: Option, + pub thresholds_min_impacted_volume: Option, + pub thresholds_tolerance: Option, + pub thresholds_diff_threshold: Option, + pub thresholds_merchant_impact: Option, + pub thresholds_alert_period: Option, + pub thresholds_min_observations: Option, + pub thresholds_min_history_volume: Option, + pub thresholds_filter_percentile: Option, + pub thresholds_current_min_volume: Option, + pub metadata: Option, + pub author: Option, + pub is_enabled: Option, + pub last_updated_at: Option, +} + +/// A stored override. +#[derive(Clone, Debug)] +pub struct MerchantThresholds { + pub id: String, + pub name: Option, + pub product: Option, + pub merchant_id: Option, + pub profile_id: String, + pub thresholds_min_volume: Option, + pub thresholds_min_impacted_volume: Option, + pub thresholds_tolerance: Option, + pub thresholds_diff_threshold: Option, + pub thresholds_merchant_impact: Option, + pub thresholds_alert_period: Option, + pub thresholds_min_observations: Option, + pub thresholds_min_history_volume: Option, + pub thresholds_filter_percentile: Option, + pub thresholds_current_min_volume: Option, + pub metadata: Option, + pub author: Option, + pub is_enabled: Option, + pub last_updated_at: Option, +} + +/// The filters `list_merchant_thresholds` accepts. +#[derive(Clone, Debug, Default)] +pub struct MerchantThresholdsFilter { + pub ids: Option>, + pub names: Option>, + pub products: Option>, + pub merchant_ids: Option>, + pub profile_ids: Option>, + pub authors: Option>, + pub is_enabled: Option, + pub metadata: Vec<(String, Vec)>, + pub updated_from: Option, + pub updated_to: Option, +} + +/// The keys an update or a bulk delete selects rows by. Only the keys sent filter; at least one +/// must be sent. +#[derive(Clone, Debug, Default)] +pub struct MerchantThresholdsKeyFilter { + pub name: Option, + pub product: Option, + pub merchant_id: Option, + pub profile_id: Option, +} + +/// What a write does to the `metadata` column. +#[derive(Clone, Debug, Default)] +pub enum MerchantThresholdsMetadataChange { + /// Leave the column as it is. + #[default] + Keep, + /// Set it to `NULL`. + Clear, + /// Overwrite it with the given value. + Replace(Value), + /// `metadata || value`, so a `NULL` column stays `NULL`. + Merge(Value), +} + +/// A validated change to apply to whatever rows a [`MerchantThresholdsKeyFilter`] selects. +#[derive(Clone, Debug, Default)] +pub struct MerchantThresholdsUpdate { + pub thresholds_min_volume: Option>, + pub thresholds_min_impacted_volume: Option>, + pub thresholds_tolerance: Option>, + pub thresholds_diff_threshold: Option>, + pub thresholds_merchant_impact: Option>, + pub thresholds_alert_period: Option>, + pub thresholds_min_observations: Option>, + pub thresholds_min_history_volume: Option>, + pub thresholds_filter_percentile: Option>, + pub thresholds_current_min_volume: Option>, + pub metadata: MerchantThresholdsMetadataChange, + pub author: Option>, + pub is_enabled: Option>, + pub last_updated_at: Option>, +} + +/// A validated `update_merchant_thresholds` request: the keys to select rows, and the change to +/// apply to them. +#[derive(Clone, Debug)] +pub struct MerchantThresholdsBulkUpdate { + pub filter: MerchantThresholdsKeyFilter, + pub update: MerchantThresholdsUpdate, +} + +impl MerchantThresholdsNew { + /// Reject what the table would refuse, so a bad value is a `400` rather than a database error. + /// `name`, `product`, `merchant_id` and `profile_id` may be empty, as r-apps allows, but not + /// longer than the column. + fn validate(&self) -> ObservabilityApiResult<()> { + optional_text("name", Some(&self.name), SHORT_TEXT_MAX_CHARS)?; + optional_text("product", Some(&self.product), SHORT_TEXT_MAX_CHARS)?; + optional_text("merchant_id", Some(&self.merchant_id), SHORT_TEXT_MAX_CHARS)?; + optional_text("profile_id", Some(&self.profile_id), SHORT_TEXT_MAX_CHARS)?; + optional_text("author", self.author.as_deref(), SHORT_TEXT_MAX_CHARS)?; + + Ok(()) + } + + /// What an upsert sets on a conflict: every threshold, `metadata` and `last_updated_at` that + /// was actually sent. `author` and `is_enabled` are never here — they are part of the conflict + /// key, not something a conflicting row can change. `None` when nothing was sent, so the + /// caller can turn the write into `DO NOTHING`. + pub fn conflict_update(&self) -> Option { + let mut update = MerchantThresholdsUpdate::default(); + let mut changed = false; + + macro_rules! copy_threshold { + ($field:ident) => { + if let Some(value) = self.$field { + update.$field = Some(Some(value)); + changed = true; + } + }; + } + + copy_threshold!(thresholds_min_volume); + copy_threshold!(thresholds_min_impacted_volume); + copy_threshold!(thresholds_tolerance); + copy_threshold!(thresholds_diff_threshold); + copy_threshold!(thresholds_merchant_impact); + copy_threshold!(thresholds_alert_period); + copy_threshold!(thresholds_min_observations); + copy_threshold!(thresholds_min_history_volume); + copy_threshold!(thresholds_filter_percentile); + copy_threshold!(thresholds_current_min_volume); + + if let Some(metadata) = self.metadata.clone() { + update.metadata = MerchantThresholdsMetadataChange::Replace(metadata); + changed = true; + } + + if let Some(last_updated_at) = self.last_updated_at { + update.last_updated_at = Some(Some(last_updated_at)); + changed = true; + } + + changed.then_some(update) + } +} + +impl MerchantThresholdsUpdate { + /// Reject a write that would touch nothing, and a `metadata` merge that is not an object — + /// `metadata || value` needs an object on both sides. + fn validate(&self) -> ObservabilityApiResult<()> { + let nothing_to_change = self.thresholds_min_volume.is_none() + && self.thresholds_min_impacted_volume.is_none() + && self.thresholds_tolerance.is_none() + && self.thresholds_diff_threshold.is_none() + && self.thresholds_merchant_impact.is_none() + && self.thresholds_alert_period.is_none() + && self.thresholds_min_observations.is_none() + && self.thresholds_min_history_volume.is_none() + && self.thresholds_filter_percentile.is_none() + && self.thresholds_current_min_volume.is_none() + && self.author.is_none() + && self.is_enabled.is_none() + && self.last_updated_at.is_none() + && matches!(self.metadata, MerchantThresholdsMetadataChange::Keep); + + if nothing_to_change { + Err(report!(ObservabilityError::InvalidRequest)) + .attach_printable("nothing to update")?; + } + + if let MerchantThresholdsMetadataChange::Merge(value) = &self.metadata { + if !value.is_object() { + Err(report!(ObservabilityError::InvalidRequest)) + .attach_printable("metadata must be an object")?; + } + } + + Ok(()) + } + + /// Split into what an `AsChangeset` can express directly and the `metadata || value` merge, + /// which cannot: an `AsChangeset` column can only be skipped or set, never assigned from an + /// expression over its own current value. + pub fn into_storage(self) -> (storage::MerchantThresholdsUpdate, Option) { + let (metadata, merge) = match self.metadata { + MerchantThresholdsMetadataChange::Keep => (None, None), + MerchantThresholdsMetadataChange::Clear => (Some(None), None), + MerchantThresholdsMetadataChange::Replace(value) => (Some(Some(value)), None), + MerchantThresholdsMetadataChange::Merge(value) => (None, Some(value)), + }; + + ( + storage::MerchantThresholdsUpdate { + thresholds_min_volume: self.thresholds_min_volume, + thresholds_min_impacted_volume: self.thresholds_min_impacted_volume, + thresholds_tolerance: self.thresholds_tolerance, + thresholds_diff_threshold: self.thresholds_diff_threshold, + thresholds_merchant_impact: self.thresholds_merchant_impact, + thresholds_alert_period: self.thresholds_alert_period, + thresholds_min_observations: self.thresholds_min_observations, + thresholds_min_history_volume: self.thresholds_min_history_volume, + thresholds_filter_percentile: self.thresholds_filter_percentile, + thresholds_current_min_volume: self.thresholds_current_min_volume, + metadata, + author: self.author, + is_enabled: self.is_enabled, + last_updated_at: self.last_updated_at, + }, + merge, + ) + } +} + +impl TryFrom for MerchantThresholdsNew { + type Error = error_stack::Report; + + fn try_from(request: api::MerchantThresholdsUpsertRequest) -> Result { + let new = Self { + id: generate_time_ordered_id("merchant_threshold"), + name: request.name, + product: request.product, + merchant_id: request.merchant_id, + profile_id: request.profile_id, + thresholds_min_volume: request.thresholds_min_volume, + thresholds_min_impacted_volume: request.thresholds_min_impacted_volume, + thresholds_tolerance: request.thresholds_tolerance, + thresholds_diff_threshold: request.thresholds_diff_threshold, + thresholds_merchant_impact: request.thresholds_merchant_impact, + thresholds_alert_period: request.thresholds_alert_period, + thresholds_min_observations: request.thresholds_min_observations, + thresholds_min_history_volume: request.thresholds_min_history_volume, + thresholds_filter_percentile: request.thresholds_filter_percentile, + thresholds_current_min_volume: request.thresholds_current_min_volume, + metadata: request.metadata, + author: request.author, + is_enabled: request.is_enabled, + last_updated_at: request.last_updated_at, + }; + + new.validate()?; + + Ok(new) + } +} + +impl TryFrom for MerchantThresholdsBulkUpdate { + type Error = error_stack::Report; + + fn try_from(request: api::MerchantThresholdsUpdateRequest) -> Result { + let filter = MerchantThresholdsKeyFilter { + name: request.name, + product: request.product, + merchant_id: request.merchant_id, + profile_id: request.profile_id, + }; + + if filter.name.is_none() + && filter.product.is_none() + && filter.merchant_id.is_none() + && filter.profile_id.is_none() + { + Err(report!(ObservabilityError::InvalidRequest)) + .attach_printable("name, product, merchant_id or profile_id is required")?; + } + + let metadata = match request.metadata { + None => MerchantThresholdsMetadataChange::Keep, + Some(None) => MerchantThresholdsMetadataChange::Clear, + Some(Some(value)) => MerchantThresholdsMetadataChange::Merge(value), + }; + + let update = MerchantThresholdsUpdate { + thresholds_min_volume: request.thresholds_min_volume, + thresholds_min_impacted_volume: request.thresholds_min_impacted_volume, + thresholds_tolerance: request.thresholds_tolerance, + thresholds_diff_threshold: request.thresholds_diff_threshold, + thresholds_merchant_impact: request.thresholds_merchant_impact, + thresholds_alert_period: request.thresholds_alert_period, + thresholds_min_observations: request.thresholds_min_observations, + thresholds_min_history_volume: request.thresholds_min_history_volume, + thresholds_filter_percentile: request.thresholds_filter_percentile, + thresholds_current_min_volume: request.thresholds_current_min_volume, + metadata, + author: request.author, + is_enabled: request.is_enabled, + last_updated_at: request.last_updated_at, + }; + + update.validate()?; + + Ok(Self { filter, update }) + } +} + +/// `One(v)` is the one filter value; an empty `Many` means no filter on this field. +fn text_values(filter: api::MerchantThresholdsTextFilter) -> Option> { + match filter { + api::MerchantThresholdsTextFilter::One(value) => Some(vec![value]), + api::MerchantThresholdsTextFilter::Many(values) if values.is_empty() => None, + api::MerchantThresholdsTextFilter::Many(values) => Some(values), + } +} + +impl TryFrom for MerchantThresholdsFilter { + type Error = error_stack::Report; + + fn try_from(request: api::MerchantThresholdsListRequest) -> Result { + let ids = request + .id + .and_then(text_values) + .map(|ids| ids.iter().map(|id| parse_id(id)).collect::>()) + .transpose()?; + + let metadata = request + .metadata + .into_iter() + .flatten() + .filter_map(|(key, filter)| text_values(filter).map(|values| (key, values))) + .collect(); + + let (updated_from, updated_to) = match request.last_updated_at { + Some(api::MerchantThresholdsTimeFilter::Range { start, end }) => (start, end), + Some(api::MerchantThresholdsTimeFilter::From(at)) => (Some(at), None), + None => (None, None), + }; + + Ok(Self { + ids, + names: request.name.and_then(text_values), + products: request.product.and_then(text_values), + merchant_ids: request.merchant_id.and_then(text_values), + profile_ids: request.profile_id.and_then(text_values), + authors: request.author.and_then(text_values), + is_enabled: request.is_enabled, + metadata, + updated_from, + updated_to, + }) + } +} + +impl From for MerchantThresholdsKeyFilter { + fn from(request: api::MerchantThresholdsDeleteByFilterRequest) -> Self { + Self { + name: Some(request.name), + product: Some(request.product), + merchant_id: request.merchant_id, + profile_id: request.profile_id, + } + } +} + +impl From for storage::MerchantThresholdsNew { + fn from(new: MerchantThresholdsNew) -> Self { + Self { + id: new.id, + name: new.name, + product: new.product, + merchant_id: new.merchant_id, + profile_id: new.profile_id, + thresholds_min_volume: new.thresholds_min_volume, + thresholds_min_impacted_volume: new.thresholds_min_impacted_volume, + thresholds_tolerance: new.thresholds_tolerance, + thresholds_diff_threshold: new.thresholds_diff_threshold, + thresholds_merchant_impact: new.thresholds_merchant_impact, + thresholds_alert_period: new.thresholds_alert_period, + thresholds_min_observations: new.thresholds_min_observations, + thresholds_min_history_volume: new.thresholds_min_history_volume, + thresholds_filter_percentile: new.thresholds_filter_percentile, + thresholds_current_min_volume: new.thresholds_current_min_volume, + metadata: new.metadata, + author: new.author, + is_enabled: new.is_enabled, + last_updated_at: new.last_updated_at, + } + } +} + +impl From for MerchantThresholds { + fn from(row: storage::MerchantThresholds) -> Self { + Self { + id: row.id, + name: row.name, + product: row.product, + merchant_id: row.merchant_id, + profile_id: row.profile_id, + thresholds_min_volume: row.thresholds_min_volume, + thresholds_min_impacted_volume: row.thresholds_min_impacted_volume, + thresholds_tolerance: row.thresholds_tolerance, + thresholds_diff_threshold: row.thresholds_diff_threshold, + thresholds_merchant_impact: row.thresholds_merchant_impact, + thresholds_alert_period: row.thresholds_alert_period, + thresholds_min_observations: row.thresholds_min_observations, + thresholds_min_history_volume: row.thresholds_min_history_volume, + thresholds_filter_percentile: row.thresholds_filter_percentile, + thresholds_current_min_volume: row.thresholds_current_min_volume, + metadata: row.metadata, + author: row.author, + is_enabled: row.is_enabled, + last_updated_at: row.last_updated_at, + } + } +} + +impl From for storage::MerchantThresholdsFilter { + fn from(filter: MerchantThresholdsFilter) -> Self { + Self { + ids: filter.ids, + names: filter.names, + products: filter.products, + merchant_ids: filter.merchant_ids, + profile_ids: filter.profile_ids, + authors: filter.authors, + is_enabled: filter.is_enabled, + metadata: filter.metadata, + updated_from: filter.updated_from, + updated_to: filter.updated_to, + } + } +} + +impl From for storage::MerchantThresholdsKeyFilter { + fn from(filter: MerchantThresholdsKeyFilter) -> Self { + Self { + name: filter.name, + product: filter.product, + merchant_id: filter.merchant_id, + profile_id: filter.profile_id, + } + } +} + +impl From for api::MerchantThresholdsResponse { + fn from(row: MerchantThresholds) -> Self { + Self { + id: row.id.to_string(), + name: row.name, + product: row.product, + merchant_id: row.merchant_id, + profile_id: row.profile_id, + thresholds_min_volume: row.thresholds_min_volume, + thresholds_min_impacted_volume: row.thresholds_min_impacted_volume, + thresholds_tolerance: row.thresholds_tolerance, + thresholds_diff_threshold: row.thresholds_diff_threshold, + thresholds_merchant_impact: row.thresholds_merchant_impact, + thresholds_alert_period: row.thresholds_alert_period, + thresholds_min_observations: row.thresholds_min_observations, + thresholds_min_history_volume: row.thresholds_min_history_volume, + thresholds_filter_percentile: row.thresholds_filter_percentile, + thresholds_current_min_volume: row.thresholds_current_min_volume, + metadata: row.metadata, + author: row.author, + is_enabled: row.is_enabled, + last_updated_at: row.last_updated_at, + } + } +} + +/// The body of every route that returns more than one row. A free function, since the orphan rule +/// blocks `impl From> for api::MerchantThresholdsListResponse` here. +pub fn list_response(rows: Vec) -> api::MerchantThresholdsListResponse { + let data: Vec = rows + .into_iter() + .map(api::MerchantThresholdsResponse::from) + .collect(); + + api::MerchantThresholdsListResponse { + count: data.len(), + data, + } +} + +/// Parse a path `id` or an `id` filter value into what the table's primary key actually is, so an +/// Reject a blank ID before it reaches the database. +pub fn parse_id(id: &str) -> ObservabilityApiResult { + if id.trim().is_empty() { + Err(report!(ObservabilityError::InvalidRequest)).attach_printable("id must not be empty")? + } + + Ok(id.to_owned()) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used, clippy::indexing_slicing)] +mod tests { + use time::macros::datetime; + + use super::*; + + fn upsert_request() -> api::MerchantThresholdsUpsertRequest { + api::MerchantThresholdsUpsertRequest { + name: "Volume Drop".to_owned(), + product: "payments".to_owned(), + merchant_id: "acme_store".to_owned(), + profile_id: String::new(), + thresholds_min_volume: None, + thresholds_min_impacted_volume: None, + thresholds_tolerance: None, + thresholds_diff_threshold: None, + thresholds_merchant_impact: None, + thresholds_alert_period: None, + thresholds_min_observations: None, + thresholds_min_history_volume: None, + thresholds_filter_percentile: None, + thresholds_current_min_volume: None, + metadata: None, + author: None, + is_enabled: None, + last_updated_at: None, + } + } + + #[test] + fn an_upsert_leaves_absent_fields_to_the_database() { + let new = MerchantThresholdsNew::try_from(upsert_request()).unwrap(); + + assert!(new.author.is_none()); + assert!(new.is_enabled.is_none()); + assert!(new.thresholds_min_volume.is_none()); + } + + #[test] + fn an_upsert_with_only_keys_does_nothing_on_conflict() { + let new = MerchantThresholdsNew::try_from(upsert_request()).unwrap(); + + assert!(new.conflict_update().is_none()); + } + + #[test] + fn an_upsert_conflict_sets_only_sent_non_key_fields() { + let new = MerchantThresholdsNew::try_from(api::MerchantThresholdsUpsertRequest { + thresholds_min_volume: Some(50.0), + metadata: Some(serde_json::json!({"k": "v"})), + ..upsert_request() + }) + .unwrap(); + + let update = new.conflict_update().unwrap(); + + assert_eq!(update.thresholds_min_volume, Some(Some(50.0))); + assert!(matches!( + update.metadata, + MerchantThresholdsMetadataChange::Replace(_) + )); + assert!(update.author.is_none()); + assert!(update.is_enabled.is_none()); + } + + #[test] + fn a_threshold_is_stored_as_the_value_sent() { + let new = MerchantThresholdsNew::try_from(api::MerchantThresholdsUpsertRequest { + thresholds_tolerance: Some(0.1), + thresholds_min_volume: Some(100.0), + ..upsert_request() + }) + .unwrap(); + + assert_eq!(new.thresholds_tolerance, Some(0.1)); + assert_eq!(new.thresholds_min_volume, Some(100.0)); + } + + fn update_request() -> api::MerchantThresholdsUpdateRequest { + api::MerchantThresholdsUpdateRequest { + merchant_id: Some("acme_store".to_owned()), + ..Default::default() + } + } + + #[test] + fn an_update_needs_a_key() { + let error = MerchantThresholdsBulkUpdate::try_from(api::MerchantThresholdsUpdateRequest { + merchant_id: None, + thresholds_min_volume: Some(Some(1.0)), + ..Default::default() + }) + .unwrap_err(); + + assert!(matches!( + error.current_context(), + ObservabilityError::InvalidRequest + )); + } + + #[test] + fn an_update_needs_a_change() { + let error = MerchantThresholdsBulkUpdate::try_from(update_request()).unwrap_err(); + + assert!(matches!( + error.current_context(), + ObservabilityError::InvalidRequest + )); + } + + #[test] + fn an_update_null_clears_and_absent_keeps() { + let bulk_update = + MerchantThresholdsBulkUpdate::try_from(api::MerchantThresholdsUpdateRequest { + thresholds_min_volume: Some(None), + ..update_request() + }) + .unwrap(); + + let (changeset, merge) = bulk_update.update.into_storage(); + + assert_eq!(changeset.thresholds_min_volume, Some(None)); + assert!(changeset.thresholds_tolerance.is_none()); + assert!(merge.is_none()); + } + + #[test] + fn an_update_metadata_object_merges_null_clears_other_refused() { + let merged = MerchantThresholdsBulkUpdate::try_from(api::MerchantThresholdsUpdateRequest { + metadata: Some(Some(serde_json::json!({"a": 1}))), + ..update_request() + }) + .unwrap(); + let (_, merge) = merged.update.into_storage(); + assert_eq!(merge, Some(serde_json::json!({"a": 1}))); + + let cleared = + MerchantThresholdsBulkUpdate::try_from(api::MerchantThresholdsUpdateRequest { + metadata: Some(None), + ..update_request() + }) + .unwrap(); + let (changeset, merge) = cleared.update.into_storage(); + assert_eq!(changeset.metadata, Some(None)); + assert!(merge.is_none()); + + let error = MerchantThresholdsBulkUpdate::try_from(api::MerchantThresholdsUpdateRequest { + metadata: Some(Some(serde_json::json!(["not", "an", "object"]))), + ..update_request() + }) + .unwrap_err(); + assert!(matches!( + error.current_context(), + ObservabilityError::InvalidRequest + )); + } + + #[test] + fn a_stored_row_reaches_the_response_unchanged() { + let row = MerchantThresholds { + id: "01JTESTID".to_owned(), + name: Some("Volume Drop".to_owned()), + product: Some("payments".to_owned()), + merchant_id: Some("acme_store".to_owned()), + profile_id: String::new(), + thresholds_min_volume: Some(50.0), + thresholds_min_impacted_volume: None, + thresholds_tolerance: None, + thresholds_diff_threshold: None, + thresholds_merchant_impact: None, + thresholds_alert_period: None, + thresholds_min_observations: None, + thresholds_min_history_volume: None, + thresholds_filter_percentile: None, + thresholds_current_min_volume: None, + metadata: None, + author: Some("reliability_team".to_owned()), + is_enabled: Some(false), + last_updated_at: Some(datetime!(2026-09-15 09:42:10.512)), + }; + + let response = api::MerchantThresholdsResponse::from(row); + let serialized = serde_json::to_value(&response).unwrap(); + + assert_eq!(serialized["id"], "01JTESTID"); + assert_eq!(serialized["thresholds_min_volume"], 50.0); + assert!(serialized["last_updated_at"] + .as_str() + .is_some_and(|value| value.contains('T'))); + } + + #[test] + fn a_text_filter_takes_one_value_or_a_list() { + assert_eq!( + text_values(api::MerchantThresholdsTextFilter::One("a".to_owned())), + Some(vec!["a".to_owned()]) + ); + assert_eq!( + text_values(api::MerchantThresholdsTextFilter::Many(vec![ + "a".to_owned(), + "b".to_owned() + ])), + Some(vec!["a".to_owned(), "b".to_owned()]) + ); + assert_eq!( + text_values(api::MerchantThresholdsTextFilter::Many(Vec::new())), + None + ); + } + + #[test] + fn a_time_filter_takes_a_range_or_a_bare_timestamp() { + let filter: api::MerchantThresholdsListRequest = + serde_json::from_value(serde_json::json!({ + "last_updated_at": "2026-09-15T00:00:00.000Z" + })) + .unwrap(); + let parsed = MerchantThresholdsFilter::try_from(filter).unwrap(); + assert!(parsed.updated_from.is_some()); + assert!(parsed.updated_to.is_none()); + + let filter: api::MerchantThresholdsListRequest = + serde_json::from_value(serde_json::json!({ + "last_updated_at": { + "start": "2026-09-15T00:00:00.000Z", + "end": "2026-09-16T00:00:00.000Z" + } + })) + .unwrap(); + let parsed = MerchantThresholdsFilter::try_from(filter).unwrap(); + assert!(parsed.updated_from.is_some()); + assert!(parsed.updated_to.is_some()); + } + + #[test] + fn a_blank_id_is_refused() { + assert!(parse_id("").is_err()); + assert!(parse_id("01JTESTID").is_ok()); + + let filter: api::MerchantThresholdsListRequest = + serde_json::from_value(serde_json::json!({ + "id": "01JTESTID" + })) + .unwrap(); + assert!(MerchantThresholdsFilter::try_from(filter).is_ok()); + } +} diff --git a/crates/observability/src/routes.rs b/crates/observability/src/routes.rs index 71bc70aebb8..a19a30764ad 100644 --- a/crates/observability/src/routes.rs +++ b/crates/observability/src/routes.rs @@ -1,6 +1,7 @@ //! HTTP surface, laid out as the router lays its own out: [`app`] holds the route tree, and one //! module per area holds the handlers. +pub mod alert_manager; pub mod alerts_info; pub mod app; pub mod cloudwatch; diff --git a/crates/observability/src/routes/alert_manager.rs b/crates/observability/src/routes/alert_manager.rs new file mode 100644 index 00000000000..bb0dfb1d610 --- /dev/null +++ b/crates/observability/src/routes/alert_manager.rs @@ -0,0 +1 @@ +pub mod merchant_thresholds; diff --git a/crates/observability/src/routes/alert_manager/merchant_thresholds.rs b/crates/observability/src/routes/alert_manager/merchant_thresholds.rs new file mode 100644 index 00000000000..76d340617be --- /dev/null +++ b/crates/observability/src/routes/alert_manager/merchant_thresholds.rs @@ -0,0 +1,93 @@ +//! Handlers for the per-merchant threshold override routes. The route tree that mounts them is in +//! [`crate::routes::app`]. + +use actix_web::{web, HttpRequest, HttpResponse}; +use api_models::observability::alert_manager::merchant_thresholds::{ + MerchantThresholdsDeleteByFilterRequest, MerchantThresholdsDeleteRequest, + MerchantThresholdsListRequest, MerchantThresholdsUpdateRequest, + MerchantThresholdsUpsertRequest, +}; + +use crate::{auth, core, services, state::AppState}; + +/// `POST /alerts/alerts_manager/merchant_thresholds`. +pub async fn upsert( + state: web::Data, + request: HttpRequest, + payload: web::Json, +) -> HttpResponse { + services::server_wrap( + state.get_ref().clone(), + &request, + payload.into_inner(), + core::alert_manager::merchant_thresholds::upsert_merchant_threshold, + &auth::InternalApiKeyAuth, + ) + .await +} + +/// `POST /alerts/alerts_manager/merchant_thresholds/list`. +pub async fn list( + state: web::Data, + request: HttpRequest, + payload: web::Json, +) -> HttpResponse { + services::server_wrap( + state.get_ref().clone(), + &request, + payload.into_inner(), + core::alert_manager::merchant_thresholds::list_merchant_thresholds, + &auth::InternalApiKeyAuth, + ) + .await +} + +/// `POST /alerts/alerts_manager/merchant_thresholds/update`. +pub async fn update( + state: web::Data, + request: HttpRequest, + payload: web::Json, +) -> HttpResponse { + services::server_wrap( + state.get_ref().clone(), + &request, + payload.into_inner(), + core::alert_manager::merchant_thresholds::update_merchant_thresholds, + &auth::InternalApiKeyAuth, + ) + .await +} + +/// `POST /alerts/alerts_manager/merchant_thresholds/delete`. +pub async fn delete_by_filter( + state: web::Data, + request: HttpRequest, + payload: web::Json, +) -> HttpResponse { + services::server_wrap( + state.get_ref().clone(), + &request, + payload.into_inner(), + core::alert_manager::merchant_thresholds::delete_merchant_thresholds_by_filter, + &auth::InternalApiKeyAuth, + ) + .await +} + +/// `DELETE /alerts/alerts_manager/merchant_thresholds/{id}`. +pub async fn delete( + state: web::Data, + request: HttpRequest, + id: web::Path, +) -> HttpResponse { + services::server_wrap( + state.get_ref().clone(), + &request, + MerchantThresholdsDeleteRequest { + id: id.into_inner(), + }, + core::alert_manager::merchant_thresholds::delete_merchant_threshold, + &auth::InternalApiKeyAuth, + ) + .await +} diff --git a/crates/observability/src/routes/app.rs b/crates/observability/src/routes/app.rs index a07c18520a6..2834d464eff 100644 --- a/crates/observability/src/routes/app.rs +++ b/crates/observability/src/routes/app.rs @@ -14,7 +14,7 @@ use actix_web::{web, Scope}; use crate::{ errors::types::{ApiError, ApiErrorResponse}, logger, - routes::{alerts_info, cloudwatch, health_check, notify}, + routes::{alert_manager, alerts_info, cloudwatch, health_check, notify}, state::AppState, }; @@ -59,7 +59,26 @@ impl Alerts { ) .service( web::scope("/alerts_manager") - .service(web::resource("/info").route(web::post().to(alerts_info::create))), + .service(web::resource("/info").route(web::post().to(alerts_info::create))) + .service( + web::resource("/merchant_thresholds") + .route(web::post().to(alert_manager::merchant_thresholds::upsert)), + ) + .service( + web::resource("/merchant_thresholds/list") + .route(web::post().to(alert_manager::merchant_thresholds::list)), + ) + .service( + web::resource("/merchant_thresholds/update") + .route(web::post().to(alert_manager::merchant_thresholds::update)), + ) + .service(web::resource("/merchant_thresholds/delete").route( + web::post().to(alert_manager::merchant_thresholds::delete_by_filter), + )) + .service( + web::resource("/merchant_thresholds/{id}") + .route(web::delete().to(alert_manager::merchant_thresholds::delete)), + ), ) } } diff --git a/crates/observability/tests/merchant_thresholds.rs b/crates/observability/tests/merchant_thresholds.rs new file mode 100644 index 00000000000..d5580b139b6 --- /dev/null +++ b/crates/observability/tests/merchant_thresholds.rs @@ -0,0 +1,182 @@ +//! The per-merchant threshold override routes, exercised through actix as a caller would reach +//! them. +//! +//! The store is built with no idle connections, so only what fails *before* the database is +//! reached can be tested here: the guard, and every `400` a bad request earns before a query is +//! ever issued. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::indexing_slicing)] + +use std::sync::Arc; + +use actix_web::{ + http::StatusCode, + test::{self, TestRequest}, + App, +}; +use observability::{ + auth::X_INTERNAL_API_KEY, db::Store, domain::notifier::Registry, routes::Alerts, + settings::Database, state::AppState, +}; +use serde_json::{json, Value}; + +const API_KEY: &str = "test_internal_key"; +const BASE: &str = "/alerts/alerts_manager/merchant_thresholds"; + +async fn state() -> AppState { + let conf = serde_json::from_value(json!({ + "auth": { "internal_api_key": API_KEY } + })) + .expect("the test configuration should deserialize"); + + AppState { + conf: Arc::new(conf), + chat: Arc::new(Registry::default()), + email: Arc::new(Registry::default()), + metrics: None, + store: Arc::new(lazy_store().await), + } +} + +/// A store that never connects. No idle connections are opened at build, and none of the failures +/// tested here reach the database, so these tests need no database. +async fn lazy_store() -> Store { + Store::new(&Database { + username: "unused".to_owned(), + host: "localhost".to_owned(), + dbname: "unused".to_owned(), + min_idle_pool_size: 0, + ..Default::default() + }) + .await + .expect("a pool with no idle connections builds without a database") +} + +async fn call(request: TestRequest) -> (StatusCode, Value) { + let app = test::init_service(App::new().service(Alerts::server(state().await))).await; + let response = test::call_service(&app, request.to_request()).await; + let status = response.status(); + let body = test::read_body(response).await; + + (status, serde_json::from_slice(&body).unwrap_or(Value::Null)) +} + +fn authed(request: TestRequest) -> TestRequest { + request.insert_header((X_INTERNAL_API_KEY, API_KEY)) +} + +fn post(uri: &str, body: Value) -> TestRequest { + authed(TestRequest::post().uri(uri)).set_json(body) +} + +fn delete(uri: &str) -> TestRequest { + authed(TestRequest::delete().uri(uri)) +} + +fn upsert_body() -> Value { + json!({ + "name": "Volume Drop", + "product": "payments", + "merchant_id": "acme_store", + "profile_id": "" + }) +} + +#[actix_web::test] +async fn every_route_is_behind_the_guard() { + for request in [ + TestRequest::post().uri(BASE).set_json(upsert_body()), + TestRequest::post() + .uri(&format!("{BASE}/list")) + .set_json(json!({})), + TestRequest::post() + .uri(&format!("{BASE}/update")) + .set_json(json!({ + "merchant_id": "acme_store", + "thresholds_min_volume": 50 + })), + TestRequest::post() + .uri(&format!("{BASE}/delete")) + .set_json(json!({ + "name": "Volume Drop", + "product": "payments" + })), + TestRequest::delete().uri(&format!("{BASE}/abc")), + ] { + let (status, body) = call(request).await; + + assert_eq!(status, StatusCode::UNAUTHORIZED); + assert_eq!(body["error"]["code"], "IR_01"); + } +} + +#[actix_web::test] +async fn an_upsert_without_merchant_id_is_refused() { + let mut body = upsert_body(); + body.as_object_mut().unwrap().remove("merchant_id"); + + let (status, body) = call(post(BASE, body)).await; + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], "IR_04"); +} + +#[actix_web::test] +async fn an_upsert_without_profile_id_is_refused() { + let mut body = upsert_body(); + body.as_object_mut().unwrap().remove("profile_id"); + + let (status, body) = call(post(BASE, body)).await; + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], "IR_04"); +} + +#[actix_web::test] +async fn an_unknown_field_is_refused_on_every_write() { + for uri in [ + BASE.to_owned(), + format!("{BASE}/update"), + format!("{BASE}/delete"), + ] { + let mut body = upsert_body(); + body.as_object_mut() + .unwrap() + .insert("unexpected".to_owned(), json!("x")); + + let (status, body) = call(post(&uri, body)).await; + + assert_eq!(status, StatusCode::BAD_REQUEST, "{uri} must refuse it"); + assert_eq!(body["error"]["code"], "IR_04"); + } +} + +#[actix_web::test] +async fn update_refuses_bad_input_before_the_database() { + let (status, body) = call(post(&format!("{BASE}/update"), json!({}))).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], "IR_04"); + + let (status, body) = call(post( + &format!("{BASE}/update"), + json!({ "merchant_id": "acme_store" }), + )) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], "IR_04"); +} + +#[actix_web::test] +async fn delete_refuses_blank_id_before_the_database() { + let (status, body) = call(post( + &format!("{BASE}/delete"), + json!({ "name": "Volume Drop" }), + )) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], "IR_04"); + + let (status, body) = call(delete(&format!("{BASE}/%20"))).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], "IR_04"); +}