#![forbid(unsafe_code)]

use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant};

static CAPTURE_ORIGIN: OnceLock<Instant> = OnceLock::new();
const DEFAULT_SOURCE_LAG_CAP_MS: u64 = 250;
const DEFAULT_SOURCE_LEAD_CAP_MS: u64 = 80;

fn origin() -> Instant {
    *CAPTURE_ORIGIN.get_or_init(Instant::now)
}

/// Return the shared live-capture timestamp for upstream camera/mic packets.
///
/// Inputs: none.
/// Outputs: microseconds elapsed since the relay child first stamped live media.
/// Why: camera and microphone capture pipelines run independently, so they need
/// one explicit common origin before the server can keep them on the same live
/// call timeline.
#[must_use]
pub fn capture_pts_us() -> u64 {
    origin().elapsed().as_micros().min(u64::MAX as u128) as u64
}

/// Measure how old one shared capture timestamp is right now.
///
/// Inputs: a packet timestamp previously produced by `capture_pts_us`.
/// Outputs: the elapsed age as a `Duration`.
/// Why: upstream freshness telemetry should use the same shared live clock as
/// packet timestamps so queue-age calculations stay honest.
#[must_use]
#[allow(dead_code)]
pub fn packet_age(pts_us: u64) -> Duration {
    Duration::from_micros(capture_pts_us().saturating_sub(pts_us))
}

/// Decide whether extra upstream timing instrumentation should be emitted.
///
/// Inputs: none.
/// Outputs: `true` when detailed capture/rebase timing logs are enabled.
/// Why: A/V sync work needs bursts of deep timing visibility without leaving
/// noisy logs on during normal live operation.
#[must_use]
pub fn upstream_timing_trace_enabled() -> bool {
    std::env::var("LESAVKA_UPSTREAM_TIMING_TRACE")
        .ok()
        .map(|value| {
            let trimmed = value.trim();
            !(trimmed.eq_ignore_ascii_case("0")
                || trimmed.eq_ignore_ascii_case("false")
                || trimmed.eq_ignore_ascii_case("no")
                || trimmed.eq_ignore_ascii_case("off"))
        })
        .unwrap_or(false)
}

/// Cap how far source-derived packet timestamps may trail the live capture clock.
///
/// Inputs: none.
/// Outputs: the maximum tolerated lag between a rebased source PTS and the
/// current capture clock.
/// Why: encoded appsink buffers can emerge well after the raw capture moment,
/// and trusting those delayed buffer PTS values without a guard can make the
/// server believe fresh audio/video packets are already hopelessly stale.
#[must_use]
pub fn upstream_source_lag_cap() -> Duration {
    std::env::var("LESAVKA_UPSTREAM_SOURCE_LAG_CAP_MS")
        .ok()
        .and_then(|raw| raw.trim().parse::<u64>().ok())
        .filter(|value| *value > 0)
        .map(Duration::from_millis)
        .unwrap_or_else(|| Duration::from_millis(DEFAULT_SOURCE_LAG_CAP_MS))
}

/// Cap how far source-derived packet timestamps may lead the live capture clock.
///
/// Inputs: none.
/// Outputs: the maximum tolerated future lead for source-based packet PTS.
/// Why: live sources can flush a burst of future-stamped buffers; if those
/// future timestamps escape, the server freezes media waiting for local backlog.
#[must_use]
pub fn upstream_source_lead_cap() -> Duration {
    std::env::var("LESAVKA_UPSTREAM_SOURCE_LEAD_CAP_MS")
        .ok()
        .and_then(|raw| raw.trim().parse::<u64>().ok())
        .filter(|value| *value > 0)
        .map(Duration::from_millis)
        .unwrap_or_else(|| Duration::from_millis(DEFAULT_SOURCE_LEAD_CAP_MS))
}

#[derive(Debug, Default)]
struct SourcePtsRebaserState {
    source_base_us: Option<u64>,
    capture_base_us: Option<u64>,
    last_packet_pts_us: Option<u64>,
}

/// Rebase source-buffer timestamps onto the shared client capture clock.
///
/// Inputs: optional source PTS values from one live capture pipeline.
/// Outputs: packet timestamps that share the same client clock origin across
/// camera and microphone while still advancing based on source timing rather
/// than late appsink pull time.
/// Why: camera and microphone encode paths can add different queue/encode
/// delays before appsink pull, so stamping at pull time bakes skew into the
/// packets before the server ever sees them.
#[derive(Debug, Default)]
pub struct SourcePtsRebaser {
    state: Mutex<SourcePtsRebaserState>,
}

/// Snapshot of one client-side timestamp rebasing decision.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RebasedSourcePts {
    pub packet_pts_us: u64,
    pub capture_now_us: u64,
    pub source_pts_us: Option<u64>,
    pub source_base_us: Option<u64>,
    pub capture_base_us: Option<u64>,
    pub used_source_pts: bool,
    pub lag_clamped: bool,
    pub lead_clamped: bool,
}

#[derive(Debug, Default)]
struct DurationPacedSourcePtsState {
    next_packet_pts_us: Option<u64>,
    last_packet_pts_us: Option<u64>,
}

/// Rebase encoded packet timing by anchoring once, then pacing by duration.
///
/// Inputs: optional source PTS from one encoded packet stream plus the packet's
/// declared duration and a freshness lag cap.
/// Outputs: packet timestamps on the shared client capture clock that advance
/// by actual media duration instead of trusting potentially stretched parser
/// PTS on every packet.
/// Why: encoded audio parsers can emit packet PTS values that do not track
/// real packet duration faithfully, which can make the server pace audio far
/// too slowly or quickly even when the underlying capture stream is healthy.
#[derive(Debug, Default)]
pub struct DurationPacedSourcePtsRebaser {
    anchor_rebaser: SourcePtsRebaser,
    state: Mutex<DurationPacedSourcePtsState>,
}

impl SourcePtsRebaser {
    /// Translate one source-buffer timestamp onto the shared capture clock.
    ///
    /// Inputs: the buffer PTS if available plus the minimum monotonic step.
    /// Outputs: a rebased packet timestamp and the values used to derive it.
    /// Why: source PTS should drive packet timing when available, but packets
    /// must still remain monotonic even if buffers repeat or arrive oddly.
    #[allow(dead_code)]
    #[must_use]
    pub fn rebase_or_now(&self, source_pts_us: Option<u64>, min_step_us: u64) -> RebasedSourcePts {
        self.rebase_with_lag_cap(source_pts_us, min_step_us, None)
    }

    /// Translate one source-buffer timestamp onto the shared capture clock
    /// while bounding how stale that source-derived timestamp may become.
    ///
    /// Inputs: optional source PTS, minimum monotonic step, and an optional
    /// maximum lag behind the current capture clock.
    /// Outputs: a rebased packet timestamp plus details about any lag clamp.
    /// Why: encoder/parser queues can batch source buffers, so a pure
    /// source-PTS timeline may fall far behind real packet availability and
    /// poison server-side freshness calculations.
    #[must_use]
    pub fn rebase_with_lag_cap(
        &self,
        source_pts_us: Option<u64>,
        min_step_us: u64,
        max_lag: Option<Duration>,
    ) -> RebasedSourcePts {
        let capture_now_us = capture_pts_us();
        let mut state = self
            .state
            .lock()
            .expect("source pts rebaser mutex poisoned");
        let mut packet_pts_us = capture_now_us;
        let mut used_source_pts = false;
        let mut lag_clamped = false;
        let mut lead_clamped = false;

        if let Some(source_pts_us) = source_pts_us {
            let source_base_us = *state.source_base_us.get_or_insert(source_pts_us);
            let capture_base_us = *state.capture_base_us.get_or_insert(capture_now_us);
            state.capture_base_us = Some(capture_base_us);
            packet_pts_us =
                capture_base_us.saturating_add(source_pts_us.saturating_sub(source_base_us));
            used_source_pts = true;
        }

        if used_source_pts && let Some(max_lag) = max_lag {
            let lag_floor_us =
                capture_now_us.saturating_sub(max_lag.as_micros().min(u64::MAX as u128) as u64);
            if packet_pts_us < lag_floor_us {
                packet_pts_us = lag_floor_us;
                lag_clamped = true;
            }
            let lead_ceiling_us =
                capture_now_us.saturating_add(
                    upstream_source_lead_cap().as_micros().min(u64::MAX as u128) as u64,
                );
            if packet_pts_us > lead_ceiling_us {
                packet_pts_us = lead_ceiling_us;
                lead_clamped = true;
            }
        }

        if let Some(last_packet_pts_us) = state.last_packet_pts_us
            && packet_pts_us <= last_packet_pts_us
        {
            packet_pts_us = last_packet_pts_us.saturating_add(min_step_us.max(1));
        }

        state.last_packet_pts_us = Some(packet_pts_us);
        RebasedSourcePts {
            packet_pts_us,
            capture_now_us,
            source_pts_us,
            source_base_us: state.source_base_us,
            capture_base_us: state.capture_base_us,
            used_source_pts,
            lag_clamped,
            lead_clamped,
        }
    }
}

impl DurationPacedSourcePtsRebaser {
    /// Rebase one encoded packet onto the shared capture clock.
    ///
    /// Inputs: optional packet PTS, the packet media duration in microseconds,
    /// and a freshness lag cap behind the live capture clock.
    /// Outputs: a rebased packet timestamp plus the values used to derive it.
    /// Why: once the first encoded packet is anchored, the safest pacing signal
    /// for compressed audio is its actual packet duration, with a live lag
    /// clamp to keep delayed batches from resurrecting stale timing.
    #[must_use]
    pub fn rebase_with_packet_duration(
        &self,
        source_pts_us: Option<u64>,
        packet_duration_us: u64,
        max_lag: Duration,
    ) -> RebasedSourcePts {
        let step_us = packet_duration_us.max(1);
        let mut rebased =
            self.anchor_rebaser
                .rebase_with_lag_cap(source_pts_us, step_us, Some(max_lag));
        let lag_floor_us = rebased
            .capture_now_us
            .saturating_sub(max_lag.as_micros().min(u64::MAX as u128) as u64);
        let mut state = self
            .state
            .lock()
            .expect("duration paced source pts rebaser mutex poisoned");
        let mut packet_pts_us = state.next_packet_pts_us.unwrap_or(rebased.packet_pts_us);
        if packet_pts_us < lag_floor_us {
            packet_pts_us = lag_floor_us;
            rebased.lag_clamped = true;
        }
        let lead_ceiling_us = rebased
            .capture_now_us
            .saturating_add(upstream_source_lead_cap().as_micros().min(u64::MAX as u128) as u64);
        if packet_pts_us > lead_ceiling_us {
            packet_pts_us = lead_ceiling_us;
            rebased.lead_clamped = true;
        }
        if let Some(last_packet_pts_us) = state.last_packet_pts_us
            && packet_pts_us <= last_packet_pts_us
        {
            packet_pts_us = last_packet_pts_us.saturating_add(1);
        }
        state.last_packet_pts_us = Some(packet_pts_us);
        state.next_packet_pts_us = Some(packet_pts_us.saturating_add(step_us));
        rebased.packet_pts_us = packet_pts_us;
        rebased
    }
}

#[cfg(test)]
#[path = "live_capture_clock/tests.rs"]
mod tests;