crabka-broker 0.3.6

//! `CreateTopics` (`api_key=19`). Routes through `Controller::submit_change`
//! so every topic/partition creation goes through the metadata quorum before
//! the partition directories are materialized on disk.

use std::time::Duration;

use bytes::{Bytes, BytesMut};
use crabka_metadata::{
    AclOperation, MetadataRecord, PartitionRecord, TopicConfigRecord, TopicRecord,
};
use crabka_protocol::owned::create_topics_request::CreateTopicsRequest;
use crabka_protocol::owned::create_topics_response::{CreatableTopicResult, CreateTopicsResponse};
use crabka_protocol::primitives::uuid::Uuid as ProtoUuid;
use crabka_protocol::{Decode, Encode};
use crabka_raft::RaftError;
use uuid::Uuid;

use crate::authorizer::{AuthorizationRequest, AuthorizationResult};
use crate::broker::Broker;
use crate::codes;
use crate::error::BrokerError;
use crate::replicator_supervisor::materialize_partition;

/// Round-robin replica placement.
///
/// Given a sorted broker set `bs = [b0, b1, …, bk-1]` and a partition
/// count `P`, returns a `Vec<Vec<NodeId>>` of length `P`, where each
/// inner vec is `R = replication_factor` long. Partition `p`'s leader
/// is `bs[(p) % k]`; the remaining replicas are `bs[(p + i) % k]` for
/// `i in 1..R`. Caller must guarantee `R <= k` (else returns an empty
/// outer vec and the caller surfaces `INVALID_REPLICATION_FACTOR`).
pub(crate) fn round_robin_replicas(
    sorted_brokers: &[crabka_raft::NodeId],
    num_partitions: i32,
    replication_factor: i16,
) -> Vec<Vec<crabka_raft::NodeId>> {
    let k = sorted_brokers.len();
    let r = usize::try_from(replication_factor).unwrap_or(0);
    if r == 0 || r > k {
        return Vec::new();
    }
    let p_count = usize::try_from(num_partitions).unwrap_or(0);
    (0..p_count)
        .map(|p| {
            (0..r)
                .map(|i| sorted_brokers[(p + i) % k])
                .collect::<Vec<_>>()
        })
        .collect()
}

#[allow(clippy::too_many_lines)]
pub(crate) async fn handle(
    broker: &Broker,
    version: i16,
    _correlation_id: i32,
    req_bytes: &[u8],
    ctx: &crate::handlers::RequestContext<'_>,
) -> Result<Bytes, BrokerError> {
    // ── ACL preamble ────────────────────────────────────────
    // Whole-request Cluster Create gate. On Deny, return
    // CLUSTER_AUTHORIZATION_FAILED on every topic row and short-circuit.
    {
        let image = broker.controller.current_image();
        let allow = broker.config.authorizer.authorize(
            &*image,
            &AuthorizationRequest {
                principal: ctx.principal,
                host: ctx.peer,
                resource_type: crabka_metadata::ResourceType::Cluster,
                resource_name: "kafka-cluster",
                operation: AclOperation::Create,
            },
        );
        if allow == AuthorizationResult::Deny {
            // Peek at the topic names without consuming the request fully.
            let mut cur: &[u8] = req_bytes;
            let req = CreateTopicsRequest::decode(&mut cur, version)?;
            let results: Vec<CreatableTopicResult> = req
                .topics
                .into_iter()
                .map(|t| CreatableTopicResult {
                    name: t.name,
                    topic_id: ProtoUuid([0u8; 16]),
                    error_code: codes::CLUSTER_AUTHORIZATION_FAILED,
                    error_message: Some("create-topics denied".into()),
                    ..Default::default()
                })
                .collect();
            let resp = CreateTopicsResponse {
                topics: results,
                throttle_time_ms: 0,
                ..Default::default()
            };
            let mut buf = BytesMut::with_capacity(resp.encoded_len(version));
            resp.encode(&mut buf, version)?;
            return Ok(buf.freeze());
        }
    }

    let req_bytes = req_bytes.to_vec();
    let controller = broker.controller.clone();
    let node_id = broker.config.node_id;
    let log_dirs = broker.config.all_log_dirs();
    let log_config = broker.config.log_config.clone();
    let log_dir_status = broker.log_dir_status.clone();
    let partitions_map = broker.partitions.clone();

    {
        let mut cur: &[u8] = &req_bytes;
        let req = CreateTopicsRequest::decode(&mut cur, version)?;

        // KIP-599: count mutations before running handler logic so that even
        // invalid requests consume quota (bad-faith clients can't escape by
        // sending malformed RPCs). num_partitions == -1 means "use cluster
        // default"; count it as 1 for accounting.
        #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)]
        let mutation_count: u64 = req
            .topics
            .iter()
            .map(|t| t.num_partitions.max(1) as u64)
            .sum();

        // Hoist the image once; the per-topic loop reuses it for broker-set
        // lookup instead of calling current_image() on every iteration.
        let image = controller.current_image();

        let mut results: Vec<CreatableTopicResult> = Vec::with_capacity(req.topics.len());

        for topic_req in req.topics {
            let name = topic_req.name.clone();
            let partition_count = topic_req.num_partitions;
            let replication_factor = topic_req.replication_factor;

            // Reject invalid partition counts before attempting placement.
            if partition_count <= 0 {
                results.push(CreatableTopicResult {
                    name,
                    topic_id: ProtoUuid([0u8; 16]),
                    error_code: codes::INVALID_PARTITIONS,
                    error_message: None,
                    ..Default::default()
                });
                continue;
            }

            // Read the current broker set from the controller's image; sort by
            // node_id for determinism.
            //
            // Race-tolerance: on a freshly-started cluster, the self-registration
            // V1BrokerRegistration record may not have made it into the local
            // MetadataImage yet when this handler runs (the controller's apply is
            // mostly synchronous on the leader but observable timing has slipped
            // on slow runners). If `brokers()` is empty, fall back to "this broker
            // is the only known broker" so the single-broker case (which is by
            // far the most common) doesn't silently degrade to
            // INVALID_REPLICATION_FACTOR.
            let mut sorted_brokers: Vec<crabka_raft::NodeId> =
                image.brokers().map(|b| b.node_id).collect();
            if sorted_brokers.is_empty() {
                sorted_brokers.push(node_id);
            }
            sorted_brokers.sort_unstable();

            let assignments =
                round_robin_replicas(&sorted_brokers, partition_count, replication_factor);

            if assignments.is_empty() {
                // RF > broker count. Surface INVALID_REPLICATION_FACTOR per Apache
                // Kafka semantics.
                results.push(CreatableTopicResult {
                    name,
                    topic_id: ProtoUuid([0u8; 16]),
                    error_code: codes::INVALID_REPLICATION_FACTOR,
                    error_message: None,
                    ..Default::default()
                });
                continue;
            }

            let topic_id = Uuid::new_v4();

            // Build the batch: one TopicRecord + N PartitionRecords.
            let mut records = vec![MetadataRecord::V1Topic(TopicRecord {
                name: name.clone(),
                topic_id,
                partitions: partition_count,
                replication_factor,
            })];
            for (p, replicas) in assignments.iter().enumerate() {
                let p_i32 = i32::try_from(p).unwrap_or(0);
                records.push(MetadataRecord::V1Partition(PartitionRecord {
                    topic: name.clone(),
                    partition: p_i32,
                    leader: replicas[0],
                    replicas: replicas.clone(),
                    isr: replicas.clone(),
                    leader_epoch: 0,
                    adding_replicas: vec![],
                    removing_replicas: vec![],
                    directories: vec![],
                    partition_epoch: 0,
                }));
            }

            // Persist any topic-level configs the client sent. Without
            // this, cleanup.policy / segment.bytes / retention.ms etc.
            // set at CreateTopics time would be silently dropped — clients
            // would need a follow-up AlterConfigs round-trip. Match Kafka's
            // CreateTopics semantics by emitting one V1TopicConfig record
            // covering the full override map.
            if !topic_req.configs.is_empty() {
                let mut overrides: std::collections::BTreeMap<String, String> =
                    std::collections::BTreeMap::new();
                for cfg in &topic_req.configs {
                    if let Some(value) = &cfg.value {
                        overrides.insert(cfg.name.clone(), value.clone());
                    }
                }
                if !overrides.is_empty() {
                    records.push(MetadataRecord::V1TopicConfig(TopicConfigRecord {
                        topic: name.clone(),
                        overrides,
                    }));
                }
            }

            let result = controller.submit_change(records).await;

            let error_code = match result {
                Ok(()) => {
                    // Committed to quorum — materialize on-disk partitions for
                    // every assignment where THIS broker is in `replicas`,
                    // whether as leader or follower. The replicator supervisor
                    // materializes the same partitions on the OTHER brokers
                    // lazily via the metadata-watch; this handler-side path is
                    // an optimization so producers that immediately follow the
                    // CreateTopics ack don't race the supervisor.
                    for (p, replicas) in assignments.iter().enumerate() {
                        let p_i32 = i32::try_from(p).unwrap_or(0);
                        if !replicas.contains(&node_id) {
                            continue;
                        }
                        // Use the same `materialize_partition` helper the
                        // supervisor uses — its Entry::Vacant gate ensures
                        // we don't spawn a second writer task when the
                        // supervisor has already reconciled (the metadata
                        // watch can fire mid-handler).
                        if let Err(e) = materialize_partition(
                            &partitions_map,
                            &name,
                            p_i32,
                            &log_dirs,
                            &log_config,
                            &log_dir_status,
                        ) {
                            tracing::error!(
                                topic = %name, partition = p_i32, error = %e,
                                "CreateTopics: materialize after quorum commit failed"
                            );
                            // Quorum already committed — partition will be
                            // recovered on next broker restart.
                            continue;
                        }
                        // Mirror what `ReplicatorSupervisor::reconcile` does
                        // for newly-materialized leader partitions: sync the
                        // cached leader + epoch, and (when self is leader)
                        // install the ISR for HW computation. Without this, a
                        // Produce arriving before the supervisor's
                        // metadata-watch fires sees `isr.is_empty()`, falls
                        // into `compute_hw == leader_leo`, and acks=-1 returns
                        // instantly without waiting for followers.
                        if let Some(part) = partitions_map.get(&name, p_i32) {
                            let leader = replicas[0];
                            part.install_leader_change(leader, 0).await;
                            if leader == node_id {
                                // At creation the ISR equals the full replica set.
                                part.install_isr(replicas, replicas, leader).await;
                            }
                        }
                    }
                    codes::NONE
                }
                Err(RaftError::Metadata(crabka_metadata::MetadataError::TopicExists(_))) => {
                    codes::TOPIC_ALREADY_EXISTS
                }
                Err(RaftError::Metadata(crabka_metadata::MetadataError::InvalidRecord(_))) => {
                    // E.g., `partitions <= 0` rejected by image::validate.
                    codes::INVALID_PARTITIONS
                }
                Err(RaftError::NotLeader { .. } | RaftError::LeaderUnknown) => {
                    codes::NOT_CONTROLLER
                }
                Err(e) => {
                    tracing::error!(topic = %name, error = %e, "CreateTopics submit_change failed");
                    codes::UNKNOWN_SERVER_ERROR
                }
            };

            // Convert uuid::Uuid → crabka_protocol::primitives::uuid::Uuid.
            let proto_uuid = ProtoUuid(topic_id.into_bytes());

            let mut result = CreatableTopicResult {
                name,
                topic_id: proto_uuid,
                error_code,
                error_message: None,
                ..Default::default()
            };

            if error_code == codes::NONE {
                result.num_partitions = partition_count;
                result.replication_factor = replication_factor;
                // KIP-525 (v5+): return an empty configs list to satisfy
                // clients that unconditionally call `configs().stream()`.
                result.configs = Some(Vec::new());
            }
            results.push(result);
        }

        // KIP-599: consume controller_mutation_rate quota.
        let delay = crate::quota::consume_controller_mutation_quota(
            &image,
            &broker.quota_buckets,
            &ctx.principal.name,
            ctx.client_id,
            mutation_count,
        );
        let resp = CreateTopicsResponse {
            topics: results,
            throttle_time_ms: i32::try_from(delay.as_millis()).unwrap_or(i32::MAX),
            ..Default::default()
        };
        if delay > Duration::ZERO {
            tokio::time::sleep(delay).await;
        }
        let mut buf = BytesMut::with_capacity(resp.encoded_len(version));
        resp.encode(&mut buf, version)?;
        Ok(buf.freeze())
    }
}

#[cfg(test)]
mod replica_assignment_tests {
    use super::round_robin_replicas;
    use assert2::assert;

    #[test]
    fn three_brokers_three_partitions_rf_three() {
        let bs = vec![1u64, 2, 3];
        let out = round_robin_replicas(&bs, 3, 3);
        // Every broker should lead exactly one partition.
        let leaders: Vec<_> = out.iter().map(|r| r[0]).collect();
        let mut sorted = leaders.clone();
        sorted.sort_unstable();
        assert!(sorted == vec![1, 2, 3]);
        // Each partition has all three brokers as replicas.
        for replicas in &out {
            let mut s = replicas.clone();
            s.sort_unstable();
            assert!(s == vec![1, 2, 3]);
        }
    }

    #[test]
    fn offset_per_partition_means_distinct_leaders() {
        let bs = vec![1u64, 2, 3];
        let out = round_robin_replicas(&bs, 3, 1);
        assert!(out[0] == vec![1]);
        assert!(out[1] == vec![2]);
        assert!(out[2] == vec![3]);
    }

    #[test]
    fn rf_too_high_returns_empty() {
        let bs = vec![1u64, 2, 3];
        let out = round_robin_replicas(&bs, 1, 5);
        assert!(out.is_empty());
    }

    #[test]
    fn rf_one_single_broker_preserves_replica_shape() {
        let bs = vec![1u64];
        let out = round_robin_replicas(&bs, 2, 1);
        assert!(out == vec![vec![1u64], vec![1u64]]);
    }

    #[test]
    fn consume_controller_mutation_quota_tuple_match_overage_throttles() {
        use crabka_metadata::{ClientQuotaRecord, MetadataImage, MetadataRecord, QuotaEntity};
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1ClientQuota(ClientQuotaRecord {
            entity: vec![
                QuotaEntity {
                    entity_type: "user".into(),
                    entity_name: Some("alice".into()),
                },
                QuotaEntity {
                    entity_type: "client-id".into(),
                    entity_name: Some("app-x".into()),
                },
            ],
            config_key: "controller_mutation_rate".into(),
            config_value: Some(1.0),
        }));
        let buckets = crate::quota::QuotaBuckets::new();
        let delay_match =
            crate::quota::consume_controller_mutation_quota(&img, &buckets, "alice", "app-x", 10);
        assert!(
            delay_match > std::time::Duration::ZERO,
            "tuple quota match should throttle on overage; got {delay_match:?}"
        );
        let buckets2 = crate::quota::QuotaBuckets::new();
        let delay_other =
            crate::quota::consume_controller_mutation_quota(&img, &buckets2, "alice", "other", 10);
        assert!(
            delay_other == std::time::Duration::ZERO,
            "non-matching client_id should not throttle; got {delay_other:?}"
        );
    }
}