//! The left-right write/read plumbing around `EngineSnapshot`, the actor loop that owns it,
//! and the `EngineHandle` other tasks use to talk to that actor.
//!
//! Ticket 106 (giving API handlers a route to both the primary and profile actors) resolved to
//! option (b) from its own scope note: one combined `EngineHandle`, not two separate handle
//! types. Callers never pick a target explicitly -- `EngineHandle::tx` accepts every
//! `EngineMessage` variant, and `run()` below internally proxies the metric/alert-related ones
//! to the profile actor (`super::profile::ProfileMessage`) rather than mutating the primary's
//! own snapshot; `EngineHandle::profile_reader` similarly gives read access to the profile
//! actor's snapshot for the query variants it owns (routed in `api::EngineHandle::ask`). This
//! means `reload_engine_declarations`/`activate_profile`/`Profile::load_into_engine` needed no
//! changes at all to correctly split their remove-then-redeclare sequence across both actors --
//! they already just send `EngineMessage`s the same way they did before the split.

use crate::config::SettingsHandle;
use crate::consts::{INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS};
use crate::engine::index_key::IndexKey;
use crate::engine::message::EngineMessage;
use crate::engine::metric_primitive::{MetricPrimitive, Number};
use crate::engine::snapshot::EngineSnapshot;
use crate::engine::snapshot::params::log_stream_header_transition;
use crate::engine::timestamp::Timestamp;
use crate::engine::{ClearParamsResult, IngestedLine};
use left_right::Absorb;
use std::collections::BTreeMap;
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot, watch};

struct EngineOp(Box<dyn FnMut(&mut EngineSnapshot) + Send + 'static>);

impl Absorb<EngineOp> for EngineSnapshot {
    fn absorb_first(&mut self, operation: &mut EngineOp, _other: &Self) {
        (operation.0)(self);
    }

    fn drop_first(self: Box<Self>) {}

    fn sync_with(&mut self, first: &Self) {
        *self = first.clone();
    }
}

struct IndexWriter {
    handle: left_right::WriteHandle<EngineSnapshot, EngineOp>,
    /// Bumped and re-sent on every publish so the profile actor (`super::profile`) can wake on
    /// change rather than polling; the value itself carries no meaning beyond "something new is
    /// visible; go re-read what you need through your own `IndexReader`."
    version: u64,
    version_tx: watch::Sender<u64>,
}

impl IndexWriter {
    fn mutate(&mut self, op: impl FnMut(&mut EngineSnapshot) + Send + 'static) {
        self.handle.append(EngineOp(Box::new(op)));
        self.handle.publish();
        self.version += 1;
        let _ = self.version_tx.send(self.version);
    }

    fn peek(&mut self) -> Option<left_right::ReadGuard<'_, EngineSnapshot>> {
        self.handle.enter()
    }
}

pub struct IndexReadHandle(left_right::ReadHandle<EngineSnapshot>);

impl IndexReadHandle {
    pub(crate) fn enter(&self) -> Option<left_right::ReadGuard<'_, EngineSnapshot>> {
        self.0.enter()
    }
}

pub struct IndexReader {
    pub(crate) factory: left_right::ReadHandleFactory<EngineSnapshot>,
}

impl Clone for IndexReader {
    fn clone(&self) -> Self {
        IndexReader {
            factory: self.factory.clone(),
        }
    }
}

impl IndexReader {
    pub(crate) fn handle(&self) -> IndexReadHandle {
        IndexReadHandle(self.factory.handle())
    }
}

pub struct EngineHandle {
    pub tx: mpsc::Sender<EngineMessage>,
    pub reader: IndexReader,
    /// Read-only view of the profile actor's snapshot (`super::profile`), which owns metric
    /// declarations and bucket data (ticket 104) — `MetricDeclarations`/`BucketValues`/`Promql`
    /// queries read through this instead of `reader`. Not part of the write path; the profile
    /// actor's own `mpsc` channel (proxied through `EngineMessage`'s metric variants) is that.
    pub(crate) profile_reader: IndexReader,
}

impl Clone for EngineHandle {
    fn clone(&self) -> Self {
        EngineHandle {
            tx: self.tx.clone(),
            reader: self.reader.clone(),
            profile_reader: self.profile_reader.clone(),
        }
    }
}

impl EngineHandle {
    pub async fn barrier(&self) {
        let (respond_to, response) = oneshot::channel();
        if self
            .tx
            .send(EngineMessage::Barrier(respond_to))
            .await
            .is_ok()
        {
            let _ = response.await;
        }
    }
}

/// Builds the primary actor's sender/receiver/reader/writer/version-receiver, shared by `spawn`
/// and, in tests, by callers that want the raw receiver without an actor loop draining it.
/// Stops short of building `EngineHandle` itself, since that also needs a profile-actor reader
/// that doesn't exist until the caller spawns (or fakes, see `test_handle`) the profile actor.
fn new_channel_and_handle(
    channel_capacity: usize,
) -> (
    mpsc::Sender<EngineMessage>,
    mpsc::Receiver<EngineMessage>,
    IndexReader,
    IndexWriter,
    watch::Receiver<u64>,
) {
    let (tx, rx) = mpsc::channel(channel_capacity);
    let (write, read) = left_right::new::<EngineSnapshot, EngineOp>();
    let reader = IndexReader {
        factory: read.factory(),
    };
    let (version_tx, version_rx) = watch::channel(0u64);
    (
        tx,
        rx,
        reader,
        IndexWriter {
            handle: write,
            version: 0,
            version_tx,
        },
        version_rx,
    )
}

pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle {
    let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity);
    let profile = crate::engine::profile::spawn(reader.clone(), version_rx, settings);
    let handle = EngineHandle {
        tx,
        reader,
        profile_reader: profile.reader.clone(),
    };
    tokio::spawn(run(rx, write, profile));
    handle
}

#[cfg(test)]
pub(crate) fn test_handle(
    channel_capacity: usize,
) -> (EngineHandle, mpsc::Receiver<EngineMessage>) {
    let (tx, rx, reader, _write, _version_rx) = new_channel_and_handle(channel_capacity);
    let handle = EngineHandle {
        tx,
        reader,
        // No actor loop is spawned here at all, so nothing ever publishes -- an inert reader
        // over an always-empty snapshot is all a query through this handle could see anyway.
        profile_reader: crate::engine::profile::inert_reader(),
    };
    (handle, rx)
}

/// Like `spawn`, but wires the profile actor's test-only observer instead of the normal
/// fire-and-forget one, so a test can assert the primary->profile wake/read loop actually ran.
#[cfg(test)]
pub(crate) fn spawn_with_profile_observer(
    settings: SettingsHandle,
    channel_capacity: usize,
) -> (EngineHandle, tokio::sync::mpsc::UnboundedReceiver<u64>) {
    let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity);
    let (profile, observed) =
        crate::engine::profile::spawn_observed(reader.clone(), version_rx, settings);
    let handle = EngineHandle {
        tx,
        reader,
        profile_reader: profile.reader.clone(),
    };
    tokio::spawn(run(rx, write, profile));
    (handle, observed)
}

async fn run(
    mut rx: mpsc::Receiver<EngineMessage>,
    mut write: IndexWriter,
    profile: crate::engine::profile::ProfileHandle,
) {
    let mut max_logs: Option<usize> = None;
    let mut metric_sample_sequence: u64 = 0;
    let mut log_stream_header: Option<crate::engine::logfmt::LogStreamHeader> = None;
    let mut ingest_lag_samples: Vec<(Timestamp, i64)> = Vec::new();

    while let Some(message) = rx.recv().await {
        match message {
            EngineMessage::Ingest(line) => {
                let IngestedLine {
                    sequence,
                    timestamp,
                    text,
                } = line;
                let processed_at = Timestamp::now();
                record_ingest_lag(
                    &mut ingest_lag_samples,
                    processed_at,
                    (processed_at.as_millis() - timestamp.as_millis()).max(0),
                );
                let ingest_lag_high_watermark_ms =
                    ingest_lag_high_watermark_ms(&ingest_lag_samples, processed_at);

                let is_new_header = crate::engine::logfmt::LogStreamHeader::detect(&text).is_some();
                log_stream_header = log_stream_header_transition(&log_stream_header, &text);
                let row_header = if is_new_header {
                    None
                } else {
                    log_stream_header.clone()
                };

                let text_for_op = Arc::clone(&text);
                write.mutate(move |snap: &mut EngineSnapshot| {
                    // Primary/param extraction only (ticket 101) -- metric computation
                    // (ticket 104) and alert evaluation (ticket 105) now happen on the
                    // profile actor, which wakes on this publish and reads the
                    // newly-visible line itself.
                    snap.ingest_params(sequence, timestamp, &text_for_op, row_header.as_ref());
                    snap.update_sort_helper_indexes_on_ingest(sequence, timestamp, &text_for_op);
                    snap.purge_stale_sort_helper_indexes();
                    snap.evict_excess_logs(max_logs);
                    snap.ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms;
                });
            }
            EngineMessage::IngestMetricSample(sample) => {
                let labels: BTreeMap<String, MetricPrimitive> = sample
                    .labels
                    .into_iter()
                    .map(|(name, value)| (name, MetricPrimitive::String(value)))
                    .collect();
                let metric_name = sample.metric_name;
                let sequence = metric_sample_sequence;
                let timestamp = sample.timestamp;
                let value = MetricPrimitive::Number(Number::Float(sample.value));
                write.mutate(move |snap: &mut EngineSnapshot| {
                    snap.push_bucket_value(
                        &metric_name,
                        labels.clone(),
                        sequence,
                        timestamp,
                        value.clone(),
                    );
                });
                metric_sample_sequence += 1;
            }
            EngineMessage::SetMaxLogs(new_max) => {
                max_logs = new_max;
                write.mutate(move |snap: &mut EngineSnapshot| {
                    snap.evict_excess_logs(new_max);
                });
            }
            EngineMessage::DeclareParam {
                request,
                respond_to,
            } => {
                let result = write
                    .peek()
                    .map(|guard| guard.validate_param_declaration(&request))
                    .unwrap_or_else(|| Err("engine is shutting down".to_string()));
                let succeeded = result.is_ok();
                let name = request.name.clone();
                let request_for_op = request.clone();
                write.mutate(move |snap: &mut EngineSnapshot| {
                    snap.sort_helper_indexes.remove(&name);
                    if succeeded {
                        snap.insert_param_declaration(request_for_op.clone());
                    }
                });
                let _ = respond_to.send(result);
            }
            EngineMessage::RemoveParamDeclaration { name, respond_to } => {
                let existed = write.peek().is_some_and(|guard| {
                    guard
                        .index
                        .contains_key(&IndexKey::ParamDeclaration { name: name.clone() })
                });
                let name_for_op = name.clone();
                write.mutate(move |snap: &mut EngineSnapshot| {
                    snap.index.remove(&IndexKey::ParamDeclaration {
                        name: name_for_op.clone(),
                    });
                    snap.sort_helper_indexes.remove(&name_for_op);
                    snap.param_regex_cache.remove(&name_for_op);
                });
                let _ = respond_to.send(existed);
            }
            EngineMessage::ClearParams { respond_to } => {
                let mut scratch = EngineSnapshot::from_index(
                    write
                        .peek()
                        .map(|guard| guard.index.clone())
                        .unwrap_or_default(),
                );
                let removed = scratch.clear_all_param_declarations();
                let (reextracted, logs_considered) = scratch.reextract_params_from_retained_logs();
                write.mutate(move |snap: &mut EngineSnapshot| {
                    snap.clear_all_param_declarations();
                    snap.reextract_params_from_retained_logs();
                    snap.sort_helper_indexes.clear();
                });
                let _ = respond_to.send(ClearParamsResult {
                    removed,
                    reextracted,
                    logs_considered,
                });
            }
            EngineMessage::DeclareMetric {
                request,
                respond_to,
            } => {
                // Metric declarations live on the profile actor (ticket 104), not here --
                // proxy the message rather than mutating the primary's snapshot.
                let _ = profile
                    .tx
                    .send(crate::engine::profile::ProfileMessage::DeclareMetric {
                        request,
                        respond_to,
                    })
                    .await;
            }
            EngineMessage::UpdateMetric {
                name,
                request,
                respond_to,
            } => {
                let _ = profile
                    .tx
                    .send(crate::engine::profile::ProfileMessage::UpdateMetric {
                        name,
                        request,
                        respond_to,
                    })
                    .await;
            }
            EngineMessage::RemoveMetricDeclaration { name, respond_to } => {
                let _ = profile
                    .tx
                    .send(
                        crate::engine::profile::ProfileMessage::RemoveMetricDeclaration {
                            name,
                            respond_to,
                        },
                    )
                    .await;
            }
            EngineMessage::EnsureSortHelperIndex {
                param_name,
                respond_to,
            } => {
                write.mutate(move |snap: &mut EngineSnapshot| {
                    snap.ensure_sort_helper_index_built(&param_name);
                });
                let _ = respond_to.send(());
            }
            EngineMessage::Barrier(respond_to) => {
                let _ = respond_to.send(());
            }
        }
    }
}

pub(crate) fn record_ingest_lag(samples: &mut Vec<(Timestamp, i64)>, now: Timestamp, lag_ms: i64) {
    samples.retain(|(recorded_at, _)| {
        now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS
    });
    if samples.len() < INGEST_LAG_WATERMARK_MAX_ENTRIES {
        samples.push((now, lag_ms));
        return;
    }
    let Some((min_index, &(_, min_lag))) =
        samples.iter().enumerate().min_by_key(|(_, (_, lag))| *lag)
    else {
        return;
    };
    if lag_ms > min_lag {
        samples[min_index] = (now, lag_ms);
    }
}

pub(crate) fn ingest_lag_high_watermark_ms(samples: &[(Timestamp, i64)], now: Timestamp) -> i64 {
    samples
        .iter()
        .filter(|(recorded_at, _)| {
            now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS
        })
        .map(|(_, lag)| *lag)
        .max()
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::Settings;
    use std::sync::Mutex;
    use std::time::Duration;

    /// Proves ticket 103's wake/read loop end-to-end: the profile actor wakes on the primary's
    /// publish signal and reads the newly-visible line through its own primary `IndexReader`,
    /// without the primary ever being told the profile actor exists.
    #[tokio::test]
    async fn profile_actor_observes_newly_ingested_lines_via_the_primary_watch_signal() {
        let settings: SettingsHandle = Arc::new(Mutex::new(Settings::default()));
        let (engine, mut observed) = spawn_with_profile_observer(settings, 8);

        engine
            .tx
            .send(EngineMessage::Ingest(IngestedLine {
                sequence: 0,
                timestamp: Timestamp::from_millis(0),
                text: "hello".into(),
            }))
            .await
            .unwrap();

        let observed_sequence = tokio::time::timeout(Duration::from_secs(2), observed.recv())
            .await
            .expect("timed out waiting for the profile actor to observe the new line")
            .expect("profile actor's observer channel closed");
        assert_eq!(observed_sequence, 0);
    }
}