queuerious_lapin/

config.rs

//! Lapin adapter configuration.

use std::sync::Arc;

use lapin::types::AMQPValue;

/// Configuration for the Queuerious lapin adapter.
#[derive(Clone)]
#[allow(clippy::type_complexity)]
pub struct LapinAdapterConfig {
    /// Default job type when not extractable from the message (default: "unknown").
    pub default_job_type: String,
    /// Whether to include the message payload in events (default: false).
    /// May contain PII — enable with caution.
    pub include_payload: bool,
    /// Maximum payload size in bytes to include (default: 65536 / 64 KB).
    /// Payloads exceeding this limit are replaced with a descriptive placeholder.
    pub max_payload_size: usize,
    /// Custom function to extract the job type from a delivery.
    pub job_type_extractor: Option<Arc<dyn Fn(&lapin::message::Delivery) -> String + Send + Sync>>,
    /// Custom function to extract the job ID from a delivery.
    /// Defaults to using the delivery tag as a string.
    pub job_id_extractor: Option<Arc<dyn Fn(&lapin::message::Delivery) -> String + Send + Sync>>,
    /// Custom function to extract metadata from a delivery.
    /// When set, this is used instead of the default metadata extraction.
    pub metadata_extractor:
        Option<Arc<dyn Fn(&lapin::message::Delivery) -> serde_json::Value + Send + Sync>>,
    /// Whether to include default AMQP metadata when no custom metadata_extractor is set (default: true).
    pub include_default_metadata: bool,
    /// Custom function to extract the correlation ID from a delivery.
    /// When not set, the default extraction tries: AMQP `correlation_id` property,
    /// `x-correlation-id` header, then JSON payload fields (`correlation_id`, `correlationId`).
    pub correlation_id_extractor:
        Option<Arc<dyn Fn(&lapin::message::Delivery) -> Option<String> + Send + Sync>>,
    /// Maximum retry attempts for this consumer. When set, included in `job_started`
    /// events so the server can track attempt progress.
    pub max_attempts: Option<i32>,
}

impl Default for LapinAdapterConfig {
    fn default() -> Self {
        Self {
            default_job_type: "unknown".to_string(),
            include_payload: false,
            max_payload_size: 65_536,
            job_type_extractor: None,
            job_id_extractor: None,
            metadata_extractor: None,
            include_default_metadata: true,
            correlation_id_extractor: None,
            max_attempts: None,
        }
    }
}

impl std::fmt::Debug for LapinAdapterConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("LapinAdapterConfig")
            .field("default_job_type", &self.default_job_type)
            .field("include_payload", &self.include_payload)
            .field("max_payload_size", &self.max_payload_size)
            .field(
                "job_type_extractor",
                &self.job_type_extractor.as_ref().map(|_| "..."),
            )
            .field(
                "job_id_extractor",
                &self.job_id_extractor.as_ref().map(|_| "..."),
            )
            .field(
                "metadata_extractor",
                &self.metadata_extractor.as_ref().map(|_| "..."),
            )
            .field("include_default_metadata", &self.include_default_metadata)
            .field(
                "correlation_id_extractor",
                &self.correlation_id_extractor.as_ref().map(|_| "..."),
            )
            .field("max_attempts", &self.max_attempts)
            .finish()
    }
}

/// Extract the `x-retry-count` header from a delivery's AMQP properties.
///
/// Returns the retry count as an `i32` if the header is present and holds an
/// integer value, or `None` for first-attempt deliveries (no header).
///
/// Used by [`TrackedConsumer`] to propagate the current attempt number
/// (`retry_count + 1`) in `job_started` events.
pub(crate) fn extract_retry_count(delivery: &lapin::message::Delivery) -> Option<i32> {
    delivery
        .properties
        .headers()
        .as_ref()
        .and_then(|h| h.inner().get("x-retry-count"))
        .and_then(|v| match v {
            AMQPValue::LongLongInt(n) => i32::try_from(*n).ok(),
            AMQPValue::ShortShortInt(n) => Some(i32::from(*n)),
            AMQPValue::ShortInt(n) => Some(i32::from(*n)),
            AMQPValue::LongInt(n) => Some(*n),
            _ => None,
        })
}

/// Extract a string value from an AMQP header value (`LongString` or `ShortString`).
fn amqp_value_as_string(value: &AMQPValue) -> Option<String> {
    match value {
        AMQPValue::LongString(s) => {
            let s = String::from_utf8_lossy(s.as_bytes()).into_owned();
            if s.is_empty() {
                None
            } else {
                Some(s)
            }
        }
        AMQPValue::ShortString(s) => {
            let s = s.as_str();
            if s.is_empty() {
                None
            } else {
                Some(s.to_owned())
            }
        }
        _ => None,
    }
}

/// Look up a header by key and extract its string value.
fn header_string(properties: &lapin::BasicProperties, key: &str) -> Option<String> {
    properties
        .headers()
        .as_ref()
        .and_then(|table| table.inner().get(key))
        .and_then(amqp_value_as_string)
}

/// Smart default job type extraction from AMQP message metadata.
///
/// Tries multiple sources in priority order:
/// 1. AMQP `type` property (`properties.kind()`)
/// 2. `x-job-type` header (Queuerious convention)
/// 3. `job_type` header (common convention)
/// 4. Routing key (often encodes type info, e.g. `"emails.send"`)
/// 5. Falls back to `default`
pub(crate) fn extract_job_type(
    properties: &lapin::BasicProperties,
    routing_key: &lapin::types::ShortString,
    default: &str,
) -> String {
    // 1. AMQP type property (most standard)
    if let Some(kind) = properties.kind().as_ref() {
        let s = kind.as_str();
        if !s.is_empty() {
            return s.to_owned();
        }
    }

    // 2. x-job-type header (Queuerious convention)
    if let Some(jt) = header_string(properties, "x-job-type") {
        return jt;
    }

    // 3. job_type header (common convention)
    if let Some(jt) = header_string(properties, "job_type") {
        return jt;
    }

    // 4. Routing key
    let rk = routing_key.as_str();
    if !rk.is_empty() {
        return rk.to_owned();
    }

    // 5. Final fallback
    default.to_owned()
}

/// Smart default job ID extraction from AMQP message metadata and payload.
///
/// Tries multiple sources in priority order:
/// 1. AMQP `message_id` property (standard AMQP for message identity)
/// 2. `x-job-id` header (Queuerious convention)
/// 3. JSON payload fields: `job_id`, `jobId` (top-level, then one level deep
///    for externally-tagged enums like `{"CreateEvent": {"job_id": "..."}}`)
/// 4. Falls back to `delivery_tag` as a string
///
/// Pass a pre-parsed `serde_json::Value` to avoid redundant JSON parsing when
/// multiple extractors need to inspect the same payload.
pub(crate) fn extract_job_id(
    delivery: &lapin::message::Delivery,
    parsed_payload: Option<&serde_json::Value>,
) -> String {
    // 1. AMQP message_id property (most standard)
    if let Some(msg_id) = delivery.properties.message_id().as_ref() {
        let s = msg_id.as_str();
        if !s.is_empty() {
            return s.to_owned();
        }
    }

    // 2. x-job-id header (Queuerious convention)
    if let Some(job_id) = header_string(&delivery.properties, "x-job-id") {
        return job_id;
    }

    // 3. JSON payload (common field names)
    if let Some(id) = extract_string_field_from_payload(parsed_payload, &["job_id", "jobId"]) {
        return id;
    }

    // 4. Delivery tag fallback
    delivery.delivery_tag.to_string()
}

/// Smart default correlation ID extraction from AMQP message metadata and payload.
///
/// Tries multiple sources in priority order:
/// 1. AMQP `correlation_id` property (standard AMQP)
/// 2. `x-correlation-id` header
/// 3. JSON payload fields: `correlation_id`, `correlationId` (top-level, then one
///    level deep for externally-tagged enums)
///
/// Pass a pre-parsed `serde_json::Value` to avoid redundant JSON parsing when
/// multiple extractors need to inspect the same payload.
pub(crate) fn extract_correlation_id(
    delivery: &lapin::message::Delivery,
    parsed_payload: Option<&serde_json::Value>,
) -> Option<String> {
    // 1. AMQP correlation_id property (standard AMQP)
    if let Some(cid) = delivery.properties.correlation_id().as_ref() {
        let s = cid.as_str();
        if !s.is_empty() {
            return Some(s.to_owned());
        }
    }

    // 2. x-correlation-id header
    if let Some(cid) = header_string(&delivery.properties, "x-correlation-id") {
        return Some(cid);
    }

    // 3. JSON payload (common field names)
    extract_string_field_from_payload(parsed_payload, &["correlation_id", "correlationId"])
}

/// Try to extract a string field from a pre-parsed JSON value by checking multiple field names.
///
/// Handles both flat payloads (`{"job_id": "..."}`) and externally-tagged enums
/// (`{"VariantName": {"job_id": "..."}}`).
///
/// Returns `None` if `parsed` is `None` (e.g. non-JSON payload or parse failure).
fn extract_string_field_from_payload(
    parsed: Option<&serde_json::Value>,
    field_names: &[&str],
) -> Option<String> {
    let value = parsed?;

    // Try top-level fields first
    if let Some(id) = try_extract_string_field(value, field_names) {
        return Some(id);
    }

    // For externally-tagged enums: {"VariantName": {"field": ...}}
    // Check if root is an object with a single key containing an inner object
    if let Some(obj) = value.as_object() {
        if obj.len() == 1 {
            if let Some(inner) = obj.values().next() {
                return try_extract_string_field(inner, field_names);
            }
        }
    }

    None
}

/// Try to extract one of the given field names as a non-empty string from a JSON value.
///
/// Handles both string and numeric values (many systems use integer job IDs).
fn try_extract_string_field(value: &serde_json::Value, field_names: &[&str]) -> Option<String> {
    for key in field_names {
        match value.get(*key) {
            Some(serde_json::Value::String(s)) if !s.is_empty() => return Some(s.clone()),
            Some(serde_json::Value::Number(n)) => return Some(n.to_string()),
            _ => continue,
        }
    }
    None
}

/// Extracts default metadata from AMQP message properties and delivery info.
///
/// Includes standard AMQP properties (message_id, app_id, content_type, etc.),
/// delivery metadata (routing_key, exchange), and user-defined headers
/// (excluding internal RabbitMQ headers like `x-death`).
pub fn extract_default_metadata(delivery: &lapin::message::Delivery) -> serde_json::Value {
    extract_metadata_from_parts(
        &delivery.properties,
        delivery.routing_key.as_str(),
        delivery.exchange.as_str(),
    )
}

/// Core metadata extraction from AMQP properties and delivery routing info.
///
/// This is the inner implementation used by [`extract_default_metadata`]. It is
/// separated so that the property-level logic can be unit-tested without
/// constructing a full `lapin::message::Delivery` (which requires internal
/// channel state).
fn extract_metadata_from_parts(
    props: &lapin::BasicProperties,
    routing_key: &str,
    exchange: &str,
) -> serde_json::Value {
    let mut meta = serde_json::Map::new();

    if let Some(msg_id) = props.message_id().as_ref() {
        let s = msg_id.as_str();
        if !s.is_empty() {
            meta.insert("message_id".into(), serde_json::Value::String(s.to_owned()));
        }
    }
    if let Some(app_id) = props.app_id().as_ref() {
        let s = app_id.as_str();
        if !s.is_empty() {
            meta.insert("app_id".into(), serde_json::Value::String(s.to_owned()));
        }
    }
    if let Some(content_type) = props.content_type().as_ref() {
        let s = content_type.as_str();
        if !s.is_empty() {
            meta.insert(
                "content_type".into(),
                serde_json::Value::String(s.to_owned()),
            );
        }
    }
    if let Some(content_encoding) = props.content_encoding().as_ref() {
        let s = content_encoding.as_str();
        if !s.is_empty() {
            meta.insert(
                "content_encoding".into(),
                serde_json::Value::String(s.to_owned()),
            );
        }
    }
    if let Some(priority) = props.priority() {
        meta.insert("priority".into(), serde_json::json!(*priority));
    }
    if let Some(timestamp) = props.timestamp() {
        meta.insert("timestamp".into(), serde_json::json!(*timestamp));
    }
    if let Some(reply_to) = props.reply_to().as_ref() {
        let s = reply_to.as_str();
        if !s.is_empty() {
            meta.insert("reply_to".into(), serde_json::Value::String(s.to_owned()));
        }
    }
    if let Some(expiration) = props.expiration().as_ref() {
        let s = expiration.as_str();
        if !s.is_empty() {
            meta.insert("expiration".into(), serde_json::Value::String(s.to_owned()));
        }
    }

    // Delivery metadata
    if !routing_key.is_empty() {
        meta.insert(
            "routing_key".into(),
            serde_json::Value::String(routing_key.to_owned()),
        );
    }
    if !exchange.is_empty() {
        meta.insert(
            "exchange".into(),
            serde_json::Value::String(exchange.to_owned()),
        );
    }

    // User-defined headers (exclude internal RabbitMQ headers)
    if let Some(headers) = props.headers().as_ref() {
        let mut user_headers = serde_json::Map::new();
        for (key, value) in headers.inner() {
            let key_str = key.as_str();
            if key_str.starts_with("x-death")
                || key_str == "x-first-death-exchange"
                || key_str == "x-first-death-queue"
                || key_str == "x-first-death-reason"
            {
                continue;
            }
            if let Some(s) = amqp_value_as_string(value) {
                user_headers.insert(key_str.to_owned(), serde_json::Value::String(s));
            }
        }
        if !user_headers.is_empty() {
            meta.insert("headers".into(), serde_json::Value::Object(user_headers));
        }
    }

    serde_json::Value::Object(meta)
}

#[cfg(test)]
mod tests {
    use super::*;
    use lapin::types::{FieldTable, LongString, ShortString};
    use lapin::BasicProperties;

    #[test]
    fn extracts_from_amqp_type_property() {
        let props = BasicProperties::default().with_type("SendEmail".into());
        let rk = ShortString::from("");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "SendEmail");
    }

    #[test]
    fn extracts_from_x_job_type_header() {
        let mut headers = FieldTable::default();
        headers.insert(
            "x-job-type".into(),
            AMQPValue::LongString(LongString::from("ProcessPayment")),
        );
        let props = BasicProperties::default().with_headers(headers);
        let rk = ShortString::from("some.routing.key");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "ProcessPayment");
    }

    #[test]
    fn extracts_from_job_type_header() {
        let mut headers = FieldTable::default();
        headers.insert(
            "job_type".into(),
            AMQPValue::LongString(LongString::from("GenerateReport")),
        );
        let props = BasicProperties::default().with_headers(headers);
        let rk = ShortString::from("");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "GenerateReport");
    }

    #[test]
    fn extracts_from_routing_key() {
        let props = BasicProperties::default();
        let rk = ShortString::from("orders.process");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "orders.process");
    }

    #[test]
    fn falls_back_to_default() {
        let props = BasicProperties::default();
        let rk = ShortString::from("");
        assert_eq!(extract_job_type(&props, &rk, "fallback"), "fallback");
    }

    #[test]
    fn amqp_type_takes_priority_over_headers_and_routing_key() {
        let mut headers = FieldTable::default();
        headers.insert(
            "x-job-type".into(),
            AMQPValue::LongString(LongString::from("from_header")),
        );
        let props = BasicProperties::default()
            .with_type("from_type".into())
            .with_headers(headers);
        let rk = ShortString::from("from_routing_key");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "from_type");
    }

    #[test]
    fn x_job_type_takes_priority_over_job_type_header() {
        let mut headers = FieldTable::default();
        headers.insert(
            "x-job-type".into(),
            AMQPValue::LongString(LongString::from("from_x_header")),
        );
        headers.insert(
            "job_type".into(),
            AMQPValue::LongString(LongString::from("from_generic_header")),
        );
        let props = BasicProperties::default().with_headers(headers);
        let rk = ShortString::from("");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "from_x_header");
    }

    #[test]
    fn handles_short_string_header_values() {
        let mut headers = FieldTable::default();
        headers.insert(
            "x-job-type".into(),
            AMQPValue::ShortString("ShortVal".into()),
        );
        let props = BasicProperties::default().with_headers(headers);
        let rk = ShortString::from("");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "ShortVal");
    }

    #[test]
    fn skips_empty_amqp_type() {
        let props = BasicProperties::default().with_type("".into());
        let rk = ShortString::from("fallback_rk");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "fallback_rk");
    }

    #[test]
    fn skips_empty_header_values() {
        let mut headers = FieldTable::default();
        headers.insert(
            "x-job-type".into(),
            AMQPValue::LongString(LongString::from("")),
        );
        let props = BasicProperties::default().with_headers(headers);
        let rk = ShortString::from("routing_fallback");
        assert_eq!(extract_job_type(&props, &rk, "unknown"), "routing_fallback");
    }

    // --- extract_metadata_from_parts tests ---

    #[test]
    fn metadata_extracts_standard_amqp_properties() {
        let props = BasicProperties::default()
            .with_app_id("my-app".into())
            .with_message_id("msg-123".into())
            .with_content_type("application/json".into())
            .with_content_encoding("utf-8".into())
            .with_reply_to("reply-queue".into())
            .with_expiration("60000".into())
            .with_priority(5)
            .with_timestamp(1700000000);

        let meta = extract_metadata_from_parts(&props, "", "");
        let obj = meta.as_object().unwrap();

        assert_eq!(obj.get("app_id").unwrap(), "my-app");
        assert_eq!(obj.get("message_id").unwrap(), "msg-123");
        assert_eq!(obj.get("content_type").unwrap(), "application/json");
        assert_eq!(obj.get("content_encoding").unwrap(), "utf-8");
        assert_eq!(obj.get("reply_to").unwrap(), "reply-queue");
        assert_eq!(obj.get("expiration").unwrap(), "60000");
        assert_eq!(obj.get("priority").unwrap(), 5);
        assert_eq!(obj.get("timestamp").unwrap(), 1700000000u64);
    }

    #[test]
    fn metadata_includes_routing_key_and_exchange() {
        let props = BasicProperties::default();
        let meta = extract_metadata_from_parts(&props, "orders.process", "my-exchange");
        let obj = meta.as_object().unwrap();

        assert_eq!(obj.get("routing_key").unwrap(), "orders.process");
        assert_eq!(obj.get("exchange").unwrap(), "my-exchange");
    }

    #[test]
    fn metadata_empty_for_default_properties_and_empty_routing() {
        let props = BasicProperties::default();
        let meta = extract_metadata_from_parts(&props, "", "");
        let obj = meta.as_object().unwrap();

        assert!(obj.is_empty(), "Expected empty metadata, got: {obj:?}");
    }

    #[test]
    fn metadata_skips_empty_string_properties() {
        let props = BasicProperties::default()
            .with_app_id("".into())
            .with_message_id("".into())
            .with_content_type("".into())
            .with_reply_to("".into())
            .with_expiration("".into());

        let meta = extract_metadata_from_parts(&props, "", "");
        let obj = meta.as_object().unwrap();

        assert!(
            obj.is_empty(),
            "Expected empty metadata for empty string props, got: {obj:?}"
        );
    }

    #[test]
    fn metadata_filters_internal_rabbitmq_headers() {
        let mut headers = FieldTable::default();
        headers.insert(
            "x-death".into(),
            AMQPValue::LongString(LongString::from("death-info")),
        );
        headers.insert(
            "x-death-count".into(),
            AMQPValue::LongString(LongString::from("3")),
        );
        headers.insert(
            "x-first-death-exchange".into(),
            AMQPValue::LongString(LongString::from("dlx")),
        );
        headers.insert(
            "x-first-death-queue".into(),
            AMQPValue::LongString(LongString::from("dlq")),
        );
        headers.insert(
            "x-first-death-reason".into(),
            AMQPValue::LongString(LongString::from("rejected")),
        );
        headers.insert(
            "custom-header".into(),
            AMQPValue::LongString(LongString::from("my-value")),
        );
        headers.insert(
            "correlation-id".into(),
            AMQPValue::LongString(LongString::from("abc-123")),
        );
        let props = BasicProperties::default().with_headers(headers);

        let meta = extract_metadata_from_parts(&props, "", "");
        let obj = meta.as_object().unwrap();
        let user_headers = obj.get("headers").unwrap().as_object().unwrap();

        // Internal headers must be filtered out
        assert!(!user_headers.contains_key("x-death"));
        assert!(!user_headers.contains_key("x-death-count"));
        assert!(!user_headers.contains_key("x-first-death-exchange"));
        assert!(!user_headers.contains_key("x-first-death-queue"));
        assert!(!user_headers.contains_key("x-first-death-reason"));

        // User headers must be preserved
        assert_eq!(user_headers.get("custom-header").unwrap(), "my-value");
        assert_eq!(user_headers.get("correlation-id").unwrap(), "abc-123");
    }

    #[test]
    fn metadata_includes_short_string_headers() {
        let mut headers = FieldTable::default();
        headers.insert("tenant-id".into(), AMQPValue::ShortString("org_123".into()));
        let props = BasicProperties::default().with_headers(headers);

        let meta = extract_metadata_from_parts(&props, "", "");
        let obj = meta.as_object().unwrap();
        let user_headers = obj.get("headers").unwrap().as_object().unwrap();

        assert_eq!(user_headers.get("tenant-id").unwrap(), "org_123");
    }

    #[test]
    fn metadata_omits_headers_section_when_all_filtered() {
        let mut headers = FieldTable::default();
        headers.insert(
            "x-death".into(),
            AMQPValue::LongString(LongString::from("death-info")),
        );
        headers.insert(
            "x-first-death-exchange".into(),
            AMQPValue::LongString(LongString::from("dlx")),
        );
        let props = BasicProperties::default().with_headers(headers);

        let meta = extract_metadata_from_parts(&props, "", "");
        let obj = meta.as_object().unwrap();

        assert!(
            !obj.contains_key("headers"),
            "Headers section should be omitted when all headers are internal"
        );
    }

    #[test]
    fn metadata_skips_non_string_header_values() {
        let mut headers = FieldTable::default();
        headers.insert("numeric-header".into(), AMQPValue::LongLongInt(42));
        headers.insert("bool-header".into(), AMQPValue::Boolean(true));
        headers.insert(
            "string-header".into(),
            AMQPValue::LongString(LongString::from("kept")),
        );
        let props = BasicProperties::default().with_headers(headers);

        let meta = extract_metadata_from_parts(&props, "", "");
        let obj = meta.as_object().unwrap();
        let user_headers = obj.get("headers").unwrap().as_object().unwrap();

        // Non-string values are skipped by amqp_value_as_string
        assert!(!user_headers.contains_key("numeric-header"));
        assert!(!user_headers.contains_key("bool-header"));
        assert_eq!(user_headers.get("string-header").unwrap(), "kept");
    }

    #[test]
    fn metadata_combined_properties_and_delivery_info() {
        let props = BasicProperties::default()
            .with_app_id("worker-svc".into())
            .with_content_type("application/json".into());

        let meta = extract_metadata_from_parts(&props, "emails.send", "notifications");
        let obj = meta.as_object().unwrap();

        assert_eq!(obj.get("app_id").unwrap(), "worker-svc");
        assert_eq!(obj.get("content_type").unwrap(), "application/json");
        assert_eq!(obj.get("routing_key").unwrap(), "emails.send");
        assert_eq!(obj.get("exchange").unwrap(), "notifications");
        // No headers key since none were set
        assert!(!obj.contains_key("headers"));
    }

    #[test]
    fn config_default_includes_metadata() {
        let config = LapinAdapterConfig::default();
        assert!(config.include_default_metadata);
        assert!(config.metadata_extractor.is_none());
    }

    #[test]
    fn config_debug_includes_metadata_fields() {
        let config = LapinAdapterConfig::default();
        let debug = format!("{config:?}");
        assert!(debug.contains("metadata_extractor"));
        assert!(debug.contains("include_default_metadata"));
    }

    #[test]
    fn config_default_has_no_correlation_id_extractor() {
        let config = LapinAdapterConfig::default();
        assert!(config.correlation_id_extractor.is_none());
        assert!(config.max_attempts.is_none());
    }

    #[test]
    fn config_debug_includes_new_fields() {
        let config = LapinAdapterConfig::default();
        let debug = format!("{config:?}");
        assert!(debug.contains("correlation_id_extractor"));
        assert!(debug.contains("max_attempts"));
    }

    // --- extract_string_field_from_payload tests ---

    fn parse(json: &str) -> serde_json::Value {
        serde_json::from_str(json).unwrap()
    }

    #[test]
    fn payload_extracts_flat_job_id() {
        let v = parse(r#"{"job_id": "abc-123", "name": "test"}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            Some("abc-123".to_owned())
        );
    }

    #[test]
    fn payload_extracts_camel_case_job_id() {
        let v = parse(r#"{"jobId": "def-456"}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            Some("def-456".to_owned())
        );
    }

    #[test]
    fn payload_does_not_extract_generic_id() {
        let v = parse(r#"{"id": "ghi-789", "name": "test"}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            None
        );
    }

    #[test]
    fn payload_job_id_takes_priority_over_camel_case() {
        let v = parse(r#"{"job_id": "snake", "jobId": "camel"}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            Some("snake".to_owned())
        );
    }

    #[test]
    fn payload_extracts_from_externally_tagged_enum() {
        let v = parse(r#"{"CreateEvent": {"job_id": "019-uuid", "correlation_id": "trace-1"}}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            Some("019-uuid".to_owned())
        );
    }

    #[test]
    fn payload_tagged_enum_correlation_id() {
        let v = parse(r#"{"CreateEvent": {"job_id": "019-uuid", "correlation_id": "trace-1"}}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["correlation_id", "correlationId"]),
            Some("trace-1".to_owned())
        );
    }

    #[test]
    fn payload_returns_none_for_missing_field() {
        let v = parse(r#"{"name": "test", "count": 42}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            None
        );
    }

    #[test]
    fn payload_returns_none_when_no_parsed_value() {
        assert_eq!(extract_string_field_from_payload(None, &["job_id"]), None);
    }

    #[test]
    fn payload_skips_empty_string_values() {
        let v = parse(r#"{"job_id": "", "jobId": "fallback"}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            Some("fallback".to_owned())
        );
    }

    #[test]
    fn payload_extracts_numeric_job_id() {
        let v = parse(r#"{"job_id": 12345}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            Some("12345".to_owned())
        );
    }

    #[test]
    fn payload_extracts_numeric_zero() {
        let v = parse(r#"{"job_id": 0}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id"]),
            Some("0".to_owned())
        );
    }

    #[test]
    fn payload_extracts_numeric_from_tagged_enum() {
        let v = parse(r#"{"SendEmail": {"job_id": 42}}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id"]),
            Some("42".to_owned())
        );
    }

    #[test]
    fn payload_string_preferred_over_numeric_by_field_order() {
        // When job_id is numeric but jobId is a string, job_id (first in field_names) wins
        let v = parse(r#"{"job_id": 42, "jobId": "string-id"}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            Some("42".to_owned())
        );
    }

    #[test]
    fn payload_skips_non_string_non_numeric_values() {
        let v = parse(r#"{"job_id": true, "jobId": "string-id"}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id", "jobId"]),
            Some("string-id".to_owned())
        );
    }

    #[test]
    fn payload_does_not_recurse_into_multi_key_objects() {
        let v = parse(r#"{"key1": {"job_id": "inner"}, "key2": "val"}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["job_id"]),
            None
        );
    }

    #[test]
    fn payload_handles_null_correlation_id() {
        let v = parse(r#"{"CreateEvent": {"job_id": "019", "correlation_id": null}}"#);
        assert_eq!(
            extract_string_field_from_payload(Some(&v), &["correlation_id", "correlationId"]),
            None
        );
    }
}