diff --git a/src/common/meta/organization.rs b/src/common/meta/organization.rs index fe519c7edf..be63093e84 100644 --- a/src/common/meta/organization.rs +++ b/src/common/meta/organization.rs @@ -158,6 +158,10 @@ pub struct AllOrgListDetails { #[serde(default)] pub plan: i32, pub trial_expires_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub contract_end_date: Option, + #[serde(default)] + pub billing_provider: String, } #[derive(Serialize, ToSchema)] @@ -177,6 +181,20 @@ pub struct ExtendTrialPeriodRequest { pub new_end_date: i64, } +#[cfg(feature = "cloud")] +#[derive(Serialize, Deserialize, ToSchema)] +pub struct CreateExternalContractRequest { + pub org_id: String, + pub end_date: i64, +} + +#[cfg(feature = "cloud")] +#[derive(Serialize, Deserialize, ToSchema)] +pub struct ExtendExternalContractRequest { + pub org_id: String, + pub new_end_date: i64, +} + #[derive(Serialize, Deserialize, ToSchema)] pub struct OrgSummary { pub streams: StreamSummary, @@ -683,6 +701,8 @@ mod tests { org_type: "standard".to_string(), plan: Default::default(), trial_expires_at: None, + contract_end_date: None, + billing_provider: String::new(), }; assert_eq!(details.plan, 0); @@ -729,6 +749,8 @@ mod tests { org_type: "basic".to_string(), plan: 0, trial_expires_at: None, + contract_end_date: None, + billing_provider: String::new(), }; let details2 = AllOrgListDetails { @@ -740,6 +762,8 @@ mod tests { org_type: "premium".to_string(), plan: 1, trial_expires_at: Some(1641081600), + contract_end_date: None, + billing_provider: String::new(), }; let response = AllOrganizationResponse { @@ -763,6 +787,79 @@ mod tests { assert_eq!(request.new_end_date, 1641081600); } + #[cfg(feature = "cloud")] + #[test] + fn test_create_external_contract_request_roundtrip() { + let json = r#"{"org_id":"org-ext","end_date":1893456000000000}"#; + let req: CreateExternalContractRequest = serde_json::from_str(json).unwrap(); + assert_eq!(req.org_id, "org-ext"); + assert_eq!(req.end_date, 1893456000000000); + + // Round-trip through serialize to ensure both directions agree on the + // wire format used by the /external_contract endpoint. + let serialized = serde_json::to_string(&req).unwrap(); + let req2: CreateExternalContractRequest = serde_json::from_str(&serialized).unwrap(); + assert_eq!(req2.org_id, req.org_id); + assert_eq!(req2.end_date, req.end_date); + } + + #[cfg(feature = "cloud")] + #[test] + fn test_extend_external_contract_request_roundtrip() { + let json = r#"{"org_id":"org-ext","new_end_date":1893456000000000}"#; + let req: ExtendExternalContractRequest = serde_json::from_str(json).unwrap(); + assert_eq!(req.org_id, "org-ext"); + assert_eq!(req.new_end_date, 1893456000000000); + + let serialized = serde_json::to_string(&req).unwrap(); + let req2: ExtendExternalContractRequest = serde_json::from_str(&serialized).unwrap(); + assert_eq!(req2.org_id, req.org_id); + assert_eq!(req2.new_end_date, req.new_end_date); + } + + #[test] + fn test_all_org_list_details_with_contract_fields_serialization() { + // External-contract orgs surface contract_end_date + billing_provider, + // and contract_end_date must be omitted when null. + let with_contract = AllOrgListDetails { + id: 7, + identifier: "org-ext".to_string(), + name: "External Org".to_string(), + created_at: 1640995200, + updated_at: 1640995260, + org_type: "custom".to_string(), + plan: 3, // ExternalContract + trial_expires_at: Some(1641081600), + contract_end_date: Some(1893456000000000), + billing_provider: "no_op".to_string(), + }; + + let v = serde_json::to_value(&with_contract).unwrap(); + assert_eq!(v["contract_end_date"], 1893456000000000_i64); + assert_eq!(v["billing_provider"], "no_op"); + assert_eq!(v["plan"], 3); + + let without_contract = AllOrgListDetails { + id: 8, + identifier: "org-free".to_string(), + name: "Free Org".to_string(), + created_at: 1640995200, + updated_at: 1640995260, + org_type: "custom".to_string(), + plan: 0, + trial_expires_at: Some(1641081600), + contract_end_date: None, + billing_provider: String::new(), + }; + + let v = serde_json::to_value(&without_contract).unwrap(); + assert!( + v.get("contract_end_date").is_none(), + "contract_end_date should be omitted when None: {v}" + ); + assert_eq!(v["billing_provider"], ""); + } + #[test] fn test_organization_setting_with_trial_expiry() { let setting = OrganizationSetting { diff --git a/src/common/utils/auth.rs b/src/common/utils/auth.rs index 3fb411b449..483445d823 100644 --- a/src/common/utils/auth.rs +++ b/src/common/utils/auth.rs @@ -390,8 +390,9 @@ where // this will take format of settings:{org_id} or pipelines:{org_id} etc let key = if path_columns[1].eq("invites") { "users" - } else if (path_columns[1].eq("rename") || path_columns[1].eq("extend_trial_period")) - && method.eq("PUT") + } else if ((path_columns[1].eq("rename") || path_columns[1].eq("extend_trial_period")) + && method.eq("PUT")) + || path_columns[1].eq("external_contract") { "organizations" } else { @@ -401,6 +402,7 @@ where // for organization api changes we need perms on _all_{org} let entity = match (key, path_columns[1]) { ("organizations", "extend_trial_period") => "_all__meta".to_string(), + ("organizations", "external_contract") => "_all__meta".to_string(), ("organizations", "organizations") => { // Special case: assume_service_account endpoint should check org:_all__meta if url_len == 3 && path_columns[2] == "assume_service_account" { diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 05efa0fbd7..efed2a6631 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -50,7 +50,7 @@ pub type RwAHashSet = tokio::sync::RwLock>; pub type RwBTreeMap = tokio::sync::RwLock>; // for DDL commands and migrations -pub const DB_SCHEMA_VERSION: u64 = 40; +pub const DB_SCHEMA_VERSION: u64 = 41; pub const DB_SCHEMA_KEY: &str = "/db_schema_version/"; // global version variables diff --git a/src/handler/http/models/billings.rs b/src/handler/http/models/billings.rs index eecc7f2a71..b79e53bdca 100644 --- a/src/handler/http/models/billings.rs +++ b/src/handler/http/models/billings.rs @@ -39,16 +39,21 @@ pub struct ListSubscriptionResponseBody { pub subscription_type: String, #[serde(skip_serializing_if = "Option::is_none")] pub customer_id: Option, - /// Billing provider: "stripe" or "aws" - pub provider: String, + /// Billing provider: "stripe", "aws", or "azure". Omitted for external-contract subs. + #[serde(skip_serializing_if = "Option::is_none")] + pub provider: Option, } impl From for ListSubscriptionResponseBody { fn from(value: cloud_billings::CustomerBilling) -> Self { + let provider = match value.provider { + cloud_billings::MeteringProvider::NoOp => None, + other => Some(other.to_string()), + }; Self { subscription_type: value.subscription_type.to_string(), customer_id: value.customer_id, - provider: value.provider.to_string(), + provider, } } } @@ -101,3 +106,165 @@ pub struct NewUserAttribution { pub from: String, pub company: String, } + +#[cfg(test)] +mod tests { + use cloud_billings::{ + CustomerBilling, MeteringProvider, SubscriptionType, org_usage::OrgUsageQueryResult, + }; + + use super::*; + + fn billing_with_provider(provider: MeteringProvider) -> CustomerBilling { + let mut cb = CustomerBilling::new("admin@example.com", "org-x"); + cb.provider = provider; + cb.subscription_type = SubscriptionType::Rate; + cb.customer_id = Some("cus_abc".to_string()); + cb + } + + #[test] + fn test_list_subscription_response_omits_provider_for_noop() { + // External-contract orgs (NoOp provider) must hide the provider field + // from the API response — that's the contract the frontend relies on + // to detect "this is an external contract" vs. a real billing backend. + let mut cb = billing_with_provider(MeteringProvider::NoOp); + cb.subscription_type = SubscriptionType::ExternalContract; + cb.customer_id = Some("custom".to_string()); + + let body: ListSubscriptionResponseBody = cb.into(); + + assert!(body.provider.is_none()); + assert_eq!(body.subscription_type, "external-contract"); + assert_eq!(body.customer_id.as_deref(), Some("custom")); + + let json = serde_json::to_value(&body).unwrap(); + assert!( + json.get("provider").is_none(), + "provider field must be omitted from JSON for NoOp, got {json}" + ); + } + + #[test] + fn test_list_subscription_response_includes_provider_for_stripe() { + let body: ListSubscriptionResponseBody = + billing_with_provider(MeteringProvider::Stripe).into(); + + assert_eq!(body.provider.as_deref(), Some("stripe")); + assert_eq!(body.subscription_type, "pay-as-you-go"); + } + + #[test] + fn test_list_subscription_response_includes_provider_for_aws_and_azure() { + let aws: ListSubscriptionResponseBody = billing_with_provider(MeteringProvider::Aws).into(); + assert_eq!(aws.provider.as_deref(), Some("aws")); + + let azure: ListSubscriptionResponseBody = + billing_with_provider(MeteringProvider::Azure).into(); + assert_eq!(azure.provider.as_deref(), Some("azure")); + } + + #[test] + fn test_convert_to_unit_gb_to_mb() { + let mut body = GetOrgUsageResponseBody { + data: vec![OrgUserData { + event: "ingestion".to_string(), + value: 2.0, + unit: "GB".to_string(), + }], + range: "30d".to_string(), + }; + + body.convert_to_unit("MB"); + + assert_eq!(body.data[0].unit, "MB"); + assert!((body.data[0].value - 2048.0).abs() < f64::EPSILON); + } + + #[test] + fn test_convert_to_unit_mb_to_gb() { + let mut body = GetOrgUsageResponseBody { + data: vec![OrgUserData { + event: "ingestion".to_string(), + value: 1024.0, + unit: "MB".to_string(), + }], + range: "30d".to_string(), + }; + + body.convert_to_unit("GB"); + + assert_eq!(body.data[0].unit, "GB"); + assert!((body.data[0].value - 1.0).abs() < f64::EPSILON); + } + + #[test] + fn test_convert_to_unit_is_case_insensitive() { + let mut body = GetOrgUsageResponseBody { + data: vec![OrgUserData { + event: "ingestion".to_string(), + value: 1.0, + unit: "gb".to_string(), + }], + range: "30d".to_string(), + }; + + body.convert_to_unit("mb"); + + assert_eq!(body.data[0].unit, "MB"); + assert!((body.data[0].value - 1024.0).abs() < f64::EPSILON); + } + + #[test] + fn test_convert_to_unit_no_op_when_units_match() { + let mut body = GetOrgUsageResponseBody { + data: vec![OrgUserData { + event: "ingestion".to_string(), + value: 5.5, + unit: "GB".to_string(), + }], + range: "7d".to_string(), + }; + + body.convert_to_unit("GB"); + + assert_eq!(body.data[0].unit, "GB"); + assert!((body.data[0].value - 5.5).abs() < f64::EPSILON); + } + + #[test] + fn test_convert_to_unit_unknown_pair_is_ignored() { + let mut body = GetOrgUsageResponseBody { + data: vec![OrgUserData { + event: "ingestion".to_string(), + value: 100.0, + unit: "Bytes".to_string(), + }], + range: "1d".to_string(), + }; + + body.convert_to_unit("MB"); + + // No conversion path defined for Bytes → MB; record stays untouched. + assert_eq!(body.data[0].unit, "Bytes"); + assert!((body.data[0].value - 100.0).abs() < f64::EPSILON); + } + + #[test] + fn test_org_user_data_from_query_result() { + // OrgUsageQueryResult only derives Deserialize, so build it from JSON. + // UsageEvent uses default (PascalCase) serde, not snake_case. + let json = serde_json::json!({ + "event": "Ingestion", + "unit": "GB", + "size": 12.5, + }); + let q: OrgUsageQueryResult = serde_json::from_value(json).unwrap(); + + let data: OrgUserData = q.into(); + + assert!((data.value - 12.5).abs() < f64::EPSILON); + assert_eq!(data.unit, "GB"); + assert_eq!(data.event, "Ingestion"); + } +} diff --git a/src/handler/http/request/organization/org.rs b/src/handler/http/request/organization/org.rs index 103703db34..103f019c99 100644 --- a/src/handler/http/request/organization/org.rs +++ b/src/handler/http/request/organization/org.rs @@ -27,12 +27,16 @@ use o2_enterprise::enterprise::common::config::get_config as get_o2_config; use { crate::common::meta::organization::OrganizationInvites, crate::common::meta::organization::{ - AllOrgListDetails, AllOrganizationResponse, ExtendTrialPeriodRequest, - OrganizationInviteUserRecord, + AllOrgListDetails, AllOrganizationResponse, CreateExternalContractRequest, + ExtendExternalContractRequest, ExtendTrialPeriodRequest, OrganizationInviteUserRecord, }, axum::body::Body, axum::http::StatusCode, - o2_enterprise::enterprise::cloud::list_customer_billings, + o2_enterprise::enterprise::cloud::{ + billings::{MeteringProvider, SubscriptionType}, + list_customer_billings, + }, + svix_ksuid::KsuidLike, }; use crate::{ @@ -188,30 +192,48 @@ pub async fn all_organizations( } }; - let all_subscriptions = match list_customer_billings().await { - Ok(orgs) => orgs - .into_iter() - .map(|cb| (cb.org_id, cb.subscription_type as i32)) - .collect::>(), + let all_billings = match list_customer_billings().await { + Ok(cbs) => cbs, Err(e) => { return MetaHttpResponse::internal_error(e.to_string()); } }; + let all_billing_info: HashMap<_, _> = all_billings + .iter() + .map(|cb| { + ( + cb.org_id.clone(), + (cb.subscription_type, cb.end_date, cb.provider.to_string()), + ) + }) + .collect(); + let mut id = 1; for org in all_orgs { + let billing_info = all_billing_info.get(&org.identifier); + let contract_end_date = billing_info.and_then(|(subscription, end_date, _)| { + if *subscription == SubscriptionType::ExternalContract { + *end_date + } else { + None + } + }); let org = AllOrgListDetails { id, identifier: org.identifier.clone(), name: org.org_name, org_type: org.org_type.to_string(), - plan: all_subscriptions - .get(&org.identifier) - .cloned() + plan: billing_info + .map(|(st, ..)| i16::from(*st) as i32) .unwrap_or_default(), created_at: org.created_at, updated_at: org.updated_at, trial_expires_at: Some(org.trial_ends_at), + contract_end_date, + billing_provider: billing_info + .map(|(_, _, provider)| provider.clone()) + .unwrap_or_default(), }; if !org_names.contains(&org.identifier) { org_names.insert(org.identifier.clone()); @@ -560,6 +582,212 @@ pub async fn extend_trial_period( ret } +/// CreateExternalContract +#[cfg(feature = "cloud")] +#[utoipa::path( + post, + path = "/{org_id}/external_contract", + context_path = "/api", + tag = "Organizations", + operation_id = "CreateExternalContract", + summary = "Create an external contract subscription for an organization", + security( + ("Authorization"= []) + ), + request_body(content = inline(CreateExternalContractRequest), description = "Create external contract request", content_type = "application/json"), + responses( + (status = 200, description = "Success", content_type = "application/json", body = String), + ), + extensions( + ("x-o2-mcp" = json!({"enabled": false})) + ) +)] +pub async fn create_external_contract( + Path(org_id): Path, + Json(req): Json, +) -> Response { + use o2_enterprise::enterprise::cloud::billings::CustomerBilling; + + if org_id != "_meta" { + return MetaHttpResponse::unauthorized("not authorized to access this resource"); + } + + // Validate the target org exists + if let Err(e) = infra::table::organizations::get(&req.org_id).await { + return MetaHttpResponse::not_found(format!("Organization not found: {e}")); + } + + // Get the org admin's email for the billing record + let admin = match infra::table::org_users::get_admin(&req.org_id).await { + Ok(admin) => admin, + Err(e) => { + return MetaHttpResponse::not_found(format!( + "Could not find admin for organization: {e}" + )); + } + }; + + let now = chrono::Utc::now().timestamp_micros(); + if req.end_date <= now { + return MetaHttpResponse::bad_request("end_date must be in the future"); + } + + // Block if the org already has any active billing record to avoid silently + // overwriting a live Stripe/AWS/Azure subscription. + if let Ok(Some(existing)) = + o2_enterprise::enterprise::cloud::billings::get_billing_by_org_id(&req.org_id).await + { + let msg = if matches!(existing.provider, MeteringProvider::NoOp) { + "Organization already has an active external contract" + } else { + log::warn!( + "[EXT_CONTRACT] Refusing to create external contract for org {}: existing {} subscription would be overwritten", + req.org_id, + existing.provider + ); + "Organization already has an active subscription; cancel it before creating an external contract" + }; + return MetaHttpResponse::bad_request(msg); + } + + let mut billing = CustomerBilling::new(&admin.email, &req.org_id); + billing.provider = MeteringProvider::NoOp; + billing.subscription_type = SubscriptionType::ExternalContract; + billing.customer_id = Some("custom".to_string()); + billing.subscription_id = Some(svix_ksuid::Ksuid::new(None, None).to_string()); + billing.end_date = Some(req.end_date); + billing.expiry_notified_stage = None; + + match o2_enterprise::enterprise::cloud::update_customer_billing(billing).await { + Ok(_) => MetaHttpResponse::json(serde_json::json!({"status": "ok"})), + Err(e) => { + MetaHttpResponse::internal_error(format!("Failed to create external contract: {e}")) + } + } +} + +/// ExtendExternalContract +#[cfg(feature = "cloud")] +#[utoipa::path( + put, + path = "/{org_id}/external_contract", + context_path = "/api", + tag = "Organizations", + operation_id = "ExtendExternalContract", + summary = "Extend an external contract end date", + security( + ("Authorization"= []) + ), + request_body(content = inline(ExtendExternalContractRequest), description = "Extend external contract request", content_type = "application/json"), + responses( + (status = 200, description = "Success", content_type = "application/json", body = String), + ), + extensions( + ("x-o2-mcp" = json!({"enabled": false})) + ) +)] +pub async fn extend_external_contract( + Path(org_id): Path, + Json(req): Json, +) -> Response { + if org_id != "_meta" { + return MetaHttpResponse::unauthorized("not authorized to access this resource"); + } + + let mut billing = match o2_enterprise::enterprise::cloud::billings::get_billing_by_org_id( + &req.org_id, + ) + .await + { + Ok(Some(b)) if matches!(b.provider, MeteringProvider::NoOp) => b, + Ok(_) => { + return MetaHttpResponse::not_found( + "No active external contract found for this organization", + ); + } + Err(e) => { + return MetaHttpResponse::internal_error(e.to_string()); + } + }; + + let now = chrono::Utc::now().timestamp_micros(); + if req.new_end_date <= now { + return MetaHttpResponse::bad_request("new_end_date must be in the future"); + } + if billing + .end_date + .is_some_and(|current| req.new_end_date <= current) + { + return MetaHttpResponse::bad_request( + "new_end_date must be after the current contract end date", + ); + } + + billing.end_date = Some(req.new_end_date); + // Reset expiry notifications so warnings fire again for the new window. + billing.expiry_notified_stage = None; + + match o2_enterprise::enterprise::cloud::update_customer_billing(billing).await { + Ok(_) => MetaHttpResponse::json(serde_json::json!({"status": "ok"})), + Err(e) => { + MetaHttpResponse::internal_error(format!("Failed to extend external contract: {e}")) + } + } +} + +/// RevokeExternalContract +#[cfg(feature = "cloud")] +#[utoipa::path( + delete, + path = "/{org_id}/external_contract/{target_org_id}", + context_path = "/api", + tag = "Organizations", + operation_id = "RevokeExternalContract", + summary = "Revoke an external contract", + security( + ("Authorization"= []) + ), + params( + ("org_id" = String, Path, description = "Meta organization id"), + ("target_org_id" = String, Path, description = "Target organization id"), + ), + responses( + (status = 200, description = "Success", content_type = "application/json", body = String), + ), + extensions( + ("x-o2-mcp" = json!({"enabled": false})) + ) +)] +pub async fn revoke_external_contract( + Path((org_id, target_org_id)): Path<(String, String)>, +) -> Response { + if org_id != "_meta" { + return MetaHttpResponse::unauthorized("not authorized to access this resource"); + } + + // Verify the org has an external contract before deleting + match o2_enterprise::enterprise::cloud::billings::get_billing_by_org_id(&target_org_id).await { + Ok(Some(b)) if matches!(b.provider, MeteringProvider::NoOp) => {} + Ok(_) => { + return MetaHttpResponse::not_found( + "No active external contract found for this organization", + ); + } + Err(e) => { + return MetaHttpResponse::internal_error(e.to_string()); + } + }; + + match o2_enterprise::enterprise::cloud::customer_billings::delete_by_org_id(&target_org_id) + .await + { + Ok(_) => MetaHttpResponse::json(serde_json::json!({"status": "ok"})), + Err(e) => { + MetaHttpResponse::internal_error(format!("Failed to revoke external contract: {e}")) + } + } +} + /// RenameOrganization #[utoipa::path( put, diff --git a/src/handler/http/router/mod.rs b/src/handler/http/router/mod.rs index 04d98844ea..21b1aaf850 100644 --- a/src/handler/http/router/mod.rs +++ b/src/handler/http/router/mod.rs @@ -966,6 +966,15 @@ pub fn service_routes() -> Router { "/{org_id}/extend_trial_period", put(organization::org::extend_trial_period), ) + .route( + "/{org_id}/external_contract", + post(organization::org::create_external_contract) + .put(organization::org::extend_external_contract), + ) + .route( + "/{org_id}/external_contract/{target_org_id}", + delete(organization::org::revoke_external_contract), + ) .route( "/{org_id}/aws-marketplace/link-subscription", post(cloud::aws_marketplace::link_subscription), diff --git a/src/job/cloud.rs b/src/job/cloud.rs index 6893d7fc12..f19a237e2f 100644 --- a/src/job/cloud.rs +++ b/src/job/cloud.rs @@ -41,11 +41,25 @@ const NO_INGESTION_REPORT_INTERVAL: u64 = 3600; /// DB flush interval for trial quota deductions (seconds). const TRIAL_QUOTA_FLUSH_INTERVAL: u64 = 10; +/// Interval for external contract expiry checks (1 hour). +const EXTERNAL_CONTRACT_CHECK_INTERVAL: u64 = 3600; + +/// (days-remaining threshold, stage stored after the warning fires). +/// Walked in descending-urgency order so we send at most one warning per tick. +const EXPIRY_WARNING_STAGES: &[( + i64, + o2_enterprise::enterprise::cloud::billings::ExpiryNotificationStage, +)] = { + use o2_enterprise::enterprise::cloud::billings::ExpiryNotificationStage::*; + &[(1, OneDay), (7, SevenDay), (30, ThirtyDay)] +}; + pub fn start() { tokio::spawn(async move { run_no_ingestion_period().await }); tokio::spawn(async move { run_no_ingestion_daily().await }); tokio::spawn(async move { run_org_expiry_daily().await }); tokio::spawn(async move { run_ai_quota_check().await }); + tokio::spawn(async move { run_external_contract_expiry_check().await }); } /// Start trial quota background jobs (flush + cluster sync). @@ -351,3 +365,350 @@ async fn run_org_expiry_daily() { log::info!("daily expiry reminder reporting completed"); } } + +async fn run_external_contract_expiry_check() { + let mut interval = tokio::time::interval(tokio::time::Duration::from_secs( + EXTERNAL_CONTRACT_CHECK_INTERVAL, + )); + interval.tick().await; // skip first immediate tick + loop { + interval.tick().await; + check_external_contract_expiry().await; + } +} + +async fn check_external_contract_expiry() { + use o2_enterprise::enterprise::cloud::billings::MeteringProvider; + + let billings = match o2_enterprise::enterprise::cloud::list_customer_billings().await { + Ok(cbs) => cbs, + Err(e) => { + log::error!("[EXT_CONTRACT] Failed to list customer billings: {e}"); + return; + } + }; + + let now = chrono::Utc::now().timestamp_micros(); + let day_micros = hour_micros(24); + + for cb in billings { + if !matches!(cb.provider, MeteringProvider::NoOp) { + continue; + } + + let Some(end_date) = cb.end_date else { + continue; + }; + + let org_id = &cb.org_id; + + // Check if contract has expired + if end_date <= now { + log::info!( + "[EXT_CONTRACT] External contract expired for org {org_id}, reverting to Free tier" + ); + if let Err(e) = + o2_enterprise::enterprise::cloud::customer_billings::delete_by_org_id(org_id).await + { + log::error!( + "[EXT_CONTRACT] Failed to delete expired billing for org {org_id}: {e}" + ); + } + continue; + } + + // Check if we need to send expiry warnings + let days_remaining = (end_date - now) / day_micros; + + let stage = match find_pending_expiry_stage(days_remaining, cb.expiry_notified_stage) { + Some(s) => s, + None => continue, + }; + + // Send warning email to org admin + let admin = match get_admin(org_id).await { + Ok(u) => u, + Err(e) => { + log::error!("[EXT_CONTRACT] Failed to get admin for org={org_id}: {e}"); + continue; + } + }; + + let org_name = match infra::table::organizations::get(org_id).await { + Ok(record) => record.org_name, + Err(_) => String::new(), + }; + + let (subject, body) = + build_external_contract_expiry_email(org_id, &org_name, days_remaining); + + let email = Email { + recipients: vec![admin.email.clone()], + }; + + match crate::service::alerts::alert::send_email_notification(&subject, &email, body).await { + Ok(_) => { + log::info!( + "[EXT_CONTRACT] Sent {stage:?} expiry warning to {} for org={org_id}", + admin.email + ); + } + Err(e) => { + log::error!( + "[EXT_CONTRACT] Failed to send expiry warning email for org={org_id}: {e}" + ); + continue; + } + } + + // Report to segment proxy for contract expiry tracking + report_to_segment( + "OpenObserve - Contract expiry reminder", + org_id, + &format!("{days_remaining} days"), + ) + .await; + + // Advance the stored stage so we don't resend. + let mut updated_billing = cb.clone(); + updated_billing.expiry_notified_stage = Some(stage); + if let Err(e) = + o2_enterprise::enterprise::cloud::update_customer_billing(updated_billing).await + { + log::error!("[EXT_CONTRACT] Failed to update expiry stage for org={org_id}: {e}"); + } + } +} + +fn build_external_contract_expiry_email( + org_id: &str, + org_name: &str, + days_remaining: i64, +) -> (String, String) { + let display_name = if org_name.is_empty() { + org_id + } else { + org_name + }; + + let subject = format!( + "[OpenObserve] Contract expiring in {} day{} — {}", + days_remaining, + if days_remaining == 1 { "" } else { "s" }, + display_name + ); + + let message = if days_remaining <= 1 { + "Your contract expires tomorrow. Please contact your account manager to renew \ + and avoid service interruption." + } else if days_remaining <= 7 { + "Your contract is expiring soon. Please contact your account manager to renew \ + your subscription." + } else { + "Your contract will expire in the coming weeks. Please reach out to your \ + account manager if you'd like to renew." + }; + + let display_name_esc = html_escape(display_name); + let org_id_esc = html_escape(org_id); + let body = format!( + "\ +

Contract Expiry Notice

\ +

Organization: {display_name_esc} ({org_id_esc})

\ +

{message}

\ +

Days remaining: {days_remaining}

\ +

After expiry, your organization will revert to the Free tier.

\ + " + ); + + (subject, body) +} + +fn html_escape(s: &str) -> String { + s.replace('&', "&") + .replace('<', "<") + .replace('>', ">") +} + +/// Pick the most-urgent expiry warning stage that should fire, given the current +/// `days_remaining` and the last stage already notified. +/// +/// Walks `EXPIRY_WARNING_STAGES` in descending-urgency order (30 → 7 → 1) and +/// returns the first stage whose threshold has been reached and is strictly +/// greater than `current_stage`. Returns `None` when no new warning is due. +fn find_pending_expiry_stage( + days_remaining: i64, + current_stage: Option, +) -> Option { + EXPIRY_WARNING_STAGES + .iter() + .find(|(threshold, stage)| { + days_remaining <= *threshold && current_stage.is_none_or(|s| s < *stage) + }) + .map(|(_, stage)| *stage) +} + +#[cfg(test)] +mod tests { + use o2_enterprise::enterprise::cloud::billings::ExpiryNotificationStage; + + use super::*; + + #[test] + fn test_html_escape_basic_chars() { + assert_eq!( + html_escape(""), + "<script>alert('x')</script>" + ); + assert_eq!(html_escape("a & b"), "a & b"); + assert_eq!(html_escape(""), ""); + assert_eq!(html_escape("plain text"), "plain text"); + } + + #[test] + fn test_html_escape_orders_amp_first() { + // & must be escaped before < and > so we don't double-escape entities. + assert_eq!(html_escape("&<>"), "&<>"); + assert_eq!(html_escape("&"), "&amp;"); + } + + #[test] + fn test_build_expiry_email_singular_day() { + let (subject, body) = build_external_contract_expiry_email("org-a", "Acme Corp", 1); + + // "1 day" — no plural suffix + assert!( + subject.contains("expiring in 1 day —"), + "subject: {subject}" + ); + assert!(!subject.contains("1 days")); + assert!(subject.contains("Acme Corp")); + // <=1 days uses the most urgent message + assert!(body.contains("expires tomorrow")); + assert!(body.contains("Days remaining: 1")); + assert!(body.contains("Acme Corp")); + assert!(body.contains("(org-a)")); + } + + #[test] + fn test_build_expiry_email_plural_seven_days() { + let (subject, body) = build_external_contract_expiry_email("org-b", "Beta Inc", 7); + + assert!(subject.contains("expiring in 7 days")); + // 7 days falls into the "expiring soon" bucket (>1 and <=7) + assert!(body.contains("expiring soon")); + assert!(!body.contains("expires tomorrow")); + } + + #[test] + fn test_build_expiry_email_plural_thirty_days() { + let (_subject, body) = build_external_contract_expiry_email("org-c", "Gamma LLC", 30); + + // >7 days falls into the "coming weeks" bucket + assert!(body.contains("coming weeks")); + assert!(!body.contains("expires tomorrow")); + assert!(!body.contains("expiring soon")); + } + + #[test] + fn test_build_expiry_email_falls_back_to_org_id_when_name_empty() { + let (subject, body) = build_external_contract_expiry_email("org-d", "", 5); + + // Empty org name → org_id is used as the display name in subject and body. + assert!(subject.ends_with("— org-d"), "subject: {subject}"); + assert!(body.contains("Organization: org-d (org-d)")); + } + + #[test] + fn test_build_expiry_email_escapes_html_in_org_name() { + let (_subject, body) = + build_external_contract_expiry_email("org-e", "", 10); + + // The malicious org name must be escaped in the email body. + assert!(!body.contains("")); + assert!(body.contains("<script>alert(1)</script>")); + } + + #[test] + fn test_find_pending_stage_no_warning_when_far_out() { + // Plenty of time left, no warning has been sent yet. + assert_eq!(find_pending_expiry_stage(60, None), None); + } + + #[test] + fn test_find_pending_stage_thirty_day_first_fire() { + // 30 days remaining, nothing notified yet → fire ThirtyDay. + assert_eq!( + find_pending_expiry_stage(30, None), + Some(ExpiryNotificationStage::ThirtyDay) + ); + // Same goes for 25 days remaining. + assert_eq!( + find_pending_expiry_stage(25, None), + Some(ExpiryNotificationStage::ThirtyDay) + ); + } + + #[test] + fn test_find_pending_stage_no_resend_thirty_day() { + // ThirtyDay already sent, days_remaining still in the 8..30 window → no new + // warning yet (must wait until SevenDay threshold). + assert_eq!( + find_pending_expiry_stage(20, Some(ExpiryNotificationStage::ThirtyDay)), + None + ); + } + + #[test] + fn test_find_pending_stage_seven_day_after_thirty() { + // ThirtyDay already sent and we crossed the 7-day boundary → fire SevenDay. + assert_eq!( + find_pending_expiry_stage(7, Some(ExpiryNotificationStage::ThirtyDay)), + Some(ExpiryNotificationStage::SevenDay) + ); + assert_eq!( + find_pending_expiry_stage(3, Some(ExpiryNotificationStage::ThirtyDay)), + Some(ExpiryNotificationStage::SevenDay) + ); + } + + #[test] + fn test_find_pending_stage_one_day_after_seven() { + assert_eq!( + find_pending_expiry_stage(1, Some(ExpiryNotificationStage::SevenDay)), + Some(ExpiryNotificationStage::OneDay) + ); + } + + #[test] + fn test_find_pending_stage_no_resend_after_one_day() { + // OneDay (the most urgent) has already been sent → never fire again. + assert_eq!( + find_pending_expiry_stage(1, Some(ExpiryNotificationStage::OneDay)), + None + ); + assert_eq!( + find_pending_expiry_stage(0, Some(ExpiryNotificationStage::OneDay)), + None + ); + } + + #[test] + fn test_find_pending_stage_skips_levels_when_first_seen_late() { + // Org is already inside the 7-day window when we first notice it: skip + // ThirtyDay and fire SevenDay directly, since both thresholds are met. + assert_eq!( + find_pending_expiry_stage(5, None), + Some(ExpiryNotificationStage::ThirtyDay) + ); + // First check happens with only 1 day remaining: ThirtyDay still wins + // because the iterator returns the first matching stage. The next tick + // (after marking ThirtyDay as sent) escalates to SevenDay, then OneDay. + // This mirrors the production loop's behaviour of at-most-one-notice + // per tick. + assert_eq!( + find_pending_expiry_stage(1, None), + Some(ExpiryNotificationStage::ThirtyDay) + ); + } +} diff --git a/web/src/components/settings/OrganizationManagement.spec.ts b/web/src/components/settings/OrganizationManagement.spec.ts index 9bdb78fe71..10ee994c1b 100644 --- a/web/src/components/settings/OrganizationManagement.spec.ts +++ b/web/src/components/settings/OrganizationManagement.spec.ts @@ -272,14 +272,16 @@ describe("OrganizationManagement.vue", () => { it("should have correct column configuration", () => { wrapper = createWrapper(); const columns = wrapper.vm.columns; - expect(columns).toHaveLength(7); + expect(columns).toHaveLength(9); expect(columns[0].name).toBe("#"); expect(columns[1].name).toBe("name"); expect(columns[2].name).toBe("identifier"); expect(columns[3].name).toBe("subscription_status"); - expect(columns[4].name).toBe("created_on"); - expect(columns[5].name).toBe("trial_expiry"); - expect(columns[6].name).toBe("actions"); + expect(columns[4].name).toBe("billing_provider"); + expect(columns[5].name).toBe("created_on"); + expect(columns[6].name).toBe("trial_expiry"); + expect(columns[7].name).toBe("contract_end_date"); + expect(columns[8].name).toBe("actions"); }); it("should have correct perPageOptions", () => { @@ -424,8 +426,11 @@ describe("OrganizationManagement.vue", () => { name: "Test Org 1", identifier: "test-org-1", plan: "Pay as you go", + billing_provider: "-", created_at: "2023-12-01", - trial_expires_at: "2023-12-01" + trial_expires_at: "2023-12-01", + contract_end_date: 0, + contract_end_date_display: "-" }); }); diff --git a/web/src/components/settings/OrganizationManagement.vue b/web/src/components/settings/OrganizationManagement.vue index a5073f5a21..3cd4235771 100644 --- a/web/src/components/settings/OrganizationManagement.vue +++ b/web/src/components/settings/OrganizationManagement.vue @@ -46,6 +46,42 @@ along with this program. If not, see . data-test="otg-management-extend-trial-btn" @click.stop="toggleExtendTrialDialog(props.row)" > + + + + + @@ -137,6 +175,57 @@ along with this program. If not, see . + + + + + + + + {{ contractMode === 'create' ? 'Create' : 'Extend' }} External Contract for {{ contractDataRow.name }} + + + + + + +
+
+ {{ contractMode === 'create' ? 'End Date' : 'New End Date' }} +
+ +
+
+ Current end date: {{ formatMicrosToDate(contractDataRow.contract_end_date) }} +
+ + + + + + + +