forge_dioxus/

client.rs

use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::rc::Rc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use dioxus::prelude::{Signal, WritableExt, dioxus_core::Task};
use serde::Serialize;
use serde::de::DeserializeOwned;

use crate::signals::ForgeSignals;
use crate::types::{
    ConnectionState, ForgeClientError, ForgeError, RpcEnvelopeRaw, StreamEvent,
};

type TokenProvider = Rc<dyn Fn() -> Option<String>>;
type AuthErrorHandler = Rc<dyn Fn(ForgeError)>;
type EventSender = futures_channel::mpsc::UnboundedSender<SseDispatch>;
type ConnectWaiter = futures_channel::oneshot::Sender<Result<(), ForgeClientError>>;

static NEXT_SUBSCRIPTION_ID: AtomicU64 = AtomicU64::new(1);

enum SseDispatch {
    Data(serde_json::Value),
    Error { code: String, message: String },
}

struct RegistrationMeta {
    endpoint: &'static str,
    payload: serde_json::Value,
}

#[derive(Default, Clone, Copy, PartialEq, Eq)]
enum SseState {
    #[default]
    Idle,
    Connecting,
    Connected,
}

/// Shared SSE connection state, one per ForgeClient.
#[derive(Default)]
struct SseManager {
    session_id: Option<String>,
    session_secret: Option<String>,
    state: SseState,
    ever_connected: bool,
    listeners: HashMap<String, EventSender>,
    registrations: HashMap<String, RegistrationMeta>,
    event_loop_task: Option<Task>,
    reconnect_attempts: u32,
    connect_waiters: Vec<ConnectWaiter>,
}

const MAX_RECONNECT_ATTEMPTS: u32 = 10;

#[derive(Clone)]
pub struct ForgeClientConfig {
    pub url: String,
    pub get_token: Option<TokenProvider>,
    pub on_auth_error: Option<AuthErrorHandler>,
    pub(crate) connection_state: Option<Signal<ConnectionState>>,
}

impl ForgeClientConfig {
    pub fn new(url: impl Into<String>) -> Self {
        Self {
            url: url.into(),
            get_token: None,
            on_auth_error: None,
            connection_state: None,
        }
    }

    pub fn with_token_provider(mut self, provider: impl Fn() -> Option<String> + 'static) -> Self {
        self.get_token = Some(Rc::new(provider));
        self
    }

    pub fn with_auth_error_handler(
        mut self,
        handler: impl Fn(ForgeError) + 'static,
    ) -> Self {
        self.on_auth_error = Some(Rc::new(handler));
        self
    }

    pub(crate) fn with_connection_state(mut self, state: Signal<ConnectionState>) -> Self {
        self.connection_state = Some(state);
        self
    }
}

#[derive(Clone)]
pub struct ForgeClient {
    inner: Rc<ForgeClientInner>,
}

struct ForgeClientInner {
    url: String,
    get_token: Option<TokenProvider>,
    on_auth_error: Option<AuthErrorHandler>,
    connection_state: Option<Signal<ConnectionState>>,
    sse: RefCell<SseManager>,
    signals: RefCell<Option<ForgeSignals>>,
}

impl ForgeClient {
    pub fn new(config: ForgeClientConfig) -> Self {
        Self {
            inner: Rc::new(ForgeClientInner {
                url: config.url.trim_end_matches('/').to_string(),
                get_token: config.get_token,
                on_auth_error: config.on_auth_error,
                connection_state: config.connection_state,
                sse: RefCell::new(SseManager::default()),
                signals: RefCell::new(None),
            }),
        }
    }

    /// Wire signals for correlation ID injection on RPC calls.
    pub fn set_signals(&self, signals: ForgeSignals) {
        *self.inner.signals.borrow_mut() = Some(signals);
    }

    /// Get the base URL of this client.
    pub fn get_url(&self) -> &str {
        &self.inner.url
    }

    /// Generate a correlation ID from the wired signals instance, if any.
    fn correlation_id(&self) -> Option<String> {
        self.inner.signals.borrow().as_ref().map(|s| s.next_correlation_id())
    }

    pub async fn call<TArgs, TResult>(
        &self,
        function_name: &str,
        args: TArgs,
    ) -> Result<TResult, ForgeClientError>
    where
        TArgs: Serialize,
        TResult: DeserializeOwned,
    {
        let body = serde_json::json!({ "args": args });
        let correlation_id = self.correlation_id();
        let envelope = platform::request_json(
            self,
            &format!("{}/_api/rpc/{}", self.inner.url, function_name),
            body,
            correlation_id.as_deref(),
        )
        .await?;
        self.decode_envelope(envelope)
    }

    #[cfg(target_arch = "wasm32")]
    pub async fn call_multipart<TResult>(
        &self,
        function_name: &str,
        form: web_sys::FormData,
    ) -> Result<TResult, ForgeClientError>
    where
        TResult: DeserializeOwned,
    {
        let correlation_id = self.correlation_id();
        let envelope = platform::request_multipart(
            self,
            &format!("{}/_api/rpc/{}/upload", self.inner.url, function_name),
            form,
            correlation_id.as_deref(),
        )
        .await?;
        self.decode_envelope(envelope)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub async fn call_multipart<TResult>(
        &self,
        function_name: &str,
        form: reqwest::multipart::Form,
    ) -> Result<TResult, ForgeClientError>
    where
        TResult: DeserializeOwned,
    {
        let correlation_id = self.correlation_id();
        let envelope = platform::request_multipart(
            self,
            &format!("{}/_api/rpc/{}/upload", self.inner.url, function_name),
            form,
            correlation_id.as_deref(),
        )
        .await?;
        self.decode_envelope(envelope)
    }

    pub fn subscribe_query<TArgs, TResult, F>(
        &self,
        function_name: &str,
        args: TArgs,
        callback: F,
    ) -> SubscriptionHandle
    where
        TArgs: Serialize + Clone + 'static,
        TResult: DeserializeOwned + Clone + 'static,
        F: FnMut(StreamEvent<TResult>) + 'static,
    {
        let sub_id = self.random_id("sub");
        let target = format!("sub:{sub_id}");

        let (tx, rx) = futures_channel::mpsc::unbounded::<SseDispatch>();
        self.inner.sse.borrow_mut().listeners.insert(target.clone(), tx);

        let args_value = serde_json::to_value(&args).unwrap_or(serde_json::Value::Null);
        let reg_payload = serde_json::json!({
            "id": sub_id,
            "function": function_name,
            "args": args_value,
        });
        self.inner.sse.borrow_mut().registrations.insert(
            sub_id.clone(),
            RegistrationMeta {
                endpoint: "/_api/subscribe",
                payload: reg_payload,
            },
        );

        self.spawn_subscription(sub_id, target, rx, callback, |client, envelope, cb| {
            match client.decode_envelope::<TResult>(envelope) {
                Ok(data) => cb(StreamEvent::Data(data)),
                Err(err) => cb(StreamEvent::Error(err)),
            }
        })
    }

    pub fn subscribe_job<TResult, F>(&self, job_id: String, callback: F) -> SubscriptionHandle
    where
        TResult: DeserializeOwned + Clone + 'static,
        F: FnMut(StreamEvent<TResult>) + 'static,
    {
        self.subscribe_tracker("job", serde_json::json!({ "job_id": job_id }), "/_api/subscribe-job", callback)
    }

    pub fn subscribe_workflow<TResult, F>(
        &self,
        workflow_id: String,
        callback: F,
    ) -> SubscriptionHandle
    where
        TResult: DeserializeOwned + Clone + 'static,
        F: FnMut(StreamEvent<TResult>) + 'static,
    {
        self.subscribe_tracker(
            "wf",
            serde_json::json!({ "workflow_id": workflow_id }),
            "/_api/subscribe-workflow",
            callback,
        )
    }

    fn subscribe_tracker<TResult, F>(
        &self,
        prefix: &str,
        payload: serde_json::Value,
        endpoint: &'static str,
        callback: F,
    ) -> SubscriptionHandle
    where
        TResult: DeserializeOwned + Clone + 'static,
        F: FnMut(StreamEvent<TResult>) + 'static,
    {
        let sub_id = self.random_id(prefix);
        let target = format!("{prefix}:{sub_id}");

        let (tx, rx) = futures_channel::mpsc::unbounded::<SseDispatch>();
        self.inner.sse.borrow_mut().listeners.insert(target.clone(), tx);

        let mut reg_payload = payload;
        reg_payload
            .as_object_mut()
            .expect("tracker payload must be an object")
            .insert("id".to_string(), serde_json::Value::String(sub_id.clone()));
        self.inner.sse.borrow_mut().registrations.insert(
            sub_id.clone(),
            RegistrationMeta {
                endpoint,
                payload: reg_payload,
            },
        );

        self.spawn_subscription(sub_id, target, rx, callback, |_client, envelope, cb| {
            if envelope.success {
                if let Some(data) = envelope.data {
                    match serde_json::from_value::<TResult>(data) {
                        Ok(parsed) => cb(StreamEvent::Data(parsed)),
                        Err(e) => cb(StreamEvent::Error(ForgeClientError::new(
                            "DESERIALIZATION_ERROR",
                            e.to_string(),
                            None,
                        ))),
                    }
                }
            }
        })
    }

    fn spawn_subscription<TResult, F>(
        &self,
        sub_id: String,
        target: String,
        mut rx: futures_channel::mpsc::UnboundedReceiver<SseDispatch>,
        mut callback: F,
        on_initial: impl FnOnce(&ForgeClient, RpcEnvelopeRaw, &mut F) + 'static,
    ) -> SubscriptionHandle
    where
        TResult: DeserializeOwned + Clone + 'static,
        F: FnMut(StreamEvent<TResult>) + 'static,
    {
        let client = self.clone();
        let handle = SubscriptionHandle::new(sub_id.clone(), target, self.clone());
        let handle_task = handle.clone();

        let task = dioxus::prelude::spawn(async move {
            callback(StreamEvent::Connection(ConnectionState::Connecting));

            if let Err(e) = client.ensure_connected().await {
                callback(StreamEvent::Error(e));
                callback(StreamEvent::Connection(ConnectionState::Disconnected));
                handle_task.finish();
                return;
            }

            match client.register_subscription(&sub_id).await {
                Ok(envelope) => {
                    callback(StreamEvent::Connection(ConnectionState::Connected));
                    on_initial(&client, envelope, &mut callback);
                }
                Err(err) => {
                    callback(StreamEvent::Error(err));
                    callback(StreamEvent::Connection(ConnectionState::Disconnected));
                    handle_task.finish();
                    return;
                }
            }

            while let Some(event) = futures_util::StreamExt::next(&mut rx).await {
                Self::deliver_event::<TResult, F>(&mut callback, &client, event);
            }

            handle_task.finish();
        });

        handle.set_task(task);
        handle
    }

    fn deliver_event<TResult, F>(
        callback: &mut F,
        client: &ForgeClient,
        event: SseDispatch,
    ) where
        TResult: DeserializeOwned,
        F: FnMut(StreamEvent<TResult>),
    {
        match event {
            SseDispatch::Data(value) => match serde_json::from_value::<TResult>(value) {
                Ok(data) => callback(StreamEvent::Data(data)),
                Err(e) => {
                    callback(StreamEvent::Error(ForgeClientError::new(
                        "DESERIALIZATION_ERROR",
                        e.to_string(),
                        None,
                    )));
                }
            },
            SseDispatch::Error { code, message } => {
                let err = ForgeClientError::new(&code, &message, None);
                if code == "UNAUTHORIZED" {
                    if let Some(handler) = &client.inner.on_auth_error {
                        handler(err.as_forge_error());
                    }
                }
                callback(StreamEvent::Error(err));
            }
        }
    }

    /// Ensure the shared SSE connection is established. Spawns the event loop on first call.
    async fn ensure_connected(&self) -> Result<(), ForgeClientError> {
        let rx = {
            let mut sse = self.inner.sse.borrow_mut();
            if sse.state == SseState::Connected {
                return Ok(());
            }

            let (tx, rx) = futures_channel::oneshot::channel();
            sse.connect_waiters.push(tx);

            if sse.state == SseState::Idle {
                sse.state = SseState::Connecting;
                drop(sse);
                platform::start_event_loop(self.clone());
            }

            rx
        };

        rx.await.unwrap_or_else(|_| {
            Err(ForgeClientError::new(
                "SSE_CONNECTION_FAILED",
                "Connection attempt cancelled",
                None,
            ))
        })
    }

    /// Register a subscription with the server via POST.
    /// If the server returns SESSION_NOT_FOUND, forces a reconnect and retries once.
    async fn register_subscription(
        &self,
        sub_id: &str,
    ) -> Result<RpcEnvelopeRaw, ForgeClientError> {
        let envelope = self.try_register_subscription(sub_id).await?;

        // Stale session: force reconnect and retry once
        let needs_retry = !envelope.success
            && envelope
                .error
                .as_ref()
                .is_some_and(|e| e.code == "SESSION_NOT_FOUND" || e.code == "SESSION_PRINCIPAL_MISMATCH");

        if needs_retry {
            self.force_reconnect().await;
            self.ensure_connected().await?;
            return self.try_register_subscription(sub_id).await;
        }

        Ok(envelope)
    }

    async fn try_register_subscription(
        &self,
        sub_id: &str,
    ) -> Result<RpcEnvelopeRaw, ForgeClientError> {
        let (endpoint, payload) = {
            let sse = self.inner.sse.borrow();
            let meta = sse
                .registrations
                .get(sub_id)
                .ok_or_else(|| {
                    ForgeClientError::new("INTERNAL_ERROR", "Registration metadata not found", None)
                })?;
            let session_id = sse.session_id.clone().unwrap_or_default();
            let session_secret = sse.session_secret.clone().unwrap_or_default();
            let mut payload = meta.payload.clone();
            let obj = payload
                .as_object_mut()
                .expect("registration payload must be an object");
            obj.insert("session_id".into(), serde_json::Value::String(session_id));
            obj.insert("session_secret".into(), serde_json::Value::String(session_secret));
            (meta.endpoint, payload)
        };

        let url = format!("{}{}", self.inner.url, endpoint);
        platform::request_json(self, &url, payload, None).await
    }

    async fn force_reconnect(&self) {
        let task = {
            let mut sse = self.inner.sse.borrow_mut();
            sse.session_id = None;
            sse.session_secret = None;
            sse.state = SseState::Idle;
            // No need to drain waiters; force_reconnect is only called mid-registration
            // where the caller already passed ensure_connected
            sse.event_loop_task.take()
        };
        if let Some(task) = task {
            task.cancel();
        }
        sleep(Duration::from_millis(10)).await;
    }

    async fn reregister_all(&self) {
        let sub_ids: Vec<String> = {
            let sse = self.inner.sse.borrow();
            sse.registrations.keys().cloned().collect()
        };

        for sub_id in sub_ids {
            let _ = self.register_subscription(&sub_id).await;
        }
    }

    fn dispatch_event(&self, target: &str, event: SseDispatch) {
        let tx = {
            let sse = self.inner.sse.borrow();
            sse.listeners.get(target).cloned()
        };
        if let Some(tx) = tx {
            let _ = tx.unbounded_send(event);
        }
    }

    fn broadcast_connection(&self, state: ConnectionState) {
        if let Some(mut signal) = self.inner.connection_state {
            signal.set(state);
        }
    }

    fn mark_connected(&self, session_id: String, session_secret: String) -> bool {
        let mut sse = self.inner.sse.borrow_mut();
        let is_reconnect = sse.ever_connected;
        sse.session_id = Some(session_id);
        sse.session_secret = Some(session_secret);
        sse.state = SseState::Connected;
        sse.reconnect_attempts = 0;
        sse.ever_connected = true;
        for waiter in sse.connect_waiters.drain(..) {
            let _ = waiter.send(Ok(()));
        }
        is_reconnect
    }

    fn mark_disconnected(&self) {
        let mut sse = self.inner.sse.borrow_mut();
        sse.session_id = None;
        sse.session_secret = None;
        sse.state = SseState::Idle;
        sse.event_loop_task = None;
        let err = || ForgeClientError::new("SSE_CONNECTION_FAILED", "SSE connection lost", None);
        for waiter in sse.connect_waiters.drain(..) {
            let _ = waiter.send(Err(err()));
        }
    }

    fn should_reconnect(&self) -> Option<u32> {
        let mut sse = self.inner.sse.borrow_mut();
        if sse.listeners.is_empty() {
            return None;
        }
        let attempts = sse.reconnect_attempts;
        if attempts >= MAX_RECONNECT_ATTEMPTS {
            return None;
        }
        sse.reconnect_attempts = attempts + 1;
        Some(attempts)
    }

    fn get_token(&self) -> Option<String> {
        self.inner
            .get_token
            .as_ref()
            .and_then(|provider| provider())
            .filter(|t| !t.is_empty())
    }

    fn decode_envelope<TResult>(
        &self,
        envelope: RpcEnvelopeRaw,
    ) -> Result<TResult, ForgeClientError>
    where
        TResult: DeserializeOwned,
    {
        if !envelope.success {
            let error = envelope.error.unwrap_or(ForgeError {
                code: "UNKNOWN".to_string(),
                message: "Unknown error".to_string(),
                details: None,
            });
            return Err(ForgeClientError::new(error.code, error.message, error.details));
        }

        let data = envelope.data.ok_or_else(|| {
            ForgeClientError::new("EMPTY_RESPONSE", "Server returned no data", None)
        })?;
        serde_json::from_value(data)
            .map_err(|err| ForgeClientError::new("DESERIALIZATION_ERROR", err.to_string(), None))
    }

    fn random_id(&self, prefix: &str) -> String {
        let id = NEXT_SUBSCRIPTION_ID.fetch_add(1, Ordering::Relaxed);
        format!("{prefix}-{id}")
    }
}

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

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

#[derive(Clone)]
pub struct SubscriptionHandle {
    closed: Rc<Cell<bool>>,
    task: Rc<RefCell<Option<Task>>>,
    cleanup: Rc<RefCell<Option<Box<dyn FnOnce()>>>>,
}

impl SubscriptionHandle {
    fn new(sub_id: String, target: String, client: ForgeClient) -> Self {
        let cleanup: Box<dyn FnOnce()> = Box::new(move || {
            let mut sse = client.inner.sse.borrow_mut();
            sse.listeners.remove(&target);
            sse.registrations.remove(&sub_id);
            // Jobs/workflows have server-managed lifecycles; only query subs need explicit unsubscribe
            if target.starts_with("sub:") {
                let session_id = sse.session_id.clone();
                let session_secret = sse.session_secret.clone();
                drop(sse);
                if let (Some(sid), Some(ss)) = (session_id, session_secret) {
                    let url = format!("{}/_api/unsubscribe", client.inner.url);
                    let payload = serde_json::json!({
                        "session_id": sid,
                        "session_secret": ss,
                        "id": sub_id,
                    });
                    let client = client.clone();
                    dioxus::prelude::spawn(async move {
                        let _ = platform::request_json(&client, &url, payload, None).await;
                    });
                }
            }
        });

        Self {
            closed: Rc::new(Cell::new(false)),
            task: Rc::new(RefCell::new(None)),
            cleanup: Rc::new(RefCell::new(Some(cleanup))),
        }
    }

    fn set_task(&self, task: Task) {
        *self.task.borrow_mut() = Some(task);
    }

    pub(crate) fn finish(&self) {
        if self.closed.replace(true) {
            return;
        }
        if let Some(cleanup) = self.cleanup.borrow_mut().take() {
            cleanup();
        }
        self.task.borrow_mut().take();
    }

    pub fn close(&self) {
        let task = { self.task.borrow_mut().clone() };
        self.finish();
        if let Some(task) = task {
            task.cancel();
        }
    }

    pub fn is_closed(&self) -> bool {
        self.closed.get()
    }
}

impl Drop for SubscriptionHandle {
    fn drop(&mut self) {
        self.close();
    }
}

// ── WASM platform ───────────────────────────────────────────────────────────

#[cfg(target_arch = "wasm32")]
mod platform {
    use dioxus::prelude::spawn;
    use futures_util::{StreamExt, stream};
    use gloo_net::eventsource::futures::EventSource;
    use gloo_net::http::Request;
    use js_sys::{JSON, encode_uri_component};

    use super::{ForgeClient, SseDispatch, sleep};
    use crate::signals::platform_tag;
    use crate::types::{
        ConnectedEvent, ConnectionState, ForgeClientError, RpcEnvelopeRaw, SseEnvelopeRaw,
    };

    pub(super) async fn request_json(
        client: &ForgeClient,
        url: &str,
        body: serde_json::Value,
        correlation_id: Option<&str>,
    ) -> Result<RpcEnvelopeRaw, ForgeClientError> {
        let mut request = Request::post(url)
            .header("Content-Type", "application/json")
            .header("x-forge-platform", platform_tag())
            .credentials(web_sys::RequestCredentials::Include);
        if let Some(token) = client.get_token() {
            request = request.header("Authorization", &format!("Bearer {token}"));
        }
        if let Some(cid) = correlation_id {
            request = request.header("x-correlation-id", cid);
        }

        let request = request.body(body.to_string()).map_err(request_error)?;
        request
            .send()
            .await
            .map_err(request_error)?
            .json()
            .await
            .map_err(request_error)
    }

    pub(super) async fn request_multipart(
        client: &ForgeClient,
        url: &str,
        form: web_sys::FormData,
        correlation_id: Option<&str>,
    ) -> Result<RpcEnvelopeRaw, ForgeClientError> {
        let mut request = Request::post(url)
            .header("x-forge-platform", platform_tag())
            .credentials(web_sys::RequestCredentials::Include);
        if let Some(token) = client.get_token() {
            request = request.header("Authorization", &format!("Bearer {token}"));
        }
        if let Some(cid) = correlation_id {
            request = request.header("x-correlation-id", cid);
        }

        let response = request.body(form).map_err(request_error)?;
        response
            .send()
            .await
            .map_err(request_error)?
            .json()
            .await
            .map_err(request_error)
    }

    fn message_data_as_string(message: &web_sys::MessageEvent) -> Option<String> {
        let data = message.data();
        data.as_string().or_else(|| {
            JSON::stringify(&data)
                .ok()
                .and_then(|value| value.as_string())
                .map(|raw| serde_json::from_str::<String>(&raw).unwrap_or(raw))
        })
    }

    fn events_url(client: &ForgeClient) -> String {
        match client.get_token() {
            Some(token) => format!(
                "{}/_api/events?token={}",
                client.inner.url,
                encode_uri_component(&token)
            ),
            None => format!("{}/_api/events", client.inner.url),
        }
    }

    /// Start the single shared SSE event loop.
    pub(super) fn start_event_loop(client: ForgeClient) {
        let client_for_task = client.clone();
        let task = spawn(async move {
            run_event_loop(&client_for_task).await;

            // Disconnected. Try to reconnect.
            client_for_task.mark_disconnected();
            client_for_task.broadcast_connection(ConnectionState::Disconnected);

            if let Some(attempts) = client_for_task.should_reconnect() {
                let delay = 1000 * (1u64 << attempts.min(4));
                let jitter = (js_sys::Math::random() * 500.0) as u64;
                sleep(std::time::Duration::from_millis(delay + jitter)).await;

                client_for_task.inner.sse.borrow_mut().state = super::SseState::Connecting;
                start_event_loop(client_for_task);
            }
        });

        client.inner.sse.borrow_mut().event_loop_task = Some(task);
    }

    async fn run_event_loop(client: &ForgeClient) {
        let mut event_source = match EventSource::new(&events_url(client)) {
            Ok(source) => source,
            Err(_) => {
                return;
            }
        };

        let mut connected_stream = match event_source.subscribe("connected") {
            Ok(stream) => stream,
            Err(_) => return,
        };
        let update_stream = match event_source.subscribe("update") {
            Ok(stream) => stream,
            Err(_) => return,
        };
        let error_stream = match event_source.subscribe("error") {
            Ok(stream) => stream,
            Err(_) => return,
        };

        // Wait for the connected event
        let connected_event = match connected_stream.next().await {
            Some(Ok((_kind, message))) => {
                let Some(raw) = message_data_as_string(&message) else {
                    return;
                };
                match serde_json::from_str::<ConnectedEvent>(&raw) {
                    Ok(event) => event,
                    Err(_) => return,
                }
            }
            _ => return,
        };

        let session_id = connected_event.session_id.unwrap_or_default();
        let session_secret = connected_event.session_secret.unwrap_or_default();

        if session_id.is_empty() || session_secret.is_empty() {
            return;
        }

        let is_reconnect = client.mark_connected(session_id, session_secret);
        client.broadcast_connection(ConnectionState::Connected);

        // Only re-register on reconnect; initial registrations are handled by each subscription task
        if is_reconnect {
            client.reregister_all().await;
        }

        let mut events = stream::select(update_stream, error_stream);
        while let Some(event) = events.next().await {
            match event {
                Ok((kind, message)) => {
                    let Some(raw) = message_data_as_string(&message) else {
                        continue;
                    };
                    let Ok(envelope) = serde_json::from_str::<SseEnvelopeRaw>(&raw) else {
                        continue;
                    };

                    let Some(target) = envelope.target else {
                        continue;
                    };

                    if kind == "update" {
                        if let Some(payload) = envelope.payload {
                            client.dispatch_event(&target, SseDispatch::Data(payload));
                        }
                    } else {
                        let code = envelope.code.unwrap_or_else(|| "SSE_ERROR".to_string());
                        let message = envelope.message.unwrap_or_else(|| "Subscription error".to_string());
                        client.dispatch_event(&target, SseDispatch::Error { code, message });
                    }
                }
                Err(_) => break,
            }
        }

        event_source.close();
    }

    fn request_error(err: gloo_net::Error) -> ForgeClientError {
        ForgeClientError::new("REQUEST_FAILED", err.to_string(), None)
    }
}

// ── Native platform ─────────────────────────────────────────────────────────

#[cfg(not(target_arch = "wasm32"))]
mod platform {
    use dioxus::prelude::spawn;
    use futures_util::StreamExt;
    use reqwest::Client;
    use reqwest_eventsource::{Event, EventSource};

    use super::{ForgeClient, SseDispatch, sleep};
    use crate::signals::platform_tag;
    use crate::types::{
        ConnectedEvent, ConnectionState, ForgeClientError, RpcEnvelopeRaw, SseEnvelopeRaw,
    };

    pub(super) async fn request_json(
        client: &ForgeClient,
        url: &str,
        body: serde_json::Value,
        correlation_id: Option<&str>,
    ) -> Result<RpcEnvelopeRaw, ForgeClientError> {
        let mut request = Client::new()
            .post(url)
            .header("x-forge-platform", platform_tag())
            .json(&body);
        if let Some(token) = client.get_token() {
            request = request.bearer_auth(token);
        }
        if let Some(cid) = correlation_id {
            request = request.header("x-correlation-id", cid);
        }

        request
            .send()
            .await
            .map_err(request_error)?
            .json()
            .await
            .map_err(request_error)
    }

    pub(super) async fn request_multipart(
        client: &ForgeClient,
        url: &str,
        form: reqwest::multipart::Form,
        correlation_id: Option<&str>,
    ) -> Result<RpcEnvelopeRaw, ForgeClientError> {
        let mut request = Client::new()
            .post(url)
            .header("x-forge-platform", platform_tag())
            .multipart(form);
        if let Some(token) = client.get_token() {
            request = request.bearer_auth(token);
        }
        if let Some(cid) = correlation_id {
            request = request.header("x-correlation-id", cid);
        }

        request
            .send()
            .await
            .map_err(request_error)?
            .json()
            .await
            .map_err(request_error)
    }

    /// Start the single shared SSE event loop.
    pub(super) fn start_event_loop(client: ForgeClient) {
        let client_for_task = client.clone();
        let task = spawn(async move {
            run_event_loop(&client_for_task).await;

            client_for_task.mark_disconnected();
            client_for_task.broadcast_connection(ConnectionState::Disconnected);

            if let Some(attempts) = client_for_task.should_reconnect() {
                let delay = 1000 * (1u64 << attempts.min(4));
                sleep(std::time::Duration::from_millis(delay)).await;

                client_for_task.inner.sse.borrow_mut().state = super::SseState::Connecting;
                start_event_loop(client_for_task);
            }
        });

        client.inner.sse.borrow_mut().event_loop_task = Some(task);
    }

    async fn run_event_loop(client: &ForgeClient) {
        let mut request = Client::new().get(format!("{}/_api/events", client.inner.url));
        if let Some(token) = client.get_token() {
            request = request.bearer_auth(token);
        }

        let mut event_source = match EventSource::new(request) {
            Ok(source) => source,
            Err(_) => return,
        };

        // Wait for connected event
        let connected_event = loop {
            let Some(event) = event_source.next().await else {
                return;
            };
            match event {
                Ok(Event::Open) => continue,
                Ok(Event::Message(msg)) if msg.event == "connected" => {
                    match serde_json::from_str::<ConnectedEvent>(&msg.data) {
                        Ok(event) => break event,
                        Err(_) => return,
                    }
                }
                Ok(Event::Message(_)) => continue,
                Err(_) => return,
            }
        };

        let session_id = connected_event.session_id.unwrap_or_default();
        let session_secret = connected_event.session_secret.unwrap_or_default();

        if session_id.is_empty() || session_secret.is_empty() {
            return;
        }

        let is_reconnect = client.mark_connected(session_id, session_secret);
        client.broadcast_connection(ConnectionState::Connected);

        if is_reconnect {
            client.reregister_all().await;
        }

        while let Some(event) = event_source.next().await {
            match event {
                Ok(Event::Open) => {}
                Ok(Event::Message(msg)) if msg.event == "update" || msg.event == "error" => {
                    let Ok(envelope) = serde_json::from_str::<SseEnvelopeRaw>(&msg.data) else {
                        continue;
                    };
                    let Some(target) = envelope.target else {
                        continue;
                    };

                    if msg.event == "update" {
                        if let Some(payload) = envelope.payload {
                            client.dispatch_event(&target, SseDispatch::Data(payload));
                        }
                    } else {
                        let code = envelope.code.unwrap_or_else(|| "SSE_ERROR".to_string());
                        let message =
                            envelope.message.unwrap_or_else(|| "Subscription error".to_string());
                        client.dispatch_event(&target, SseDispatch::Error { code, message });
                    }
                }
                Ok(Event::Message(_)) => {}
                Err(_) => break,
            }
        }

        event_source.close();
    }

    fn request_error(err: reqwest::Error) -> ForgeClientError {
        ForgeClientError::new("REQUEST_FAILED", err.to_string(), None)
    }
}