Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The broker in a browser tab.
3 : //!
4 : //! Everything above the socket is unchanged — the same axum router, the same
5 : //! handlers, the same memory store through the store seam. The ONE thing that
6 : //! cannot cross is the TCP listener (browsers have no inbound sockets), so a
7 : //! Service Worker feeds requests in instead and `handle` drives the
8 : //! router directly with `tower::Service::call`, exactly as the native
9 : //! `main.rs` accept loop does per connection.
10 : //!
11 : //! What the browser build is NOT: no NATS, no MQTT, no Postgres, no
12 : //! roles. `bus=local` and the memory store are the only shapes that exist
13 : //! here, which is why this crate turns `antares-api`'s default features off.
14 : #![cfg_attr(not(test), warn(clippy::expect_used))]
15 :
16 : use axum::body::Body;
17 : use http_body_util::BodyExt;
18 : use tower::Service;
19 :
20 : /// One broker instance: the composed router plus the state it owns.
21 : pub struct Broker {
22 : router: axum::Router,
23 : }
24 :
25 : impl Default for Broker {
26 0 : fn default() -> Self {
27 0 : Self::new()
28 0 : }
29 : }
30 :
31 : impl Broker {
32 : /// Build the router over an in-memory store. Mirrors the native wiring
33 : /// minus the pieces that need a socket or a pool.
34 0 : pub fn new() -> Self {
35 0 : Self::with_store(antares_sql::store::Store::default(), "memory")
36 0 : }
37 :
38 : /// The same wiring over an externally-constructed store — the OPFS-backed
39 : /// store enters here; `mode` is what `/q/health` reports.
40 0 : pub fn with_store(store: antares_sql::store::Store, mode: &str) -> Self {
41 0 : Self::with_store_alias(store, mode, None)
42 0 : }
43 :
44 : /// `host_alias` names this instance in Via chains — must be distinct per
45 : /// instance in a federation, or loop detection 508s every forward.
46 0 : pub fn with_store_alias(
47 0 : store: antares_sql::store::Store,
48 0 : mode: &str,
49 0 : host_alias: Option<String>,
50 0 : ) -> Self {
51 0 : let store = std::sync::Arc::new(antares_sql::store::any::AnyStore::Mem(store));
52 : // wasm compositions only ever carry the Mem arm — anything but a
53 : // known Mem-arm mode name is a caller bug, defaulted to memory.
54 0 : let mut state = antares_api::AppState::with_drivers(
55 0 : host_alias.unwrap_or_else(|| "antares-wasm".to_owned()),
56 0 : store.clone(),
57 0 : store,
58 0 : mode,
59 : );
60 : // Same in-process matcher/notifier path as bus=local: the
61 : // store's change hook feeds it, no bus process exists to talk to.
62 : // The browser constructors are JS-callable and cannot await, and
63 : // every driver in a wasm composition is the in-process memory arm,
64 : // whose futures complete on their first poll. One that suspended
65 : // would mean this composition grew a step no browser build has: the
66 : // matcher then takes the store-scan fallback `wire` documents
67 : // instead of the page losing its broker.
68 0 : if futures_util::FutureExt::now_or_never(antares_api::wire(&mut state)).is_none() {
69 0 : tracing::error!(
70 : "wiring did not complete synchronously; \
71 : matching falls back to a store scan per change"
72 : );
73 0 : }
74 : // 5.8.1.4: distributed subscriptions hand this URL to the remote
75 : // broker as the notification callback. wasm32 has no process env
76 : // (the native default reads ANTARES_PUBLIC_URL + appends the port),
77 : // so the SAME variable comes off `globalThis` — set it before
78 : // construction, like ANTARES_SWEEP_SECS below. Absent → the portless
79 : // host-alias default, which no peer outside a browser can dial.
80 : #[cfg(target_arch = "wasm32")]
81 : if let Some(url) = js_sys::Reflect::get(&js_sys::global(), &"ANTARES_PUBLIC_URL".into())
82 : .ok()
83 : .and_then(|v| v.as_string())
84 : .filter(|s| !s.is_empty())
85 : {
86 : state.public_url = url;
87 : }
88 : // 6.3.10 ceiling (ANTARES_TEMPORAL_INSTANCE_LIMIT): the same
89 : // variable off `globalThis`, set before construction; the ETSI cells
90 : // set 9 for the suite's 206 cases. Absent → the native default.
91 : #[cfg(target_arch = "wasm32")]
92 : if let Some(n) =
93 : js_sys::Reflect::get(&js_sys::global(), &"ANTARES_TEMPORAL_INSTANCE_LIMIT".into())
94 : .ok()
95 : .and_then(|v| v.as_f64())
96 : .filter(|n| *n >= 1.0)
97 : {
98 : state.temporal_instance_limit = n as usize;
99 : }
100 : // 4.22 GC: the native broker sweeps on a tokio interval (main.rs,
101 : // ANTARES_SWEEP_SECS). The browser has no env, so the SAME variable is
102 : // read off `globalThis.ANTARES_SWEEP_SECS` (seconds, set before
103 : // construction — the playground forwards a ?ANTARES_SWEEP_SECS= URL
104 : // param into both the worker and in-page contexts); absent → 60 s.
105 : // Without this loop the OPFS file grows without bound under ticking
106 : // transient attributes — reads filter expired instances but nothing
107 : // would ever delete them.
108 : #[cfg(target_arch = "wasm32")]
109 : {
110 : // both sweeps of the native loop: the store's 4.22 GC, then the
111 : // documents with their own expiry (Registrations, Snapshots,
112 : // EntityMaps). The store sweep was a future never awaited, so
113 : // nothing was reaped.
114 : let sweeper = state.clone();
115 : let sweep_ms = js_sys::Reflect::get(&js_sys::global(), &"ANTARES_SWEEP_SECS".into())
116 : .ok()
117 : .and_then(|v| v.as_f64())
118 : .filter(|s| *s > 0.0)
119 : .map(|s| (s * 1000.0) as u32)
120 : .unwrap_or(60_000);
121 : wasm_bindgen_futures::spawn_local(async move {
122 : loop {
123 : gloo_timers::future::TimeoutFuture::new(sweep_ms).await;
124 : sweeper.store.sweep_expired().await;
125 : antares_api::sweep_expired_docs(&sweeper).await;
126 : }
127 : });
128 : }
129 0 : Self {
130 0 : router: antares_api::router(state),
131 0 : }
132 0 : }
133 :
134 : /// Serve ONE request. The signature is the seam every front end reduces
135 : /// to: the Service Worker, the in-page API, and the Node shim
136 : /// all funnel here.
137 : ///
138 : /// `&self` on purpose: a federation forward to the loopback host re-enters
139 : /// this same instance WHILE an outer `handle` is suspended — `&mut`
140 : /// would make that a wasm-bindgen recursive-borrow error. Router clone is
141 : /// a cheap Arc bump and shares all state.
142 0 : pub async fn handle(&self, req: http::Request<Vec<u8>>) -> http::Response<Vec<u8>> {
143 0 : let (mut parts, body) = req.into_parts();
144 : // The native binary routes under NormalizePathLayer::trim_trailing_slash
145 : // (6.3 URLs arrive both with and without a trailing '/'); this seam is
146 : // the wasm equivalent — same trim, applied before the router sees it.
147 0 : if let Some(pq) = parts.uri.path_and_query() {
148 0 : let path = pq.path();
149 0 : let trimmed = path.trim_end_matches('/');
150 0 : if trimmed.len() != path.len() && !trimmed.is_empty() {
151 0 : let new = match pq.query() {
152 0 : Some(q) => format!("{trimmed}?{q}"),
153 0 : None => trimmed.to_owned(),
154 : };
155 0 : if let Ok(uri) = new.parse() {
156 0 : parts.uri = uri;
157 0 : }
158 0 : }
159 0 : }
160 0 : let req = http::Request::from_parts(parts, Body::from(body));
161 0 : let mut router = self.router.clone();
162 0 : let resp = match router.call(req).await {
163 0 : Ok(r) => r,
164 : // The router is Infallible; keep the arm honest rather than
165 : // unwrapping (workspace lints deny unwrap outside tests).
166 : Err(_) => {
167 0 : return http::Response::builder()
168 0 : .status(500)
169 0 : .body(Vec::new())
170 0 : .unwrap_or_default()
171 : }
172 : };
173 0 : let (parts, body) = resp.into_parts();
174 0 : let bytes = body
175 0 : .collect()
176 0 : .await
177 0 : .map(|c| c.to_bytes().to_vec())
178 0 : .unwrap_or_default();
179 0 : http::Response::from_parts(parts, bytes)
180 0 : }
181 : }
182 :
183 : #[cfg(target_arch = "wasm32")]
184 : mod browser;
185 : #[cfg(target_arch = "wasm32")]
186 : mod opfs;
187 : #[cfg(target_arch = "wasm32")]
188 : pub use browser::*;
|
