plushie-renderer-lib 0.7.1

//! Rate-limited event emission with coalescing.
//!
//! Buffers high-frequency events (mouse moves, scroll, animation frames)
//! and emits them at a configurable rate. Non-coalescable events (clicks,
//! key presses) flush the buffer immediately before emitting.
//!
//! The host controls rates via three mechanisms (highest priority first):
//! 1. Per-widget `event_rate` prop
//! 2. Per-subscription `max_rate` field on Subscribe
//! 3. Global `default_event_rate` in Settings

use std::collections::HashMap;
use std::sync::Arc;

use parking_lot::Mutex;

use iced::time::{Duration, Instant};

use iced::Task;

use plushie_widget_sdk::protocol::{CoalesceHint, OutgoingEvent};
use plushie_widget_sdk::runtime::Message;

use crate::emitters::{EventSink, SinkMutex};

// ---------------------------------------------------------------------------
// Platform-aware sleep
// ---------------------------------------------------------------------------

#[cfg(not(target_arch = "wasm32"))]
async fn platform_sleep(duration: Duration) {
    tokio::time::sleep(duration).await;
}

#[cfg(target_arch = "wasm32")]
async fn platform_sleep(duration: Duration) {
    wasmtimer::tokio::sleep(duration).await;
}

// ---------------------------------------------------------------------------
// Capacity cap
// ---------------------------------------------------------------------------

/// Hard cap on the pending coalesce map. When exceeded the map is
/// fully flushed and a diagnostic emitted. Defensive only; typical
/// subscription-tag counts are well below this.
const PENDING_CAP: usize = 4096;

// ---------------------------------------------------------------------------
// Coalesce key
// ---------------------------------------------------------------------------

/// Identifies a stream of events that can be coalesced together.
#[derive(Clone, Debug, Hash, Eq, PartialEq, Ord, PartialOrd)]
pub enum CoalesceKey {
    /// Subscription event keyed by entry tag (e.g. "on_pointer_move" or
    /// "on_pointer_move:main" for window-scoped subscriptions).
    Subscription(String),
    /// Widget event keyed by (widget_id, event_family).
    Widget(String, String),
}

// ---------------------------------------------------------------------------
// Pending event buffer
// ---------------------------------------------------------------------------

enum PendingEvent {
    /// Latest-value-wins: only the most recent event is kept.
    Replace(OutgoingEvent),
    /// Named-field accumulation: the listed fields in `data` are summed
    /// across arrivals. All other fields keep the latest event's values.
    Accumulate {
        base: OutgoingEvent,
        fields: Vec<String>,
        totals: HashMap<String, f64>,
    },
}

impl PendingEvent {
    fn from_hint(event: OutgoingEvent, hint: &CoalesceHint) -> Self {
        match hint {
            CoalesceHint::Replace => PendingEvent::Replace(event),
            CoalesceHint::Accumulate(fields) => {
                let mut totals = HashMap::new();
                if let Some(value) = &event.value {
                    for field in fields {
                        if let Some(val) = value.get(field).and_then(|v| v.as_f64()) {
                            totals.insert(field.clone(), val);
                        }
                    }
                }
                PendingEvent::Accumulate {
                    base: event,
                    fields: fields.clone(),
                    totals,
                }
            }
        }
    }

    fn merge(&mut self, event: OutgoingEvent) {
        match self {
            PendingEvent::Replace(existing) => *existing = event,
            PendingEvent::Accumulate {
                base,
                fields,
                totals,
            } => {
                if let Some(value) = &event.value {
                    for field in fields.iter() {
                        if let Some(val) = value.get(field).and_then(|v| v.as_f64()) {
                            *totals.entry(field.clone()).or_insert(0.0) += val;
                        }
                    }
                }
                *base = event;
            }
        }
    }

    fn into_event(self) -> OutgoingEvent {
        match self {
            PendingEvent::Replace(ev) => ev,
            PendingEvent::Accumulate {
                mut base, totals, ..
            } => {
                // Patch accumulated totals back into the event's value.
                if let Some(ref mut value) = base.value
                    && let Some(obj) = value.as_object_mut()
                {
                    for (field, total) in totals {
                        obj.insert(field, serde_json::json!(total));
                    }
                }
                base
            }
        }
    }
}

// ---------------------------------------------------------------------------
// EventEmitter
// ---------------------------------------------------------------------------

/// Rate-limited event emission with coalescing.
///
/// Sits between the iced message handlers and the wire protocol. Events
/// classified as coalescable are buffered and emitted at a controlled
/// rate; non-coalescable events flush the buffer and emit immediately.
pub struct EventEmitter {
    /// The output sink for emitted events.
    sink: Arc<SinkMutex>,
    /// Pending coalescable events, keyed by coalesce key.
    pending: HashMap<CoalesceKey, PendingEvent>,
    /// Timestamp of last emission per coalesce key.
    last_emits: HashMap<CoalesceKey, Instant>,
    /// Whether a `Message::FlushCoalesce` timer task is outstanding.
    flush_scheduled: bool,
    /// Global default rate from Settings. None = no limit.
    default_rate: Option<u32>,
    /// Per-subscription rates from Subscribe max_rate.
    subscription_rates: HashMap<String, u32>,
    /// Per-widget rates from event_rate prop.
    widget_rates: HashMap<String, u32>,
    /// Batch-suppression state. When the depth is > 0, outgoing
    /// events are buffered here in order; they flush when the
    /// outermost batch closes. Protected by a Mutex so the sink
    /// emit paths (which take `&self`) can still buffer safely.
    batch: Arc<Mutex<BatchState>>,
}

#[derive(Default)]
struct BatchState {
    depth: u32,
    buffer: Vec<OutgoingEvent>,
}

impl EventEmitter {
    /// Create a new EventEmitter that writes to the given sink.
    pub fn new(sink: Arc<SinkMutex>) -> Self {
        Self {
            sink,
            pending: HashMap::new(),
            last_emits: HashMap::new(),
            flush_scheduled: false,
            default_rate: None,
            subscription_rates: HashMap::new(),
            widget_rates: HashMap::new(),
            batch: Arc::new(Mutex::new(BatchState::default())),
        }
    }

    /// Begin an atomic batch: outgoing events are buffered until
    /// [`end_batch`](Self::end_batch) is called at the matching
    /// depth. Nested calls are counted so callers don't have to
    /// coordinate.
    pub fn begin_batch(&self) {
        let mut state = self.batch.lock();
        state.depth = state.depth.saturating_add(1);
    }

    /// End an atomic batch. When the outermost batch closes, all
    /// buffered events are emitted through the sink in order.
    pub fn end_batch(&self) {
        let buffered = {
            let mut state = self.batch.lock();
            if state.depth == 0 {
                log::warn!("event emitter batch ended without a matching begin");
                debug_assert!(state.depth > 0, "event emitter batch depth underflow");
                return;
            }
            state.depth -= 1;
            if state.depth == 0 {
                std::mem::take(&mut state.buffer)
            } else {
                Vec::new()
            }
        };
        if buffered.is_empty() {
            return;
        }
        // Hold the sink lock for the whole batch so the writes appear
        // contiguous on the wire and so we only pay one buffer flush.
        let mut guard = self.sink.lock();
        for event in buffered {
            if let Err(e) = guard.emit_event(event) {
                log::error!("event sink write error: {e}");
            }
        }
        if let Err(e) = guard.flush_output() {
            log::error!("event sink flush error: {e}");
        }
    }

    /// Get a clone of the sink Arc for passing to async callbacks.
    pub fn sink(&self) -> Arc<SinkMutex> {
        self.sink.clone()
    }

    /// Set the global default rate from Settings.
    pub fn set_default_rate(&mut self, rate: Option<u32>) {
        self.default_rate = rate;
    }

    /// Set (or update) the rate for a subscription kind.
    pub fn set_subscription_rate(&mut self, kind: &str, rate: u32) {
        self.subscription_rates.insert(kind.to_string(), rate);
    }

    /// Remove rate tracking for a subscription kind. Also drops the
    /// matching `last_emits` entry so the timestamp map doesn't
    /// accumulate stale keys for unsubscribed streams.
    pub fn remove_subscription_rate(&mut self, kind: &str) {
        self.subscription_rates.remove(kind);
        self.last_emits
            .remove(&CoalesceKey::Subscription(kind.to_string()));
    }

    /// Set the rate for a specific widget (from `event_rate` prop).
    pub fn set_widget_rate(&mut self, widget_id: &str, rate: u32) {
        self.widget_rates.insert(widget_id.to_string(), rate);
    }

    /// Clear all widget rates (called on Snapshot, tree replaced). Also
    /// drops widget-keyed `last_emits` entries so timestamps for nodes
    /// that didn't survive the snapshot don't linger.
    pub fn clear_widget_rates(&mut self) {
        self.widget_rates.clear();
        self.last_emits
            .retain(|key, _| !matches!(key, CoalesceKey::Widget(_, _)));
    }

    /// Check whether a widget rate is already cached.
    pub fn has_widget_rate(&self, widget_id: &str) -> bool {
        self.widget_rates.contains_key(widget_id)
    }

    /// Iterate over the subscription rate keys.
    pub fn subscription_rate_keys(&self) -> impl Iterator<Item = &str> {
        self.subscription_rates.keys().map(|s| s.as_str())
    }

    #[cfg(test)]
    pub(crate) fn subscription_rate_for(&self, tag: &str) -> Option<u32> {
        self.subscription_rates.get(tag).copied()
    }

    /// Resolve the effective rate for a given key, following the
    /// priority hierarchy: widget > subscription > global default.
    fn effective_rate(&self, key: &CoalesceKey) -> Option<u32> {
        match key {
            CoalesceKey::Widget(widget_id, _family) => {
                if let Some(&rate) = self.widget_rates.get(widget_id) {
                    return Some(rate);
                }
                self.default_rate
            }
            CoalesceKey::Subscription(tag) => {
                if let Some(&rate) = self.subscription_rates.get(tag) {
                    return Some(rate);
                }
                self.default_rate
            }
        }
    }

    /// Emit a coalescable event, buffering it if the rate limit has
    /// not elapsed. The coalescing strategy is read from the event's
    /// [`CoalesceHint`]. Returns a Task if a flush timer needs scheduling.
    pub fn coalesce(&mut self, key: CoalesceKey, mut event: OutgoingEvent) -> Task<Message> {
        // Take the hint out of the event: it's consumed by the emitter
        // and not needed downstream (not serialized to the wire).
        let hint = match event.take_coalesce() {
            Some(h) => h,
            None => {
                // No hint: treat as non-coalescable (immediate delivery).
                return self.emit_immediate(event);
            }
        };

        let rate = self.effective_rate(&key);

        // Zero rate = muted, silently drop.
        if rate == Some(0) {
            return Task::none();
        }

        // No rate limit = emit immediately.
        let Some(rate) = rate else {
            let flush_task = self.flush_key(&key);
            return Task::batch([flush_task, self.do_emit(event)]);
        };

        let min_interval = Duration::from_secs_f64(1.0 / rate as f64);
        let now = Instant::now();

        let can_emit_now = self
            .last_emits
            .get(&key)
            .map(|last| now.duration_since(*last) >= min_interval)
            .unwrap_or(true);

        if can_emit_now {
            self.pending.remove(&key);
            self.last_emits.insert(key, now);
            return self.do_emit(event);
        }

        // Buffer the event.
        let buffer_task = self.buffer_event(&key, event, &hint);

        // Schedule a flush timer if one isn't already running.
        if !self.flush_scheduled {
            self.flush_scheduled = true;
            let remaining = self
                .last_emits
                .get(&key)
                .map(|last| min_interval.saturating_sub(now.duration_since(*last)))
                .unwrap_or(min_interval);
            return Task::batch([
                buffer_task,
                Task::perform(
                    async move {
                        platform_sleep(remaining).await;
                    },
                    |_| Message::FlushCoalesce,
                ),
            ]);
        }

        buffer_task
    }

    /// Emit a non-coalescable event immediately, flushing pending
    /// events first to preserve ordering.
    pub fn emit_immediate(&mut self, event: OutgoingEvent) -> Task<Message> {
        let flush_task = self.flush_all();
        Task::batch([flush_task, self.do_emit(event)])
    }

    /// Flush all pending events. Called by the `Message::FlushCoalesce`
    /// handler.
    pub fn flush(&mut self) -> Task<Message> {
        self.flush_scheduled = false;
        self.flush_all()
    }

    /// Flush pending events for a specific key.
    ///
    /// Returns any tasks produced by the underlying writes. On a
    /// broken pipe `do_emit` yields `iced::exit()`; propagating the
    /// task is what keeps the renderer from wedging when a
    /// high-frequency coalesced event flushes after the host has
    /// disconnected.
    pub fn flush_key(&mut self, key: &CoalesceKey) -> Task<Message> {
        if let Some(pending) = self.pending.remove(key) {
            let now = Instant::now();
            self.last_emits.insert(key.clone(), now);
            return self.do_emit(pending.into_event());
        }
        Task::none()
    }

    /// Flush all pending events (internal).
    fn flush_all(&mut self) -> Task<Message> {
        // `drain()` moves ownership of each `(key, PendingEvent)` pair
        // out of the map without cloning keys - key clones would copy
        // two Strings per entry under `CoalesceKey::Widget`, which for
        // a busy app adds up on every FlushCoalesce tick.
        //
        // Sort by `CoalesceKey` before emitting so flush order is
        // deterministic across runs. `HashMap::drain` returns entries
        // in an unspecified, randomised order; tests and replay tools
        // need a stable sequence to reproduce host-visible event
        // streams.
        let mut drained: Vec<_> = self.pending.drain().collect();
        drained.sort_by(|(a, _), (b, _)| a.cmp(b));
        let now = Instant::now();
        let mut tasks: Vec<Task<Message>> = Vec::new();
        for (key, pending) in drained {
            self.last_emits.insert(key, now);
            tasks.push(self.do_emit_no_flush(pending.into_event()));
        }
        if tasks.is_empty() {
            return Task::none();
        }
        // One flush for the whole drained batch: the BufWriter inside
        // the sink has been accumulating bytes for every event in the
        // loop, and we want a single syscall to push them to the host.
        tasks.push(self.flush_output());
        Task::batch(tasks)
    }

    /// Buffer an event under the given key.
    ///
    /// If the existing entry uses a different strategy (e.g. Replace vs
    /// Accumulate), the old entry is flushed first and a fresh buffer is
    /// started. This handles the edge case where a widget changes
    /// its coalesce hint between events for the same key.
    ///
    /// The pending map is hard-capped at [`PENDING_CAP`] entries to
    /// prevent unbounded growth under pathological
    /// (subscription-tag * widgets) combinations. When the cap is
    /// exceeded we flush the entire pending map and log a diagnostic
    /// with code `emitter_coalesce_cap_exceeded`.
    fn buffer_event(
        &mut self,
        key: &CoalesceKey,
        event: OutgoingEvent,
        hint: &CoalesceHint,
    ) -> Task<Message> {
        let mut tasks: Vec<Task<Message>> = Vec::new();
        if let Some(existing) = self.pending.get_mut(key) {
            // Check for strategy mismatch. Replace-vs-Replace is always compatible.
            // Accumulate-vs-Accumulate is only compatible when the tracked field
            // set is identical: merging into a buffer that tracks different
            // fields would silently miscount totals.
            let compatible = match (&*existing, hint) {
                (PendingEvent::Replace(_), CoalesceHint::Replace) => true,
                (
                    PendingEvent::Accumulate {
                        fields: existing_fields,
                        ..
                    },
                    CoalesceHint::Accumulate(new_fields),
                ) => existing_fields == new_fields,
                _ => false,
            };
            if compatible {
                existing.merge(event);
                return Task::none();
            }
            // Strategy changed (or Accumulate field list changed); flush the old
            // entry and start fresh.
            tasks.push(self.flush_key(key));
        }
        if self.pending.len() >= PENDING_CAP {
            // Typed diagnostic emitted through Display so the log line
            // stays consistent with other typed sites. We deliberately
            // do not route this through the outgoing event sink: the
            // emitter fires this branch exactly when the sink is under
            // pressure, so adding another event would feed the jam we
            // are trying to drain.
            plushie_widget_sdk::diagnostics::warn(
                plushie_core::Diagnostic::EmitterCoalesceCapExceeded { cap: PENDING_CAP },
            );
            tasks.push(self.flush_all());
        }
        self.pending
            .insert(key.clone(), PendingEvent::from_hint(event, hint));
        if tasks.is_empty() {
            Task::none()
        } else {
            Task::batch(tasks)
        }
    }

    /// Emit an event through the sink, returning a Result.
    ///
    /// Used by methods that return `io::Result` (e.g. event handlers
    /// in events.rs, apply.rs).
    ///
    /// These callers wait on the result before continuing (e.g. they
    /// emit a `session_error` and exit on write failure), so this
    /// path flushes the sink before returning instead of leaving the
    /// event in the BufWriter for the next coalesce-flush boundary.
    pub fn emit_event(&self, event: OutgoingEvent) -> std::io::Result<()> {
        self.with_sink(|sink| {
            sink.emit_event(event)?;
            sink.flush_output()
        })
    }

    /// Write an event directly to the sink, bypassing rate limiting.
    ///
    /// Used for subscription events and system events that don't
    /// participate in widget-level coalescing. Returns Task::none()
    /// on success, iced::exit() on broken pipe.
    pub fn emit_direct(&self, event: OutgoingEvent) -> Task<Message> {
        self.do_emit(event)
    }

    /// Emit an effect response through the sink.
    pub fn emit_effect_response(
        &self,
        response: plushie_widget_sdk::protocol::EffectResponse,
    ) -> std::io::Result<()> {
        self.with_sink(|sink| sink.emit_effect_response(response))
    }

    /// Emit a query response through the sink.
    pub fn emit_query_response(
        &self,
        kind: &str,
        tag: &str,
        data: &serde_json::Value,
    ) -> std::io::Result<()> {
        self.with_sink(|sink| sink.emit_query_response(kind, tag, data))
    }

    /// Emit a screenshot response through the sink.
    pub fn emit_screenshot_response(
        &self,
        id: &str,
        name: &str,
        hash: &str,
        width: u32,
        height: u32,
        rgba_bytes: &[u8],
    ) -> std::io::Result<()> {
        self.with_sink(|sink| {
            sink.emit_screenshot_response(id, name, hash, width, height, rgba_bytes)
        })
    }

    /// Write pre-encoded bytes through the sink.
    pub fn write_raw(&self, bytes: &[u8]) -> std::io::Result<()> {
        self.with_sink(|sink| sink.write_raw(bytes))
    }

    /// Write a single event without flushing the underlying buffer.
    ///
    /// Used inside loops that drain the pending coalesce map: a
    /// single `flush_output()` call at the end of the loop produces
    /// one syscall for the whole batch. For one-shot callers,
    /// [`do_emit`](Self::do_emit) chains the flush automatically.
    fn do_emit_no_flush(&self, event: OutgoingEvent) -> Task<Message> {
        {
            let mut state = self.batch.lock();
            if state.depth > 0 {
                state.buffer.push(event);
                return Task::none();
            }
        }
        match self.with_sink(|sink| sink.emit_event(event)) {
            Ok(()) => Task::none(),
            Err(e) => {
                log::error!("write error: {e}");
                iced::exit()
            }
        }
    }

    fn do_emit(&self, event: OutgoingEvent) -> Task<Message> {
        // Inside an active batch the event goes into the in-memory
        // buffer and the sink sees nothing until end_batch; skip
        // flushing in that case.
        let buffered = {
            let state = self.batch.lock();
            state.depth > 0
        };
        let emit = self.do_emit_no_flush(event);
        if buffered {
            return emit;
        }
        Task::batch([emit, self.flush_output()])
    }

    /// Flush the sink's output buffer.
    ///
    /// Returns `Task::none()` on success, `iced::exit()` on a broken
    /// pipe so the renderer exits cleanly when the host has
    /// disconnected mid-batch.
    fn flush_output(&self) -> Task<Message> {
        match self.with_sink(|sink| sink.flush_output()) {
            Ok(()) => Task::none(),
            Err(e) => {
                log::error!("flush error: {e}");
                iced::exit()
            }
        }
    }

    fn with_sink<R>(
        &self,
        f: impl FnOnce(&mut dyn EventSink) -> std::io::Result<R>,
    ) -> std::io::Result<R> {
        // sink is the innermost lock on this path: do not add nested
        // locks (no calling back into iced or user code while held).
        let mut guard = self.sink.lock();
        f(&mut **guard)
    }
}

/// Build a CoalesceKey for a widget event.
pub fn widget_coalesce_key(event: &OutgoingEvent) -> CoalesceKey {
    CoalesceKey::Widget(event.id.clone(), event.family.clone())
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use plushie_widget_sdk::protocol::{CoalesceHint, OutgoingEvent};
    use serde_json::json;

    /// No-op sink for unit tests that only exercise rate limiting
    /// and coalescing logic, not actual event delivery.
    struct NullSink;
    impl EventSink for NullSink {
        fn emit_event(&mut self, _: OutgoingEvent) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_effect_response(
            &mut self,
            _: plushie_widget_sdk::protocol::EffectResponse,
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_query_response(
            &mut self,
            _: &str,
            _: &str,
            _: &serde_json::Value,
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_screenshot_response(
            &mut self,
            _: &str,
            _: &str,
            _: &str,
            _: u32,
            _: u32,
            _: &[u8],
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_hello(
            &mut self,
            _: &str,
            _: &str,
            _: &[&str],
            _: &[&str],
            _: &str,
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_diagnostic(
            &mut self,
            _: plushie_widget_sdk::protocol::DiagnosticMessage,
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn write_raw(&mut self, _: &[u8]) -> std::io::Result<()> {
            Ok(())
        }
    }

    fn test_emitter() -> EventEmitter {
        let sink: Arc<SinkMutex> = Arc::new(Mutex::new(Box::new(NullSink)));
        EventEmitter::new(sink)
    }

    /// Sink that records every emit attempt and always returns
    /// `BrokenPipe`. Used to verify the flush paths actually call
    /// the sink (and therefore hit `do_emit`'s error branch, where
    /// the exit Task is constructed).
    struct FailingSink {
        events: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    }
    impl EventSink for FailingSink {
        fn emit_event(&mut self, _: OutgoingEvent) -> std::io::Result<()> {
            self.events
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Err(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
        }
        fn emit_effect_response(
            &mut self,
            _: plushie_widget_sdk::protocol::EffectResponse,
        ) -> std::io::Result<()> {
            Err(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
        }
        fn emit_query_response(
            &mut self,
            _: &str,
            _: &str,
            _: &serde_json::Value,
        ) -> std::io::Result<()> {
            Err(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
        }
        fn emit_screenshot_response(
            &mut self,
            _: &str,
            _: &str,
            _: &str,
            _: u32,
            _: u32,
            _: &[u8],
        ) -> std::io::Result<()> {
            Err(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
        }
        fn emit_hello(
            &mut self,
            _: &str,
            _: &str,
            _: &[&str],
            _: &[&str],
            _: &str,
        ) -> std::io::Result<()> {
            Err(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
        }
        fn emit_diagnostic(
            &mut self,
            _: plushie_widget_sdk::protocol::DiagnosticMessage,
        ) -> std::io::Result<()> {
            Err(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
        }
        fn write_raw(&mut self, _: &[u8]) -> std::io::Result<()> {
            Err(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
        }
    }

    fn make_event(family: &str, id: &str) -> OutgoingEvent {
        OutgoingEvent::generic(family, id, None)
    }

    fn make_event_with_data(family: &str, id: &str, data: serde_json::Value) -> OutgoingEvent {
        OutgoingEvent::generic(family, id, Some(data))
    }

    // -- effective_rate hierarchy --

    #[test]
    fn effective_rate_no_config_returns_none() {
        let emitter = test_emitter();
        let key = CoalesceKey::Subscription("on_pointer_move".into());
        assert_eq!(emitter.effective_rate(&key), None);
    }

    #[test]
    fn effective_rate_uses_default() {
        let mut emitter = test_emitter();
        emitter.set_default_rate(Some(60));
        let key = CoalesceKey::Subscription("on_pointer_move".into());
        assert_eq!(emitter.effective_rate(&key), Some(60));
    }

    #[test]
    fn effective_rate_subscription_overrides_default() {
        let mut emitter = test_emitter();
        emitter.set_default_rate(Some(60));
        emitter.set_subscription_rate("on_pointer_move", 30);
        let key = CoalesceKey::Subscription("on_pointer_move".into());
        assert_eq!(emitter.effective_rate(&key), Some(30));
    }

    #[test]
    fn effective_rate_widget_overrides_default() {
        let mut emitter = test_emitter();
        emitter.set_default_rate(Some(60));
        emitter.set_widget_rate("slider-1", 15);
        let key = CoalesceKey::Widget("slider-1".into(), "slide".into());
        assert_eq!(emitter.effective_rate(&key), Some(15));
    }

    #[test]
    fn effective_rate_widget_without_override_falls_to_default() {
        let mut emitter = test_emitter();
        emitter.set_default_rate(Some(60));
        let key = CoalesceKey::Widget("slider-1".into(), "slide".into());
        assert_eq!(emitter.effective_rate(&key), Some(60));
    }

    // -- clear_widget_rates --

    #[test]
    fn clear_widget_rates_removes_all() {
        let mut emitter = test_emitter();
        emitter.set_widget_rate("a", 10);
        emitter.set_widget_rate("b", 20);
        emitter.clear_widget_rates();
        assert!(emitter.widget_rates.is_empty());
    }

    // -- remove_subscription_rate --

    #[test]
    fn remove_subscription_rate_clears_rate() {
        let mut emitter = test_emitter();
        emitter.set_subscription_rate("on_pointer_move", 30);
        emitter.remove_subscription_rate("on_pointer_move");
        assert!(!emitter.subscription_rates.contains_key("on_pointer_move"));
    }

    #[cfg(debug_assertions)]
    #[test]
    #[should_panic(expected = "event emitter batch depth underflow")]
    fn end_batch_without_begin_panics_in_debug_builds() {
        let emitter = test_emitter();
        emitter.end_batch();
    }

    #[cfg(not(debug_assertions))]
    #[test]
    fn end_batch_without_begin_is_noop_in_release_builds() {
        let emitter = test_emitter();
        emitter.end_batch();
    }

    // -- buffer_event --

    #[test]
    fn buffer_replace_keeps_latest() {
        let mut emitter = test_emitter();
        let key = CoalesceKey::Widget("w1".into(), "slide".into());
        let hint = CoalesceHint::Replace;

        let ev1 = make_event("slide", "w1");
        let _ = emitter.buffer_event(&key, ev1, &hint);

        let ev2 = make_event("slide", "w1");
        let _ = emitter.buffer_event(&key, ev2, &hint);

        assert_eq!(emitter.pending.len(), 1);
    }

    #[test]
    fn buffer_accumulate_sums_deltas() {
        let mut emitter = test_emitter();
        let key = CoalesceKey::Widget("ma1".into(), "scroll".into());
        let hint = CoalesceHint::Accumulate(vec!["delta_x".into(), "delta_y".into()]);

        let ev1 = make_event_with_data("scroll", "ma1", json!({"delta_x": 1.0, "delta_y": 2.0}));
        let _ = emitter.buffer_event(&key, ev1, &hint);

        let ev2 = make_event_with_data("scroll", "ma1", json!({"delta_x": 3.0, "delta_y": 4.0}));
        let _ = emitter.buffer_event(&key, ev2, &hint);

        match emitter.pending.get(&key).unwrap() {
            PendingEvent::Accumulate { totals, .. } => {
                assert!((totals["delta_x"] - 4.0).abs() < f64::EPSILON);
                assert!((totals["delta_y"] - 6.0).abs() < f64::EPSILON);
            }
            _ => panic!("expected Accumulate variant"),
        }
    }

    // -- PendingEvent::into_event --

    #[test]
    fn accumulate_into_event_patches_totals() {
        let base = make_event_with_data(
            "canvas_scroll",
            "c1",
            json!({"delta_x": 1.0, "delta_y": 2.0, "x": 50.0}),
        );
        let mut totals = HashMap::new();
        totals.insert("delta_x".to_string(), 10.0);
        totals.insert("delta_y".to_string(), 20.0);
        let pending = PendingEvent::Accumulate {
            base,
            fields: vec!["delta_x".into(), "delta_y".into()],
            totals,
        };
        let event = pending.into_event();
        let value = event.value.unwrap();
        assert_eq!(value["delta_x"], 10.0);
        assert_eq!(value["delta_y"], 20.0);
        // Other fields preserved.
        assert_eq!(value["x"], 50.0);
    }

    // -- CoalesceHint on constructors --

    #[test]
    fn constructors_set_replace_hint() {
        let events = vec![
            OutgoingEvent::slide("s1", 0.5),
            OutgoingEvent::cursor_moved("t", 1.0, 2.0),
            OutgoingEvent::pointer_move(
                "m1",
                1.0,
                2.0,
                "mouse",
                None,
                plushie_widget_sdk::protocol::KeyModifiers::default(),
            ),
            OutgoingEvent::resize("s1", 100.0, 200.0),
            OutgoingEvent::pane_resized("p1", "s0", 0.5),
            OutgoingEvent::animation_frame("t", 16000),
            OutgoingEvent::theme_changed("t", "dark"),
            OutgoingEvent::finger_moved("t", 1, 10.0, 20.0),
            OutgoingEvent::modifiers_changed(
                "t",
                plushie_widget_sdk::protocol::KeyModifiers::default(),
            ),
            OutgoingEvent::scroll("s1", 0.0, 0.0, 0.0, 0.0, 100.0, 200.0, 300.0, 400.0),
        ];
        for event in events {
            assert!(
                matches!(event.coalesce_hint(), Some(CoalesceHint::Replace)),
                "expected Replace hint on {}",
                event.family
            );
        }
    }

    #[test]
    fn constructors_set_accumulate_hint() {
        let events = vec![
            OutgoingEvent::wheel_scrolled("t", 0.0, -3.0, "line"),
            OutgoingEvent::pointer_scroll(
                "m1",
                0.0,
                0.0,
                0.0,
                -3.0,
                "mouse",
                plushie_widget_sdk::protocol::KeyModifiers::default(),
            ),
        ];
        for event in events {
            assert!(
                matches!(event.coalesce_hint(), Some(CoalesceHint::Accumulate(_))),
                "expected Accumulate hint on {}",
                event.family
            );
        }
    }

    #[test]
    fn constructors_set_no_hint_for_discrete() {
        let events = vec![
            OutgoingEvent::click("b1"),
            OutgoingEvent::input("i1", "text"),
            OutgoingEvent::submit("f1", "data"),
            OutgoingEvent::toggle("c1", true),
            OutgoingEvent::select("p1", "opt"),
            OutgoingEvent::paste("i1", "text"),
            OutgoingEvent::slide_release("s1", 0.5),
            OutgoingEvent::pointer_press(
                "c1",
                1.0,
                2.0,
                "Left",
                "mouse",
                None,
                plushie_widget_sdk::protocol::KeyModifiers::default(),
            ),
            OutgoingEvent::pointer_release(
                "c1",
                1.0,
                2.0,
                "Left",
                "mouse",
                None,
                plushie_widget_sdk::protocol::KeyModifiers::default(),
            ),
            OutgoingEvent::option_hovered("cb1", "opt"),
            OutgoingEvent::cursor_entered("t"),
            OutgoingEvent::cursor_left("t"),
            OutgoingEvent::button_pressed("t", "Left"),
            OutgoingEvent::button_released("t", "Left"),
            OutgoingEvent::pointer_enter("m1"),
            OutgoingEvent::pointer_exit("m1"),
            OutgoingEvent::pane_clicked("pg1", "pane_a"),
            OutgoingEvent::pane_focus_cycle("pg1", "pane_a"),
            OutgoingEvent::pane_dragged("pg1", "picked", "pane_a", None, None, None),
        ];
        for event in events {
            assert!(
                event.coalesce_hint().is_none(),
                "expected no hint on {}",
                event.family
            );
        }
    }

    // -- Accumulate with missing fields --

    #[test]
    fn accumulate_missing_fields_graceful() {
        let hint = CoalesceHint::Accumulate(vec!["dx".into(), "dy".into()]);
        // Event only has dx, not dy.
        let ev = make_event_with_data("custom", "w1", json!({"dx": 5.0}));
        let pending = PendingEvent::from_hint(ev, &hint);
        match &pending {
            PendingEvent::Accumulate { totals, .. } => {
                assert_eq!(totals.get("dx"), Some(&5.0));
                assert_eq!(totals.get("dy"), None);
            }
            _ => panic!("expected Accumulate"),
        }
    }

    // -- Mixed hinted/unhinted events (ordering guarantee) --

    #[test]
    fn emit_immediate_flushes_pending_first() {
        let mut emitter = test_emitter();
        let key = CoalesceKey::Widget("w1".into(), "cursor_pos".into());
        let hint = CoalesceHint::Replace;

        // Buffer a coalescable event.
        let ev = make_event("cursor_pos", "w1");
        let _ = emitter.buffer_event(&key, ev, &hint);
        assert_eq!(emitter.pending.len(), 1);

        // emit_immediate should flush pending events first (even though
        // it can't actually write to stdout in tests, the flush clears
        // the pending buffer).
        let discrete = make_event("click", "w1");
        let _ = emitter.emit_immediate(discrete);

        // The pending buffer should be empty after flush.
        assert!(emitter.pending.is_empty());
    }

    // -- Strategy mismatch (widget changes hint between events) --

    #[test]
    fn buffer_event_flushes_on_strategy_mismatch() {
        let mut emitter = test_emitter();
        let key = CoalesceKey::Widget("w1".into(), "update".into());

        // Buffer a Replace event.
        let ev1 = make_event_with_data("update", "w1", json!({"x": 1.0}));
        let _ = emitter.buffer_event(&key, ev1, &CoalesceHint::Replace);
        assert_eq!(emitter.pending.len(), 1);

        // Buffer an Accumulate event with the same key (strategy mismatch).
        // The old Replace entry should be flushed and a new Accumulate started.
        let ev2 = make_event_with_data("update", "w1", json!({"dx": 5.0}));
        let acc_hint = CoalesceHint::Accumulate(vec!["dx".into()]);
        let _ = emitter.buffer_event(&key, ev2, &acc_hint);

        // Should still have one pending entry, but now it's Accumulate.
        assert_eq!(emitter.pending.len(), 1);
        assert!(matches!(
            emitter.pending.get(&key),
            Some(PendingEvent::Accumulate { .. })
        ));
    }

    // -- Accumulate with custom fields --

    #[test]
    fn accumulate_custom_fields() {
        let mut emitter = test_emitter();
        let key = CoalesceKey::Widget("w1".into(), "physics".into());
        let hint = CoalesceHint::Accumulate(vec!["impulse_x".into(), "impulse_y".into()]);

        let ev1 = make_event_with_data(
            "physics",
            "w1",
            json!({"x": 10.0, "y": 20.0, "impulse_x": 1.0, "impulse_y": 2.0}),
        );
        let _ = emitter.buffer_event(&key, ev1, &hint);

        let ev2 = make_event_with_data(
            "physics",
            "w1",
            json!({"x": 15.0, "y": 25.0, "impulse_x": 3.0, "impulse_y": 4.0}),
        );
        let _ = emitter.buffer_event(&key, ev2, &hint);

        let result = emitter.pending.remove(&key).unwrap().into_event();
        let value = result.value.unwrap();
        // Position fields: latest value wins.
        assert_eq!(value["x"], 15.0);
        assert_eq!(value["y"], 25.0);
        // Impulse fields: accumulated.
        assert_eq!(value["impulse_x"], 4.0);
        assert_eq!(value["impulse_y"], 6.0);
    }

    // -- Broken-pipe propagation through the coalesced flush path --

    /// A buffered, coalesced event must reach the sink when the
    /// emitter flushes. Pre-fix `flush_all` discarded the Task
    /// returned by `do_emit`; the broken-pipe error never produced
    /// an exit signal and the renderer would wedge silently.
    /// Counting writes through a failing sink covers the path: if
    /// the sink is hit, the Task carrying `iced::exit()` is the
    /// only thing the caller can observe, so propagation is the
    /// load-bearing piece.
    #[test]
    fn flush_all_drives_writes_after_buffering() {
        let counter = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let sink: Arc<SinkMutex> = Arc::new(Mutex::new(Box::new(FailingSink {
            events: counter.clone(),
        })));
        let mut emitter = EventEmitter::new(sink);

        let key = CoalesceKey::Subscription("on_pointer_move".into());
        let hint = CoalesceHint::Replace;
        let _ = emitter.buffer_event(&key, make_event("pointer_moved", ""), &hint);
        assert_eq!(emitter.pending.len(), 1);

        let _ = emitter.flush();
        assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 1);
        assert!(emitter.pending.is_empty());
    }

    #[test]
    fn flush_key_drives_writes() {
        let counter = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let sink: Arc<SinkMutex> = Arc::new(Mutex::new(Box::new(FailingSink {
            events: counter.clone(),
        })));
        let mut emitter = EventEmitter::new(sink);

        let key = CoalesceKey::Subscription("on_pointer_move".into());
        let hint = CoalesceHint::Replace;
        let _ = emitter.buffer_event(&key, make_event("pointer_moved", ""), &hint);

        let _ = emitter.flush_key(&key);
        assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 1);
    }

    /// A sink that accepts writes (so the BufWriter-style coalesced
    /// path ends up calling `flush_output` once at the end of the
    /// drain) but reports broken-pipe on the buffer flush. The
    /// emitter must surface that as an exit task instead of
    /// swallowing it.
    struct FlushFailingSink {
        flushes: std::sync::Arc<std::sync::atomic::AtomicUsize>,
        writes: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    }
    impl EventSink for FlushFailingSink {
        fn emit_event(&mut self, _: OutgoingEvent) -> std::io::Result<()> {
            self.writes
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Ok(())
        }
        fn emit_effect_response(
            &mut self,
            _: plushie_widget_sdk::protocol::EffectResponse,
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_query_response(
            &mut self,
            _: &str,
            _: &str,
            _: &serde_json::Value,
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_screenshot_response(
            &mut self,
            _: &str,
            _: &str,
            _: &str,
            _: u32,
            _: u32,
            _: &[u8],
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_hello(
            &mut self,
            _: &str,
            _: &str,
            _: &[&str],
            _: &[&str],
            _: &str,
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn emit_diagnostic(
            &mut self,
            _: plushie_widget_sdk::protocol::DiagnosticMessage,
        ) -> std::io::Result<()> {
            Ok(())
        }
        fn write_raw(&mut self, _: &[u8]) -> std::io::Result<()> {
            Ok(())
        }
        fn flush_output(&mut self) -> std::io::Result<()> {
            self.flushes
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Err(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
        }
    }

    /// Coalesce-flush path must surface a broken-pipe-on-flush
    /// error. With BufWriter wrapping the underlying writer, write
    /// errors only show up at flush time; the renderer would wedge
    /// if the flush failure didn't produce an exit task.
    #[test]
    fn flush_all_propagates_flush_output_error() {
        let writes = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let flushes = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let sink: Arc<SinkMutex> = Arc::new(Mutex::new(Box::new(FlushFailingSink {
            writes: writes.clone(),
            flushes: flushes.clone(),
        })));
        let mut emitter = EventEmitter::new(sink);

        let key = CoalesceKey::Subscription("on_pointer_move".into());
        let hint = CoalesceHint::Replace;
        let _ = emitter.buffer_event(&key, make_event("pointer_moved", ""), &hint);
        assert_eq!(emitter.pending.len(), 1);

        // Drive the FlushCoalesce timer entry point; the buffered
        // event drains, the BufWriter flushes, the flush errors,
        // and the resulting Task carries iced::exit. We can't
        // introspect the Task directly without running an iced
        // runtime, but we can prove the flush path ran exactly
        // once for the whole drained batch.
        let _task = emitter.flush();
        assert_eq!(writes.load(std::sync::atomic::Ordering::SeqCst), 1);
        assert_eq!(flushes.load(std::sync::atomic::Ordering::SeqCst), 1);
        assert!(emitter.pending.is_empty());
    }

    /// Direct emit path: a coalescable event whose rate limit
    /// allows immediate dispatch must still flush after writing
    /// (otherwise the BufWriter holds the bytes indefinitely).
    #[test]
    fn coalesce_immediate_path_flushes_after_emit() {
        let writes = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let flushes = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let sink: Arc<SinkMutex> = Arc::new(Mutex::new(Box::new(FlushFailingSink {
            writes: writes.clone(),
            flushes: flushes.clone(),
        })));
        let mut emitter = EventEmitter::new(sink);

        // No rate -> coalesce takes the immediate emit branch.
        let event = make_event("pointer_moved", "").with_coalesce(CoalesceHint::Replace);
        let key = CoalesceKey::Subscription("on_pointer_move".into());
        let _task = emitter.coalesce(key, event);

        assert_eq!(writes.load(std::sync::atomic::Ordering::SeqCst), 1);
        assert_eq!(flushes.load(std::sync::atomic::Ordering::SeqCst), 1);
    }

    /// Drained batch produces one syscall worth of work for the
    /// whole flush, not one per event.
    #[test]
    fn flush_all_uses_single_flush_for_whole_batch() {
        let writes = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let flushes = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        struct SuccessSink {
            writes: std::sync::Arc<std::sync::atomic::AtomicUsize>,
            flushes: std::sync::Arc<std::sync::atomic::AtomicUsize>,
        }
        impl EventSink for SuccessSink {
            fn emit_event(&mut self, _: OutgoingEvent) -> std::io::Result<()> {
                self.writes
                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                Ok(())
            }
            fn emit_effect_response(
                &mut self,
                _: plushie_widget_sdk::protocol::EffectResponse,
            ) -> std::io::Result<()> {
                Ok(())
            }
            fn emit_query_response(
                &mut self,
                _: &str,
                _: &str,
                _: &serde_json::Value,
            ) -> std::io::Result<()> {
                Ok(())
            }
            fn emit_screenshot_response(
                &mut self,
                _: &str,
                _: &str,
                _: &str,
                _: u32,
                _: u32,
                _: &[u8],
            ) -> std::io::Result<()> {
                Ok(())
            }
            fn emit_hello(
                &mut self,
                _: &str,
                _: &str,
                _: &[&str],
                _: &[&str],
                _: &str,
            ) -> std::io::Result<()> {
                Ok(())
            }
            fn emit_diagnostic(
                &mut self,
                _: plushie_widget_sdk::protocol::DiagnosticMessage,
            ) -> std::io::Result<()> {
                Ok(())
            }
            fn write_raw(&mut self, _: &[u8]) -> std::io::Result<()> {
                Ok(())
            }
            fn flush_output(&mut self) -> std::io::Result<()> {
                self.flushes
                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                Ok(())
            }
        }

        let sink: Arc<SinkMutex> = Arc::new(Mutex::new(Box::new(SuccessSink {
            writes: writes.clone(),
            flushes: flushes.clone(),
        })));
        let mut emitter = EventEmitter::new(sink);

        for i in 0..5 {
            let key = CoalesceKey::Subscription(format!("sub_{i}"));
            let _ = emitter.buffer_event(&key, make_event("ev", ""), &CoalesceHint::Replace);
        }
        let _task = emitter.flush();
        assert_eq!(writes.load(std::sync::atomic::Ordering::SeqCst), 5);
        assert_eq!(flushes.load(std::sync::atomic::Ordering::SeqCst), 1);
    }
}