ruststream_sqlx/inbox/events.rs
1//! One trait per event of the queue: what a service implements when its table names the event as
2//! its own, in `#[inbox(custom(..))]` or with `InboxSpec::own`, and `Publish`, which has no
3//! default.
4
5use std::future::Future;
6use std::time::Duration;
7
8use ruststream::OutgoingMessage;
9use sqlx::{Database, Error};
10
11use super::InboxRow;
12use super::time::LeaseRow;
13
14/// Claims up to `limit` rows of the queue `queue` and returns their ids, inside the claim's
15/// transaction.
16///
17/// The dialect builds it from the table's description; a table names `claim` as its own to take it
18/// over (`custom(claim)`, or [`own::Claim`](crate::spec::own::Claim) by hand), for a database
19/// without a built-in dialect or a claim of its own. The crate then reads the
20/// rows with the service's [`Fetch`], or with its own fetch by a list of ids, which Postgres
21/// alone runs: on MySQL and SQLite a claim of the service's own comes with a `Fetch` of its own,
22/// and a subscription without one stops at startup. In the lease form the crate also leases each
23/// id the claim returns, in the same transaction, and passes over an id whose row another lease
24/// holds.
25///
26/// # Examples
27///
28/// ```
29/// # #[cfg(feature = "postgres")]
30/// # mod demo {
31/// use ruststream_sqlx::{Claim, Inbox};
32/// use sqlx::{PgConnection, Postgres};
33///
34/// #[derive(Inbox, sqlx::FromRow)]
35/// #[inbox(table = "jobs", custom(claim))]
36/// pub struct Job {
37/// #[field(id)]
38/// id: i64,
39/// #[field(payload)]
40/// payload: Vec<u8>,
41/// }
42///
43/// impl Claim<Postgres> for Job {
44/// async fn claim(
45/// conn: &mut PgConnection,
46/// _queue: &str,
47/// limit: i64,
48/// ) -> Result<Vec<i64>, sqlx::Error> {
49/// // The oldest rows first, skipping the ones another worker holds.
50/// sqlx::query_scalar("SELECT id FROM jobs ORDER BY id LIMIT $1 FOR UPDATE SKIP LOCKED")
51/// .bind(limit)
52/// .fetch_all(conn)
53/// .await
54/// }
55/// }
56/// # }
57/// # fn main() {}
58/// ```
59#[diagnostic::on_unimplemented(
60 message = "`{Self}` runs its own `claim` (`custom(claim)` in `#[inbox(..)]`, or `own::Claim` in its `InboxSpec`) and does not implement `Claim<{DB}>`",
61 label = "the service's own claim is missing",
62 note = "implement `Claim<{DB}>` for `{Self}`, or drop `claim` from `custom(..)` (by hand: `.own::<own::Claim>()` and `own::Claim` in `type Table`)"
63)]
64pub trait Claim<DB: Database>: InboxRow {
65 /// Claims up to `limit` rows and returns their ids.
66 ///
67 /// # Errors
68 ///
69 /// The database's error; the subscription waits one second and claims again.
70 fn claim(
71 conn: &mut DB::Connection,
72 queue: &str,
73 limit: i64,
74 ) -> impl Future<Output = Result<Vec<Self::Id>, Error>> + Send;
75}
76
77/// Reads the rows of claimed ids, on the claim's connection.
78///
79/// The dialect builds it, for a flat table and for a message assembled from header fields; a table
80/// names `fetch` as its own (`custom(fetch)`, or [`own::Fetch`](crate::spec::own::Fetch) by hand)
81/// to assemble messages itself, from other tables
82/// ([a fetch over a join](https://powersemmi.github.io/ruststream-sqlx/latest/components/headers/fetch/)). Rows are matched to the claimed ids by their `id` field; a
83/// claimed id with no row settles by the decode-failure policy before its handler runs, and the
84/// log names the id. It runs inside the claim's transaction, or right after a lease claim that
85/// leased the rows and committed in one statement, as on Postgres and SQLite. In the advisory lock
86/// form it runs once per row, with that row's id alone, after the take that counted its attempt.
87/// Either way a delivery reports the attempt its row held before the claim. It runs only after a
88/// claim that took ids, so `ids` is never empty.
89///
90/// # Examples
91///
92/// ```
93/// # #[cfg(feature = "postgres")]
94/// # mod demo {
95/// use ruststream_sqlx::{Fetch, Inbox};
96/// use sqlx::{PgConnection, Postgres};
97///
98/// /// A job whose message is the body of the order it refers to.
99/// #[derive(Inbox, sqlx::FromRow)]
100/// #[inbox(table = "jobs", custom(fetch))]
101/// pub struct Job {
102/// #[field(id)]
103/// id: i64,
104/// #[field(payload)]
105/// payload: Vec<u8>,
106/// }
107///
108/// impl Fetch<Postgres> for Job {
109/// async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
110/// sqlx::query_as(
111/// "SELECT j.id, o.body AS payload FROM jobs j JOIN orders o ON o.id = j.order_id \
112/// WHERE j.id = ANY($1)",
113/// )
114/// .bind(ids)
115/// .fetch_all(conn)
116/// .await
117/// }
118/// }
119/// # }
120/// # fn main() {}
121/// ```
122///
123/// In the headers layout two messages may flatten one headers struct: the default fetch reads one
124/// from the queue table, and the service's own fetch reads the other over a join. A claimed id the
125/// join finds no row for settles by the decode policy before the handler runs:
126///
127/// ```no_run
128/// # #[cfg(all(feature = "inbox", feature = "postgres"))]
129/// # mod demo {
130/// use ruststream_sqlx::Fetch;
131/// use ruststream_sqlx::prelude::*;
132/// use serde::Deserialize;
133/// use sqlx::{PgConnection, PgPool, Postgres};
134///
135/// // order_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, attempt SMALLINT DEFAULT 1, tenant TEXT,
136/// // trace TEXT NULL, order_id BIGINT, note TEXT NULL
137/// // orders: id BIGINT PRIMARY KEY, customer TEXT, total BIGINT
138/// /// The queue table: the columns that run the queue, and the headers `tenant`, `trace` and
139/// /// `order_id`.
140/// #[derive(Debug, Clone, InboxHeaders, sqlx::FromRow)]
141/// #[inbox(table = "order_jobs")]
142/// pub struct OrderHeaders {
143/// #[field(id, generated)]
144/// job_id: i64,
145/// #[field(group)]
146/// name: String,
147/// #[field(attempt, generated)]
148/// attempt: i16,
149/// tenant: String,
150/// trace: Option<String>,
151/// order_id: i64,
152/// }
153///
154/// /// A job and its note, read from the queue table by the default fetch.
155/// #[derive(Debug, Clone, Inbox, sqlx::FromRow)]
156/// pub struct OrderJob {
157/// #[field(headers)]
158/// #[sqlx(flatten)]
159/// headers: OrderHeaders,
160/// note: Option<String>,
161/// }
162///
163/// /// A job and the order it is for, read by the service's own fetch over a join.
164/// #[derive(Debug, Clone, Inbox, sqlx::FromRow)]
165/// #[inbox(custom(fetch))]
166/// pub struct OrderMail {
167/// #[field(headers)]
168/// #[sqlx(flatten)]
169/// headers: OrderHeaders,
170/// customer: String,
171/// total: i64,
172/// }
173///
174/// impl Fetch<Postgres> for OrderMail {
175/// async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
176/// sqlx::query_as(
177/// "SELECT j.job_id, j.name, j.attempt, j.tenant, j.trace, j.order_id, \
178/// o.customer, o.total \
179/// FROM order_jobs j JOIN orders o ON o.id = j.order_id \
180/// WHERE j.job_id = ANY($1)",
181/// )
182/// .bind(ids)
183/// .fetch_all(conn)
184/// .await
185/// }
186/// }
187///
188/// #[derive(Deserialize)]
189/// pub struct Tenant {
190/// tenant: String,
191/// trace: Option<String>,
192/// }
193///
194/// #[subscriber(InboxQueue::<OrderJob>::new("packing"))]
195/// async fn pack(job: &OrderJob) -> HandlerOutcome {
196/// tracing::info!(attempt = job.headers.attempt, note = ?job.note, "packing");
197/// HandlerOutcome::ack()
198/// }
199///
200/// // The header map is built from `OrderHeaders` on its first read, here by `Headers<Tenant>`.
201/// #[subscriber(InboxQueue::<OrderMail>::new("receipts"))]
202/// async fn send_receipt(mail: &OrderMail, Headers(tenant): Headers<Tenant>) -> HandlerOutcome {
203/// tracing::info!(
204/// tenant = %tenant.tenant,
205/// trace = ?tenant.trace,
206/// customer = %mail.customer,
207/// total = mail.total,
208/// "sending the receipt"
209/// );
210/// HandlerOutcome::ack()
211/// }
212///
213/// pub fn app(pool: PgPool) -> RustStream {
214/// RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
215/// b.include(pack);
216/// // A job whose order is gone fails to decode: the default policy drops it.
217/// b.include(send_receipt);
218/// })
219/// }
220/// # }
221/// # fn main() {}
222/// ```
223#[diagnostic::on_unimplemented(
224 message = "`{Self}` runs its own `fetch` (`custom(fetch)` in `#[inbox(..)]`, or `own::Fetch` in its `InboxSpec`) and does not implement `Fetch<{DB}>`",
225 label = "the service's own fetch is missing",
226 note = "implement `Fetch<{DB}>` for `{Self}`, or drop `fetch` from `custom(..)` (by hand: `.own::<own::Fetch>()` and `own::Fetch` in `type Table`)"
227)]
228pub trait Fetch<DB: Database>: InboxRow {
229 /// The rows of `ids`, in any order.
230 ///
231 /// # Errors
232 ///
233 /// The database's error; the claim fails and its rows return to the queue.
234 fn fetch(
235 conn: &mut DB::Connection,
236 ids: &[Self::Id],
237 ) -> impl Future<Output = Result<Vec<Self>, Error>> + Send;
238}
239
240macro_rules! settle_event {
241 (
242 $(#[$doc:meta])*
243 $trait:ident, $method:ident, $event:literal,
244 gate = $gate:literal, $(uses = $uses:literal,)? $(fields = $fields:literal,)? body = $body:literal,
245 message = $message:literal, note = $note:literal
246 $(, $extra:ident: $extra_ty:ty = $extra_value:literal)?
247 ) => {
248 $(#[$doc])*
249 #[doc = ""]
250 #[doc = "# Examples"]
251 #[doc = ""]
252 #[doc = "```"]
253 #[doc = concat!("# #[cfg(", $gate, ")]")]
254 #[doc = "# mod demo {"]
255 #[doc = "use std::time::Duration;"]
256 #[doc = ""]
257 #[doc = concat!("use ruststream_sqlx::{", stringify!($trait), ", Inbox};")]
258 #[doc = "use sqlx::{PgConnection, Postgres};"]
259 $(#[doc = $uses])?
260 #[doc = ""]
261 #[doc = "#[derive(Inbox, sqlx::FromRow)]"]
262 #[doc = concat!("#[inbox(table = \"jobs\", custom(", $event, "))]")]
263 #[doc = "pub struct Job {"]
264 #[doc = " #[field(id)]"]
265 #[doc = " id: i64,"]
266 $(#[doc = $fields])?
267 #[doc = " #[field(payload)]"]
268 #[doc = " payload: Vec<u8>,"]
269 #[doc = "}"]
270 #[doc = ""]
271 #[doc = concat!("impl ", stringify!($trait), "<Postgres> for Job {")]
272 #[doc = concat!(" async fn ", stringify!($method), "(")]
273 #[doc = " conn: &mut PgConnection,"]
274 #[doc = " id: &i64,"]
275 $(#[doc = concat!(" ", stringify!($extra), ": ", stringify!($extra_ty), ",")])?
276 #[doc = " ) -> Result<(), sqlx::Error> {"]
277 #[doc = $body]
278 #[doc = " Ok(())"]
279 #[doc = " }"]
280 #[doc = "}"]
281 #[doc = "# }"]
282 #[doc = "# fn main() {}"]
283 #[doc = "```"]
284 #[diagnostic::on_unimplemented(
285 message = $message,
286 label = "the service's own event is missing",
287 note = $note
288 )]
289 pub trait $trait<DB: Database>: InboxRow {
290 #[doc = concat!("Runs the `", $event, "` event for the row `id` inside a transaction the crate commits: the claim's, or in the lease form one that first confirms the row still holds the delivery's lease.")]
291 #[doc = ""]
292 #[doc = "# Errors"]
293 #[doc = ""]
294 #[doc = "The database's error; the transaction rolls back, and the row returns to the queue at once, or in the lease form once its lease runs out."]
295 fn $method(
296 conn: &mut DB::Connection,
297 id: &Self::Id,
298 $($extra: $extra_ty,)?
299 ) -> impl Future<Output = Result<(), Error>> + Send;
300 }
301 };
302}
303
304settle_event!(
305 /// Acknowledges a row: by default it deletes the row, or sets `processed_at`.
306 ///
307 /// The service's own acknowledgement takes the row out of what the claim selects: it deletes
308 /// the row, moves it, or marks it in a column the claim passes over. A row left claimable is
309 /// delivered again.
310 Ack, ack, "ack",
311 gate = "feature = \"postgres\"",
312 body = " // The finished job moves to an archive in one statement.
313 sqlx::query(
314 \"WITH done AS (DELETE FROM jobs WHERE id = $1 RETURNING *) \\
315 INSERT INTO jobs_done SELECT * FROM done\",
316 )
317 .bind(id)
318 .execute(conn)
319 .await?;",
320 message = "`{Self}` runs its own `ack` (`custom(ack)` in `#[inbox(..)]`, or `own::Ack` in its `InboxSpec`) and does not implement `Ack<{DB}>`",
321 note = "implement `Ack<{DB}>` for `{Self}`, or drop `ack` from `custom(..)` (by hand: `.own::<own::Ack>()` and `own::Ack` in `type Table`)"
322);
323
324settle_event!(
325 /// Releases a row for another attempt at once: by default it counts the attempt, or leaves the
326 /// row for the rollback to release; in the lease form it clears the lease.
327 ///
328 /// In the lease form the row comes back once `locked_until` no longer holds it: the service's
329 /// own retry clears the column to release the row at once, and otherwise the row waits for its
330 /// lease to run out.
331 Retry, retry, "retry",
332 gate = "feature = \"postgres\"",
333 body = " sqlx::query(\"UPDATE jobs SET tries = tries + 1 WHERE id = $1\")
334 .bind(id)
335 .execute(conn)
336 .await?;",
337 message = "`{Self}` runs its own `retry` (`custom(retry)` in `#[inbox(..)]`, or `own::Retry` in its `InboxSpec`) and does not implement `Retry<{DB}>`",
338 note = "implement `Retry<{DB}>` for `{Self}`, or drop `retry` from `custom(..)` (by hand: `.own::<own::Retry>()` and `own::Retry` in `type Table`)"
339);
340
341settle_event!(
342 /// Releases a row for another attempt after `delay`: by default it sets `retry_after`.
343 ///
344 /// A row with this event, or a `retry_after` field, is redelivered by the database's own
345 /// clock. The service's own event delays a row only where the claim passes over it until
346 /// then: the default claim reads the `retry_after` field.
347 RetryAfter, retry_after, "retry_after",
348 gate = "all(feature = \"postgres\", feature = \"chrono\")",
349 uses = "use chrono::{DateTime, Utc};",
350 fields = " #[field(retry_after)]
351 retry_after: DateTime<Utc>,",
352 body = " // The delay counts from the database's clock, whatever the host's says.
353 sqlx::query(
354 \"UPDATE jobs SET retry_after = now() + make_interval(secs => $2) WHERE id = $1\",
355 )
356 .bind(id)
357 .bind(delay.as_secs_f64())
358 .execute(conn)
359 .await?;",
360 message = "`{Self}` runs its own `retry_after` (`custom(retry_after)` in `#[inbox(..)]`, or `own::RetryAfter` in its `InboxSpec`) and does not implement `RetryAfter<{DB}>`",
361 note = "implement `RetryAfter<{DB}>` for `{Self}`, or drop `retry_after` from `custom(..)` (by hand: `.own::<own::RetryAfter>()` and `own::RetryAfter` in `type Table`)",
362 delay: Duration = "Duration::from_secs(30)"
363);
364
365settle_event!(
366 /// Drops a row: by default it deletes the row, or sets `processed_at`.
367 ///
368 /// The service's own drop takes the row out of what the claim selects, as an acknowledgement
369 /// does.
370 Discard, discard, "discard",
371 gate = "feature = \"postgres\"",
372 body = " sqlx::query(\"DELETE FROM jobs WHERE id = $1\")
373 .bind(id)
374 .execute(conn)
375 .await?;",
376 message = "`{Self}` runs its own `discard` (`custom(discard)` in `#[inbox(..)]`, or `own::Discard` in its `InboxSpec`) and does not implement `Discard<{DB}>`",
377 note = "implement `Discard<{DB}>` for `{Self}`, or drop `discard` from `custom(..)` (by hand: `.own::<own::Discard>()` and `own::Discard` in `type Table`)"
378);
379
380settle_event!(
381 /// Moves a row whose attempts are spent to `destination`: by default it moves the row to that
382 /// group (on a table with a `group` column) or into that table.
383 ///
384 /// The service's own move takes the row out of what the claim selects; a row copied and left
385 /// behind is moved again at every delivery.
386 DeadLetter, dead_letter, "dead_letter",
387 gate = "feature = \"postgres\"",
388 body = " // One dead-letter table serves the whole service, whatever the destination names.
389 sqlx::query(
390 \"WITH dead AS (DELETE FROM jobs WHERE id = $1 RETURNING *) \\
391 INSERT INTO dead_jobs SELECT * FROM dead\",
392 )
393 .bind(id)
394 .execute(conn)
395 .await?;
396 tracing::warn!(id, destination, \"a job's attempts are spent\");",
397 message = "`{Self}` runs its own `dead_letter` (`custom(dead_letter)` in `#[inbox(..)]`, or `own::DeadLetter` in its `InboxSpec`) and does not implement `DeadLetter<{DB}>`",
398 note = "implement `DeadLetter<{DB}>` for `{Self}`, or drop `dead_letter` from `custom(..)` (by hand: `.own::<own::DeadLetter>()` and `own::DeadLetter` in `type Table`)",
399 destination: &str = "\"jobs.dead\""
400);
401
402/// Extends the lease on a row: writes `until` into `locked_until` while the row still holds
403/// `held`, and says whether it did.
404///
405/// The dialect builds it for every table in the lease form; a table names `extend` as its own
406/// (`custom(extend)`, or [`own::Extend`](crate::spec::own::Extend) by hand) to take it over, for a
407/// database without a built-in dialect. A
408/// subscription runs it each half lease for every delivery in work, on one connection. The crate
409/// also runs it with `until` equal to `held` to confirm a delivery's lease, inside the transaction
410/// where an event of the service's own then runs.
411///
412/// # Examples
413///
414/// ```
415/// # #[cfg(all(feature = "postgres", feature = "chrono"))]
416/// # mod demo {
417/// use chrono::{DateTime, Utc};
418/// use ruststream_sqlx::{Extend, Inbox};
419/// use sqlx::{PgConnection, Postgres};
420///
421/// #[derive(Inbox, sqlx::FromRow)]
422/// #[inbox(table = "jobs", custom(extend))]
423/// pub struct Job {
424/// #[field(id)]
425/// id: i64,
426/// #[field(locked_until)]
427/// locked_until: Option<DateTime<Utc>>,
428/// #[field(payload)]
429/// payload: Vec<u8>,
430/// }
431///
432/// impl Extend<Postgres> for Job {
433/// async fn extend(
434/// conn: &mut PgConnection,
435/// id: &i64,
436/// held: &DateTime<Utc>,
437/// until: &DateTime<Utc>,
438/// ) -> Result<bool, sqlx::Error> {
439/// // The row keeps its lease only while it holds the expiry the delivery knows.
440/// let extended =
441/// sqlx::query("UPDATE jobs SET locked_until = $1 WHERE id = $2 AND locked_until = $3")
442/// .bind(until)
443/// .bind(id)
444/// .bind(held)
445/// .execute(conn)
446/// .await?;
447/// Ok(extended.rows_affected() == 1)
448/// }
449/// }
450/// # }
451/// # fn main() {}
452/// ```
453#[diagnostic::on_unimplemented(
454 message = "`{Self}` runs its own `extend` (`custom(extend)` in `#[inbox(..)]`, or `own::Extend` in its `InboxSpec`) and does not implement `Extend<{DB}>`",
455 label = "the service's own extension is missing",
456 note = "implement `Extend<{DB}>` for `{Self}`, or drop `extend` from `custom(..)` (by hand: `.own::<own::Extend>()` and `own::Extend` in `type Table`)"
457)]
458pub trait Extend<DB: Database>: LeaseRow {
459 /// Writes `until` into the lease of the row `id` while it still holds `held`; `true` when it
460 /// did, `false` when the row no longer holds `held`.
461 ///
462 /// # Errors
463 ///
464 /// The database's error.
465 fn extend(
466 conn: &mut DB::Connection,
467 id: &Self::Id,
468 held: &Self::Lease,
469 until: &Self::Lease,
470 ) -> impl Future<Output = Result<bool, Error>> + Send;
471}
472
473/// Takes the advisory lock on a row's key for the session that will hold its delivery, without
474/// waiting: the `lock` event of the advisory lock form.
475///
476/// The dialect takes the lock itself on Postgres and MySQL, and the process keeps the keys in work
477/// on SQLite. A table names `lock` and `unlock` as its own together (`custom(lock, unlock)`, or
478/// [`own::Lock`](crate::spec::own::Lock) and [`own::Unlock`](crate::spec::own::Unlock) by hand) to
479/// run the service's own SQL for them instead, as a database without a built-in dialect does, MSSQL with
480/// `sp_getapplock` and `sp_releaseapplock`; the dialect still selects the candidates and takes
481/// each row. A claim calls it once per candidate with the key the template renders, on the
482/// connection that holds the delivery; [`Unlock`] releases the lock when the delivery settles,
483/// and where the delivery is dropped unsettled the connection closes after that unlock. With a
484/// lock of the service's own the process keeps no registry of keys, on SQLite too.
485///
486/// The dialect's candidate select cannot see a lock of the service's own, so a claim reads as many
487/// candidates past its limit as the subscription has deliveries in work: a key its deliveries hold
488/// does not hold back the rows behind it. A key the service's lock holds elsewhere, in another
489/// process or system, is passed over within that margin only: while such keys head the claim
490/// order, the claim may end empty and wait for its next poll.
491///
492/// # Examples
493///
494/// ```no_run
495/// # #[cfg(feature = "postgres")]
496/// # mod demo {
497/// use ruststream_sqlx::{Inbox, Lock, Unlock};
498/// use sqlx::{PgConnection, Postgres};
499///
500/// #[derive(Inbox, sqlx::FromRow)]
501/// #[inbox(table = "jobs", advisory_lock = "jobs-{id}", custom(lock, unlock))]
502/// pub struct Job {
503/// #[field(id)]
504/// id: i64,
505/// #[field(payload)]
506/// payload: Vec<u8>,
507/// }
508///
509/// // The pair MSSQL writes with `sp_getapplock` and `sp_releaseapplock`, written here with
510/// // Postgres's session locks on a 32-bit hash of the key.
511/// impl Lock<Postgres> for Job {
512/// async fn lock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
513/// sqlx::query_scalar("SELECT pg_try_advisory_lock(hashtext($1))")
514/// .bind(key)
515/// .fetch_one(conn)
516/// .await
517/// }
518/// }
519///
520/// impl Unlock<Postgres> for Job {
521/// async fn unlock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
522/// sqlx::query_scalar("SELECT pg_advisory_unlock(hashtext($1))")
523/// .bind(key)
524/// .fetch_one(conn)
525/// .await
526/// }
527/// }
528/// # }
529/// # fn main() {}
530/// ```
531///
532/// A queue that takes the lock another system already holds on an account, so a payout never
533/// runs beside that system's work on it:
534///
535/// ```no_run
536/// # #[cfg(all(feature = "inbox", feature = "postgres"))]
537/// # mod demo {
538/// use ruststream_sqlx::prelude::*;
539/// use ruststream_sqlx::{Lock, Unlock};
540/// use serde::Deserialize;
541/// use sqlx::{PgConnection, PgPool, Postgres};
542///
543/// // payouts: id BIGSERIAL PRIMARY KEY, account_id BIGINT NOT NULL, payload BYTEA NOT NULL
544/// #[derive(Inbox, sqlx::FromRow)]
545/// #[inbox(table = "payouts", advisory_lock = "{account_id}", custom(lock, unlock))]
546/// pub struct Payout {
547/// #[field(id, generated)]
548/// id: i64,
549/// account_id: i64,
550/// #[field(payload)]
551/// payload: Vec<u8>,
552/// }
553///
554/// // The billing system holds `pg_advisory_lock(account_id)` while it changes an account. The queue
555/// // takes the same lock, so a payout never runs beside a billing run of its account.
556/// impl Lock<Postgres> for Payout {
557/// async fn lock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
558/// sqlx::query_scalar("SELECT pg_try_advisory_lock($1::bigint)")
559/// .bind(key)
560/// .fetch_one(conn)
561/// .await
562/// }
563/// }
564///
565/// impl Unlock<Postgres> for Payout {
566/// async fn unlock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
567/// sqlx::query_scalar("SELECT pg_advisory_unlock($1::bigint)")
568/// .bind(key)
569/// .fetch_one(conn)
570/// .await
571/// }
572/// }
573///
574/// #[derive(Deserialize)]
575/// pub struct Transfer {
576/// cents: i64,
577/// }
578///
579/// # async fn transfer_to_bank(transfer: &Transfer) { let _ = transfer.cents; }
580/// #[subscriber(InboxQueue::<Payout>::new("payouts"))]
581/// async fn pay(transfer: &Transfer) -> HandlerOutcome {
582/// transfer_to_bank(transfer).await;
583/// HandlerOutcome::ack()
584/// }
585///
586/// pub fn app(pool: PgPool) -> RustStream {
587/// RustStream::new(AppInfo::new("payouts", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
588/// b.include(pay);
589/// })
590/// }
591/// # }
592/// # fn main() {}
593/// ```
594#[diagnostic::on_unimplemented(
595 message = "`{Self}` runs its own `lock` (`custom(lock)` in `#[inbox(..)]`, or `own::Lock` in its `InboxSpec`) and does not implement `Lock<{DB}>`",
596 label = "the service's own lock is missing",
597 note = "implement `Lock<{DB}>` for `{Self}`, or drop `lock` from `custom(..)` (by hand: `.own::<own::Lock>()` and `own::Lock` in `type Table`)"
598)]
599pub trait Lock<DB: Database>: InboxRow {
600 /// Tries the advisory lock on `key` for the session of `conn`, without waiting: `true` when it
601 /// took the lock. The session holds it until `Unlock` releases it or the session ends.
602 ///
603 /// # Errors
604 ///
605 /// The database's error; the claim fails, its connection closes, and the next claim waits one
606 /// second.
607 fn lock(
608 conn: &mut DB::Connection,
609 key: &str,
610 ) -> impl Future<Output = Result<bool, Error>> + Send;
611}
612
613/// Releases the advisory lock on a delivery's key: the `unlock` event of the advisory lock form,
614/// which a table names as its own beside [`Lock`].
615///
616/// A settlement calls it after its statement, on the connection that holds the delivery, and the
617/// connection goes back to the pool once it answered `true`. An unlock that answers `false` or
618/// fails closes the connection instead, which ends the session and its locks. A delivery dropped
619/// unsettled runs it before its connection closes, and `shutdown` runs it for every delivery in
620/// work.
621///
622/// # Examples
623///
624/// ```no_run
625/// # #[cfg(feature = "postgres")]
626/// # mod demo {
627/// use ruststream_sqlx::{Inbox, Lock, Unlock};
628/// use sqlx::{PgConnection, Postgres};
629///
630/// #[derive(Inbox, sqlx::FromRow)]
631/// #[inbox(table = "jobs", advisory_lock = "jobs-{id}", custom(lock, unlock))]
632/// pub struct Job {
633/// #[field(id)]
634/// id: i64,
635/// #[field(payload)]
636/// payload: Vec<u8>,
637/// }
638///
639/// // MSSQL releases with `sp_releaseapplock`; Postgres's session lock stands in for it here.
640/// impl Unlock<Postgres> for Job {
641/// async fn unlock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
642/// sqlx::query_scalar("SELECT pg_advisory_unlock(hashtext($1))")
643/// .bind(key)
644/// .fetch_one(conn)
645/// .await
646/// }
647/// }
648///
649/// impl Lock<Postgres> for Job {
650/// async fn lock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
651/// sqlx::query_scalar("SELECT pg_try_advisory_lock(hashtext($1))")
652/// .bind(key)
653/// .fetch_one(conn)
654/// .await
655/// }
656/// }
657/// # }
658/// # fn main() {}
659/// ```
660#[diagnostic::on_unimplemented(
661 message = "`{Self}` runs its own `unlock` (`custom(unlock)` in `#[inbox(..)]`, or `own::Unlock` in its `InboxSpec`) and does not implement `Unlock<{DB}>`",
662 label = "the service's own unlock is missing",
663 note = "implement `Unlock<{DB}>` for `{Self}`, or drop `unlock` from `custom(..)` (by hand: `.own::<own::Unlock>()` and `own::Unlock` in `type Table`)"
664)]
665pub trait Unlock<DB: Database>: InboxRow {
666 /// Releases the advisory lock on `key` the session of `conn` holds: `true` when the session
667 /// held it.
668 ///
669 /// # Errors
670 ///
671 /// The database's error; the connection closes, which ends the session and its locks.
672 fn unlock(
673 conn: &mut DB::Connection,
674 key: &str,
675 ) -> impl Future<Output = Result<bool, Error>> + Send;
676}
677
678/// Writes a published message into the table: the name, the bytes and the headers reach the
679/// service's SQL, which lays them out in its columns.
680///
681/// It has no default. With it, the struct's [`Repository`](crate::Repository) is a publisher and
682/// a route can lead names to it; without it, a mount that needs one does not compile, rather than
683/// publish into nothing. A message carrying a header the struct cannot hold byte for byte (any
684/// header, without a `#[field(headers)]` field; see [`HeaderColumn`](crate::HeaderColumn)) is
685/// refused before this runs.
686///
687/// # Examples
688///
689/// ```
690/// # #[cfg(feature = "postgres")]
691/// # mod demo {
692/// use ruststream::OutgoingMessage;
693/// use ruststream_sqlx::{Inbox, Publish};
694/// use sqlx::{PgConnection, Postgres};
695///
696/// #[derive(Inbox, sqlx::FromRow)]
697/// #[inbox(table = "email_jobs")]
698/// pub struct SendEmail {
699/// #[field(id, generated)]
700/// job_id: i64,
701/// #[field(group)]
702/// name: String,
703/// #[field(payload)]
704/// payload: Vec<u8>,
705/// }
706///
707/// impl Publish<Postgres> for SendEmail {
708/// async fn publish(
709/// conn: &mut PgConnection,
710/// message: &OutgoingMessage<'_>,
711/// ) -> Result<(), sqlx::Error> {
712/// sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
713/// .bind(message.name())
714/// .bind(message.payload())
715/// .execute(conn)
716/// .await?;
717/// Ok(())
718/// }
719/// }
720/// # }
721/// # fn main() {}
722/// ```
723#[diagnostic::on_unimplemented(
724 message = "`{Self}` does not implement `Publish<{DB}>`, so nothing writes a published message into its table",
725 label = "no `Publish` for this row",
726 note = "implement `Publish<{DB}>` for `{Self}`: its SQL lays the name, the bytes and the headers out in the table"
727)]
728pub trait Publish<DB: Database>: InboxRow {
729 /// Writes `message` into the table.
730 ///
731 /// # Errors
732 ///
733 /// The database's error, which the publish returns.
734 fn publish(
735 conn: &mut DB::Connection,
736 message: &OutgoingMessage<'_>,
737 ) -> impl Future<Output = Result<(), Error>> + Send;
738}
739
740/// Inserts the row: every field except the `generated` ones, the queue's mechanics included, so
741/// a delayed task is a row whose `retry_after` lies ahead.
742///
743/// The derive implements it for the connection of each built-in dialect, `PgConnection` (feature
744/// `postgres`), `MySqlConnection` (feature `mysql`) and `SqliteConnection` (feature `sqlite`), with
745/// the statements built at compile time. A headers struct gets it from
746/// [`InboxHeaders`](derive@crate::InboxHeaders): the queue table's columns.
747/// A struct with a `#[sqlx(flatten)]` field gets none: the derive cannot see the nested struct's
748/// columns, so the service writes that insert itself.
749/// A table described by hand implements it over the text
750/// [`dialect::insert`](crate::dialect::insert) renders from its description, binding the columns
751/// in the description's order, as below.
752///
753/// # Examples
754///
755/// ```
756/// # #[cfg(feature = "postgres")]
757/// # mod demo {
758/// use ruststream::OutgoingMessage;
759/// use ruststream_sqlx::{Inbox, Insert, Publish};
760/// use sqlx::{PgConnection, Postgres};
761///
762/// #[derive(Inbox, sqlx::FromRow)]
763/// #[inbox(table = "email_jobs")]
764/// pub struct SendEmail {
765/// #[field(id, generated)]
766/// job_id: i64,
767/// #[field(group)]
768/// name: String,
769/// #[field(payload)]
770/// payload: Vec<u8>,
771/// }
772///
773/// impl Publish<Postgres> for SendEmail {
774/// async fn publish(
775/// conn: &mut PgConnection,
776/// message: &OutgoingMessage<'_>,
777/// ) -> Result<(), sqlx::Error> {
778/// let job = SendEmail {
779/// job_id: 0,
780/// name: message.name().to_owned(),
781/// payload: message.payload().to_vec(),
782/// };
783/// // `job_id` is generated: the database fills it in.
784/// job.insert(conn).await
785/// }
786/// }
787/// # }
788/// # fn main() {}
789/// ```
790///
791/// A table described by hand in row mode writes its insert over the text
792/// [`dialect::insert`](crate::dialect::insert) renders, in a constant:
793///
794/// ```no_run
795/// # #[cfg(all(feature = "inbox", feature = "postgres"))]
796/// # mod demo {
797/// use ruststream::runtime::{Input, SoloCarried};
798/// use ruststream_sqlx::dialect::Column;
799/// use ruststream_sqlx::dialect::insert::{self, Sql};
800/// use ruststream_sqlx::prelude::*;
801/// use ruststream_sqlx::spec::Attempt;
802/// use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable};
803/// use serde::{Deserialize, Serialize};
804/// use sqlx::{PgConnection, PgPool};
805///
806/// #[derive(sqlx::FromRow, Clone)]
807/// pub struct SendEmail {
808/// job_id: i64,
809/// name: String,
810/// attempt: i16,
811/// to: String,
812/// subject: Option<String>,
813/// #[sqlx(json)]
814/// attachments: Vec<Attachment>,
815/// }
816///
817/// #[derive(Clone, Serialize, Deserialize)]
818/// pub struct Attachment {
819/// file: String,
820/// }
821///
822/// impl InboxTable for SendEmail {
823/// type Id = i64;
824/// type Table = InboxSpec<(Attempt,)>;
825/// const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
826/// .group(Column::new("name"))
827/// .attempt(Column::new("attempt").generated())
828/// .data(&[Column::new("to"), Column::new("subject"), Column::new("attachments")]);
829///
830/// fn id(&self) -> &i64 {
831/// &self.job_id
832/// }
833/// }
834///
835/// // Row mode: a handler takes the row itself.
836/// impl Input for SendEmail {
837/// type Axis = SoloCarried<Self>;
838/// }
839///
840/// impl AttemptRow for SendEmail {
841/// type Attempt = i16;
842///
843/// fn attempt(&self) -> &i16 {
844/// &self.attempt
845/// }
846/// }
847///
848/// // INSERT INTO "email_jobs" ("name", "to", "subject", "attachments") VALUES ($1, $2, $3, $4)
849/// const INSERT: Sql<128> = insert::postgres(&SendEmail::TABLE.spec());
850///
851/// impl Insert<PgConnection> for SendEmail {
852/// async fn insert(&self, conn: &mut PgConnection) -> Result<(), sqlx::Error> {
853/// sqlx::query(INSERT.as_str())
854/// .bind(&self.name)
855/// .bind(&self.to)
856/// .bind(&self.subject)
857/// .bind(sqlx::types::Json(&self.attachments))
858/// .execute(conn)
859/// .await?;
860/// Ok(())
861/// }
862/// }
863/// # async fn deliver(_: &SendEmail) -> bool { true }
864/// # #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
865/// # async fn send(email: &SendEmail) -> HandlerOutcome {
866/// # if deliver(email).await { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
867/// # }
868/// # pub fn app(pool: PgPool) -> RustStream {
869/// # RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
870/// # b.include(send);
871/// # })
872/// # }
873/// # }
874/// # fn main() {}
875/// ```
876#[diagnostic::on_unimplemented(
877 message = "`{Self}` has no generated insert for `{Connection}`",
878 label = "no `Insert` for this connection",
879 note = "the derive generates it for the connections of the built-in dialects (features \
880 `postgres`, `mysql` and `sqlite`), unless a `#[sqlx(flatten)]` field hides columns \
881 from it: write that insert in the service"
882)]
883pub trait Insert<Connection>: Sync {
884 /// Inserts the row on `conn`.
885 ///
886 /// # Errors
887 ///
888 /// The database's error.
889 ///
890 /// # Examples
891 ///
892 /// ```
893 /// # #[cfg(feature = "postgres")]
894 /// # mod demo {
895 /// use ruststream_sqlx::prelude::*;
896 /// use serde::Deserialize;
897 /// use sqlx::{PgPool, Postgres};
898 ///
899 /// #[derive(Inbox, sqlx::FromRow)]
900 /// #[inbox(table = "signup_jobs")]
901 /// pub struct Signup {
902 /// #[field(id, generated)]
903 /// id: i64,
904 /// #[field(payload)]
905 /// payload: Vec<u8>,
906 /// }
907 ///
908 /// /// A welcome mail to send, in row mode: the handler of `welcome_jobs` takes the row itself.
909 /// #[derive(Debug, Clone, Inbox, sqlx::FromRow)]
910 /// #[inbox(table = "welcome_jobs")]
911 /// pub struct Welcome {
912 /// #[field(id, generated)]
913 /// id: i64,
914 /// email: String,
915 /// }
916 ///
917 /// #[derive(Deserialize)]
918 /// struct Account {
919 /// email: String,
920 /// }
921 ///
922 /// // The welcome row commits with the signup's acknowledgement, or neither does.
923 /// #[subscriber(InboxQueue::<Signup>::new("signups"))]
924 /// async fn open_account(account: &Account, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
925 /// let welcome = Welcome { id: 0, email: account.email.clone() };
926 /// match welcome.insert(&mut *tx).await {
927 /// Ok(()) => HandlerOutcome::ack(),
928 /// Err(_) => HandlerOutcome::retry(),
929 /// }
930 /// }
931 ///
932 /// pub fn app(pool: PgPool) -> RustStream {
933 /// RustStream::new(AppInfo::new("accounts", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
934 /// b.include(open_account.transactional());
935 /// })
936 /// }
937 /// # }
938 /// # fn main() {}
939 /// ```
940 fn insert<'c>(
941 &'c self,
942 conn: &'c mut Connection,
943 ) -> impl Future<Output = Result<(), Error>> + Send + 'c;
944}