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 : // 4.22 GC: the native broker sweeps on a tokio interval (main.rs,
89 : // ANTARES_SWEEP_SECS). The browser has no env, so the SAME variable is
90 : // read off `globalThis.ANTARES_SWEEP_SECS` (seconds, set before
91 : // construction — the playground forwards a ?ANTARES_SWEEP_SECS= URL
92 : // param into both the worker and in-page contexts); absent → 60 s.
93 : // Without this loop the OPFS file grows without bound under ticking
94 : // transient attributes — reads filter expired instances but nothing
95 : // would ever delete them.
96 : #[cfg(target_arch = "wasm32")]
97 : {
98 : let store = state.store.clone();
99 : let sweep_ms = js_sys::Reflect::get(&js_sys::global(), &"ANTARES_SWEEP_SECS".into())
100 : .ok()
101 : .and_then(|v| v.as_f64())
102 : .filter(|s| *s > 0.0)
103 : .map(|s| (s * 1000.0) as u32)
104 : .unwrap_or(60_000);
105 : wasm_bindgen_futures::spawn_local(async move {
106 : loop {
107 : gloo_timers::future::TimeoutFuture::new(sweep_ms).await;
108 : store.sweep_expired();
109 : }
110 : });
111 : }
112 0 : Self {
113 0 : router: antares_api::router(state),
114 0 : }
115 0 : }
116 :
117 : /// Serve ONE request. The signature is the seam every front end reduces
118 : /// to: the Service Worker, the in-page API, and the Node shim
119 : /// all funnel here.
120 : ///
121 : /// `&self` on purpose: a federation forward to the loopback host re-enters
122 : /// this same instance WHILE an outer `handle` is suspended — `&mut`
123 : /// would make that a wasm-bindgen recursive-borrow error. Router clone is
124 : /// a cheap Arc bump and shares all state.
125 0 : pub async fn handle(&self, req: http::Request<Vec<u8>>) -> http::Response<Vec<u8>> {
126 0 : let (mut parts, body) = req.into_parts();
127 : // The native binary routes under NormalizePathLayer::trim_trailing_slash
128 : // (6.3 URLs arrive both with and without a trailing '/'); this seam is
129 : // the wasm equivalent — same trim, applied before the router sees it.
130 0 : if let Some(pq) = parts.uri.path_and_query() {
131 0 : let path = pq.path();
132 0 : let trimmed = path.trim_end_matches('/');
133 0 : if trimmed.len() != path.len() && !trimmed.is_empty() {
134 0 : let new = match pq.query() {
135 0 : Some(q) => format!("{trimmed}?{q}"),
136 0 : None => trimmed.to_owned(),
137 : };
138 0 : if let Ok(uri) = new.parse() {
139 0 : parts.uri = uri;
140 0 : }
141 0 : }
142 0 : }
143 0 : let req = http::Request::from_parts(parts, Body::from(body));
144 0 : let mut router = self.router.clone();
145 0 : let resp = match router.call(req).await {
146 0 : Ok(r) => r,
147 : // The router is Infallible; keep the arm honest rather than
148 : // unwrapping (workspace lints deny unwrap outside tests).
149 : Err(_) => {
150 0 : return http::Response::builder()
151 0 : .status(500)
152 0 : .body(Vec::new())
153 0 : .unwrap_or_default()
154 : }
155 : };
156 0 : let (parts, body) = resp.into_parts();
157 0 : let bytes = body
158 0 : .collect()
159 0 : .await
160 0 : .map(|c| c.to_bytes().to_vec())
161 0 : .unwrap_or_default();
162 0 : http::Response::from_parts(parts, bytes)
163 0 : }
164 : }
165 :
166 : #[cfg(target_arch = "wasm32")]
167 : mod browser;
168 : #[cfg(target_arch = "wasm32")]
169 : mod opfs;
170 : #[cfg(target_arch = "wasm32")]
171 : pub use browser::*;
|
