cratestack_sqlx/
topic.rs

1//! A schema `topic`'s runtime handle on `db = Postgres`
2//! (`docs/design/subscriptions-websocket.md` §3.4, slice S4): the
3//! backplane-only [`TopicHandle`] from `cratestack-core`, plus
4//! `publish_in_tx`, which sends from inside a transaction so the message is
5//! delivered only if it commits, in commit order with the transaction's model
6//! events. It needs a backplane the transaction can notify
7//! (`PgNotifyBackplane`, the default).
8//!
9//! A topic taken from an `@isolation` procedure's handle is bound to the
10//! procedure's attempt: its `publish` is delivered only if the attempt
11//! commits, whichever backplane carries it (`PgNotifyBackplane` notifies
12//! inside the transaction, any other sends after the commit).
13
14use std::fmt::Display;
15use std::ops::Deref;
16use std::sync::Arc;
17
18use cratestack_core::{CratestackError, SubscriptionBackplane, TopicHandle};
19use serde::Serialize;
20use serde::de::DeserializeOwned;
21
22use crate::SqlxRuntime;
23use crate::backplane::notify_payload::{NOTIFY_CHANNEL, encode};
24use crate::bound::{BoundTx, nested_in_bound};
25use crate::error::cratestack_error_from_sqlx;
26use crate::sqlx;
27
28/// One topic under one key. Cheap to build; hold none. `subscribe` and
29/// `channel` are the [`TopicHandle`]'s; `publish` is this type's own, so a
30/// topic bound to an `@isolation` attempt rides it.
31pub struct Topic<T> {
32    handle: TopicHandle<T>,
33    /// The `@isolation` attempt a `publish` rides, when there is one.
34    attempt: Option<Arc<BoundTx>>,
35}
36
37impl<T> Deref for Topic<T> {
38    type Target = TopicHandle<T>;
39
40    fn deref(&self) -> &Self::Target {
41        &self.handle
42    }
43}
44
45impl<T> Topic<T>
46where
47    T: Serialize + DeserializeOwned + Send + 'static,
48{
49    /// `topic` is the declared name (`RoomMessage`), `key` the key's value.
50    pub fn new(backplane: Arc<dyn SubscriptionBackplane>, topic: &str, key: impl Display) -> Self {
51        Self {
52            handle: TopicHandle::new(backplane, topic, key),
53            attempt: None,
54        }
55    }
56
57    /// Binds `publish` to `runtime`'s `@isolation` attempt. A runtime that
58    /// is not one, or none, leaves the topic as it was.
59    #[doc(hidden)]
60    pub fn in_attempt(mut self, runtime: Option<&SqlxRuntime>) -> Self {
61        self.attempt = runtime.and_then(|runtime| runtime.bound.clone());
62        self
63    }
64
65    /// Delivers `message` to every current subscriber, on every replica.
66    /// From an `@isolation` procedure's handle it is delivered only if the
67    /// attempt commits, never for one that rolls back or retries: with
68    /// `PgNotifyBackplane` as [`Self::publish_in_tx`] in a savepoint of the
69    /// attempt, with any other backplane queued and sent after the commit.
70    /// Inside `db.transaction(..)` there, use `publish_in_tx(tx, ..)`.
71    pub async fn publish(&self, message: &T) -> Result<(), CratestackError> {
72        let Some(attempt) = &self.attempt else {
73            return self.handle.publish(message).await;
74        };
75        let payload = self.encode(message)?;
76        if !self.backplane().publishes_model_events_in_tx() {
77            attempt.defer_publish(self.backplane().clone(), self.channel().to_owned(), payload);
78            return Ok(());
79        }
80        attempt.lock().map(drop).map_err(|_| {
81            CratestackError::Internal(
82                "a topic publish is refused while `db.transaction(..)` runs on this handle: \
83                 call `publish_in_tx(tx, &message)` with its `tx`"
84                    .to_owned(),
85            )
86        })?;
87        nested_in_bound(attempt, async |tx| self.notify(tx, &payload).await).await
88    }
89
90    /// Delivers `message` only if `tx` commits, when it commits. On a topic
91    /// from an `@isolation` procedure's handle, a backplane the transaction
92    /// cannot notify gets it queued until the attempt commits (dropped if
93    /// this `db.transaction(..)` rolls back). Otherwise such a backplane
94    /// (`InProcessBackplane`) is refused: call [`Self::publish`] after the
95    /// commit instead.
96    pub async fn publish_in_tx(
97        &self,
98        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
99        message: &T,
100    ) -> Result<(), CratestackError> {
101        if !self.backplane().publishes_model_events_in_tx() {
102            if let Some(attempt) = &self.attempt {
103                let payload = self.encode(message)?;
104                attempt.defer_publish(self.backplane().clone(), self.channel().to_owned(), payload);
105                return Ok(());
106            }
107            return Err(CratestackError::Internal(
108                "`publish_in_tx` needs a backplane the transaction notifies (PgNotifyBackplane); \
109                 with this one, `publish` after the commit"
110                    .to_owned(),
111            ));
112        }
113        self.notify(tx, &self.encode(message)?).await
114    }
115
116    /// `pg_notify` of `payload` inside `tx`.
117    async fn notify(
118        &self,
119        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
120        payload: &str,
121    ) -> Result<(), CratestackError> {
122        let notification = encode(self.channel(), payload, None).ok_or_else(|| {
123            CratestackError::BadRequest(
124                "a topic message must stay under Postgres' 8000-byte NOTIFY limit".to_owned(),
125            )
126        })?;
127        sqlx::query("SELECT pg_notify($1, $2)")
128            .bind(NOTIFY_CHANNEL)
129            .bind(notification)
130            .execute(&mut **tx)
131            .await
132            .map_err(cratestack_error_from_sqlx)?;
133        Ok(())
134    }
135}
cratestack_sqlx/topic.rs

cratestack_sqlx/
topic.rs
