bindy/reconcilers/

dnszone.rs

// Copyright (c) 2025 Erick Bourgeois, firestoned
#![allow(clippy::uninlined_format_args)]
#![allow(clippy::doc_markdown)]
// SPDX-License-Identifier: MIT

//! DNS zone reconciliation logic.
//!
//! This module handles the creation and management of DNS zones on BIND9 servers.
//! It supports both primary and secondary zone configurations.

// Module imports
pub mod bind9_config;
pub mod cleanup;
pub mod constants;
pub mod discovery;
pub mod helpers;
pub mod primary;
pub mod secondary;
pub mod status_helpers;
pub mod types;
pub mod validation;

#[cfg(test)]
#[path = "dnszone/helpers_tests.rs"]
mod helpers_tests;

// Bind9Instance and InstanceReferenceWithStatus are used by dead_code marked functions (Phase 2 cleanup)
use self::types::DuplicateZoneInfo;
#[allow(unused_imports)]
use crate::crd::{Condition, DNSZone, DNSZoneStatus};
use anyhow::{anyhow, Result};
use bindcar::{ZONE_TYPE_PRIMARY, ZONE_TYPE_SECONDARY};
use futures::stream::{self, StreamExt};
use k8s_openapi::api::core::v1::{Pod, Service};
use kube::{api::ListParams, client::Client, Api, ResourceExt};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::{debug, error, info, warn};

/// Creates a map of nameserver hostnames to IP addresses by:
/// 1. Checking for Service external IPs first (`LoadBalancer` or `NodePort`)
/// 2. Falling back to pod IPs if no external IPs are available
///
/// Nameservers are named: `ns1.{zone_name}.`, `ns2.{zone_name}.`, etc.
/// Order: Primary instances first, then secondary instances.
///
/// # Arguments
///
/// * `client` - Kubernetes API client
/// * `zone_name` - DNS zone name (e.g., "example.com")
/// * `instance_refs` - All instance references (primaries and secondaries)
///
/// # Returns
///
/// `HashMap` of nameserver hostnames to IP addresses, or None if no IPs found
///
/// # Errors
///
/// Returns an error if Kubernetes API calls fail.
pub async fn generate_nameserver_ips(
    client: &Client,
    zone_name: &str,
    instance_refs: &[crate::crd::InstanceReference],
) -> Result<Option<HashMap<String, String>>> {
    if instance_refs.is_empty() {
        return Ok(None);
    }

    let mut nameserver_ips = HashMap::new();
    let mut ns_index = 1;

    // Process primaries first, then secondaries
    for instance_ref in instance_refs {
        // Try to get Service external IP first
        let service_api: Api<Service> = Api::namespaced(client.clone(), &instance_ref.namespace);

        let ip = match service_api.get(&instance_ref.name).await {
            Ok(service) => {
                // Check for LoadBalancer external IP
                if let Some(status) = &service.status {
                    if let Some(load_balancer) = &status.load_balancer {
                        if let Some(ingress_list) = &load_balancer.ingress {
                            if let Some(ingress) = ingress_list.first() {
                                if let Some(lb_ip) = &ingress.ip {
                                    debug!(
                                        "Using LoadBalancer IP {} for instance {}/{}",
                                        lb_ip, instance_ref.namespace, instance_ref.name
                                    );
                                    Some(lb_ip.clone())
                                } else {
                                    None
                                }
                            } else {
                                None
                            }
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                } else {
                    None
                }
            }
            Err(e) => {
                debug!(
                    "Failed to get service for instance {}/{}: {}. Will try pod IP.",
                    instance_ref.namespace, instance_ref.name, e
                );
                None
            }
        };

        // If no service external IP, fallback to pod IP
        let ip = if ip.is_none() {
            // Get pod IP
            let pod_api: Api<Pod> = Api::namespaced(client.clone(), &instance_ref.namespace);
            let label_selector = format!("app=bind9,instance={}", instance_ref.name);
            let lp = ListParams::default().labels(&label_selector);

            match pod_api.list(&lp).await {
                Ok(pods) => {
                    // Find first running pod
                    pods.items
                        .iter()
                        .find(|pod| {
                            let phase = pod
                                .status
                                .as_ref()
                                .and_then(|s| s.phase.as_ref())
                                .map_or("Unknown", std::string::String::as_str);
                            phase == "Running"
                        })
                        .and_then(|pod| {
                            pod.status
                                .as_ref()
                                .and_then(|s| s.pod_ip.as_ref())
                                .map(|ip| {
                                    debug!(
                                        "Using pod IP {} for instance {}/{}",
                                        ip, instance_ref.namespace, instance_ref.name
                                    );
                                    ip.clone()
                                })
                        })
                }
                Err(e) => {
                    warn!(
                        "Failed to list pods for instance {}/{}: {}. Skipping.",
                        instance_ref.namespace, instance_ref.name, e
                    );
                    None
                }
            }
        } else {
            ip
        };

        // Add to nameserver map if we found an IP
        if let Some(ip) = ip {
            let ns_hostname = format!("ns{ns_index}.{zone_name}.");
            nameserver_ips.insert(ns_hostname, ip);
            ns_index += 1;
        }
    }

    if nameserver_ips.is_empty() {
        Ok(None)
    } else {
        Ok(Some(nameserver_ips))
    }
}

/// Get the effective nameservers list for a DNSZone, handling both new and deprecated fields.
///
/// This function provides backward compatibility by:
/// 1. Preferring the new `name_servers` field if present
/// 2. Falling back to the deprecated `name_server_ips` field with automatic migration
/// 3. Logging deprecation warnings when the old field is used
///
/// # Arguments
/// * `spec` - The DNSZone spec containing nameserver configuration
///
/// # Returns
/// `Option<Vec<NameServer>>` - The effective list of nameservers, or `None` if neither field is set
///
/// # Examples
///
/// ```
/// # #[allow(deprecated)]
/// # use bindy::crd::{DNSZoneSpec, NameServer, SOARecord};
/// # use std::collections::HashMap;
/// // New field takes precedence
/// let spec = DNSZoneSpec {
///     zone_name: "example.com".into(),
///     soa_record: SOARecord {
///         primary_ns: "ns1.example.com.".into(),
///         admin_email: "admin.example.com.".into(),
///         serial: 1,
///         refresh: 3600,
///         retry: 600,
///         expire: 604800,
///         negative_ttl: 86400,
///     },
///     ttl: None,
///     cluster_ref: None,
///     name_servers: Some(vec![NameServer {
///         hostname: "ns2.example.com.".into(),
///         ipv4_address: None,
///         ipv6_address: None,
///     }]),
///     name_server_ips: Some(HashMap::from([("ns3.example.com.".into(), "192.0.2.3".into())])),
///     records_from: None,
///     bind9_instances_from: None,
///     dnssec_policy: None,
/// };
/// // Returns name_servers (new field), ignoring name_server_ips
/// ```
fn get_effective_name_servers(
    spec: &crate::crd::DNSZoneSpec,
) -> Option<Vec<crate::crd::NameServer>> {
    use crate::crd::NameServer;

    // New field takes precedence
    if let Some(ref new_servers) = spec.name_servers {
        debug!(
            "Using new `nameServers` field with {} server(s)",
            new_servers.len()
        );
        return Some(new_servers.clone());
    }

    // Fallback to deprecated field with migration warning
    #[allow(deprecated)]
    if let Some(ref old_ips) = spec.name_server_ips {
        warn!(
            "DNSZone uses deprecated `nameServerIps` field. \
             Migrate to `nameServers` for better functionality and IPv6 support. \
             See migration guide at docs/src/operations/migration-guide.md"
        );

        // Convert HashMap<String, String> to Vec<NameServer>
        // Old format: {"ns2.example.com.": "192.0.2.2"}
        // New format: vec![NameServer { hostname: "ns2.example.com.", ipv4_address: Some("192.0.2.2"), .. }]
        let servers: Vec<NameServer> = old_ips
            .iter()
            .map(|(hostname, ip)| NameServer {
                hostname: hostname.clone(),
                ipv4_address: Some(ip.clone()),
                ipv6_address: None, // Old field doesn't support IPv6
            })
            .collect();

        debug!(
            "Migrated {} server(s) from deprecated `nameServerIps` to new format",
            servers.len()
        );

        return Some(servers);
    }

    // Neither field set
    None
}

/// Re-fetch a DNSZone to get the latest status.
///
/// The `dnszone` parameter from the watch event might have stale status from the cache.
/// We need the latest `status.bind9Instances` which may have been updated by the
/// Bind9Instance reconciler.
///
/// # Arguments
/// * `client` - Kubernetes client
/// * `namespace` - Namespace of the DNSZone
/// * `name` - Name of the DNSZone
///
/// # Returns
/// The freshly fetched DNSZone with current status
///
/// # Errors
/// Returns an error if the Kubernetes API call fails
async fn refetch_zone(client: &kube::Client, namespace: &str, name: &str) -> Result<DNSZone> {
    let zones_api: Api<DNSZone> = Api::namespaced(client.clone(), namespace);
    let zone = zones_api.get(name).await?;
    Ok(zone)
}

/// Handle duplicate zone conflicts by setting Ready=False and stopping reconciliation.
///
/// When a duplicate zone is detected, this function:
/// 1. Logs a warning with details about the conflict
/// 2. Updates the status with Ready=False and DuplicateZone condition
/// 3. Applies the status to the API server
///
/// # Arguments
/// * `client` - Kubernetes client
/// * `namespace` - Namespace of the conflicting DNSZone
/// * `name` - Name of the conflicting DNSZone
/// * `duplicate_info` - Information about the duplicate zone conflict
/// * `status_updater` - Status updater to apply the condition
///
/// # Errors
/// Returns an error if the status update fails
async fn handle_duplicate_zone(
    client: &kube::Client,
    namespace: &str,
    name: &str,
    duplicate_info: &DuplicateZoneInfo,
    status_updater: &mut crate::reconcilers::status::DNSZoneStatusUpdater,
) -> Result<()> {
    warn!(
        "Duplicate zone detected: {}/{} cannot claim '{}' because it is already configured by: {:?}",
        namespace, name, duplicate_info.zone_name, duplicate_info.conflicting_zones
    );

    // Build list of conflicting zones in namespace/name format
    let conflicting_zone_refs: Vec<String> = duplicate_info
        .conflicting_zones
        .iter()
        .map(|z| format!("{}/{}", z.namespace, z.name))
        .collect();

    // Set Ready=False with DuplicateZone reason
    status_updater.set_duplicate_zone_condition(&duplicate_info.zone_name, &conflicting_zone_refs);

    // Apply status and stop processing
    status_updater.apply(client).await?;

    Ok(())
}

/// Detect if the zone spec has changed since last reconciliation.
///
/// Compares current generation with observed generation to determine
/// if this is first reconciliation or if spec changed.
///
/// # Arguments
///
/// * `zone` - The DNSZone resource
///
/// # Returns
///
/// Tuple of (first_reconciliation, spec_changed)
fn detect_spec_changes(zone: &DNSZone) -> (bool, bool) {
    let current_generation = zone.metadata.generation;
    let observed_generation = zone.status.as_ref().and_then(|s| s.observed_generation);

    let first_reconciliation = observed_generation.is_none();
    let spec_changed =
        crate::reconcilers::should_reconcile(current_generation, observed_generation);

    (first_reconciliation, spec_changed)
}

/// Detect if the instance list changed between watch event and re-fetch.
///
/// This is critical for detecting when:
/// 1. New instances are added to `status.bind9Instances` (via `bind9InstancesFrom` selectors)
/// 2. Instance `lastReconciledAt` timestamps are cleared (e.g., instance deleted, needs reconfiguration)
///
/// NOTE: `InstanceReference` `PartialEq` ignores `lastReconciledAt`, so we must check timestamps separately!
///
/// # Arguments
///
/// * `namespace` - Namespace for logging
/// * `name` - Zone name for logging
/// * `watch_instances` - Instances from the watch event that triggered reconciliation
/// * `current_instances` - Instances after re-fetching (current state)
///
/// # Returns
///
/// `true` if instances changed (list or timestamps), `false` otherwise
fn detect_instance_changes(
    namespace: &str,
    name: &str,
    watch_instances: Option<&Vec<crate::crd::InstanceReference>>,
    current_instances: &[crate::crd::InstanceReference],
) -> bool {
    let Some(watch_instances) = watch_instances else {
        // No instances in watch event, first reconciliation or error
        return true;
    };

    // Get the instance names from the watch event (what triggered us)
    let watch_instance_names: std::collections::HashSet<_> =
        watch_instances.iter().map(|r| &r.name).collect();

    // Get the instance names after re-fetching (current state)
    let current_instance_names: std::collections::HashSet<_> =
        current_instances.iter().map(|r| &r.name).collect();

    // Check if instance list changed (added/removed instances)
    let list_changed = watch_instance_names != current_instance_names;

    if list_changed {
        info!(
            "Instance list changed during reconciliation for zone {}/{}: watch_event={:?}, current={:?}",
            namespace, name, watch_instance_names, current_instance_names
        );
        return true;
    }

    // List is the same, but check if any lastReconciledAt timestamps changed
    // Use InstanceReference as HashMap key (uses its Hash impl which hashes identity fields)
    let watch_timestamps: std::collections::HashMap<&crate::crd::InstanceReference, Option<&str>> =
        watch_instances
            .iter()
            .map(|inst| (inst, inst.last_reconciled_at.as_deref()))
            .collect();

    let current_timestamps: std::collections::HashMap<
        &crate::crd::InstanceReference,
        Option<&str>,
    > = current_instances
        .iter()
        .map(|inst| (inst, inst.last_reconciled_at.as_deref()))
        .collect();

    let timestamps_changed = watch_timestamps.iter().any(|(inst_ref, watch_ts)| {
        current_timestamps
            .get(inst_ref)
            .is_some_and(|current_ts| current_ts != watch_ts)
    });

    if timestamps_changed {
        info!(
            "Instance lastReconciledAt timestamps changed for zone {}/{}",
            namespace, name
        );
    }

    timestamps_changed
}

/// Reconciles a `DNSZone` resource.
///
/// Creates or updates DNS zone files on BIND9 instances that match the zone's
/// instance selector. Supports both primary and secondary zone types.
///
/// # Zone Types
///
/// - **Primary**: Authoritative zone with SOA record and local zone file
/// - **Secondary**: Replica zone that transfers from primary servers
///
/// # Arguments
///
/// * `client` - Kubernetes API client for finding matching `Bind9Instances`
/// * `dnszone` - The `DNSZone` resource to reconcile
/// * `zone_manager` - BIND9 manager for creating zone files
///
/// # Returns
///
/// * `Ok(())` - If zone was created/updated successfully
/// * `Err(_)` - If zone creation failed or configuration is invalid
///
/// # Example
///
/// ```rust,no_run,ignore
/// use bindy::reconcilers::reconcile_dnszone;
/// use bindy::crd::DNSZone;
/// use bindy::bind9::Bind9Manager;
/// use bindy::context::Context;
/// use std::sync::Arc;
///
/// async fn handle_zone(ctx: Arc<Context>, zone: DNSZone) -> anyhow::Result<()> {
///     let manager = Bind9Manager::new();
///     reconcile_dnszone(ctx, zone, &manager).await?;
///     Ok(())
/// }
/// ```
///
/// # Errors
///
/// Returns an error if Kubernetes API operations fail or BIND9 zone operations fail.
#[allow(clippy::too_many_lines)]
pub async fn reconcile_dnszone(
    ctx: Arc<crate::context::Context>,
    dnszone: DNSZone,
    zone_manager: &crate::bind9::Bind9Manager,
) -> Result<()> {
    let client = ctx.client.clone();
    let bind9_instances_store = &ctx.stores.bind9_instances;

    let namespace = dnszone.namespace().unwrap_or_default();
    let name = dnszone.name_any();

    info!("Reconciling DNSZone: {}/{}", namespace, name);
    debug!(
        namespace = %namespace,
        name = %name,
        generation = ?dnszone.metadata.generation,
        "Starting DNSZone reconciliation"
    );

    // Save the instance list from the watch event (before re-fetching)
    // This represents the instances that triggered this reconciliation
    let watch_event_instances =
        validation::get_instances_from_zone(&dnszone, bind9_instances_store).ok();

    // CRITICAL: Re-fetch the zone to get the latest status
    let dnszone = refetch_zone(&client, &namespace, &name).await?;

    // Create centralized status updater to batch all status changes
    let mut status_updater = crate::reconcilers::status::DNSZoneStatusUpdater::new(&dnszone);

    // Extract spec
    let spec = &dnszone.spec;

    // Validate that zone has instances assigned (via spec.bind9Instances or status.bind9Instances)
    // This will fail early if zone is not selected by any instance
    let instance_refs = validation::get_instances_from_zone(&dnszone, bind9_instances_store)?;

    info!(
        "DNSZone {}/{} is assigned to {} instance(s): {:?}",
        namespace,
        name,
        instance_refs.len(),
        instance_refs.iter().map(|r| &r.name).collect::<Vec<_>>()
    );

    // CRITICAL: Check for duplicate zones BEFORE any configuration
    // If another zone already claims this zone name, set Ready=False with DuplicateZone reason
    // and stop processing to prevent conflicting DNS configurations
    let zones_store = &ctx.stores.dnszones;
    if let Some(duplicate_info) = validation::check_for_duplicate_zones(&dnszone, zones_store) {
        handle_duplicate_zone(
            &client,
            &namespace,
            &name,
            &duplicate_info,
            &mut status_updater,
        )
        .await?;
        return Ok(());
    }

    // Determine if this is the first reconciliation or if spec has changed
    let (first_reconciliation, spec_changed) = detect_spec_changes(&dnszone);

    // Check if the instance list or lastReconciledAt timestamps changed between watch event and re-fetch
    let instances_changed = detect_instance_changes(
        &namespace,
        &name,
        watch_event_instances.as_ref(),
        &instance_refs,
    );

    // Check if any instances need reconciliation (never reconciled or reconciliation failed)
    let unreconciled_instances =
        validation::filter_instances_needing_reconciliation(&instance_refs);
    let has_unreconciled_instances = !unreconciled_instances.is_empty();

    if has_unreconciled_instances {
        info!(
            "Found {} unreconciled instance(s) for zone {}/{}: {:?}",
            unreconciled_instances.len(),
            namespace,
            name,
            unreconciled_instances
                .iter()
                .map(|i| format!("{}/{}", i.namespace, i.name))
                .collect::<Vec<_>>()
        );
    } else {
        debug!(
            "No unreconciled instances for zone {}/{} - all {} instance(s) already configured (lastReconciledAt set)",
            namespace,
            name,
            instance_refs.len()
        );
    }

    // CRITICAL: Cleanup deleted instances BEFORE early return check
    // If we skip reconciliation due to no changes, we still need to remove deleted instances from status
    match cleanup::cleanup_deleted_instances(&client, &dnszone, &mut status_updater).await {
        Ok(deleted_count) if deleted_count > 0 => {
            info!(
                "Cleaned up {} deleted instance(s) from zone {}/{} status",
                deleted_count, namespace, name
            );
        }
        Ok(_) => {
            debug!(
                "No deleted instances found for zone {}/{} status",
                namespace, name
            );
        }
        Err(e) => {
            warn!(
                "Failed to cleanup deleted instances for zone {}/{}: {} (continuing with reconciliation)",
                namespace, name, e
            );
            // Don't fail reconciliation for cleanup errors
        }
    }

    // CRITICAL: We CANNOT skip reconciliation entirely, even if spec and instances haven't changed.
    // Reconciliation may be triggered by ARecord/AAAA/TXT/etc changes via watches, and we MUST
    // run record discovery to tag newly created records with status.zoneRef.
    //
    // However, we CAN skip BIND9 configuration if nothing changed (handled later in the flow).
    // This ensures record discovery ALWAYS runs while still optimizing BIND9 API calls.

    if instances_changed {
        info!(
            "Instances changed for zone {}/{} - reconciling to configure new instances",
            namespace, name
        );
    }

    info!(
        "Reconciling zone {} (first_reconciliation={}, spec_changed={})",
        spec.zone_name, first_reconciliation, spec_changed
    );

    // Cleanup stale records from status.records[] before main reconciliation
    // This ensures status stays in sync with actual Kubernetes resources
    match cleanup::cleanup_stale_records(
        &client,
        &dnszone,
        &mut status_updater,
        bind9_instances_store,
    )
    .await
    {
        Ok(stale_count) if stale_count > 0 => {
            info!(
                "Cleaned up {} stale record(s) from zone {}/{} status",
                stale_count, namespace, name
            );
        }
        Ok(_) => {
            debug!(
                "No stale records found in zone {}/{} status",
                namespace, name
            );
        }
        Err(e) => {
            warn!(
                "Failed to cleanup stale records for zone {}/{}: {} (continuing with reconciliation)",
                namespace, name, e
            );
            // Don't fail reconciliation for cleanup errors
        }
    }

    // BIND9 configuration: Always ensure zones exist on all instances
    // This implements true declarative reconciliation - if a pod restarts without
    // persistent storage, the reconciler will detect the missing zone and recreate it.
    // The add_zones() function is idempotent, so this is safe to call every reconciliation.
    //
    // NOTE: We ALWAYS configure zones, not just when spec changes. This ensures:
    // - Zones are recreated if pods restart without persistent volumes
    // - New instances added to the cluster get zones automatically
    // - Drift detection: if someone manually deletes a zone, it's recreated
    // - True Kubernetes declarative reconciliation: actual state continuously matches desired state
    let (primary_outcome, secondary_outcome) = bind9_config::configure_zone_on_instances(
        ctx.clone(),
        &dnszone,
        zone_manager,
        &mut status_updater,
        &instance_refs,
        &unreconciled_instances,
    )
    .await?;

    // Discover DNS records and update status
    let (record_refs, records_count) =
        discovery::discover_and_update_records(&client, &dnszone, &mut status_updater, &ctx.stores)
            .await?;

    // Replay the zone's records whenever the zone had to be (re)created on any
    // server, or a previous replay did not finish. Without this a wiped pod
    // comes back authoritative for a zone containing only SOA and NS - see
    // `records::replay_zone_records` for the full rationale.
    replay_records_if_zone_was_recreated(
        &ctx,
        &dnszone,
        &mut status_updater,
        &instance_refs,
        &record_refs,
        primary_outcome.zones_created + secondary_outcome.zones_created,
    )
    .await?;

    // Check if all discovered records are ready and trigger zone transfers if needed
    if records_count > 0 {
        let all_records_ready =
            discovery::check_all_records_ready(&client, &namespace, &record_refs).await?;

        if all_records_ready {
            info!(
                "All {} record(s) for zone {} are ready, triggering zone transfers to secondaries",
                records_count, spec.zone_name
            );

            // Trigger zone transfers to all secondaries
            // Zone transfers are triggered automatically by BIND9 via NOTIFY messages
            // No manual trigger needed in the new architecture
            info!(
                "Zone {} configured on instances - BIND9 will handle zone transfers via NOTIFY",
                spec.zone_name
            );
        } else {
            info!("Not all records for zone {} are ready yet", spec.zone_name);
        }
    }
    // Calculate expected counts and finalize status
    let (expected_primary_count, expected_secondary_count) =
        status_helpers::calculate_expected_instance_counts(&client, &instance_refs).await?;

    status_helpers::finalize_zone_status(
        &mut status_updater,
        &client,
        &spec.zone_name,
        &namespace,
        &name,
        primary_outcome,
        secondary_outcome,
        expected_primary_count,
        expected_secondary_count,
        records_count,
        dnszone.metadata.generation,
    )
    .await?;

    Ok(())
}

/// Replay all of a zone's records into BIND9 when the zone was (re)created.
///
/// # Why
///
/// BIND9 operand pods keep zone data in ephemeral storage. Any event that
/// replaces a pod - an operator upgrade, a `placement` change that rolls the
/// Deployment, an eviction, a node reboot, a manual `kubectl delete pod` -
/// brings the pod back with no zones. The zone reconciler then recreates the
/// zone from `spec`, which yields SOA and NS records ONLY. The pod is now
/// *authoritative* for a zone with no data: it answers authoritative NXDOMAIN,
/// or, when `global.recursion` and `global.forwarders` are set, silently
/// forwards the query and returns the PUBLIC answer for an internal name.
///
/// Nothing about the record CRs changed, so their own controllers have no
/// reason to act. This function is the missing link: the moment the zone
/// reconciler observes that it created a zone, it pushes every record CR the
/// zone selects back into BIND9.
///
/// The intent is recorded in `status.recordsResyncPending` before the replay is
/// attempted, so an operator crash mid-replay, or a partial failure, is retried
/// on the next reconciliation instead of being forgotten. While the flag is set
/// the zone reports `Ready=False` / `Degraded=True`, so a server authoritative
/// for an empty zone is never advertised as healthy.
///
/// # Arguments
///
/// * `ctx` - Application context (Kubernetes client and reflector stores)
/// * `dnszone` - The zone being reconciled
/// * `status_updater` - Status updater collecting in-memory condition changes
/// * `instance_refs` - All instances assigned to the zone
/// * `record_refs` - The records just discovered for the zone
/// * `zones_created` - Number of endpoints where the zone was newly created
///
/// # Errors
///
/// Returns an error if the PRIMARY instances cannot be determined. Individual
/// record push failures are reported through the `Degraded` condition and
/// retried on the next reconciliation rather than aborting the zone reconcile.
async fn replay_records_if_zone_was_recreated(
    ctx: &Arc<crate::context::Context>,
    dnszone: &DNSZone,
    status_updater: &mut crate::reconcilers::status::DNSZoneStatusUpdater,
    instance_refs: &[crate::crd::InstanceReference],
    record_refs: &[crate::crd::RecordReferenceWithTimestamp],
    zones_created: usize,
) -> Result<()> {
    let client = ctx.client.clone();
    let namespace = dnszone.namespace().unwrap_or_default();
    let name = dnszone.name_any();
    let zone_name = &dnszone.spec.zone_name;

    // A replay left over from a previous reconciliation is just as binding as
    // one triggered right now.
    let resync_outstanding = dnszone
        .status
        .as_ref()
        .is_some_and(|status| status.records_resync_pending);

    if zones_created == 0 && !resync_outstanding {
        return Ok(());
    }

    if zones_created > 0 {
        warn!(
            "Zone {} was created on {} endpoint(s) during this reconciliation of DNSZone {}/{} - \
             those servers hold SOA and NS records only. Replaying {} record(s).",
            zone_name,
            zones_created,
            namespace,
            name,
            record_refs.len()
        );
    } else {
        info!(
            "DNSZone {}/{} still has an outstanding record resync - retrying {} record(s) for zone {}",
            namespace,
            name,
            record_refs.len(),
            zone_name
        );
    }

    // A zone with no records is fully described by its SOA and NS records, so
    // recreating it already restored the declared state - nothing to replay and
    // nothing to keep the zone out of Ready.
    if record_refs.is_empty() {
        debug!(
            "Zone {} selects no records - nothing to replay for DNSZone {}/{}",
            zone_name, namespace, name
        );
        status_updater.set_records_resync_pending(false);
        return Ok(());
    }

    // Persist the intent BEFORE touching BIND9: if the operator dies mid-replay
    // the flag survives and the next reconciliation retries.
    status_updater.set_records_resync_pending(true);
    status_updater.apply(&client).await?;

    // Records are written to PRIMARY servers only; secondaries pull the zone
    // via AXFR once the primary's serial advances.
    let primary_refs = primary::filter_primary_instances(&client, instance_refs).await?;

    if primary_refs.is_empty() {
        let message = format!(
            "Zone {zone_name} must replay {} record(s) but has no primary instances to write them to",
            record_refs.len()
        );
        warn!("DNSZone {}/{}: {}", namespace, name, message);
        status_updater.set_condition("Degraded", "True", "RecordsResyncPending", &message);
        return Ok(());
    }

    let outcome = crate::reconcilers::records::replay_zone_records(
        &client,
        &ctx.stores,
        zone_name,
        record_refs,
        &primary_refs,
    )
    .await;

    if outcome.is_complete() {
        info!(
            "Record resync complete for DNSZone {}/{}: {}",
            namespace,
            name,
            outcome.summary(zone_name)
        );
        status_updater.set_records_resync_pending(false);
        return Ok(());
    }

    let message = outcome.summary(zone_name);
    warn!(
        "Record resync incomplete for DNSZone {}/{}: {} - zone stays Degraded and will be retried",
        namespace, name, message
    );
    status_updater.set_condition("Degraded", "True", "RecordsResyncPending", &message);

    Ok(())
}

/// Adds a DNS zone to all primary instances.
///
/// # Arguments
///
/// * `client` - Kubernetes API client
/// * `dnszone` - The `DNSZone` resource
/// * `zone_manager` - BIND9 manager for adding zone
///
/// # Returns
///
/// * `Ok(ZoneConfigOutcome)` - Per-instance and per-endpoint configuration counts.
///   An instance counts as configured only if ALL of its ready endpoints accepted
///   the zone, so the instance count is directly comparable with the expected
///   PRIMARY instance count when computing readiness.
/// * `Err(_)` - If zone addition failed on every endpoint
///
/// # Errors
///
/// Returns an error if BIND9 zone addition fails or if no instances are assigned.
///
/// # Panics
///
/// Panics if the RNDC key is not loaded by the helper function (should never happen in practice).
#[allow(clippy::too_many_lines)]
pub async fn add_dnszone(
    ctx: Arc<crate::context::Context>,
    dnszone: DNSZone,
    zone_manager: &crate::bind9::Bind9Manager,
    status_updater: &mut crate::reconcilers::status::DNSZoneStatusUpdater,
    instance_refs: &[crate::crd::InstanceReference],
) -> Result<types::ZoneConfigOutcome> {
    let client = ctx.client.clone();
    let namespace = dnszone.namespace().unwrap_or_default();
    let name = dnszone.name_any();
    let spec = &dnszone.spec;

    info!("Adding DNSZone {}/{}", namespace, name);

    // PHASE 2 OPTIMIZATION: Use the filtered instance list passed by the caller
    // This ensures we only process instances that need reconciliation (lastReconciledAt == None)

    info!(
        "DNSZone {}/{} will be added to {} instance(s): {:?}",
        namespace,
        name,
        instance_refs.len(),
        instance_refs
            .iter()
            .map(|i| format!("{}/{}", i.namespace, i.name))
            .collect::<Vec<_>>()
    );

    // Filter to only PRIMARY instances
    let primary_instance_refs = primary::filter_primary_instances(&client, instance_refs).await?;

    if primary_instance_refs.is_empty() {
        return Err(anyhow!(
            "DNSZone {}/{} has no PRIMARY instances assigned. Instances: {:?}",
            namespace,
            name,
            instance_refs
                .iter()
                .map(|i| format!("{}/{}", i.namespace, i.name))
                .collect::<Vec<_>>()
        ));
    }

    info!(
        "Found {} PRIMARY instance(s) for DNSZone {}/{}",
        primary_instance_refs.len(),
        namespace,
        name
    );

    // Find all secondary instances for zone transfer configuration
    let secondary_instance_refs =
        secondary::filter_secondary_instances(&client, instance_refs).await?;
    let secondary_ips =
        secondary::find_secondary_pod_ips_from_instances(&client, &secondary_instance_refs).await?;

    if secondary_ips.is_empty() {
        warn!(
            "No secondary servers found for DNSZone {}/{} - zone transfers will not be configured",
            namespace, name
        );
    } else {
        info!(
            "Found {} secondary server(s) for DNSZone {}/{} - zone transfers will be configured: {:?}",
            secondary_ips.len(),
            namespace,
            name,
            secondary_ips
        );
    }

    // Get effective nameservers (supports both new `nameServers` and deprecated `nameServerIps`)
    let effective_name_servers = get_effective_name_servers(spec);

    // Generate legacy nameserver IPs format for backward compatibility with bindcar API
    // If user didn't provide either field, auto-generate from instance IPs
    let name_server_ips = if effective_name_servers.is_none() {
        info!(
            "DNSZone {}/{} has no explicit nameServers - auto-generating from {} instance(s)",
            namespace,
            name,
            instance_refs.len()
        );

        // Build ordered list: primaries first, then secondaries
        let mut ordered_instances = primary_instance_refs.clone();
        ordered_instances.extend(secondary_instance_refs.clone());

        match generate_nameserver_ips(&client, &spec.zone_name, &ordered_instances).await {
            Ok(Some(generated_ips)) => {
                info!(
                    "Auto-generated {} nameserver(s) for DNSZone {}/{}: {:?}",
                    generated_ips.len(),
                    namespace,
                    name,
                    generated_ips
                );
                Some(generated_ips)
            }
            Ok(None) => {
                warn!(
                    "Failed to auto-generate nameserver IPs for DNSZone {}/{} - no IPs available",
                    namespace, name
                );
                None
            }
            Err(e) => {
                warn!(
                    "Error auto-generating nameserver IPs for DNSZone {}/{}: {}",
                    namespace, name, e
                );
                None
            }
        }
    } else {
        // Convert effective_name_servers to HashMap<String, String> for bindcar API compatibility
        // Only include IPv4 addresses (bindcar doesn't support IPv6 glue records in this field)
        // SAFETY: We know effective_name_servers is Some because we're in the else block
        let name_server_map: HashMap<String, String> =
            if let Some(ref ns_list) = effective_name_servers {
                ns_list
                    .iter()
                    .filter_map(|ns| {
                        ns.ipv4_address
                            .as_ref()
                            .map(|ip| (ns.hostname.clone(), ip.clone()))
                    })
                    .collect()
            } else {
                HashMap::new()
            };

        info!(
            "Using explicit nameServers for DNSZone {}/{} ({} with IPv4 glue records)",
            namespace,
            name,
            name_server_map.len()
        );

        if name_server_map.is_empty() {
            None
        } else {
            Some(name_server_map)
        }
    };

    // Extract list of ALL nameserver hostnames (primary from SOA + all from nameServers field)
    // This is used by bindcar to generate NS records in the zone file
    let all_nameserver_hostnames: Vec<String> = {
        let mut hostnames = vec![spec.soa_record.primary_ns.clone()];

        if let Some(ref ns_list) = effective_name_servers {
            for ns in ns_list {
                // Avoid duplicates - don't add primary NS again if it's in the list
                if ns.hostname != spec.soa_record.primary_ns {
                    hostnames.push(ns.hostname.clone());
                }
            }
        }

        hostnames
    };

    info!(
        "Zone {}/{} will be configured with {} nameserver(s): {:?}",
        namespace,
        name,
        all_nameserver_hostnames.len(),
        all_nameserver_hostnames
    );

    // Extract DNSSEC policy if configured
    let dnssec_policy = spec.dnssec_policy.as_deref();
    if let Some(policy) = dnssec_policy {
        info!(
            "DNSSEC policy '{}' will be applied to zone {}/{}",
            policy, namespace, name
        );
    }

    // Process all primary instances concurrently using async streams
    // Mark each instance as reconciled immediately after first successful endpoint configuration
    let first_endpoint = Arc::new(Mutex::new(None::<String>));
    let total_endpoints = Arc::new(Mutex::new(0_usize));
    // Endpoints where the zone did NOT exist and had to be created. A created
    // zone holds only SOA + NS, so every one of these endpoints is missing all
    // of the zone's record data and needs a replay.
    let zones_created = Arc::new(Mutex::new(0_usize));
    let errors = Arc::new(Mutex::new(Vec::<String>::new()));
    let status_updater_shared = Arc::new(Mutex::new(status_updater));

    // Create a stream of futures for all instances.
    // Each instance future resolves to `true` only if EVERY endpoint of that
    // instance accepted the zone (added or already present) - this is the
    // per-INSTANCE success signal used for readiness computation.
    let instance_results = stream::iter(primary_instance_refs.iter())
        .then(|instance_ref| {
            let client = client.clone();
            let zone_manager = zone_manager.clone();
            let zone_name = spec.zone_name.clone();
            let soa_record = spec.soa_record.clone();
            let all_nameserver_hostnames = all_nameserver_hostnames.clone();
            let name_server_ips = name_server_ips.clone();
            let secondary_ips = secondary_ips.clone();
            let first_endpoint = Arc::clone(&first_endpoint);
            let total_endpoints = Arc::clone(&total_endpoints);
            let zones_created = Arc::clone(&zones_created);
            let errors = Arc::clone(&errors);
            let status_updater_shared = Arc::clone(&status_updater_shared);
            let instance_ref = instance_ref.clone();
            let _zone_namespace = namespace.clone();
            let _zone_name_ref = name.clone();

            async move {
                info!(
                    "Processing endpoints for primary instance {}/{}",
                    instance_ref.namespace, instance_ref.name
                );

                // Load RNDC key for this specific instance
                let key_data = match helpers::load_rndc_key(&client, &instance_ref.namespace, &instance_ref.name).await {
                    Ok(key) => key,
                    Err(e) => {
                        let err_msg = format!("instance {}/{}: failed to load RNDC key: {e}", instance_ref.namespace, instance_ref.name);
                        errors.lock().await.push(err_msg);
                        return false;
                    }
                };

                // Get all endpoints for this instance
                let endpoints = match helpers::get_endpoint(&client, &instance_ref.namespace, &instance_ref.name, "http").await {
                    Ok(eps) => eps,
                    Err(e) => {
                        let err_msg = format!("instance {}/{}: failed to get endpoints: {e}", instance_ref.namespace, instance_ref.name);
                        errors.lock().await.push(err_msg);
                        return false;
                    }
                };

                info!(
                    "Found {} endpoint(s) for primary instance {}/{}",
                    endpoints.len(),
                    instance_ref.namespace,
                    instance_ref.name
                );

                // Process endpoints concurrently for this instance
                let endpoint_results = stream::iter(endpoints.iter())
                    .then(|endpoint| {
                        let zone_manager = zone_manager.clone();
                        let zone_name = zone_name.clone();
                        let key_data = key_data.clone();
                        let soa_record = soa_record.clone();
                        let all_nameserver_hostnames = all_nameserver_hostnames.clone();
                        let name_server_ips = name_server_ips.clone();
                        let secondary_ips = secondary_ips.clone();
                        let first_endpoint = Arc::clone(&first_endpoint);
                        let total_endpoints = Arc::clone(&total_endpoints);
                        let zones_created = Arc::clone(&zones_created);
                        let errors = Arc::clone(&errors);
                        let instance_ref = instance_ref.clone();
                        let endpoint = endpoint.clone();

                        async move {
                            let pod_endpoint = format!("{}:{}", endpoint.ip, endpoint.port);

                            // Save the first endpoint (globally)
                            {
                                let mut first = first_endpoint.lock().await;
                                if first.is_none() {
                                    *first = Some(pod_endpoint.clone());
                                }
                            }

                            // Check if zone already exists before attempting creation
                            let zone_exists = match zone_manager.zone_exists(&zone_name, &pod_endpoint).await {
                                Ok(exists) => exists,
                                Err(e) => {
                                    error!(
                                        "Failed to check if zone {} exists on endpoint {} (instance {}/{}): {}",
                                        zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name, e
                                    );
                                    // Treat errors as "zone might not exist" - proceed with add_zones
                                    false
                                }
                            };

                            if zone_exists {
                                debug!(
                                    "Zone {} already exists on endpoint {} (instance {}/{}), skipping creation",
                                    zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name
                                );
                                *total_endpoints.lock().await += 1;
                                // Return false to indicate zone was not newly added
                                return Ok(false);
                            }

                            // Pass secondary IPs for zone transfer configuration
                            let secondary_ips_ref = if secondary_ips.is_empty() {
                                None
                            } else {
                                Some(secondary_ips.as_slice())
                            };

                            match zone_manager
                                .add_zones(
                                    &zone_name,
                                    ZONE_TYPE_PRIMARY,
                                    &pod_endpoint,
                                    &key_data,
                                    Some(&soa_record),
                                    Some(&all_nameserver_hostnames),
                                    name_server_ips.as_ref(),
                                    secondary_ips_ref,
                                    None, // primary_ips only for secondary zones
                                    dnssec_policy,
                                )
                                .await
                            {
                                Ok(was_added) => {
                                    if was_added {
                                        // The zone was absent from this pod and has just been
                                        // recreated from spec - it currently holds only SOA and
                                        // NS records, so the pod is authoritative for an empty
                                        // zone until the records are replayed.
                                        warn!(
                                            "Zone {} was MISSING on endpoint {} (instance: {}/{}) and has been recreated \
                                             with SOA and NS records only - all records for this zone will be replayed",
                                            zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name
                                        );
                                        *zones_created.lock().await += 1;
                                    }
                                    *total_endpoints.lock().await += 1;
                                    // Return was_added so we can check if zone was actually configured
                                    Ok(was_added)
                                }
                                Err(e) => {
                                    error!(
                                        "Failed to add zone {} to endpoint {} (instance {}/{}): {}",
                                        zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name, e
                                    );
                                    errors.lock().await.push(format!(
                                        "endpoint {pod_endpoint} (instance {}/{}): {e}",
                                        instance_ref.namespace, instance_ref.name
                                    ));
                                    Err(())
                                }
                            }
                        }
                    })
                    .collect::<Vec<Result<bool, ()>>>()
                    .await;

                // Mark this instance as configured ONLY if at least one endpoint actually added the zone
                // This prevents updating lastReconciledAt when zone already exists (avoids tight loop)
                let zone_was_configured = endpoint_results.iter().any(|r| r.is_ok() && *r.as_ref().unwrap());
                if zone_was_configured {
                    status_updater_shared
                        .lock()
                        .await
                        .update_instance_status(
                            &instance_ref.name,
                            &instance_ref.namespace,
                            crate::crd::InstanceStatus::Configured,
                            Some("Zone successfully configured on primary instance".to_string()),
                        );
                    info!(
                        "Marked primary instance {}/{} as configured for zone {}",
                        instance_ref.namespace, instance_ref.name, zone_name
                    );
                }

                // The instance counts as fully configured only if every one of
                // its ready endpoints accepted the zone (added OR already
                // present). A single failed endpoint means the instance is NOT
                // fully serving the zone and must not count towards readiness.
                !endpoint_results.is_empty() && endpoint_results.iter().all(Result::is_ok)
            }
        })
        .collect::<Vec<bool>>()
        .await;

    let instances_configured = instance_results.iter().filter(|ok| **ok).count();

    let first_endpoint = Arc::try_unwrap(first_endpoint)
        .expect("Failed to unwrap first_endpoint Arc")
        .into_inner();
    let total_endpoints = Arc::try_unwrap(total_endpoints)
        .expect("Failed to unwrap total_endpoints Arc")
        .into_inner();
    let zones_created = Arc::try_unwrap(zones_created)
        .expect("Failed to unwrap zones_created Arc")
        .into_inner();
    let errors = Arc::try_unwrap(errors)
        .expect("Failed to unwrap errors Arc")
        .into_inner();
    let _status_updater = Arc::try_unwrap(status_updater_shared)
        .map_err(|_| anyhow!("Failed to unwrap status_updater - multiple references remain"))?
        .into_inner();

    // If ALL operations failed, return an error
    if total_endpoints == 0 && !errors.is_empty() {
        return Err(anyhow!(
            "Failed to add zone {} to all primary instances. Errors: {}",
            spec.zone_name,
            errors.join("; ")
        ));
    }

    info!(
        "Successfully added zone {} to {} endpoint(s) across {}/{} fully configured primary instance(s)",
        spec.zone_name,
        total_endpoints,
        instances_configured,
        primary_instance_refs.len()
    );

    // Auto-generate NS records and glue records from nameServers field
    if let Some(ref name_servers) = effective_name_servers {
        if !name_servers.is_empty() {
            info!(
                "Auto-generating NS records for {} nameserver(s) in zone {}",
                name_servers.len(),
                spec.zone_name
            );

            if let Err(e) = auto_generate_ns_records(
                &client,
                name_servers,
                &spec.zone_name,
                spec.ttl,
                &primary_instance_refs,
            )
            .await
            {
                warn!(
                    "Failed to auto-generate some NS records for zone {}: {}. \
                     Zone is functional but may have incomplete NS records.",
                    spec.zone_name, e
                );
                // Don't fail reconciliation - zone is functional even without all NS records
            }
        }
    }

    // Note: We don't need to reload after addzone because:
    // 1. rndc addzone immediately adds the zone to BIND9's running config
    // 2. The zone file will be created automatically when records are added via dynamic updates
    // 3. Reloading would fail if the zone file doesn't exist yet

    // Notify secondaries about the new zone via the first endpoint
    // This triggers zone transfer (AXFR) from primary to secondaries
    if let Some(first_pod_endpoint) = first_endpoint {
        info!("Notifying secondaries about new zone {}", spec.zone_name);
        if let Err(e) = zone_manager
            .notify_zone(&spec.zone_name, &first_pod_endpoint)
            .await
        {
            // Don't fail if NOTIFY fails - the zone was successfully created
            // Secondaries will sync via SOA refresh timer
            warn!(
                "Failed to notify secondaries for zone {}: {}. Secondaries will sync via SOA refresh timer.",
                spec.zone_name, e
            );
        }
    } else {
        warn!(
            "No endpoints found for zone {}, cannot notify secondaries",
            spec.zone_name
        );
    }

    Ok(types::ZoneConfigOutcome {
        instances_configured,
        endpoints_configured: total_endpoints,
        zones_created,
    })
}

/// Adds a DNS zone to all secondary instances in the cluster with primaries configured.
///
/// Creates secondary zones on all secondary instances, configuring them to transfer
/// from the provided primary server IPs. If a zone already exists on a secondary,
/// it checks if the primaries list matches and updates it if necessary.
///
/// # Arguments
///
/// * `client` - Kubernetes API client
/// * `dnszone` - The `DNSZone` resource
/// * `zone_manager` - BIND9 manager for adding zone
/// * `primary_ips` - List of primary server IPs to configure in the primaries field
///
/// # Returns
///
/// * `Ok(ZoneConfigOutcome)` - Per-instance and per-endpoint configuration counts.
///   An instance counts as configured only if ALL of its ready endpoints accepted
///   the zone, so the instance count is directly comparable with the expected
///   SECONDARY instance count when computing readiness.
/// * `Err(_)` - If zone addition failed on every endpoint
///
/// # Errors
///
/// Returns an error if BIND9 zone addition fails on any secondary instance.
///
/// # Panics
///
/// Panics if internal Arc unwrapping fails (should not happen in normal operation).
#[allow(clippy::too_many_lines)]
pub async fn add_dnszone_to_secondaries(
    ctx: Arc<crate::context::Context>,
    dnszone: DNSZone,
    zone_manager: &crate::bind9::Bind9Manager,
    primary_ips: &[String],
    status_updater: &mut crate::reconcilers::status::DNSZoneStatusUpdater,
    instance_refs: &[crate::crd::InstanceReference],
) -> Result<types::ZoneConfigOutcome> {
    let client = ctx.client.clone();
    let namespace = dnszone.namespace().unwrap_or_default();
    let name = dnszone.name_any();
    let spec = &dnszone.spec;

    if primary_ips.is_empty() {
        warn!(
            "No primary IPs provided for secondary zone {}/{} - skipping secondary configuration",
            namespace, spec.zone_name
        );
        return Ok(types::ZoneConfigOutcome::default());
    }

    info!(
        "Adding DNSZone {}/{} to secondary instances with primaries: {:?}",
        namespace, name, primary_ips
    );

    // PHASE 2 OPTIMIZATION: Use the filtered instance list passed by the caller
    // This ensures we only process instances that need reconciliation (lastReconciledAt == None)

    // Filter to only SECONDARY instances
    let secondary_instance_refs =
        secondary::filter_secondary_instances(&client, instance_refs).await?;

    if secondary_instance_refs.is_empty() {
        info!(
            "No secondary instances found for DNSZone {}/{} - skipping secondary zone configuration",
            namespace, name
        );
        return Ok(types::ZoneConfigOutcome::default());
    }

    info!(
        "Found {} secondary instance(s) for DNSZone {}/{}",
        secondary_instance_refs.len(),
        namespace,
        name
    );

    // Process all secondary instances concurrently using async streams
    // Mark each instance as reconciled immediately after first successful endpoint configuration
    let total_endpoints = Arc::new(Mutex::new(0_usize));
    // Endpoints where the secondary zone had to be created (see the primary
    // path for why this matters).
    let zones_created = Arc::new(Mutex::new(0_usize));
    let errors = Arc::new(Mutex::new(Vec::<String>::new()));
    let status_updater_shared = Arc::new(Mutex::new(status_updater));

    // Create a stream of futures for all secondary instances.
    // Each instance future resolves to `true` only if EVERY endpoint of that
    // instance accepted the zone (added or already present) - this is the
    // per-INSTANCE success signal used for readiness computation.
    let instance_results = stream::iter(secondary_instance_refs.iter())
        .then(|instance_ref| {
            let client = client.clone();
            let zone_manager = zone_manager.clone();
            let zone_name = spec.zone_name.clone();
            let primary_ips = primary_ips.to_vec();
            let total_endpoints = Arc::clone(&total_endpoints);
            let zones_created = Arc::clone(&zones_created);
            let errors = Arc::clone(&errors);
            let status_updater_shared = Arc::clone(&status_updater_shared);
            let instance_ref = instance_ref.clone();
            let _zone_namespace = namespace.clone();
            let _zone_name_ref = name.clone();

            async move {
                info!(
                    "Processing secondary instance {}/{} for zone {}",
                    instance_ref.namespace, instance_ref.name, zone_name
                );

                // Load RNDC key for this specific instance
                // Each instance has its own RNDC secret for security isolation
                let key_data = match helpers::load_rndc_key(&client, &instance_ref.namespace, &instance_ref.name).await {
                    Ok(key) => key,
                    Err(e) => {
                        let err_msg = format!("instance {}/{}: failed to load RNDC key: {e}", instance_ref.namespace, instance_ref.name);
                        errors.lock().await.push(err_msg);
                        return false;
                    }
                };

                // Get all endpoints for this secondary instance
                let endpoints = match helpers::get_endpoint(&client, &instance_ref.namespace, &instance_ref.name, "http").await {
                    Ok(eps) => eps,
                    Err(e) => {
                        let err_msg = format!("instance {}/{}: failed to get endpoints: {e}", instance_ref.namespace, instance_ref.name);
                        errors.lock().await.push(err_msg);
                        return false;
                    }
                };

                info!(
                    "Found {} endpoint(s) for secondary instance {}/{}",
                    endpoints.len(),
                    instance_ref.namespace,
                    instance_ref.name
                );

                // Process endpoints concurrently for this instance
                let endpoint_results = stream::iter(endpoints.iter())
                    .then(|endpoint| {
                        let zone_manager = zone_manager.clone();
                        let zone_name = zone_name.clone();
                        let key_data = key_data.clone();
                        let primary_ips = primary_ips.clone();
                        let total_endpoints = Arc::clone(&total_endpoints);
                        let zones_created = Arc::clone(&zones_created);
                        let errors = Arc::clone(&errors);
                        let instance_ref = instance_ref.clone();
                        let endpoint = endpoint.clone();

                        async move {
                            let pod_endpoint = format!("{}:{}", endpoint.ip, endpoint.port);

                            // Check if zone already exists before attempting creation
                            let zone_exists = match zone_manager.zone_exists(&zone_name, &pod_endpoint).await {
                                Ok(exists) => exists,
                                Err(e) => {
                                    error!(
                                        "Failed to check if zone {} exists on endpoint {} (instance {}/{}): {}",
                                        zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name, e
                                    );
                                    // Treat errors as "zone might not exist" - proceed with add_zones
                                    false
                                }
                            };

                            // Variable to track if zone was added
                            let was_added = if zone_exists {
                                debug!(
                                    "Secondary zone {} already exists on endpoint {} (instance {}/{}), skipping creation",
                                    zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name
                                );
                                *total_endpoints.lock().await += 1;
                                false // Zone not newly added
                            } else {
                                info!(
                                    "Adding secondary zone {} to endpoint {} (instance: {}/{}) with primaries: {:?}",
                                    zone_name,
                                    pod_endpoint,
                                    instance_ref.namespace,
                                    instance_ref.name,
                                    primary_ips
                                );

                                match zone_manager
                                    .add_zones(
                                        &zone_name,
                                        ZONE_TYPE_SECONDARY,
                                        &pod_endpoint,
                                        &key_data,
                                        None, // No SOA record for secondary zones
                                        None, // No name_servers for secondary zones
                                        None, // No name_server_ips for secondary zones
                                        None, // No secondary_ips for secondary zones
                                        Some(&primary_ips),
                                        None, // No DNSSEC policy for secondary zones
                                    )
                                    .await
                                {
                                    Ok(added) => {
                                        if added {
                                            info!(
                                                "Successfully added secondary zone {} to endpoint {} (instance: {}/{})",
                                                zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name
                                            );
                                            *zones_created.lock().await += 1;
                                        } else {
                                            info!(
                                                "Secondary zone {} already exists on endpoint {} (instance: {}/{})",
                                                zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name
                                            );
                                        }
                                        *total_endpoints.lock().await += 1;
                                        added
                                    }
                                    Err(e) => {
                                        error!(
                                            "Failed to add secondary zone {} to endpoint {} (instance {}/{}): {}",
                                            zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name, e
                                        );
                                        errors.lock().await.push(format!(
                                            "endpoint {pod_endpoint} (instance {}/{}): {e}",
                                            instance_ref.namespace, instance_ref.name
                                        ));
                                        return Err(());
                                    }
                                }
                            };

                            // CRITICAL: Immediately trigger zone transfer to load the zone data
                            // This is necessary because:
                            // 1. `rndc addzone` only adds the zone to BIND9's config (in-memory)
                            // 2. The zone file doesn't exist yet on the secondary
                            // 3. Queries will return SERVFAIL until data is transferred from primary
                            // 4. `rndc retransfer` forces an immediate AXFR from primary to secondary
                            //
                            // This ensures the zone is LOADED and SERVING queries immediately after
                            // secondary pod restart or zone creation.
                            // NOTE: We trigger transfer even if zone already existed to ensure it's up to date
                            info!(
                                "Triggering immediate zone transfer for {} on secondary {} to load zone data",
                                zone_name, pod_endpoint
                            );
                            if let Err(e) = zone_manager
                                .retransfer_zone(&zone_name, &pod_endpoint)
                                .await
                            {
                                // Don't fail reconciliation if retransfer fails - zone will sync via SOA refresh
                                warn!(
                                    "Failed to trigger immediate zone transfer for {} on {}: {}. Zone will sync via SOA refresh timer.",
                                    zone_name, pod_endpoint, e
                                );
                            } else {
                                info!(
                                    "Successfully triggered zone transfer for {} on {}",
                                    zone_name, pod_endpoint
                                );
                            }

                            // Return was_added so we can check if zone was actually configured
                            Ok(was_added)
                        }
                    })
                    .collect::<Vec<Result<bool, ()>>>()
                    .await;

                // Mark this instance as configured ONLY if at least one endpoint actually added the zone
                // This prevents updating lastReconciledAt when zone already exists (avoids tight loop)
                let zone_was_configured = endpoint_results.iter().any(|r| r.is_ok() && *r.as_ref().unwrap());
                if zone_was_configured {
                    status_updater_shared
                        .lock()
                        .await
                        .update_instance_status(
                            &instance_ref.name,
                            &instance_ref.namespace,
                            crate::crd::InstanceStatus::Configured,
                            Some("Zone successfully configured on secondary instance".to_string()),
                        );
                    info!(
                        "Marked secondary instance {}/{} as configured for zone {}",
                        instance_ref.namespace, instance_ref.name, zone_name
                    );
                }

                // The instance counts as fully configured only if every one of
                // its ready endpoints accepted the zone (added OR already
                // present). A single failed endpoint means the instance is NOT
                // fully serving the zone and must not count towards readiness.
                !endpoint_results.is_empty() && endpoint_results.iter().all(Result::is_ok)
            }
        })
        .collect::<Vec<bool>>()
        .await;

    let instances_configured = instance_results.iter().filter(|ok| **ok).count();

    let total_endpoints = Arc::try_unwrap(total_endpoints)
        .expect("Failed to unwrap total_endpoints Arc")
        .into_inner();
    let zones_created = Arc::try_unwrap(zones_created)
        .expect("Failed to unwrap zones_created Arc")
        .into_inner();
    let errors = Arc::try_unwrap(errors)
        .expect("Failed to unwrap errors Arc")
        .into_inner();

    // If ALL operations failed, return an error
    if total_endpoints == 0 && !errors.is_empty() {
        return Err(anyhow!(
            "Failed to add zone {} to all secondary instances. Errors: {}",
            spec.zone_name,
            errors.join("; ")
        ));
    }

    info!(
        "Successfully configured secondary zone {} on {} endpoint(s) across {}/{} fully configured secondary instance(s)",
        spec.zone_name,
        total_endpoints,
        instances_configured,
        secondary_instance_refs.len()
    );

    Ok(types::ZoneConfigOutcome {
        instances_configured,
        endpoints_configured: total_endpoints,
        zones_created,
    })
}

/// Deletes a DNS zone and its associated zone files.
///
/// # Arguments
///
/// * `_client` - Kubernetes API client (unused, for future extensions)
/// * `dnszone` - The `DNSZone` resource to delete
/// * `zone_manager` - BIND9 manager for removing zone files
///
/// # Returns
///
/// * `Ok(())` - If zone was deleted successfully
/// * `Err(_)` - If zone deletion failed
///
/// # Errors
///
/// Returns an error if BIND9 zone deletion fails.
pub async fn delete_dnszone(
    ctx: Arc<crate::context::Context>,
    dnszone: DNSZone,
    zone_manager: &crate::bind9::Bind9Manager,
) -> Result<()> {
    let client = ctx.client.clone();
    let bind9_instances_store = &ctx.stores.bind9_instances;
    let namespace = dnszone.namespace().unwrap_or_default();
    let name = dnszone.name_any();
    let spec = &dnszone.spec;

    info!("Deleting DNSZone {}/{}", namespace, name);

    // Get instances from new architecture (spec.bind9Instances or status.bind9Instances)
    // If zone has no instances assigned (e.g., orphaned zone), still allow deletion
    let instance_refs = match validation::get_instances_from_zone(&dnszone, bind9_instances_store) {
        Ok(refs) => refs,
        Err(e) => {
            warn!(
                "DNSZone {}/{} has no instances assigned: {}. Allowing deletion anyway.",
                namespace, name, e
            );
            return Ok(());
        }
    };

    // Filter to primary and secondary instances
    let primary_instance_refs = primary::filter_primary_instances(&client, &instance_refs).await?;
    let secondary_instance_refs =
        secondary::filter_secondary_instances(&client, &instance_refs).await?;

    // Delete from all primary instances.
    // Deletion cleanup uses SkipUnavailable: an instance with zero ready
    // endpoints must not block finalizer removal forever - its DNS data is
    // unreachable (and lost anyway with ephemeral storage). Real API errors
    // still propagate so the next reconcile retries.
    if !primary_instance_refs.is_empty() {
        let (_first_endpoint, total_endpoints) = helpers::for_each_instance_endpoint_with_policy(
            &client,
            &primary_instance_refs,
            false, // with_rndc_key = false for zone deletion
            "http", // Use HTTP API port for zone deletion via bindcar API
            helpers::EndpointFailurePolicy::SkipUnavailable,
            |pod_endpoint, instance_name, _rndc_key| {
                let zone_name = spec.zone_name.clone();
                let zone_manager = zone_manager.clone();

                async move {
                    info!(
                        "Deleting zone {} from endpoint {} (instance: {})",
                        zone_name, pod_endpoint, instance_name
                    );

                    // Attempt to delete zone - if it fails (zone not found, endpoint unreachable, etc.),
                    // log a warning but don't fail the deletion. This ensures DNSZones can be deleted
                    // even if BIND9 instances are unavailable or the zone was already removed.
                    // Pass freeze_before_delete=true for primary zones to prevent updates during deletion
                    if let Err(e) = zone_manager.delete_zone(&zone_name, &pod_endpoint, true).await {
                        warn!(
                            "Failed to delete zone {} from endpoint {} (instance: {}): {}. Continuing with deletion anyway.",
                            zone_name, pod_endpoint, instance_name, e
                        );
                    } else {
                        debug!(
                            "Successfully deleted zone {} from endpoint {} (instance: {})",
                            zone_name, pod_endpoint, instance_name
                        );
                    }

                    Ok(())
                }
            },
        )
        .await?;

        info!(
            "Successfully deleted zone {} from {} primary endpoint(s)",
            spec.zone_name, total_endpoints
        );
    }

    // Delete from all secondary instances
    if !secondary_instance_refs.is_empty() {
        let mut secondary_endpoints_deleted = 0;

        for instance_ref in &secondary_instance_refs {
            // Deletion cleanup: skip secondary instances with no reachable
            // endpoints instead of blocking finalizer removal forever. Real
            // (potentially transient) API errors still propagate for retry.
            let endpoints = match helpers::get_endpoint(
                &client,
                &instance_ref.namespace,
                &instance_ref.name,
                "http",
            )
            .await
            {
                Ok(eps) => eps,
                Err(e) if helpers::is_unavailable_for_deletion(&e) => {
                    warn!(
                        "SKIPPING secondary instance {}/{} during zone deletion: no reachable endpoints ({e:#}). \
                         Zone data on this instance cannot be cleaned up and may be orphaned.",
                        instance_ref.namespace, instance_ref.name
                    );
                    continue;
                }
                Err(e) => return Err(e),
            };

            for endpoint in &endpoints {
                let pod_endpoint = format!("{}:{}", endpoint.ip, endpoint.port);

                info!(
                    "Deleting zone {} from secondary endpoint {} (instance: {}/{})",
                    spec.zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name
                );

                // Attempt to delete zone - if it fails, log a warning but don't fail the deletion
                // Pass freeze_before_delete=false for secondary zones (they are read-only, no need to freeze)
                if let Err(e) = zone_manager
                    .delete_zone(&spec.zone_name, &pod_endpoint, false)
                    .await
                {
                    warn!(
                        "Failed to delete zone {} from secondary endpoint {} (instance: {}/{}): {}. Continuing with deletion anyway.",
                        spec.zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name, e
                    );
                } else {
                    debug!(
                        "Successfully deleted zone {} from secondary endpoint {} (instance: {}/{})",
                        spec.zone_name, pod_endpoint, instance_ref.namespace, instance_ref.name
                    );
                    secondary_endpoints_deleted += 1;
                }
            }
        }

        info!(
            "Successfully deleted zone {} from {} secondary endpoint(s)",
            spec.zone_name, secondary_endpoints_deleted
        );
    }

    // Note: We don't need to reload after delzone because:
    // 1. rndc delzone immediately removes the zone from BIND9's running config
    // 2. BIND9 will clean up the zone file and journal files automatically

    Ok(())
}

/// Auto-generates NS records for all nameservers in the zone.
///
/// This function is called after zone creation to add NS records for secondary nameservers
/// specified in the `nameServers` field. The primary nameserver NS record is already created
/// by bindcar during zone initialization (from SOA).
///
/// # Arguments
/// * `client` - Kubernetes client for loading RNDC keys and getting endpoints
/// * `effective_name_servers` - List of nameservers from `nameServers` field
/// * `zone_name` - The DNS zone name
/// * `ttl` - TTL for the NS and glue records
/// * `primary_instance_refs` - List of primary instances to update
///
/// # Returns
/// Result indicating success or failure
///
/// # Errors
/// Returns error if NS record or glue record addition fails
#[allow(clippy::too_many_lines)]
async fn auto_generate_ns_records(
    client: &kube::Client,
    effective_name_servers: &[crate::crd::NameServer],
    zone_name: &str,
    ttl: Option<i32>,
    primary_instance_refs: &[crate::crd::InstanceReference],
) -> Result<()> {
    if effective_name_servers.is_empty() {
        return Ok(());
    }

    info!(
        "Auto-generating {} NS record(s) for zone {}",
        effective_name_servers.len(),
        zone_name
    );

    for nameserver in effective_name_servers {
        // Add NS record at zone apex (@)
        info!(
            "Adding NS record: {} IN NS {}",
            zone_name, nameserver.hostname
        );

        for instance_ref in primary_instance_refs {
            // Load RNDC key for this instance
            let key_data = match helpers::load_rndc_key(
                client,
                &instance_ref.namespace,
                &instance_ref.name,
            )
            .await
            {
                Ok(key) => key,
                Err(e) => {
                    warn!(
                        "Failed to load RNDC key for instance {}/{}: {}. Skipping NS record addition.",
                        instance_ref.namespace, instance_ref.name, e
                    );
                    continue;
                }
            };

            // Get endpoints for this instance
            let endpoints = match helpers::get_endpoint(
                client,
                &instance_ref.namespace,
                &instance_ref.name,
                "dns-tcp",
            )
            .await
            {
                Ok(eps) => eps,
                Err(e) => {
                    warn!(
                        "Failed to get endpoints for instance {}/{}: {}. Skipping NS record addition.",
                        instance_ref.namespace, instance_ref.name, e
                    );
                    continue;
                }
            };

            // Add NS record to all endpoints of this instance
            for endpoint in &endpoints {
                let pod_endpoint = format!("{}:{}", endpoint.ip, endpoint.port);

                if let Err(e) = crate::bind9::records::ns::add_ns_record(
                    zone_name,
                    "@", // Zone apex
                    &nameserver.hostname,
                    ttl,
                    &pod_endpoint,
                    &key_data,
                )
                .await
                {
                    warn!(
                        "Failed to add NS record for {} to endpoint {} (instance {}/{}): {}",
                        nameserver.hostname,
                        pod_endpoint,
                        instance_ref.namespace,
                        instance_ref.name,
                        e
                    );
                    // Continue with other endpoints - partial success is acceptable
                }
            }
        }

        // Add glue records if IPs provided (for in-zone nameservers)
        if let Some(ref ipv4) = nameserver.ipv4_address {
            add_glue_record(
                client,
                zone_name,
                &nameserver.hostname,
                ipv4,
                hickory_proto::rr::RecordType::A,
                ttl,
                primary_instance_refs,
            )
            .await?;
        }

        if let Some(ref ipv6) = nameserver.ipv6_address {
            add_glue_record(
                client,
                zone_name,
                &nameserver.hostname,
                ipv6,
                hickory_proto::rr::RecordType::AAAA,
                ttl,
                primary_instance_refs,
            )
            .await?;
        }
    }

    info!(
        "Successfully auto-generated NS records and glue records for zone {}",
        zone_name
    );

    Ok(())
}

/// Adds a glue record (A or AAAA) for an in-zone nameserver.
///
/// Glue records provide IP addresses for nameservers within the zone's own domain.
/// This is necessary to avoid circular dependencies when resolving the nameserver itself.
///
/// # Arguments
/// * `client` - Kubernetes client for loading RNDC keys and getting endpoints
/// * `zone_name` - The DNS zone name
/// * `hostname` - Full nameserver hostname (e.g., "ns2.example.com.")
/// * `ip_address` - IP address (IPv4 or IPv6)
/// * `record_type` - Type of glue record (A or AAAA)
/// * `ttl` - TTL for the glue record
/// * `primary_instance_refs` - List of primary instances to update
///
/// # Returns
/// Result indicating success or failure
///
/// # Errors
/// Returns error if glue record addition fails on all instances
#[allow(clippy::too_many_lines)]
async fn add_glue_record(
    client: &kube::Client,
    zone_name: &str,
    hostname: &str,
    ip_address: &str,
    record_type: hickory_proto::rr::RecordType,
    ttl: Option<i32>,
    primary_instance_refs: &[crate::crd::InstanceReference],
) -> Result<()> {
    // Extract record name from hostname
    // Example: "ns2.example.com." in zone "example.com" → name = "ns2"
    let name = hostname
        .trim_end_matches('.')
        .strip_suffix(&format!(".{}", zone_name.trim_end_matches('.')))
        .unwrap_or_else(|| hostname.trim_end_matches('.'));

    // Check if this is actually an in-zone nameserver
    if name == hostname.trim_end_matches('.') {
        // Hostname doesn't end with zone name - this is an out-of-zone nameserver
        // No glue record needed
        debug!(
            "Skipping glue record for out-of-zone nameserver {} (not in zone {})",
            hostname, zone_name
        );
        return Ok(());
    }

    info!(
        "Adding {} glue record: {} IN {} {}",
        if record_type == hickory_proto::rr::RecordType::A {
            "A"
        } else {
            "AAAA"
        },
        name,
        if record_type == hickory_proto::rr::RecordType::A {
            "A"
        } else {
            "AAAA"
        },
        ip_address
    );

    let mut success_count = 0;
    let mut errors = Vec::new();

    for instance_ref in primary_instance_refs {
        // Load RNDC key for this instance
        let key_data = match helpers::load_rndc_key(
            client,
            &instance_ref.namespace,
            &instance_ref.name,
        )
        .await
        {
            Ok(key) => key,
            Err(e) => {
                warn!(
                    "Failed to load RNDC key for instance {}/{}: {}. Skipping glue record addition.",
                    instance_ref.namespace, instance_ref.name, e
                );
                continue;
            }
        };

        // Get endpoints for this instance
        let endpoints = match helpers::get_endpoint(
            client,
            &instance_ref.namespace,
            &instance_ref.name,
            "dns-tcp",
        )
        .await
        {
            Ok(eps) => eps,
            Err(e) => {
                warn!(
                    "Failed to get endpoints for instance {}/{}: {}. Skipping glue record addition.",
                    instance_ref.namespace, instance_ref.name, e
                );
                continue;
            }
        };

        // Add glue record to all endpoints of this instance
        for endpoint in &endpoints {
            let pod_endpoint = format!("{}:{}", endpoint.ip, endpoint.port);

            let result = match record_type {
                hickory_proto::rr::RecordType::A => {
                    crate::bind9::records::a::add_a_record(
                        zone_name,
                        name,
                        &[ip_address.to_string()],
                        ttl,
                        &pod_endpoint,
                        &key_data,
                    )
                    .await
                }
                hickory_proto::rr::RecordType::AAAA => {
                    crate::bind9::records::a::add_aaaa_record(
                        zone_name,
                        name,
                        &[ip_address.to_string()],
                        ttl,
                        &pod_endpoint,
                        &key_data,
                    )
                    .await
                }
                _ => {
                    return Err(anyhow::anyhow!(
                        "Invalid record type for glue record: {:?}",
                        record_type
                    ))
                }
            };

            match result {
                Ok(()) => {
                    success_count += 1;
                }
                Err(e) => {
                    warn!(
                        "Failed to add glue record {} to endpoint {} (instance {}/{}): {}",
                        name, pod_endpoint, instance_ref.namespace, instance_ref.name, e
                    );
                    errors.push(format!(
                        "endpoint {} (instance {}/{}): {}",
                        pod_endpoint, instance_ref.namespace, instance_ref.name, e
                    ));
                }
            }
        }
    }

    // Accept partial success - at least one endpoint updated
    if success_count > 0 {
        Ok(())
    } else {
        Err(anyhow::anyhow!(
            "Failed to add glue record {} to all instances. Errors: {}",
            name,
            errors.join("; ")
        ))
    }
}

#[cfg(test)]
#[path = "dnszone_tests.rs"]
mod dnszone_tests;