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

use crate::StorageAPI;
use crate::bucket::bucket_target_sys::BucketTargetSys;
use crate::bucket::metadata_sys;
use crate::bucket::replication::ResyncOpts;
use crate::bucket::replication::ResyncStatusType;
use crate::bucket::replication::replicate_delete;
use crate::bucket::replication::replicate_object;
use crate::bucket::replication::replication_resyncer::{
    BucketReplicationResyncStatus, DeletedObjectReplicationInfo, REPLICATION_DIR, RESYNC_FILE_NAME, ReplicationConfig,
    ReplicationResyncer, decode_resync_file, get_heal_replicate_object_info,
};
use crate::bucket::replication::replication_state::ReplicationStats;
use crate::config::com::read_config;
use crate::disk::BUCKET_META_PREFIX;
use crate::error::Error as EcstoreError;
use crate::store_api::ObjectInfo;
use lazy_static::lazy_static;
use rustfs_filemeta::MrfReplicateEntry;
use rustfs_filemeta::ReplicateDecision;
use rustfs_filemeta::ReplicateObjectInfo;
use rustfs_filemeta::ReplicatedTargetInfo;
use rustfs_filemeta::ReplicationStatusType;
use rustfs_filemeta::ReplicationType;
use rustfs_filemeta::ReplicationWorkerOperation;
use rustfs_filemeta::ResyncDecision;
use rustfs_filemeta::VersionPurgeStatusType;
use rustfs_filemeta::replication_statuses_map;
use rustfs_filemeta::version_purge_statuses_map;
use rustfs_filemeta::{REPLICATE_EXISTING, REPLICATE_HEAL, REPLICATE_HEAL_DELETE};
use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str};
use std::any::Any;
use std::sync::Arc;
use std::sync::atomic::AtomicI32;
use std::sync::atomic::Ordering;
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use tokio::sync::Mutex;
use tokio::sync::RwLock;
use tokio::sync::mpsc;
use tokio::sync::mpsc::Receiver;
use tokio::sync::mpsc::Sender;
use tokio::task::JoinHandle;
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;
use tracing::{info, instrument, warn};

// Worker limits
pub const WORKER_MAX_LIMIT: usize = 500;
pub const WORKER_MIN_LIMIT: usize = 50;
pub const WORKER_AUTO_DEFAULT: usize = 100;
pub const MRF_WORKER_MAX_LIMIT: usize = 8;
pub const MRF_WORKER_MIN_LIMIT: usize = 2;
pub const MRF_WORKER_AUTO_DEFAULT: usize = 4;
pub const LARGE_WORKER_COUNT: usize = 10;
pub const MIN_LARGE_OBJ_SIZE: i64 = 128 * 1024 * 1024; // 128MiB

/// Priority levels for replication
#[derive(Debug, Clone, PartialEq)]
pub enum ReplicationPriority {
    Fast,
    Slow,
    Auto,
}

impl std::str::FromStr for ReplicationPriority {
    type Err = ();

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "fast" => Ok(ReplicationPriority::Fast),
            "slow" => Ok(ReplicationPriority::Slow),
            "auto" => Ok(ReplicationPriority::Auto),
            _ => Ok(ReplicationPriority::Auto), // Default to Auto for unknown values
        }
    }
}

impl ReplicationPriority {
    pub fn as_str(&self) -> &'static str {
        match self {
            ReplicationPriority::Fast => "fast",
            ReplicationPriority::Slow => "slow",
            ReplicationPriority::Auto => "auto",
        }
    }
}

/// Enum for different types of replication operations
#[derive(Debug)]
pub enum ReplicationOperation {
    Object(Box<ReplicateObjectInfo>),
    Delete(Box<DeletedObjectReplicationInfo>),
}

impl ReplicationWorkerOperation for ReplicationOperation {
    fn as_any(&self) -> &dyn Any {
        self
    }

    fn to_mrf_entry(&self) -> MrfReplicateEntry {
        match self {
            ReplicationOperation::Object(obj) => obj.to_mrf_entry(),
            ReplicationOperation::Delete(del) => del.to_mrf_entry(),
        }
    }

    fn get_bucket(&self) -> &str {
        match self {
            ReplicationOperation::Object(obj) => obj.get_bucket(),
            ReplicationOperation::Delete(del) => del.get_bucket(),
        }
    }

    fn get_object(&self) -> &str {
        match self {
            ReplicationOperation::Object(obj) => obj.get_object(),
            ReplicationOperation::Delete(del) => del.get_object(),
        }
    }

    fn get_size(&self) -> i64 {
        match self {
            ReplicationOperation::Object(obj) => obj.get_size(),
            ReplicationOperation::Delete(del) => del.get_size(),
        }
    }

    fn is_delete_marker(&self) -> bool {
        match self {
            ReplicationOperation::Object(obj) => obj.is_delete_marker(),
            ReplicationOperation::Delete(del) => del.is_delete_marker(),
        }
    }

    fn get_op_type(&self) -> ReplicationType {
        match self {
            ReplicationOperation::Object(obj) => obj.get_op_type(),
            ReplicationOperation::Delete(del) => del.get_op_type(),
        }
    }
}

/// Replication pool options
#[derive(Debug, Clone)]
pub struct ReplicationPoolOpts {
    pub priority: ReplicationPriority,
    pub max_workers: Option<usize>,
    pub max_l_workers: Option<usize>,
}

impl Default for ReplicationPoolOpts {
    fn default() -> Self {
        Self {
            priority: ReplicationPriority::Auto,
            max_workers: None,
            max_l_workers: None,
        }
    }
}
/// Main replication pool structure
#[derive(Debug)]
pub struct ReplicationPool<S: StorageAPI> {
    // Atomic counters for active workers
    active_workers: Arc<AtomicI32>,
    active_lrg_workers: Arc<AtomicI32>,
    active_mrf_workers: Arc<AtomicI32>,

    storage: Arc<S>,

    // Configuration
    priority: RwLock<ReplicationPriority>,
    max_workers: RwLock<usize>,
    max_l_workers: RwLock<usize>,

    // Statistics
    stats: Arc<ReplicationStats>,

    // Worker channels
    workers: RwLock<Vec<Sender<ReplicationOperation>>>,
    lrg_workers: RwLock<Vec<Sender<ReplicationOperation>>>,

    // MRF (Most Recent Failures) channels
    mrf_replica_tx: Sender<ReplicationOperation>,
    mrf_replica_rx: Mutex<Option<Receiver<ReplicationOperation>>>,
    mrf_save_tx: Sender<MrfReplicateEntry>,
    mrf_save_rx: Mutex<Option<Receiver<MrfReplicateEntry>>>,

    // Control channels
    mrf_worker_kill_tx: Sender<()>,
    mrf_stop_tx: Sender<()>,

    // Worker size tracking
    mrf_worker_size: AtomicI32,

    // Task handles for cleanup
    task_handles: Mutex<Vec<JoinHandle<()>>>,

    // Replication resyncer for handling bucket resync operations
    resyncer: Arc<ReplicationResyncer>,
}

impl<S: StorageAPI> ReplicationPool<S> {
    /// Creates a new replication pool with specified options
    pub async fn new(opts: ReplicationPoolOpts, stats: Arc<ReplicationStats>, storage: Arc<S>) -> Arc<Self> {
        let max_workers = opts.max_workers.unwrap_or(WORKER_MAX_LIMIT);

        let (workers, failed_workers) = match opts.priority {
            ReplicationPriority::Fast => (WORKER_MAX_LIMIT, MRF_WORKER_MAX_LIMIT),
            ReplicationPriority::Slow => (WORKER_MIN_LIMIT, MRF_WORKER_MIN_LIMIT),
            ReplicationPriority::Auto => (WORKER_AUTO_DEFAULT, MRF_WORKER_AUTO_DEFAULT),
        };

        let workers = std::cmp::min(workers, max_workers);
        let failed_workers = std::cmp::min(failed_workers, max_workers);

        let max_l_workers = opts.max_l_workers.unwrap_or(LARGE_WORKER_COUNT);

        // Create MRF channels
        let (mrf_replica_tx, mrf_replica_rx) = mpsc::channel(100000);
        let (mrf_save_tx, mrf_save_rx) = mpsc::channel(100000);
        let (mrf_worker_kill_tx, _mrf_worker_kill_rx) = mpsc::channel(failed_workers);
        let (mrf_stop_tx, _mrf_stop_rx) = mpsc::channel(1);

        let pool = Arc::new(Self {
            active_workers: Arc::new(AtomicI32::new(0)),
            active_lrg_workers: Arc::new(AtomicI32::new(0)),
            active_mrf_workers: Arc::new(AtomicI32::new(0)),
            priority: RwLock::new(opts.priority),
            max_workers: RwLock::new(max_workers),
            max_l_workers: RwLock::new(max_l_workers),
            stats,
            storage,
            workers: RwLock::new(Vec::new()),
            lrg_workers: RwLock::new(Vec::new()),
            mrf_replica_tx,
            mrf_replica_rx: Mutex::new(Some(mrf_replica_rx)),
            mrf_save_tx,
            mrf_save_rx: Mutex::new(Some(mrf_save_rx)),
            mrf_worker_kill_tx,
            mrf_stop_tx,
            mrf_worker_size: AtomicI32::new(0),
            task_handles: Mutex::new(Vec::new()),
            resyncer: Arc::new(ReplicationResyncer::new().await),
        });

        // Initialize workers
        pool.resize_lrg_workers(max_l_workers, 0).await;
        pool.resize_workers(workers, 0).await;
        pool.resize_failed_workers(failed_workers as i32).await;

        // Start background tasks
        pool.start_mrf_processor().await;
        pool.start_mrf_persister().await;

        pool
    }

    /// Returns the number of active workers handling replication traffic
    pub fn active_workers(&self) -> i32 {
        self.active_workers.load(Ordering::SeqCst)
    }

    /// Returns the number of active workers handling replication failures
    pub fn active_mrf_workers(&self) -> i32 {
        self.active_mrf_workers.load(Ordering::SeqCst)
    }

    /// Returns the number of active workers handling traffic > 128MiB object size
    pub fn active_lrg_workers(&self) -> i32 {
        self.active_lrg_workers.load(Ordering::SeqCst)
    }

    /// Resizes the large workers pool
    pub async fn resize_lrg_workers(&self, n: usize, check_old: usize) {
        let mut lrg_workers = self.lrg_workers.write().await;

        if (check_old > 0 && lrg_workers.len() != check_old) || n == lrg_workers.len() || n < 1 {
            return;
        }

        // Add workers if needed
        while lrg_workers.len() < n {
            let (tx, rx) = mpsc::channel(100000);
            lrg_workers.push(tx);

            let active_counter = self.active_lrg_workers.clone();
            let storage = self.storage.clone();

            let handle = tokio::spawn(async move {
                let mut rx = rx;
                while let Some(operation) = rx.recv().await {
                    active_counter.fetch_add(1, Ordering::SeqCst);

                    match operation {
                        ReplicationOperation::Object(obj_info) => {
                            replicate_object(*obj_info, storage.clone()).await;
                        }
                        ReplicationOperation::Delete(del_info) => {
                            replicate_delete(*del_info, storage.clone()).await;
                        }
                    }

                    active_counter.fetch_sub(1, Ordering::SeqCst);
                }
            });

            self.task_handles.lock().await.push(handle);
        }

        // Remove workers if needed
        while lrg_workers.len() > n {
            if let Some(worker) = lrg_workers.pop() {
                drop(worker); // Closing the channel will terminate the worker
            }
        }
    }

    /// Resizes the regular workers pool
    pub async fn resize_workers(&self, n: usize, check_old: usize) {
        let mut workers = self.workers.write().await;

        if (check_old > 0 && workers.len() != check_old) || n == workers.len() || n < 1 {
            warn!(
                "resize_workers: skipping resize - check_old_mismatch={}, same_size={}, invalid_n={}",
                check_old > 0 && workers.len() != check_old,
                n == workers.len(),
                n < 1
            );
            return;
        }

        // Add workers if needed
        if workers.len() < n {
            info!("resize_workers: adding workers from {} to {}", workers.len(), n);
        }

        while workers.len() < n {
            let (tx, rx) = mpsc::channel(10000);
            workers.push(tx);

            let active_counter = self.active_workers.clone();
            let stats = self.stats.clone();
            let storage = self.storage.clone();

            let handle = tokio::spawn(async move {
                let mut rx = rx;
                while let Some(operation) = rx.recv().await {
                    active_counter.fetch_add(1, Ordering::SeqCst);

                    match operation {
                        ReplicationOperation::Object(obj_info) => {
                            stats
                                .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
                                .await;

                            // Perform actual replication (placeholder)
                            replicate_object(obj_info.as_ref().clone(), storage.clone()).await;

                            stats
                                .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
                                .await;
                        }
                        ReplicationOperation::Delete(del_info) => {
                            stats.inc_q(&del_info.bucket, 0, true, del_info.op_type).await;
                            // Perform actual delete replication (placeholder)
                            replicate_delete(del_info.as_ref().clone(), storage.clone()).await;

                            stats.dec_q(&del_info.bucket, 0, true, del_info.op_type).await;
                        }
                    }

                    active_counter.fetch_sub(1, Ordering::SeqCst);
                }
            });

            self.task_handles.lock().await.push(handle);
        }

        // Remove workers if needed
        if workers.len() > n {
            warn!("resize_workers: removing workers from {} to {}", workers.len(), n);
        }

        while workers.len() > n {
            if let Some(worker) = workers.pop() {
                drop(worker); // Closing the channel will terminate the worker
            }
        }
    }

    /// Resizes the failed workers pool
    pub async fn resize_failed_workers(&self, n: i32) {
        // Add workers if needed
        while self.mrf_worker_size.load(Ordering::SeqCst) < n {
            self.mrf_worker_size.fetch_add(1, Ordering::SeqCst);

            let active_counter = self.active_mrf_workers.clone();
            let stats = self.stats.clone();
            let storage = self.storage.clone();
            let mrf_rx = self.mrf_replica_rx.lock().await.take();

            if let Some(rx) = mrf_rx {
                let handle = tokio::spawn(async move {
                    let mut rx = rx;
                    while let Some(operation) = rx.recv().await {
                        active_counter.fetch_add(1, Ordering::SeqCst);

                        match operation {
                            ReplicationOperation::Object(obj_info) => {
                                stats
                                    .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
                                    .await;

                                replicate_object(obj_info.as_ref().clone(), storage.clone()).await;

                                stats
                                    .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
                                    .await;
                            }
                            ReplicationOperation::Delete(del_info) => {
                                replicate_delete(*del_info, storage.clone()).await;
                            }
                        }

                        active_counter.fetch_sub(1, Ordering::SeqCst);
                    }
                });
                self.task_handles.lock().await.push(handle);
                break; // Only one receiver can be taken
            }
        }

        // Remove workers if needed
        while self.mrf_worker_size.load(Ordering::SeqCst) > n {
            self.mrf_worker_size.fetch_sub(1, Ordering::SeqCst);
            let _ = self.mrf_worker_kill_tx.try_send(()); // Signal worker to stop
        }
    }

    /// Resizes worker priority and counts
    pub async fn resize_worker_priority(
        &self,
        pri: ReplicationPriority,
        max_workers: Option<usize>,
        max_l_workers: Option<usize>,
    ) {
        let (workers, mrf_workers) = match pri {
            ReplicationPriority::Fast => (WORKER_MAX_LIMIT, MRF_WORKER_MAX_LIMIT),
            ReplicationPriority::Slow => (WORKER_MIN_LIMIT, MRF_WORKER_MIN_LIMIT),
            ReplicationPriority::Auto => {
                let mut workers = WORKER_AUTO_DEFAULT;
                let mut mrf_workers = MRF_WORKER_AUTO_DEFAULT;

                let current_workers = self.workers.read().await.len();
                if current_workers < WORKER_AUTO_DEFAULT {
                    workers = std::cmp::min(current_workers + 1, WORKER_AUTO_DEFAULT);
                }

                let current_mrf = self.mrf_worker_size.load(Ordering::SeqCst) as usize;
                if current_mrf < MRF_WORKER_AUTO_DEFAULT {
                    mrf_workers = std::cmp::min(current_mrf + 1, MRF_WORKER_AUTO_DEFAULT);
                }
                (workers, mrf_workers)
            }
        };

        let (final_workers, final_mrf_workers) = if let Some(max_w) = max_workers {
            *self.max_workers.write().await = max_w;
            (std::cmp::min(workers, max_w), std::cmp::min(mrf_workers, max_w))
        } else {
            (workers, mrf_workers)
        };

        let max_l_workers_val = max_l_workers.unwrap_or(LARGE_WORKER_COUNT);
        *self.max_l_workers.write().await = max_l_workers_val;
        *self.priority.write().await = pri;

        self.resize_workers(final_workers, 0).await;
        self.resize_failed_workers(final_mrf_workers as i32).await;
        self.resize_lrg_workers(max_l_workers_val, 0).await;
    }

    /// Gets a worker channel deterministically based on bucket and object names
    async fn get_worker_ch(&self, bucket: &str, object: &str, _size: i64) -> Option<Sender<ReplicationOperation>> {
        use std::collections::hash_map::DefaultHasher;
        use std::hash::{Hash, Hasher};

        let mut hasher = DefaultHasher::new();
        format!("{bucket}{object}").hash(&mut hasher);
        let hash = hasher.finish();

        let workers = self.workers.read().await;
        if workers.is_empty() {
            return None;
        }

        let index = (hash as usize) % workers.len();
        workers.get(index).cloned()
    }

    /// Queues a replica task
    pub async fn queue_replica_task(&self, ri: ReplicateObjectInfo) {
        // If object is large, queue it to a static set of large workers
        if ri.size >= MIN_LARGE_OBJ_SIZE {
            use std::collections::hash_map::DefaultHasher;
            use std::hash::{Hash, Hasher};

            let mut hasher = DefaultHasher::new();
            format!("{}{}", ri.bucket, ri.name).hash(&mut hasher);
            let hash = hasher.finish();

            let lrg_workers = self.lrg_workers.read().await;

            if !lrg_workers.is_empty() {
                let index = (hash as usize) % lrg_workers.len();

                if let Some(worker) = lrg_workers.get(index)
                    && worker.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_err()
                {
                    // Queue to MRF if worker is busy
                    let _ = self.mrf_save_tx.try_send(ri.to_mrf_entry());

                    // Try to add more workers if possible
                    let max_l_workers = *self.max_l_workers.read().await;
                    let existing = lrg_workers.len();
                    if self.active_lrg_workers() < std::cmp::min(max_l_workers, LARGE_WORKER_COUNT) as i32 {
                        let workers = std::cmp::min(existing + 1, max_l_workers);

                        drop(lrg_workers);
                        self.resize_lrg_workers(workers, existing).await;
                    }
                }
            }
            return;
        }

        // Handle regular sized objects

        let ch = match ri.op_type {
            ReplicationType::Heal | ReplicationType::ExistingObject => Some(self.mrf_replica_tx.clone()),
            _ => self.get_worker_ch(&ri.bucket, &ri.name, ri.size).await,
        };

        if let Some(channel) = ch
            && channel.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_err()
        {
            // Queue to MRF if all workers are busy
            let _ = self.mrf_save_tx.try_send(ri.to_mrf_entry());

            // Try to scale up workers based on priority
            let priority = self.priority.read().await.clone();
            let max_workers = *self.max_workers.read().await;

            match priority {
                ReplicationPriority::Fast => {
                    // Log warning about unable to keep up
                    info!("Warning: Unable to keep up with incoming traffic");
                }
                ReplicationPriority::Slow => {
                    info!("Warning: Unable to keep up with incoming traffic - recommend increasing replication priority to auto");
                }
                ReplicationPriority::Auto => {
                    let max_w = std::cmp::min(max_workers, WORKER_MAX_LIMIT);
                    let active_workers = self.active_workers();

                    if active_workers < max_w as i32 {
                        let workers = self.workers.read().await;
                        let new_count = std::cmp::min(workers.len() + 1, max_w);
                        let existing = workers.len();

                        drop(workers);
                        self.resize_workers(new_count, existing).await;
                    }

                    let max_mrf_workers = std::cmp::min(max_workers, MRF_WORKER_MAX_LIMIT);
                    let active_mrf = self.active_mrf_workers();

                    if active_mrf < max_mrf_workers as i32 {
                        let current_mrf = self.mrf_worker_size.load(Ordering::SeqCst);
                        let new_mrf = std::cmp::min(current_mrf + 1, max_mrf_workers as i32);

                        self.resize_failed_workers(new_mrf).await;
                    }
                }
            }
        }
    }

    /// Queues a replica delete task
    pub async fn queue_replica_delete_task(&self, doi: DeletedObjectReplicationInfo) {
        let ch = match doi.op_type {
            ReplicationType::Heal | ReplicationType::ExistingObject => Some(self.mrf_replica_tx.clone()),
            _ => self.get_worker_ch(&doi.bucket, &doi.delete_object.object_name, 0).await,
        };

        if let Some(channel) = ch
            && channel.try_send(ReplicationOperation::Delete(Box::new(doi.clone()))).is_err()
        {
            let _ = self.mrf_save_tx.try_send(doi.to_mrf_entry());

            let priority = self.priority.read().await.clone();
            let max_workers = *self.max_workers.read().await;

            match priority {
                ReplicationPriority::Fast => {
                    info!("Warning: Unable to keep up with incoming deletes");
                }
                ReplicationPriority::Slow => {
                    info!("Warning: Unable to keep up with incoming deletes - recommend increasing replication priority to auto");
                }
                ReplicationPriority::Auto => {
                    let max_w = std::cmp::min(max_workers, WORKER_MAX_LIMIT);
                    if self.active_workers() < max_w as i32 {
                        let workers = self.workers.read().await;
                        let new_count = std::cmp::min(workers.len() + 1, max_w);
                        let existing = workers.len();
                        drop(workers);
                        self.resize_workers(new_count, existing).await;
                    }
                }
            }
        }
    }

    /// Queues an MRF save operation
    async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
        let _ = self.mrf_save_tx.try_send(entry);
    }

    /// Starts the MRF processor background task
    async fn start_mrf_processor(&self) {
        // This would start a background task to process MRF entries
        // Implementation depends on the actual MRF processing logic
    }

    /// Starts the MRF persister background task  
    async fn start_mrf_persister(&self) {
        // This would start a background task to persist MRF entries to disk
        // Implementation depends on the actual persistence logic
    }

    /// Worker function for handling regular replication operations
    async fn add_worker(
        &self,
        mut rx: Receiver<ReplicationOperation>,
        active_counter: Arc<AtomicI32>,
        stats: Arc<ReplicationStats>,
    ) {
        while let Some(operation) = rx.recv().await {
            active_counter.fetch_add(1, Ordering::SeqCst);

            match operation {
                ReplicationOperation::Object(obj_info) => {
                    stats
                        .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
                        .await;

                    // Perform actual replication (placeholder)
                    replicate_object(obj_info.as_ref().clone(), self.storage.clone()).await;

                    stats
                        .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
                        .await;
                }
                ReplicationOperation::Delete(del_info) => {
                    stats.inc_q(&del_info.bucket, 0, true, del_info.op_type).await;

                    // Perform actual delete replication (placeholder)
                    replicate_delete(del_info.as_ref().clone(), self.storage.clone()).await;

                    stats.dec_q(&del_info.bucket, 0, true, del_info.op_type).await;
                }
            }

            active_counter.fetch_sub(1, Ordering::SeqCst);
        }
    }

    /// Worker function for handling large object replication operations
    async fn add_large_worker(&self, mut rx: Receiver<ReplicationOperation>, active_counter: Arc<AtomicI32>, storage: Arc<S>) {
        while let Some(operation) = rx.recv().await {
            active_counter.fetch_add(1, Ordering::SeqCst);

            match operation {
                ReplicationOperation::Object(obj_info) => {
                    replicate_object(*obj_info, storage.clone()).await;
                }
                ReplicationOperation::Delete(del_info) => {
                    replicate_delete(*del_info, storage.clone()).await;
                }
            }

            active_counter.fetch_sub(1, Ordering::SeqCst);
        }
    }

    /// Worker function for handling MRF (Most Recent Failures) operations
    async fn add_mrf_worker(
        &self,
        mut rx: Receiver<ReplicationOperation>,
        active_counter: Arc<AtomicI32>,
        stats: Arc<ReplicationStats>,
    ) {
        while let Some(operation) = rx.recv().await {
            active_counter.fetch_add(1, Ordering::SeqCst);

            match operation {
                ReplicationOperation::Object(obj_info) => {
                    stats
                        .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
                        .await;

                    replicate_object(obj_info.as_ref().clone(), self.storage.clone()).await;

                    stats
                        .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
                        .await;
                }
                ReplicationOperation::Delete(del_info) => {
                    replicate_delete(*del_info, self.storage.clone()).await;
                }
            }

            active_counter.fetch_sub(1, Ordering::SeqCst);
        }
    }

    /// Delete resync metadata from replication resync state in memory
    pub async fn delete_resync_metadata(&self, bucket: &str) {
        let mut status_map = self.resyncer.status_map.write().await;
        status_map.remove(bucket);
        // Note: global site resync metrics deletion would be handled here
        // global_site_resync_metrics.delete_bucket(bucket);
    }

    /// Initialize bucket replication resync for all buckets
    pub async fn init_resync_internal(
        self: Arc<Self>,
        cancellation_token: CancellationToken,
        buckets: Vec<String>,
    ) -> Result<(), EcstoreError> {
        // Load bucket metadata system in background
        let pool_clone = self.clone();

        tokio::spawn(async move {
            pool_clone.start_resync_routine(buckets, cancellation_token).await;
        });

        Ok(())
    }

    /// Start the resync routine that runs in a loop
    async fn start_resync_routine(self: Arc<Self>, buckets: Vec<String>, cancellation_token: CancellationToken) {
        // Run the replication resync in a loop
        loop {
            let self_clone = self.clone();
            let ctx = cancellation_token.clone();
            tokio::select! {
                _ = cancellation_token.cancelled() => {
                    return;
                }
                result = self_clone.load_resync(&buckets, ctx) => {
                    if result.is_ok() {
                        return;
                    }
                }
            }

            // Generate random duration between 0 and 1 minute
            use rand::RngExt;
            let duration_millis = rand::rng().random_range(0..60_000);
            let mut duration = Duration::from_millis(duration_millis);

            // Make sure to sleep at least a second to avoid high CPU ticks
            if duration < Duration::from_secs(1) {
                duration = Duration::from_secs(1);
            }

            tokio::time::sleep(duration).await;
        }
    }

    /// Load bucket replication resync statuses into memory
    #[instrument(skip(cancellation_token))]
    async fn load_resync(self: Arc<Self>, buckets: &[String], cancellation_token: CancellationToken) -> Result<(), EcstoreError> {
        // TODO: add leader_lock
        // Make sure only one node running resync on the cluster
        // Note: Leader lock implementation would be needed here
        // let _lock_guard = global_leader_lock.get_lock().await?;

        for bucket in buckets {
            let meta = match load_bucket_resync_metadata(bucket, self.storage.clone()).await {
                Ok(meta) => meta,
                Err(err) => {
                    if !matches!(err, EcstoreError::VolumeNotFound) {
                        warn!("Error loading resync metadata for bucket {bucket}: {err:?}");
                    }
                    continue;
                }
            };

            // Store metadata in resyncer
            {
                let mut status_map = self.resyncer.status_map.write().await;
                status_map.insert(bucket.clone(), meta.clone());
            }

            // Process target statistics
            let target_stats = meta.clone_tgt_stats();
            for (arn, stats) in target_stats {
                match stats.resync_status {
                    ResyncStatusType::ResyncFailed | ResyncStatusType::ResyncStarted | ResyncStatusType::ResyncPending => {
                        // Note: This would spawn a resync task in a real implementation
                        // For now, we just log the resync request

                        let ctx = cancellation_token.clone();
                        let bucket_clone = bucket.clone();
                        let resync = self.resyncer.clone();
                        let storage = self.storage.clone();
                        tokio::spawn(async move {
                            resync
                                .resync_bucket(
                                    ctx,
                                    storage,
                                    true,
                                    ResyncOpts {
                                        bucket: bucket_clone,
                                        arn,
                                        resync_id: stats.resync_id,
                                        resync_before: stats.resync_before_date,
                                    },
                                )
                                .await;
                        });
                    }
                    _ => {}
                }
            }
        }

        Ok(())
    }
}

/// Load bucket resync metadata from disk
async fn load_bucket_resync_metadata<S: StorageAPI>(
    bucket: &str,
    obj_api: Arc<S>,
) -> Result<BucketReplicationResyncStatus, EcstoreError> {
    let mut brs = BucketReplicationResyncStatus::new();

    let resync_dir_path = format!("{BUCKET_META_PREFIX}/{bucket}/{REPLICATION_DIR}");
    let resync_file_path = format!("{resync_dir_path}/{RESYNC_FILE_NAME}");

    let data = match read_config(obj_api, &resync_file_path).await {
        Ok(data) => data,
        Err(EcstoreError::ConfigNotFound) => return Ok(brs),
        Err(err) => return Err(err),
    };

    if data.is_empty() {
        // Seems to be empty
        return Ok(brs);
    }

    brs = decode_resync_file(&data)?;

    Ok(brs)
}

// Define a trait object type for the replication pool
pub type DynReplicationPool = dyn ReplicationPoolTrait + Send + Sync;

/// Trait that abstracts the replication pool operations
#[async_trait::async_trait]
pub trait ReplicationPoolTrait: std::fmt::Debug {
    async fn queue_replica_task(&self, ri: ReplicateObjectInfo);
    async fn queue_replica_delete_task(&self, ri: DeletedObjectReplicationInfo);
    async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize);
    async fn init_resync(
        self: Arc<Self>,
        cancellation_token: CancellationToken,
        buckets: Vec<String>,
    ) -> Result<(), EcstoreError>;
}

// Implement the trait for ReplicationPool
#[async_trait::async_trait]
impl<S: StorageAPI> ReplicationPoolTrait for ReplicationPool<S> {
    async fn queue_replica_task(&self, ri: ReplicateObjectInfo) {
        self.queue_replica_task(ri).await;
    }

    async fn queue_replica_delete_task(&self, ri: DeletedObjectReplicationInfo) {
        self.queue_replica_delete_task(ri).await;
    }

    async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize) {
        self.resize(priority, max_workers, max_l_workers).await;
    }

    async fn init_resync(
        self: Arc<Self>,
        cancellation_token: CancellationToken,
        buckets: Vec<String>,
    ) -> Result<(), EcstoreError> {
        self.init_resync_internal(cancellation_token, buckets).await
    }
}

lazy_static! {
    pub static ref GLOBAL_REPLICATION_POOL: tokio::sync::OnceCell<Arc<DynReplicationPool>> = tokio::sync::OnceCell::new();
    pub static ref GLOBAL_REPLICATION_STATS: tokio::sync::OnceCell<Arc<ReplicationStats>> = tokio::sync::OnceCell::new();
}

/// Initializes background replication with the given options
pub async fn init_background_replication<S: StorageAPI>(storage: Arc<S>) {
    let stats = GLOBAL_REPLICATION_STATS
        .get_or_init(|| async {
            let stats = Arc::new(ReplicationStats::new());
            stats.start_background_tasks().await;
            stats
        })
        .await;

    let _pool = GLOBAL_REPLICATION_POOL
        .get_or_init(|| async {
            let pool = ReplicationPool::new(ReplicationPoolOpts::default(), stats.clone(), storage).await;
            pool as Arc<DynReplicationPool>
        })
        .await;

    assert!(GLOBAL_REPLICATION_STATS.get().is_some());
    assert!(GLOBAL_REPLICATION_POOL.get().is_some());
}

pub fn get_global_replication_pool() -> Option<Arc<DynReplicationPool>> {
    GLOBAL_REPLICATION_POOL.get().cloned()
}

pub async fn schedule_replication<S: StorageAPI>(oi: ObjectInfo, o: Arc<S>, dsc: ReplicateDecision, op_type: ReplicationType) {
    let tgt_statuses = replication_statuses_map(&oi.replication_status_internal.clone().unwrap_or_default());
    let purge_statuses = version_purge_statuses_map(&oi.version_purge_status_internal.clone().unwrap_or_default());
    let tm = get_str(&oi.user_defined, SUFFIX_REPLICATION_TIMESTAMP)
        .map(|v| OffsetDateTime::parse(&v, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH));
    let mut rstate = oi.replication_state();
    rstate.replicate_decision_str = dsc.to_string();
    let asz = oi.get_actual_size().unwrap_or_default();

    let mut ri = ReplicateObjectInfo {
        name: oi.name,
        size: oi.size,
        actual_size: asz,
        bucket: oi.bucket,
        version_id: oi.version_id,
        etag: oi.etag,
        mod_time: oi.mod_time,
        replication_status: oi.replication_status,
        replication_status_internal: oi.replication_status_internal,
        delete_marker: oi.delete_marker,
        version_purge_status_internal: oi.version_purge_status_internal,
        version_purge_status: oi.version_purge_status,

        replication_state: Some(rstate),
        op_type,
        dsc: dsc.clone(),
        target_statuses: tgt_statuses,
        target_purge_statuses: purge_statuses,
        replication_timestamp: tm,
        user_tags: oi.user_tags,
        checksum: None,
        retry_count: 0,
        event_type: "".to_string(),
        existing_obj_resync: ResyncDecision::default(),
        ssec: false,
    };

    if ri.ssec {
        ri.checksum = oi.checksum
    }
    if dsc.is_synchronous() {
        replicate_object(ri, o).await
    } else if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
        pool.queue_replica_task(ri).await;
    }
}

pub async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) {
    if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
        pool.queue_replica_delete_task(dv.clone()).await;
    }

    if let (Some(rs), Some(stats)) = (dv.delete_object.replication_state, GLOBAL_REPLICATION_STATS.get()) {
        for (k, _v) in rs.targets.iter() {
            let ri = ReplicatedTargetInfo {
                arn: k.clone(),
                size: 0,
                duration: Duration::default(),
                op_type: ReplicationType::Delete,
                ..Default::default()
            };
            stats
                .update(&dv.bucket, &ri, ReplicationStatusType::Pending, ReplicationStatusType::Empty)
                .await;
        }
    }
}

/// QueueReplicationHeal is a wrapper for queue_replication_heal_internal
pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u32) {
    // ignore modtime zero objects
    if oi.mod_time.is_none() || oi.mod_time == Some(OffsetDateTime::UNIX_EPOCH) {
        return;
    }

    let rcfg = match metadata_sys::get_replication_config(bucket).await {
        Ok((config, _)) => config,
        Err(err) => {
            warn!("Failed to get replication config for bucket {}: {}", bucket, err);

            return;
        }
    };

    let tgts = match BucketTargetSys::get().list_bucket_targets(bucket).await {
        Ok(targets) => Some(targets),
        Err(err) => {
            warn!("Failed to list bucket targets for bucket {}: {}", bucket, err);
            None
        }
    };

    let rcfg_wrapper = ReplicationConfig::new(Some(rcfg), tgts);
    queue_replication_heal_internal(bucket, oi, rcfg_wrapper, retry_count).await;
}

/// queue_replication_heal_internal enqueues objects that failed replication OR eligible for resyncing through
/// an ongoing resync operation or via existing objects replication configuration setting.
pub async fn queue_replication_heal_internal(
    _bucket: &str,
    oi: ObjectInfo,
    rcfg: ReplicationConfig,
    retry_count: u32,
) -> ReplicateObjectInfo {
    let mut roi = ReplicateObjectInfo::default();

    // ignore modtime zero objects
    if oi.mod_time.is_none() || oi.mod_time == Some(OffsetDateTime::UNIX_EPOCH) {
        return roi;
    }

    if rcfg.config.is_none() || rcfg.remotes.is_none() {
        return roi;
    }

    roi = get_heal_replicate_object_info(&oi, &rcfg).await;
    roi.retry_count = retry_count;

    if !roi.dsc.replicate_any() {
        return roi;
    }

    // early return if replication already done, otherwise we need to determine if this
    // version is an existing object that needs healing.
    if roi.replication_status == ReplicationStatusType::Completed
        && roi.version_purge_status.is_empty()
        && !roi.existing_obj_resync.must_resync()
    {
        return roi;
    }

    if roi.delete_marker || !roi.version_purge_status.is_empty() {
        let (version_id, dm_version_id) = if roi.version_purge_status.is_empty() {
            (None, roi.version_id)
        } else {
            (roi.version_id, None)
        };

        let dv = DeletedObjectReplicationInfo {
            delete_object: crate::store_api::DeletedObject {
                object_name: roi.name.clone(),
                delete_marker_version_id: dm_version_id,
                version_id,
                replication_state: roi.replication_state.clone(),
                delete_marker_mtime: roi.mod_time,
                delete_marker: roi.delete_marker,
                ..Default::default()
            },
            bucket: roi.bucket.clone(),
            op_type: ReplicationType::Heal,
            event_type: REPLICATE_HEAL_DELETE.to_string(),
            ..Default::default()
        };

        // heal delete marker replication failure or versioned delete replication failure
        if roi.replication_status == ReplicationStatusType::Pending
            || roi.replication_status == ReplicationStatusType::Failed
            || roi.version_purge_status == VersionPurgeStatusType::Failed
            || roi.version_purge_status == VersionPurgeStatusType::Pending
        {
            if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
                pool.queue_replica_delete_task(dv).await;
            }
            return roi;
        }

        // if replication status is Complete on DeleteMarker and existing object resync required
        let existing_obj_resync = roi.existing_obj_resync.clone();
        if existing_obj_resync.must_resync()
            && (roi.replication_status == ReplicationStatusType::Completed || roi.replication_status.is_empty())
        {
            queue_replicate_deletes_wrapper(dv, existing_obj_resync).await;
            return roi;
        }

        return roi;
    }

    if roi.existing_obj_resync.must_resync() {
        roi.op_type = ReplicationType::ExistingObject;
    }

    match roi.replication_status {
        ReplicationStatusType::Pending | ReplicationStatusType::Failed => {
            roi.event_type = REPLICATE_HEAL.to_string();
            if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
                pool.queue_replica_task(roi.clone()).await;
            }
            return roi;
        }
        _ => {}
    }

    if roi.existing_obj_resync.must_resync() {
        roi.event_type = REPLICATE_EXISTING.to_string();
        if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
            pool.queue_replica_task(roi.clone()).await;
        }
    }

    roi
}

/// Wrapper function for queueing replicate deletes with resync decision
async fn queue_replicate_deletes_wrapper(doi: DeletedObjectReplicationInfo, existing_obj_resync: ResyncDecision) {
    for (k, v) in existing_obj_resync.targets.iter() {
        if v.replicate {
            let mut dv = doi.clone();
            dv.reset_id = v.reset_id.clone();
            dv.target_arn = k.clone();
            if let Some(pool) = GLOBAL_REPLICATION_POOL.get() {
                pool.queue_replica_delete_task(dv).await;
            }
        }
    }
}