cratefield_core/ports/
realtime.rs

1//! The `Realtime` port (issue #103): rooms of WebSocket connections that share
2//! a clock and a chat — Yoginini's "together rooms", 2-20 people practising the
3//! same sequence.
4//!
5//! The harness is otherwise one stateless Worker over ports, with no way to
6//! hold a connection or coordinate between requests. This port is the
7//! exception: on Cloudflare a Durable Object holds the sockets (with the
8//! hibernation API, so an idle room costs nothing); on the native runtime an
9//! in-process registry does.
10//!
11//! **Split of responsibilities.** The port owns the sockets; the module owns
12//! the protocol. A module implements [`RoomHandler`] (`on_join`, `on_message`,
13//! `on_leave`, `on_alarm`) and the runtime calls it as events arrive, passing a
14//! [`RoomContext`] to reach the sockets. A module's ordinary HTTP handlers poke
15//! a room from outside a socket through the [`Realtime`] trait
16//! (`broadcast`/`members`). Because a handler holds no socket itself — it
17//! reaches them only through the context it is handed — a hibernating Durable
18//! Object can drop the handler between events and reconstruct it on the next.
19//!
20//! **The Durable Object is a coordinator, not the record.** Anything that must
21//! survive (a chat transcript) is the module's job to write to the database;
22//! room state is ephemeral coordination.
23
24use async_trait::async_trait;
25use std::time::Duration;
26use thiserror::Error;
27
28/// A connected participant, identified by the venture *after* it verified the
29/// upgrade request's bearer token (through the `Signer` or the auth client).
30/// A [`RoomHandler`] only ever sees this verified id, never a raw token.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct Member {
33    pub id: String,
34}
35
36impl Member {
37    #[must_use]
38    pub fn new(id: impl Into<String>) -> Self {
39        Self { id: id.into() }
40    }
41}
42
43/// Realtime failures.
44#[derive(Debug, Clone, Error)]
45pub enum RealtimeError {
46    /// No realtime adapter configured (no Durable Object binding): the caller
47    /// should degrade, not fail.
48    #[error("realtime is not configured")]
49    NotConfigured,
50    /// A message arrived from a socket whose member is not in the room (it left
51    /// or never joined): drop it.
52    #[error("sender is not a member of this room")]
53    NotAMember,
54    /// The room does not exist or has no connected members.
55    #[error("no such room: {0}")]
56    NoSuchRoom(String),
57    /// A transport or storage failure.
58    #[error("realtime operation failed: {0}")]
59    Operation(String),
60}
61
62/// The operations a [`RoomHandler`] performs on its room while handling an
63/// event. The handler holds no sockets of its own — it reaches them through
64/// this context — so a Durable Object can hibernate between events.
65#[async_trait]
66pub trait RoomContext: Send + Sync {
67    /// The room this event is for.
68    fn room_id(&self) -> &str;
69
70    /// The members currently connected.
71    async fn members(&self) -> Vec<Member>;
72
73    /// Send `message` to every connected socket.
74    async fn broadcast(&self, message: &[u8]) -> Result<(), RealtimeError>;
75
76    /// Send `message` to one member's socket(s).
77    async fn send(&self, member_id: &str, message: &[u8]) -> Result<(), RealtimeError>;
78
79    /// Schedule [`RoomHandler::on_alarm`] to fire once after `delay` — the
80    /// shared clock. Setting it again replaces the pending alarm.
81    async fn set_alarm(&self, delay: Duration) -> Result<(), RealtimeError>;
82}
83
84/// The protocol for one kind of room. A module implements it; the runtime owns
85/// the sockets and calls these as events arrive.
86///
87/// Implementations must be **stateless** — all room state lives in the room and
88/// is reached through [`RoomContext`] — so a hibernating Durable Object can
89/// reconstruct the handler on the next event. `route()` is the path prefix the
90/// runtime mounts the upgrade endpoint under (`GET <route>/<room_id>`).
91#[async_trait]
92pub trait RoomHandler: Send + Sync {
93    /// The path prefix for this room's upgrade endpoint, e.g. `"/rooms"`. The
94    /// runtime serves `GET <route>/<room_id>` as the WebSocket upgrade.
95    fn route(&self) -> &'static str {
96        "/rooms"
97    }
98
99    /// A member's socket has opened and joined the room.
100    async fn on_join(&self, ctx: &dyn RoomContext, member: &Member) -> Result<(), RealtimeError> {
101        let _ = (ctx, member);
102        Ok(())
103    }
104
105    /// A message arrived from a member's socket.
106    async fn on_message(
107        &self,
108        ctx: &dyn RoomContext,
109        member: &Member,
110        message: &[u8],
111    ) -> Result<(), RealtimeError>;
112
113    /// A member's socket has closed and left the room.
114    async fn on_leave(&self, ctx: &dyn RoomContext, member: &Member) -> Result<(), RealtimeError> {
115        let _ = (ctx, member);
116        Ok(())
117    }
118
119    /// A previously scheduled alarm fired (the shared clock tick).
120    async fn on_alarm(&self, ctx: &dyn RoomContext) -> Result<(), RealtimeError> {
121        let _ = ctx;
122        Ok(())
123    }
124}
125
126/// Poke a room from outside a socket — a module's HTTP handler (e.g. an admin
127/// ends a session) or a cron. The socket lifecycle itself is the runtime's job
128/// (it calls [`RoomHandler`]); this is the part a module holds via [`Ports`].
129///
130/// [`Ports`]: crate::Ports
131#[async_trait]
132pub trait Realtime: Send + Sync {
133    /// Send `message` to every socket in `room_id`.
134    async fn broadcast(&self, room_id: &str, message: &[u8]) -> Result<(), RealtimeError>;
135
136    /// The members currently connected to `room_id`.
137    async fn members(&self, room_id: &str) -> Result<Vec<Member>, RealtimeError>;
138}
cratefield_core/ports/realtime.rs

cratefield_core/ports/
realtime.rs
