hammerwork 3.0.0

//! Webhook system for delivering job lifecycle events to external systems.
//!
//! This module provides a robust webhook delivery system for job lifecycle events.
//! It includes retry policies, failure handling, and configurable delivery options.
//! This is separate from the alerting webhook system and focuses on real-time
//! event delivery to external integrations.
//!
//! # Overview
//!
//! The webhook system consists of several key components:
//!
//! - [`WebhookManager`] - Manages webhook configurations and delivery
//! - [`WebhookConfig`] - Configuration for individual webhook endpoints
//! - [`RetryPolicy`] - Configurable retry behavior for failed deliveries
//! - [`WebhookAuth`] - Authentication methods (Bearer, Basic, API Key, Custom)
//! - [`WebhookDelivery`] - Results and metadata from delivery attempts
//!
//! # Features
//!
//! - **Multiple authentication methods** - Bearer tokens, Basic auth, API keys, custom headers
//! - **HMAC signatures** - Secure webhook verification with SHA-256 HMAC
//! - **Flexible retry policies** - Exponential backoff with configurable parameters
//! - **Event filtering** - Only deliver events matching specific criteria
//! - **Delivery tracking** - Comprehensive statistics and delivery history
//! - **Rate limiting** - Configurable concurrent delivery limits
//! - **Custom payload templates** - Reshape events into the JSON a receiver expects
//!   (see [`PayloadTemplate`])
//!
//! # Examples
//!
//! ## Basic Webhook Setup
//!
//! ```rust
//! use hammerwork::webhooks::{WebhookManager, WebhookConfig, HttpMethod, WebhookManagerConfig};
//! use hammerwork::events::{EventManager, EventFilter, JobLifecycleEventType};
//! use std::sync::Arc;
//! use std::collections::HashMap;
//!
//! # #[tokio::main]
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let event_manager = Arc::new(EventManager::new_default());
//! let webhook_manager = WebhookManager::new(
//!     event_manager.clone(),
//!     WebhookManagerConfig::default()
//! )?;
//!
//! let webhook = WebhookConfig::new(
//!     "job_completion_hook".to_string(),
//!     "https://api.example.com/webhooks/jobs".to_string(),
//! )
//! .with_method(HttpMethod::Post)
//! .with_header("Content-Type".to_string(), "application/json".to_string())
//! .with_filter(
//!     EventFilter::new()
//!         .with_event_types(vec![JobLifecycleEventType::Completed])
//! );
//!
//! webhook_manager.add_webhook(webhook).await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Webhook with Authentication and HMAC
//!
//! ```rust
//! use hammerwork::webhooks::{WebhookConfig, WebhookAuth, HttpMethod};
//! use hammerwork::events::EventFilter;
//!
//! let webhook = WebhookConfig::new(
//!     "secure_webhook".to_string(),
//!     "https://secure-api.example.com/events".to_string(),
//! )
//! .with_auth(WebhookAuth::Bearer {
//!     token: "your-bearer-token".to_string()
//! })
//! .with_secret("webhook-signing-secret".to_string())
//! .with_timeout_secs(30);
//! ```
//!
//! ## Custom Retry Policy
//!
//! ```rust
//! use hammerwork::webhooks::{WebhookConfig, RetryPolicy};
//!
//! let webhook = WebhookConfig::new(
//!     "reliable_webhook".to_string(),
//!     "https://api.example.com/events".to_string(),
//! )
//! .with_retry_policy(RetryPolicy::default()
//!     .with_max_attempts(5)
//!     .with_initial_delay_secs(2)
//!     .with_max_delay_secs(300)
//!     .with_backoff_multiplier(2.5)
//!     .with_retry_on_status_codes(vec![500, 502, 503, 504, 429]));
//! ```

use crate::events::{EventFilter, EventManager, EventSubscription, JobLifecycleEvent};
use crate::task_tracker::{ListenerHandle, TaskTracker, clamp_permits};
use chrono::{DateTime, Utc};
use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::{collections::HashMap, sync::Arc, time::Duration};
use tokio::{
    sync::{OwnedSemaphorePermit, RwLock, Semaphore, broadcast, oneshot},
    time::{Instant, sleep, timeout_at},
};
use uuid::Uuid;

/// Module for serializing UUID as string for TOML compatibility
mod uuid_string {
    use serde::{Deserialize, Deserializer, Serializer};
    use uuid::Uuid;

    pub fn serialize<S>(uuid: &Uuid, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&uuid.to_string())
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<Uuid, D::Error>
    where
        D: Deserializer<'de>,
    {
        use serde::de::Error;
        let s = String::deserialize(deserializer)?;
        Uuid::parse_str(&s).map_err(D::Error::custom)
    }
}

/// Configuration for a webhook endpoint.
///
/// WebhookConfig defines how events should be delivered to a specific webhook endpoint.
/// It includes authentication, retry behavior, filtering criteria, and delivery options.
///
/// # Examples
///
/// ```rust
/// use hammerwork::webhooks::{WebhookConfig, WebhookAuth, HttpMethod, RetryPolicy};
/// use hammerwork::events::{EventFilter, JobLifecycleEventType};
/// use hammerwork::priority::JobPriority;
/// use std::collections::HashMap;
///
/// // Basic webhook configuration
/// let webhook = WebhookConfig::new(
///     "payment_webhook".to_string(),
///     "https://api.example.com/webhooks/payments".to_string(),
/// )
/// .with_method(HttpMethod::Post)
/// .with_header("Content-Type".to_string(), "application/json".to_string())
/// .with_header("X-API-Version".to_string(), "v1".to_string());
///
/// // Webhook with authentication and filtering
/// let secure_webhook = WebhookConfig::new(
///     "critical_alerts".to_string(),
///     "https://alerts.example.com/webhook".to_string(),
/// )
/// .with_auth(WebhookAuth::ApiKey {
///     header_name: "X-API-Key".to_string(),
///     api_key: "your-api-key".to_string(),
/// })
/// .with_filter(
///     EventFilter::new()
///         .with_event_types(vec![JobLifecycleEventType::Failed, JobLifecycleEventType::Dead])
///         .with_priorities(vec![JobPriority::Critical])
/// )
/// .with_secret("hmac-signing-secret".to_string())
/// .with_timeout_secs(15);
/// ```
///
/// `Debug` never shows the signing secret, header values, credentials, or the path of
/// the URL (webhook URLs of chat services are themselves credentials).
#[derive(Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct WebhookConfig {
    /// Unique identifier for this webhook
    #[serde(with = "uuid_string")]
    pub id: Uuid,
    /// Human-readable name for this webhook
    pub name: String,
    /// URL to deliver events to
    pub url: String,
    /// HTTP method to use (typically POST)
    pub method: HttpMethod,
    /// Custom headers to include in requests
    pub headers: HashMap<String, String>,
    /// Event filter to determine which events to deliver
    pub filter: EventFilter,
    /// Retry policy for failed deliveries
    pub retry_policy: RetryPolicy,
    /// Authentication configuration
    pub auth: Option<WebhookAuth>,
    /// Timeout for one delivery attempt, in seconds: sending the request and reading
    /// the response, body included. `0` (also the value when a configuration file
    /// leaves it out) uses the manager's
    /// [`default_timeout_secs`](WebhookManagerConfig::default_timeout_secs).
    #[serde(default)]
    pub timeout_secs: u64,
    /// Whether this webhook is currently enabled
    pub enabled: bool,
    /// Secret for HMAC signature verification (optional)
    pub secret: Option<String>,
    /// Custom payload template (optional).
    ///
    /// A JSON document with `{{ event.* }}` placeholders; see [`PayloadTemplate`] for
    /// the syntax. When set, the rendered template is sent instead of the raw event.
    /// It is validated when the webhook is added.
    pub payload_template: Option<String>,
}

impl Default for WebhookConfig {
    fn default() -> Self {
        Self {
            id: Uuid::new_v4(),
            name: "Default Webhook".to_string(),
            url: "https://example.com/webhook".to_string(),
            method: HttpMethod::Post,
            headers: HashMap::new(),
            filter: EventFilter::default(),
            retry_policy: RetryPolicy::default(),
            auth: None,
            timeout_secs: 30,
            enabled: true,
            secret: None,
            payload_template: None,
        }
    }
}

impl std::fmt::Debug for WebhookConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut header_names: Vec<&String> = self.headers.keys().collect();
        header_names.sort();
        f.debug_struct("WebhookConfig")
            .field("id", &self.id)
            .field("name", &self.name)
            .field("url", &crate::config::redact_url_path(&self.url))
            .field("method", &self.method)
            .field("headers", &header_names)
            .field("filter", &self.filter)
            .field("retry_policy", &self.retry_policy)
            .field("auth", &self.auth)
            .field("timeout_secs", &self.timeout_secs)
            .field("enabled", &self.enabled)
            .field("secret", &self.secret.as_ref().map(|_| "[REDACTED]"))
            .field("payload_template", &self.payload_template)
            .finish()
    }
}

/// HTTP methods supported for webhooks
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "UPPERCASE")]
#[non_exhaustive]
pub enum HttpMethod {
    Post,
    Put,
    Patch,
}

impl std::fmt::Display for HttpMethod {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            HttpMethod::Post => write!(f, "POST"),
            HttpMethod::Put => write!(f, "PUT"),
            HttpMethod::Patch => write!(f, "PATCH"),
        }
    }
}

/// Authentication methods for webhooks
///
/// `Debug` shows the kind of authentication and header names, never tokens, passwords,
/// API keys or header values.
#[derive(Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
#[non_exhaustive]
pub enum WebhookAuth {
    /// Bearer token authentication
    Bearer { token: String },
    /// Basic authentication
    Basic { username: String, password: String },
    /// API key in header
    ApiKey {
        header_name: String,
        api_key: String,
    },
    /// Custom header authentication
    Custom { headers: HashMap<String, String> },
}

impl std::fmt::Debug for WebhookAuth {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        const REDACTED: &str = "[REDACTED]";
        match self {
            WebhookAuth::Bearer { .. } => {
                f.debug_struct("Bearer").field("token", &REDACTED).finish()
            }
            WebhookAuth::Basic { username, .. } => f
                .debug_struct("Basic")
                .field("username", username)
                .field("password", &REDACTED)
                .finish(),
            WebhookAuth::ApiKey { header_name, .. } => f
                .debug_struct("ApiKey")
                .field("header_name", header_name)
                .field("api_key", &REDACTED)
                .finish(),
            WebhookAuth::Custom { headers } => {
                let mut names: Vec<&String> = headers.keys().collect();
                names.sort();
                f.debug_struct("Custom").field("headers", &names).finish()
            }
        }
    }
}

/// Retry policy configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct RetryPolicy {
    /// Maximum number of delivery attempts, including the first (0 is treated as 1:
    /// every event is attempted at least once)
    pub max_attempts: u32,
    /// Initial delay between retries (in seconds)
    pub initial_delay_secs: u64,
    /// Maximum delay between retries (in seconds)
    pub max_delay_secs: u64,
    /// Multiplier for exponential backoff
    pub backoff_multiplier: f64,
    /// Whether to retry on specific HTTP status codes
    pub retry_on_status_codes: Vec<u16>,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            initial_delay_secs: 1,
            max_delay_secs: 300, // 5 minutes
            backoff_multiplier: 2.0,
            retry_on_status_codes: vec![408, 429, 500, 502, 503, 504],
        }
    }
}

impl RetryPolicy {
    /// Calculate delay for a specific retry attempt
    pub fn calculate_delay(&self, attempt: u32) -> Duration {
        let delay_secs = (self.initial_delay_secs as f64)
            * self
                .backoff_multiplier
                .powi(attempt.min(i32::MAX as u32) as i32);
        let delay_secs = delay_secs.min(self.max_delay_secs as f64);
        Duration::from_secs(delay_secs as u64)
    }

    /// Check if we should retry based on status code
    pub fn should_retry_status(&self, status_code: u16) -> bool {
        self.retry_on_status_codes.contains(&status_code)
    }
}

/// Webhook delivery attempt result
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct WebhookDelivery {
    /// Unique identifier for this delivery attempt
    #[serde(with = "uuid_string")]
    pub delivery_id: Uuid,
    /// Webhook configuration ID
    #[serde(with = "uuid_string")]
    pub webhook_id: Uuid,
    /// Event that was delivered
    #[serde(with = "uuid_string")]
    pub event_id: Uuid,
    /// HTTP status code received
    pub status_code: Option<u16>,
    /// Response body (truncated if too long)
    pub response_body: Option<String>,
    /// Error message if delivery failed
    pub error_message: Option<String>,
    /// When the delivery was attempted
    pub attempted_at: DateTime<Utc>,
    /// How long the request took (in milliseconds)
    pub duration_ms: Option<u64>,
    /// Attempt number (1-based)
    pub attempt_number: u32,
    /// Whether this delivery was successful
    pub success: bool,
}

/// Webhook delivery statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct WebhookStats {
    /// Webhook configuration ID
    #[serde(with = "uuid_string")]
    pub webhook_id: Uuid,
    /// Total number of HTTP delivery attempts, retries included: one event can take
    /// several attempts.
    pub total_attempts: u64,
    /// Number of HTTP delivery attempts that failed (an error, a timeout or a non-2xx
    /// response), retried ones included. `total_attempts - failed_attempts` attempts
    /// succeeded. Compare [`failed_deliveries`](Self::failed_deliveries), which counts
    /// events whose delivery failed for good.
    #[serde(default)]
    pub failed_attempts: u64,
    /// Number of events delivered successfully (by any attempt)
    pub successful_deliveries: u64,
    /// Number of events whose delivery ultimately failed (retries exhausted, or a
    /// non-retryable response)
    pub failed_deliveries: u64,
    /// Current success rate (0.0 to 1.0): `successful_deliveries` out of the events
    /// whose delivery has finished (`successful_deliveries + failed_deliveries`)
    pub success_rate: f64,
    /// Average response time in milliseconds
    pub avg_response_time_ms: f64,
    /// Last successful delivery timestamp
    pub last_success_at: Option<DateTime<Utc>>,
    /// Last failure timestamp
    pub last_failure_at: Option<DateTime<Utc>>,
    /// Events this webhook never attempted because its listener fell behind: while
    /// [`max_pending_deliveries`](WebhookManagerConfig::max_pending_deliveries)
    /// deliveries are pending the listener waits, and events that overflow the event
    /// manager's buffer in the meantime are dropped (oldest first).
    #[serde(default)]
    pub dropped_events: u64,
    /// Time window these statistics cover
    pub calculated_at: DateTime<Utc>,
}

impl WebhookStats {
    /// Statistics of a webhook with no deliveries yet.
    pub(crate) fn empty(webhook_id: Uuid) -> Self {
        Self {
            webhook_id,
            total_attempts: 0,
            failed_attempts: 0,
            successful_deliveries: 0,
            failed_deliveries: 0,
            success_rate: 0.0,
            avg_response_time_ms: 0.0,
            last_success_at: None,
            last_failure_at: None,
            dropped_events: 0,
            calculated_at: Utc::now(),
        }
    }
}

/// Overall webhook manager statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct WebhookManagerStats {
    /// Total number of configured webhooks
    pub total_webhooks: usize,
    /// Number of active/enabled webhooks
    pub active_webhooks: usize,
    /// Total HTTP delivery attempts across all webhooks, retries included
    pub total_attempts: u64,
    /// Total failed HTTP delivery attempts across all webhooks, retries included
    #[serde(default)]
    pub failed_attempts: u64,
    /// Total events delivered successfully across all webhooks
    pub successful_deliveries: u64,
    /// Total events whose delivery ultimately failed across all webhooks
    pub failed_deliveries: u64,
}

/// Webhook delivery system for managing and delivering job lifecycle events.
///
/// WebhookManager is the central component for webhook delivery. It manages webhook
/// configurations, handles event subscriptions, delivers events to configured endpoints,
/// and provides statistics about delivery performance.
///
/// # Features
///
/// - **Concurrent delivery** - Configurable concurrent webhook delivery limits
/// - **Automatic retries** - Exponential backoff with configurable retry policies
/// - **Authentication support** - Bearer tokens, Basic auth, API keys, and custom headers
/// - **HMAC signatures** - Secure webhook verification with SHA-256 HMAC
/// - **Event filtering** - Only deliver events matching webhook-specific criteria
/// - **Statistics tracking** - Comprehensive delivery metrics and success rates
/// - **Rate limiting** - Prevents overwhelming webhook endpoints
///
/// # Examples
///
/// ```rust
/// use hammerwork::webhooks::{WebhookManager, WebhookConfig, WebhookManagerConfig, HttpMethod};
/// use hammerwork::events::{EventManager, EventFilter, JobLifecycleEventType};
/// use std::sync::Arc;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// // Create webhook manager
/// let event_manager = Arc::new(EventManager::new_default());
/// let config = WebhookManagerConfig::default()
///     .with_max_concurrent_deliveries(20)
///     .with_max_response_body_size(64 * 1024)
///     .with_log_deliveries(true)
///     .with_default_timeout_secs(30)
///     .with_max_pending_deliveries(1_000)
///     .with_user_agent("MyApp/1.0".to_string());
/// let webhook_manager = WebhookManager::new(event_manager, config)?;
///
/// // Add a webhook
/// let webhook = WebhookConfig::new(
///     "completion_notifier".to_string(),
///     "https://api.example.com/job-completed".to_string(),
/// )
/// .with_filter(
///     EventFilter::new()
///         .with_event_types(vec![JobLifecycleEventType::Completed])
/// );
///
/// webhook_manager.add_webhook(webhook).await?;
///
/// // Get delivery statistics
/// let stats = webhook_manager.get_stats().await;
/// println!("Active webhooks: {}", stats.active_webhooks);
/// # Ok(())
/// # }
/// ```
pub struct WebhookManager {
    /// HTTP client for making requests
    http_client: Client,
    /// Active webhook configurations
    webhooks: Arc<RwLock<HashMap<Uuid, WebhookConfig>>>,
    /// The running listener of each webhook
    subscriptions: Arc<RwLock<HashMap<Uuid, ListenerHandle>>>,
    /// Event manager for subscribing to events
    event_manager: Arc<EventManager>,
    /// Rate limiting semaphore for concurrent deliveries
    delivery_semaphore: Arc<Semaphore>,
    /// Delivery statistics
    stats: Arc<RwLock<HashMap<Uuid, WebhookStats>>>,
    /// Configuration
    config: WebhookManagerConfig,
    /// Long-lived per-webhook event listener tasks
    listeners: TaskTracker,
    /// In-flight delivery tasks (including their retries)
    deliveries: TaskTracker,
}

/// Configuration for the webhook manager
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct WebhookManagerConfig {
    /// Maximum number of HTTP requests in flight at once, across all webhooks.
    /// A delivery holds a slot only while a request is in flight, not while it waits
    /// to retry.
    pub max_concurrent_deliveries: usize,
    /// Maximum response body size to read and store (in bytes); the rest of a
    /// longer body is not read.
    pub max_response_body_size: usize,
    /// Whether to log webhook deliveries
    pub log_deliveries: bool,
    /// Timeout, in seconds, for webhooks whose
    /// [`timeout_secs`](WebhookConfig::timeout_secs) is `0`.
    pub default_timeout_secs: u64,
    /// Maximum number of deliveries pending per webhook: waiting for a request slot,
    /// in flight, or waiting to retry.
    ///
    /// When a webhook reaches it (for example because its endpoint is down and every
    /// delivery is retrying), its listener stops taking new events until a delivery
    /// finishes. Events keep arriving in the [`EventManager`]'s broadcast buffer
    /// (`EventConfig::max_buffer_size`); once that overflows, the oldest events are
    /// dropped for this webhook, logged at `warn` and counted in
    /// [`WebhookStats::dropped_events`]. Memory use stays bounded however long the
    /// endpoint is down. Values below 1 are treated as 1.
    pub max_pending_deliveries: usize,
    /// User agent string for webhook requests
    pub user_agent: String,
}

impl Default for WebhookManagerConfig {
    fn default() -> Self {
        Self {
            max_concurrent_deliveries: 100,
            max_response_body_size: 64 * 1024, // 64KB
            log_deliveries: true,
            default_timeout_secs: 30,
            max_pending_deliveries: 1_000,
            user_agent: format!("hammerwork-webhooks/{}", env!("CARGO_PKG_VERSION")),
        }
    }
}

impl From<&crate::config::WebhookGlobalSettings> for WebhookManagerConfig {
    fn from(settings: &crate::config::WebhookGlobalSettings) -> Self {
        Self {
            max_concurrent_deliveries: settings.max_concurrent_deliveries,
            max_response_body_size: settings.max_response_body_size,
            log_deliveries: settings.log_deliveries,
            default_timeout_secs: settings.default_timeout_secs,
            max_pending_deliveries: settings.max_pending_deliveries,
            user_agent: settings.user_agent.clone(),
        }
    }
}

/// Clamp a configured delivery concurrency into the range `Semaphore` accepts.
///
/// `0` permits would deadlock every delivery task, and more than
/// `Semaphore::MAX_PERMITS` panics inside tokio.
fn clamp_delivery_permits(configured: usize) -> usize {
    clamp_permits(configured, "max_concurrent_deliveries")
}

/// The time allowed for one delivery attempt of `webhook`.
fn attempt_timeout(webhook: &WebhookConfig, config: &WebhookManagerConfig) -> Duration {
    let secs = if webhook.timeout_secs == 0 {
        config.default_timeout_secs
    } else {
        webhook.timeout_secs
    };
    Duration::from_secs(secs)
}

/// Read at most `max_bytes` of `response`'s body. Stops reading (and drops the
/// connection) once the limit is reached, so a huge or endless body costs nothing more.
async fn read_body_limited(
    mut response: reqwest::Response,
    max_bytes: usize,
) -> reqwest::Result<String> {
    let mut body = Vec::new();
    let mut truncated = false;
    while let Some(chunk) = response.chunk().await? {
        let room = max_bytes.saturating_sub(body.len());
        if chunk.len() > room {
            body.extend_from_slice(&chunk[..room]);
            truncated = true;
            break;
        }
        body.extend_from_slice(&chunk);
    }
    Ok(response_body_text(body, truncated))
}

/// The response body as text, marked when it was cut off.
///
/// The cut can fall inside a multi-byte UTF-8 character; the incomplete character is
/// dropped rather than replaced. Invalid UTF-8 elsewhere becomes U+FFFD.
fn response_body_text(mut body: Vec<u8>, truncated: bool) -> String {
    if truncated
        && let Err(e) = std::str::from_utf8(&body)
        && e.error_len().is_none()
    {
        body.truncate(e.valid_up_to());
    }
    let mut text = String::from_utf8_lossy(&body).into_owned();
    if truncated {
        text.push_str("... [truncated]");
    }
    text
}

impl WebhookManager {
    /// Create a new webhook manager.
    ///
    /// Fails with [`HammerworkError::Webhook`](crate::HammerworkError::Webhook) when the
    /// HTTP client cannot be built (for example when `config.user_agent` is not a valid
    /// header value, or the TLS backend fails to initialise).
    ///
    /// `max_concurrent_deliveries` is clamped to `1..=Semaphore::MAX_PERMITS`: `0`
    /// would make every delivery wait forever and a huge value would panic.
    pub fn new(
        event_manager: Arc<EventManager>,
        config: WebhookManagerConfig,
    ) -> crate::Result<Self> {
        let http_client = Client::builder()
            .user_agent(&config.user_agent)
            .build()
            .map_err(|e| crate::error::HammerworkError::Webhook {
                message: format!("Failed to create webhook HTTP client: {e}"),
            })?;

        let permits = clamp_delivery_permits(config.max_concurrent_deliveries);

        Ok(Self {
            http_client,
            webhooks: Arc::new(RwLock::new(HashMap::new())),
            subscriptions: Arc::new(RwLock::new(HashMap::new())),
            event_manager,
            delivery_semaphore: Arc::new(Semaphore::new(permits)),
            stats: Arc::new(RwLock::new(HashMap::new())),
            config,
            listeners: TaskTracker::new("webhook listener"),
            deliveries: TaskTracker::new("webhook delivery"),
        })
    }

    /// Stop the manager and wait for in-flight deliveries.
    ///
    /// Stops every webhook's event listener (no new events are picked up), then waits
    /// up to `grace` for deliveries that are already running, including their retries.
    /// Deliveries still running after `grace` are aborted. Returns the number aborted.
    ///
    /// After shutdown the manager no longer starts delivery tasks; webhooks added
    /// later are registered but never delivered.
    pub async fn shutdown(&self, grace: Duration) -> usize {
        self.listeners.close();
        self.listeners.abort_all().await;
        // The listeners are stopped: take their subscriptions off the event manager too.
        let handles: Vec<_> = self.subscriptions.write().await.drain().collect();
        for (_, handle) in handles {
            if let Err(e) = self.event_manager.unsubscribe(handle.subscription_id).await {
                tracing::warn!("Failed to unsubscribe a listener during shutdown: {e}");
            }
        }
        let aborted = self.deliveries.drain(grace).await;

        if self.config.log_deliveries {
            tracing::info!("Webhook manager shut down ({aborted} deliveries aborted)");
        }
        aborted
    }

    /// Number of background webhook tasks (listeners or deliveries) that have panicked.
    ///
    /// Panics are also logged at error level when they happen.
    pub fn task_panics(&self) -> u64 {
        self.listeners.panic_count() + self.deliveries.panic_count()
    }

    /// Create a new webhook manager with default configuration. Fails like
    /// [`new`](Self::new).
    pub fn new_default(event_manager: Arc<EventManager>) -> crate::Result<Self> {
        Self::new(event_manager, WebhookManagerConfig::default())
    }

    /// Create a webhook manager from the `webhooks` section of a
    /// [`HammerworkConfig`](crate::HammerworkConfig), registering every configured webhook.
    pub async fn from_config(
        event_manager: Arc<EventManager>,
        config: &crate::config::WebhookConfigs,
    ) -> crate::Result<Self> {
        let manager = Self::new(event_manager, (&config.global_settings).into())?;
        for webhook in &config.webhooks {
            manager.add_webhook(webhook.clone()).await?;
        }
        Ok(manager)
    }

    /// Add a new webhook configuration.
    ///
    /// The webhook receives every matching event published after this returns.
    /// Adding a webhook with the id of an existing one replaces it (and resets its
    /// statistics); use [`update_webhook`](Self::update_webhook) to change a webhook
    /// without a gap in delivery.
    ///
    /// Fails with a configuration error when the webhook is invalid (see
    /// [`WebhookConfig::validate`]), for example an invalid payload template.
    pub async fn add_webhook(&self, webhook: WebhookConfig) -> crate::Result<()> {
        webhook.validate()?;
        let webhook_id = webhook.id;

        // Subscribe before returning, and hand this receiver to the listener: events
        // published from now on are buffered for it even before its task first runs.
        let subscription = self.event_manager.subscribe(webhook.filter.clone()).await?;
        let (handle, stop) = ListenerHandle::new(subscription.id);

        {
            let mut webhooks = self.webhooks.write().await;
            webhooks.insert(webhook_id, webhook.clone());
        }

        // Initialize statistics
        {
            let mut stats = self.stats.write().await;
            stats.insert(webhook_id, WebhookStats::empty(webhook_id));
        }

        // Replacing the handle of an existing listener stops that listener.
        let replaced = self.subscriptions.write().await.insert(webhook_id, handle);
        if let Some(replaced) = replaced {
            self.event_manager
                .unsubscribe(replaced.subscription_id)
                .await?;
        }

        self.start_webhook_delivery_task(webhook_id, subscription, stop);

        if self.config.log_deliveries {
            // The URL path of chat webhooks (Slack, Discord, Teams) is a credential.
            tracing::info!(
                "Added webhook: {} -> {}",
                webhook.name,
                crate::config::redact_url_path(&webhook.url)
            );
        }

        Ok(())
    }

    /// Remove a webhook configuration.
    ///
    /// Its listener stops right away. Deliveries already started finish (including
    /// their retries).
    pub async fn remove_webhook(&self, webhook_id: Uuid) -> crate::Result<()> {
        {
            let mut webhooks = self.webhooks.write().await;
            webhooks.remove(&webhook_id);
        }

        // Dropping the handle stops the listener.
        let handle = self.subscriptions.write().await.remove(&webhook_id);
        if let Some(handle) = handle {
            self.event_manager
                .unsubscribe(handle.subscription_id)
                .await?;
        }

        {
            let mut stats = self.stats.write().await;
            stats.remove(&webhook_id);
        }

        if self.config.log_deliveries {
            tracing::info!("Removed webhook: {}", webhook_id);
        }

        Ok(())
    }

    /// Get webhook configuration
    pub async fn get_webhook(&self, webhook_id: Uuid) -> Option<WebhookConfig> {
        let webhooks = self.webhooks.read().await;
        webhooks.get(&webhook_id).cloned()
    }

    /// List all webhook configurations
    pub async fn list_webhooks(&self) -> Vec<WebhookConfig> {
        let webhooks = self.webhooks.read().await;
        webhooks.values().cloned().collect()
    }

    /// Enable a webhook
    pub async fn enable_webhook(&self, webhook_id: Uuid) -> crate::Result<()> {
        let mut webhooks = self.webhooks.write().await;
        if let Some(webhook) = webhooks.get_mut(&webhook_id) {
            webhook.enabled = true;
            Ok(())
        } else {
            Err(crate::HammerworkError::Queue {
                message: format!("Webhook {} not found", webhook_id),
            })
        }
    }

    /// Disable a webhook
    pub async fn disable_webhook(&self, webhook_id: Uuid) -> crate::Result<()> {
        let mut webhooks = self.webhooks.write().await;
        if let Some(webhook) = webhooks.get_mut(&webhook_id) {
            webhook.enabled = false;
            Ok(())
        } else {
            Err(crate::HammerworkError::Queue {
                message: format!("Webhook {} not found", webhook_id),
            })
        }
    }

    /// Update a webhook's configuration.
    ///
    /// The new configuration (URL, filter, method, auth, ...) applies to every event
    /// the webhook's listener handles from now on. The listener and the statistics
    /// are kept, so no event is missed or delivered twice because of the update.
    /// Deliveries already started finish with the configuration they started with.
    /// A webhook with an unknown id is added.
    pub async fn update_webhook(&self, webhook: WebhookConfig) -> crate::Result<()> {
        // Validate first so an invalid update leaves the existing webhook in place.
        webhook.validate()?;

        {
            let mut webhooks = self.webhooks.write().await;
            if let Some(existing) = webhooks.get_mut(&webhook.id) {
                *existing = webhook;
                return Ok(());
            }
        }

        self.add_webhook(webhook).await
    }

    /// Get webhook statistics
    pub async fn get_webhook_stats(&self, webhook_id: Uuid) -> Option<WebhookStats> {
        let stats = self.stats.read().await;
        stats.get(&webhook_id).cloned()
    }

    /// Get statistics for all webhooks
    pub async fn get_all_webhook_stats(&self) -> Vec<WebhookStats> {
        let stats = self.stats.read().await;
        stats.values().cloned().collect()
    }

    /// Get general webhook manager statistics
    pub async fn get_stats(&self) -> WebhookManagerStats {
        let webhooks = self.webhooks.read().await;
        let stats = self.stats.read().await;

        let total_webhooks = webhooks.len();
        let active_webhooks = webhooks.values().filter(|w| w.enabled).count();
        let total_attempts = stats.values().map(|s| s.total_attempts).sum();
        let failed_attempts = stats.values().map(|s| s.failed_attempts).sum();
        let successful_deliveries = stats.values().map(|s| s.successful_deliveries).sum();
        let failed_deliveries = stats.values().map(|s| s.failed_deliveries).sum();

        WebhookManagerStats {
            total_webhooks,
            active_webhooks,
            total_attempts,
            failed_attempts,
            successful_deliveries,
            failed_deliveries,
        }
    }

    /// Start the listener of a webhook: it reads `receiver` until `stop` fires (its
    /// [`ListenerHandle`] is dropped) and starts a delivery for every matching event.
    fn start_webhook_delivery_task(
        &self,
        webhook_id: Uuid,
        // The listener holds the whole subscription: dropping it (when the listener
        // stops) unregisters the subscription from the event manager.
        mut subscription: EventSubscription,
        mut stop: oneshot::Receiver<()>,
    ) {
        let webhooks = self.webhooks.clone();
        let http_client = self.http_client.clone();
        let stats = self.stats.clone();
        let delivery_semaphore = self.delivery_semaphore.clone();
        let config = self.config.clone();
        let deliveries = self.deliveries.clone();
        let event_manager = self.event_manager.clone();
        let pending = Arc::new(Semaphore::new(clamp_permits(
            config.max_pending_deliveries,
            "max_pending_deliveries",
        )));

        self.listeners.spawn("listener", async move {
            // Whether the listener is waiting for pending deliveries (logged once per pause).
            let mut paused = false;
            loop {
                let received = tokio::select! {
                    biased;
                    _ = &mut stop => break,
                    received = subscription.receiver.recv() => received,
                };
                let mut event = match received {
                    Ok(event) => event,
                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
                        tracing::warn!(
                            "Webhook {webhook_id} listener fell behind and dropped {skipped} \
                             events; the endpoint may be slow or down, or the event buffer \
                             too small"
                        );
                        if let Some(webhook_stats) = stats.write().await.get_mut(&webhook_id) {
                            webhook_stats.dropped_events += skipped;
                        }
                        continue;
                    }
                    // Event manager dropped
                    Err(broadcast::error::RecvError::Closed) => break,
                };

                // Use the current configuration: it may have been updated, enabled or
                // disabled while waiting. A disabled webhook drops events but keeps
                // listening, so that enabling it again resumes delivery.
                let webhook = match webhooks.read().await.get(&webhook_id) {
                    Some(current) => current.clone(),
                    None => break,
                };
                if !webhook.enabled || !webhook.filter.matches(&event) {
                    continue;
                }
                // Another subscriber may have asked for payloads; this one only gets
                // them when its filter (or the event manager's default) asks for them.
                if !event_manager.delivers_payload(&webhook.filter) {
                    event.payload = None;
                }

                // Bound the backlog: wait for a pending-delivery slot. While waiting,
                // new events queue up in the broadcast buffer (and the oldest are
                // dropped once it is full) instead of in unbounded delivery tasks.
                let full = pending.available_permits() == 0;
                if full && !paused && config.log_deliveries {
                    tracing::warn!(
                        "Webhook {} has {} deliveries pending; pausing its listener",
                        webhook.name,
                        config.max_pending_deliveries
                    );
                }
                paused = full;
                let slot = tokio::select! {
                    biased;
                    _ = &mut stop => break,
                    slot = pending.clone().acquire_owned() => match slot {
                        Ok(slot) => slot,
                        Err(_) => break,
                    },
                };

                let http_client = http_client.clone();
                let stats = stats.clone();
                let config = config.clone();
                let delivery_semaphore = delivery_semaphore.clone();
                deliveries.spawn("delivery", async move {
                    Self::deliver_webhook_event(
                        webhook,
                        event,
                        http_client,
                        stats,
                        config,
                        delivery_semaphore,
                        slot,
                    )
                    .await;
                });
            }
        });
    }

    /// Deliver an event to a webhook endpoint with retries.
    ///
    /// Each attempt holds a slot of `delivery_semaphore` while its request is in
    /// flight; the backoff between attempts does not. `_pending` is the webhook's
    /// pending-delivery slot, released when the delivery is finished.
    async fn deliver_webhook_event(
        webhook: WebhookConfig,
        event: JobLifecycleEvent,
        http_client: Client,
        stats: Arc<RwLock<HashMap<Uuid, WebhookStats>>>,
        config: WebhookManagerConfig,
        delivery_semaphore: Arc<Semaphore>,
        _pending: OwnedSemaphorePermit,
    ) {
        let mut attempt = 0;
        // Every event is attempted at least once, even with `max_attempts: 0`.
        let max_attempts = webhook.retry_policy.max_attempts.max(1);
        // How the delivery of this event ended (its last attempt), counted once.
        let mut outcome: Option<bool> = None;

        while attempt < max_attempts {
            attempt += 1;

            let delivery_result = {
                let Ok(_permit) = delivery_semaphore.acquire().await else {
                    tracing::error!("webhook delivery semaphore closed");
                    return;
                };
                Self::attempt_webhook_delivery(&webhook, &event, &http_client, &config, attempt)
                    .await
            };

            // Update statistics
            Self::update_webhook_stats(&webhook.id, &delivery_result, stats.clone()).await;
            outcome = Some(delivery_result.success);

            if delivery_result.success {
                if config.log_deliveries {
                    tracing::debug!(
                        "Webhook delivery successful: {} -> {} (attempt {})",
                        webhook.name,
                        crate::config::redact_url_path(&webhook.url),
                        attempt
                    );
                }
                break;
            } else if attempt < max_attempts {
                // Check if we should retry based on status code
                let should_retry = delivery_result
                    .status_code
                    .map(|code| webhook.retry_policy.should_retry_status(code))
                    .unwrap_or(true); // Retry on network errors

                if should_retry {
                    let delay = webhook.retry_policy.calculate_delay(attempt - 1);
                    if config.log_deliveries {
                        tracing::warn!(
                            "Webhook delivery failed, retrying in {:?}: {} -> {} (attempt {}/{})",
                            delay,
                            webhook.name,
                            crate::config::redact_url_path(&webhook.url),
                            attempt,
                            max_attempts
                        );
                    }
                    sleep(delay).await;
                } else {
                    if config.log_deliveries {
                        tracing::error!(
                            "Webhook delivery failed with non-retryable status {}: {} -> {}",
                            delivery_result.status_code.unwrap_or(0),
                            webhook.name,
                            crate::config::redact_url_path(&webhook.url)
                        );
                    }
                    break;
                }
            } else if config.log_deliveries {
                tracing::error!(
                    "Webhook delivery failed after {} attempts: {} -> {}",
                    max_attempts,
                    webhook.name,
                    crate::config::redact_url_path(&webhook.url)
                );
            }
        }

        if let Some(success) = outcome {
            Self::record_delivery_outcome(&webhook.id, success, stats).await;
        }
    }

    /// Attempt a single webhook delivery.
    ///
    /// The webhook's timeout covers the whole attempt: connecting, sending, the
    /// response headers and reading the response body. When the status arrived but
    /// the body did not finish in time, the attempt counts by its status (the endpoint
    /// did answer) and the body is not recorded.
    async fn attempt_webhook_delivery(
        webhook: &WebhookConfig,
        event: &JobLifecycleEvent,
        http_client: &Client,
        config: &WebhookManagerConfig,
        attempt_number: u32,
    ) -> WebhookDelivery {
        let delivery_id = Uuid::new_v4();
        let start_time = std::time::Instant::now();
        let elapsed_ms = || u64::try_from(start_time.elapsed().as_millis()).unwrap_or(u64::MAX);
        let failure = |status_code: Option<u16>, error: String, duration_ms: u64| WebhookDelivery {
            delivery_id,
            webhook_id: webhook.id,
            event_id: event.event_id,
            status_code,
            response_body: None,
            error_message: Some(error),
            attempted_at: Utc::now(),
            duration_ms: Some(duration_ms),
            attempt_number,
            success: false,
        };

        // Prepare payload
        let payload = match Self::build_payload(webhook, event) {
            Ok(payload) => payload,
            Err(e) => return failure(None, format!("failed to build payload: {e}"), 0),
        };

        // Build request
        let mut request = match webhook.method {
            HttpMethod::Post => http_client.post(&webhook.url),
            HttpMethod::Put => http_client.put(&webhook.url),
            HttpMethod::Patch => http_client.patch(&webhook.url),
        };

        // Add headers
        for (key, value) in &webhook.headers {
            request = request.header(key, value);
        }

        // Add authentication
        if let Some(ref auth) = webhook.auth {
            request = Self::add_auth_headers(request, auth);
        }

        // Add HMAC signature if secret is configured
        if let Some(ref secret) = webhook.secret {
            let signature = Self::calculate_hmac_signature(secret, &payload);
            request = request.header("X-Hammerwork-Signature", signature);
        }

        // Set content type and body
        request = request
            .header("Content-Type", "application/json")
            .json(&payload);

        // One deadline for the whole attempt, body included.
        let deadline = Instant::now() + attempt_timeout(webhook, config);

        let response = match timeout_at(deadline, request.send()).await {
            Ok(Ok(response)) => response,
            Ok(Err(err)) => return failure(None, err.to_string(), elapsed_ms()),
            Err(_) => return failure(Some(408), "Request timeout".to_string(), elapsed_ms()),
        };

        let status_code = response.status().as_u16();
        let success = response.status().is_success();
        let response_body = match timeout_at(
            deadline,
            read_body_limited(response, config.max_response_body_size),
        )
        .await
        {
            Ok(Ok(body)) => Some(body),
            Ok(Err(e)) => {
                tracing::debug!(
                    "Webhook {}: failed to read response body: {e}",
                    webhook.name
                );
                None
            }
            Err(_) => {
                tracing::debug!(
                    "Webhook {}: timed out reading the response body",
                    webhook.name
                );
                None
            }
        };

        WebhookDelivery {
            delivery_id,
            webhook_id: webhook.id,
            event_id: event.event_id,
            status_code: Some(status_code),
            response_body,
            error_message: if success {
                None
            } else {
                Some(format!("HTTP {}", status_code))
            },
            attempted_at: Utc::now(),
            duration_ms: Some(elapsed_ms()),
            attempt_number,
            success,
        }
    }

    /// The request body for `event`: the rendered payload template, or the event itself.
    fn build_payload(
        webhook: &WebhookConfig,
        event: &JobLifecycleEvent,
    ) -> crate::Result<serde_json::Value> {
        match &webhook.payload_template {
            Some(template) => PayloadTemplate::parse(template)?.render(event),
            None => Ok(serde_json::to_value(event)?),
        }
    }

    /// Add authentication headers to request
    fn add_auth_headers(
        mut request: reqwest::RequestBuilder,
        auth: &WebhookAuth,
    ) -> reqwest::RequestBuilder {
        match auth {
            WebhookAuth::Bearer { token } => {
                request = request.header("Authorization", format!("Bearer {}", token));
            }
            WebhookAuth::Basic { username, password } => {
                request = request.basic_auth(username, Some(password));
            }
            WebhookAuth::ApiKey {
                header_name,
                api_key,
            } => {
                request = request.header(header_name, api_key);
            }
            WebhookAuth::Custom { headers } => {
                for (key, value) in headers {
                    request = request.header(key, value);
                }
            }
        }
        request
    }

    /// Calculate HMAC signature for webhook verification
    fn calculate_hmac_signature(secret: &str, payload: &serde_json::Value) -> String {
        use hmac::{Hmac, Mac};
        use sha2::Sha256;

        type HmacSha256 = Hmac<Sha256>;

        let payload_bytes = serde_json::to_string(payload).unwrap_or_default();
        let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).unwrap();
        mac.update(payload_bytes.as_bytes());
        let result = mac.finalize();
        format!("sha256={}", hex::encode(result.into_bytes()))
    }

    /// Update the per-attempt statistics: the attempt counters, the average response
    /// time and the last success and failure timestamps.
    async fn update_webhook_stats(
        webhook_id: &Uuid,
        delivery: &WebhookDelivery,
        stats: Arc<RwLock<HashMap<Uuid, WebhookStats>>>,
    ) {
        let mut stats_map = stats.write().await;
        if let Some(webhook_stats) = stats_map.get_mut(webhook_id) {
            webhook_stats.total_attempts += 1;

            if delivery.success {
                webhook_stats.last_success_at = Some(delivery.attempted_at);
            } else {
                webhook_stats.failed_attempts += 1;
                webhook_stats.last_failure_at = Some(delivery.attempted_at);
            }

            // Update average response time (per attempt)
            if let Some(duration) = delivery.duration_ms {
                let total_time =
                    webhook_stats.avg_response_time_ms * (webhook_stats.total_attempts - 1) as f64;
                webhook_stats.avg_response_time_ms =
                    (total_time + duration as f64) / webhook_stats.total_attempts as f64;
            }

            webhook_stats.calculated_at = Utc::now();
        }
    }

    /// Count the final outcome of one event's delivery (after its retries): a success on
    /// the third attempt is one successful delivery, not two failed ones and a success.
    async fn record_delivery_outcome(
        webhook_id: &Uuid,
        success: bool,
        stats: Arc<RwLock<HashMap<Uuid, WebhookStats>>>,
    ) {
        let mut stats_map = stats.write().await;
        if let Some(webhook_stats) = stats_map.get_mut(webhook_id) {
            if success {
                webhook_stats.successful_deliveries += 1;
            } else {
                webhook_stats.failed_deliveries += 1;
            }
            let finished = webhook_stats.successful_deliveries + webhook_stats.failed_deliveries;
            webhook_stats.success_rate =
                webhook_stats.successful_deliveries as f64 / finished as f64;
            webhook_stats.calculated_at = Utc::now();
        }
    }
}

/// Helper for creating webhook configurations
impl WebhookConfig {
    /// Create a new webhook configuration
    pub fn new(name: String, url: String) -> Self {
        Self {
            id: Uuid::new_v4(),
            name,
            url,
            method: HttpMethod::Post,
            headers: HashMap::new(),
            filter: EventFilter::default(),
            retry_policy: RetryPolicy::default(),
            auth: None,
            timeout_secs: 30,
            enabled: true,
            secret: None,
            payload_template: None,
        }
    }

    /// Set HTTP method
    pub fn with_method(mut self, method: HttpMethod) -> Self {
        self.method = method;
        self
    }

    /// Add custom header
    pub fn with_header(mut self, key: String, value: String) -> Self {
        self.headers.insert(key, value);
        self
    }

    /// Set event filter
    pub fn with_filter(mut self, filter: EventFilter) -> Self {
        self.filter = filter;
        self
    }

    /// Set retry policy
    pub fn with_retry_policy(mut self, retry_policy: RetryPolicy) -> Self {
        self.retry_policy = retry_policy;
        self
    }

    /// Set authentication
    pub fn with_auth(mut self, auth: WebhookAuth) -> Self {
        self.auth = Some(auth);
        self
    }

    /// Set request timeout
    pub fn with_timeout_secs(mut self, timeout_secs: u64) -> Self {
        self.timeout_secs = timeout_secs;
        self
    }

    /// Set webhook secret for HMAC signatures
    pub fn with_secret(mut self, secret: String) -> Self {
        self.secret = Some(secret);
        self
    }

    /// Enable or disable webhook
    pub fn enabled(mut self, enabled: bool) -> Self {
        self.enabled = enabled;
        self
    }

    /// Send a rendered [`PayloadTemplate`] instead of the raw event.
    ///
    /// The template is validated by [`validate`](Self::validate), which
    /// [`WebhookManager::add_webhook`] calls.
    pub fn with_payload_template(mut self, template: impl Into<String>) -> Self {
        self.payload_template = Some(template.into());
        self
    }

    /// Check the configuration, returning a configuration error for an invalid
    /// payload template.
    pub fn validate(&self) -> crate::Result<()> {
        if let Some(template) = &self.payload_template {
            PayloadTemplate::parse(template).map_err(|e| {
                crate::HammerworkError::Config(format!("webhook '{}': {}", self.name, e))
            })?;
        }
        Ok(())
    }
}

/// A parsed webhook payload template.
///
/// A template is a JSON document. Any JSON string inside it (at any depth, but not
/// object keys) may contain placeholders of the form `{{ path }}`, which are filled
/// in from the [`JobLifecycleEvent`] being delivered:
///
/// - A string that consists of exactly one placeholder, such as `"{{event.payload}}"`,
///   is replaced by the JSON value itself, keeping its type (object, number, `null`,
///   and so on).
/// - A placeholder embedded in other text, such as `"Job {{event.job_id}} failed"`,
///   is replaced by the value's text: strings are inserted as-is, `null` becomes an
///   empty string, and other values are inserted as compact JSON.
///
/// Substitution happens on the parsed JSON tree, never by pasting text into the
/// template source, so event data cannot break out of a string or inject JSON: the
/// rendered payload is always valid JSON, with values escaped by the serializer.
///
/// # Available paths
///
/// | Path | Value |
/// |------|-------|
/// | `event` | the whole event, as sent when no template is configured |
/// | `event.event_id`, `event.job_id` | UUID strings |
/// | `event.queue_name` | queue name |
/// | `event.event_type` | e.g. `"completed"`, `"failed"` |
/// | `event.priority` | e.g. `"Normal"` |
/// | `event.timestamp` | RFC 3339 timestamp |
/// | `event.processing_time_ms` | number or `null` |
/// | `event.error` | error object or `null` |
/// | `event.error.message`, `.error_type`, `.details`, `.retry_attempt` | error fields |
/// | `event.payload` | job payload (when included in events) or `null` |
/// | `event.payload.<key>...` | a nested payload value; numeric segments index arrays |
/// | `event.metadata` | metadata object |
/// | `event.metadata.<key>` | one metadata value |
///
/// A path that is valid but absent from a particular event (for example
/// `event.error.message` on a completed job) renders as `null`.
///
/// Templates are validated when the webhook is added: invalid JSON, an unknown
/// path, an unterminated `{{`, or a placeholder inside an object key is a
/// configuration error. There is no escape for a literal `{{`.
///
/// # Examples
///
/// ```rust
/// use hammerwork::webhooks::PayloadTemplate;
///
/// let template = PayloadTemplate::parse(r#"{
///     "text": "Job {{event.job_id}} in {{event.queue_name}} is {{event.event_type}}",
///     "job": "{{event.job_id}}",
///     "data": "{{event.payload}}"
/// }"#).unwrap();
///
/// assert!(PayloadTemplate::parse(r#"{"x": "{{event.nope}}"}"#).is_err());
/// ```
#[derive(Debug, Clone)]
pub struct PayloadTemplate {
    root: serde_json::Value,
}

/// One piece of a template string.
enum TemplatePart<'a> {
    Text(&'a str),
    Placeholder(Vec<&'a str>),
}

impl PayloadTemplate {
    /// Parse and validate a template.
    pub fn parse(template: &str) -> crate::Result<Self> {
        let root: serde_json::Value = serde_json::from_str(template).map_err(|e| {
            crate::HammerworkError::Config(format!("payload_template is not valid JSON: {e}"))
        })?;
        Self::validate_value(&root)?;
        Ok(Self { root })
    }

    /// Render the template for `event`.
    pub fn render(&self, event: &JobLifecycleEvent) -> crate::Result<serde_json::Value> {
        let event_value = serde_json::to_value(event)?;
        Ok(Self::render_value(&self.root, &event_value))
    }

    fn validate_value(value: &serde_json::Value) -> crate::Result<()> {
        match value {
            serde_json::Value::String(text) => {
                for part in Self::split(text)? {
                    if let TemplatePart::Placeholder(path) = part {
                        Self::validate_path(&path)?;
                    }
                }
                Ok(())
            }
            serde_json::Value::Array(items) => items.iter().try_for_each(Self::validate_value),
            serde_json::Value::Object(map) => {
                for (key, value) in map {
                    if key.contains("{{") {
                        return Err(crate::HammerworkError::Config(format!(
                            "payload_template: placeholders are not allowed in object keys \
                             ('{key}')"
                        )));
                    }
                    Self::validate_value(value)?;
                }
                Ok(())
            }
            _ => Ok(()),
        }
    }

    fn validate_path(path: &[&str]) -> crate::Result<()> {
        let unknown = || {
            crate::HammerworkError::Config(format!(
                "payload_template: unknown placeholder '{{{{{}}}}}'",
                path.join(".")
            ))
        };
        if path.iter().any(|segment| segment.is_empty()) {
            return Err(unknown());
        }
        match path {
            ["event"] => Ok(()),
            ["event", field, rest @ ..] => match *field {
                "event_id" | "job_id" | "queue_name" | "event_type" | "priority" | "timestamp"
                | "processing_time_ms"
                    if rest.is_empty() =>
                {
                    Ok(())
                }
                "error" => match rest {
                    [] | ["message" | "error_type" | "details" | "retry_attempt"] => Ok(()),
                    _ => Err(unknown()),
                },
                "metadata" if rest.len() <= 1 => Ok(()),
                "payload" => Ok(()),
                _ => Err(unknown()),
            },
            _ => Err(unknown()),
        }
    }

    /// Split a template string into text and placeholders.
    fn split(text: &str) -> crate::Result<Vec<TemplatePart<'_>>> {
        let mut parts = Vec::new();
        let mut rest = text;
        while let Some(start) = rest.find("{{") {
            if start > 0 {
                parts.push(TemplatePart::Text(&rest[..start]));
            }
            let after = &rest[start + 2..];
            let Some(end) = after.find("}}") else {
                return Err(crate::HammerworkError::Config(format!(
                    "payload_template: unterminated '{{{{' in \"{text}\""
                )));
            };
            let path = after[..end].trim();
            parts.push(TemplatePart::Placeholder(
                path.split('.').map(str::trim).collect(),
            ));
            rest = &after[end + 2..];
        }
        if !rest.is_empty() {
            parts.push(TemplatePart::Text(rest));
        }
        Ok(parts)
    }

    fn render_value(value: &serde_json::Value, event: &serde_json::Value) -> serde_json::Value {
        match value {
            serde_json::Value::String(text) => Self::render_string(text, event),
            serde_json::Value::Array(items) => serde_json::Value::Array(
                items
                    .iter()
                    .map(|item| Self::render_value(item, event))
                    .collect(),
            ),
            serde_json::Value::Object(map) => serde_json::Value::Object(
                map.iter()
                    .map(|(key, value)| (key.clone(), Self::render_value(value, event)))
                    .collect(),
            ),
            other => other.clone(),
        }
    }

    fn render_string(text: &str, event: &serde_json::Value) -> serde_json::Value {
        // Validated at parse time, so splitting cannot fail here.
        let Ok(parts) = Self::split(text) else {
            return serde_json::Value::String(text.to_string());
        };
        if let [TemplatePart::Placeholder(path)] = parts.as_slice() {
            return Self::lookup(event, path)
                .cloned()
                .unwrap_or(serde_json::Value::Null);
        }
        let mut rendered = String::with_capacity(text.len());
        for part in parts {
            match part {
                TemplatePart::Text(text) => rendered.push_str(text),
                TemplatePart::Placeholder(path) => match Self::lookup(event, &path) {
                    None | Some(serde_json::Value::Null) => {}
                    Some(serde_json::Value::String(s)) => rendered.push_str(s),
                    Some(other) => rendered.push_str(&other.to_string()),
                },
            }
        }
        serde_json::Value::String(rendered)
    }

    /// Resolve `path` (starting with `event`) against the serialized event.
    fn lookup<'v>(event: &'v serde_json::Value, path: &[&str]) -> Option<&'v serde_json::Value> {
        let mut current = event;
        for segment in path.get(1..)? {
            current = match current {
                serde_json::Value::Object(map) => map.get(*segment)?,
                serde_json::Value::Array(items) => items.get(segment.parse::<usize>().ok()?)?,
                _ => return None,
            };
        }
        Some(current)
    }
}

/// Generate HMAC-SHA256 signature for webhook payload verification.
///
/// This function creates a secure signature that can be used to verify webhook authenticity.
/// The signature is generated using HMAC-SHA256 and returned as a hexadecimal string.
///
/// # Arguments
///
/// * `secret` - The shared secret key for HMAC generation
/// * `payload` - The raw payload bytes to sign
///
/// # Examples
///
/// ```rust
/// use hammerwork::webhooks::generate_hmac_signature;
///
/// let secret = "my-webhook-secret";
/// let payload = b"{'event': 'job.completed', 'job_id': '123'}";
/// let signature = generate_hmac_signature(secret, payload);
///
/// // Deliveries send it as the X-Hammerwork-Signature header
/// println!("X-Hammerwork-Signature: sha256={}", signature);
/// ```
///
/// # Security Notes
///
/// - Use a cryptographically secure random secret of at least 32 characters
/// - Always compare signatures using constant-time comparison to prevent timing attacks
/// - Rotate secrets periodically for enhanced security
pub fn generate_hmac_signature(secret: &str, payload: &[u8]) -> String {
    use hmac::{Hmac, Mac};
    use sha2::Sha256;

    let mut mac =
        Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
    mac.update(payload);
    let result = mac.finalize();
    hex::encode(result.into_bytes())
}

/// Verify HMAC-SHA256 signature for webhook payload authenticity.
///
/// This function verifies that a received webhook payload matches the expected signature,
/// providing protection against tampering and ensuring the webhook came from a trusted source.
/// Uses constant-time comparison to prevent timing attacks.
///
/// # Arguments
///
/// * `secret` - The shared secret key used for verification
/// * `payload` - The raw payload bytes to verify
/// * `signature` - The signature to verify against: the `X-Hammerwork-Signature` header
///   value of a delivery (`sha256=<hex>`), or the bare hexadecimal signature
///
/// # Returns
///
/// Returns `true` if the signature is valid, `false` otherwise.
///
/// # Examples
///
/// ```rust
/// use hammerwork::webhooks::{generate_hmac_signature, verify_hmac_signature};
///
/// let secret = "my-webhook-secret";
/// let payload = b"{'event': 'job.completed', 'job_id': '123'}";
///
/// // Generate signature (sender side)
/// let signature = generate_hmac_signature(secret, payload);
///
/// // Verify signature (receiver side)
/// let is_valid = verify_hmac_signature(secret, payload, &signature);
/// assert!(is_valid);
///
/// // Invalid signature
/// let is_invalid = verify_hmac_signature(secret, payload, "invalid-signature");
/// assert!(!is_invalid);
/// ```
///
/// # Security Notes
///
/// - Always verify signatures before processing webhook payloads
/// - Use the same secret that was used to generate the signature
/// - This function uses constant-time comparison to prevent timing attacks
pub fn verify_hmac_signature(secret: &str, payload: &[u8], signature: &str) -> bool {
    use hmac::{Hmac, Mac};
    use sha2::Sha256;

    // Deliveries send `X-Hammerwork-Signature: sha256=<hex>`; accept that header value as
    // well as the bare hex of `generate_hmac_signature`.
    let signature = signature.trim();
    let hex_signature = signature.strip_prefix("sha256=").unwrap_or(signature);
    let Ok(signature_bytes) = hex::decode(hex_signature) else {
        return false;
    };
    let Ok(mut mac) = Hmac::<Sha256>::new_from_slice(secret.as_bytes()) else {
        return false;
    };
    mac.update(payload);
    // `verify_slice` compares in constant time.
    mac.verify_slice(&signature_bytes).is_ok()
}

with_setters! {
    RetryPolicy {
        with_max_attempts(max_attempts): u32,
        with_initial_delay_secs(initial_delay_secs): u64,
        with_max_delay_secs(max_delay_secs): u64,
        with_backoff_multiplier(backoff_multiplier): f64,
        with_retry_on_status_codes(retry_on_status_codes): Vec<u16>,
    }
}

with_setters! {
    WebhookManagerConfig {
        with_max_concurrent_deliveries(max_concurrent_deliveries): usize,
        with_max_response_body_size(max_response_body_size): usize,
        with_log_deliveries(log_deliveries): bool,
        with_default_timeout_secs(default_timeout_secs): u64,
        with_max_pending_deliveries(max_pending_deliveries): usize,
        with_user_agent(user_agent): Into(String),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn verify_hmac_signature_accepts_the_delivered_signature_header() {
        let secret = "webhook-secret";
        let payload = serde_json::json!({"event": "job.completed", "job_id": "123"});
        let body = serde_json::to_vec(&payload).unwrap();
        // What a delivery sends in X-Hammerwork-Signature: "sha256=<hex>"
        let header = WebhookManager::calculate_hmac_signature(secret, &payload);
        assert!(header.starts_with("sha256="));
        assert!(verify_hmac_signature(secret, &body, &header));
        // The bare hex form keeps working
        assert!(verify_hmac_signature(
            secret,
            &body,
            &generate_hmac_signature(secret, &body)
        ));
        // Wrong secret, tampered body, garbage and empty signatures are rejected
        assert!(!verify_hmac_signature("other", &body, &header));
        assert!(!verify_hmac_signature(secret, b"{}", &header));
        assert!(!verify_hmac_signature(secret, &body, "sha256=zz"));
        assert!(!verify_hmac_signature(secret, &body, ""));
        assert!(!verify_hmac_signature(secret, &body, "sha256="));
    }

    #[test]
    fn test_debug_does_not_print_webhook_secrets() {
        let mut config = WebhookConfig::new(
            "chat".to_string(),
            "https://hooks.slack.com/services/T0/B0/url-hunter2".to_string(),
        )
        .with_secret("sign-hunter2".to_string());
        config
            .headers
            .insert("X-Api-Token".to_string(), "header-hunter2".to_string());
        for auth in [
            WebhookAuth::Bearer {
                token: "bearer-hunter2".to_string(),
            },
            WebhookAuth::Basic {
                username: "svc".to_string(),
                password: "basic-hunter2".to_string(),
            },
            WebhookAuth::ApiKey {
                header_name: "X-Key".to_string(),
                api_key: "api-hunter2".to_string(),
            },
            WebhookAuth::Custom {
                headers: HashMap::from([("X-Custom".to_string(), "custom-hunter2".to_string())]),
            },
        ] {
            config.auth = Some(auth);
            let debug = format!("{:?}", config);
            assert!(!debug.contains("hunter2"), "{debug}");
            assert!(debug.contains("https://hooks.slack.com/***"), "{debug}");
            assert!(debug.contains("X-Api-Token"), "{debug}");
        }
    }

    #[tokio::test]
    async fn test_webhook_manager_from_config() {
        let mut config = crate::config::WebhookConfigs::default();
        config.global_settings.max_concurrent_deliveries = 7;
        config.global_settings.user_agent = "custom-agent".to_string();
        config.webhooks.push(WebhookConfig::new(
            "hook".to_string(),
            "https://example.com/webhook".to_string(),
        ));

        let manager_config = WebhookManagerConfig::from(&config.global_settings);
        assert_eq!(manager_config.max_concurrent_deliveries, 7);
        assert_eq!(manager_config.user_agent, "custom-agent");

        let manager = WebhookManager::from_config(Arc::new(EventManager::new_default()), &config)
            .await
            .unwrap();
        assert_eq!(manager.list_webhooks().await.len(), 1);
    }

    #[test]
    fn test_webhook_config_creation() {
        let webhook = WebhookConfig::new(
            "test-webhook".to_string(),
            "https://example.com/webhook".to_string(),
        )
        .with_method(HttpMethod::Post)
        .with_header("X-Custom-Header".to_string(), "custom-value".to_string())
        .with_timeout_secs(60)
        .with_secret("secret-key".to_string());

        assert_eq!(webhook.name, "test-webhook");
        assert_eq!(webhook.url, "https://example.com/webhook");
        assert_eq!(webhook.method, HttpMethod::Post);
        assert_eq!(webhook.timeout_secs, 60);
        assert_eq!(webhook.secret, Some("secret-key".to_string()));
        assert!(webhook.enabled);
        assert_eq!(
            webhook.headers.get("X-Custom-Header"),
            Some(&"custom-value".to_string())
        );
    }

    #[test]
    fn test_retry_policy_delay_calculation() {
        let policy = RetryPolicy {
            initial_delay_secs: 2,
            backoff_multiplier: 2.0,
            max_delay_secs: 60,
            ..Default::default()
        };

        assert_eq!(policy.calculate_delay(0), Duration::from_secs(2));
        assert_eq!(policy.calculate_delay(1), Duration::from_secs(4));
        assert_eq!(policy.calculate_delay(2), Duration::from_secs(8));
        assert_eq!(policy.calculate_delay(10), Duration::from_secs(60)); // Capped at max
    }

    #[test]
    fn test_retry_policy_status_codes() {
        let policy = RetryPolicy::default();

        assert!(policy.should_retry_status(500));
        assert!(policy.should_retry_status(502));
        assert!(policy.should_retry_status(429));
        assert!(!policy.should_retry_status(200));
        assert!(!policy.should_retry_status(400));
        assert!(!policy.should_retry_status(404));
    }

    #[test]
    fn test_webhook_auth_types() {
        let bearer = WebhookAuth::Bearer {
            token: "token123".to_string(),
        };
        let basic = WebhookAuth::Basic {
            username: "user".to_string(),
            password: "pass".to_string(),
        };
        let api_key = WebhookAuth::ApiKey {
            header_name: "X-API-Key".to_string(),
            api_key: "key123".to_string(),
        };

        match bearer {
            WebhookAuth::Bearer { token } => assert_eq!(token, "token123"),
            _ => panic!("Wrong auth type"),
        }

        match basic {
            WebhookAuth::Basic { username, password } => {
                assert_eq!(username, "user");
                assert_eq!(password, "pass");
            }
            _ => panic!("Wrong auth type"),
        }

        match api_key {
            WebhookAuth::ApiKey {
                header_name,
                api_key,
            } => {
                assert_eq!(header_name, "X-API-Key");
                assert_eq!(api_key, "key123");
            }
            _ => panic!("Wrong auth type"),
        }
    }

    #[test]
    fn test_http_method_display() {
        assert_eq!(HttpMethod::Post.to_string(), "POST");
        assert_eq!(HttpMethod::Put.to_string(), "PUT");
        assert_eq!(HttpMethod::Patch.to_string(), "PATCH");
    }

    #[test]
    fn test_webhook_delivery_structure() {
        let delivery = WebhookDelivery {
            delivery_id: Uuid::new_v4(),
            webhook_id: Uuid::new_v4(),
            event_id: Uuid::new_v4(),
            status_code: Some(200),
            response_body: Some("OK".to_string()),
            error_message: None,
            attempted_at: Utc::now(),
            duration_ms: Some(150),
            attempt_number: 1,
            success: true,
        };

        assert!(delivery.success);
        assert_eq!(delivery.status_code, Some(200));
        assert_eq!(delivery.attempt_number, 1);
        assert!(delivery.error_message.is_none());
    }

    #[test]
    fn test_webhook_stats_calculation() {
        let mut stats = WebhookStats::empty(Uuid::new_v4());

        // Simulate successful delivery
        stats.total_attempts = 1;
        stats.successful_deliveries = 1;
        stats.success_rate = 1.0;
        stats.avg_response_time_ms = 100.0;

        assert_eq!(stats.success_rate, 1.0);
        assert_eq!(stats.failed_deliveries, 0);

        // Simulate failed delivery
        stats.total_attempts = 2;
        stats.failed_deliveries = 1;
        stats.success_rate = 0.5;

        assert_eq!(stats.success_rate, 0.5);
        assert_eq!(stats.total_attempts, 2);
    }

    #[test]
    fn test_clamp_delivery_permits() {
        assert_eq!(clamp_delivery_permits(0), 1);
        assert_eq!(clamp_delivery_permits(7), 7);
        assert_eq!(clamp_delivery_permits(usize::MAX), Semaphore::MAX_PERMITS);
    }

    #[tokio::test]
    async fn test_new_with_extreme_concurrency_does_not_panic() {
        for value in [0, usize::MAX] {
            let config = WebhookManagerConfig {
                max_concurrent_deliveries: value,
                ..Default::default()
            };
            let manager =
                WebhookManager::new(Arc::new(EventManager::new_default()), config).unwrap();
            let permits = manager.delivery_semaphore.available_permits();
            assert!((1..=Semaphore::MAX_PERMITS).contains(&permits));
        }
    }

    #[test]
    fn test_new_rejects_invalid_user_agent() {
        let config = WebhookManagerConfig {
            user_agent: "bad\nagent".to_string(),
            ..Default::default()
        };
        // #130: an error, not a panic.
        let result = WebhookManager::new(Arc::new(EventManager::new_default()), config);
        assert!(matches!(
            result,
            Err(crate::HammerworkError::Webhook { .. })
        ));
    }

    #[test]
    fn test_retry_policy_huge_attempt_does_not_panic() {
        let policy = RetryPolicy::default();
        assert_eq!(
            policy.calculate_delay(u32::MAX),
            Duration::from_secs(policy.max_delay_secs)
        );
    }

    #[tokio::test]
    async fn test_webhook_manager_creation() {
        let config = WebhookManagerConfig::default();
        let event_manager = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new(event_manager, config).unwrap();

        let stats = manager.get_stats().await;
        assert_eq!(stats.total_webhooks, 0);
        assert_eq!(stats.active_webhooks, 0);
    }

    #[tokio::test]
    async fn test_webhook_manager_add_remove_webhook() {
        let config = WebhookManagerConfig::default();
        let event_manager = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new(event_manager, config).unwrap();

        let webhook_config = WebhookConfig {
            id: Uuid::new_v4(),
            name: "test-webhook".to_string(),
            url: "https://api.example.com/webhook".to_string(),
            method: HttpMethod::Post,
            headers: HashMap::new(),
            auth: None,
            secret: Some("test-secret".to_string()),
            filter: EventFilter::new(),
            retry_policy: RetryPolicy::default(),
            enabled: true,
            timeout_secs: 30,
            payload_template: None,
        };

        let webhook_id = webhook_config.id;

        // Add webhook
        manager.add_webhook(webhook_config).await.unwrap();

        let stats = manager.get_stats().await;
        assert_eq!(stats.total_webhooks, 1);
        assert_eq!(stats.active_webhooks, 1);

        // Remove webhook
        manager.remove_webhook(webhook_id).await.unwrap();

        let stats = manager.get_stats().await;
        assert_eq!(stats.total_webhooks, 0);
        assert_eq!(stats.active_webhooks, 0);
    }

    #[tokio::test]
    async fn test_webhook_manager_enable_disable() {
        let config = WebhookManagerConfig::default();
        let event_manager = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new(event_manager, config).unwrap();

        let webhook_config = WebhookConfig {
            id: Uuid::new_v4(),
            name: "test-webhook".to_string(),
            url: "https://api.example.com/webhook".to_string(),
            method: HttpMethod::Post,
            headers: HashMap::new(),
            auth: None,
            secret: None,
            filter: EventFilter::new(),
            retry_policy: RetryPolicy::default(),
            enabled: true,
            timeout_secs: 30,
            payload_template: None,
        };

        let webhook_id = webhook_config.id;
        manager.add_webhook(webhook_config).await.unwrap();

        // Initially enabled
        let stats = manager.get_stats().await;
        assert_eq!(stats.active_webhooks, 1);

        // Disable webhook
        manager.disable_webhook(webhook_id).await.unwrap();
        let stats = manager.get_stats().await;
        assert_eq!(stats.active_webhooks, 0);
        assert_eq!(stats.total_webhooks, 1); // Still exists but disabled

        // Re-enable webhook
        manager.enable_webhook(webhook_id).await.unwrap();
        let stats = manager.get_stats().await;
        assert_eq!(stats.active_webhooks, 1);
    }

    #[test]
    fn test_hmac_signature_generation() {
        let secret = "test-secret";
        let payload = b"test payload";

        let signature1 = generate_hmac_signature(secret, payload);
        let signature2 = generate_hmac_signature(secret, payload);

        // Same input should produce same signature
        assert_eq!(signature1, signature2);

        // Different payload should produce different signature
        let different_payload = b"different payload";
        let signature3 = generate_hmac_signature(secret, different_payload);
        assert_ne!(signature1, signature3);

        // Signature should be hex-encoded
        assert!(signature1.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[test]
    fn test_hmac_signature_verification() {
        let secret = "test-secret";
        let payload = b"test payload";

        let signature = generate_hmac_signature(secret, payload);
        assert!(verify_hmac_signature(secret, payload, &signature));

        // Wrong signature should fail
        assert!(!verify_hmac_signature(secret, payload, "wrong-signature"));

        // Wrong secret should fail
        assert!(!verify_hmac_signature("wrong-secret", payload, &signature));

        // Wrong payload should fail
        assert!(!verify_hmac_signature(secret, b"wrong payload", &signature));
    }

    #[test]
    fn test_webhook_config_serialization() {
        let webhook_config = WebhookConfig {
            id: Uuid::new_v4(),
            name: "test-webhook".to_string(),
            url: "https://api.example.com/webhook".to_string(),
            method: HttpMethod::Post,
            headers: {
                let mut headers = HashMap::new();
                headers.insert("Content-Type".to_string(), "application/json".to_string());
                headers
            },
            auth: Some(WebhookAuth::Bearer {
                token: "token123".to_string(),
            }),
            secret: Some("webhook-secret".to_string()),
            filter: EventFilter::new()
                .with_event_types(vec![crate::events::JobLifecycleEventType::Completed]),
            retry_policy: RetryPolicy {
                max_attempts: 5,
                initial_delay_secs: 1,
                max_delay_secs: 300,
                backoff_multiplier: 2.0,
                retry_on_status_codes: vec![500, 502, 503],
            },
            enabled: true,
            timeout_secs: 60,
            payload_template: Some(r#"{"job": "{{event.job_id}}"}"#.to_string()),
        };

        let serialized = serde_json::to_string(&webhook_config).unwrap();
        let deserialized: WebhookConfig = serde_json::from_str(&serialized).unwrap();

        assert_eq!(deserialized.name, webhook_config.name);
        assert_eq!(deserialized.url, webhook_config.url);
        assert_eq!(deserialized.enabled, webhook_config.enabled);
        assert_eq!(deserialized.timeout_secs, webhook_config.timeout_secs);
        assert!(deserialized.auth.is_some());
        assert!(deserialized.secret.is_some());
    }

    #[test]
    fn test_webhook_manager_config_defaults() {
        let config = WebhookManagerConfig::default();

        assert_eq!(config.max_concurrent_deliveries, 100);
        assert_eq!(config.default_timeout_secs, 30);
        assert!(config.log_deliveries);
        assert_eq!(config.max_response_body_size, 64 * 1024);
        assert!(config.user_agent.contains("hammerwork-webhooks"));
    }

    #[test]
    fn test_webhook_manager_config_customization() {
        let config = WebhookManagerConfig {
            max_concurrent_deliveries: 50,
            default_timeout_secs: 60,
            log_deliveries: false,
            ..Default::default()
        };

        assert_eq!(config.max_concurrent_deliveries, 50);
        assert_eq!(config.default_timeout_secs, 60);
        assert!(!config.log_deliveries);
    }

    #[tokio::test]
    async fn test_webhook_delivery_timeout() {
        let delivery = WebhookDelivery {
            delivery_id: Uuid::new_v4(),
            webhook_id: Uuid::new_v4(),
            event_id: Uuid::new_v4(),
            status_code: None,
            response_body: None,
            error_message: Some("Request timeout".to_string()),
            attempted_at: Utc::now(),
            duration_ms: Some(30000), // 30 seconds
            attempt_number: 1,
            success: false,
        };

        assert!(!delivery.success);
        assert!(delivery.error_message.is_some());
        assert_eq!(delivery.duration_ms, Some(30000));
        assert!(delivery.status_code.is_none());
    }

    #[test]
    fn test_webhook_filter_integration() {
        let filter = EventFilter::new()
            .with_event_types(vec![crate::events::JobLifecycleEventType::Completed])
            .with_queue_names(vec!["important".to_string()])
            .with_priorities(vec![crate::priority::JobPriority::High]);

        let matching_event = crate::events::JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "important".to_string(),
            event_type: crate::events::JobLifecycleEventType::Completed,
            priority: crate::priority::JobPriority::High,
            timestamp: Utc::now(),
            processing_time_ms: Some(1000),
            error: None,
            payload: None,
            metadata: HashMap::new(),
        };

        let non_matching_event = crate::events::JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "unimportant".to_string(),
            event_type: crate::events::JobLifecycleEventType::Failed,
            priority: crate::priority::JobPriority::Low,
            timestamp: Utc::now(),
            processing_time_ms: None,
            error: None,
            payload: None,
            metadata: HashMap::new(),
        };

        assert!(filter.matches(&matching_event));
        assert!(!filter.matches(&non_matching_event));
    }

    #[test]
    fn test_retry_policy_edge_cases() {
        let policy = RetryPolicy {
            max_attempts: 3,
            initial_delay_secs: 1,
            max_delay_secs: 10,
            backoff_multiplier: 3.0,
            retry_on_status_codes: vec![500, 502],
        };

        // Test delay calculation with large attempt numbers
        assert_eq!(policy.calculate_delay(0), Duration::from_secs(1));
        assert_eq!(policy.calculate_delay(1), Duration::from_secs(3));
        assert_eq!(policy.calculate_delay(2), Duration::from_secs(9));
        assert_eq!(policy.calculate_delay(3), Duration::from_secs(10)); // Capped

        // Test status code checking
        assert!(policy.should_retry_status(500));
        assert!(policy.should_retry_status(502));
        assert!(!policy.should_retry_status(503)); // Not in list
        assert!(!policy.should_retry_status(200)); // Success
        assert!(!policy.should_retry_status(400)); // Client error
    }

    #[test]
    fn test_webhook_delivery_serialization() {
        let delivery = WebhookDelivery {
            delivery_id: Uuid::new_v4(),
            webhook_id: Uuid::new_v4(),
            event_id: Uuid::new_v4(),
            status_code: Some(201),
            response_body: Some("{\"status\":\"success\"}".to_string()),
            error_message: None,
            attempted_at: Utc::now(),
            duration_ms: Some(250),
            attempt_number: 2,
            success: true,
        };

        let serialized = serde_json::to_string(&delivery).unwrap();
        let deserialized: WebhookDelivery = serde_json::from_str(&serialized).unwrap();

        assert_eq!(deserialized.delivery_id, delivery.delivery_id);
        assert_eq!(deserialized.status_code, delivery.status_code);
        assert_eq!(deserialized.success, delivery.success);
        assert_eq!(deserialized.attempt_number, delivery.attempt_number);
    }

    #[tokio::test]
    async fn test_webhook_stats_aggregation() {
        let webhook_id = Uuid::new_v4();
        let mut stats = WebhookStats {
            webhook_id,
            total_attempts: 10,
            failed_attempts: 3,
            successful_deliveries: 7,
            failed_deliveries: 3,
            success_rate: 0.7,
            avg_response_time_ms: 150.5,
            last_success_at: Some(Utc::now() - chrono::Duration::minutes(5)),
            last_failure_at: Some(Utc::now() - chrono::Duration::minutes(10)),
            dropped_events: 0,
            calculated_at: Utc::now(),
        };

        // Verify stats calculations
        assert_eq!(
            stats.total_attempts,
            stats.successful_deliveries + stats.failed_deliveries
        );
        assert!((stats.success_rate - 0.7).abs() < f64::EPSILON);
        assert!(stats.last_success_at.is_some());
        assert!(stats.last_failure_at.is_some());

        // Update stats with new delivery
        stats.total_attempts += 1;
        stats.successful_deliveries += 1;
        stats.success_rate = stats.successful_deliveries as f64 / stats.total_attempts as f64;

        assert_eq!(stats.total_attempts, 11);
        assert_eq!(stats.successful_deliveries, 8);
        assert!((stats.success_rate - 8.0 / 11.0).abs() < f64::EPSILON);
    }

    #[test]
    fn test_response_body_text_respects_utf8_boundaries() {
        let cut = |body: &str, max: usize| {
            let bytes = body.as_bytes();
            let truncated = bytes.len() > max;
            response_body_text(bytes[..max.min(bytes.len())].to_vec(), truncated)
        };
        // Short bodies are returned unchanged.
        assert_eq!(cut("ok", 10), "ok");

        // "é" is 2 bytes and "😀" is 4; a cut inside either drops the partial character.
        let body = "aé😀b"; // bytes: a(1) é(2) 😀(4) b(1) = 8
        assert_eq!(cut(body, 2), "a... [truncated]");
        assert_eq!(cut(body, 3), "aé... [truncated]");
        for max in 0..body.len() {
            let truncated = cut(body, max);
            assert!(truncated.ends_with("... [truncated]"));
            assert!(truncated.len() <= max + "... [truncated]".len());
        }
        assert_eq!(cut(body, body.len()), body);

        // Invalid UTF-8 from the endpoint is replaced, not an error.
        assert_eq!(
            response_body_text(vec![b'a', 0xff, b'b'], false),
            "a\u{fffd}b"
        );
    }

    fn completed_event() -> JobLifecycleEvent {
        JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "shutdown_queue".to_string(),
            event_type: crate::events::JobLifecycleEventType::Completed,
            priority: crate::priority::JobPriority::Normal,
            timestamp: Utc::now(),
            processing_time_ms: Some(1),
            error: None,
            payload: None,
            metadata: HashMap::new(),
        }
    }

    /// Minimal HTTP server that answers every request with `200 OK` after `delay`.
    /// Returns its URL and a counter of requests received.
    async fn slow_http_server(delay: Duration) -> (String, Arc<std::sync::atomic::AtomicUsize>) {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}/hook", listener.local_addr().unwrap());
        let received = Arc::new(AtomicUsize::new(0));
        let counter = received.clone();
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                let counter = counter.clone();
                tokio::spawn(async move {
                    let mut buf = vec![0u8; 16 * 1024];
                    let _ = socket.read(&mut buf).await;
                    counter.fetch_add(1, Ordering::SeqCst);
                    tokio::time::sleep(delay).await;
                    let _ = socket
                        .write_all(
                            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
                        )
                        .await;
                });
            }
        });
        (url, received)
    }

    async fn wait_for(received: &std::sync::atomic::AtomicUsize, count: usize) {
        for _ in 0..200 {
            if received.load(std::sync::atomic::Ordering::SeqCst) >= count {
                return;
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        panic!("server did not receive {count} request(s)");
    }

    #[tokio::test]
    async fn test_shutdown_waits_for_in_flight_delivery() {
        let (url, received) = slow_http_server(Duration::from_millis(300)).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook = WebhookConfig::new("slow".to_string(), url);
        let webhook_id = webhook.id;
        manager.add_webhook(webhook).await.unwrap();

        events.publish_event(completed_event()).await.unwrap();
        wait_for(&received, 1).await;

        // The delivery is mid-flight: shutdown must wait for it, not abandon it.
        assert_eq!(manager.shutdown(Duration::from_secs(10)).await, 0);
        let stats = manager.get_webhook_stats(webhook_id).await.unwrap();
        assert_eq!(stats.successful_deliveries, 1);
        assert_eq!(manager.task_panics(), 0);
    }

    #[tokio::test]
    async fn test_shutdown_aborts_deliveries_after_grace_period() {
        let (url, received) = slow_http_server(Duration::from_secs(30)).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        manager
            .add_webhook(WebhookConfig::new("stuck".to_string(), url))
            .await
            .unwrap();

        events.publish_event(completed_event()).await.unwrap();
        wait_for(&received, 1).await;

        let started = std::time::Instant::now();
        assert_eq!(manager.shutdown(Duration::from_millis(100)).await, 1);
        assert!(started.elapsed() < Duration::from_secs(5));
    }

    #[tokio::test]
    async fn test_no_deliveries_after_shutdown() {
        let (url, received) = slow_http_server(Duration::ZERO).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        manager
            .add_webhook(WebhookConfig::new("late".to_string(), url))
            .await
            .unwrap();

        manager.shutdown(Duration::from_secs(1)).await;
        events.publish_event(completed_event()).await.unwrap();
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(received.load(std::sync::atomic::Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn test_listener_delivers_every_event_in_a_burst() {
        let (url, received) = slow_http_server(Duration::ZERO).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        manager
            .add_webhook(WebhookConfig::new("burst".to_string(), url))
            .await
            .unwrap();

        for _ in 0..100 {
            events.publish_event(completed_event()).await.unwrap();
        }

        // Every event published back to back must be delivered.
        wait_for(&received, 100).await;
        manager.shutdown(Duration::from_secs(5)).await;
        assert_eq!(received.load(std::sync::atomic::Ordering::SeqCst), 100);
    }

    #[tokio::test]
    async fn test_listener_keeps_delivering_after_falling_behind() {
        let (url, received) = slow_http_server(Duration::ZERO).await;
        let events = Arc::new(EventManager::new(crate::events::EventConfig {
            max_buffer_size: 4,
            ..Default::default()
        }));
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook = WebhookConfig::new("lagging".to_string(), url);
        let id = webhook.id;
        manager.add_webhook(webhook).await.unwrap();

        // Overrun the 4-event channel so the listener sees RecvError::Lagged.
        for _ in 0..50 {
            events.publish_event(completed_event()).await.unwrap();
        }
        // Every event is either delivered or counted as dropped.
        let mut accounted = false;
        for _ in 0..400 {
            let dropped = manager.get_webhook_stats(id).await.unwrap().dropped_events;
            let delivered = received.load(std::sync::atomic::Ordering::SeqCst) as u64;
            if dropped > 0 && dropped + delivered == 50 {
                accounted = true;
                break;
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        assert!(accounted, "dropped events are counted");
        let before = received.load(std::sync::atomic::Ordering::SeqCst);
        assert!(before >= 1, "some events from the burst must be delivered");

        // The listener must survive the lag and deliver later events.
        events.publish_event(completed_event()).await.unwrap();
        wait_for(&received, before + 1).await;
        manager.shutdown(Duration::from_secs(5)).await;
    }

    #[tokio::test]
    async fn test_panicking_background_task_is_observed() {
        let manager = WebhookManager::new_default(Arc::new(EventManager::new_default())).unwrap();
        assert_eq!(manager.task_panics(), 0);

        manager.deliveries.spawn("delivery", async {
            panic!("expected test panic in a delivery task");
        });
        manager.listeners.spawn("listener", async {
            panic!("expected test panic in a listener task");
        });

        // Let both tasks run (and panic) before shutting down.
        for _ in 0..200 {
            if manager.task_panics() == 2 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        manager.shutdown(Duration::from_secs(5)).await;
        assert_eq!(manager.task_panics(), 2);
    }

    fn template_event() -> JobLifecycleEvent {
        let mut metadata = HashMap::new();
        metadata.insert("region".to_string(), "eu-west-1".to_string());
        JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "emails".to_string(),
            event_type: crate::events::JobLifecycleEventType::Failed,
            priority: crate::priority::JobPriority::High,
            timestamp: Utc::now(),
            processing_time_ms: Some(42),
            error: Some(crate::events::JobError {
                message: "SMTP timeout".to_string(),
                error_type: None,
                details: None,
                retry_attempt: Some(2),
            }),
            payload: Some(serde_json::json!({
                "user": {"id": 7, "tags": ["a", "b"]},
                "note": "said \"hi\""
            })),
            metadata,
        }
    }

    #[test]
    fn test_payload_template_renders_values_with_their_types() {
        let event = template_event();
        let template = PayloadTemplate::parse(
            r#"{
                "job": "{{ event.job_id }}",
                "queue": "{{event.queue_name}}",
                "took": "{{event.processing_time_ms}}",
                "attempt": "{{event.error.retry_attempt}}",
                "user": "{{event.payload.user}}",
                "second_tag": "{{event.payload.user.tags.1}}",
                "region": "{{event.metadata.region}}",
                "missing": "{{event.payload.nope}}",
                "details": "{{event.error.details}}",
                "constant": [1, true, null, "plain"],
                "whole": "{{event}}"
            }"#,
        )
        .unwrap();
        let rendered = template.render(&event).unwrap();

        assert_eq!(rendered["job"], event.job_id.to_string());
        assert_eq!(rendered["queue"], "emails");
        assert_eq!(rendered["took"], 42);
        assert_eq!(rendered["attempt"], 2);
        assert_eq!(
            rendered["user"],
            serde_json::json!({"id": 7, "tags": ["a", "b"]})
        );
        assert_eq!(rendered["second_tag"], "b");
        assert_eq!(rendered["region"], "eu-west-1");
        assert_eq!(rendered["missing"], serde_json::Value::Null);
        assert_eq!(rendered["details"], serde_json::Value::Null);
        assert_eq!(
            rendered["constant"],
            serde_json::json!([1, true, null, "plain"])
        );
        assert_eq!(rendered["whole"], serde_json::to_value(&event).unwrap());
    }

    #[test]
    fn test_payload_template_interpolates_text() {
        let event = template_event();
        let template = PayloadTemplate::parse(
            r#"{"text": "Job {{event.job_id}} in {{event.queue_name}} is {{event.event_type}} after {{event.processing_time_ms}}ms ({{event.payload.user.tags}}){{event.payload.nope}}"}"#,
        )
        .unwrap();
        let rendered = template.render(&event).unwrap();
        assert_eq!(
            rendered["text"],
            format!(
                "Job {} in emails is failed after 42ms ([\"a\",\"b\"])",
                event.job_id
            )
        );
    }

    #[test]
    fn test_payload_template_escapes_event_data() {
        let mut event = template_event();
        // Data that would break out of the string if it were pasted into the JSON source.
        event.queue_name = r#"q", "injected": true, "x": "\"#.to_string();
        let template = PayloadTemplate::parse(
            r#"{"queue": "{{event.queue_name}}", "text": "queue={{event.queue_name}} note={{event.payload.note}}"}"#,
        )
        .unwrap();
        let rendered = template.render(&event).unwrap();

        // The rendered payload round-trips through its serialized form unchanged.
        let body = serde_json::to_string(&rendered).unwrap();
        let reparsed: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(reparsed, rendered);
        let object = reparsed.as_object().unwrap();
        assert_eq!(object.len(), 2);
        assert!(!object.contains_key("injected"));
        assert_eq!(object["queue"], event.queue_name.as_str());
        assert_eq!(
            object["text"],
            format!("queue={} note=said \"hi\"", event.queue_name)
        );
    }

    #[test]
    fn test_payload_template_rejects_invalid_templates() {
        let invalid = [
            ("not json", "not valid JSON"),
            (
                r#"{"a": "{{event.nope}}"}"#,
                "unknown placeholder '{{event.nope}}'",
            ),
            (r#"{"a": "{{job.id}}"}"#, "unknown placeholder"),
            (r#"{"a": "{{event.queue_name.x}}"}"#, "unknown placeholder"),
            (r#"{"a": "{{event.error.stack}}"}"#, "unknown placeholder"),
            (r#"{"a": "{{event.metadata.a.b}}"}"#, "unknown placeholder"),
            (r#"{"a": "{{event.}}"}"#, "unknown placeholder"),
            (r#"{"a": "{{}}"}"#, "unknown placeholder"),
            (r#"{"a": ["ok", "{{event.job_id"]}"#, "unterminated"),
            (r#"{"{{event.job_id}}": 1}"#, "object keys"),
        ];
        for (template, expected) in invalid {
            match PayloadTemplate::parse(template) {
                Err(crate::HammerworkError::Config(message)) => {
                    assert!(message.contains(expected), "{template}: {message}");
                }
                other => panic!("{template}: expected a config error, got {other:?}"),
            }
        }

        // Plain JSON without placeholders is a valid (constant) template.
        assert!(PayloadTemplate::parse(r#"{"static": [1, 2, 3]}"#).is_ok());
        assert!(PayloadTemplate::parse(r#""{{event}}""#).is_ok());
    }

    #[tokio::test]
    async fn test_add_webhook_rejects_invalid_template() {
        let manager = WebhookManager::new_default(Arc::new(EventManager::new_default())).unwrap();
        let webhook = WebhookConfig::new("bad".to_string(), "http://127.0.0.1:1/".to_string())
            .with_payload_template(r#"{"a": "{{event.unknown}}"}"#);
        match manager.add_webhook(webhook).await {
            Err(crate::HammerworkError::Config(message)) => {
                assert!(message.contains("webhook 'bad'"), "{message}");
                assert!(message.contains("unknown placeholder"), "{message}");
            }
            other => panic!("expected a config error, got {other:?}"),
        }
        assert!(manager.list_webhooks().await.is_empty());

        // An invalid update leaves the existing webhook untouched.
        let good = WebhookConfig::new("good".to_string(), "http://127.0.0.1:1/".to_string());
        let id = good.id;
        manager.add_webhook(good.clone()).await.unwrap();
        let mut bad_update = good.with_payload_template("{");
        bad_update.id = id;
        assert!(manager.update_webhook(bad_update).await.is_err());
        let current = manager.get_webhook(id).await.unwrap();
        assert!(current.payload_template.is_none());
        manager.shutdown(Duration::from_secs(1)).await;
    }

    /// An HTTP server that answers 200 and forwards every request body.
    async fn capturing_http_server() -> (String, tokio::sync::mpsc::UnboundedReceiver<String>) {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}/hook", listener.local_addr().unwrap());
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                let tx = tx.clone();
                tokio::spawn(async move {
                    let mut data = Vec::new();
                    let mut buf = [0u8; 4096];
                    loop {
                        let Ok(n) = socket.read(&mut buf).await else {
                            return;
                        };
                        if n == 0 {
                            return;
                        }
                        data.extend_from_slice(&buf[..n]);
                        let text = String::from_utf8_lossy(&data).to_string();
                        let Some(header_end) = text.find("\r\n\r\n") else {
                            continue;
                        };
                        let length = text[..header_end]
                            .lines()
                            .find_map(|line| {
                                let (name, value) = line.split_once(':')?;
                                name.eq_ignore_ascii_case("content-length")
                                    .then(|| value.trim().parse::<usize>().ok())?
                            })
                            .unwrap_or(0);
                        if data.len() >= header_end + 4 + length {
                            let body = String::from_utf8_lossy(
                                &data[header_end + 4..header_end + 4 + length],
                            )
                            .to_string();
                            let _ = tx.send(body);
                            let _ = socket
                                .write_all(
                                    b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
                                )
                                .await;
                            return;
                        }
                    }
                });
            }
        });
        (url, rx)
    }

    #[tokio::test]
    async fn test_delivery_sends_rendered_template() {
        let (url, mut bodies) = capturing_http_server().await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook = WebhookConfig::new("templated".to_string(), url).with_payload_template(
            r#"{"text": "{{event.queue_name}} job {{event.job_id}}", "job_id": "{{event.job_id}}"}"#,
        );
        manager.add_webhook(webhook).await.unwrap();

        let event = completed_event();
        events.publish_event(event.clone()).await.unwrap();
        let body = tokio::time::timeout(Duration::from_secs(10), bodies.recv())
            .await
            .expect("webhook was not delivered")
            .unwrap();
        manager.shutdown(Duration::from_secs(5)).await;

        let body: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(
            body,
            serde_json::json!({
                "text": format!("shutdown_queue job {}", event.job_id),
                "job_id": event.job_id.to_string(),
            })
        );
    }

    /// A request received by [`recording_http_server`].
    #[derive(Debug)]
    struct Recorded {
        request_line: String,
        headers: HashMap<String, String>,
        body: String,
    }

    /// A local HTTP server answering every request with `status` (after `delay`) and
    /// recording the requests it receives.
    async fn recording_http_server(
        status: u16,
        delay: Duration,
    ) -> (String, tokio::sync::mpsc::UnboundedReceiver<Recorded>) {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}/hook", listener.local_addr().unwrap());
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        tokio::spawn(async move {
            while let Ok((mut socket, _)) = listener.accept().await {
                let tx = tx.clone();
                tokio::spawn(async move {
                    let mut buffer = Vec::new();
                    let mut chunk = [0u8; 4096];
                    let (head, body) = loop {
                        if let Some(end) = buffer.windows(4).position(|w| w == b"\r\n\r\n") {
                            let head = String::from_utf8_lossy(&buffer[..end]).to_string();
                            let length: usize = head
                                .lines()
                                .find_map(|l| {
                                    l.to_lowercase()
                                        .strip_prefix("content-length:")
                                        .map(|v| v.trim().to_string())
                                })
                                .and_then(|v| v.parse().ok())
                                .unwrap_or(0);
                            if buffer.len() >= end + 4 + length {
                                break (head, buffer[end + 4..end + 4 + length].to_vec());
                            }
                        }
                        match socket.read(&mut chunk).await {
                            Ok(0) | Err(_) => return,
                            Ok(n) => buffer.extend_from_slice(&chunk[..n]),
                        }
                    };
                    let mut lines = head.lines();
                    let request_line = lines.next().unwrap_or_default().to_string();
                    let headers = lines
                        .filter_map(|l| l.split_once(':'))
                        .map(|(k, v)| (k.trim().to_lowercase(), v.trim().to_string()))
                        .collect();
                    let _ = tx.send(Recorded {
                        request_line,
                        headers,
                        body: String::from_utf8_lossy(&body).to_string(),
                    });
                    tokio::time::sleep(delay).await;
                    let response = format!(
                        "HTTP/1.1 {status} X\r\ncontent-length: 4\r\nconnection: close\r\n\r\nnope"
                    );
                    let _ = socket.write_all(response.as_bytes()).await;
                });
            }
        });
        (url, rx)
    }

    async fn next_recorded(rx: &mut tokio::sync::mpsc::UnboundedReceiver<Recorded>) -> Recorded {
        tokio::time::timeout(Duration::from_secs(10), rx.recv())
            .await
            .expect("no request received")
            .unwrap()
    }

    fn no_retries() -> RetryPolicy {
        RetryPolicy {
            max_attempts: 1,
            ..RetryPolicy::default()
        }
    }

    /// Add `webhook`, publish one event and return the request the endpoint received.
    async fn deliver_one(
        webhook: WebhookConfig,
        url_rx: &mut tokio::sync::mpsc::UnboundedReceiver<Recorded>,
    ) -> Recorded {
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        manager.add_webhook(webhook).await.unwrap();
        events.publish_event(completed_event()).await.unwrap();
        let recorded = next_recorded(url_rx).await;
        manager.shutdown(Duration::from_secs(5)).await;
        recorded
    }

    #[tokio::test]
    async fn a_zero_max_attempts_still_delivers_once() {
        let (url, mut rx) = recording_http_server(200, Duration::ZERO).await;
        let webhook = WebhookConfig::new("zero".to_string(), url).with_retry_policy(RetryPolicy {
            max_attempts: 0,
            ..RetryPolicy::default()
        });
        let recorded = deliver_one(webhook, &mut rx).await;
        assert!(recorded.request_line.starts_with("POST"));
    }

    #[tokio::test]
    async fn test_auth_methods_and_signature() {
        let (url, mut rx) = recording_http_server(200, Duration::ZERO).await;
        let cases = [
            (
                WebhookAuth::Bearer {
                    token: "t0ken".to_string(),
                },
                "authorization",
                "Bearer t0ken".to_string(),
            ),
            (
                WebhookAuth::Basic {
                    username: "user".to_string(),
                    password: "pass".to_string(),
                },
                "authorization",
                "Basic dXNlcjpwYXNz".to_string(),
            ),
            (
                WebhookAuth::ApiKey {
                    header_name: "X-Api-Key".to_string(),
                    api_key: "k123".to_string(),
                },
                "x-api-key",
                "k123".to_string(),
            ),
            (
                WebhookAuth::Custom {
                    headers: HashMap::from([("X-Custom".to_string(), "c".to_string())]),
                },
                "x-custom",
                "c".to_string(),
            ),
        ];
        for (auth, header, expected) in cases {
            let webhook = WebhookConfig::new("auth".to_string(), url.clone())
                .with_auth(auth)
                .with_retry_policy(no_retries());
            let recorded = deliver_one(webhook, &mut rx).await;
            assert_eq!(
                recorded.headers.get(header),
                Some(&expected),
                "{recorded:?}"
            );
        }

        // Method, extra headers and the HMAC signature of the body.
        let webhook = WebhookConfig::new("signed".to_string(), url.clone())
            .with_method(HttpMethod::Put)
            .with_header("X-Team".to_string(), "queues".to_string())
            .with_secret("s3cret".to_string())
            .with_retry_policy(no_retries());
        let recorded = deliver_one(webhook, &mut rx).await;
        assert!(
            recorded.request_line.starts_with("PUT /hook"),
            "{recorded:?}"
        );
        assert_eq!(
            recorded.headers.get("x-team").map(String::as_str),
            Some("queues")
        );
        use hmac::{Hmac, Mac};
        let mut mac = Hmac::<sha2::Sha256>::new_from_slice(b"s3cret").unwrap();
        let body: serde_json::Value = serde_json::from_str(&recorded.body).unwrap();
        mac.update(serde_json::to_string(&body).unwrap().as_bytes());
        let expected = format!("sha256={}", hex::encode(mac.finalize().into_bytes()));
        assert_eq!(
            recorded.headers.get("x-hammerwork-signature"),
            Some(&expected)
        );

        let webhook = WebhookConfig::new("patch".to_string(), url)
            .with_method(HttpMethod::Patch)
            .with_retry_policy(no_retries());
        let recorded = deliver_one(webhook, &mut rx).await;
        assert!(recorded.request_line.starts_with("PATCH /hook"));
    }

    #[tokio::test]
    async fn test_non_retryable_status_and_timeout_are_failures() {
        // 404 is not in retry_on_status_codes: one attempt only.
        let (url, mut rx) = recording_http_server(404, Duration::ZERO).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook = WebhookConfig::new("404".to_string(), url).with_retry_policy(RetryPolicy {
            max_attempts: 3,
            initial_delay_secs: 0,
            ..RetryPolicy::default()
        });
        let id = webhook.id;
        manager.add_webhook(webhook).await.unwrap();
        events.publish_event(completed_event()).await.unwrap();
        next_recorded(&mut rx).await;
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert!(rx.try_recv().is_err(), "a 404 is not retried");
        let stats = manager.get_webhook_stats(id).await.unwrap();
        assert_eq!((stats.total_attempts, stats.failed_deliveries), (1, 1));
        assert!(stats.last_failure_at.is_some());
        assert_eq!(manager.get_all_webhook_stats().await.len(), 1);
        manager.shutdown(Duration::from_secs(5)).await;

        // An endpoint slower than the webhook timeout fails the delivery.
        let (url, mut rx) = recording_http_server(200, Duration::from_secs(5)).await;
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook = WebhookConfig::new("slow".to_string(), url)
            .with_timeout_secs(1)
            .with_retry_policy(no_retries());
        let id = webhook.id;
        manager.add_webhook(webhook).await.unwrap();
        events.publish_event(completed_event()).await.unwrap();
        next_recorded(&mut rx).await;
        let mut failed = false;
        for _ in 0..100 {
            if manager
                .get_webhook_stats(id)
                .await
                .is_some_and(|s| s.failed_deliveries == 1)
            {
                failed = true;
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert!(failed, "the timed-out delivery is recorded as failed");
        manager.shutdown(Duration::from_secs(5)).await;
    }

    /// Disabling a webhook stops delivery, enabling it resumes, updating replaces it,
    /// and unknown ids are errors.
    #[tokio::test]
    async fn test_enable_disable_update_webhook() {
        let (url, mut rx) = recording_http_server(200, Duration::ZERO).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook =
            WebhookConfig::new("toggle".to_string(), url.clone()).with_retry_policy(no_retries());
        let id = webhook.id;
        manager.add_webhook(webhook.clone()).await.unwrap();

        manager.disable_webhook(id).await.unwrap();
        assert_eq!(manager.get_stats().await.active_webhooks, 0);
        events.publish_event(completed_event()).await.unwrap();
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert!(rx.try_recv().is_err(), "disabled webhooks send nothing");

        manager.enable_webhook(id).await.unwrap();
        let event = completed_event();
        let event_id = event.event_id;
        events.publish_event(event).await.unwrap();
        let recorded = next_recorded(&mut rx).await;
        assert!(
            recorded.body.contains(&event_id.to_string()),
            "{recorded:?}"
        );

        // Updating replaces the configuration (here: the request method).
        manager
            .update_webhook(webhook.with_method(HttpMethod::Put))
            .await
            .unwrap();
        assert_eq!(
            manager.get_webhook(id).await.unwrap().method,
            HttpMethod::Put
        );
        events.publish_event(completed_event()).await.unwrap();
        assert!(next_recorded(&mut rx).await.request_line.starts_with("PUT"));

        let unknown = Uuid::new_v4();
        for result in [
            manager.enable_webhook(unknown).await,
            manager.disable_webhook(unknown).await,
        ] {
            assert!(result.unwrap_err().to_string().contains("not found"));
        }
        manager.shutdown(Duration::from_secs(5)).await;
    }

    /// Wait until `condition` holds, polling every 10 ms for up to 10 seconds.
    async fn eventually<F, Fut>(what: &str, mut condition: F)
    where
        F: FnMut() -> Fut,
        Fut: std::future::Future<Output = bool>,
    {
        for _ in 0..1000 {
            if condition().await {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("timed out waiting until {what}");
    }

    /// #114: deliveries are counted by final outcome per event, attempts separately.
    #[tokio::test]
    async fn test_stats_count_final_outcomes_not_attempts() {
        let id = Uuid::new_v4();
        let stats = Arc::new(RwLock::new(HashMap::from([(id, WebhookStats::empty(id))])));
        let attempt = |success: bool| WebhookDelivery {
            delivery_id: Uuid::new_v4(),
            webhook_id: id,
            event_id: Uuid::new_v4(),
            status_code: Some(if success { 200 } else { 500 }),
            response_body: None,
            error_message: None,
            attempted_at: Utc::now(),
            duration_ms: Some(10),
            attempt_number: 1,
            success,
        };

        // An event delivered on its third attempt.
        for success in [false, false, true] {
            WebhookManager::update_webhook_stats(&id, &attempt(success), stats.clone()).await;
        }
        WebhookManager::record_delivery_outcome(&id, true, stats.clone()).await;
        {
            let stats = stats.read().await;
            let s = &stats[&id];
            assert_eq!((s.total_attempts, s.failed_attempts), (3, 2));
            assert_eq!((s.successful_deliveries, s.failed_deliveries), (1, 0));
            assert_eq!(s.success_rate, 1.0);
            assert!(s.last_success_at.is_some() && s.last_failure_at.is_some());
        }

        // An event that failed on every attempt.
        for _ in 0..3 {
            WebhookManager::update_webhook_stats(&id, &attempt(false), stats.clone()).await;
        }
        WebhookManager::record_delivery_outcome(&id, false, stats.clone()).await;
        let stats = stats.read().await;
        let s = &stats[&id];
        assert_eq!((s.total_attempts, s.failed_attempts), (6, 5));
        assert_eq!((s.successful_deliveries, s.failed_deliveries), (1, 1));
        assert_eq!(s.success_rate, 0.5);
    }

    /// #114: shutting the manager down takes its listeners off the event manager.
    #[tokio::test]
    async fn test_shutdown_unsubscribes_listeners() {
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        for name in ["a", "b"] {
            manager
                .add_webhook(WebhookConfig::new(
                    name.to_string(),
                    "http://127.0.0.1:1/hook".to_string(),
                ))
                .await
                .unwrap();
        }
        assert_eq!(events.subscription_count().await, 2);
        manager.shutdown(Duration::from_secs(1)).await;
        assert_eq!(events.subscription_count().await, 0);
    }

    /// M9: the subscription exists when `add_webhook` returns, so an event published
    /// immediately afterwards (before the listener task has even run) is delivered.
    #[tokio::test]
    async fn test_event_published_right_after_add_is_delivered() {
        let (url, mut rx) = recording_http_server(200, Duration::ZERO).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let event = completed_event();
        manager
            .add_webhook(WebhookConfig::new("immediate".to_string(), url))
            .await
            .unwrap();
        events.publish_event(event.clone()).await.unwrap();

        let recorded = next_recorded(&mut rx).await;
        assert!(recorded.body.contains(&event.event_id.to_string()));
        manager.shutdown(Duration::from_secs(5)).await;
    }

    /// H11: updating a webhook (or re-adding it under the same id) leaves exactly one
    /// listener, so an event is delivered once, not once per update.
    #[tokio::test]
    async fn test_updates_do_not_duplicate_delivery() {
        let (url, mut rx) = recording_http_server(200, Duration::ZERO).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook =
            WebhookConfig::new("updated".to_string(), url.clone()).with_retry_policy(no_retries());
        let id = webhook.id;
        manager.add_webhook(webhook.clone()).await.unwrap();

        for i in 0..5 {
            manager
                .update_webhook(
                    webhook
                        .clone()
                        .with_header("X-Version".to_string(), i.to_string()),
                )
                .await
                .unwrap();
        }
        // Re-adding and removing then adding under the same id replace the listener too.
        manager.add_webhook(webhook.clone()).await.unwrap();
        manager.remove_webhook(id).await.unwrap();
        manager
            .add_webhook(webhook.with_header("X-Version".to_string(), "last".to_string()))
            .await
            .unwrap();
        assert_eq!(events.subscription_count().await, 1);

        let event = completed_event();
        events.publish_event(event.clone()).await.unwrap();
        let recorded = next_recorded(&mut rx).await;
        assert_eq!(
            recorded.headers.get("x-version").map(String::as_str),
            Some("last")
        );
        assert!(recorded.body.contains(&event.event_id.to_string()));

        // No duplicate arrives, and exactly one attempt was made.
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert!(
            rx.try_recv().is_err(),
            "the event was delivered more than once"
        );
        assert_eq!(
            manager.get_webhook_stats(id).await.unwrap().total_attempts,
            1
        );
        manager.shutdown(Duration::from_secs(5)).await;
    }

    /// An updated configuration applies to the next event without re-adding.
    #[tokio::test]
    async fn test_update_keeps_stats_and_applies_new_filter() {
        let (url, mut rx) = recording_http_server(200, Duration::ZERO).await;
        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook =
            WebhookConfig::new("filtered".to_string(), url).with_retry_policy(no_retries());
        let id = webhook.id;
        manager.add_webhook(webhook.clone()).await.unwrap();
        events.publish_event(completed_event()).await.unwrap();
        next_recorded(&mut rx).await;
        eventually("the first delivery is recorded", || async {
            manager.get_webhook_stats(id).await.unwrap().total_attempts == 1
        })
        .await;

        // Only failed events from now on.
        manager
            .update_webhook(
                webhook.with_filter(
                    EventFilter::new()
                        .with_event_types(vec![crate::events::JobLifecycleEventType::Failed]),
                ),
            )
            .await
            .unwrap();
        events.publish_event(completed_event()).await.unwrap();
        let mut failed = completed_event();
        failed.event_type = crate::events::JobLifecycleEventType::Failed;
        events.publish_event(failed.clone()).await.unwrap();

        let recorded = next_recorded(&mut rx).await;
        assert!(recorded.body.contains(&failed.event_id.to_string()));
        eventually("the second delivery is recorded", || async {
            manager.get_webhook_stats(id).await.unwrap().total_attempts == 2
        })
        .await;
        manager.shutdown(Duration::from_secs(5)).await;
    }

    /// A local HTTP server that holds every request until `gate` has a permit (one per
    /// response), then answers `200 OK`. Returns its URL and a request counter.
    async fn gated_http_server(
        gate: Arc<Semaphore>,
    ) -> (String, Arc<std::sync::atomic::AtomicUsize>) {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}/hook", listener.local_addr().unwrap());
        let received = Arc::new(AtomicUsize::new(0));
        let counter = received.clone();
        tokio::spawn(async move {
            while let Ok((mut socket, _)) = listener.accept().await {
                let counter = counter.clone();
                let gate = gate.clone();
                tokio::spawn(async move {
                    let mut buf = vec![0u8; 16 * 1024];
                    let _ = socket.read(&mut buf).await;
                    counter.fetch_add(1, Ordering::SeqCst);
                    if let Ok(permit) = gate.acquire().await {
                        permit.forget();
                    }
                    let _ = socket
                        .write_all(
                            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
                        )
                        .await;
                });
            }
        });
        (url, received)
    }

    /// H12: with the endpoint stuck, at most `max_pending_deliveries` deliveries exist
    /// for the webhook; the listener waits and the overflow is dropped and counted.
    #[tokio::test]
    async fn test_pending_deliveries_are_bounded() {
        use std::sync::atomic::Ordering;
        let gate = Arc::new(Semaphore::new(0));
        let (url, received) = gated_http_server(gate.clone()).await;
        let events = Arc::new(EventManager::new(crate::events::EventConfig {
            max_buffer_size: 4,
            ..Default::default()
        }));
        let manager = WebhookManager::new(
            events.clone(),
            WebhookManagerConfig {
                max_pending_deliveries: 2,
                ..Default::default()
            },
        )
        .unwrap();
        let webhook = WebhookConfig::new("stuck".to_string(), url).with_retry_policy(no_retries());
        let id = webhook.id;
        manager.add_webhook(webhook).await.unwrap();

        for _ in 0..20 {
            events.publish_event(completed_event()).await.unwrap();
            tokio::task::yield_now().await;
        }
        eventually("two requests are in flight", || async {
            received.load(Ordering::SeqCst) == 2
        })
        .await;
        // The listener is waiting: no third delivery is started.
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(received.load(Ordering::SeqCst), 2);
        assert!(
            manager.deliveries.len() <= 2,
            "{}",
            manager.deliveries.len()
        );

        // Once the endpoint answers, the rest is delivered or counted as dropped.
        gate.add_permits(1_000);
        eventually("every event is delivered or dropped", || async {
            let stats = manager.get_webhook_stats(id).await.unwrap();
            stats.successful_deliveries + stats.dropped_events == 20
        })
        .await;
        let stats = manager.get_webhook_stats(id).await.unwrap();
        assert!(stats.dropped_events > 0, "{stats:?}");
        manager.shutdown(Duration::from_secs(5)).await;
    }

    /// H13: an endpoint that sends its headers and then stalls the body cannot hold a
    /// delivery longer than the webhook's timeout.
    #[tokio::test]
    async fn test_timeout_covers_the_response_body() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}/hook", listener.local_addr().unwrap());
        tokio::spawn(async move {
            while let Ok((mut socket, _)) = listener.accept().await {
                tokio::spawn(async move {
                    let mut buf = vec![0u8; 16 * 1024];
                    let _ = socket.read(&mut buf).await;
                    let _ = socket
                        .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1000\r\n\r\npartial")
                        .await;
                    tokio::time::sleep(Duration::from_secs(60)).await;
                });
            }
        });

        let events = Arc::new(EventManager::new_default());
        let manager = WebhookManager::new_default(events.clone()).unwrap();
        let webhook = WebhookConfig::new("stalling".to_string(), url)
            .with_timeout_secs(1)
            .with_retry_policy(no_retries());
        let id = webhook.id;
        manager.add_webhook(webhook).await.unwrap();
        let started = std::time::Instant::now();
        events.publish_event(completed_event()).await.unwrap();

        eventually("the attempt finishes", || async {
            manager.get_webhook_stats(id).await.unwrap().total_attempts == 1
        })
        .await;
        assert!(started.elapsed() < Duration::from_secs(5));
        // The endpoint did answer 200, so the delivery counts as successful.
        let stats = manager.get_webhook_stats(id).await.unwrap();
        assert_eq!(stats.successful_deliveries, 1);
        assert_eq!(
            manager.shutdown(Duration::ZERO).await,
            0,
            "nothing still running"
        );
    }

    /// H13: only `max_bytes` of a response body are read, even an endless one.
    #[tokio::test]
    async fn test_response_body_read_is_bounded() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}/", listener.local_addr().unwrap());
        tokio::spawn(async move {
            let Ok((mut socket, _)) = listener.accept().await else {
                return;
            };
            let mut buf = vec![0u8; 16 * 1024];
            let _ = socket.read(&mut buf).await;
            let _ = socket
                .write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n")
                .await;
            // An endless body.
            while socket.write_all(b"400\r\n").await.is_ok()
                && socket.write_all(&[b'x'; 1024]).await.is_ok()
                && socket.write_all(b"\r\n").await.is_ok()
            {}
        });

        let response = Client::new().get(&url).send().await.unwrap();
        let body = tokio::time::timeout(Duration::from_secs(5), read_body_limited(response, 10))
            .await
            .expect("reading stops at the limit")
            .unwrap();
        assert_eq!(body, "xxxxxxxxxx... [truncated]");
    }

    /// M4: `timeout_secs = 0` (or missing in a file) uses the manager's default.
    #[test]
    fn test_zero_timeout_uses_manager_default() {
        let config = WebhookManagerConfig {
            default_timeout_secs: 7,
            ..Default::default()
        };
        let webhook = WebhookConfig::new("w".to_string(), "http://x".to_string());
        assert_eq!(
            attempt_timeout(&webhook.clone().with_timeout_secs(3), &config),
            Duration::from_secs(3)
        );
        assert_eq!(
            attempt_timeout(&webhook.clone().with_timeout_secs(0), &config),
            Duration::from_secs(7)
        );

        let mut value = serde_json::to_value(&webhook).unwrap();
        value.as_object_mut().unwrap().remove("timeout_secs");
        let parsed: WebhookConfig = serde_json::from_value(value).unwrap();
        assert_eq!(parsed.timeout_secs, 0);
        assert_eq!(attempt_timeout(&parsed, &config), Duration::from_secs(7));

        let settings = crate::config::WebhookGlobalSettings {
            default_timeout_secs: 9,
            max_pending_deliveries: 5,
            ..Default::default()
        };
        let from_settings = WebhookManagerConfig::from(&settings);
        assert_eq!(from_settings.default_timeout_secs, 9);
        assert_eq!(from_settings.max_pending_deliveries, 5);
    }

    #[test]
    fn test_webhook_config_builders() {
        let filter = EventFilter::new().with_queue_names(vec!["q".to_string()]);
        let webhook = WebhookConfig::new("w".to_string(), "http://x".to_string())
            .with_filter(filter)
            .with_retry_policy(no_retries())
            .enabled(false);
        assert_eq!(webhook.filter.queue_names, vec!["q".to_string()]);
        assert_eq!(webhook.retry_policy.max_attempts, 1);
        assert!(!webhook.enabled);
    }

    /// #81: webhook settings built from their defaults with setters.
    #[test]
    fn test_webhook_settings_builders() {
        let retry = RetryPolicy::default()
            .with_max_attempts(2)
            .with_initial_delay_secs(1)
            .with_max_delay_secs(5)
            .with_backoff_multiplier(3.0)
            .with_retry_on_status_codes(vec![503]);
        assert_eq!(
            (
                retry.max_attempts,
                retry.initial_delay_secs,
                retry.max_delay_secs
            ),
            (2, 1, 5)
        );
        assert_eq!(retry.backoff_multiplier, 3.0);
        assert!(retry.should_retry_status(503) && !retry.should_retry_status(500));

        let manager = WebhookManagerConfig::default()
            .with_max_concurrent_deliveries(3)
            .with_max_response_body_size(100)
            .with_log_deliveries(false)
            .with_default_timeout_secs(4)
            .with_max_pending_deliveries(6)
            .with_user_agent("acme");
        assert_eq!(manager.max_concurrent_deliveries, 3);
        assert_eq!(manager.max_response_body_size, 100);
        assert!(!manager.log_deliveries);
        assert_eq!(manager.default_timeout_secs, 4);
        assert_eq!(manager.max_pending_deliveries, 6);
        assert_eq!(manager.user_agent, "acme");
    }
}