//! EventSource (Server-Sent Events) support.

use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::sync::mpsc;
use std::thread::{self, JoinHandle};
use std::time::Duration;

use crate::builtins::{make_native, set_builtin_prop};
use crate::gc::GcRef;
use crate::vm::{HeapObject, NativeContext, ObjectData, Property, RuntimeError, Value, Vm};

const EVENTSOURCE_ID_KEY: &str = "__eventsource_id__";
const DEFAULT_RETRY_MS: u64 = 3000;
const MAX_RETRY_MS: u64 = 60_000;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SseEvent {
    pub event_type: String,
    pub data: String,
    pub last_event_id: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SseParserOutput {
    Event(SseEvent),
    Retry(u64),
}

#[derive(Debug, Clone)]
pub struct SseParser {
    line: Vec<u8>,
    saw_cr: bool,
    data: String,
    event_type: String,
    last_event_id: String,
}

impl SseParser {
    pub fn new(last_event_id: String) -> Self {
        Self {
            line: Vec::new(),
            saw_cr: false,
            data: String::new(),
            event_type: String::new(),
            last_event_id,
        }
    }

    pub fn last_event_id(&self) -> &str {
        &self.last_event_id
    }

    pub fn push_bytes(&mut self, bytes: &[u8]) -> Vec<SseParserOutput> {
        let mut out = Vec::new();
        for &byte in bytes {
            if self.saw_cr {
                self.saw_cr = false;
                if byte == b'\n' {
                    continue;
                }
            }

            match byte {
                b'\r' => {
                    self.saw_cr = true;
                    self.finish_line(&mut out);
                }
                b'\n' => self.finish_line(&mut out),
                _ => self.line.push(byte),
            }
        }
        out
    }

    fn finish_line(&mut self, out: &mut Vec<SseParserOutput>) {
        let line = String::from_utf8_lossy(&self.line).into_owned();
        self.line.clear();

        if line.is_empty() {
            if self.data.is_empty() {
                self.event_type.clear();
                return;
            }
            if self.data.ends_with('\n') {
                self.data.pop();
            }
            let event_type = if self.event_type.is_empty() {
                "message".to_string()
            } else {
                self.event_type.clone()
            };
            out.push(SseParserOutput::Event(SseEvent {
                event_type,
                data: std::mem::take(&mut self.data),
                last_event_id: self.last_event_id.clone(),
            }));
            self.event_type.clear();
            return;
        }

        if line.starts_with(':') {
            return;
        }

        let (field, value) = match line.split_once(':') {
            Some((field, value)) => {
                let value = value.strip_prefix(' ').unwrap_or(value);
                (field, value)
            }
            None => (line.as_str(), ""),
        };

        match field {
            "data" => {
                self.data.push_str(value);
                self.data.push('\n');
            }
            "event" => self.event_type = value.to_string(),
            "id" if !value.contains('\0') => self.last_event_id = value.to_string(),
            "retry" if value.bytes().all(|b| b.is_ascii_digit()) => {
                if let Ok(ms) = value.parse::<u64>() {
                    out.push(SseParserOutput::Retry(ms.min(MAX_RETRY_MS)));
                }
            }
            _ => {}
        }
    }
}

enum EventSourceControl {
    Close,
}

enum EventSourceTraffic {
    Open,
    Message(SseEvent),
    Error { closed: bool },
    Closed,
}

struct EventSourceInstance {
    control_tx: mpsc::Sender<EventSourceControl>,
    traffic_rx: mpsc::Receiver<EventSourceTraffic>,
    thread: Option<JoinHandle<()>>,
    js_object: GcRef,
}

struct EventSourceRegistry {
    next_id: u64,
    sources: HashMap<u64, EventSourceInstance>,
}

impl EventSourceRegistry {
    fn new() -> Self {
        Self {
            next_id: 1,
            sources: HashMap::new(),
        }
    }
}

thread_local! {
    static REGISTRY: RefCell<EventSourceRegistry> = RefCell::new(EventSourceRegistry::new());
}

pub fn eventsource_gc_roots() -> Vec<GcRef> {
    REGISTRY.with(|r| r.borrow().sources.values().map(|s| s.js_object).collect())
}

pub fn has_live_event_sources() -> bool {
    REGISTRY.with(|r| !r.borrow().sources.is_empty())
}

pub fn reset_eventsource_registry() {
    REGISTRY.with(|r| {
        let mut reg = r.borrow_mut();
        for (_, mut source) in reg.sources.drain() {
            let _ = source.control_tx.send(EventSourceControl::Close);
            if let Some(thread) = source.thread.take() {
                let _ = thread.join();
            }
        }
        reg.next_id = 1;
    });
}

fn eventsource_constructor(args: &[Value], ctx: &mut NativeContext) -> Result<Value, RuntimeError> {
    let url = match args.first() {
        Some(v) => v.to_js_string(ctx.gc),
        None => {
            return Err(RuntimeError::type_error(
                "EventSource constructor requires a URL argument",
            ))
        }
    };

    let mut with_credentials = false;
    if let Some(Value::Object(opts_ref)) = args.get(1) {
        if let Some(HeapObject::Object(opts)) = ctx.gc.get(*opts_ref) {
            if let Some(prop) = opts.get_property("withCredentials", ctx.shapes) {
                with_credentials = matches!(prop.value, Value::Boolean(true));
            }
        }
    }

    let source_id = REGISTRY.with(|r| {
        let mut reg = r.borrow_mut();
        let id = reg.next_id;
        reg.next_id += 1;
        id
    });

    let js_object = build_eventsource_object(ctx, source_id, &url, with_credentials);
    let (control_tx, control_rx) = mpsc::channel();
    let (traffic_tx, traffic_rx) = mpsc::channel();
    let document_origin = crate::fetch::get_document_origin();
    let thread_url = url.clone();
    let thread = thread::Builder::new()
        .name(format!("we-eventsource:{thread_url}"))
        .spawn(move || {
            run_eventsource_thread(
                thread_url,
                with_credentials,
                document_origin,
                control_rx,
                traffic_tx,
            );
        })
        .map_err(|e| {
            RuntimeError::type_error(format!("EventSource: failed to spawn thread: {e}"))
        })?;

    REGISTRY.with(|r| {
        r.borrow_mut().sources.insert(
            source_id,
            EventSourceInstance {
                control_tx,
                traffic_rx,
                thread: Some(thread),
                js_object,
            },
        );
    });

    Ok(Value::Object(js_object))
}

fn build_eventsource_object(
    ctx: &mut NativeContext,
    source_id: u64,
    url: &str,
    with_credentials: bool,
) -> GcRef {
    let mut data = ObjectData::new();
    data.insert_property(
        EVENTSOURCE_ID_KEY.to_string(),
        Property::builtin(Value::Number(source_id as f64)),
        ctx.shapes,
    );
    data.insert_property(
        "url".to_string(),
        Property::data(Value::String(url.to_string())),
        ctx.shapes,
    );
    data.insert_property(
        "readyState".to_string(),
        Property::data(Value::Number(0.0)),
        ctx.shapes,
    );
    data.insert_property(
        "withCredentials".to_string(),
        Property::data(Value::Boolean(with_credentials)),
        ctx.shapes,
    );
    data.insert_property(
        "CONNECTING".to_string(),
        Property::builtin(Value::Number(0.0)),
        ctx.shapes,
    );
    data.insert_property(
        "OPEN".to_string(),
        Property::builtin(Value::Number(1.0)),
        ctx.shapes,
    );
    data.insert_property(
        "CLOSED".to_string(),
        Property::builtin(Value::Number(2.0)),
        ctx.shapes,
    );
    data.insert_property(
        "onopen".to_string(),
        Property::data(Value::Null),
        ctx.shapes,
    );
    data.insert_property(
        "onmessage".to_string(),
        Property::data(Value::Null),
        ctx.shapes,
    );
    data.insert_property(
        "onerror".to_string(),
        Property::data(Value::Null),
        ctx.shapes,
    );

    let obj_ref = ctx.gc.alloc(HeapObject::Object(data));
    let close = make_native(ctx.gc, "close", eventsource_close);
    set_builtin_prop(ctx.gc, ctx.shapes, obj_ref, "close", Value::Function(close));
    let add = make_native(ctx.gc, "addEventListener", eventsource_add_event_listener);
    set_builtin_prop(
        ctx.gc,
        ctx.shapes,
        obj_ref,
        "addEventListener",
        Value::Function(add),
    );
    let remove = make_native(
        ctx.gc,
        "removeEventListener",
        eventsource_remove_event_listener,
    );
    set_builtin_prop(
        ctx.gc,
        ctx.shapes,
        obj_ref,
        "removeEventListener",
        Value::Function(remove),
    );
    ensure_listener_array(ctx, obj_ref, "open");
    ensure_listener_array(ctx, obj_ref, "message");
    ensure_listener_array(ctx, obj_ref, "error");
    obj_ref
}

fn run_eventsource_thread(
    url: String,
    with_credentials: bool,
    document_origin: Option<String>,
    control_rx: mpsc::Receiver<EventSourceControl>,
    traffic_tx: mpsc::Sender<EventSourceTraffic>,
) {
    let mut retry_ms = DEFAULT_RETRY_MS;
    let mut last_event_id = String::new();

    loop {
        if matches!(control_rx.try_recv(), Ok(EventSourceControl::Close)) {
            let _ = traffic_tx.send(EventSourceTraffic::Closed);
            return;
        }

        let mut headers = we_net::http::Headers::new();
        headers.set("Accept", "text/event-stream");
        headers.set("Cache-Control", "no-store");
        if !last_event_id.is_empty() {
            headers.set("Last-Event-ID", &last_event_id);
        }
        if let Some(origin) = &document_origin {
            headers.set("Origin", origin);
        }

        let result = connect_once(
            &url,
            with_credentials,
            document_origin.as_deref(),
            &headers,
            &control_rx,
            &traffic_tx,
            &mut retry_ms,
            &mut last_event_id,
        );

        match result {
            ConnectOutcome::ClosedByUser => {
                let _ = traffic_tx.send(EventSourceTraffic::Closed);
                return;
            }
            ConnectOutcome::Fatal => {
                let _ = traffic_tx.send(EventSourceTraffic::Error { closed: true });
                return;
            }
            ConnectOutcome::Reconnect => {
                let _ = traffic_tx.send(EventSourceTraffic::Error { closed: false });
                if wait_for_reconnect_or_close(&control_rx, retry_ms) {
                    let _ = traffic_tx.send(EventSourceTraffic::Closed);
                    return;
                }
            }
        }
    }
}

fn wait_for_reconnect_or_close(
    control_rx: &mpsc::Receiver<EventSourceControl>,
    retry_ms: u64,
) -> bool {
    let mut remaining = retry_ms;
    while remaining > 0 {
        if matches!(control_rx.try_recv(), Ok(EventSourceControl::Close)) {
            return true;
        }
        let step = remaining.min(10);
        thread::sleep(Duration::from_millis(step));
        remaining -= step;
    }
    matches!(control_rx.try_recv(), Ok(EventSourceControl::Close))
}

enum ConnectOutcome {
    Reconnect,
    Fatal,
    ClosedByUser,
}

#[allow(clippy::too_many_arguments)]
fn connect_once(
    url: &str,
    with_credentials: bool,
    document_origin: Option<&str>,
    headers: &we_net::http::Headers,
    control_rx: &mpsc::Receiver<EventSourceControl>,
    traffic_tx: &mpsc::Sender<EventSourceTraffic>,
    retry_ms: &mut u64,
    last_event_id: &mut String,
) -> ConnectOutcome {
    let parsed = match we_url::Url::parse(url) {
        Ok(url) => url,
        Err(_) => return ConnectOutcome::Fatal,
    };
    let mut client = we_net::client::HttpClient::new();
    let mut parser = SseParser::new(last_event_id.clone());
    let opened = Cell::new(false);
    let rejected = Cell::new(false);

    let response = client.request_streaming(
        we_net::http::Method::Get,
        &parsed,
        headers,
        None,
        |response| {
            if is_terminal_status(response.status_code) {
                rejected.set(true);
                return Err(we_net::client::ClientError::ConnectionClosed);
            }
            if !(200..300).contains(&response.status_code) {
                return Ok(());
            }
            if !is_event_stream(&response.headers) {
                rejected.set(true);
                return Err(we_net::client::ClientError::ConnectionClosed);
            }
            if requires_cors(&parsed, document_origin)
                && !cors_allows(
                    &response.headers,
                    document_origin.unwrap_or(""),
                    with_credentials,
                )
            {
                rejected.set(true);
                return Err(we_net::client::ClientError::ConnectionClosed);
            }
            if !opened.get() {
                opened.set(true);
                let _ = traffic_tx.send(EventSourceTraffic::Open);
            }
            Ok(())
        },
        |chunk| {
            if matches!(control_rx.try_recv(), Ok(EventSourceControl::Close)) {
                return Err(we_net::client::ClientError::ConnectionClosed);
            }
            if !opened.get() {
                opened.set(true);
                let _ = traffic_tx.send(EventSourceTraffic::Open);
            }
            for item in parser.push_bytes(chunk) {
                match item {
                    SseParserOutput::Event(event) => {
                        *last_event_id = parser.last_event_id().to_string();
                        let _ = traffic_tx.send(EventSourceTraffic::Message(event));
                    }
                    SseParserOutput::Retry(ms) => *retry_ms = ms,
                }
            }
            Ok(())
        },
    );

    let response = match response {
        Ok(response) => response,
        Err(we_net::client::ClientError::ConnectionClosed) => {
            if rejected.get() {
                return ConnectOutcome::Fatal;
            }
            return ConnectOutcome::ClosedByUser;
        }
        Err(_) => return ConnectOutcome::Reconnect,
    };

    if !(200..300).contains(&response.status_code) {
        return ConnectOutcome::Reconnect;
    }
    if !opened.get() {
        let _ = traffic_tx.send(EventSourceTraffic::Open);
    }

    ConnectOutcome::Reconnect
}

fn is_event_stream(headers: &we_net::http::Headers) -> bool {
    headers
        .get("Content-Type")
        .map(|value| {
            value
                .split(';')
                .next()
                .unwrap_or("")
                .trim()
                .eq_ignore_ascii_case("text/event-stream")
        })
        .unwrap_or(false)
}

fn is_terminal_status(status: u16) -> bool {
    matches!(status, 204 | 401 | 403 | 404 | 410) || status >= 500
}

fn requires_cors(url: &we_url::Url, document_origin: Option<&str>) -> bool {
    let Some(origin) = document_origin else {
        return false;
    };
    let Ok(origin_url) = we_url::Url::parse(&format!("{origin}/")) else {
        return true;
    };
    !origin_url.origin().same_origin(&url.origin())
}

fn cors_allows(headers: &we_net::http::Headers, origin: &str, with_credentials: bool) -> bool {
    let allow_origin = headers.get("Access-Control-Allow-Origin").unwrap_or("");
    let origin_ok = allow_origin == "*" || allow_origin == origin;
    if !origin_ok {
        return false;
    }
    if with_credentials {
        allow_origin == origin
            && headers
                .get("Access-Control-Allow-Credentials")
                .map(|v| v.eq_ignore_ascii_case("true"))
                .unwrap_or(false)
    } else {
        true
    }
}

fn eventsource_id_from_this(ctx: &NativeContext) -> Option<u64> {
    let this_ref = ctx.this.gc_ref()?;
    match ctx.gc.get(this_ref) {
        Some(HeapObject::Object(data)) => data
            .get_property(EVENTSOURCE_ID_KEY, ctx.shapes)
            .map(|p| p.value.to_number() as u64),
        _ => None,
    }
}

fn eventsource_close(_args: &[Value], ctx: &mut NativeContext) -> Result<Value, RuntimeError> {
    let Some(id) = eventsource_id_from_this(ctx) else {
        return Ok(Value::Undefined);
    };
    let source = REGISTRY.with(|r| r.borrow_mut().sources.remove(&id));
    if let Some(mut source) = source {
        let _ = source.control_tx.send(EventSourceControl::Close);
        if let Some(thread) = source.thread.take() {
            let _ = thread.join();
        }
        set_object_prop(ctx, source.js_object, "readyState", Value::Number(2.0));
    }
    Ok(Value::Undefined)
}

fn eventsource_add_event_listener(
    args: &[Value],
    ctx: &mut NativeContext,
) -> Result<Value, RuntimeError> {
    let this_ref = match ctx.this.gc_ref() {
        Some(r) => r,
        None => return Ok(Value::Undefined),
    };
    let event_type = args
        .first()
        .map(|v| v.to_js_string(ctx.gc))
        .unwrap_or_default();
    let callback = match args.get(1) {
        Some(Value::Function(r)) => *r,
        _ => return Ok(Value::Undefined),
    };
    let listeners_ref = ensure_listener_array(ctx, this_ref, &event_type);
    append_to_array(listeners_ref, Value::Function(callback), ctx);
    Ok(Value::Undefined)
}

fn eventsource_remove_event_listener(
    args: &[Value],
    ctx: &mut NativeContext,
) -> Result<Value, RuntimeError> {
    let this_ref = match ctx.this.gc_ref() {
        Some(r) => r,
        None => return Ok(Value::Undefined),
    };
    let event_type = args
        .first()
        .map(|v| v.to_js_string(ctx.gc))
        .unwrap_or_default();
    let callback = match args.get(1) {
        Some(Value::Function(r)) => *r,
        _ => return Ok(Value::Undefined),
    };
    if let Some(Value::Object(listeners_ref)) =
        get_object_prop(ctx.gc, ctx.shapes, this_ref, &listener_key(&event_type))
    {
        remove_from_array(listeners_ref, callback, ctx);
    }
    Ok(Value::Undefined)
}

fn listener_key(event_type: &str) -> String {
    format!("__eventsource_listeners_{event_type}__")
}

fn ensure_listener_array(ctx: &mut NativeContext, obj_ref: GcRef, event_type: &str) -> GcRef {
    let key = listener_key(event_type);
    if let Some(Value::Object(existing)) = get_object_prop(ctx.gc, ctx.shapes, obj_ref, &key) {
        return existing;
    }
    let arr = ctx.gc.alloc(HeapObject::Object(empty_array(ctx.shapes)));
    set_builtin_prop(ctx.gc, ctx.shapes, obj_ref, &key, Value::Object(arr));
    arr
}

fn empty_array(shapes: &mut crate::shape::ShapeTable) -> ObjectData {
    let mut data = ObjectData::new();
    data.insert_property(
        "length".to_string(),
        Property {
            value: Value::Number(0.0),
            getter: None,
            setter: None,
            writable: true,
            enumerable: false,
            configurable: false,
        },
        shapes,
    );
    data
}

fn set_object_prop(ctx: &mut NativeContext, obj_ref: GcRef, key: &str, value: Value) {
    if let Some(HeapObject::Object(data)) = ctx.gc.get_mut(obj_ref) {
        data.insert_property(key.to_string(), Property::data(value), ctx.shapes);
    }
}

fn set_object_prop_vm(vm: &mut Vm, obj_ref: GcRef, key: &str, value: Value) {
    if let Some(HeapObject::Object(data)) = vm.gc.get_mut(obj_ref) {
        data.insert_property(key.to_string(), Property::data(value), &mut vm.shapes);
    }
}

fn get_object_prop(
    gc: &crate::gc::Gc<HeapObject>,
    shapes: &crate::shape::ShapeTable,
    obj_ref: GcRef,
    key: &str,
) -> Option<Value> {
    match gc.get(obj_ref) {
        Some(HeapObject::Object(data)) => data.get_property(key, shapes).map(|p| p.value),
        _ => None,
    }
}

fn append_to_array(arr_ref: GcRef, val: Value, ctx: &mut NativeContext) {
    if let Some(HeapObject::Object(data)) = ctx.gc.get_mut(arr_ref) {
        let len = data
            .get_property("length", ctx.shapes)
            .map(|p| p.value.to_number() as usize)
            .unwrap_or(0);
        data.insert_property(len.to_string(), Property::data(val), ctx.shapes);
        data.insert_property(
            "length".to_string(),
            Property::data(Value::Number((len + 1) as f64)),
            ctx.shapes,
        );
    }
}

fn remove_from_array(arr_ref: GcRef, target: GcRef, ctx: &mut NativeContext) {
    if let Some(HeapObject::Object(data)) = ctx.gc.get_mut(arr_ref) {
        let len = data
            .get_property("length", ctx.shapes)
            .map(|p| p.value.to_number() as usize)
            .unwrap_or(0);
        let mut kept = Vec::with_capacity(len);
        for i in 0..len {
            if let Some(prop) = data.get_property(&i.to_string(), ctx.shapes) {
                if !matches!(prop.value, Value::Function(r) if r == target) {
                    kept.push(prop.value);
                }
            }
        }
        for i in 0..len {
            data.insert_property(i.to_string(), Property::data(Value::Undefined), ctx.shapes);
        }
        for (i, value) in kept.iter().enumerate() {
            data.insert_property(i.to_string(), Property::data(value.clone()), ctx.shapes);
        }
        data.insert_property(
            "length".to_string(),
            Property::data(Value::Number(kept.len() as f64)),
            ctx.shapes,
        );
    }
}

struct DrainedEventSourceTraffic {
    events: Vec<(GcRef, EventSourceTraffic)>,
    closed: Vec<u64>,
}

fn drain_traffic() -> DrainedEventSourceTraffic {
    let mut events = Vec::new();
    let mut closed = Vec::new();
    REGISTRY.with(|r| {
        let reg = r.borrow();
        for (id, source) in &reg.sources {
            while let Ok(event) = source.traffic_rx.try_recv() {
                if matches!(event, EventSourceTraffic::Closed) {
                    closed.push(*id);
                }
                events.push((source.js_object, event));
            }
        }
    });
    DrainedEventSourceTraffic { events, closed }
}

pub fn drain_eventsource_events(vm: &mut Vm) -> Result<(), RuntimeError> {
    let DrainedEventSourceTraffic { events, closed } = drain_traffic();
    for (source_ref, event) in events {
        match event {
            EventSourceTraffic::Open => {
                set_object_prop_vm(vm, source_ref, "readyState", Value::Number(1.0));
                dispatch_event(vm, source_ref, "open", None, "");
            }
            EventSourceTraffic::Message(event) => {
                dispatch_event(
                    vm,
                    source_ref,
                    &event.event_type,
                    Some((event.data, event.last_event_id)),
                    "",
                );
            }
            EventSourceTraffic::Error { closed } => {
                set_object_prop_vm(
                    vm,
                    source_ref,
                    "readyState",
                    Value::Number(if closed { 2.0 } else { 0.0 }),
                );
                dispatch_event(vm, source_ref, "error", None, "");
            }
            EventSourceTraffic::Closed => {
                set_object_prop_vm(vm, source_ref, "readyState", Value::Number(2.0));
            }
        }
    }

    if !closed.is_empty() {
        REGISTRY.with(|r| {
            let mut reg = r.borrow_mut();
            for id in closed {
                reg.sources.remove(&id);
            }
        });
    }
    Ok(())
}

fn dispatch_event(
    vm: &mut Vm,
    source_ref: GcRef,
    event_type: &str,
    message: Option<(String, String)>,
    origin: &str,
) {
    let event_ref = build_event(vm, event_type, message, origin);
    let event_val = Value::Object(event_ref);
    let handler_name = match event_type {
        "open" => "onopen",
        "message" => "onmessage",
        "error" => "onerror",
        _ => "",
    };
    if !handler_name.is_empty() {
        if let Some(Value::Function(handler)) =
            get_object_prop(&vm.gc, &vm.shapes, source_ref, handler_name)
        {
            let prev_this = vm.get_global("this").cloned().unwrap_or(Value::Undefined);
            vm.set_global("this", Value::Object(source_ref));
            let _ = vm.call_function(handler, std::slice::from_ref(&event_val));
            vm.set_global("this", prev_this);
        }
    }

    if let Some(Value::Object(listeners_ref)) =
        get_object_prop(&vm.gc, &vm.shapes, source_ref, &listener_key(event_type))
    {
        for handler in collect_funcs(vm, listeners_ref) {
            let prev_this = vm.get_global("this").cloned().unwrap_or(Value::Undefined);
            vm.set_global("this", Value::Object(source_ref));
            let _ = vm.call_function(handler, std::slice::from_ref(&event_val));
            vm.set_global("this", prev_this);
        }
    }
}

fn build_event(
    vm: &mut Vm,
    event_type: &str,
    message: Option<(String, String)>,
    origin: &str,
) -> GcRef {
    let mut data = ObjectData::new();
    data.insert_property(
        "type".to_string(),
        Property::data(Value::String(event_type.to_string())),
        &mut vm.shapes,
    );
    let (data_value, last_event_id) = message.unwrap_or_else(|| (String::new(), String::new()));
    data.insert_property(
        "data".to_string(),
        Property::data(Value::String(data_value)),
        &mut vm.shapes,
    );
    data.insert_property(
        "lastEventId".to_string(),
        Property::data(Value::String(last_event_id)),
        &mut vm.shapes,
    );
    data.insert_property(
        "origin".to_string(),
        Property::data(Value::String(origin.to_string())),
        &mut vm.shapes,
    );
    vm.gc.alloc(HeapObject::Object(data))
}

fn collect_funcs(vm: &Vm, arr_ref: GcRef) -> Vec<GcRef> {
    let mut out = Vec::new();
    if let Some(HeapObject::Object(data)) = vm.gc.get(arr_ref) {
        let len = data
            .get_property("length", &vm.shapes)
            .map(|p| p.value.to_number() as usize)
            .unwrap_or(0);
        for i in 0..len {
            if let Some(prop) = data.get_property(&i.to_string(), &vm.shapes) {
                if let Value::Function(r) = prop.value {
                    out.push(r);
                }
            }
        }
    }
    out
}

pub fn init_eventsource_api(vm: &mut Vm) {
    let ctor = make_native(&mut vm.gc, "EventSource", eventsource_constructor);
    if let Some(HeapObject::Function(func)) = vm.gc.get_mut(ctor) {
        func.properties.insert(
            "CONNECTING".to_string(),
            Property::builtin(Value::Number(0.0)),
        );
        func.properties
            .insert("OPEN".to_string(), Property::builtin(Value::Number(1.0)));
        func.properties
            .insert("CLOSED".to_string(), Property::builtin(Value::Number(2.0)));
    }
    vm.set_global("EventSource", Value::Function(ctor));
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::compiler;
    use crate::parser::Parser;
    use std::io::{Read, Write};
    use std::net::TcpListener;
    use std::sync::{Arc, Mutex};
    use std::time::Instant;

    #[test]
    fn parser_handles_multiline_custom_id_retry_comments_and_crlf() {
        let mut parser = SseParser::new(String::new());
        let out = parser
            .push_bytes(b": ignored\r\nevent: foo\rdata: one\ndata: two\nid: 7\nretry: 25\n\n");
        assert_eq!(out[0], SseParserOutput::Retry(25));
        assert_eq!(
            out[1],
            SseParserOutput::Event(SseEvent {
                event_type: "foo".to_string(),
                data: "one\ntwo".to_string(),
                last_event_id: "7".to_string(),
            })
        );
    }

    #[test]
    fn parser_ignores_nul_id_and_invalid_retry() {
        let mut parser = SseParser::new("old".to_string());
        let out = parser.push_bytes(b"id: bad\0id\nretry: -1\nretry: abc\ndata: x\n\n");
        assert_eq!(
            out,
            vec![SseParserOutput::Event(SseEvent {
                event_type: "message".to_string(),
                data: "x".to_string(),
                last_event_id: "old".to_string(),
            })]
        );
    }

    #[test]
    fn parser_empty_id_clears_last_event_id() {
        let mut parser = SseParser::new("old".to_string());
        let out = parser.push_bytes(b"id:\ndata: x\n\n");
        assert_eq!(
            out,
            vec![SseParserOutput::Event(SseEvent {
                event_type: "message".to_string(),
                data: "x".to_string(),
                last_event_id: String::new(),
            })]
        );
    }

    #[test]
    fn eventsource_connects_dispatches_and_reconnects_with_last_event_id() {
        reset_eventsource_registry();
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let seen_last_id = Arc::new(Mutex::new(false));
        let seen_clone = seen_last_id.clone();

        let server = thread::spawn(move || {
            for request_index in 0..2 {
                let (mut stream, _) = listener.accept().unwrap();
                let mut request = [0; 2048];
                let n = stream.read(&mut request).unwrap();
                let request = String::from_utf8_lossy(&request[..n]);
                if request_index == 1 && request.contains("Last-Event-ID: 42") {
                    *seen_clone.lock().unwrap() = true;
                }
                let body = if request_index == 0 {
                    "retry: 10\nid: 42\nevent: foo\ndata: first\n\n"
                } else {
                    "data: second\n\n"
                };
                write!(
                    stream,
                    "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n{body}"
                )
                .unwrap();
            }
        });

        let mut vm = Vm::new();
        init_eventsource_api(&mut vm);
        let source = format!(
            r#"
            var es = new EventSource("http://{addr}/events");
            es.events = "";
            es.addEventListener("foo", function(e) {{ es.events = es.events + e.type + ":" + e.data + ":" + e.lastEventId + ";"; }});
            es.onmessage = function(e) {{ es.events = es.events + e.type + ":" + e.data + ":" + e.lastEventId + ";"; es.close(); }};
            "#
        );
        let program = Parser::parse(&source).unwrap();
        let func = compiler::compile(&program).unwrap();
        vm.execute(&func).unwrap();

        let deadline = Instant::now() + Duration::from_secs(2);
        while Instant::now() < deadline {
            vm.pump_event_loop().unwrap();
            let done = match vm.get_global("es") {
                Some(Value::Object(es)) => match vm.gc.get(*es) {
                    Some(HeapObject::Object(data)) => data
                        .get_property("events", &vm.shapes)
                        .map(|p| p.value.to_js_string(&vm.gc) == "foo:first:42;message:second:42;")
                        .unwrap_or(false),
                    _ => false,
                },
                _ => false,
            };
            if done {
                break;
            }
            thread::sleep(Duration::from_millis(5));
        }

        vm.pump_event_loop().unwrap();
        server.join().unwrap();
        assert!(*seen_last_id.lock().unwrap());
        match vm.get_global("es") {
            Some(Value::Object(es)) => match vm.gc.get(*es) {
                Some(HeapObject::Object(data)) => {
                    let events = data
                        .get_property("events", &vm.shapes)
                        .map(|p| p.value.to_js_string(&vm.gc))
                        .unwrap_or_default();
                    assert_eq!(events, "foo:first:42;message:second:42;");
                }
                _ => panic!("es object missing"),
            },
            _ => panic!("es global missing"),
        }
    }
}