use sqlx::Connection as _;
use sqlx::postgres::{PgConnectOptions, PgConnection};
use std::str::FromStr as _;
use crate::secret::Secret;
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum InspectError {
#[error("{0}")]
InvalidInput(String),
#[error("could not connect to the database: {0}")]
Connect(String),
#[error("permission denied: {0}")]
Permission(String),
#[error("a catalog read failed: {0}")]
Query(String),
#[error("a write was attempted during the inspection; nothing was committed")]
WriteDetected,
#[error("{0}")]
ManagementUnauthorized(String),
}
impl InspectError {
#[must_use]
pub fn is_access(&self) -> bool {
matches!(
self,
Self::Connect(_) | Self::Permission(_) | Self::ManagementUnauthorized(_)
)
}
}
pub(crate) fn redact(message: &str, url: &Secret) -> String {
let mut message = message.replace(url.expose(), "[redacted]");
if let Ok(parsed) = url::Url::parse(url.expose())
&& let Some(password) = parsed.password()
&& !password.is_empty()
{
message = message.replace(password, "[redacted]");
let decoded = percent_decode(password);
if !decoded.is_empty() {
message = message.replace(&decoded, "[redacted]");
}
}
cratefield_core::scrub_text(&message)
}
fn percent_decode(text: &str) -> String {
let bytes = text.as_bytes();
let mut out = Vec::with_capacity(bytes.len());
let mut index = 0;
while index < bytes.len() {
if bytes[index] == b'%'
&& index + 2 < bytes.len()
&& text.is_char_boundary(index + 3)
&& let Ok(byte) = u8::from_str_radix(&text[index + 1..index + 3], 16)
{
out.push(byte);
index += 3;
} else {
out.push(bytes[index]);
index += 1;
}
}
String::from_utf8_lossy(&out).into_owned()
}
pub(crate) fn classify(error: &sqlx::Error, url: &Secret) -> InspectError {
let text = redact(&error.to_string(), url);
match error {
sqlx::Error::Database(db) => match db.code().as_deref() {
Some("42501") => InspectError::Permission(text),
Some(code) if code.starts_with("28") || code.starts_with("08") || code == "3D000" => {
InspectError::Connect(text)
}
_ => InspectError::Query(text),
},
sqlx::Error::Io(_)
| sqlx::Error::Tls(_)
| sqlx::Error::Configuration(_)
| sqlx::Error::Protocol(_)
| sqlx::Error::PoolTimedOut
| sqlx::Error::PoolClosed => InspectError::Connect(text),
_ => InspectError::Query(text),
}
}
pub struct ReadOnlySession {
pub(crate) conn: PgConnection,
pub(crate) url: Secret,
}
impl ReadOnlySession {
pub async fn open(url: &Secret) -> Result<Self, InspectError> {
let options = PgConnectOptions::from_str(url.expose())
.map_err(|_| {
InspectError::InvalidInput(
"the database URL is not a valid Postgres connection string".to_owned(),
)
})?
.application_name("cratefield-import-supabase")
.options([
("default_transaction_read_only", "on"),
("statement_timeout", "120000"),
]);
let mut conn = PgConnection::connect_with(&options)
.await
.map_err(|error| classify(&error, url))?;
for statement in [
"SET default_transaction_read_only = on",
"BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY",
] {
sqlx::raw_sql(statement)
.execute(&mut conn)
.await
.map_err(|error| classify(&error, url))?;
}
Ok(Self {
conn,
url: url.clone(),
})
}
pub(crate) fn error(&self, error: &sqlx::Error) -> InspectError {
classify(error, &self.url)
}
#[doc(hidden)]
pub async fn execute_for_test(&mut self, sql: &str) -> Result<(), InspectError> {
sqlx::raw_sql(sql)
.execute(&mut self.conn)
.await
.map(|_| ())
.map_err(|error| self.error(&error))
}
pub(crate) async fn finish(mut self) -> Result<bool, InspectError> {
let assigned: Option<String> =
sqlx::query_scalar("SELECT txid_current_if_assigned()::text")
.fetch_one(&mut self.conn)
.await
.map_err(|error| self.error(&error))?;
let _ = sqlx::raw_sql("ROLLBACK").execute(&mut self.conn).await;
let _ = self.conn.close().await;
if assigned.is_some() {
return Err(InspectError::WriteDetected);
}
Ok(true)
}
}