rocketmq-protocol 1.0.0

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

use std::fmt;

use cheetah_string::CheetahString;
use rocketmq_error::Sensitive;
use rocketmq_macros::RequestHeaderCodec;
use serde::Deserialize;
use serde::Serialize;

/// Common header for every privileged maintenance request.
///
/// Authentication fields remain owned by the existing signed Remoting command.
/// These fields bind the authenticated operation to a policy revision, absolute
/// deadline, idempotency key, and Kubernetes Lease fencing token.
#[derive(Clone, Deserialize, Eq, PartialEq, RequestHeaderCodec, Serialize)]
#[header(type_id = "rocketmq_protocol::protocol::header::maintenance_request_header::MaintenanceRequestHeader")]
#[serde(rename_all = "camelCase")]
pub struct MaintenanceRequestHeader {
    /// Idempotency key generated by the release coordinator.
    #[header(required)]
    pub operation_id: CheetahString,
    /// Version of the maintenance policy used by the caller.
    #[header(required)]
    pub policy_version: u64,
    /// Absolute Unix deadline in milliseconds.
    #[header(required)]
    pub deadline_unix_millis: u64,
    /// Non-zero token derived from the current Kubernetes Lease revision.
    #[header(required)]
    pub fencing_token: u64,
}

impl fmt::Debug for MaintenanceRequestHeader {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("MaintenanceRequestHeader")
            .field("operation_id", &Sensitive::new(&self.operation_id))
            .field("policy_version", &self.policy_version)
            .field("deadline_unix_millis", &self.deadline_unix_millis)
            .field("fencing_token", &Sensitive::new(self.fencing_token))
            .finish()
    }
}

impl MaintenanceRequestHeader {
    /// Validates request-local invariants before authorization.
    ///
    /// # Errors
    ///
    /// Returns a static reason when an idempotency key, policy version,
    /// deadline, or fencing token is missing or invalid.
    pub fn validate(&self) -> Result<(), &'static str> {
        if self.operation_id.is_empty() {
            return Err("operation_id cannot be empty");
        }
        if self.policy_version == 0 {
            return Err("policy_version must be greater than zero");
        }
        if self.deadline_unix_millis == 0 {
            return Err("deadline_unix_millis must be greater than zero");
        }
        if self.fencing_token == 0 {
            return Err("fencing_token must be greater than zero");
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use super::MaintenanceRequestHeader;
    use crate::protocol::command_custom_header::CommandCustomHeader;
    use crate::protocol::command_custom_header::FromMap;
    use cheetah_string::CheetahString;

    #[test]
    fn privileged_maintenance_header_round_trips_all_fencing_fields() {
        let header = MaintenanceRequestHeader {
            operation_id: "rollback-42-controller-snapshot".into(),
            policy_version: 7,
            deadline_unix_millis: 1_800_000_000_000,
            fencing_token: 42,
        };

        let map = header.to_map().expect("encode maintenance header");
        let decoded = <MaintenanceRequestHeader as FromMap>::from(&map).expect("decode maintenance header");

        assert_eq!(decoded, header);
        assert_eq!(
            map.get(&CheetahString::from_static_str("fencingToken")),
            Some(&CheetahString::from_static_str("42"))
        );
        assert!(decoded.validate().is_ok());
    }

    #[test]
    fn privileged_maintenance_header_requires_every_field() {
        let missing = HashMap::new();

        assert!(<MaintenanceRequestHeader as FromMap>::from(&missing).is_err());
    }

    #[test]
    fn rust_extension_preserves_the_full_unsigned_domain() {
        let header = MaintenanceRequestHeader {
            operation_id: "full-u64-domain".into(),
            policy_version: u64::MAX,
            deadline_unix_millis: u64::MAX,
            fencing_token: u64::MAX,
        };

        let map = header.to_map().expect("encode full unsigned values");
        let decoded = <MaintenanceRequestHeader as FromMap>::from(&map).expect("decode full unsigned values");

        assert_eq!(decoded, header);
    }

    #[test]
    fn privileged_maintenance_header_debug_redacts_sensitive_fields() {
        let header = MaintenanceRequestHeader {
            operation_id: "release-operation-secret".into(),
            policy_version: 7,
            deadline_unix_millis: 1_800_000_000_000,
            fencing_token: 987_654_321,
        };

        let debug = format!("{header:?}");

        assert!(debug.contains("<redacted>"));
        assert!(!debug.contains("release-operation-secret"));
        assert!(!debug.contains("987654321"));
    }
}