use crate::types::{Header, Headers, TopicName, TopicOffset};

/// Prefix reserved for server-set system headers.
/// Clients cannot publish headers with keys starting with this prefix.
pub const RESERVED_HEADER_PREFIX: &[u8] = b"_fafb.";

/// Reason a message was routed to the dead-letter queue.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DlqReason {
    /// Automatic DLQ after exceeding `max_redelivery_attempts`.
    MaxAttemptsExceeded,
    /// Client explicitly rejected the message via Nack frame.
    ClientNack { reason: String },
    /// Message TTL expired while inflight (delivered but not acked in time).
    MessageExpired,
}

impl DlqReason {
    /// Returns the reason label string for metrics.
    pub fn reason_label(&self) -> &'static str {
        match self {
            DlqReason::MaxAttemptsExceeded => "max_attempts",
            DlqReason::ClientNack { .. } => "client_nack",
            DlqReason::MessageExpired => "message_expired",
        }
    }
}

// ---------------------------------------------------------------------------
// DLQ system header key constants (`_fafb.dlq.*`)
// ---------------------------------------------------------------------------

pub const DLQ_HEADER_SOURCE_TOPIC: &[u8] = b"_fafb.dlq.source_topic";
pub const DLQ_HEADER_ORIGINAL_OFFSET: &[u8] = b"_fafb.dlq.original_offset";
pub const DLQ_HEADER_DELIVERY_COUNT: &[u8] = b"_fafb.dlq.delivery_count";
pub const DLQ_HEADER_REASON: &[u8] = b"_fafb.dlq.reason";
pub const DLQ_HEADER_FIRST_DELIVERY_TS: &[u8] = b"_fafb.dlq.first_delivery_ts";
pub const DLQ_HEADER_TIMESTAMP: &[u8] = b"_fafb.dlq.timestamp";
pub const DLQ_HEADER_NACK_REASON: &[u8] = b"_fafb.dlq.nack_reason";

/// Build `_fafb.dlq.*` system headers for a DLQ message.
///
/// Merges the DLQ provenance headers with any existing application headers
/// from the original message. The original headers come first, followed by
/// system headers.
pub fn build_dlq_headers(
    source_topic: &TopicName,
    original_offset: TopicOffset,
    delivery_count: u32,
    reason: &DlqReason,
    first_delivered_at: u64,
    dlq_timestamp: u64,
    original_headers: &Headers,
) -> Headers {
    let mut headers = original_headers.clone();

    headers.push(Header {
        key: DLQ_HEADER_SOURCE_TOPIC.to_vec(),
        value: source_topic.0.as_bytes().to_vec(),
    });
    headers.push(Header {
        key: DLQ_HEADER_ORIGINAL_OFFSET.to_vec(),
        value: original_offset.0.to_string().into_bytes(),
    });
    headers.push(Header {
        key: DLQ_HEADER_DELIVERY_COUNT.to_vec(),
        value: delivery_count.to_string().into_bytes(),
    });
    headers.push(Header {
        key: DLQ_HEADER_REASON.to_vec(),
        value: reason.reason_label().as_bytes().to_vec(),
    });
    headers.push(Header {
        key: DLQ_HEADER_FIRST_DELIVERY_TS.to_vec(),
        value: first_delivered_at.to_string().into_bytes(),
    });
    headers.push(Header {
        key: DLQ_HEADER_TIMESTAMP.to_vec(),
        value: dlq_timestamp.to_string().into_bytes(),
    });

    if let DlqReason::ClientNack { reason } = reason {
        headers.push(Header {
            key: DLQ_HEADER_NACK_REASON.to_vec(),
            value: reason.as_bytes().to_vec(),
        });
    }

    headers
}

/// Parsed DLQ metadata extracted from `_fafb.dlq.*` system headers.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DlqMetadata {
    pub source_topic: String,
    pub original_offset: u64,
    pub delivery_count: u32,
    pub reason: String,
    pub first_delivery_ts: u64,
    pub dlq_timestamp: u64,
    pub nack_reason: Option<String>,
}

/// Parse `_fafb.dlq.*` system headers into [`DlqMetadata`].
///
/// Returns `None` if the required DLQ headers are missing (message is not a
/// DLQ message). This is a convenience for DLQ consumers — the server sets
/// headers but does not read them back.
pub fn parse_dlq_headers(headers: &Headers) -> Option<DlqMetadata> {
    let find = |key: &[u8]| -> Option<String> {
        headers
            .iter()
            .find(|h| h.key == key)
            .and_then(|h| String::from_utf8(h.value.clone()).ok())
    };

    let source_topic = find(DLQ_HEADER_SOURCE_TOPIC)?;
    let original_offset = find(DLQ_HEADER_ORIGINAL_OFFSET)?.parse::<u64>().ok()?;
    let delivery_count = find(DLQ_HEADER_DELIVERY_COUNT)?.parse::<u32>().ok()?;
    let reason = find(DLQ_HEADER_REASON)?;
    let first_delivery_ts = find(DLQ_HEADER_FIRST_DELIVERY_TS)?.parse::<u64>().ok()?;
    let dlq_timestamp = find(DLQ_HEADER_TIMESTAMP)?.parse::<u64>().ok()?;
    let nack_reason = find(DLQ_HEADER_NACK_REASON);

    Some(DlqMetadata {
        source_topic,
        original_offset,
        delivery_count,
        reason,
        first_delivery_ts,
        dlq_timestamp,
        nack_reason,
    })
}

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

    #[test]
    fn build_parse_round_trip_max_attempts() {
        let topic = TopicName("orders".into());
        let dlq_ts = 1_700_000_060_000u64;

        let headers = build_dlq_headers(
            &topic,
            TopicOffset(42),
            6,
            &DlqReason::MaxAttemptsExceeded,
            1_700_000_000_000,
            dlq_ts,
            &vec![],
        );
        let meta = parse_dlq_headers(&headers).expect("should parse DLQ headers");

        assert_eq!(meta.source_topic, "orders");
        assert_eq!(meta.original_offset, 42);
        assert_eq!(meta.delivery_count, 6);
        assert_eq!(meta.reason, "max_attempts");
        assert_eq!(meta.first_delivery_ts, 1_700_000_000_000);
        assert_eq!(meta.dlq_timestamp, dlq_ts);
        assert!(meta.nack_reason.is_none());
    }

    #[test]
    fn build_parse_round_trip_client_nack() {
        let topic = TopicName("events.user".into());
        let dlq_ts = 1_700_000_005_000u64;
        let reason = DlqReason::ClientNack {
            reason: "invalid JSON".into(),
        };

        let headers = build_dlq_headers(
            &topic,
            TopicOffset(99),
            1,
            &reason,
            1_700_000_000_000,
            dlq_ts,
            &vec![],
        );
        let meta = parse_dlq_headers(&headers).expect("should parse DLQ headers");

        assert_eq!(meta.source_topic, "events.user");
        assert_eq!(meta.original_offset, 99);
        assert_eq!(meta.delivery_count, 1);
        assert_eq!(meta.reason, "client_nack");
        assert_eq!(meta.first_delivery_ts, 1_700_000_000_000);
        assert_eq!(meta.dlq_timestamp, dlq_ts);
        assert_eq!(meta.nack_reason.as_deref(), Some("invalid JSON"));
    }

    #[test]
    fn build_parse_round_trip_empty_nack_reason() {
        let reason = DlqReason::ClientNack {
            reason: String::new(),
        };
        let headers = build_dlq_headers(
            &TopicName("orders".into()),
            TopicOffset(0),
            1,
            &reason,
            0,
            100,
            &vec![],
        );
        let meta = parse_dlq_headers(&headers).unwrap();
        assert_eq!(meta.nack_reason.as_deref(), Some(""));
    }

    #[test]
    fn parse_returns_none_for_non_dlq_message() {
        let headers = vec![Header {
            key: b"content-type".to_vec(),
            value: b"application/json".to_vec(),
        }];
        assert!(parse_dlq_headers(&headers).is_none());
    }

    #[test]
    fn parse_returns_none_for_empty_headers() {
        assert!(parse_dlq_headers(&vec![]).is_none());
    }

    #[test]
    fn original_headers_preserved() {
        let original = vec![
            Header {
                key: b"content-type".to_vec(),
                value: b"application/json".to_vec(),
            },
            Header {
                key: b"correlation-id".to_vec(),
                value: b"req-123".to_vec(),
            },
        ];
        let headers = build_dlq_headers(
            &TopicName("orders".into()),
            TopicOffset(42),
            6,
            &DlqReason::MaxAttemptsExceeded,
            1_700_000_000_000,
            100,
            &original,
        );

        // Original headers come first
        assert_eq!(headers[0].key, b"content-type");
        assert_eq!(headers[1].key, b"correlation-id");
        // DLQ headers follow
        assert_eq!(headers[2].key, DLQ_HEADER_SOURCE_TOPIC);

        // Parse still works
        let meta = parse_dlq_headers(&headers).unwrap();
        assert_eq!(meta.source_topic, "orders");
    }

    #[test]
    fn reason_label_values() {
        assert_eq!(
            DlqReason::MaxAttemptsExceeded.reason_label(),
            "max_attempts"
        );
        assert_eq!(
            DlqReason::ClientNack {
                reason: "bad".into()
            }
            .reason_label(),
            "client_nack"
        );
    }

    #[test]
    fn reserved_prefix_constant_matches_dlq_keys() {
        // All DLQ header keys must start with the reserved prefix
        for key in [
            DLQ_HEADER_SOURCE_TOPIC,
            DLQ_HEADER_ORIGINAL_OFFSET,
            DLQ_HEADER_DELIVERY_COUNT,
            DLQ_HEADER_REASON,
            DLQ_HEADER_FIRST_DELIVERY_TS,
            DLQ_HEADER_TIMESTAMP,
            DLQ_HEADER_NACK_REASON,
        ] {
            assert!(
                key.starts_with(RESERVED_HEADER_PREFIX),
                "DLQ header key {:?} should start with reserved prefix",
                std::str::from_utf8(key).unwrap()
            );
        }
    }
}
