openrtc 0.2.1

//! WebChannel protocol client for Firestore Listen.
//!
//! Implements Google's WebChannel framing protocol to stream real-time document
//! changes from Firestore. Works on both native (reqwest) and WASM (web-sys fetch).

use crate::logging::log_warn;
use anyhow::{Context, Result};
use std::sync::{Arc, Mutex};

/// Percent-encode a string for use in URL query parameters.
fn url_encode(s: &str) -> String {
    s.bytes()
        .flat_map(|b| match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                vec![b as char]
            }
            _ => format!("%{:02X}", b).chars().collect::<Vec<_>>(),
        })
        .collect()
}

/// Percent-encode for form body values (space → +, others → %XX).
fn form_encode(s: &str) -> String {
    s.bytes()
        .flat_map(|b| match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'*' => {
                vec![b as char]
            }
            b' ' => vec!['+'],
            _ => format!("%{:02X}", b).chars().collect::<Vec<_>>(),
        })
        .collect()
}

const WEBCHANNEL_BASE_PROD: &str =
    "https://firestore.googleapis.com/google.firestore.v1.Firestore/Listen/channel";

#[cfg(target_arch = "wasm32")]
fn wasm_global_string(key: &str) -> Option<String> {
    let global = js_sys::global();
    js_sys::Reflect::get(&global, &wasm_bindgen::JsValue::from_str(key))
        .ok()
        .and_then(|v| v.as_string())
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
}

fn webchannel_base() -> String {
    #[cfg(target_arch = "wasm32")]
    {
        if let Some(host) = wasm_global_string("__OPENRTC_FIRESTORE_EMULATOR_HOST__") {
            return format!(
                "http://{}/google.firestore.v1.Firestore/Listen/channel",
                host
            );
        }
    }
    if let Ok(host) = std::env::var("OPENRTC_FIRESTORE_EMULATOR_HOST") {
        let host = host.trim();
        if !host.is_empty() {
            return format!(
                "http://{}/google.firestore.v1.Firestore/Listen/channel",
                host
            );
        }
    }
    WEBCHANNEL_BASE_PROD.to_string()
}
const WEBCHANNEL_VERSION: u8 = 8;
const WEBCHANNEL_CVER: u8 = 22;

/// Events emitted by the Firestore Listen stream.
#[derive(Debug, Clone)]
pub(crate) enum ListenEvent {
    DocumentChange {
        document: serde_json::Value,
        /// Required by the WebChannel wire format even though the current device
        /// listener only tracks a single target and does not read per-target IDs.
        #[allow(dead_code)]
        target_ids: Vec<i32>,
    },
    DocumentDelete {
        document_path: String,
        /// Required by the WebChannel wire format even though remove handling
        /// only needs the document path for the current listener implementation.
        #[allow(dead_code)]
        target_ids: Vec<i32>,
    },
    DocumentRemove {
        document_path: String,
        /// Required by the WebChannel wire format even though remove handling
        /// only needs the document path for the current listener implementation.
        #[allow(dead_code)]
        target_ids: Vec<i32>,
    },
    TargetChange {
        target_change_type: String,
        target_ids: Vec<i32>,
        resume_token: Option<String>,
        cause: Option<ListenError>,
    },
    Filter {
        /// Required by the WebChannel wire format; retained so filter frames can
        /// be decoded without losing protocol fidelity.
        #[allow(dead_code)]
        target_id: i32,
        /// Required by the WebChannel wire format; retained so filter frames can
        /// be decoded without losing protocol fidelity.
        #[allow(dead_code)]
        count: i32,
    },
}

#[derive(Debug, Clone)]
pub(crate) struct ListenError {
    pub code: i32,
    pub message: String,
}

/// Active WebChannel session state.
pub(crate) struct WebChannelSession {
    sid: String,
    gsession_id: String,
    next_rid: u32,
    last_aid: i64,
    backward_channel_counter: u32,
}

impl WebChannelSession {
    pub(crate) fn short_sid(&self) -> String {
        format!(
            "{}...{}",
            &self.sid[..self.sid.len().min(4)],
            &self.sid[self.sid.len().saturating_sub(4)..]
        )
    }
}

/// WebChannel client for Firestore Listen.
pub(crate) struct WebChannelClient {
    project_id: String,
    database_path: String,
    token_provider: Arc<Mutex<Box<dyn Fn() -> Option<String> + Send + Sync>>>,
    #[cfg(not(target_arch = "wasm32"))]
    http: reqwest::Client,
}

impl WebChannelClient {
    pub fn new(
        project_id: String,
        token_provider: Arc<Mutex<Box<dyn Fn() -> Option<String> + Send + Sync>>>,
    ) -> Self {
        let database_path = format!("projects/{}/databases/(default)", project_id);
        Self {
            project_id,
            database_path,
            token_provider,
            #[cfg(not(target_arch = "wasm32"))]
            http: reqwest::Client::new(),
        }
    }

    fn get_token(&self) -> Option<String> {
        match self.token_provider.lock() {
            Ok(provider) => (provider)(),
            Err(poisoned) => {
                log_warn("[WebChannel] token provider mutex poisoned; recovering inner provider");
                (poisoned.into_inner())()
            }
        }
    }

    fn encoded_database(&self) -> String {
        url_encode(&self.database_path)
    }

    /// Create a new WebChannel session (handshake).
    pub async fn create_session(&self) -> Result<WebChannelSession> {
        let url = format!(
            "{}?VER={}&database={}&RID=0&CVER={}&X-HTTP-Session-Id=gsessionid",
            &webchannel_base(),
            WEBCHANNEL_VERSION,
            self.encoded_database(),
            WEBCHANNEL_CVER,
        );

        let body = "count=0";
        let (response_text, gsession_id) = self.post_form(&url, body).await?;

        // Parse: [[0,["c","<SID>","",8,14,30000]]]
        let parsed: serde_json::Value = serde_json::from_str(&response_text)
            .with_context(|| format!("Failed to parse handshake response: {}", response_text))?;

        let sid = parsed
            .as_array()
            .and_then(|arr| arr.first())
            .and_then(|item| item.as_array())
            .and_then(|pair| pair.get(1))
            .and_then(|data| data.as_array())
            .and_then(|data| data.get(1))
            .and_then(|sid| sid.as_str())
            .context("Missing SID in handshake response")?
            .to_string();

        Ok(WebChannelSession {
            sid,
            gsession_id,
            next_rid: 1,
            last_aid: 0,
            backward_channel_counter: 0,
        })
    }

    /// Send an addTarget request through the forward channel.
    pub async fn add_target(
        &self,
        session: &mut WebChannelSession,
        target_id: i32,
        parent_path: &str,
        collection_id: &str,
        resume_token: Option<&str>,
    ) -> Result<()> {
        let mut listen_req = serde_json::json!({
            "database": self.database_path,
            "addTarget": {
                "query": {
                    "parent": format!(
                        "projects/{}/databases/(default)/documents/{}",
                        self.project_id, parent_path
                    ),
                    "structuredQuery": {
                        "from": [{"collectionId": collection_id}]
                    }
                },
                "targetId": target_id
            }
        });

        if let Some(token) = resume_token {
            listen_req["addTarget"]["resumeToken"] = serde_json::Value::String(token.to_string());
        }

        self.send_forward(session, &listen_req).await
    }

    /// Send a removeTarget request through the forward channel.
    /// Required for full WebChannel protocol coverage even though the current
    /// listener lifecycle relies on session teardown instead of explicit removal.
    #[allow(dead_code)]
    pub async fn remove_target(
        &self,
        session: &mut WebChannelSession,
        target_id: i32,
    ) -> Result<()> {
        let listen_req = serde_json::json!({
            "database": self.database_path,
            "removeTarget": target_id
        });

        self.send_forward(session, &listen_req).await
    }

    /// Poll the backward channel for events. Blocks until events arrive or timeout.
    pub async fn poll_backward(&self, session: &mut WebChannelSession) -> Result<Vec<ListenEvent>> {
        let url = format!(
            "{}?VER={}&database={}&gsessionid={}&SID={}&RID=rpc&AID={}&CI=0&TYPE=xmlhttp&t={}",
            &webchannel_base(),
            WEBCHANNEL_VERSION,
            self.encoded_database(),
            url_encode(&session.gsession_id),
            url_encode(&session.sid),
            session.last_aid,
            session.backward_channel_counter,
        );
        session.backward_channel_counter += 1;

        #[cfg(not(target_arch = "wasm32"))]
        let frames = {
            let response_text = self.get_with_auth(&url).await?;
            parse_framed_response(&response_text)?
        };

        #[cfg(target_arch = "wasm32")]
        let frames = self.get_streaming_frames_with_auth(&url).await?;

        let mut events = Vec::new();

        for (aid, messages) in frames {
            if aid > session.last_aid {
                session.last_aid = aid;
            }
            for msg in messages {
                if let Some(event) = parse_listen_event(&msg) {
                    events.push(event);
                }
            }
        }

        Ok(events)
    }

    /// Send a message through the forward channel.
    async fn send_forward(
        &self,
        session: &mut WebChannelSession,
        data: &serde_json::Value,
    ) -> Result<()> {
        let url = format!(
            "{}?VER={}&database={}&gsessionid={}&SID={}&RID={}&AID={}&CI=0",
            &webchannel_base(),
            WEBCHANNEL_VERSION,
            self.encoded_database(),
            url_encode(&session.gsession_id),
            url_encode(&session.sid),
            session.next_rid,
            session.last_aid,
        );
        session.next_rid += 1;

        let data_str = serde_json::to_string(data)?;
        let body = format!("count=1&ofs=0&req0___data__={}", form_encode(&data_str));

        let (response_text, _) = self.post_form(&url, &body).await?;
        if let Ok(frames) = parse_framed_response(&response_text) {
            for (aid, _) in frames {
                if aid > session.last_aid {
                    session.last_aid = aid;
                }
            }
        }
        Ok(())
    }

    // ── Platform-specific HTTP methods ──────────────────────────────────

    #[cfg(not(target_arch = "wasm32"))]
    async fn post_form(&self, url: &str, body: &str) -> Result<(String, String)> {
        let mut req = self
            .http
            .post(url)
            .header("Content-Type", "application/x-www-form-urlencoded")
            .body(body.to_string());

        if let Some(token) = self.get_token() {
            req = req.header("Authorization", format!("Bearer {}", token));
        }

        let res = req.send().await.context("WebChannel POST failed")?;

        let gsession_id = res
            .headers()
            .get("x-http-session-id")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("")
            .to_string();

        if !res.status().is_success() {
            let status = res.status();
            let text = res.text().await.unwrap_or_default();
            anyhow::bail!("WebChannel POST {} returned {}: {}", url, status, text);
        }

        let text = res.text().await.context("Failed to read POST response")?;
        // Strip length prefix if present
        let stripped = strip_length_prefix(&text);
        Ok((stripped.to_string(), gsession_id))
    }

    #[cfg(target_arch = "wasm32")]
    async fn post_form(&self, url: &str, body: &str) -> Result<(String, String)> {
        use wasm_bindgen::JsCast;
        use wasm_bindgen_futures::JsFuture;

        let opts = web_sys::RequestInit::new();
        opts.set_method("POST");
        opts.set_body(&wasm_bindgen::JsValue::from_str(body));

        let headers =
            web_sys::Headers::new().map_err(|e| anyhow::anyhow!("Headers::new failed: {:?}", e))?;
        headers
            .set("Content-Type", "application/x-www-form-urlencoded")
            .map_err(|e| anyhow::anyhow!("headers.set failed: {:?}", e))?;

        if let Some(token) = self.get_token() {
            headers
                .set("Authorization", &format!("Bearer {}", token))
                .map_err(|e| anyhow::anyhow!("headers.set auth failed: {:?}", e))?;
        }

        opts.set_headers(&headers);

        let request = web_sys::Request::new_with_str_and_init(url, &opts)
            .map_err(|e| anyhow::anyhow!("Request::new failed: {:?}", e))?;

        let window = web_sys::window().context("no window")?;
        let resp_value = JsFuture::from(window.fetch_with_request(&request))
            .await
            .map_err(|e| anyhow::anyhow!("fetch failed: {:?}", e))?;

        let resp: web_sys::Response = resp_value
            .dyn_into()
            .map_err(|_| anyhow::anyhow!("response is not a Response"))?;

        let gsession_id = resp
            .headers()
            .get("x-http-session-id")
            .ok()
            .flatten()
            .unwrap_or_default();

        if !resp.ok() {
            let text = JsFuture::from(
                resp.text()
                    .map_err(|e| anyhow::anyhow!("text() failed: {:?}", e))?,
            )
            .await
            .map_err(|e| anyhow::anyhow!("text await failed: {:?}", e))?
            .as_string()
            .unwrap_or_default();
            anyhow::bail!("WebChannel POST returned {}: {}", resp.status(), text);
        }

        let text = JsFuture::from(
            resp.text()
                .map_err(|e| anyhow::anyhow!("text() failed: {:?}", e))?,
        )
        .await
        .map_err(|e| anyhow::anyhow!("text await failed: {:?}", e))?
        .as_string()
        .unwrap_or_default();

        let stripped = strip_length_prefix(&text);
        Ok((stripped.to_string(), gsession_id))
    }

    /// Native GET fallback retained for parity with the WASM implementation and
    /// targeted diagnostics, even though the streaming path is preferred today.
    #[cfg(not(target_arch = "wasm32"))]
    #[allow(dead_code)]
    async fn get_with_auth(&self, url: &str) -> Result<String> {
        let mut req = self.http.get(url);

        if let Some(token) = self.get_token() {
            req = req.header("Authorization", format!("Bearer {}", token));
        }

        let res = req.send().await.context("WebChannel GET failed")?;

        if res.status() == reqwest::StatusCode::UNAUTHORIZED {
            anyhow::bail!("WebChannel: unauthorized (401)");
        }

        if !res.status().is_success() {
            let status = res.status();
            let text = res.text().await.unwrap_or_default();
            anyhow::bail!("WebChannel GET returned {}: {}", status, text);
        }

        res.text().await.context("Failed to read GET response")
    }

    /// WASM GET fallback retained for parity with the native implementation and
    /// targeted diagnostics, even though the streaming path is preferred today.
    #[cfg(target_arch = "wasm32")]
    #[allow(dead_code)]
    async fn get_with_auth(&self, url: &str) -> Result<String> {
        use wasm_bindgen::JsCast;
        use wasm_bindgen_futures::JsFuture;

        let opts = web_sys::RequestInit::new();
        opts.set_method("GET");

        let headers =
            web_sys::Headers::new().map_err(|e| anyhow::anyhow!("Headers::new failed: {:?}", e))?;

        if let Some(token) = self.get_token() {
            headers
                .set("Authorization", &format!("Bearer {}", token))
                .map_err(|e| anyhow::anyhow!("headers.set auth failed: {:?}", e))?;
        }

        opts.set_headers(&headers);

        let request = web_sys::Request::new_with_str_and_init(url, &opts)
            .map_err(|e| anyhow::anyhow!("Request::new failed: {:?}", e))?;

        let window = web_sys::window().context("no window")?;
        let resp_value = JsFuture::from(window.fetch_with_request(&request))
            .await
            .map_err(|e| anyhow::anyhow!("fetch failed: {:?}", e))?;

        let resp: web_sys::Response = resp_value
            .dyn_into()
            .map_err(|_| anyhow::anyhow!("response is not a Response"))?;

        if resp.status() == 401 {
            anyhow::bail!("WebChannel: unauthorized (401)");
        }

        if !resp.ok() {
            let text = JsFuture::from(
                resp.text()
                    .map_err(|e| anyhow::anyhow!("text() failed: {:?}", e))?,
            )
            .await
            .map_err(|e| anyhow::anyhow!("text await failed: {:?}", e))?
            .as_string()
            .unwrap_or_default();
            anyhow::bail!("WebChannel GET returned {}: {}", resp.status(), text);
        }

        let text = JsFuture::from(
            resp.text()
                .map_err(|e| anyhow::anyhow!("text() failed: {:?}", e))?,
        )
        .await
        .map_err(|e| anyhow::anyhow!("text await failed: {:?}", e))?
        .as_string()
        .unwrap_or_default();

        Ok(text)
    }

    #[cfg(target_arch = "wasm32")]
    async fn get_streaming_frames_with_auth(
        &self,
        url: &str,
    ) -> Result<Vec<(i64, Vec<serde_json::Value>)>> {
        use wasm_bindgen::JsCast;
        use wasm_bindgen_futures::JsFuture;

        let opts = web_sys::RequestInit::new();
        opts.set_method("GET");

        let headers =
            web_sys::Headers::new().map_err(|e| anyhow::anyhow!("Headers::new failed: {:?}", e))?;

        if let Some(token) = self.get_token() {
            headers
                .set("Authorization", &format!("Bearer {}", token))
                .map_err(|e| anyhow::anyhow!("headers.set auth failed: {:?}", e))?;
        }

        opts.set_headers(&headers);

        let request = web_sys::Request::new_with_str_and_init(url, &opts)
            .map_err(|e| anyhow::anyhow!("Request::new failed: {:?}", e))?;

        let window = web_sys::window().context("no window")?;
        let resp_value = JsFuture::from(window.fetch_with_request(&request))
            .await
            .map_err(|e| anyhow::anyhow!("fetch failed: {:?}", e))?;

        let resp: web_sys::Response = resp_value
            .dyn_into()
            .map_err(|_| anyhow::anyhow!("response is not a Response"))?;

        if resp.status() == 401 {
            anyhow::bail!("WebChannel: unauthorized (401)");
        }

        if !resp.ok() {
            let text = JsFuture::from(
                resp.text()
                    .map_err(|e| anyhow::anyhow!("text() failed: {:?}", e))?,
            )
            .await
            .map_err(|e| anyhow::anyhow!("text await failed: {:?}", e))?
            .as_string()
            .unwrap_or_default();
            anyhow::bail!("WebChannel GET returned {}: {}", resp.status(), text);
        }

        let body = resp
            .body()
            .ok_or_else(|| anyhow::anyhow!("WebChannel GET response missing body"))?;
        let reader = body
            .get_reader()
            .dyn_into::<web_sys::ReadableStreamDefaultReader>()
            .map_err(|_| anyhow::anyhow!("failed to acquire ReadableStreamDefaultReader"))?;

        let mut buffer = Vec::<u8>::new();

        loop {
            let chunk = JsFuture::from(reader.read())
                .await
                .map_err(|e| anyhow::anyhow!("reader.read failed: {:?}", e))?;
            let done = js_sys::Reflect::get(&chunk, &wasm_bindgen::JsValue::from_str("done"))
                .map_err(|e| anyhow::anyhow!("Reflect.get(done) failed: {:?}", e))?
                .as_bool()
                .unwrap_or(false);

            let value = js_sys::Reflect::get(&chunk, &wasm_bindgen::JsValue::from_str("value"))
                .map_err(|e| anyhow::anyhow!("Reflect.get(value) failed: {:?}", e))?;

            if !value.is_undefined() && !value.is_null() {
                let bytes = js_sys::Uint8Array::new(&value);
                let mut chunk_bytes = vec![0u8; bytes.length() as usize];
                bytes.copy_to(&mut chunk_bytes);
                buffer.extend_from_slice(&chunk_bytes);

                if let Some((_, frames)) = parse_length_prefixed_frame_batches(&buffer)? {
                    let _ = JsFuture::from(reader.cancel()).await;
                    return Ok(frames);
                }
            }

            if done {
                break;
            }
        }

        if let Some((_, frames)) = parse_length_prefixed_frame_batches(&buffer)? {
            return Ok(frames);
        }

        if buffer.is_empty() {
            return Ok(Vec::new());
        }

        let text = String::from_utf8(buffer)
            .map_err(|e| anyhow::anyhow!("WebChannel body was not valid UTF-8: {}", e))?;
        parse_framed_response(&text)
    }
}

// ── Frame parsing ──────────────────────────────────────────────────────

/// Strip the optional length prefix from a WebChannel response.
/// Format: `<decimal_length>\n<json_payload>`
fn strip_length_prefix(text: &str) -> &str {
    let trimmed = text.trim();
    if let Some(newline_pos) = trimmed.find('\n') {
        let prefix = &trimmed[..newline_pos];
        if prefix.chars().all(|c| c.is_ascii_digit()) {
            return &trimmed[newline_pos + 1..];
        }
    }
    trimmed
}

fn parse_length_prefixed_frame_batches(
    buffer: &[u8],
) -> Result<Option<(usize, Vec<(i64, Vec<serde_json::Value>)>)>> {
    let mut cursor = 0usize;
    let mut all_frames = Vec::new();

    loop {
        while cursor < buffer.len() && buffer[cursor].is_ascii_whitespace() {
            cursor += 1;
        }

        if cursor >= buffer.len() {
            break;
        }

        if buffer[cursor] == b'[' || buffer[cursor] == b'{' {
            // Consume exactly one JSON frame here. Firebase sometimes emits
            // adjacent `<json><length>\n<json>` sequences in a single chunk,
            // so we alternate between JSON and length-prefixed blocks instead
            // of handing the remainder to a streaming deserializer (which
            // would choke on the intermediate decimal length prefix).
            match parse_one_json_frame_prefix(&buffer[cursor..])? {
                Some((consumed, frames)) => {
                    all_frames.extend(frames);
                    cursor += consumed;
                    continue;
                }
                None => break,
            }
        }

        let len_start = cursor;
        while cursor < buffer.len() && buffer[cursor].is_ascii_digit() {
            cursor += 1;
        }

        if cursor == len_start {
            anyhow::bail!("Unexpected WebChannel framing byte: {}", buffer[cursor]);
        }

        if cursor >= buffer.len() {
            break;
        }

        if buffer[cursor] != b'\n' {
            anyhow::bail!(
                "Invalid WebChannel frame delimiter: expected newline, found {}",
                buffer[cursor]
            );
        }

        let payload_len = std::str::from_utf8(&buffer[len_start..cursor])
            .context("WebChannel frame length was not valid UTF-8")?
            .parse::<usize>()
            .context("WebChannel frame length was not a valid number")?;
        cursor += 1;

        if buffer.len().saturating_sub(cursor) < payload_len {
            cursor = len_start;
            break;
        }

        let payload = std::str::from_utf8(&buffer[cursor..cursor + payload_len])
            .context("WebChannel payload was not valid UTF-8")?;
        all_frames.extend(parse_frame_payload(payload)?);
        cursor += payload_len;
    }

    if all_frames.is_empty() {
        Ok(None)
    } else {
        Ok(Some((cursor, all_frames)))
    }
}

fn parse_frame_payload(payload: &str) -> Result<Vec<(i64, Vec<serde_json::Value>)>> {
    let trimmed = payload.trim();
    if trimmed.is_empty() {
        return Ok(Vec::new());
    }

    let mut deserializer =
        serde_json::Deserializer::from_str(trimmed).into_iter::<serde_json::Value>();
    let mut frames = Vec::new();
    let mut saw_value = false;

    while let Some(value) = deserializer.next() {
        let parsed =
            value.with_context(|| format!("Failed to parse WebChannel frame: {}", trimmed))?;
        saw_value = true;
        frames.extend(parse_frame_value(parsed)?);
    }

    if !saw_value {
        anyhow::bail!("Failed to parse WebChannel frame: {}", trimmed);
    }

    Ok(frames)
}

fn parse_frame_value(parsed: serde_json::Value) -> Result<Vec<(i64, Vec<serde_json::Value>)>> {
    let outer = parsed.as_array().context("Frame is not an array")?;
    let mut frames = Vec::new();

    for item in outer {
        let pair = item.as_array().context("Frame item is not an array")?;
        let aid = pair.first().and_then(|v| v.as_i64()).unwrap_or(0);
        let messages = pair
            .get(1)
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();
        frames.push((aid, messages));
    }

    Ok(frames)
}

fn parse_one_json_frame_prefix(
    buffer: &[u8],
) -> Result<Option<(usize, Vec<(i64, Vec<serde_json::Value>)>)>> {
    let text = std::str::from_utf8(buffer).context("WebChannel payload was not valid UTF-8")?;
    let mut stream = serde_json::Deserializer::from_str(text).into_iter::<serde_json::Value>();
    let Some(value) = stream.next() else {
        return Ok(None);
    };

    let parsed = match value {
        Ok(parsed) => parsed,
        Err(err) if err.is_eof() => {
            // JSON is truncated mid-value — treat as "need more bytes" rather
            // than a hard parse failure so streaming readers can buffer more.
            return Ok(None);
        }
        Err(err) => {
            return Err(anyhow::Error::new(err)
                .context(format!("Failed to parse WebChannel frame: {}", text.trim())));
        }
    };
    let consumed = stream.byte_offset();
    let frames = parse_frame_value(parsed)?;
    Ok(Some((consumed, frames)))
}

/// Parse WebChannel frames into (aid, messages) tuples.
/// Supports either a raw JSON payload or one-or-more adjacent
/// `<length>\n<json_payload>` batches in a single response body.
fn parse_framed_response(text: &str) -> Result<Vec<(i64, Vec<serde_json::Value>)>> {
    let stripped = strip_length_prefix(text);
    if stripped.is_empty() {
        return Ok(Vec::new());
    }

    let trimmed = text.trim();
    let bytes = trimmed.as_bytes();
    let mut cursor = 0usize;
    let mut all_frames = Vec::new();

    loop {
        while cursor < bytes.len() && bytes[cursor].is_ascii_whitespace() {
            cursor += 1;
        }

        if cursor >= bytes.len() {
            break;
        }

        let remaining = &bytes[cursor..];

        match remaining[0] {
            b'[' | b'{' => {
                let Some((consumed, frames)) = parse_one_json_frame_prefix(remaining)? else {
                    break;
                };
                all_frames.extend(frames);
                cursor += consumed;
            }
            b'0'..=b'9' => {
                let Some((consumed, frames)) = parse_length_prefixed_frame_batches(remaining)?
                else {
                    anyhow::bail!(
                        "Incomplete length-prefixed WebChannel payload: {}",
                        std::str::from_utf8(remaining).unwrap_or("<non-utf8>")
                    );
                };
                all_frames.extend(frames);
                cursor += consumed;
            }
            other => {
                anyhow::bail!(
                    "Unexpected leading WebChannel byte {} in payload: {}",
                    other,
                    std::str::from_utf8(remaining).unwrap_or("<non-utf8>")
                );
            }
        }
    }

    Ok(all_frames)
}

/// Parse a single ListenResponse JSON value into a ListenEvent.
fn parse_listen_event(value: &serde_json::Value) -> Option<ListenEvent> {
    if let Some(tc) = value.get("targetChange") {
        let change_type = tc
            .get("targetChangeType")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();
        let target_ids = tc
            .get("targetIds")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| v.as_i64().map(|n| n as i32))
                    .collect()
            })
            .unwrap_or_default();
        let resume_token = tc
            .get("resumeToken")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        let cause = tc.get("cause").map(|c| ListenError {
            code: c.get("code").and_then(|v| v.as_i64()).unwrap_or(0) as i32,
            message: c
                .get("message")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string(),
        });
        return Some(ListenEvent::TargetChange {
            target_change_type: change_type,
            target_ids,
            resume_token,
            cause,
        });
    }

    if let Some(dc) = value.get("documentChange") {
        let document = dc
            .get("document")
            .cloned()
            .unwrap_or(serde_json::Value::Null);
        let target_ids = dc
            .get("targetIds")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| v.as_i64().map(|n| n as i32))
                    .collect()
            })
            .unwrap_or_default();
        return Some(ListenEvent::DocumentChange {
            document,
            target_ids,
        });
    }

    if let Some(dd) = value.get("documentDelete") {
        let document_path = dd
            .get("document")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();
        let target_ids = dd
            .get("removedTargetIds")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| v.as_i64().map(|n| n as i32))
                    .collect()
            })
            .unwrap_or_default();
        return Some(ListenEvent::DocumentDelete {
            document_path,
            target_ids,
        });
    }

    if let Some(dr) = value.get("documentRemove") {
        let document_path = dr
            .get("document")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();
        let target_ids = dr
            .get("removedTargetIds")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| v.as_i64().map(|n| n as i32))
                    .collect()
            })
            .unwrap_or_default();
        return Some(ListenEvent::DocumentRemove {
            document_path,
            target_ids,
        });
    }

    if let Some(f) = value.get("filter") {
        let target_id = f.get("targetId").and_then(|v| v.as_i64()).unwrap_or(0) as i32;
        let count = f.get("count").and_then(|v| v.as_i64()).unwrap_or(0) as i32;
        return Some(ListenEvent::Filter { target_id, count });
    }

    None
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_strip_length_prefix() {
        assert_eq!(
            strip_length_prefix("51\n[[0,[\"c\",\"abc\",\"\",8,14,30000]]]"),
            "[[0,[\"c\",\"abc\",\"\",8,14,30000]]]"
        );
        assert_eq!(strip_length_prefix("[]"), "[]");
        assert_eq!(strip_length_prefix("  42\n{\"a\":1}  "), "{\"a\":1}");
    }

    #[test]
    fn test_parse_frames() {
        let input = "[[1,[{\"targetChange\":{\"targetChangeType\":\"ADD\",\"targetIds\":[2]}}]],[2,[{\"targetChange\":{\"targetChangeType\":\"CURRENT\",\"targetIds\":[2]}}]]]";
        let frames = parse_framed_response(input).unwrap();
        assert_eq!(frames.len(), 2);
        assert_eq!(frames[0].0, 1);
        assert_eq!(frames[1].0, 2);
    }

    #[test]
    fn test_parse_listen_event_target_change() {
        let val: serde_json::Value =
            serde_json::from_str(r#"{"targetChange":{"targetChangeType":"ADD","targetIds":[1]}}"#)
                .unwrap();
        let event = parse_listen_event(&val).unwrap();
        match event {
            ListenEvent::TargetChange {
                target_change_type,
                target_ids,
                ..
            } => {
                assert_eq!(target_change_type, "ADD");
                assert_eq!(target_ids, vec![1]);
            }
            _ => assert!(false, "Expected TargetChange"),
        }
    }

    #[test]
    fn test_parse_listen_event_document_change() {
        let val: serde_json::Value = serde_json::from_str(
            r#"{"documentChange":{"document":{"name":"projects/p/databases/(default)/documents/col/doc1","fields":{"a":{"stringValue":"b"}}},"targetIds":[1]}}"#,
        )
        .unwrap();
        let event = parse_listen_event(&val).unwrap();
        match event {
            ListenEvent::DocumentChange {
                document,
                target_ids,
            } => {
                assert_eq!(target_ids, vec![1]);
                assert!(document.get("name").is_some());
            }
            _ => assert!(false, "Expected DocumentChange"),
        }
    }

    #[test]
    fn test_parse_listen_event_document_delete() {
        let val: serde_json::Value = serde_json::from_str(
            r#"{"documentDelete":{"document":"projects/p/databases/(default)/documents/col/doc1","removedTargetIds":[1]}}"#,
        )
        .unwrap();
        let event = parse_listen_event(&val).unwrap();
        match event {
            ListenEvent::DocumentDelete {
                document_path,
                target_ids,
            } => {
                assert!(document_path.ends_with("col/doc1"));
                assert_eq!(target_ids, vec![1]);
            }
            _ => assert!(false, "Expected DocumentDelete"),
        }
    }

    #[test]
    fn test_parse_listen_event_with_error_cause() {
        let val: serde_json::Value = serde_json::from_str(
            r#"{"targetChange":{"targetChangeType":"REMOVE","targetIds":[2],"cause":{"code":7,"message":"Missing or insufficient permissions."}}}"#,
        )
        .unwrap();
        let event = parse_listen_event(&val).unwrap();
        match event {
            ListenEvent::TargetChange { cause, .. } => {
                let cause = cause.unwrap();
                assert_eq!(cause.code, 7);
                assert!(cause.message.contains("permissions"));
            }
            _ => assert!(false, "Expected TargetChange"),
        }
    }

    #[test]
    fn test_parse_frames_with_length_prefix() {
        let payload = r#"[[1,[{"targetChange":{"targetChangeType":"ADD","targetIds":[2]}}]]]"#;
        let input = format!("{}\n{}", payload.len(), payload);
        let frames = parse_framed_response(&input).unwrap();
        assert_eq!(frames.len(), 1);
        assert_eq!(frames[0].0, 1);
    }

    #[test]
    fn test_parse_length_prefixed_frame_batches() {
        let payload = r#"[[1,[{"targetChange":{"targetChangeType":"ADD","targetIds":[2]}}]]]"#;
        let input = format!("{}\n{}", payload.len(), payload);
        let parsed = parse_length_prefixed_frame_batches(input.as_bytes())
            .unwrap()
            .unwrap();
        assert_eq!(parsed.1.len(), 1);
        assert_eq!(parsed.1[0].0, 1);
    }

    #[test]
    fn test_parse_length_prefixed_frame_batches_waits_for_complete_payload() {
        let partial = b"70\n[[1,[{\"targetChange\":{\"targetChangeType\":\"ADD\"";
        assert!(parse_length_prefixed_frame_batches(partial)
            .unwrap()
            .is_none());
    }

    #[test]
    fn test_parse_length_prefixed_frame_batches_mixed_raw_then_prefixed() {
        // Streamed body where a raw JSON array is immediately followed by a
        // length-prefixed frame in the same chunk. Prior to the fix, the
        // streaming WASM path would bail with "Failed to parse WebChannel
        // frame" because serde_json choked on the intermediate decimal length
        // prefix.
        let payload_one = r#"[[1,[{"targetChange":{"targetChangeType":"ADD","targetIds":[2]}}]]]"#;
        let payload_two = r#"[[2,[{"documentChange":{"document":{"name":"projects/p/databases/(default)/documents/col/doc1"}}}]]]"#;
        let input = format!("{}{}\n{}", payload_one, payload_two.len(), payload_two);
        let parsed = parse_length_prefixed_frame_batches(input.as_bytes())
            .unwrap()
            .unwrap();
        assert_eq!(parsed.1.len(), 2);
        assert_eq!(parsed.1[0].0, 1);
        assert_eq!(parsed.1[1].0, 2);
    }

    #[test]
    fn test_parse_length_prefixed_frame_batches_multiple_raw_adjacent() {
        let payload_one = r#"[[5,[{"documentChange":{"document":{"name":"projects/p/databases/(default)/documents/apps/app/devices/device-1"}}}]]]"#;
        let payload_two =
            r#"[[6,[{"targetChange":{"targetChangeType":"CURRENT","targetIds":[1]}}]]]"#;
        let input = format!("{}{}", payload_one, payload_two);
        let parsed = parse_length_prefixed_frame_batches(input.as_bytes())
            .unwrap()
            .unwrap();
        assert_eq!(parsed.1.len(), 2);
        assert_eq!(parsed.1[0].0, 5);
        assert_eq!(parsed.1[1].0, 6);
    }

    #[test]
    fn test_parse_framed_response_with_multiple_adjacent_length_prefixed_batches() {
        let payload_one = r#"[[1,[{"targetChange":{"targetChangeType":"ADD","targetIds":[2]}}]]]"#;
        let payload_two = r#"[[2,[{"documentChange":{"document":{"name":"projects/p/databases/(default)/documents/col/doc1"}}}]]]"#;
        let input = format!(
            "{}\n{}{}\n{}",
            payload_one.len(),
            payload_one,
            payload_two.len(),
            payload_two
        );

        let frames = parse_framed_response(&input).unwrap();
        assert_eq!(frames.len(), 2);
        assert_eq!(frames[0].0, 1);
        assert_eq!(frames[1].0, 2);
    }

    #[test]
    fn test_parse_framed_response_with_multiple_adjacent_raw_json_arrays() {
        let payload_one = r#"[[5,[{"documentChange":{"document":{"name":"projects/p/databases/(default)/documents/apps/app/devices/device-1"}}}]]]"#;
        let payload_two =
            r#"[[6,[{"targetChange":{"targetChangeType":"CURRENT","targetIds":[1]}}]]]"#;
        let input = format!("{}{}", payload_one, payload_two);

        let frames = parse_framed_response(&input).unwrap();
        assert_eq!(frames.len(), 2);
        assert_eq!(frames[0].0, 5);
        assert_eq!(frames[1].0, 6);
    }

    #[test]
    fn test_parse_framed_response_with_raw_frame_then_length_prefixed_batch() {
        let payload_one = r#"[[5,[{"documentChange":{"document":{"name":"projects/p/databases/(default)/documents/apps/app/devices/device-1"}}}]]]"#;
        let payload_two =
            r#"[[6,[{"targetChange":{"targetChangeType":"CURRENT","targetIds":[1]}}]]]"#;
        let input = format!("{}{}\n{}", payload_one, payload_two.len(), payload_two);

        let frames = parse_framed_response(&input).unwrap();
        assert_eq!(frames.len(), 2);
        assert_eq!(frames[0].0, 5);
        assert_eq!(frames[1].0, 6);
    }
}