//! Data plane: builds batches, POSTs them, advances offset on ack.

use crate::manifest::BatchManifest;
use crate::state::SenderState;
use thiserror::Error;
use uuid::Uuid;

#[derive(Debug, Error)]
pub enum AckError {
    #[error("server's high_water_event_id {0} not in batch manifest")]
    HighWaterUnknown(Uuid),
    #[error("manifest empty — should not happen for a non-empty batch")]
    EmptyManifest,
}

/// Given the server-provided `high_water_event_id`, compute the
/// `SenderState` that should be persisted after this batch is acked.
/// Spec §3.8.3 — high_water is the last event_id in the batch in
/// submission order, regardless of accept/reject.
pub fn apply_ack(
    manifest: &BatchManifest,
    high_water_event_id: Uuid,
) -> Result<SenderState, AckError> {
    if manifest.is_empty() {
        return Err(AckError::EmptyManifest);
    }
    let entry = manifest
        .lookup(&high_water_event_id)
        .ok_or(AckError::HighWaterUnknown(high_water_event_id))?;
    Ok(SenderState {
        current_file: entry.current_file.clone(),
        byte_offset: entry.byte_range.end,
        last_acked_sequence: entry.provisional_sequence,
    })
}

use std::time::Duration;

/// Exponential backoff for retryable failures.
/// Spec §3.8.2 — ServerBusy (5xx): start 5s, cap 5min.
///                PermanentReject (426): start 5min, cap 1h.
#[derive(Clone, Copy, Debug)]
pub struct BackoffPolicy {
    pub initial: Duration,
    pub cap: Duration,
}

impl BackoffPolicy {
    pub fn server_busy() -> Self {
        BackoffPolicy {
            initial: Duration::from_secs(5),
            cap: Duration::from_secs(300),
        }
    }
    pub fn upgrade_required() -> Self {
        BackoffPolicy {
            initial: Duration::from_secs(300),
            cap: Duration::from_secs(3600),
        }
    }
    /// Returns the next sleep duration given the consecutive-failure count.
    /// `attempts == 0` returns `initial`. Doubles per attempt, clamped to `cap`.
    pub fn next_delay(&self, attempts: u32) -> Duration {
        let base = self.initial.as_secs();
        let cap = self.cap.as_secs();
        let mult = 1u64.checked_shl(attempts).unwrap_or(u64::MAX);
        let secs = base.saturating_mul(mult).min(cap);
        Duration::from_secs(secs)
    }
}

use crate::transport::{classify_send_error, classify_status, SendOutcome};
use crate::wire::{Envelope, EventsAccepted, EventsRequest};
use reqwest::Client;
use std::path::PathBuf;
use time::OffsetDateTime;

/// Inputs for one batch send (no loop).
pub struct BatchSendCtx<'a> {
    pub client: &'a Client,
    pub server_base_url: &'a str,
    pub host_id: &'a str,
    pub agent_version: &'a str,
    pub sender_version: &'a str,
    pub events: Vec<crate::wire::EventEntry>,
}

/// Outcome of `send_one_batch` — caller decides what to do with offset
/// advance and event emission.
#[derive(Debug)]
pub enum BatchOutcome {
    Accepted(EventsAccepted),
    PermanentReject { status: u16, body: String },
    ServerBusy { status: u16, body: String },
    TlsFailure(String),
    Network(String),
    ProtocolViolation(String),
}

pub async fn send_one_batch(ctx: BatchSendCtx<'_>) -> BatchOutcome {
    let req = EventsRequest {
        envelope: Envelope {
            schema_version: 1,
            batch_id: uuid::Uuid::now_v7(),
            host_id: ctx.host_id.to_string(),
            agent_version: ctx.agent_version.to_string(),
            sender_version: ctx.sender_version.to_string(),
            sent_at: OffsetDateTime::now_utc(),
        },
        events: ctx.events,
    };
    let url = format!("{}/v1/events", ctx.server_base_url.trim_end_matches('/'));
    let resp = match ctx.client.post(&url).json(&req).send().await {
        Ok(r) => r,
        Err(e) => match classify_send_error::<EventsAccepted>(e) {
            SendOutcome::TlsFailure(s) => return BatchOutcome::TlsFailure(s),
            SendOutcome::Network(s) => return BatchOutcome::Network(s),
            other => return BatchOutcome::ProtocolViolation(format!("{other:?}")),
        },
    };
    let status = resp.status().as_u16();
    let body_bytes = match resp.bytes().await {
        Ok(b) => b,
        Err(e) => {
            return BatchOutcome::ProtocolViolation(format!("body read: {e}"));
        }
    };
    let body_text = String::from_utf8_lossy(&body_bytes).to_string();
    let parsed: Option<EventsAccepted> = if (200..300).contains(&status) {
        serde_json::from_slice(&body_bytes).ok()
    } else {
        None
    };
    match classify_status(status, body_text.clone(), parsed) {
        SendOutcome::Ok2xx(r) => BatchOutcome::Accepted(r),
        SendOutcome::PermanentReject { status, body } => {
            BatchOutcome::PermanentReject { status, body }
        }
        SendOutcome::ServerBusy { status, body } => BatchOutcome::ServerBusy { status, body },
        SendOutcome::TlsFailure(s) => BatchOutcome::TlsFailure(s),
        SendOutcome::Network(s) => BatchOutcome::Network(s),
        SendOutcome::ProtocolViolation(s) => BatchOutcome::ProtocolViolation(s),
    }
}

// (silence unused-import warning when path types only used in tests)
#[allow(dead_code)]
fn _path_marker(_p: PathBuf) {}

use crate::batch_reader::read_next_batch;
use crate::config::SenderConfig;
use crate::dead_letter;
use crate::heartbeat::SharedStats;
use crate::state;
use sigil_core::event::{
    Event, Evidence, Severity, SourceKind, Subject, AGENT_VERSION, SCHEMA_VERSION,
};
use tokio_util::sync::CancellationToken;

/// Idle poll interval when no batch is available.
const IDLE_POLL: Duration = Duration::from_millis(500);

pub struct DataTaskCtx {
    pub client: Client,
    pub config: SenderConfig,
    pub host_id: String,
    pub agent_version: String,
    pub sender_version: String,
    pub stats: SharedStats,
    pub shutdown: CancellationToken,
}

/// Build a local diagnostic Event tagged with the sender's host_id.
/// Used for HostUnknown / TlsFailure / ProtocolViolation / per-event
/// rejections, all of which get appended to the dead-letter spool.
pub fn local_event(host_id: &str, evidence: Evidence) -> Event {
    Event {
        schema_version: SCHEMA_VERSION,
        event_id: uuid::Uuid::now_v7(),
        ts: OffsetDateTime::now_utc(),
        host_id: host_id.to_string(),
        agent_version: AGENT_VERSION.to_string(),
        severity: Severity::Warn,
        source: SourceKind::Agent,
        subject: Subject::Self_,
        evidence,
        target_id: None,
    }
}

/// Top-level data plane loop: read JSONL → POST /v1/events → apply_ack
/// → state::store → dead-letter rejections → update stats. Handles all
/// `BatchOutcome` variants per spec §3.8.2.
pub async fn run(ctx: DataTaskCtx) {
    let backoff_busy = BackoffPolicy::server_busy();
    let backoff_perm = BackoffPolicy::upgrade_required();
    let mut state = match state::load_or_empty(&ctx.config.offset_path) {
        Ok(s) => s,
        Err(e) => {
            tracing::warn!(error = ?e, "load_or_empty failed; starting empty");
            SenderState::empty()
        }
    };
    let mut consecutive_failures: u32 = 0;

    loop {
        if ctx.shutdown.is_cancelled() {
            break;
        }

        if state.current_file.is_empty() {
            if let Some(seg) = discover_first_segment(&ctx.config.events_dir) {
                state.current_file = seg;
            } else if !idle_sleep(&ctx.shutdown, IDLE_POLL).await {
                break;
            } else {
                continue;
            }
        }

        let read_result = read_next_batch(
            &ctx.config.events_dir,
            &state,
            ctx.config.max_batch_events,
            ctx.config.max_batch_bytes,
        );
        let (events, manifest) = match read_result {
            Ok(r) => r,
            Err(e) => {
                tracing::warn!(
                    error = ?e,
                    current_file = %state.current_file,
                    "read_next_batch failed"
                );
                if !idle_sleep(&ctx.shutdown, Duration::from_secs(1)).await {
                    break;
                }
                continue;
            }
        };

        if events.is_empty() {
            if let Some(next) = next_segment(&ctx.config.events_dir, &state) {
                state.current_file = next;
                state.byte_offset = 0;
                continue;
            }
            if !idle_sleep(&ctx.shutdown, IDLE_POLL).await {
                break;
            }
            continue;
        }

        let batch_size = events.len();
        let outcome = send_one_batch(BatchSendCtx {
            client: &ctx.client,
            server_base_url: &ctx.config.server_base_url,
            host_id: &ctx.host_id,
            agent_version: &ctx.agent_version,
            sender_version: &ctx.sender_version,
            events,
        })
        .await;

        match outcome {
            BatchOutcome::Accepted(accepted) => {
                consecutive_failures = 0;
                let new_state = match apply_ack(&manifest, accepted.high_water_event_id) {
                    Ok(s) => s,
                    Err(e) => {
                        tracing::error!(error = ?e, "apply_ack failed");
                        let evt = local_event(
                            &ctx.host_id,
                            Evidence::ServerProtocolViolation {
                                detail: format!("apply_ack: {e}"),
                            },
                        );
                        let _ = dead_letter::append(&ctx.config.dead_letter_dir, &evt);
                        continue;
                    }
                };
                for r in &accepted.rejected {
                    tracing::warn!(
                        host_id = %ctx.host_id,
                        reason = %r.reason,
                        event_id = %r.event_id,
                        "server rejected event; dead-lettering"
                    );
                    let evt = local_event(
                        &ctx.host_id,
                        Evidence::EventUnprocessableLocal {
                            original_event_id: r.event_id,
                            server_reason: r.reason.clone(),
                        },
                    );
                    if let Err(e) = dead_letter::append(&ctx.config.dead_letter_dir, &evt) {
                        tracing::warn!(
                            error = ?e,
                            event_id = %r.event_id,
                            "dead_letter::append failed"
                        );
                    }
                }
                if let Err(e) = state::store(&ctx.config.offset_path, &new_state) {
                    tracing::warn!(error = ?e, "state::store failed; will retry");
                    continue;
                }
                state = new_state;
                {
                    let mut s = ctx.stats.write();
                    s.last_server_ack_at = Some(OffsetDateTime::now_utc());
                    s.last_server_response_code = Some(200);
                    s.lag_events = s.lag_events.saturating_sub(batch_size as u64);
                }
            }
            BatchOutcome::PermanentReject { status, body } => {
                consecutive_failures += 1;
                tracing::warn!(
                    host_id = %ctx.host_id,
                    status,
                    count = batch_size,
                    "server rejected batch; dead-lettering"
                );
                let evt = local_event(
                    &ctx.host_id,
                    Evidence::ServerProtocolViolation {
                        detail: format!("batch rejected status={status}: {body}"),
                    },
                );
                let _ = dead_letter::append(&ctx.config.dead_letter_dir, &evt);
                {
                    let mut s = ctx.stats.write();
                    s.last_server_response_code = Some(status);
                }
                let delay = backoff_perm.next_delay(consecutive_failures.saturating_sub(1));
                if !idle_sleep(&ctx.shutdown, delay).await {
                    break;
                }
            }
            BatchOutcome::ServerBusy { status, .. } => {
                consecutive_failures += 1;
                {
                    let mut s = ctx.stats.write();
                    s.last_server_response_code = Some(status);
                }
                let delay = backoff_busy.next_delay(consecutive_failures.saturating_sub(1));
                if !idle_sleep(&ctx.shutdown, delay).await {
                    break;
                }
            }
            BatchOutcome::TlsFailure(reason) => {
                consecutive_failures += 1;
                let evt = local_event(&ctx.host_id, Evidence::TlsFailure { reason });
                let _ = dead_letter::append(&ctx.config.dead_letter_dir, &evt);
                let delay = backoff_busy.next_delay(consecutive_failures.saturating_sub(1));
                if !idle_sleep(&ctx.shutdown, delay).await {
                    break;
                }
            }
            BatchOutcome::Network(reason) => {
                consecutive_failures += 1;
                tracing::warn!(reason = %reason, "network failure; will retry");
                let delay = backoff_busy.next_delay(consecutive_failures.saturating_sub(1));
                if !idle_sleep(&ctx.shutdown, delay).await {
                    break;
                }
            }
            BatchOutcome::ProtocolViolation(detail) => {
                consecutive_failures += 1;
                let evt = local_event(&ctx.host_id, Evidence::ServerProtocolViolation { detail });
                let _ = dead_letter::append(&ctx.config.dead_letter_dir, &evt);
                let delay = backoff_busy.next_delay(consecutive_failures.saturating_sub(1));
                if !idle_sleep(&ctx.shutdown, delay).await {
                    break;
                }
            }
        }
    }
}

/// Sleep for `delay` unless cancelled. Returns `false` if cancelled (caller breaks).
async fn idle_sleep(cancel: &CancellationToken, delay: Duration) -> bool {
    tokio::select! {
        biased;
        _ = cancel.cancelled() => false,
        _ = tokio::time::sleep(delay) => true,
    }
}

/// Find the lexically smallest `events-*.jsonl` segment in `events_dir`.
/// Used on first run when no `current_file` has been persisted yet.
fn discover_first_segment(events_dir: &std::path::Path) -> Option<String> {
    let mut entries: Vec<String> = std::fs::read_dir(events_dir)
        .ok()?
        .filter_map(|e| e.ok())
        .map(|e| e.file_name().to_string_lossy().to_string())
        .filter(|n| n.starts_with("events-") && n.ends_with(".jsonl"))
        .collect();
    entries.sort();
    entries.into_iter().next()
}

/// Find the lexically next `events-*.jsonl` segment after the current one.
/// Returns `None` if `current_file` is the latest. Used to roll forward
/// after the current segment is fully drained.
fn next_segment(events_dir: &std::path::Path, state: &SenderState) -> Option<String> {
    let mut entries: Vec<String> = std::fs::read_dir(events_dir)
        .ok()?
        .filter_map(|e| e.ok())
        .map(|e| e.file_name().to_string_lossy().to_string())
        .filter(|n| n.starts_with("events-") && n.ends_with(".jsonl"))
        .collect();
    entries.sort();
    let cur = &state.current_file;
    entries.into_iter().find(|n| n > cur)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::manifest::{ByteRange, ManifestEntry};

    fn entry(id: u128, seq: u64, end: u64) -> ManifestEntry {
        ManifestEntry {
            event_id: Uuid::from_u128(id),
            byte_range: ByteRange {
                start: end - 100,
                end,
            },
            provisional_sequence: seq,
            current_file: "events-1.jsonl".into(),
        }
    }

    #[test]
    fn high_water_at_end_advances_to_last() {
        let mut m = BatchManifest::new();
        m.push(entry(1, 10, 100));
        m.push(entry(2, 11, 200));
        m.push(entry(3, 12, 300));
        let s = apply_ack(&m, Uuid::from_u128(3)).unwrap();
        assert_eq!(s.byte_offset, 300);
        assert_eq!(s.last_acked_sequence, 12);
    }

    #[test]
    fn high_water_in_middle_advances_to_middle() {
        let mut m = BatchManifest::new();
        m.push(entry(1, 10, 100));
        m.push(entry(2, 11, 200));
        m.push(entry(3, 12, 300));
        let s = apply_ack(&m, Uuid::from_u128(2)).unwrap();
        assert_eq!(s.byte_offset, 200);
        assert_eq!(s.last_acked_sequence, 11);
    }

    #[test]
    fn unknown_high_water_is_error() {
        let mut m = BatchManifest::new();
        m.push(entry(1, 10, 100));
        let err = apply_ack(&m, Uuid::from_u128(999)).unwrap_err();
        assert!(matches!(err, AckError::HighWaterUnknown(_)));
    }

    #[test]
    fn empty_manifest_is_error() {
        let m = BatchManifest::new();
        let err = apply_ack(&m, Uuid::from_u128(1)).unwrap_err();
        assert!(matches!(err, AckError::EmptyManifest));
    }

    #[test]
    fn server_busy_starts_at_5s() {
        let b = BackoffPolicy::server_busy();
        assert_eq!(b.next_delay(0).as_secs(), 5);
    }

    #[test]
    fn server_busy_doubles_then_caps_at_5min() {
        let b = BackoffPolicy::server_busy();
        assert_eq!(b.next_delay(1).as_secs(), 10);
        assert_eq!(b.next_delay(2).as_secs(), 20);
        assert_eq!(b.next_delay(20).as_secs(), 300); // capped
    }

    #[test]
    fn upgrade_required_starts_at_5min_caps_at_1h() {
        let b = BackoffPolicy::upgrade_required();
        assert_eq!(b.next_delay(0).as_secs(), 300);
        assert_eq!(b.next_delay(20).as_secs(), 3600);
    }
}
