ruststream_sqlx/inbox/named/
row.rs

1//! The row a by-name subscription reads by role: its parts, how its statements bind, and how it
2//! is held to the types of the route's own struct.
3
4use std::fmt;
5use std::marker::PhantomData;
6use std::mem;
7
8#[cfg(feature = "chrono")]
9use chrono::{DateTime, Utc};
10use ruststream::HeaderMap;
11use sqlx::{Column, Database, Error, FromRow, Row, ValueRef};
12#[cfg(feature = "time")]
13use time::OffsetDateTime;
14
15use super::by_name::ByName;
16use super::database::RoleColumns;
17use super::kinds::{IntKind, Kinds};
18use crate::inbox::engine::{Claimed, undecodable};
19use crate::inbox::{PayloadLane, PayloadRow, QueueRow};
20
21/// A claimed row of a by-name subscription, read by the role aliases of `ClaimShape::Roles`, its
22/// headers and times read and bound by the dialect `D`. Machinery: a by-name subscription
23/// delivers it in place of the route's own row.
24#[doc(hidden)]
25pub struct NamedRow<D> {
26    id: NamedId,
27    payload: NamedBytes,
28    /// Taken at decode: the alias `headers`, read as JSON.
29    headers: HeaderMap,
30    key: Option<NamedBytes>,
31    attempt: Option<u64>,
32    fits: Fits,
33    dialect: PhantomData<fn() -> D>,
34}
35
36impl<D> fmt::Debug for NamedRow<D> {
37    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
38        f.debug_struct("NamedRow")
39            .field("id", &self.id)
40            .field("payload", &self.payload)
41            .field("headers", &self.headers)
42            .field("key", &self.key)
43            .field("attempt", &self.attempt)
44            .field("fits", &self.fits)
45            .finish()
46    }
47}
48
49/// Which of the types a by-name subscription reads each role column holds, a bit per type in the
50/// order it tries them: what the types the struct reads are held to.
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
52struct Fits {
53    id: u8,
54    payload: u8,
55    /// Every bit for a null, which an optional key reads whatever the column's type.
56    key: u8,
57    attempt: u8,
58}
59
60impl<D> NamedRow<D> {
61    /// The role column whose type in the table does not hold the type the struct reads it as,
62    /// and that type, if one does not.
63    fn unfit(&self, kinds: Kinds) -> Option<(&'static str, &'static str)> {
64        if self.fits.id & kinds.id.bit() == 0 {
65            return Some(("id", kinds.id.name()));
66        }
67        if self.fits.payload & kinds.payload.bit() == 0 {
68            return Some(("payload", kinds.payload.name()));
69        }
70        if let Some(key) = kinds.key
71            && self.fits.key & key.bit() == 0
72        {
73            return Some(("partition_key", key.name()));
74        }
75        if let Some(attempt) = kinds.attempt
76            && !attempt_fits(attempt, self.fits.attempt)
77        {
78            return Some(("attempt", attempt.name()));
79        }
80        None
81    }
82
83    /// The row's attempt where its column holds the type the struct reads it as, as the struct's
84    /// own `FromRow` would read it.
85    fn fitting_attempt(&self, kinds: Kinds) -> Option<u64> {
86        let kind = kinds.attempt?;
87        self.attempt
88            .filter(|_| attempt_fits(kind, self.fits.attempt))
89    }
90}
91
92/// Whether an attempt column that holds the types `held` holds `kind`.
93const fn attempt_fits(kind: IntKind, held: u8) -> bool {
94    held & kind.bit() != 0
95}
96
97/// Holds a claimed row to the types its struct reads, as the struct's own `FromRow` would: a row
98/// whose role column does not hold its type goes to the decode-failure policy, and an id that
99/// does not fails the claim.
100fn hold_to<D: 'static>(kinds: Kinds, claimed: &mut Claimed<NamedRow<D>>) -> Result<(), Error> {
101    let Claimed::Row(row) = claimed else {
102        return Ok(());
103    };
104    let Some((alias, read_as)) = row.unfit(kinds) else {
105        return Ok(());
106    };
107    let error = Error::ColumnDecode {
108        index: format!("{alias:?}"),
109        source: format!(
110            "the struct reads this column as {read_as}, which its type in the table does not hold"
111        )
112        .into(),
113    };
114    if alias == "id" {
115        return Err(error);
116    }
117    // The row is dropped in place of its delivery; its id moves into the one that replaces it.
118    let attempt = row.fitting_attempt(kinds);
119    let id = mem::replace(&mut row.id, NamedId::I64(0));
120    *claimed = Claimed::Undecodable {
121        id,
122        attempt,
123        error: Box::new(undecodable(error)),
124    };
125    Ok(())
126}
127
128/// The id of a [`NamedRow`], in the type its column holds. Machinery.
129#[doc(hidden)]
130#[derive(PartialEq, Eq, Hash)]
131pub enum NamedId {
132    /// A `SMALLINT` id.
133    I16(i16),
134    /// An `INTEGER` id.
135    I32(i32),
136    /// A `BIGINT` id.
137    I64(i64),
138    /// A text id.
139    Text(String),
140    /// A byte id.
141    Bytes(Vec<u8>),
142}
143
144// Written out rather than derived: `clone_from` copies a text or byte id into the storage the id
145// it replaces held, which a lease book does for every delivery it enters.
146impl Clone for NamedId {
147    fn clone(&self) -> Self {
148        match self {
149            Self::I16(id) => Self::I16(*id),
150            Self::I32(id) => Self::I32(*id),
151            Self::I64(id) => Self::I64(*id),
152            Self::Text(id) => Self::Text(id.clone()),
153            Self::Bytes(id) => Self::Bytes(id.clone()),
154        }
155    }
156
157    fn clone_from(&mut self, source: &Self) {
158        match (self, source) {
159            (Self::Text(kept), Self::Text(id)) => kept.clone_from(id),
160            (Self::Bytes(kept), Self::Bytes(id)) => kept.clone_from(id),
161            (kept, id) => *kept = id.clone(),
162        }
163    }
164}
165
166// Written out rather than derived: logs name a row by its id as the table holds it, the way they
167// name a row of the route's own struct.
168impl fmt::Debug for NamedId {
169    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
170        match self {
171            Self::I16(id) => fmt::Debug::fmt(id, f),
172            Self::I32(id) => fmt::Debug::fmt(id, f),
173            Self::I64(id) => fmt::Debug::fmt(id, f),
174            Self::Text(id) => fmt::Debug::fmt(id, f),
175            Self::Bytes(id) => fmt::Debug::fmt(id, f),
176        }
177    }
178}
179
180/// A payload or a key of a [`NamedRow`], as bytes or as text. Machinery.
181#[doc(hidden)]
182#[derive(Debug, Clone, PartialEq, Eq, Hash)]
183pub enum NamedBytes {
184    /// A byte column.
185    Bytes(Vec<u8>),
186    /// A text column.
187    Text(String),
188}
189
190impl NamedBytes {
191    fn as_bytes(&self) -> &[u8] {
192        match self {
193            Self::Bytes(bytes) => bytes,
194            Self::Text(text) => text.as_bytes(),
195        }
196    }
197}
198
199/// A time a by-name subscription binds, in the type its column holds: the current time, a delayed
200/// retry's, or in the lease form a lease's expiry.
201///
202/// A dialect binds it in [`ByName::bind_time`](crate::ByName::bind_time). Each variant comes with
203/// the feature of its time type, so a match keeps a wildcard arm.
204///
205/// # Examples
206///
207/// ```
208/// # #[cfg(all(feature = "postgres", feature = "chrono"))]
209/// # mod demo {
210/// # use std::num::NonZeroUsize;
211/// # use ruststream::OutgoingMessage;
212/// # use ruststream_sqlx::dialect::{self, ClaimShape, Dialect, Opening, RowLock, Statement, StatementError, TableName, TableSpec};
213/// # use sqlx::PgConnection;
214/// use ruststream::HeaderMap;
215/// use ruststream_sqlx::prelude::*;
216/// use ruststream_sqlx::{BuiltIn, ByName, NamedTime};
217/// use serde::Deserialize;
218/// use sqlx::error::BoxDynError;
219/// use sqlx::postgres::{PgArguments, PgValueRef};
220/// use sqlx::{Arguments, PgPool, Postgres};
221///
222/// /// A dialect of the service's own over Postgres, whose tables keep `chrono` times.
223/// #[derive(Debug)]
224/// pub struct Audited;
225/// # impl Dialect for Audited {
226/// #     fn name(&self) -> &'static str { "audited" }
227/// #     fn quote_into(&self, ident: &str, out: &mut String) { dialect::Postgres.quote_into(ident, out) }
228/// #     fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { dialect::Postgres.placeholder_into(index, out) }
229/// #     fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.fetch(spec) }
230/// #     fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.ack(spec) }
231/// #     fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { dialect::Postgres.retry(spec) }
232/// #     fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.retry_after(spec) }
233/// #     fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.discard(spec) }
234/// #     fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.dead_letter_group(spec) }
235/// #     fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { dialect::Postgres.dead_letter_table(spec, target) }
236/// #     fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.insert(spec) }
237/// #     fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> { dialect::Postgres.begin(opening) }
238/// #     fn fifo_guard(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { dialect::Postgres.fifo_guard(spec) }
239/// # }
240/// # impl RowLock for Audited {
241/// #     fn lock_claim(&self, spec: &TableSpec<'_>, shape: ClaimShape) -> Result<Statement, StatementError> { dialect::Postgres.lock_claim(spec, shape) }
242/// # }
243///
244/// impl ByName<Postgres> for Audited {
245///     fn headers(value: PgValueRef<'_>) -> Result<HeaderMap, BoxDynError> {
246///         <BuiltIn<Postgres> as ByName<Postgres>>::headers(value)
247///     }
248///
249///     fn bind_time(arguments: &mut PgArguments, time: NamedTime) -> Result<(), sqlx::Error> {
250///         match time {
251///             NamedTime::Chrono(at) => arguments.add(at).map_err(sqlx::Error::Encode),
252///             _ => Err(sqlx::Error::Configuration("these tables keep `chrono` times".into())),
253///         }
254///     }
255/// }
256///
257/// #[derive(Inbox, sqlx::FromRow)]
258/// #[inbox(table = "email_jobs")]
259/// pub struct SendEmail {
260///     #[field(id, generated)]
261///     job_id: i64,
262///     #[field(group)]
263///     name: String,
264///     #[field(payload)]
265///     payload: Vec<u8>,
266/// }
267/// # impl Publish<Postgres> for SendEmail {
268/// #     async fn publish(_: &mut PgConnection, _: &OutgoingMessage<'_>) -> Result<(), sqlx::Error> { Ok(()) }
269/// # }
270///
271/// #[derive(Deserialize)]
272/// struct Email {
273///     to: String,
274/// }
275///
276/// #[subscriber("emails")]
277/// async fn send(email: &Email) -> HandlerOutcome {
278///     tracing::info!(to = %email.to, "sending");
279///     HandlerOutcome::ack()
280/// }
281///
282/// pub fn app(pool: PgPool) -> RustStream {
283///     let broker = SqlxBroker::with_dialect(pool, Audited).route::<SendEmail>("emails");
284///     RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
285///         b.include(send);
286///     })
287/// }
288/// # }
289/// # fn main() {}
290/// ```
291#[derive(Debug, Clone, Copy, PartialEq, Eq)]
292#[non_exhaustive]
293pub enum NamedTime {
294    /// A `chrono` column.
295    #[cfg(feature = "chrono")]
296    Chrono(DateTime<Utc>),
297    /// A `time` column.
298    #[cfg(feature = "time")]
299    Time(OffsetDateTime),
300}
301
302/// Reads a claimed row by the role aliases of `ClaimShape::Roles`: one look at each column's name
303/// and one decode per column, as a struct's own `FromRow` does; the dialect `D` reads the headers.
304fn read<DB, D>(row: &DB::Row) -> Result<NamedRow<D>, Error>
305where
306    DB: RoleColumns,
307    D: ByName<DB>,
308{
309    let mut id = None;
310    let mut payload = None;
311    let mut headers = HeaderMap::new();
312    let mut key = None;
313    let mut attempt = None;
314    let mut fits = Fits {
315        id: 0,
316        payload: 0,
317        key: 0,
318        attempt: 0,
319    };
320    for column in row.columns() {
321        let alias = column.name();
322        let failed = |source| Error::ColumnDecode {
323            index: format!("{alias:?}"),
324            source,
325        };
326        let value = || DB::value(row, column.ordinal());
327        match alias {
328            "id" => {
329                let (read, held) = DB::id(value()?).map_err(failed)?;
330                (id, fits.id) = (Some(read), held);
331            }
332            "payload" => {
333                let (read, held) = DB::bytes(value()?).map_err(failed)?;
334                (payload, fits.payload) = (Some(read), held);
335            }
336            "partition_key" => {
337                let value = value()?;
338                if value.is_null() {
339                    fits.key = u8::MAX;
340                } else {
341                    let (read, held) = DB::bytes(value).map_err(failed)?;
342                    (key, fits.key) = (Some(read), held);
343                }
344            }
345            "headers" => {
346                let value = value()?;
347                if !value.is_null() {
348                    headers = D::headers(value).map_err(failed)?;
349                }
350            }
351            "attempt" => {
352                let (read, held) = DB::attempt(value()?).map_err(failed)?;
353                (attempt, fits.attempt) = (Some(read), held);
354            }
355            _ => {}
356        }
357    }
358    Ok(NamedRow {
359        id: id.ok_or_else(|| Error::ColumnNotFound("id".to_owned()))?,
360        payload: payload.ok_or_else(|| Error::ColumnNotFound("payload".to_owned()))?,
361        headers,
362        key,
363        attempt,
364        fits,
365        dialect: PhantomData,
366    })
367}
368
369impl<'r, R, D> FromRow<'r, R> for NamedRow<D>
370where
371    R: Row,
372    R::Database: RoleColumns + Database<Row = R>,
373    D: ByName<R::Database>,
374{
375    fn from_row(row: &'r R) -> Result<Self, Error> {
376        read::<R::Database, D>(row)
377    }
378}
379
380impl<D: 'static> QueueRow for NamedRow<D> {
381    type Id = NamedId;
382    type Lane = PayloadLane;
383}
384
385impl<D: 'static> PayloadRow for NamedRow<D> {
386    type Column = NamedBytes;
387
388    fn payload(&self) -> &[u8] {
389        self.payload.as_bytes()
390    }
391}
392
393mod events;
394
395#[cfg(test)]
396mod tests {
397    //! How a by-name row holds its columns to the struct's types.
398
399    use std::marker::PhantomData;
400
401    use ruststream::HeaderMap;
402    use ruststream::codec::CodecError;
403    use sqlx::Error;
404
405    use super::{Fits, NamedBytes, NamedId, NamedRow, hold_to};
406    use crate::inbox::engine::{Claimed, undecodable};
407    use crate::inbox::named::kinds::{BytesKind, ClockKind, IdKind, IntKind, Kinds};
408
409    /// A struct of an `i64` id, a byte payload, a text key and an `i16` attempt.
410    pub(super) const KINDS: Kinds = Kinds {
411        clock: ClockKind::System,
412        id: IdKind::I64,
413        payload: BytesKind::Bytes,
414        key: Some(BytesKind::Text),
415        attempt: Some(IntKind::I16),
416        retry_after: None,
417        processed_at: None,
418        locked_until: None,
419    };
420
421    /// A row whose columns hold the types `KINDS` reads, and only those, for the dialect `D`.
422    pub(super) fn row<D>(fits: Fits) -> NamedRow<D> {
423        NamedRow {
424            id: NamedId::I64(7),
425            payload: NamedBytes::Bytes(b"{}".to_vec()),
426            headers: HeaderMap::new(),
427            key: Some(NamedBytes::Text("acme".to_owned())),
428            attempt: Some(2),
429            fits,
430            dialect: PhantomData,
431        }
432    }
433
434    pub(super) const FITTING: Fits = Fits {
435        id: IdKind::I64.bit(),
436        payload: BytesKind::Bytes.bit(),
437        key: BytesKind::Text.bit(),
438        attempt: IntKind::I16.bit(),
439    };
440
441    /// The driver's error a delivery of an undecodable row reports.
442    fn driver_error(error: &CodecError) -> Option<&Error> {
443        match error {
444            CodecError::Decode(source) => source.downcast_ref::<Error>(),
445            _ => None,
446        }
447    }
448
449    /// The column a held row names as not holding its struct's type, and the attempt it keeps.
450    fn unfit_column(claimed: &Claimed<NamedRow<()>>) -> Option<(String, Option<u64>)> {
451        match claimed {
452            Claimed::Undecodable { id, attempt, error } => {
453                assert_eq!(id, &NamedId::I64(7), "the row keeps its id for the policy");
454                match driver_error(error) {
455                    Some(Error::ColumnDecode { index, .. }) => Some((index.clone(), *attempt)),
456                    other => panic!("not a column's error: {other:?}"),
457                }
458            }
459            Claimed::Row(_) | Claimed::Missing(_) => None,
460        }
461    }
462
463    #[test]
464    fn logs_name_a_row_by_its_id_as_the_table_holds_it() {
465        let ids = [
466            NamedId::I16(1),
467            NamedId::I32(2),
468            NamedId::I64(3),
469            NamedId::Text("job-4".to_owned()),
470            NamedId::Bytes(vec![5]),
471        ];
472        let logged: Vec<String> = ids.iter().map(|id| format!("{id:?}")).collect();
473        assert_eq!(logged, ["1", "2", "3", "\"job-4\"", "[5]"]);
474    }
475
476    #[test]
477    fn an_id_is_copied_into_the_storage_of_the_one_it_replaces() {
478        let mut kept = NamedId::Text("job-0001".to_owned());
479        let storage = match &kept {
480            NamedId::Text(text) => text.as_ptr(),
481            other => panic!("not a text id: {other:?}"),
482        };
483        kept.clone_from(&NamedId::Text("job-0002".to_owned()));
484        assert!(
485            matches!(&kept, NamedId::Text(text) if text == "job-0002" && text.as_ptr() == storage),
486            "{kept:?}"
487        );
488        let mut bytes = NamedId::Bytes(vec![1, 2]);
489        bytes.clone_from(&NamedId::Bytes(vec![3, 4]));
490        assert_eq!(bytes, NamedId::Bytes(vec![3, 4]));
491        // An id of another type takes the new one's type.
492        kept.clone_from(&NamedId::I64(7));
493        assert_eq!(kept, NamedId::I64(7));
494        assert_eq!(kept.clone(), NamedId::I64(7));
495        let ids = [
496            NamedId::I16(1),
497            NamedId::I32(2),
498            NamedId::Text("job-3".to_owned()),
499            NamedId::Bytes(vec![4]),
500        ];
501        for id in ids {
502            assert_eq!(id.clone(), id);
503        }
504    }
505
506    #[test]
507    fn a_row_whose_columns_hold_its_structs_types_stays_a_row() -> Result<(), Error> {
508        let mut claimed = Claimed::Row(row::<()>(FITTING));
509        hold_to(KINDS, &mut claimed)?;
510        assert!(matches!(claimed, Claimed::Row(_)));
511        // A null key fits whatever its column's type.
512        let mut keyless = Claimed::Row(row::<()>(Fits {
513            key: u8::MAX,
514            ..FITTING
515        }));
516        hold_to(KINDS, &mut keyless)?;
517        assert!(matches!(keyless, Claimed::Row(_)));
518        Ok(())
519    }
520
521    #[test]
522    fn a_column_that_does_not_hold_its_structs_type_sends_the_row_to_the_policy()
523    -> Result<(), Error> {
524        let text_payload = Fits {
525            payload: BytesKind::Text.bit(),
526            ..FITTING
527        };
528        let byte_key = Fits {
529            key: BytesKind::Bytes.bit(),
530            ..FITTING
531        };
532        let wide_attempt = Fits {
533            attempt: IntKind::I64.bit(),
534            ..FITTING
535        };
536        let text_payload_wide_attempt = Fits {
537            attempt: IntKind::I64.bit(),
538            ..text_payload
539        };
540        // The attempt goes along, so the cap spends the row, unless its own column is one the
541        // struct does not read: the struct's `FromRow` reads no attempt from it either.
542        for (fits, column, attempt) in [
543            (text_payload, "\"payload\"", Some(2)),
544            (byte_key, "\"partition_key\"", Some(2)),
545            (wide_attempt, "\"attempt\"", None),
546            (text_payload_wide_attempt, "\"payload\"", None),
547        ] {
548            let mut claimed = Claimed::Row(row::<()>(fits));
549            hold_to(KINDS, &mut claimed)?;
550            assert_eq!(
551                unfit_column(&claimed),
552                Some((column.to_owned(), attempt)),
553                "{fits:?}"
554            );
555        }
556        Ok(())
557    }
558
559    #[test]
560    fn an_id_that_does_not_hold_its_structs_type_fails_the_claim() {
561        let mut claimed = Claimed::Row(row::<()>(Fits {
562            id: IdKind::I32.bit(),
563            ..FITTING
564        }));
565        let failed = hold_to(KINDS, &mut claimed);
566        assert!(
567            matches!(&failed, Err(Error::ColumnDecode { index, .. }) if index == "\"id\""),
568            "{failed:?}"
569        );
570    }
571
572    #[test]
573    fn a_row_already_undecodable_is_left_to_the_policy() -> Result<(), Error> {
574        let mut claimed = Claimed::<NamedRow<()>>::Undecodable {
575            id: NamedId::I64(7),
576            attempt: Some(3),
577            error: Box::new(undecodable(Error::ColumnNotFound("payload".to_owned()))),
578        };
579        hold_to(KINDS, &mut claimed)?;
580        assert!(matches!(
581            &claimed,
582            Claimed::Undecodable { attempt: Some(3), error, .. }
583                if matches!(driver_error(error), Some(Error::ColumnNotFound(_)))
584        ));
585        Ok(())
586    }
587}
ruststream_sqlx/inbox/named/row.rs

ruststream_sqlx/inbox/named/
row.rs
