use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::watch;

use crate::config::SharedConfig;
use crate::namespace::NamespaceManager;
use crate::queue::{QueueEngine, RemoveTopicResult};
use crate::server::state::ServerState;
use crate::storage::StorageBackend;
use crate::types::{NamespaceId, TopicName, TopicOffset, now_millis};

/// Metadata about a single segment file, used by retention and compaction.
pub(crate) struct SegmentInfo {
    pub(crate) base_offset: TopicOffset,
    pub(crate) path: PathBuf,
    pub(crate) size_bytes: u64,
    pub(crate) mtime_millis: u64,
}

/// Scans a segment directory and returns metadata for each `.seg` file, sorted
/// by base offset ascending.
pub(crate) async fn scan_segments(seg_dir: &Path) -> Vec<SegmentInfo> {
    let Ok(mut read_dir) = tokio::fs::read_dir(seg_dir).await else {
        return Vec::new();
    };

    let mut segments = Vec::new();
    while let Ok(Some(entry)) = read_dir.next_entry().await {
        let path = entry.path();
        if path.extension().and_then(|e| e.to_str()) != Some("seg") {
            continue;
        }
        let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else {
            continue;
        };
        let Ok(base) = stem.parse::<u64>() else {
            continue;
        };
        let Ok(meta) = tokio::fs::metadata(&path).await else {
            continue;
        };
        let mtime_millis = meta
            .modified()
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);

        segments.push(SegmentInfo {
            base_offset: TopicOffset(base),
            path,
            size_bytes: meta.len(),
            mtime_millis,
        });
    }

    segments.sort_by_key(|s| s.base_offset);
    segments
}

/// Metadata stored for each topic in the grace period.
#[derive(Debug, Clone)]
pub struct GracePendingEntry {
    /// When the topic entered the grace period (monotonic).
    pub entered: std::time::Instant,
    /// Grace period duration configured at the time the topic entered grace.
    pub grace_period_secs: u64,
    /// Last activity timestamp (epoch millis) at the time the topic entered grace.
    pub last_activity_millis: u64,
    /// Inactivity threshold that was exceeded (seconds).
    pub threshold_secs: u64,
}

/// Map of topics currently in the grace period before being reaped.
pub type GracePendingMap = HashMap<(NamespaceId, TopicName), GracePendingEntry>;

/// Background retention task that periodically scans segment directories and
/// deletes segments according to per-namespace retention policies.
///
/// Reads retention policies from `SharedConfig` on each sweep, so config
/// hot-reload takes effect without restarting the task.
pub struct RetentionTask<S: StorageBackend, M: NamespaceManager, Q: QueueEngine> {
    storage: std::sync::Arc<S>,
    manager: std::sync::Arc<M>,
    engine: std::sync::Arc<Q>,
    shared_config: SharedConfig,
    grace_pending: Arc<tokio::sync::RwLock<GracePendingMap>>,
}

/// Resolve the effective inactivity threshold for a topic by checking
/// pre-compiled override patterns. First match wins; falls back to `default_secs`.
/// A returned value of `0` means the topic is exempt from inactivity GC.
pub(crate) fn resolve_inactive_threshold(
    topic: &TopicName,
    default_secs: u64,
    overrides: &[crate::config::CompiledInactiveOverride],
) -> u64 {
    for ov in overrides {
        if ov.pattern.matches(&topic.0) {
            return ov.inactive_secs;
        }
    }
    default_secs
}

impl<S: StorageBackend + 'static, M: NamespaceManager + 'static, Q: QueueEngine + 'static>
    RetentionTask<S, M, Q>
{
    pub fn new(
        storage: std::sync::Arc<S>,
        manager: std::sync::Arc<M>,
        engine: std::sync::Arc<Q>,
        shared_config: SharedConfig,
    ) -> Self {
        Self {
            storage,
            manager,
            engine,
            shared_config,
            grace_pending: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
        }
    }

    /// Returns a clone of the `grace_pending` map Arc, for use by admin API endpoints.
    pub fn grace_pending(&self) -> Arc<tokio::sync::RwLock<GracePendingMap>> {
        Arc::clone(&self.grace_pending)
    }

    /// Replace the internal `grace_pending` map with an externally provided
    /// `Arc`. This allows the admin HTTP layer to share the same map across
    /// `RetentionTask` restarts — pass the same handle to every spawned task so
    /// the admin endpoint always reflects the live grace state.
    pub fn with_grace_pending(mut self, handle: Arc<tokio::sync::RwLock<GracePendingMap>>) -> Self {
        self.grace_pending = handle;
        self
    }

    /// Run the retention loop until the shutdown signal is received.
    pub async fn run(self, mut shutdown_rx: watch::Receiver<ServerState>) {
        // Read initial config for startup logging.
        let cfg = self.shared_config.load();
        let active_count: usize = cfg
            .namespace_retention
            .values()
            .filter(|p| p.is_active())
            .count();
        let interval_secs = cfg.retention_check_interval_secs;
        drop(cfg);

        if active_count == 0 {
            tracing::info!("retention task: no active retention policies, exiting");
            return;
        }
        tracing::info!(
            interval_secs,
            namespaces = active_count,
            "retention task started"
        );

        let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
        // Skip the first immediate tick
        interval.tick().await;

        loop {
            tokio::select! {
                _ = shutdown_rx.changed() => break,
                _ = interval.tick() => {
                    self.sweep().await;
                }
            }
        }

        tracing::info!("retention task stopped");
    }

    /// One full sweep across all namespaces and topics.
    ///
    /// Uses `storage.list_topics()` to discover all persisted topics — this
    /// catches orphaned segments for topics that have been compacted out of the
    /// engine's in-memory map.
    async fn sweep(&self) {
        let sweep_start = std::time::Instant::now();
        let mut total_deleted = 0u64;

        // Load current config snapshot for this sweep.
        let namespace_retention = self.shared_config.load().namespace_retention.clone();

        // Discover all topics with data on disk, not just in-memory topics.
        let all_topics = match self.storage.list_topics().await {
            Ok(t) => t,
            Err(e) => {
                tracing::warn!(error = %e, "retention: failed to list topics from storage");
                return;
            }
        };

        for (ns, topic) in &all_topics {
            let Some(policy) = namespace_retention.get(ns) else {
                continue;
            };
            if !policy.is_active() {
                continue;
            }

            let seg_dir = self.storage.segments_dir(ns, topic);
            let segments = scan_segments(&seg_dir).await;
            if segments.is_empty() {
                continue;
            }

            let floor = self.manager.min_live_offset(ns, topic).await;
            let mut delete_before: Option<TopicOffset> = None;

            // 1. Age-based retention
            if let Some(max_age) = policy.max_age_secs {
                let cutoff = now_millis().saturating_sub(max_age * 1000);
                for (i, seg) in segments.iter().enumerate() {
                    if seg.mtime_millis < cutoff {
                        // This segment is expired. Delete up to the next segment's base.
                        let next_base = if i + 1 < segments.len() {
                            segments[i + 1].base_offset
                        } else {
                            // Don't delete the only remaining segment
                            continue;
                        };
                        delete_before = Some(match delete_before {
                            Some(current) => current.max(next_base),
                            None => next_base,
                        });
                    }
                }
            }

            // 2. Size-based retention
            if let Some(max_bytes) = policy.max_bytes {
                let total: u64 = segments.iter().map(|s| s.size_bytes).sum();
                if total > max_bytes {
                    let mut remaining = total;
                    for (i, seg) in segments.iter().enumerate() {
                        if remaining <= max_bytes || i + 1 >= segments.len() {
                            break;
                        }
                        remaining -= seg.size_bytes;
                        let next_base = segments[i + 1].base_offset;
                        delete_before = Some(match delete_before {
                            Some(current) => current.max(next_base),
                            None => next_base,
                        });
                    }
                }
            }

            // 3. Count-based retention
            if let Some(max_segs) = policy.max_segments
                && segments.len() > max_segs
            {
                let excess = segments.len() - max_segs;
                let next_base = segments[excess].base_offset;
                delete_before = Some(match delete_before {
                    Some(current) => current.max(next_base),
                    None => next_base,
                });
            }

            // Apply floor: never delete below min_live_offset.
            // Note: the floor is a raw offset, not segment-aligned. This is
            // correct because delete_segments uses `base_offset < before_offset`,
            // so a segment is only deleted if ALL its offsets are below the floor.
            if let (Some(delete), Some(floor_offset)) = (delete_before, floor)
                && delete > floor_offset
            {
                tracing::warn!(
                    namespace = %ns, topic = %topic,
                    wanted_delete_before = %delete, clamped_to = %floor_offset,
                    "retention clamped: slow subscriber is preventing segment cleanup"
                );
                delete_before = Some(floor_offset);
            }

            // Execute deletion
            if let Some(offset) = delete_before {
                // Compact in-memory first (crash-safe ordering)
                if let Err(e) = self.engine.compact(ns, topic, offset).await {
                    tracing::warn!(
                        namespace = %ns, topic = %topic,
                        error = %e, "retention: compact failed"
                    );
                    continue;
                }

                // Delete segment files
                match self
                    .storage
                    .delete_segments(ns, topic, Some(offset), None, floor)
                    .await
                {
                    Ok(count) if count > 0 => {
                        total_deleted += count as u64;
                        metrics::counter!(
                            "fafb_retention_segments_deleted_total",
                            "namespace" => ns.0.clone(),
                            "topic" => topic.0.clone()
                        )
                        .increment(count as u64);
                        tracing::info!(
                            namespace = %ns, topic = %topic,
                            count, before_offset = %offset,
                            "retention: deleted segments"
                        );
                    }
                    Ok(_) => {}
                    Err(e) => {
                        tracing::warn!(
                            namespace = %ns, topic = %topic,
                            error = %e, "retention: delete_segments failed"
                        );
                    }
                }
            }
        }

        // --- Topic inactivity GC ---
        self.sweep_inactive_topics(&all_topics).await;

        let elapsed = sweep_start.elapsed();
        metrics::histogram!("fafb_retention_sweep_duration_seconds").record(elapsed.as_secs_f64());
        if total_deleted > 0 {
            tracing::debug!(
                total_deleted,
                elapsed_ms = elapsed.as_millis(),
                "retention sweep complete"
            );
        }
    }

    /// Scan all topics for inactivity and reap those that exceed the configured
    /// threshold after a grace period has elapsed.
    async fn sweep_inactive_topics(&self, all_topics: &[(NamespaceId, TopicName)]) {
        let cfg = self.shared_config.load();
        let namespace_retention = &cfg.namespace_retention;
        let compiled_overrides = &cfg.compiled_inactive_overrides;

        let now_millis = crate::types::now_millis();
        let now_instant = std::time::Instant::now();
        let default_grace_secs: u64 = 86400;

        for (ns, topic) in all_topics {
            let Some(policy) = namespace_retention.get(ns) else {
                continue;
            };
            let Some(ns_inactive_secs) = policy.topic_inactive_secs else {
                continue;
            };

            let overrides = compiled_overrides
                .get(ns)
                .map(|v| v.as_slice())
                .unwrap_or(&[]);
            let threshold_secs =
                Self::resolve_inactive_threshold(topic, ns_inactive_secs, overrides);
            if threshold_secs == 0 {
                continue;
            }

            let engine_activity = self
                .engine
                .topic_last_activity(ns, topic)
                .await
                .unwrap_or(0);
            let manager_activity = self
                .manager
                .topic_last_subscribe(ns, topic)
                .await
                .unwrap_or(0);
            let effective = engine_activity.max(manager_activity);
            let inactive_millis = now_millis.saturating_sub(effective);
            let threshold_millis = threshold_secs * 1000;

            let key = (ns.clone(), topic.clone());
            if inactive_millis < threshold_millis {
                let mut grace = self.grace_pending.write().await;
                if grace.remove(&key).is_some() {
                    tracing::info!(
                        namespace = %ns, topic = %topic,
                        "topic inactivity GC: activity resumed, removed from grace"
                    );
                    metrics::gauge!(
                        "fafb_topic_reap_grace_pending",
                        "namespace" => ns.0.clone()
                    )
                    .decrement(1.0);
                }
                continue;
            }

            metrics::counter!(
                "fafb_topic_inactive_threshold_exceeded",
                "namespace" => ns.0.clone()
            )
            .increment(1);

            if self.manager.has_active_subscribers(ns, topic).await {
                metrics::counter!(
                    "fafb_topic_reap_skipped_active",
                    "namespace" => ns.0.clone()
                )
                .increment(1);
                continue;
            }
            if self.manager.has_inflight_messages(ns, topic).await {
                metrics::counter!(
                    "fafb_topic_reap_skipped_active",
                    "namespace" => ns.0.clone()
                )
                .increment(1);
                continue;
            }

            let grace_period_secs = policy
                .topic_reap_grace_period_secs
                .unwrap_or(default_grace_secs);

            let mut grace = self.grace_pending.write().await;
            match grace.entry(key) {
                std::collections::hash_map::Entry::Vacant(e) => {
                    e.insert(GracePendingEntry {
                        entered: now_instant,
                        grace_period_secs,
                        last_activity_millis: effective,
                        threshold_secs,
                    });
                    metrics::gauge!(
                        "fafb_topic_reap_grace_pending",
                        "namespace" => ns.0.clone()
                    )
                    .increment(1.0);
                    tracing::warn!(
                        namespace = %ns, topic = %topic,
                        inactive_secs = inactive_millis / 1000,
                        threshold_secs, grace_period_secs,
                        "topic entered reap grace period"
                    );
                    continue;
                }
                std::collections::hash_map::Entry::Occupied(e) => {
                    let entry = e.get();
                    if now_instant.duration_since(entry.entered).as_secs() < entry.grace_period_secs
                    {
                        continue;
                    }
                    e.remove();
                }
            }
            drop(grace);

            self.reap_topic(ns, topic, inactive_millis / 1000).await;
            metrics::gauge!(
                "fafb_topic_reap_grace_pending",
                "namespace" => ns.0.clone()
            )
            .decrement(1.0);
        }
    }

    /// Resolve the effective inactivity threshold for a topic by checking
    /// pre-compiled override patterns. First match wins; falls back to `default_secs`.
    fn resolve_inactive_threshold(
        topic: &TopicName,
        default_secs: u64,
        overrides: &[crate::config::CompiledInactiveOverride],
    ) -> u64 {
        resolve_inactive_threshold(topic, default_secs, overrides)
    }

    /// Execute the full reap sequence for a single topic: teardown delivery
    /// state, close storage writer, remove from engine, delete segment directory.
    async fn reap_topic(&self, ns: &NamespaceId, topic: &TopicName, inactive_secs: u64) {
        if let Err(e) = self.manager.teardown_topic(ns, topic).await {
            tracing::warn!(
                namespace = %ns, topic = %topic, error = %e,
                "reap: teardown_topic failed"
            );
            return;
        }
        if let Err(e) = self.storage.close_topic_writer(ns, topic).await {
            tracing::warn!(
                namespace = %ns, topic = %topic, error = %e,
                "reap: close_topic_writer failed"
            );
        }
        let RemoveTopicResult {
            bytes_freed,
            messages_freed,
        } = match self.engine.remove_topic(ns, topic).await {
            Ok(result) => result,
            Err(e) => {
                tracing::warn!(
                    namespace = %ns, topic = %topic, error = %e,
                    "reap: remove_topic failed"
                );
                RemoveTopicResult {
                    bytes_freed: 0,
                    messages_freed: 0,
                }
            }
        };
        if let Err(e) = self.storage.delete_topic_dir(ns, topic).await {
            tracing::warn!(
                namespace = %ns, topic = %topic, error = %e,
                "reap: delete_topic_dir failed"
            );
            return;
        }
        tracing::warn!(
            namespace = %ns, topic = %topic,
            inactive_days = inactive_secs / 86400,
            bytes_freed, messages_freed,
            "topic reaped"
        );
        metrics::counter!(
            "fafb_topic_reaped_total",
            "namespace" => ns.0.clone(),
            "topic" => topic.0.clone()
        )
        .increment(1);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::CompiledInactiveOverride;
    use crate::wildcard::TopicPattern;
    use std::io::Write;
    use tempfile::TempDir;

    fn create_segment(dir: &Path, base_offset: u64, content: &[u8]) -> std::path::PathBuf {
        let path = dir.join(format!("{base_offset:020}.seg"));
        let mut f = std::fs::File::create(&path).unwrap();
        f.write_all(content).unwrap();
        path
    }

    #[tokio::test]
    async fn scan_segments_empty_dir() {
        let tmp = TempDir::new().unwrap();
        let result = scan_segments(tmp.path()).await;
        assert!(result.is_empty());
    }

    #[tokio::test]
    async fn scan_segments_nonexistent_dir() {
        let result = scan_segments(Path::new("/nonexistent/path")).await;
        assert!(result.is_empty());
    }

    #[tokio::test]
    async fn scan_segments_parses_offsets() {
        let tmp = TempDir::new().unwrap();
        create_segment(tmp.path(), 0, b"data0");
        create_segment(tmp.path(), 1000, b"data1");
        create_segment(tmp.path(), 2000, b"data2");

        let segments = scan_segments(tmp.path()).await;
        assert_eq!(segments.len(), 3);
        assert_eq!(segments[0].base_offset, TopicOffset(0));
        assert_eq!(segments[1].base_offset, TopicOffset(1000));
        assert_eq!(segments[2].base_offset, TopicOffset(2000));
    }

    #[tokio::test]
    async fn scan_segments_returns_size() {
        let tmp = TempDir::new().unwrap();
        create_segment(tmp.path(), 0, &[0u8; 100]);

        let segments = scan_segments(tmp.path()).await;
        assert_eq!(segments.len(), 1);
        assert_eq!(segments[0].size_bytes, 100);
    }

    #[tokio::test]
    async fn scan_segments_ignores_non_seg_files() {
        let tmp = TempDir::new().unwrap();
        create_segment(tmp.path(), 0, b"data");
        // Create a non-.seg file
        std::fs::write(tmp.path().join("notes.txt"), b"not a segment").unwrap();

        let segments = scan_segments(tmp.path()).await;
        assert_eq!(segments.len(), 1);
    }

    #[tokio::test]
    async fn scan_segments_sorted_by_offset() {
        let tmp = TempDir::new().unwrap();
        // Create in reverse order
        create_segment(tmp.path(), 2000, b"c");
        create_segment(tmp.path(), 0, b"a");
        create_segment(tmp.path(), 1000, b"b");

        let segments = scan_segments(tmp.path()).await;
        assert_eq!(segments[0].base_offset, TopicOffset(0));
        assert_eq!(segments[1].base_offset, TopicOffset(1000));
        assert_eq!(segments[2].base_offset, TopicOffset(2000));
    }

    // --- resolve_inactive_threshold tests ---

    #[test]
    fn resolve_threshold_no_overrides_returns_default() {
        let topic = TopicName("any.topic".into());
        let result = resolve_inactive_threshold(&topic, 3600, &[]);
        assert_eq!(result, 3600);
    }

    #[test]
    fn resolve_threshold_first_match_wins() {
        let topic = TopicName("orders.us".into());
        let overrides = vec![
            CompiledInactiveOverride {
                pattern: TopicPattern::parse("orders.*").unwrap(),
                inactive_secs: 100,
            },
            CompiledInactiveOverride {
                pattern: TopicPattern::parse("orders.>").unwrap(),
                inactive_secs: 200,
            },
        ];
        let result = resolve_inactive_threshold(&topic, 3600, &overrides);
        assert_eq!(result, 100); // first match wins
    }

    #[test]
    fn resolve_threshold_exempt_returns_zero() {
        let topic = TopicName("system.health".into());
        let overrides = vec![CompiledInactiveOverride {
            pattern: TopicPattern::parse("system.>").unwrap(),
            inactive_secs: 0,
        }];
        let result = resolve_inactive_threshold(&topic, 3600, &overrides);
        assert_eq!(result, 0);
    }

    #[test]
    fn resolve_threshold_no_match_uses_default() {
        let topic = TopicName("events.click".into());
        let overrides = vec![CompiledInactiveOverride {
            pattern: TopicPattern::parse("orders.*").unwrap(),
            inactive_secs: 100,
        }];
        let result = resolve_inactive_threshold(&topic, 7200, &overrides);
        assert_eq!(result, 7200);
    }
}
