hey_sdk/services/
postings.rs

1//! Acting on a selection of postings — seen, moved, trashed, muted, filed, bubbled up —
2//! and following a box's changes feed.
3//!
4//! Every HEY posting endpoint is a bulk one, so the methods here take the ids of the
5//! postings to act on. An empty selection is refused before anything is sent.
6
7use url::Url;
8
9use crate::client::Response;
10use crate::error::Error;
11use crate::generated::routes;
12use crate::generated::types::{
13    AddPostingsToBoxGroupRequestContent, CreateFolderForPostingsRequestContent, DeletedPosting,
14    FilePostingsRequestContent, FolderPayload, GetBoxPostingChangesResponseContent,
15    MarkPostingsRequestContent, MovePostingsRequestContent, Posting,
16    SchedulePostingsBubbleUpRequestContent, TrashPostingsRequestContent,
17};
18use crate::operation::Operation;
19use crate::pagination::next_link;
20use crate::route::Route;
21use crate::security::is_same_origin;
22use crate::services::boxes::BoxKind;
23use crate::types::Date;
24
25pub use crate::generated::services::postings::*;
26
27/// The status HEY answers when the cursor is too far behind for an increment to carry the
28/// difference: read the box in full instead.
29const TOO_FAR_BEHIND: u16 = 409;
30
31/// When a posting bubbles back up.
32///
33/// HEY resurfaces a posting at its morning hour of the day the slot names —
34/// [`BubbleUpSlot::LaterToday`] at its evening hour of the current day instead — and reads
35/// both hours in UTC, like every hour it takes out of a JSON request.
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37#[non_exhaustive]
38pub enum BubbleUpSlot {
39    /// This evening. HEY's `today`.
40    LaterToday,
41    /// Tomorrow morning.
42    Tomorrow,
43    /// Saturday.
44    ThisWeekend,
45    /// Monday.
46    NextWeek,
47    /// A day of the caller's choosing. HEY does not refuse one that has already passed —
48    /// those postings bubble up on the next run of its scheduler.
49    Custom(Date),
50}
51
52impl BubbleUpSlot {
53    /// The slot as HEY's `slot` parameter names it.
54    pub fn as_str(&self) -> &'static str {
55        match self {
56            BubbleUpSlot::LaterToday => "today",
57            BubbleUpSlot::Tomorrow => "tomorrow",
58            BubbleUpSlot::ThisWeekend => "weekend",
59            BubbleUpSlot::NextWeek => "next_week",
60            BubbleUpSlot::Custom(_) => "custom",
61        }
62    }
63
64    fn date(self) -> Option<String> {
65        match self {
66            BubbleUpSlot::Custom(date) => Some(date.to_string()),
67            _ => None,
68        }
69    }
70}
71
72/// Where a read of a box's changes feed starts.
73///
74/// `since` is an ISO 8601 timestamp with milliseconds and is exclusive; `version` is the
75/// contract version the caller speaks. A box's `posting_changes_url` carries the pair to
76/// begin with — read it with [`PostingChangesCursor::from_url`] rather than picking the
77/// query apart.
78#[derive(Debug, Clone, Default, PartialEq, Eq)]
79pub struct PostingChangesCursor {
80    /// The instant the changes come after.
81    pub since: String,
82    /// The contract version the feed speaks, HEY's `v`.
83    pub version: Option<String>,
84    /// The page within an increment, while it has more than one.
85    pub page: Option<String>,
86    /// How many changes a page holds, when the URL named a size.
87    pub per_page: Option<String>,
88}
89
90impl PostingChangesCursor {
91    /// Reads a cursor out of a changes URL HEY issued, either a box's `posting_changes_url`
92    /// or a `Link` header the feed answered with.
93    pub fn from_url(changes_url: &str) -> Result<PostingChangesCursor, Error> {
94        let url = Url::parse(changes_url).map_err(|error| {
95            Error::usage(format!(
96                "failed to read changes URL {changes_url:?}: {error}"
97            ))
98        })?;
99        let mut cursor = PostingChangesCursor::default();
100        for (name, value) in url.query_pairs() {
101            match name.as_ref() {
102                "since" => cursor.since = value.into_owned(),
103                "v" => cursor.version = Some(value.into_owned()),
104                "page" => cursor.page = Some(value.into_owned()),
105                "per_page" => cursor.per_page = Some(value.into_owned()),
106                _ => {}
107            }
108        }
109        Ok(cursor)
110    }
111}
112
113/// Everything that happened to a box's postings since a cursor.
114///
115/// `next_page` is set while this increment has more pages to read now. `next_cursor` is set
116/// on the last page and is where the next read resumes; it is `None` when nothing changed,
117/// in which case the cursor that produced this page still stands. `full_sync_required` is
118/// set when the cursor is too far behind for an increment to carry the difference, and the
119/// box has to be read in full instead.
120#[derive(Debug, Clone, Default, PartialEq)]
121#[non_exhaustive]
122pub struct PostingChanges {
123    /// The postings that appeared in the box.
124    pub added: Vec<Posting>,
125    /// The postings that changed.
126    pub updated: Vec<Posting>,
127    /// The postings that left the box.
128    pub deleted: Vec<DeletedPosting>,
129    /// The next page of this increment, while it has one.
130    pub next_page: Option<PostingChangesCursor>,
131    /// Where the next read resumes, once the increment is read to its end.
132    pub next_cursor: Option<PostingChangesCursor>,
133    /// Whether the cursor is too far behind and the box has to be read in full.
134    pub full_sync_required: bool,
135}
136
137impl Postings<'_> {
138    /// Marks a selection of postings as seen.
139    pub async fn mark_postings_seen(&self, posting_ids: &[i64]) -> Result<(), Error> {
140        self.mark(&routes::MARK_POSTINGS_SEEN, posting_ids).await
141    }
142
143    /// Marks a selection of postings as unseen.
144    pub async fn mark_postings_unseen(&self, posting_ids: &[i64]) -> Result<(), Error> {
145        self.mark(&routes::MARK_POSTINGS_UNSEEN, posting_ids).await
146    }
147
148    /// Moves a selection of postings to a box. Box ids come from [`Boxes::list`].
149    ///
150    /// [`Boxes::list`]: crate::services::boxes::Boxes::list
151    pub async fn move_to_box(&self, box_id: i64, posting_ids: &[i64]) -> Result<(), Error> {
152        let mut operation = self.selection(&routes::MOVE_POSTINGS, posting_ids)?;
153        operation.json(&MovePostingsRequestContent {
154            posting_ids: posting_ids.to_vec(),
155            box_id,
156        })?;
157        self.client().send_unit(operation).await
158    }
159
160    /// Moves a selection of postings to the box of a kind, resolving the kind through
161    /// [`Boxes::id_by_kind`], which answers from the one reading of the box index the
162    /// client keeps. Only the first move by kind costs that read.
163    ///
164    /// [`Boxes::id_by_kind`]: crate::services::boxes::Boxes::id_by_kind
165    pub async fn move_to_kind(&self, kind: BoxKind, posting_ids: &[i64]) -> Result<(), Error> {
166        require_ids(posting_ids)?;
167        let box_id = self.client().boxes().id_by_kind(kind).await?;
168        self.move_to_box(box_id, posting_ids).await
169    }
170
171    /// Moves a selection of postings to the Imbox.
172    pub async fn move_to_imbox(&self, posting_ids: &[i64]) -> Result<(), Error> {
173        self.move_to_kind(BoxKind::Imbox, posting_ids).await
174    }
175
176    /// Moves a selection of postings to The Feed.
177    pub async fn move_to_feed(&self, posting_ids: &[i64]) -> Result<(), Error> {
178        self.move_to_kind(BoxKind::Feed, posting_ids).await
179    }
180
181    /// Moves a selection of postings to Set Aside.
182    pub async fn move_to_set_aside(&self, posting_ids: &[i64]) -> Result<(), Error> {
183        self.move_to_kind(BoxKind::SetAside, posting_ids).await
184    }
185
186    /// Moves a selection of postings to Reply Later.
187    pub async fn move_to_reply_later(&self, posting_ids: &[i64]) -> Result<(), Error> {
188        self.move_to_kind(BoxKind::ReplyLater, posting_ids).await
189    }
190
191    /// Moves a selection of postings to the Paper Trail.
192    pub async fn move_to_paper_trail(&self, posting_ids: &[i64]) -> Result<(), Error> {
193        self.move_to_kind(BoxKind::PaperTrail, posting_ids).await
194    }
195
196    /// Moves a selection of postings to the trash. On a shared topic HEY removes your own
197    /// access rather than trashing the thread for everybody on it.
198    pub async fn move_to_trash(&self, posting_ids: &[i64]) -> Result<(), Error> {
199        self.trash_selection(None, posting_ids).await
200    }
201
202    /// Moves a selection of postings to the trash, and trashes a shared topic for everybody
203    /// on it rather than only dropping your own access.
204    pub async fn trash_for_everyone(&self, posting_ids: &[i64]) -> Result<(), Error> {
205        self.trash_selection(Some("false"), posting_ids).await
206    }
207
208    /// Mutes a selection of postings, so their threads stop notifying.
209    pub async fn mute_postings(&self, posting_ids: &[i64]) -> Result<(), Error> {
210        self.mark(&routes::MUTE_POSTINGS, posting_ids).await
211    }
212
213    /// Unmutes a selection of postings.
214    pub async fn unmute_postings(&self, posting_ids: &[i64]) -> Result<(), Error> {
215        self.by_ids(&routes::UNMUTE_POSTINGS, posting_ids).await
216    }
217
218    /// Marks a selection of postings as spam. Past ten postings HEY hands the work to a
219    /// background job, so the call comes back before the postings have moved.
220    pub async fn mark_postings_spam(&self, posting_ids: &[i64]) -> Result<(), Error> {
221        self.mark(&routes::MARK_POSTINGS_SPAM, posting_ids).await
222    }
223
224    /// Files a selection of postings into an existing Set Aside group.
225    pub async fn add_postings_to_box_group(
226        &self,
227        box_id: i64,
228        box_group_id: i64,
229        posting_ids: &[i64],
230    ) -> Result<(), Error> {
231        let mut operation = self.selection(&routes::ADD_POSTINGS_TO_BOX_GROUP, posting_ids)?;
232        operation.json(&AddPostingsToBoxGroupRequestContent {
233            posting_ids: posting_ids.to_vec(),
234            box_id,
235            box_group_id,
236        })?;
237        self.client().send_unit(operation).await
238    }
239
240    /// Takes a selection of postings out of whatever Set Aside group they are in.
241    pub async fn remove_postings_from_box_group(&self, posting_ids: &[i64]) -> Result<(), Error> {
242        self.by_ids(&routes::REMOVE_POSTINGS_FROM_BOX_GROUP, posting_ids)
243            .await
244    }
245
246    /// Labels a selection of postings with an existing folder.
247    pub async fn file_postings(&self, folder_id: i64, posting_ids: &[i64]) -> Result<(), Error> {
248        let mut operation = self.selection(&routes::FILE_POSTINGS, posting_ids)?;
249        operation.json(&FilePostingsRequestContent {
250            posting_ids: posting_ids.to_vec(),
251            folder_id,
252        })?;
253        self.client().send_unit(operation).await
254    }
255
256    /// Takes a label off a selection of postings, or every label when `folder_id` is 0.
257    ///
258    /// Zero is not "every folder" to HEY, it is a folder that does not exist, so it is left
259    /// out of the request rather than sent.
260    pub async fn unfile_postings(&self, folder_id: i64, posting_ids: &[i64]) -> Result<(), Error> {
261        let mut operation = self.selection(&routes::UNFILE_POSTINGS, posting_ids)?;
262        operation.query("posting_ids", join_ids(posting_ids));
263        if folder_id != 0 {
264            operation.query("folder_id", folder_id);
265        }
266        self.client().send_unit(operation).await
267    }
268
269    /// Creates a folder and files a selection of postings into it. HEY serves no JSON
270    /// endpoint for creating a folder on its own.
271    pub async fn create_folder_for_postings(
272        &self,
273        name: &str,
274        posting_ids: &[i64],
275    ) -> Result<(), Error> {
276        let mut operation = self.selection(&routes::CREATE_FOLDER_FOR_POSTINGS, posting_ids)?;
277        operation.json(&CreateFolderForPostingsRequestContent {
278            posting_ids: posting_ids.to_vec(),
279            folder: FolderPayload {
280                name: name.to_string(),
281                status: None,
282            },
283        })?;
284        self.client().send_unit(operation).await
285    }
286
287    /// Bubbles a selection of postings up right away.
288    pub async fn bubble_up_postings_now(&self, posting_ids: &[i64]) -> Result<(), Error> {
289        self.mark(&routes::BUBBLE_UP_POSTINGS_NOW, posting_ids)
290            .await
291    }
292
293    /// Schedules a selection of postings to bubble back up at a slot.
294    pub async fn schedule_postings_bubble_up(
295        &self,
296        slot: BubbleUpSlot,
297        posting_ids: &[i64],
298    ) -> Result<(), Error> {
299        let mut operation = self.selection(&routes::SCHEDULE_POSTINGS_BUBBLE_UP, posting_ids)?;
300        operation.json(&SchedulePostingsBubbleUpRequestContent {
301            posting_ids: posting_ids.to_vec(),
302            slot: slot.as_str().to_string(),
303            date: slot.date(),
304        })?;
305        self.client().send_unit(operation).await
306    }
307
308    /// Drops the scheduled bubble up on a selection of postings.
309    pub async fn cancel_postings_bubble_up(&self, posting_ids: &[i64]) -> Result<(), Error> {
310        self.by_ids(&routes::CANCEL_POSTINGS_BUBBLE_UP, posting_ids)
311            .await
312    }
313
314    /// Reads a box's changes feed from a cursor to the end of the increment, following the
315    /// pages HEY names.
316    ///
317    /// A full sync comes back as soon as HEY asks for one, with whatever was read before it
318    /// dropped: the box has to be read in full anyway. Reading stops at the client's page
319    /// limit, and the answer then carries the cursor of the last page read rather than the
320    /// end of the feed.
321    pub async fn all_changes(
322        &self,
323        box_id: i64,
324        cursor: &PostingChangesCursor,
325    ) -> Result<PostingChanges, Error> {
326        let mut all = PostingChanges::default();
327        let mut cursor = cursor.clone();
328        for _ in 0..self.client().max_pages() {
329            let mut changes = self.changes(box_id, &cursor).await?;
330            if changes.full_sync_required {
331                return Ok(changes);
332            }
333            all.added.append(&mut changes.added);
334            all.updated.append(&mut changes.updated);
335            all.deleted.append(&mut changes.deleted);
336            all.next_cursor = changes.next_cursor;
337            match changes.next_page {
338                Some(next) => cursor = next,
339                None => return Ok(all),
340            }
341        }
342        crate::trace::warning!(
343            max_pages = self.client().max_pages(),
344            "posting changes pagination capped"
345        );
346        Ok(all)
347    }
348
349    /// Reads one page of what changed among a box's postings since a cursor.
350    ///
351    /// This is the incremental sync feed the mail clients follow rather than re-reading a
352    /// box. HEY answers 409 when the cursor is too far behind for an increment to carry the
353    /// difference, which comes back as a `full_sync_required` answer rather than a failure:
354    /// read the box in full instead. The hooks still see the 409 for what it was.
355    pub async fn changes(
356        &self,
357        box_id: i64,
358        cursor: &PostingChangesCursor,
359    ) -> Result<PostingChanges, Error> {
360        if cursor.since.is_empty() {
361            return Err(Error::usage(
362                "a since cursor is required — start from the box's posting_changes_url",
363            ));
364        }
365
366        let mut operation = self
367            .client()
368            .operation(&routes::GET_BOX_POSTING_CHANGES, &[&box_id]);
369        operation.resource_id(box_id);
370        operation.query("since", &cursor.since);
371        operation.query_optional("v", cursor.version.as_ref());
372        operation.query_optional("page", cursor.page.as_ref());
373        operation.query_optional("per_page", cursor.per_page.as_ref());
374        // A cursor URL never repeats, so a cached answer would never be revalidated and a
375        // long-running watch would grow the cache one dead entry per read.
376        operation.no_cache();
377
378        let response = match self.client().execute(operation).await {
379            Ok(response) => response,
380            Err(error) if error.http_status() == Some(TOO_FAR_BEHIND) => {
381                return Ok(PostingChanges {
382                    full_sync_required: true,
383                    ..PostingChanges::default()
384                });
385            }
386            Err(error) => return Err(error),
387        };
388
389        let body: GetBoxPostingChangesResponseContent = response.json()?;
390        let mut changes = PostingChanges {
391            added: body.added.unwrap_or_default(),
392            updated: body.updated.unwrap_or_default(),
393            deleted: body.deleted.unwrap_or_default(),
394            ..PostingChanges::default()
395        };
396        if let Some(next) = self.next_cursor(&response)? {
397            // The feed names a page while an increment has more of them to read, and a
398            // fresh since cursor on the last one.
399            if next.page.is_some() {
400                changes.next_page = Some(next);
401            } else {
402                changes.next_cursor = Some(next);
403            }
404        }
405        Ok(changes)
406    }
407
408    async fn mark(&self, route: &'static Route, posting_ids: &[i64]) -> Result<(), Error> {
409        let mut operation = self.selection(route, posting_ids)?;
410        operation.json(&MarkPostingsRequestContent {
411            posting_ids: posting_ids.to_vec(),
412        })?;
413        self.client().send_unit(operation).await
414    }
415
416    /// Sends the selection in the query, comma-joined, for the endpoints whose method
417    /// carries no body.
418    async fn by_ids(&self, route: &'static Route, posting_ids: &[i64]) -> Result<(), Error> {
419        let mut operation = self.selection(route, posting_ids)?;
420        operation.query("posting_ids", join_ids(posting_ids));
421        self.client().send_unit(operation).await
422    }
423
424    /// A `remove_access` of `None` is left out of the request, which HEY reads as removing
425    /// only your own access from a shared topic.
426    async fn trash_selection(
427        &self,
428        remove_access: Option<&str>,
429        posting_ids: &[i64],
430    ) -> Result<(), Error> {
431        let mut operation = self.selection(&routes::TRASH_POSTINGS, posting_ids)?;
432        operation.json(&TrashPostingsRequestContent {
433            posting_ids: posting_ids.to_vec(),
434            remove_access: remove_access.map(str::to_string),
435        })?;
436        self.client().send_unit(operation).await
437    }
438
439    /// The operation for a bulk endpoint: an empty selection is refused before anything is
440    /// sent, and a selection of one names the posting it acts on.
441    fn selection(&self, route: &'static Route, posting_ids: &[i64]) -> Result<Operation, Error> {
442        require_ids(posting_ids)?;
443        let mut operation = self.client().operation(route, &[]);
444        if let [posting_id] = posting_ids {
445            operation.resource_id(*posting_id);
446        }
447        Ok(operation)
448    }
449
450    fn next_cursor(&self, response: &Response) -> Result<Option<PostingChangesCursor>, Error> {
451        match response.header("link").and_then(next_link) {
452            None => Ok(None),
453            Some(target) => {
454                let next = response.url.join(&target)?;
455                if is_same_origin(&next, self.client().base_url()) {
456                    PostingChangesCursor::from_url(next.as_str()).map(Some)
457                } else {
458                    Err(Error::usage(format!(
459                        "changes Link header points to a different origin: {next}"
460                    )))
461                }
462            }
463        }
464    }
465}
466
467fn require_ids(posting_ids: &[i64]) -> Result<(), Error> {
468    if posting_ids.is_empty() {
469        Err(Error::usage("at least one posting ID is required"))
470    } else {
471        Ok(())
472    }
473}
474
475fn join_ids(posting_ids: &[i64]) -> String {
476    posting_ids
477        .iter()
478        .map(i64::to_string)
479        .collect::<Vec<_>>()
480        .join(",")
481}
hey_sdk/services/postings.rs

hey_sdk/services/
postings.rs
