// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::admin::{
    auth::authorize_admin_request,
    handlers::notify_runtime_access::{get_notification_system, load_notification_config_snapshot},
    handlers::supervise_admin_mutation,
    handlers::target_descriptor::{
        AdminTargetSpec, EndpointKey, RuntimeHealthStatus, TargetEndpointSource, admin_target_spec_from_builtin,
        build_enabled_target_kvs, build_json_response, collect_runtime_statuses, extract_supported_target_params,
        merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason,
        target_mutation_block_reason as shared_target_mutation_block_reason,
    },
    router::{AdminOperation, Operation, S3Router},
    runtime_sources::{AppContext, app_context_from_req},
    service::config::{preflight_dynamic_config_reload_for_context, signal_dynamic_config_reload_checked_for_context},
};
use crate::server::{
    ADMIN_PREFIX, is_notify_module_enabled, refresh_notify_module_enabled, refresh_persisted_module_switches_from_store,
};
use http::StatusCode;
use hyper::Method;
use matchit::Params;
use rustfs_config::notify::NOTIFY_ROUTE_PREFIX;
use rustfs_config::server_config::Config;
use rustfs_config::{EVENT_DEFAULT_DIR, MAX_ADMIN_REQUEST_BODY_SIZE};
use rustfs_policy::policy::action::{Action, AdminAction};
use rustfs_targets::catalog::builtin::builtin_notify_target_admin_descriptors;
use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::LazyLock;
use tracing::{error, info, warn};

const LOG_COMPONENT_ADMIN_API: &str = "admin_api";

async fn converge_target_mutation_on_cluster(
    context: Option<&AppContext>,
    target_type: &str,
    local_result: Result<(), rustfs_notify::NotificationError>,
) -> S3Result<()> {
    let subsystem = notification_target_subsystem(target_type)?;
    let peer_result = signal_dynamic_config_reload_checked_for_context(context, subsystem).await;

    match (local_result, peer_result) {
        (Ok(()), Ok(())) => Ok(()),
        (Err(local), Ok(())) => {
            warn!(target_type, error = %local, "Local notification runtime failed to converge");
            Err(s3_error!(InternalError, "local notification runtime failed to converge"))
        }
        (Ok(()), Err(peer)) => Err(peer),
        (Err(local), Err(peer)) => {
            warn!(target_type, error = %local, "Local notification runtime failed while peer convergence also failed");
            Err(s3_error!(
                InternalError,
                "local notification runtime and peer convergence failed: {}",
                peer
            ))
        }
    }
}

fn notification_target_subsystem(target_type: &str) -> S3Result<&'static str> {
    notification_target_specs()
        .iter()
        .find(|spec| spec.subsystem == target_type)
        .map(|spec| spec.subsystem)
        .ok_or_else(|| s3_error!(InvalidArgument, "unsupported notification target type: {}", target_type))
}

async fn preflight_target_mutation_on_cluster(context: Option<&AppContext>, target_type: &str) -> S3Result<()> {
    preflight_dynamic_config_reload_for_context(context, notification_target_subsystem(target_type)?).await
}

const LOG_SUBSYSTEM_NOTIFICATION_TARGET: &str = "notification_target";
const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected";
const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed";
const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted";
const EVENT_ADMIN_OPERATION_BLOCKED: &str = "admin_operation_blocked";

macro_rules! log_notification_target_operation_blocked {
    ($operation:expr, Some($target_type:expr), Some($target_name:expr), $reason:expr) => {
        warn!(
            event = EVENT_ADMIN_OPERATION_BLOCKED,
            component = LOG_COMPONENT_ADMIN_API,
            subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
            operation = $operation,
            result = "blocked",
            target_type = $target_type,
            target_name = $target_name,
            reason = $reason,
            "admin request rejected"
        );
    };
    ($operation:expr, None, None, $reason:expr) => {
        warn!(
            event = EVENT_ADMIN_OPERATION_BLOCKED,
            component = LOG_COMPONENT_ADMIN_API,
            subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
            operation = $operation,
            result = "blocked",
            reason = $reason,
            "admin request rejected"
        );
    };
}

macro_rules! log_notification_target_body_read_failed {
    ($target_type:expr, $target_name:expr, $err:expr) => {
        warn!(
            event = EVENT_ADMIN_REQUEST_REJECTED,
            component = LOG_COMPONENT_ADMIN_API,
            subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
            operation = "set_target_config",
            result = "rejected",
            reason = "request_body_read_failed",
            target_type = $target_type,
            target_name = $target_name,
            error = %$err,
            "admin request rejected"
        );
    };
}

macro_rules! log_notification_target_body_decode_failed {
    ($target_type:expr, $target_name:expr, $err:expr) => {
        warn!(
            event = EVENT_ADMIN_REQUEST_REJECTED,
            component = LOG_COMPONENT_ADMIN_API,
            subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
            operation = "set_target_config",
            result = "rejected",
            reason = "request_body_decode_failed",
            target_type = $target_type,
            target_name = $target_name,
            error = %$err,
            "admin request rejected"
        );
    };
}

macro_rules! log_notification_target_config_updated {
    ($operation:expr, $target_type:expr, $target_name:expr) => {
        info!(
            event = EVENT_ADMIN_RESPONSE_EMITTED,
            component = LOG_COMPONENT_ADMIN_API,
            subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
            operation = $operation,
            result = "success",
            target_type = $target_type,
            target_name = $target_name,
            "admin response emitted"
        );
    };
}

macro_rules! log_notification_target_request_failed {
    ($operation:expr, $reason:expr, Some($target_type:expr), Some($target_name:expr), $err:expr) => {
        error!(
            event = EVENT_ADMIN_REQUEST_FAILED,
            component = LOG_COMPONENT_ADMIN_API,
            subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
            operation = $operation,
            result = "failed",
            reason = $reason,
            target_type = $target_type,
            target_name = $target_name,
            error = %$err,
            "admin request failed"
        );
    };
    ($operation:expr, $reason:expr, None, None, $err:expr) => {
        error!(
            event = EVENT_ADMIN_REQUEST_FAILED,
            component = LOG_COMPONENT_ADMIN_API,
            subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
            operation = $operation,
            result = "failed",
            reason = $reason,
            error = %$err,
            "admin request failed"
        );
    };
}

pub fn register_notification_target_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
    r.insert(
        Method::GET,
        format!("{}{}", ADMIN_PREFIX, "/v3/target/list").as_str(),
        AdminOperation(&ListNotificationTargets {}),
    )?;

    r.insert(
        Method::PUT,
        format!("{}{}", ADMIN_PREFIX, "/v3/target/{target_type}/{target_name}").as_str(),
        AdminOperation(&NotificationTarget {}),
    )?;

    r.insert(
        Method::DELETE,
        format!("{}{}", ADMIN_PREFIX, "/v3/target/{target_type}/{target_name}/reset").as_str(),
        AdminOperation(&RemoveNotificationTarget {}),
    )?;

    r.insert(
        Method::GET,
        format!("{}{}", ADMIN_PREFIX, "/v3/target/arns").as_str(),
        AdminOperation(&ListTargetsArns {}),
    )?;

    Ok(())
}

#[derive(Debug, Deserialize)]
pub struct KeyValue {
    pub key: String,
    pub value: String,
}

#[derive(Debug, Deserialize)]
pub struct NotificationTargetBody {
    pub key_values: Vec<KeyValue>,
}

#[derive(Serialize, Debug)]
struct NotificationEndpoint {
    account_id: String,
    service: String,
    status: String,
    health_state: String,
    health_reason: String,
    source: TargetEndpointSource,
}

#[derive(Serialize, Debug)]
struct NotificationEndpointsResponse {
    notify_enabled: bool,
    notification_endpoints: Vec<NotificationEndpoint>,
}

static NOTIFICATION_TARGET_SPECS: LazyLock<Vec<AdminTargetSpec>> = LazyLock::new(|| {
    builtin_notify_target_admin_descriptors()
        .into_iter()
        .map(|descriptor| admin_target_spec_from_builtin(&descriptor))
        .collect()
});

fn notification_target_specs() -> &'static [AdminTargetSpec] {
    &NOTIFICATION_TARGET_SPECS
}

// --- Helper Functions ---

/// The pre-check keeps these endpoints' historical missing-credentials message;
/// the shared gate reports "get cred failed".
async fn authorize_notification_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
    if req.credentials.is_none() {
        return Err(s3_error!(InvalidRequest, "credentials not found"));
    }
    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
    Ok(())
}

fn target_mutation_block_reason(config: &Config, target_type: &str, target_name: &str) -> S3Result<Option<String>> {
    shared_target_mutation_block_reason(
        notification_target_specs(),
        NOTIFY_ROUTE_PREFIX,
        config,
        target_type,
        target_name,
        "target",
    )
}

async fn notification_target_operation_block_reason(action: &str) -> Option<String> {
    if let Err(err) = refresh_persisted_module_switches_from_store().await {
        warn!(
            event = EVENT_ADMIN_REQUEST_REJECTED,
            component = LOG_COMPONENT_ADMIN_API,
            subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
            operation = "reload_persisted_module_switches",
            result = "failed",
            error = %err,
            "admin request rejected"
        );
    }
    refresh_notify_module_enabled();
    target_module_disabled_reason("notify", rustfs_config::ENV_NOTIFY_ENABLE, is_notify_module_enabled(), action)
}

fn merge_notification_endpoints(
    config: &Config,
    runtime_statuses: HashMap<EndpointKey, RuntimeHealthStatus>,
) -> S3Result<Vec<NotificationEndpoint>> {
    Ok(
        shared_merge_target_endpoints(notification_target_specs(), NOTIFY_ROUTE_PREFIX, config, runtime_statuses)?
            .into_iter()
            .map(|endpoint| NotificationEndpoint {
                account_id: endpoint.account_id,
                service: endpoint.service,
                status: endpoint.status,
                health_state: endpoint.health_state,
                health_reason: endpoint.health_reason,
                source: endpoint.source,
            })
            .collect(),
    )
}

fn collect_online_target_arns(region: &str, target_statuses: Vec<(rustfs_targets::arn::TargetID, String)>) -> Vec<String> {
    target_statuses
        .into_iter()
        .filter_map(|(target_id, status)| (status == "online").then(|| target_id.to_arn(region).to_string()))
        .collect()
}

// --- Operations ---

pub struct NotificationTarget {}
#[async_trait::async_trait]
impl Operation for NotificationTarget {
    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let (target_type, target_name) = extract_target_params(&params)?;
        let context = app_context_from_req(&req);

        authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
        if let Some(reason) = notification_target_operation_block_reason("managing notification targets from the console").await {
            log_notification_target_operation_blocked!("set_target_config", Some(target_type), Some(target_name), &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }
        let (ns, config_snapshot) = load_notification_config_snapshot().await?;
        if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name)? {
            log_notification_target_operation_blocked!("set_target_config", Some(target_type), Some(target_name), &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }

        let mut input = req.input;
        let body_bytes = input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await.map_err(|e| {
            log_notification_target_body_read_failed!(target_type, target_name, e);
            s3_error!(InvalidRequest, "failed to read request body")
        })?;

        let notification_body: NotificationTargetBody = serde_json::from_slice(&body_bytes).map_err(|e| {
            log_notification_target_body_decode_failed!(target_type, target_name, e);
            s3_error!(InvalidArgument, "invalid json body for target config: {}", e)
        })?;

        let specs = notification_target_specs();
        let kvs = build_enabled_target_kvs(
            specs,
            notification_body
                .key_values
                .iter()
                .map(|kv| (kv.key.as_str(), kv.value.as_str())),
            target_type,
            EVENT_DEFAULT_DIR,
            "target",
        )
        .await?;

        let mutation_target_type = target_type.to_owned();
        let mutation_target_name = target_name.to_owned();
        supervise_admin_mutation("notification target mutation", async move {
            preflight_target_mutation_on_cluster(context.as_deref(), &mutation_target_type).await?;
            let local_result = ns.set_target_config(&mutation_target_type, &mutation_target_name, kvs).await;
            converge_target_mutation_on_cluster(context.as_deref(), &mutation_target_type, local_result).await
        })
        .await
        .map_err(|e| {
            log_notification_target_request_failed!(
                "set_target_config",
                "set_target_config_failed",
                Some(target_type),
                Some(target_name),
                e
            );
            e
        })?;
        log_notification_target_config_updated!("set_target_config", target_type, target_name);

        Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id")))
    }
}

pub struct ListNotificationTargets {}
#[async_trait::async_trait]
impl Operation for ListNotificationTargets {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;
        refresh_persisted_module_switches_from_store().await.map_err(|err| {
            warn!(
                event = EVENT_ADMIN_REQUEST_FAILED,
                component = LOG_COMPONENT_ADMIN_API,
                subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
                operation = "list_targets",
                result = "failed",
                reason = "module_switch_refresh_failed",
                error = %err,
                "admin request failed"
            );
            s3_error!(InternalError, "failed to refresh notification module state")
        })?;
        let notify_enabled = refresh_notify_module_enabled();
        let (ns, config) = load_notification_config_snapshot().await?;
        let runtime_statuses = collect_runtime_statuses(ns.get_target_values().await).await;
        let notification_endpoints = merge_notification_endpoints(&config, runtime_statuses)?;

        let data = serde_json::to_vec(&NotificationEndpointsResponse {
            notify_enabled,
            notification_endpoints,
        })
        .map_err(|e| {
            log_notification_target_request_failed!("list_targets", "serialize_targets_failed", None, None, e);
            s3_error!(InternalError, "failed to serialize targets: {}", e)
        })?;

        Ok(build_json_response(StatusCode::OK, Body::from(data), req.headers.get("x-request-id")))
    }
}

pub struct ListTargetsArns {}
#[async_trait::async_trait]
impl Operation for ListTargetsArns {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;
        if let Some(reason) = notification_target_operation_block_reason(
            "querying notification target ARNs for bucket associations from the console",
        )
        .await
        {
            log_notification_target_operation_blocked!("list_target_arns", None, None, &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }
        let ns = get_notification_system().await?;

        let region = req.region.clone().ok_or_else(|| {
            warn!(
                event = EVENT_ADMIN_REQUEST_REJECTED,
                component = LOG_COMPONENT_ADMIN_API,
                subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET,
                operation = "list_target_arns",
                result = "rejected",
                reason = "region_not_found",
                "admin request rejected"
            );
            s3_error!(InvalidRequest, "region not found")
        })?;
        let target_statuses = collect_runtime_statuses(ns.get_target_values().await)
            .await
            .into_iter()
            .map(|((account_id, service), health)| (rustfs_targets::arn::TargetID::new(account_id, service), health.status))
            .collect();

        let data_target_arn_list = collect_online_target_arns(region.as_str(), target_statuses);

        let data = serde_json::to_vec(&data_target_arn_list).map_err(|e| {
            log_notification_target_request_failed!("list_target_arns", "serialize_targets_failed", None, None, e);
            s3_error!(InternalError, "failed to serialize targets: {}", e)
        })?;

        Ok(build_json_response(StatusCode::OK, Body::from(data), req.headers.get("x-request-id")))
    }
}

pub struct RemoveNotificationTarget {}
#[async_trait::async_trait]
impl Operation for RemoveNotificationTarget {
    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let (target_type, target_name) = extract_target_params(&params)?;
        let context = app_context_from_req(&req);

        authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
        if let Some(reason) = notification_target_operation_block_reason("managing notification targets from the console").await {
            log_notification_target_operation_blocked!("remove_target_config", Some(target_type), Some(target_name), &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }
        let (ns, config_snapshot) = load_notification_config_snapshot().await?;
        if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name)? {
            log_notification_target_operation_blocked!("remove_target_config", Some(target_type), Some(target_name), &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }

        let mutation_target_type = target_type.to_owned();
        let mutation_target_name = target_name.to_owned();
        supervise_admin_mutation("notification target mutation", async move {
            preflight_target_mutation_on_cluster(context.as_deref(), &mutation_target_type).await?;
            let local_result = ns.remove_target_config(&mutation_target_type, &mutation_target_name).await;
            converge_target_mutation_on_cluster(context.as_deref(), &mutation_target_type, local_result).await
        })
        .await
        .map_err(|e| {
            log_notification_target_request_failed!(
                "remove_target_config",
                "remove_target_config_failed",
                Some(target_type),
                Some(target_name),
                e
            );
            e
        })?;
        log_notification_target_config_updated!("remove_target_config", target_type, target_name);

        Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id")))
    }
}

fn extract_target_params<'a>(params: &'a Params<'_, '_>) -> S3Result<(&'a str, &'a str)> {
    extract_supported_target_params(notification_target_specs(), params, "notification")
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::admin::handlers::target_descriptor::{
        allowed_target_keys, collect_validated_key_values as shared_collect_validated_key_values,
    };
    use matchit::Router;
    use rustfs_config::notify::{NOTIFY_AMQP_SUB_SYS, NOTIFY_KAFKA_SUB_SYS, NOTIFY_MQTT_SUB_SYS, NOTIFY_WEBHOOK_SUB_SYS};
    use rustfs_config::server_config::{KV, KVS};
    use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY};
    use rustfs_targets::arn::TargetID;
    use serial_test::serial;
    use std::collections::{HashMap, HashSet};
    use temp_env::{with_var, with_vars};

    fn enabled_kvs(value: &str) -> KVS {
        KVS(vec![KV {
            key: ENABLE_KEY.to_string(),
            value: value.to_string(),
            hidden_if_empty: false,
        }])
    }

    fn runtime_health(status: &str) -> RuntimeHealthStatus {
        let online = status == "online";
        RuntimeHealthStatus {
            status: status.to_string(),
            state: if online { "online" } else { "offline" }.to_string(),
            reason: if online { "reachable" } else { "unreachable" }.to_string(),
        }
    }

    #[test]
    fn notification_target_subsystem_resolves_admin_route_type() {
        assert_eq!(
            notification_target_subsystem(NOTIFY_WEBHOOK_SUB_SYS).expect("webhook subsystem should resolve"),
            NOTIFY_WEBHOOK_SUB_SYS
        );
        assert!(notification_target_subsystem("webhook").is_err());
    }

    #[test]
    fn notification_endpoints_response_includes_required_module_state() {
        let response = NotificationEndpointsResponse {
            notify_enabled: true,
            notification_endpoints: vec![NotificationEndpoint {
                account_id: "primary".to_string(),
                service: "webhook".to_string(),
                status: "online".to_string(),
                health_state: "online".to_string(),
                health_reason: "reachable".to_string(),
                source: TargetEndpointSource::Config,
            }],
        };

        assert_eq!(
            serde_json::to_value(response).expect("notification target response should serialize"),
            serde_json::json!({
                "notify_enabled": true,
                "notification_endpoints": [{
                    "account_id": "primary",
                    "service": "webhook",
                    "status": "online",
                    "health_state": "online",
                    "health_reason": "reachable",
                    "source": "config"
                }]
            })
        );
    }

    #[test]
    fn merge_notification_endpoints_keeps_configured_targets_after_runtime_loss() {
        let mut cfg_map = HashMap::new();
        cfg_map.insert(
            NOTIFY_WEBHOOK_SUB_SYS.to_string(),
            HashMap::from([("webhook-a".to_string(), enabled_kvs("on"))]),
        );
        cfg_map.insert(
            NOTIFY_MQTT_SUB_SYS.to_string(),
            HashMap::from([("mqtt-a".to_string(), enabled_kvs("on"))]),
        );
        let config = Config(cfg_map);

        let runtime = HashMap::from([(
            ("webhook-a".to_string(), "webhook".to_string()),
            RuntimeHealthStatus {
                status: "offline".to_string(),
                state: "error".to_string(),
                reason: "timed_out".to_string(),
            },
        )]);
        let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints");

        let mqtt = merged
            .iter()
            .find(|entry| entry.account_id == "mqtt-a" && entry.service == "mqtt")
            .expect("mqtt-a should be present");
        assert_eq!(mqtt.status, "offline");
        assert_eq!(mqtt.source, TargetEndpointSource::Config);

        let webhook = merged
            .iter()
            .find(|entry| entry.account_id == "webhook-a" && entry.service == "webhook")
            .expect("webhook-a should be present");
        assert_eq!(webhook.status, "offline");
        assert_eq!(webhook.health_state, "error");
        assert_eq!(webhook.health_reason, "timed_out");
        assert_eq!(webhook.source, TargetEndpointSource::Config);
    }

    #[test]
    fn merge_notification_endpoints_skips_disabled_and_default_entries() {
        let mut webhook_targets = HashMap::new();
        webhook_targets.insert(DEFAULT_DELIMITER.to_string(), enabled_kvs("on"));
        webhook_targets.insert("webhook-disabled".to_string(), enabled_kvs("off"));
        webhook_targets.insert("webhook-enabled".to_string(), enabled_kvs("on"));
        let config = Config(HashMap::from([(NOTIFY_WEBHOOK_SUB_SYS.to_string(), webhook_targets)]));

        let runtime = HashMap::from([
            (("webhook-enabled".to_string(), "webhook".to_string()), runtime_health("online")),
            (("env-only".to_string(), "mqtt".to_string()), runtime_health("offline")),
        ]);
        let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints");

        let env_only = merged
            .iter()
            .find(|entry| entry.account_id == "env-only" && entry.service == "mqtt")
            .expect("env-only should be present");
        assert_eq!(env_only.status, "offline");
        assert_eq!(env_only.source, TargetEndpointSource::Runtime);

        let enabled = merged
            .iter()
            .find(|entry| entry.account_id == "webhook-enabled" && entry.service == "webhook")
            .expect("webhook-enabled should be present");
        assert_eq!(enabled.status, "online");
        assert_eq!(enabled.source, TargetEndpointSource::Config);
    }

    #[test]
    #[serial]
    fn merge_notification_endpoints_marks_env_and_mixed_sources() {
        let config = Config(HashMap::from([
            (
                NOTIFY_WEBHOOK_SUB_SYS.to_string(),
                HashMap::from([("mixed-target".to_string(), enabled_kvs("on"))]),
            ),
            (
                NOTIFY_MQTT_SUB_SYS.to_string(),
                HashMap::from([("config-target".to_string(), enabled_kvs("on"))]),
            ),
        ]));

        with_vars(
            [
                ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_MIXED-TARGET", Some("https://example.com/hook")),
                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_ENV-ONLY", Some("on")),
                ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_ENV-ONLY", Some("https://example.com/env")),
            ],
            || {
                let runtime = HashMap::from([
                    (("mixed-target".to_string(), "webhook".to_string()), runtime_health("online")),
                    (("env-only".to_string(), "webhook".to_string()), runtime_health("online")),
                ]);
                let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints");

                let mixed = merged
                    .iter()
                    .find(|entry| entry.account_id == "mixed-target")
                    .expect("mixed target should be present");
                assert_eq!(mixed.source, TargetEndpointSource::Mixed);

                let env_only = merged
                    .iter()
                    .find(|entry| entry.account_id == "env-only")
                    .expect("env-only target should be present");
                assert_eq!(env_only.source, TargetEndpointSource::Env);

                let config_only = merged
                    .iter()
                    .find(|entry| entry.account_id == "config-target")
                    .expect("config target should be present");
                assert_eq!(config_only.source, TargetEndpointSource::Config);
            },
        );
    }

    #[test]
    #[serial]
    fn merge_notification_endpoints_marks_kafka_env_and_mixed_sources() {
        let config = Config(HashMap::from([(
            NOTIFY_KAFKA_SUB_SYS.to_string(),
            HashMap::from([("mixed-kafka".to_string(), enabled_kvs("on"))]),
        )]));

        with_vars(
            [
                ("RUSTFS_NOTIFY_KAFKA_ENABLE_MIXED-KAFKA", Some("on")),
                ("RUSTFS_NOTIFY_KAFKA_BROKERS_MIXED-KAFKA", Some("127.0.0.1:9092")),
                ("RUSTFS_NOTIFY_KAFKA_ENABLE_ENV-KAFKA", Some("on")),
                ("RUSTFS_NOTIFY_KAFKA_BROKERS_ENV-KAFKA", Some("127.0.0.1:9093")),
            ],
            || {
                let runtime = HashMap::from([
                    (("mixed-kafka".to_string(), "kafka".to_string()), runtime_health("online")),
                    (("env-kafka".to_string(), "kafka".to_string()), runtime_health("online")),
                ]);
                let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints");

                let mixed = merged
                    .iter()
                    .find(|entry| entry.account_id == "mixed-kafka" && entry.service == "kafka")
                    .expect("mixed kafka target should be present");
                assert_eq!(mixed.source, TargetEndpointSource::Mixed);

                let env_only = merged
                    .iter()
                    .find(|entry| entry.account_id == "env-kafka" && entry.service == "kafka")
                    .expect("env kafka target should be present");
                assert_eq!(env_only.source, TargetEndpointSource::Env);
            },
        );
    }

    #[test]
    #[serial]
    fn merge_notification_endpoints_marks_amqp_env_and_mixed_sources() {
        let config = Config(HashMap::from([(
            NOTIFY_AMQP_SUB_SYS.to_string(),
            HashMap::from([("mixed-amqp".to_string(), enabled_kvs("on"))]),
        )]));

        with_vars(
            [
                ("RUSTFS_NOTIFY_AMQP_ENABLE_MIXED-AMQP", Some("on")),
                ("RUSTFS_NOTIFY_AMQP_URL_MIXED-AMQP", Some("amqp://127.0.0.1:5672/%2f")),
                ("RUSTFS_NOTIFY_AMQP_ENABLE_ENV-AMQP", Some("on")),
                ("RUSTFS_NOTIFY_AMQP_URL_ENV-AMQP", Some("amqp://127.0.0.1:5672/%2f")),
            ],
            || {
                let runtime = HashMap::from([
                    (("mixed-amqp".to_string(), "amqp".to_string()), runtime_health("online")),
                    (("env-amqp".to_string(), "amqp".to_string()), runtime_health("online")),
                ]);
                let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints");

                let mixed = merged
                    .iter()
                    .find(|entry| entry.account_id == "mixed-amqp" && entry.service == "amqp")
                    .expect("mixed amqp target should be present");
                assert_eq!(mixed.source, TargetEndpointSource::Mixed);

                let env_only = merged
                    .iter()
                    .find(|entry| entry.account_id == "env-amqp" && entry.service == "amqp")
                    .expect("env amqp target should be present");
                assert_eq!(env_only.source, TargetEndpointSource::Env);
            },
        );
    }

    #[test]
    #[serial]
    fn target_mutation_block_reason_rejects_env_managed_target() {
        with_vars(
            [
                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY", Some("on")),
                ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("https://example.com/hook")),
            ],
            || {
                let config = Config(HashMap::new());
                let reason = target_mutation_block_reason(&config, NOTIFY_WEBHOOK_SUB_SYS, "primary")
                    .expect("target mutation block reason");
                assert!(reason.is_some());
                assert!(reason.unwrap().contains("managed by environment variables"));
            },
        );
    }

    #[test]
    #[serial]
    fn notification_target_operation_block_reason_requires_notify_module_enable() {
        with_var(rustfs_config::ENV_NOTIFY_ENABLE, Some("false"), || {
            let reason = futures::executor::block_on(notification_target_operation_block_reason(
                "managing notification targets from the console",
            ));
            assert!(reason.is_some());
            assert!(reason.unwrap().contains("set RUSTFS_NOTIFY_ENABLE=true"));
        });
    }

    #[test]
    #[serial]
    fn notification_target_operation_block_reason_allows_when_notify_module_enabled() {
        with_var(rustfs_config::ENV_NOTIFY_ENABLE, Some("true"), || {
            assert!(
                futures::executor::block_on(notification_target_operation_block_reason(
                    "managing notification targets from the console"
                ))
                .is_none()
            );
        });
    }

    #[test]
    #[serial]
    fn target_mutation_block_reason_rejects_mixed_target() {
        with_var("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("https://example.com/hook"), || {
            let config = Config(HashMap::from([(
                NOTIFY_WEBHOOK_SUB_SYS.to_string(),
                HashMap::from([("primary".to_string(), enabled_kvs("on"))]),
            )]));
            let reason =
                target_mutation_block_reason(&config, NOTIFY_WEBHOOK_SUB_SYS, "primary").expect("target mutation block reason");
            assert!(reason.is_some());
            assert!(reason.unwrap().contains("both persisted config and environment variables"));
        });
    }

    #[test]
    fn target_mutation_block_reason_allows_config_only_target() {
        let target_name = "config-only-target";
        let config = Config(HashMap::from([(
            NOTIFY_WEBHOOK_SUB_SYS.to_string(),
            HashMap::from([(target_name.to_string(), enabled_kvs("on"))]),
        )]));
        assert!(
            target_mutation_block_reason(&config, NOTIFY_WEBHOOK_SUB_SYS, target_name)
                .expect("target mutation block reason")
                .is_none()
        );
    }

    #[test]
    #[serial]
    fn merge_notification_endpoints_marks_disabled_config_with_env_override_as_mixed() {
        let config = Config(HashMap::from([(
            NOTIFY_WEBHOOK_SUB_SYS.to_string(),
            HashMap::from([("mixed-disabled".to_string(), enabled_kvs("off"))]),
        )]));

        with_vars(
            [
                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_MIXED-DISABLED", Some("on")),
                ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_MIXED-DISABLED", Some("https://example.com/hook")),
            ],
            || {
                let merged = merge_notification_endpoints(&config, HashMap::new()).expect("merge notification endpoints");
                let mixed = merged
                    .iter()
                    .find(|entry| entry.account_id == "mixed-disabled")
                    .expect("mixed target should be present");
                assert_eq!(mixed.source, TargetEndpointSource::Mixed);
                assert_eq!(mixed.status, "offline");
            },
        );
    }

    #[test]
    #[serial]
    fn merge_notification_endpoints_includes_env_only_target_without_runtime_status() {
        let config = Config(HashMap::new());

        with_vars(
            [
                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_ENV-ONLY", Some("on")),
                ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_ENV-ONLY", Some("https://example.com/env")),
            ],
            || {
                let merged = merge_notification_endpoints(&config, HashMap::new()).expect("merge notification endpoints");
                let env_only = merged
                    .iter()
                    .find(|entry| entry.account_id == "env-only")
                    .expect("env-only target should be present");
                assert_eq!(env_only.source, TargetEndpointSource::Env);
                assert_eq!(env_only.status, "offline");
            },
        );
    }

    #[test]
    fn collect_validated_key_values_rejects_duplicate_keys() {
        let allowed_keys: HashSet<&str> = ["endpoint", "auth_token"].into_iter().collect();
        let key_values = [
            KeyValue {
                key: "endpoint".to_string(),
                value: "https://example.com/one".to_string(),
            },
            KeyValue {
                key: "endpoint".to_string(),
                value: "https://example.com/two".to_string(),
            },
        ];

        let err = shared_collect_validated_key_values(
            key_values.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())),
            &allowed_keys,
            NOTIFY_WEBHOOK_SUB_SYS,
            "target",
        )
        .unwrap_err();
        assert!(err.to_string().contains("duplicate key"));
    }

    #[test]
    #[serial]
    fn merge_notification_endpoints_marks_mixed_with_case_insensitive_instance_id() {
        let config = Config(HashMap::from([(
            NOTIFY_WEBHOOK_SUB_SYS.to_string(),
            HashMap::from([("PrimaryCase".to_string(), enabled_kvs("on"))]),
        )]));

        with_vars(
            [
                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARYCASE", Some("on")),
                ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARYCASE", Some("https://example.com/hook")),
            ],
            || {
                let runtime = HashMap::from([(("PrimaryCase".to_string(), "webhook".to_string()), runtime_health("online"))]);
                let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints");
                let mixed = merged
                    .iter()
                    .find(|entry| entry.account_id == "PrimaryCase" && entry.service == "webhook")
                    .expect("mixed target should be present");
                assert_eq!(mixed.source, TargetEndpointSource::Mixed);
            },
        );
    }

    #[test]
    fn collect_online_target_arns_filters_offline_targets() {
        let arns = collect_online_target_arns(
            "us-east-1",
            vec![
                (TargetID::new("webhook-a".to_string(), "webhook".to_string()), "online".to_string()),
                (TargetID::new("mqtt-a".to_string(), "mqtt".to_string()), "offline".to_string()),
            ],
        );

        assert_eq!(arns, vec!["arn:rustfs:sqs:us-east-1:webhook-a:webhook".to_string()]);
    }

    #[test]
    #[serial]
    fn target_mutation_block_reason_allows_case_insensitive_config_target_lookup() {
        let config = Config(HashMap::from([(
            NOTIFY_WEBHOOK_SUB_SYS.to_string(),
            HashMap::from([("PrimaryCase".to_string(), enabled_kvs("on"))]),
        )]));

        with_vars(
            [
                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARYCASE", None::<&str>),
                ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARYCASE", None::<&str>),
            ],
            || {
                assert!(
                    target_mutation_block_reason(&config, NOTIFY_WEBHOOK_SUB_SYS, "primarycase")
                        .expect("target mutation block reason")
                        .is_none()
                );
            },
        );
    }

    /// These endpoints authorize through the shared admin gate, which reports
    /// "get cred failed" for a credential-less request. The pre-check keeps the
    /// message they have always returned (rustfs/backlog#1829).
    #[tokio::test]
    async fn notification_target_gate_keeps_its_missing_credentials_message() {
        let req = S3Request {
            input: Body::from(String::new()),
            method: Method::PUT,
            uri: http::Uri::from_static("/rustfs/admin/v3/notification/target"),
            headers: http::HeaderMap::new(),
            extensions: http::Extensions::new(),
            credentials: None,
            region: None,
            service: None,
            trailing_headers: None,
        };

        let err = authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction)
            .await
            .expect_err("a request without credentials must be rejected");
        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
        assert_eq!(err.message(), Some("credentials not found"));
    }

    #[test]
    fn notification_target_handlers_require_admin_authorization_contract() {
        let src = include_str!("event.rs");
        let put_block =
            extract_block_between_markers(src, "impl Operation for NotificationTarget", "pub struct ListNotificationTargets");
        let list_block =
            extract_block_between_markers(src, "impl Operation for ListNotificationTargets", "pub struct ListTargetsArns");
        let arns_block =
            extract_block_between_markers(src, "impl Operation for ListTargetsArns", "pub struct RemoveNotificationTarget");
        let delete_block = extract_block_between_markers(src, "impl Operation for RemoveNotificationTarget", "fn extract_param");

        for block in [put_block, list_block, arns_block, delete_block] {
            assert!(
                !block.contains(".enter()"),
                "async notification handlers must rely on request-future instrumentation instead of holding span guards across awaits"
            );
        }

        assert!(
            put_block.contains("authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"),
            "notification target writes should require SetBucketTargetAction"
        );
        assert!(
            put_block.contains("notification_target_operation_block_reason(")
                && put_block.contains("\"managing notification targets from the console\""),
            "notification target writes should reject requests when the notify module is disabled"
        );
        assert!(
            list_block.contains("authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;"),
            "notification target list should require GetBucketTargetAction"
        );
        let authorize_index = list_block
            .find("authorize_notification_admin_request")
            .expect("target list should authorize the request");
        let refresh_index = list_block
            .find("refresh_persisted_module_switches_from_store().await.map_err")
            .expect("target list should fail when persisted module state cannot be refreshed");
        let effective_state_index = list_block
            .find("let notify_enabled = refresh_notify_module_enabled();")
            .expect("target list should resolve the effective env and persisted module state");
        let load_index = list_block
            .find("load_notification_config_snapshot().await?")
            .expect("target list should load notification config");
        assert!(
            authorize_index < refresh_index && refresh_index < effective_state_index && effective_state_index < load_index,
            "target list should authorize, refresh persisted state, resolve effective state, then load targets"
        );
        assert!(
            arns_block.contains("authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;"),
            "notification target arn listing should require GetBucketTargetAction"
        );
        assert!(
            arns_block.contains("notification_target_operation_block_reason(")
                && arns_block.contains("\"querying notification target ARNs for bucket associations from the console\""),
            "notification target arn listing should reject requests when the notify module is disabled"
        );
        assert!(
            delete_block.contains("authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"),
            "notification target deletion should require SetBucketTargetAction"
        );
        assert!(
            delete_block.contains("notification_target_operation_block_reason(")
                && delete_block.contains("\"managing notification targets from the console\""),
            "notification target deletion should reject requests when the notify module is disabled"
        );
    }

    #[test]
    fn extract_target_params_accepts_kafka_target_type() {
        let mut router = Router::new();
        router
            .insert("/v3/target/{target_type}/{target_name}", ())
            .expect("route should insert");

        let params = router
            .at("/v3/target/notify_kafka/streaming")
            .expect("route should match")
            .params;
        let (target_type, target_name) = extract_target_params(&params).expect("kafka target type should be accepted");
        assert_eq!(target_type, NOTIFY_KAFKA_SUB_SYS);
        assert_eq!(target_name, "streaming");
    }

    #[test]
    fn extract_target_params_accepts_amqp_target_type() {
        let mut router = Router::new();
        router
            .insert("/v3/target/{target_type}/{target_name}", ())
            .expect("route should insert");

        let params = router
            .at("/v3/target/notify_amqp/rabbitmq")
            .expect("route should match")
            .params;
        let (target_type, target_name) = extract_target_params(&params).expect("amqp target type should be accepted");
        assert_eq!(target_type, NOTIFY_AMQP_SUB_SYS);
        assert_eq!(target_name, "rabbitmq");
    }

    #[test]
    fn collect_validated_key_values_accepts_amqp_keys() {
        let specs = notification_target_specs();
        let allowed_keys = allowed_target_keys(specs, NOTIFY_AMQP_SUB_SYS);

        let kv_map = shared_collect_validated_key_values(
            [
                (rustfs_config::AMQP_URL, "amqp://127.0.0.1:5672/%2f"),
                (rustfs_config::AMQP_EXCHANGE, "rustfs.events"),
                (rustfs_config::AMQP_ROUTING_KEY, "objects"),
            ],
            &allowed_keys,
            NOTIFY_AMQP_SUB_SYS,
            "target",
        )
        .expect("amqp keys should be accepted");

        assert_eq!(kv_map.get(rustfs_config::AMQP_URL).map(String::as_str), Some("amqp://127.0.0.1:5672/%2f"));
        assert!(allowed_keys.contains(rustfs_config::AMQP_MANDATORY));
        assert!(allowed_keys.contains(rustfs_config::AMQP_PERSISTENT));
    }

    fn extract_block_between_markers<'a>(src: &'a str, start_marker: &str, end_marker: &str) -> &'a str {
        let start = src
            .find(start_marker)
            .unwrap_or_else(|| panic!("Expected marker `{start_marker}` in source"));
        let after_start = &src[start..];
        let end = after_start
            .find(end_marker)
            .unwrap_or_else(|| panic!("Expected end marker `{end_marker}` in source"));
        &after_start[..end]
    }
}
