awa_model/
bridge.rs

1//! Transactional enqueue for non-sqlx Postgres stacks.
2//!
3//! The core [`crate::insert::insert`] and [`crate::insert::insert_with`]
4//! functions require sqlx's `PgExecutor` trait. This module provides
5//! feature-gated adapters so users of other Postgres libraries can insert
6//! jobs within their existing transactions.
7//!
8//! All adapters share the same preparation logic in
9//! [`crate::adapter::postgres::prepare_job_insert`] — validation, state
10//! determination, unique key computation, and ordering-key propagation happen
11//! exactly once, avoiding semantic drift between drivers.
12//!
13//! # Available adapters
14//!
15//! | Feature | Module | Works with |
16//! |---------|--------|------------|
17//! | `tokio-postgres` | [`tokio_pg`] | `tokio_postgres::Client` and `Transaction` |
18
19// ---------------------------------------------------------------------------
20// tokio-postgres adapter
21// ---------------------------------------------------------------------------
22
23/// Adapter for inserting Awa jobs via `tokio_postgres` connections and
24/// transactions.
25///
26/// All functions are bounded on [`tokio_postgres::GenericClient`], which is
27/// implemented for `tokio_postgres::Client` and `tokio_postgres::Transaction`.
28///
29/// Pool wrappers like `deadpool_postgres::Client` typically deref to the
30/// underlying `tokio_postgres::Client`. Call `.transaction()` on the wrapper
31/// first, then pass the resulting `tokio_postgres::Transaction` to these
32/// functions.
33#[cfg(feature = "tokio-postgres")]
34pub mod tokio_pg {
35    use crate::adapter::postgres::{
36        prepare_job_insert, prepare_raw_job_insert, PreparedJobInsert, INSERT_JOB_SQL,
37    };
38    use crate::error::AwaError;
39    use crate::job::{InsertOpts, JobRow, JobState};
40    use crate::JobArgs;
41    use ::tokio_postgres::GenericClient;
42
43    /// Insert a job with default options.
44    ///
45    /// Mirrors [`crate::insert::insert`] but works with any
46    /// [`tokio_postgres::GenericClient`].
47    pub async fn insert_job<C: GenericClient>(
48        client: &C,
49        args: &impl JobArgs,
50    ) -> Result<JobRow, AwaError> {
51        insert_job_with(client, args, InsertOpts::default()).await
52    }
53
54    /// Insert a job with custom options.
55    ///
56    /// Mirrors [`crate::insert::insert_with`] but works with any
57    /// [`tokio_postgres::GenericClient`].
58    pub async fn insert_job_with<C: GenericClient>(
59        client: &C,
60        args: &impl JobArgs,
61        opts: InsertOpts,
62    ) -> Result<JobRow, AwaError> {
63        let prepared = prepare_job_insert(args, opts)?;
64        execute(client, &prepared).await
65    }
66
67    /// Insert a job from raw kind + JSON args + options.
68    ///
69    /// Use when you don't have a `JobArgs` impl — e.g. forwarding from
70    /// a dynamic source or another language.
71    pub async fn insert_job_raw<C: GenericClient>(
72        client: &C,
73        kind: String,
74        args: serde_json::Value,
75        opts: InsertOpts,
76    ) -> Result<JobRow, AwaError> {
77        let prepared = prepare_raw_job_insert(kind, args, opts)?;
78        execute(client, &prepared).await
79    }
80
81    async fn execute<C: GenericClient>(
82        client: &C,
83        prepared: &PreparedJobInsert,
84    ) -> Result<JobRow, AwaError> {
85        let state_str = prepared.state_db_str();
86
87        let row = client
88            .query_one(
89                INSERT_JOB_SQL,
90                &[
91                    &prepared.kind,
92                    &prepared.queue,
93                    &prepared.args,
94                    &state_str,
95                    &prepared.priority,
96                    &prepared.max_attempts,
97                    &prepared.run_at,
98                    &prepared.metadata,
99                    &prepared.tags,
100                    &prepared.unique_key,
101                    &prepared.unique_states,
102                    &prepared.ordering_key,
103                ],
104            )
105            .await
106            .map_err(|err| {
107                if let Some(db_err) = err.as_db_error() {
108                    if db_err.code() == &::tokio_postgres::error::SqlState::UNIQUE_VIOLATION {
109                        return AwaError::UniqueConflict {
110                            constraint: db_err.constraint().map(|c| c.to_string()),
111                        };
112                    }
113                }
114                AwaError::TokioPg(err)
115            })?;
116
117        JobRow::try_from(&row)
118    }
119
120    fn parse_state(state_str: &str) -> Result<JobState, AwaError> {
121        state_str
122            .parse()
123            .map_err(|e: String| AwaError::Validation(e))
124    }
125
126    /// Decode a column from a tokio-postgres row, mapping errors to
127    /// [`AwaError::Validation`] instead of panicking.
128    fn col<'a, T: ::tokio_postgres::types::FromSql<'a>>(
129        row: &'a ::tokio_postgres::Row,
130        name: &str,
131    ) -> Result<T, AwaError> {
132        row.try_get(name)
133            .map_err(|e| AwaError::Validation(format!("failed to decode column {name}: {e}")))
134    }
135
136    impl TryFrom<&::tokio_postgres::Row> for JobRow {
137        type Error = AwaError;
138
139        /// Convert a `RETURNING *, state::text AS state_str` row into a
140        /// full [`JobRow`].
141        fn try_from(row: &::tokio_postgres::Row) -> Result<Self, Self::Error> {
142            let state_str: String = col(row, "state_str")?;
143
144            Ok(JobRow {
145                id: col(row, "id")?,
146                kind: col(row, "kind")?,
147                queue: col(row, "queue")?,
148                args: col(row, "args")?,
149                state: parse_state(&state_str)?,
150                priority: col(row, "priority")?,
151                attempt: col(row, "attempt")?,
152                run_lease: col(row, "run_lease")?,
153                max_attempts: col(row, "max_attempts")?,
154                run_at: col(row, "run_at")?,
155                heartbeat_at: col(row, "heartbeat_at")?,
156                deadline_at: col(row, "deadline_at")?,
157                attempted_at: col(row, "attempted_at")?,
158                finalized_at: col(row, "finalized_at")?,
159                created_at: col(row, "created_at")?,
160                errors: col(row, "errors")?,
161                metadata: col(row, "metadata")?,
162                tags: col(row, "tags")?,
163                unique_key: col(row, "unique_key")?,
164                unique_states: None, // BIT(8) — no direct tokio-postgres mapping
165                callback_id: col(row, "callback_id")?,
166                callback_timeout_at: col(row, "callback_timeout_at")?,
167                callback_filter: col(row, "callback_filter")?,
168                callback_on_complete: col(row, "callback_on_complete")?,
169                callback_on_fail: col(row, "callback_on_fail")?,
170                callback_transform: col(row, "callback_transform")?,
171                progress: col(row, "progress")?,
172            })
173        }
174    }
175}
awa_model/bridge.rs

awa_model/
bridge.rs
