ably_chat/
messages.rs

1//! The messages handle and message read operations (ADR-0010).
2
3use std::collections::BTreeMap;
4use std::future::{Future, IntoFuture};
5use std::pin::Pin;
6
7use futures::Stream;
8use reqwest::Method;
9use serde_json::{Map, Value};
10
11use crate::client::Client;
12use crate::dispatch::{decode_json, message_path, room_path};
13use crate::error::Result;
14use crate::pagination::{Fetch, Page, run_stream};
15use crate::reactions::Reactions;
16use crate::types::{Direction, Message, Metadata, RoomName, Serial, Timestamp};
17
18/// Converts a string→string map into a JSON object value (infallible).
19fn string_map(map: &BTreeMap<String, String>) -> Value {
20    Value::Object(
21        map.iter()
22            .map(|(k, v)| (k.clone(), Value::String(v.clone())))
23            .collect(),
24    )
25}
26
27/// Builds the idempotency-key query and the retry-eligibility flag (ADR-0006):
28/// a write is only retry-safe when the caller supplied an idempotency key.
29fn idempotency(key: &Option<String>) -> (Vec<(&'static str, String)>, bool) {
30    match key {
31        Some(k) => (vec![("idempotencyKey", k.clone())], true),
32        None => (Vec::new(), false),
33    }
34}
35
36/// Message operations for a room.
37///
38/// Cheap to `Clone` (`Arc`-backed via [`Client`]) and `Send + Sync`.
39#[derive(Clone, Debug)]
40pub struct Messages {
41    pub(crate) client: Client,
42    pub(crate) room: RoomName,
43}
44
45impl Messages {
46    pub(crate) fn new(client: Client, room: RoomName) -> Self {
47        Self { client, room }
48    }
49
50    /// Reaction operations on messages in this room.
51    pub fn reactions(&self) -> Reactions {
52        Reactions::new(self.client.clone(), self.room.clone())
53    }
54
55    /// Sends a new message to this room.
56    ///
57    /// `POST /chat/v4/rooms/{roomName}/messages`. Only retry-safe when an
58    /// [`idempotency_key`](SendMessage::idempotency_key) is supplied (ADR-0006).
59    pub fn send(&self, text: impl Into<String>) -> SendMessage {
60        SendMessage {
61            client: self.client.clone(),
62            room: self.room.clone(),
63            text: text.into(),
64            metadata: None,
65            headers: None,
66            idempotency_key: None,
67        }
68    }
69
70    /// Fetches a single message by its serial (latest version).
71    ///
72    /// `GET /chat/v4/rooms/{roomName}/messages/{serial}`. Retry-safe.
73    pub fn get(&self, serial: impl Into<Serial>) -> GetMessage {
74        GetMessage {
75            client: self.client.clone(),
76            room: self.room.clone(),
77            serial: serial.into(),
78        }
79    }
80
81    /// Updates (edits) a message, **fully replacing** its content.
82    ///
83    /// `PUT /chat/v4/rooms/{roomName}/messages/{serial}`. This is a
84    /// **full replace**: the supplied `text` and any
85    /// [`metadata`](UpdateMessage::metadata) /
86    /// [`headers`](UpdateMessage::headers) become the message's entire new
87    /// content. Omitted fields are **reset to empty**, not left unchanged — to
88    /// preserve existing metadata/headers you must resend them. Produces a new
89    /// version with action `message.update`. Only retry-safe when an
90    /// [`idempotency_key`](UpdateMessage::idempotency_key) is supplied (ADR-0006).
91    pub fn update(&self, serial: impl Into<Serial>, text: impl Into<String>) -> UpdateMessage {
92        UpdateMessage {
93            client: self.client.clone(),
94            room: self.room.clone(),
95            serial: serial.into(),
96            text: text.into(),
97            metadata: None,
98            headers: None,
99            description: None,
100            idempotency_key: None,
101        }
102    }
103
104    /// Soft-deletes a message.
105    ///
106    /// `POST /chat/v4/rooms/{roomName}/messages/{serial}/delete` — a `POST` to a
107    /// `/delete` sub-resource, **not** an HTTP `DELETE`. Produces a new version
108    /// with action `message.delete`; the message remains retrievable with its
109    /// delete action applied. Only retry-safe when an
110    /// [`idempotency_key`](DeleteMessage::idempotency_key) is supplied (ADR-0006).
111    pub fn delete(&self, serial: impl Into<Serial>) -> DeleteMessage {
112        DeleteMessage {
113            client: self.client.clone(),
114            room: self.room.clone(),
115            serial: serial.into(),
116            description: None,
117            metadata: None,
118            idempotency_key: None,
119        }
120    }
121
122    /// Queries message history.
123    ///
124    /// `GET /chat/v4/rooms/{roomName}/messages`, paginated. Retry-safe. Defaults
125    /// to **newest first** (`direction = backwards`) and `limit = 100`, matching
126    /// the JS SDK. `.await` for the first [`Page<Message>`], or
127    /// [`into_stream`](History::into_stream) to follow all pages.
128    pub fn history(&self) -> History {
129        History {
130            client: self.client.clone(),
131            room: self.room.clone(),
132            start: None,
133            end: None,
134            direction: Direction::Backwards,
135            limit: 100,
136            from_serial: None,
137        }
138    }
139
140    /// Queries all versions (create, updates, deletes) of a message.
141    ///
142    /// `GET /chat/v4/rooms/{roomName}/messages/{serial}/versions`, paginated.
143    /// Retry-safe. `.await` for the first [`Page<Message>`], or
144    /// [`into_stream`](Versions::into_stream) to follow all pages.
145    pub fn versions(&self, serial: impl Into<Serial>) -> Versions {
146        Versions {
147            client: self.client.clone(),
148            room: self.room.clone(),
149            serial: serial.into(),
150        }
151    }
152}
153
154/// Builder for [`Messages::get`]; `.await` it to fetch a [`Message`].
155#[derive(Clone, Debug)]
156pub struct GetMessage {
157    client: Client,
158    room: RoomName,
159    serial: Serial,
160}
161
162impl IntoFuture for GetMessage {
163    type Output = Result<Message>;
164    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send>>;
165
166    fn into_future(self) -> Self::IntoFuture {
167        Box::pin(async move {
168            let resp = self
169                .client
170                .inner
171                .send(
172                    Method::GET,
173                    &message_path(self.room.as_str(), self.serial.as_str(), ""),
174                    &[],
175                    None,
176                    false,
177                )
178                .await?;
179            decode_json(&resp.body)
180        })
181    }
182}
183
184/// Builder for [`Messages::send`]; `.await` it to publish the message and
185/// receive the created [`Message`].
186#[derive(Clone, Debug)]
187pub struct SendMessage {
188    client: Client,
189    room: RoomName,
190    text: String,
191    metadata: Option<Metadata>,
192    headers: Option<BTreeMap<String, String>>,
193    idempotency_key: Option<String>,
194}
195
196impl SendMessage {
197    /// Attaches opaque user-defined metadata to the message.
198    pub fn metadata(mut self, metadata: Metadata) -> Self {
199        self.metadata = Some(metadata);
200        self
201    }
202
203    /// Attaches user-defined string headers to the message.
204    pub fn headers(mut self, headers: BTreeMap<String, String>) -> Self {
205        self.headers = Some(headers);
206        self
207    }
208
209    /// Supplies an idempotency key, making the send safe to retry (ADR-0006).
210    pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
211        self.idempotency_key = Some(key.into());
212        self
213    }
214}
215
216impl IntoFuture for SendMessage {
217    type Output = Result<Message>;
218    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send>>;
219
220    fn into_future(self) -> Self::IntoFuture {
221        Box::pin(async move {
222            let mut obj = Map::new();
223            obj.insert("text".to_owned(), Value::String(self.text));
224            if let Some(metadata) = self.metadata {
225                obj.insert("metadata".to_owned(), Value::Object(metadata));
226            }
227            if let Some(headers) = &self.headers {
228                obj.insert("headers".to_owned(), string_map(headers));
229            }
230            let (query, has_idem) = idempotency(&self.idempotency_key);
231            let resp = self
232                .client
233                .inner
234                .send(
235                    Method::POST,
236                    &room_path(self.room.as_str(), "/messages"),
237                    &query,
238                    Some(Value::Object(obj)),
239                    has_idem,
240                )
241                .await?;
242            decode_json(&resp.body)
243        })
244    }
245}
246
247/// Builder for [`Messages::update`]; `.await` it to apply the edit and receive
248/// the updated [`Message`].
249///
250/// Update semantics are **full-replace**: see [`Messages::update`].
251#[derive(Clone, Debug)]
252pub struct UpdateMessage {
253    client: Client,
254    room: RoomName,
255    serial: Serial,
256    text: String,
257    metadata: Option<Metadata>,
258    headers: Option<BTreeMap<String, String>>,
259    description: Option<String>,
260    idempotency_key: Option<String>,
261}
262
263impl UpdateMessage {
264    /// Sets the message's new metadata. Omitting this resets metadata to empty
265    /// (full-replace; see [`Messages::update`]).
266    pub fn metadata(mut self, metadata: Metadata) -> Self {
267        self.metadata = Some(metadata);
268        self
269    }
270
271    /// Sets the message's new headers. Omitting this resets headers to empty
272    /// (full-replace; see [`Messages::update`]).
273    pub fn headers(mut self, headers: BTreeMap<String, String>) -> Self {
274        self.headers = Some(headers);
275        self
276    }
277
278    /// Attaches an optional description of the update operation.
279    pub fn description(mut self, description: impl Into<String>) -> Self {
280        self.description = Some(description.into());
281        self
282    }
283
284    /// Supplies an idempotency key, making the update safe to retry (ADR-0006).
285    pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
286        self.idempotency_key = Some(key.into());
287        self
288    }
289}
290
291impl IntoFuture for UpdateMessage {
292    type Output = Result<Message>;
293    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send>>;
294
295    fn into_future(self) -> Self::IntoFuture {
296        Box::pin(async move {
297            let mut message = Map::new();
298            message.insert("text".to_owned(), Value::String(self.text));
299            if let Some(metadata) = self.metadata {
300                message.insert("metadata".to_owned(), Value::Object(metadata));
301            }
302            if let Some(headers) = &self.headers {
303                message.insert("headers".to_owned(), string_map(headers));
304            }
305            let mut obj = Map::new();
306            obj.insert("message".to_owned(), Value::Object(message));
307            if let Some(description) = self.description {
308                obj.insert("description".to_owned(), Value::String(description));
309            }
310            let (query, has_idem) = idempotency(&self.idempotency_key);
311            let resp = self
312                .client
313                .inner
314                .send(
315                    Method::PUT,
316                    &message_path(self.room.as_str(), self.serial.as_str(), ""),
317                    &query,
318                    Some(Value::Object(obj)),
319                    has_idem,
320                )
321                .await?;
322            decode_json(&resp.body)
323        })
324    }
325}
326
327/// Builder for [`Messages::delete`]; `.await` it to soft-delete the message and
328/// receive the resulting [`Message`] (action `message.delete`).
329#[derive(Clone, Debug)]
330pub struct DeleteMessage {
331    client: Client,
332    room: RoomName,
333    serial: Serial,
334    description: Option<String>,
335    metadata: Option<BTreeMap<String, String>>,
336    idempotency_key: Option<String>,
337}
338
339impl DeleteMessage {
340    /// Attaches an optional description of the delete operation.
341    pub fn description(mut self, description: impl Into<String>) -> Self {
342        self.description = Some(description.into());
343        self
344    }
345
346    /// Attaches optional string metadata describing the delete operation.
347    pub fn metadata(mut self, metadata: BTreeMap<String, String>) -> Self {
348        self.metadata = Some(metadata);
349        self
350    }
351
352    /// Supplies an idempotency key, making the delete safe to retry (ADR-0006).
353    pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
354        self.idempotency_key = Some(key.into());
355        self
356    }
357}
358
359impl IntoFuture for DeleteMessage {
360    type Output = Result<Message>;
361    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send>>;
362
363    fn into_future(self) -> Self::IntoFuture {
364        Box::pin(async move {
365            let mut obj = Map::new();
366            if let Some(description) = self.description {
367                obj.insert("description".to_owned(), Value::String(description));
368            }
369            if let Some(metadata) = &self.metadata {
370                obj.insert("metadata".to_owned(), string_map(metadata));
371            }
372            // The request body is optional; omit it entirely when unset.
373            let body = if obj.is_empty() {
374                None
375            } else {
376                Some(Value::Object(obj))
377            };
378            let (query, has_idem) = idempotency(&self.idempotency_key);
379            let resp = self
380                .client
381                .inner
382                .send(
383                    Method::POST,
384                    &message_path(self.room.as_str(), self.serial.as_str(), "/delete"),
385                    &query,
386                    body,
387                    has_idem,
388                )
389                .await?;
390            decode_json(&resp.body)
391        })
392    }
393}
394
395/// Serializes a [`Direction`] to its wire query value.
396fn direction_str(d: Direction) -> &'static str {
397    match d {
398        Direction::Forwards => "forwards",
399        Direction::Backwards => "backwards",
400    }
401}
402
403/// Builder for [`Messages::history`]. `.await` yields the first
404/// [`Page<Message>`]; [`into_stream`](Self::into_stream) follows all pages.
405#[derive(Clone, Debug)]
406pub struct History {
407    client: Client,
408    room: RoomName,
409    start: Option<i64>,
410    end: Option<i64>,
411    direction: Direction,
412    limit: u32,
413    from_serial: Option<Serial>,
414}
415
416impl History {
417    /// Earliest timestamp to include (epoch millis; inclusive).
418    pub fn start(mut self, start: impl Into<Timestamp>) -> Self {
419        self.start = Some(start.into().as_millis());
420        self
421    }
422
423    /// Latest timestamp to include (epoch millis; exclusive).
424    pub fn end(mut self, end: impl Into<Timestamp>) -> Self {
425        self.end = Some(end.into().as_millis());
426        self
427    }
428
429    /// Ordering. Defaults to [`Direction::Backwards`] (newest first).
430    pub fn direction(mut self, direction: Direction) -> Self {
431        self.direction = direction;
432        self
433    }
434
435    /// Maximum messages per page (1..=1000). Defaults to `100`.
436    pub fn limit(mut self, limit: u32) -> Self {
437        self.limit = limit;
438        self
439    }
440
441    /// Region-scoped serial to page from.
442    pub fn from_serial(mut self, serial: impl Into<Serial>) -> Self {
443        self.from_serial = Some(serial.into());
444        self
445    }
446
447    /// Builds the query for the first request. `direction` and `limit` are
448    /// always sent so the effective defaults are explicit on the wire.
449    fn query(&self) -> Vec<(&'static str, String)> {
450        let mut query: Vec<(&'static str, String)> = Vec::new();
451        if let Some(start) = self.start {
452            query.push(("start", start.to_string()));
453        }
454        if let Some(end) = self.end {
455            query.push(("end", end.to_string()));
456        }
457        query.push(("direction", direction_str(self.direction).to_owned()));
458        query.push(("limit", self.limit.to_string()));
459        if let Some(from_serial) = &self.from_serial {
460            query.push(("fromSerial", from_serial.as_str().to_owned()));
461        }
462        query
463    }
464
465    /// Streams every message across all history pages, following `next` links
466    /// until exhausted.
467    pub fn into_stream(self) -> impl Stream<Item = Result<Message>> + Send {
468        let path = room_path(self.room.as_str(), "/messages");
469        let query = self.query();
470        run_stream(self.client, Vec::new(), Fetch::First { path, query })
471    }
472}
473
474impl IntoFuture for History {
475    type Output = Result<Page<Message>>;
476    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send>>;
477
478    fn into_future(self) -> Self::IntoFuture {
479        Box::pin(async move {
480            let path = room_path(self.room.as_str(), "/messages");
481            let query = self.query();
482            Page::fetch_first(self.client, path, query).await
483        })
484    }
485}
486
487/// Builder for [`Messages::versions`]. `.await` yields the first
488/// [`Page<Message>`]; [`into_stream`](Self::into_stream) follows all pages.
489#[derive(Clone, Debug)]
490pub struct Versions {
491    client: Client,
492    room: RoomName,
493    serial: Serial,
494}
495
496impl Versions {
497    fn path(&self) -> String {
498        message_path(self.room.as_str(), self.serial.as_str(), "/versions")
499    }
500
501    /// Streams every version across all pages, following `next` links until
502    /// exhausted.
503    pub fn into_stream(self) -> impl Stream<Item = Result<Message>> + Send {
504        let path = self.path();
505        run_stream(
506            self.client,
507            Vec::new(),
508            Fetch::First {
509                path,
510                query: Vec::new(),
511            },
512        )
513    }
514}
515
516impl IntoFuture for Versions {
517    type Output = Result<Page<Message>>;
518    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send>>;
519
520    fn into_future(self) -> Self::IntoFuture {
521        Box::pin(async move {
522            let path = self.path();
523            Page::fetch_first(self.client, path, Vec::new()).await
524        })
525    }
526}
ably_chat/messages.rs

ably_chat/
messages.rs
