use core::fmt;
use reliar_core::{Classify, FailureKind, MessageId};
use reliar_outbox::OutboxRecordId;
use crate::error::{classify_sqlstate, is_undefined_table};
#[cfg_attr(not(feature = "json"), doc = "```ignore")]
#[cfg_attr(feature = "json", doc = "```no_run")]
#[derive(Debug)]
#[non_exhaustive]
pub enum PostgresOutboxError {
SchemaNotOnSearchPath {
configured: String,
observed: String,
},
NotMigrated {
schema: String,
},
SchemaOutOfDate {
schema: String,
missing: &'static str,
},
Database {
source: sqlx::Error,
},
Decode {
id: OutboxRecordId,
message_id: MessageId,
detail: String,
},
UnknownMetadataVersion {
id: OutboxRecordId,
message_id: MessageId,
version: i32,
},
DuplicateMessage {
id: MessageId,
},
InvalidSchema {
schema: String,
},
UnsupportedServerVersion {
required: u32,
detected: u32,
},
}
impl fmt::Display for PostgresOutboxError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::SchemaNotOnSearchPath {
configured,
observed,
} => write!(
f,
"outbox did not resolve to schema \"{configured}\" (observed search_path: \
\"{observed}\"); set search_path so \"{configured}\" comes first, e.g. \
ALTER ROLE <role> SET search_path = {configured}, public"
),
Self::NotMigrated { schema } => write!(
f,
"relation \"{schema}.outbox\" does not exist; call \
reliar_store_postgres::migrate(&pool, ..) before constructing the store"
),
Self::SchemaOutOfDate { schema, missing } => write!(
f,
"relation \"{schema}.outbox\" does not satisfy required column \"{missing}\" \
(absent, or present but still nullable — migrations 0005-0010 have not all \
completed); run reliar_store_postgres::migrate(&pool, ..) before constructing \
the store"
),
Self::Database { source } => write!(f, "database error: {source}"),
Self::Decode {
id,
message_id,
detail,
} => write!(
f,
"row {id} (message {message_id}) could not be decoded: {detail}"
),
Self::UnknownMetadataVersion {
id,
message_id,
version,
} => write!(
f,
"row {id} (message {message_id}) carries unknown metadata_version {version}"
),
Self::DuplicateMessage { id } => {
write!(f, "message id {id} already exists in the outbox")
}
Self::InvalidSchema { schema } => write!(
f,
"{schema:?} is not a valid PostgreSQL identifier (expected \
[a-z_][a-z0-9_$]*, at most 63 bytes, lowercase only — PostgreSQL folds an \
unquoted identifier to lowercase, so an uppercase name would resolve \
inconsistently)"
),
Self::UnsupportedServerVersion { required, detected } => write!(
f,
"PostgreSQL 18 or newer is required (server_version_num >= {required}); \
detected {detected} — there is no supported way to run Reliar below the floor"
),
}
}
}
impl std::error::Error for PostgresOutboxError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Database { source } => Some(source),
_ => None,
}
}
}
impl Classify for PostgresOutboxError {
fn kind(&self) -> FailureKind {
match self {
Self::SchemaNotOnSearchPath { .. }
| Self::NotMigrated { .. }
| Self::SchemaOutOfDate { .. }
| Self::InvalidSchema { .. }
| Self::DuplicateMessage { .. }
| Self::Decode { .. }
| Self::UnknownMetadataVersion { .. }
| Self::UnsupportedServerVersion { .. } => FailureKind::Permanent,
Self::Database { source } => classify_sqlstate(source),
}
}
}
impl From<sqlx::Error> for PostgresOutboxError {
fn from(source: sqlx::Error) -> Self {
Self::Database { source }
}
}
pub(crate) fn map_operational_error(schema: &str, err: sqlx::Error) -> PostgresOutboxError {
if is_undefined_table(&err) {
return PostgresOutboxError::NotMigrated {
schema: schema.to_owned(),
};
}
PostgresOutboxError::Database { source: err }
}
#[cfg_attr(not(feature = "json"), doc = "```ignore")]
#[cfg_attr(feature = "json", doc = "```no_run")]
#[derive(Debug)]
#[non_exhaustive]
pub enum EnqueueError<E> {
Serialize {
source: E,
},
Duplicate {
id: MessageId,
},
Database {
source: sqlx::Error,
},
}
impl<E: fmt::Display> fmt::Display for EnqueueError<E> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Serialize { source } => {
write!(f, "failed to serialize the envelope body: {source}")
}
Self::Duplicate { id } => write!(f, "message id {id} already exists in the outbox"),
Self::Database { source } => write!(f, "database error: {source}"),
}
}
}
impl<E: std::error::Error + 'static> std::error::Error for EnqueueError<E> {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Serialize { source } => Some(source),
Self::Database { source } => Some(source),
Self::Duplicate { .. } => None,
}
}
}
impl<E: std::error::Error + Send + Sync + 'static> Classify for EnqueueError<E> {
fn kind(&self) -> FailureKind {
match self {
Self::Serialize { .. } | Self::Duplicate { .. } => FailureKind::Permanent,
Self::Database { source } => classify_sqlstate(source),
}
}
}
pub(crate) fn map_enqueue_error<E>(id: MessageId, err: sqlx::Error) -> EnqueueError<E> {
if is_constraint_violation(&err, "ix_outbox_message_id") {
return EnqueueError::Duplicate { id };
}
EnqueueError::Database { source: err }
}
pub(crate) fn is_constraint_violation(err: &sqlx::Error, constraint: &str) -> bool {
match err {
sqlx::Error::Database(db) => db.constraint() == Some(constraint),
_ => false,
}
}