brokkr-broker::db

Rust

Database connection pool management using diesel and r2d2.

For detailed documentation, see the Brokkr Documentation.

Structs

struct QueryMetrics

pub

Derives: Debug, Default

Per-connection diesel hook that feeds the database query metrics (BROKKR-T-0265).

A connection runs one query at a time, so one start time per connection is enough: StartQuery stores it, FinishQuery takes it. The hook holds no heap data and reads neither the SQL text nor the bind values.

Methods

start

pub

fn start(&mut self)

Marks the start of a query on this connection.

Source

    pub fn start(&mut self) {
        self.started = Some(Instant::now());
    }

finish

pub

fn finish(&mut self, failed: bool)

Records the finish of a query on this connection.

A finish with no seen start is counted but not timed.

Source

    pub fn finish(&mut self, failed: bool) {
        let elapsed = self.started.take().map(|t| t.elapsed().as_secs_f64());
        crate::metrics::record_database_query(elapsed, failed);
    }

struct ConnectionPool

pub

Derives: Clone

Represents a pool of PostgreSQL database connections.

Fields

Name	Type	Description
pool	Pool<ConnectionManager<PgConnection>>	The actual connection pool.
schema	Option<String>	Optional schema name for multi-tenant deployments.

Methods

get

pub

fn get(
    &self,
) -> Result<
    diesel::r2d2::PooledConnection<ConnectionManager<PgConnection>>,
    r2d2::Error,
>

Gets a connection from the pool with automatic schema search_path configuration.

If a schema is configured, this method executes SET search_path TO <schema> on the connection, so all queries execute in that schema only. There is no public fallback (BROKKR-T-0361): a table that the schema does not have gives “relation does not exist”, and is never read from or written to public. pg_catalog stays on the path implicitly, so built-in functions such as gen_random_uuid() still resolve.

Returns:

Returns a pooled connection ready for use.

Panics:

• This method will panic if:
• Unable to get a connection from the pool
• The schema name is invalid
• Failed to set the search path

Source

    pub fn get(
        &self,
    ) -> Result<diesel::r2d2::PooledConnection<ConnectionManager<PgConnection>>, r2d2::Error> {
        use diesel::prelude::*;

        let mut conn = self.pool.get()?;

        if let Some(ref schema) = self.schema {
            // Validate schema name to prevent SQL injection
            validate_schema_name(schema).expect("Invalid schema name");

            // Set search_path for this connection: the schema only, no
            // `public` fallback (BROKKR-T-0361).
            let sql = format!("SET search_path TO {}", schema);
            diesel::sql_query(&sql)
                .execute(&mut conn)
                .expect("Failed to set search_path");
        }

        Ok(conn)
    }

setup_​schema

pub

fn setup_schema(&self, schema: &str) -> Result<(), String>

Sets up a PostgreSQL schema for multi-tenant isolation.

This method creates the schema if it doesn’t exist and prepares it for migrations. It should be called during application startup before running migrations.

Parameters:

Name	Type	Description
schema	&str	The schema name to set up

Returns:

Returns Ok(()) on success, or an error if schema setup fails.

Source

    pub fn setup_schema(&self, schema: &str) -> Result<(), String> {
        use diesel::prelude::*;

        // Validate schema name
        validate_schema_name(schema).map_err(|e| format!("Invalid schema name: {}", e))?;

        let mut conn = self
            .pool
            .get()
            .map_err(|e| format!("Failed to get connection: {}", e))?;

        // Create schema if it doesn't exist
        let create_schema_sql = format!("CREATE SCHEMA IF NOT EXISTS {}", schema);
        diesel::sql_query(&create_schema_sql)
            .execute(&mut conn)
            .map_err(|e| format!("Failed to create schema '{}': {}", schema, e))?;

        // Set search_path for subsequent operations
        let set_search_path_sql = format!("SET search_path TO {}", schema);
        diesel::sql_query(&set_search_path_sql)
            .execute(&mut conn)
            .map_err(|e| format!("Failed to set search path: {}", e))?;

        Ok(())
    }

check_​schema_​migrated

pub

fn check_schema_migrated(&self) -> Result<(), String>

Checks that the configured schema holds the Brokkr tables.

Each checkout sets search_path TO <schema>, and PostgreSQL accepts a schema that does not exist. Before BROKKR-T-0361 the path also held public, so a command run with a schema that is missing, misspelled or not migrated read and wrote the tables in public (BROKKR-T-0327). Now such a command fails on its first query with “relation does not exist”; this check fails first, with an error that names the schema and says what to do. serve creates and migrates the schema before it uses it, so only the one-shot commands need this check.

The check looks for diesel’s migration table in the schema itself, by a qualified name. With no schema configured there is nothing to check.

Source

    pub fn check_schema_migrated(&self) -> Result<(), String> {
        use diesel::prelude::*;
        use diesel::sql_types::{Bool, Text};

        let Some(ref schema) = self.schema else {
            return Ok(());
        };
        validate_schema_name(schema).map_err(|e| format!("Invalid schema name: {}", e))?;

        #[derive(QueryableByName)]
        struct Present {
            #[diesel(sql_type = Bool)]
            present: bool,
        }

        let mut conn = self
            .pool
            .get()
            .map_err(|e| format!("Failed to get connection: {}", e))?;
        let row: Present = diesel::sql_query("SELECT to_regclass($1) IS NOT NULL AS present")
            .bind::<Text, _>(format!("{}.__diesel_schema_migrations", schema))
            .get_result(&mut conn)
            .map_err(|e| format!("Failed to check schema '{}': {}", schema, e))?;

        if row.present {
            Ok(())
        } else {
            Err(format!(
                "database schema '{}' has no Brokkr tables. Check BROKKR__DATABASE__SCHEMA, \
                 or start the broker once with this schema so that it creates and migrates it",
                schema
            ))
        }
    }

Functions

fn create_​shared_​connection_​pool

pub

fn create_shared_connection_pool(
    base_url: &str,
    database_name: Option<&str>,
    max_size: u32,
    schema: Option<&str>,
) -> ConnectionPool

Creates a shared connection pool for PostgreSQL databases.

History

database_name was once a required &str and every production caller passed the literal "brokkr", so set_path silently discarded whatever database the operator had configured (BROKKR-T-0306). The chart renders postgresql.external.database into that URL, so any non-default name was ignored — connecting to a database the operator never named, or failing with an error that named one they had never heard of.

Parameters:

Name	Type	Description
base_url	&str	The full database URL, including the database name (e.g. postgres://username:password@localhost:5432/mydb)
database_name	Option<&str>	Optional override for the database component of base_url. Pass None in production, so the operator’s configured URL is honoured verbatim; it exists for test harnesses that point a shared base URL at a per-test database.
max_size	u32	The maximum number of connections the pool should maintain
schema	Option<&str>	Optional schema name for multi-tenant isolation

Returns:

Returns a ConnectionPool instance containing the created connection pool.

Panics:

• This function will panic if:
• The base URL is invalid
• The connection pool creation fails

Source

pub fn create_shared_connection_pool(
    base_url: &str,
    database_name: Option<&str>,
    max_size: u32,
    schema: Option<&str>,
) -> ConnectionPool {
    let url = resolve_database_url(base_url, database_name);

    // Create a connection manager
    let manager = ConnectionManager::<PgConnection>::new(url.as_str());

    // Build the connection pool. The customizer attaches the query metrics
    // hook to each connection (BROKKR-T-0265).
    let pool = Pool::builder()
        .max_size(max_size)
        .connection_customizer(Box::new(QueryMetricsCustomizer))
        .build(manager)
        .expect("Failed to create connection pool");

    ConnectionPool {
        pool,
        schema: schema.map(String::from),
    }
}

fn validate_​schema_​name

pub

fn validate_schema_name(schema: &str) -> Result<(), String>

Validates a PostgreSQL schema name to prevent SQL injection.

Schema names must start with a letter and contain only alphanumeric characters and underscores.

Parameters:

Name	Type	Description
schema	&str	The schema name to validate

Returns:

Returns Ok(()) if valid, or an error message if invalid.

Source

pub fn validate_schema_name(schema: &str) -> Result<(), String> {
    if schema.is_empty() {
        return Err("Schema name cannot be empty".to_string());
    }

    // Check first character is a letter
    if !schema.chars().next().unwrap().is_ascii_alphabetic() {
        return Err("Schema name must start with a letter".to_string());
    }

    // Check all characters are alphanumeric or underscore
    if !schema
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '_')
    {
        return Err("Schema name can only contain letters, numbers, and underscores".to_string());
    }

    Ok(())
}