ocre/runtime/
realtime.rs

1use axum::response::Response;
2use wasm_bindgen::JsValue;
3use worker::{Env, Headers, Method, Request, RequestInit, Stub, send::SendFuture};
4
5use super::Ctx;
6use crate::{
7    Error, Result,
8    realtime::{
9        CHANNELS_BINDING, LOG_PREFIX, SUBSCRIBER_HEADER, WebSocketUpgrade, channel_error, missing_binding, record,
10    },
11};
12
13// `#[durable_object]` generates public wasm-bindgen glue (constructor and runtime callbacks) inside
14// a `const _` block, which cannot carry docs and which an `allow` on the struct does not reach.
15#[allow(missing_docs)]
16mod channel;
17
18pub use channel::OcreChannel;
19
20/// Address of requests from the Worker to a channel object; only the method matters.
21const CHANNEL_URL: &str = "https://ocre-channel/";
22
23/// The channel object for `channel`.
24fn stub(env: &Env, channel: &str) -> Result<Stub> {
25    let namespace = env.durable_object(CHANNELS_BINDING).map_err(|err| missing_binding(&err.to_string()))?;
26    Ok(namespace.get_by_name(channel)?)
27}
28
29/// Sends `message` (an HTML fragment or JSON text) to every browser connected to `channel`.
30///
31/// Returns once the channel's [`OcreChannel`] object has sent it. Build HTML
32/// messages with [`prepend`](crate::realtime::prepend),
33/// [`append`](crate::realtime::append), [`update`](crate::realtime::update),
34/// [`remove`](crate::realtime::remove), or an element with an `id` that
35/// replaces the page element with that id; several can go in one message.
36/// Works from handlers and jobs. The returned future is `Send`, so axum
37/// handlers can await it.
38///
39/// Every failure is logged as `[ocre realtime] broadcast to <channel> failed:
40/// ...` and returned; callers that treat updates as best effort (the
41/// generated controllers do) can ignore it with `.ok()`.
42///
43/// Free plan: each call is one Durable Object request (100,000 a day), even
44/// when the channel has no subscribers, and one subrequest of the current
45/// request; the messages to browsers are free.
46///
47/// # Errors
48///
49/// - [`Error::Internal`] when `channel` is not a valid name (1 to
50///   [`MAX_CHANNEL_LEN`](crate::realtime::MAX_CHANNEL_LEN) ASCII letters,
51///   digits, `_`, `-`, `.` or `:`).
52/// - [`Error::Internal`] when the `CHANNELS` Durable Object binding is
53///   missing; the message names the cloudflare.config.ts entries to add.
54/// - [`Error::Internal`] when the channel object cannot be reached (e.g. the
55///   free-plan quota is exhausted) or answers with a non-200 status.
56///
57/// # Examples
58///
59/// ```no_run
60/// use axum::extract::{Path, State};
61/// use ocre::{Ctx, Result, realtime};
62///
63/// async fn destroy(State(ctx): State<Ctx>, Path(id): Path<i64>) -> Result<()> {
64///     // ... delete the post, then remove its row from every open index page:
65///     realtime::broadcast(&ctx, "posts", &realtime::remove(&format!("post_{id}"))).await.ok();
66///     Ok(())
67/// }
68/// ```
69pub fn broadcast(ctx: &Ctx, channel: &str, message: &str) -> impl Future<Output = Result<()>> + Send + use<> {
70    let env = ctx.env().clone();
71    let invalid = channel_error(channel);
72    let channel = channel.to_owned();
73    let message = message.to_owned();
74    SendFuture::new(async move {
75        let result = match invalid {
76            Some(invalid) => Err(Error::internal(invalid)),
77            None => send(&env, &channel, &message).await,
78        };
79        match &result {
80            Ok(()) => record(&channel, &message),
81            Err(err) => worker::console_error!("{LOG_PREFIX} broadcast to {channel} failed: {err}"),
82        }
83        result
84    })
85}
86
87async fn send(env: &Env, channel: &str, message: &str) -> Result<()> {
88    let mut init = RequestInit::new();
89    init.with_method(Method::Post).with_body(Some(JsValue::from_str(message)));
90    let mut response = stub(env, channel)?.fetch_with_request(Request::new_with_init(CHANNEL_URL, &init)?).await?;
91    match response.status_code() {
92        200 => Ok(()),
93        status => Err(Error::internal(format!("channel object answered {status}: {}", response.text().await?))),
94    }
95}
96
97impl WebSocketUpgrade {
98    /// Connects the browser to `channel`, returning the `101 Switching Protocols` response to send back.
99    ///
100    /// The handshake is forwarded to the channel's [`OcreChannel`] object,
101    /// which accepts the WebSocket (hibernating). Check who may listen before
102    /// calling it: list the allowed channels, or use a channel per record or
103    /// user (`post:12`). The returned future is `Send`.
104    ///
105    /// Free plan: one Durable Object request per connection and reconnection.
106    ///
107    /// # Errors
108    ///
109    /// - [`Error::BadRequest`] (400) when `channel` is not a valid name (1 to
110    ///   [`MAX_CHANNEL_LEN`](crate::realtime::MAX_CHANNEL_LEN) ASCII letters,
111    ///   digits, `_`, `-`, `.` or `:`).
112    /// - [`Error::Internal`] (500) when the `CHANNELS` Durable Object binding
113    ///   is missing (the message names the cloudflare.config.ts entries to add) or
114    ///   the channel object cannot be reached.
115    ///
116    /// # Examples
117    ///
118    /// ```no_run
119    /// use axum::{extract::{Path, State}, response::Response};
120    /// use ocre::{Ctx, Error, Result, realtime::WebSocketUpgrade};
121    ///
122    /// async fn connect(State(ctx): State<Ctx>, Path(channel): Path<String>, upgrade: WebSocketUpgrade)
123    ///     -> Result<Response> {
124    ///     match channel.as_str() {
125    ///         "posts" => {}
126    ///         _ => return Err(Error::NotFound),
127    ///     }
128    ///     upgrade.connect(&ctx, &channel).await
129    /// }
130    /// ```
131    pub fn connect(self, ctx: &Ctx, channel: &str) -> impl Future<Output = Result<Response>> + Send + use<> {
132        let env = ctx.env().clone();
133        let invalid = channel_error(channel);
134        let channel = channel.to_owned();
135        let subscriber = self.subscriber_header();
136        SendFuture::new(async move {
137            if let Some(invalid) = invalid {
138                return Err(Error::bad_request(invalid));
139            }
140            let headers = Headers::new();
141            headers.set("Upgrade", "websocket")?;
142            headers.set(SUBSCRIBER_HEADER, &subscriber?)?;
143            let mut init = RequestInit::new();
144            init.with_headers(headers);
145            let response =
146                stub(&env, &channel)?.fetch_with_request(Request::new_with_init(CHANNEL_URL, &init)?).await?;
147            Ok(Response::from(response))
148        })
149    }
150}
ocre/runtime/realtime.rs

ocre/runtime/
realtime.rs
