brokkr-broker::cli::commands

Rust

Functions

fn connection_​pool_​from_​settings

pub

fn connection_pool_from_settings(
    config: &Settings,
    max_size: u32,
) -> ConnectionPool

Builds the database connection pool for a CLI entry point from config, honoring database.schema (BROKKR__DATABASE__SCHEMA).

Every command in this module that talks to the database must go through this: it is the single place that plumbs the configured schema into the pool. The schema is not applied once at construction — it is applied per connection by ConnectionPool::get, which issues SET search_path TO <schema> on each checkout. Callers must therefore take connections via pool.get() (which the DAL does internally), never pool.pool.get(), or the connection lands on public regardless of configuration.

Opening a bare PgConnection::establish(&config.database.url) bypasses all of this and silently operates on public; that was BROKKR-T-0297.

max_size is per-command: serve needs a large pool, one-shot subcommands need one connection (two where the operation emits webhook events, which need a second connection while the first is still held).

Source

pub fn connection_pool_from_settings(config: &Settings, max_size: u32) -> ConnectionPool {
    create_shared_connection_pool(
        &config.database.url,
        // `None`: honour the database named in `database.url`. This used to be
        // the literal "brokkr", which overwrote the operator's configured
        // database name (BROKKR-T-0306).
        None,
        max_size,
        config.database.schema.as_deref(),
    )
}

fn serve

pub

async fn serve(
    config: &Settings) -> Result<(,
) -> Result<(), Box<dyn std::error::Error>>

Function to start the Brokkr Broker server

This function initializes the database, runs migrations, checks for first-time setup, configures API routes, and starts the server with graceful shutdown support.

Source

pub async fn serve(config: &Settings) -> Result<(), Box<dyn std::error::Error>> {
    info!("Starting Brokkr Broker application");

    // Create database connection pool
    // Pool size needs to accommodate:
    // - 5 background tasks (diagnostic cleanup, work order maintenance, webhook delivery/cleanup, audit cleanup)
    // - HTTP requests (middleware holds 1 connection while DAL methods need another)
    // - Concurrent request handling and webhook event emission
    info!("Creating database connection pool");
    let connection_pool = connection_pool_from_settings(config, 50);
    info!("Database connection pool created successfully");

    // Set up schema if configured (for multi-tenant deployments)
    if let Some(ref schema) = config.database.schema {
        info!("Setting up schema: {}", schema);
        connection_pool
            .setup_schema(schema)
            .expect("Failed to set up schema");
        info!("Schema '{}' set up successfully", schema);
    }

    // Run pending migrations
    info!("Running pending database migrations");
    let mut conn = connection_pool.get().expect("Failed to get DB connection");

    conn.run_pending_migrations(MIGRATIONS)
        .expect("Failed to run migrations");
    info!("Database migrations completed successfully");

    // Check if this is the first time running the application
    let is_first_run = conn
        .transaction(|conn| {
            let result: Count =
                sql_query("SELECT COUNT(*) as count FROM app_initialization").get_result(conn)?;
            if result.count == 0 {
                // If it's the first run, insert a record into app_initialization
                sql_query("INSERT INTO app_initialization DEFAULT VALUES").execute(conn)?;
                Ok::<bool, DieselError>(true)
            } else {
                Ok::<bool, DieselError>(false)
            }
        })
        .expect("Failed to check initialization status");

    // Perform first-time setup if necessary
    if is_first_run {
        info!("First time application startup detected. Creating admin role...");
        utils::first_startup(&mut conn, config)?;
    } else {
        info!("Existing application detected. Proceeding with normal startup.");
    }

    // Backstop for the publicly-known default admin PAK (BROKKR-T-0298).
    //
    // Deliberately a warning, not a refusal — the opposite of the webhook
    // encryption-key guard below. Refusing here would need a development
    // opt-out flag (`angreal local up`, the compose harness and the integration
    // suites all boot with the shipped default), and such flags end up copied
    // into production manifests: a worse footgun than the one being closed.
    // Documentation (BROKKR-T-0286) is the primary control; this is the net
    // under it.
    //
    // Runs after the first-run branch so `admin_role` reflects whatever
    // `upsert_admin` just wrote, and reads the stored hash as well as the
    // configured one: an install that first booted with the default keeps
    // accepting it even after the config is corrected.
    let stored_admin_hash = utils::stored_admin_pak_hash(&mut conn).unwrap_or_else(|e| {
        warn!("Could not read the stored admin PAK hash to check it against the shipped default (the check is a backstop, so startup continues): {}", e);
        None
    });
    let default_admin_pak = utils::detect_default_admin_pak_hash(
        config.broker.pak_hash.as_deref(),
        stored_admin_hash.as_deref(),
    );
    utils::report_default_admin_pak_hash(&default_admin_pak);
    utils::start_default_admin_pak_reminder_task(default_admin_pak);

    // Initialize Data Access Layer
    info!("Initializing Data Access Layer");
    let auth_cache_ttl = config.broker.auth_cache_ttl_seconds.unwrap_or(60);
    let dal = DAL::new_with_auth_cache(connection_pool.clone(), auth_cache_ttl);
    info!(
        "Auth cache TTL: {}s ({})",
        auth_cache_ttl,
        if auth_cache_ttl > 0 {
            "enabled"
        } else {
            "disabled"
        }
    );

    // Guard against the unset-encryption-key trap (BROKKR-T-0288) before the
    // key is initialized. An unset key means a fresh random per-process key, so
    // any subscription written by a previous process is already undecryptable
    // and can never deliver again. Runs here because it needs migrations to
    // have created `webhook_subscriptions` and must precede
    // `init_encryption_key`, which would otherwise silently install the random
    // key. Fresh installs (zero rows) fall through to the warning path.
    let existing_webhook_subscriptions: Count =
        sql_query("SELECT COUNT(*) as count FROM webhook_subscriptions")
            .get_result(&mut conn)
            .expect("Failed to count webhook subscriptions");
    if let Err(e) = utils::encryption::check_webhook_encryption_key(
        config.broker.webhook_encryption_key.as_deref(),
        existing_webhook_subscriptions.count,
    ) {
        // Log as well as return: `main` renders the error with `Debug`, which
        // would escape this message into an unreadable single quoted string.
        error!("{}", e);
        return Err(e.into());
    }

    // Initialize encryption key for webhooks
    info!("Initializing encryption key");
    utils::encryption::init_encryption_key(config.broker.webhook_encryption_key.as_deref())
        .expect("Failed to initialize encryption key");

    // Mint the ephemeral read-only UI PAK for the served operator console
    // (BROKKR-T-0267). In-memory only; the token itself is never logged.
    info!("Minting ephemeral read-only UI PAK for the operator console");
    utils::ui_pak::init()?;

    // Provision system generator (idempotent — creates it on first run, no-op thereafter)
    info!("Provisioning system generator");
    let system_generator_id = dal
        .generators()
        .provision_system_generator()
        .expect("Failed to provision system generator");
    info!(%system_generator_id, "system generator ready");

    // Initialize audit logger for compliance tracking
    info!("Initializing audit logger");
    utils::audit::init_audit_logger(dal.clone()).expect("Failed to initialize audit logger");

    // Reloadable configuration is built *before* the background tasks, because
    // they take a handle to it and re-read their tunables on every tick
    // (BROKKR-T-0292). It used to be constructed after they had already
    // spawned, which is part of why a reload changed nothing.
    info!("Initializing reloadable configuration");
    let reloadable_config =
        ReloadableConfig::from_settings(config.clone(), std::env::var("BROKKR_CONFIG_FILE").ok());

    // Start background tasks
    info!("Starting background tasks");
    let cleanup_config = utils::background_tasks::DiagnosticCleanupConfig {
        interval_seconds: config
            .broker
            .diagnostic_cleanup_interval_seconds
            .unwrap_or(900),
        max_age_hours: config.broker.diagnostic_max_age_hours.unwrap_or(1),
    };
    utils::background_tasks::start_diagnostic_cleanup_task(
        dal.clone(),
        cleanup_config,
        Some(reloadable_config.clone()),
    );

    // Start work order maintenance task (retry processing and stale claim detection)
    let work_order_config = utils::background_tasks::WorkOrderMaintenanceConfig::default();
    utils::background_tasks::start_work_order_maintenance_task(dal.clone(), work_order_config);

    // Start agent metrics refresh task (keeps the active-agent and
    // heartbeat-age gauges fresh independent of GET /agents traffic).
    let agent_metrics_refresh_config =
        utils::background_tasks::AgentMetricsRefreshConfig::default();
    utils::background_tasks::start_agent_metrics_refresh_task(
        dal.clone(),
        agent_metrics_refresh_config,
    );

    // Start webhook delivery worker
    let webhook_delivery_config = utils::background_tasks::WebhookDeliveryConfig {
        interval_seconds: config.broker.webhook_delivery_interval_seconds.unwrap_or(5),
        batch_size: config.broker.webhook_delivery_batch_size.unwrap_or(50),
    };
    utils::background_tasks::start_webhook_delivery_task(
        dal.clone(),
        webhook_delivery_config,
        Some(reloadable_config.clone()),
    );

    // Start webhook cleanup task
    let webhook_cleanup_config = utils::background_tasks::WebhookCleanupConfig {
        interval_seconds: 3600, // Every hour
        retention_days: config.broker.webhook_cleanup_retention_days.unwrap_or(7),
    };
    utils::background_tasks::start_webhook_cleanup_task(
        dal.clone(),
        webhook_cleanup_config,
        Some(reloadable_config.clone()),
    );

    // Start audit log cleanup task
    let audit_cleanup_config = utils::background_tasks::AuditLogCleanupConfig {
        interval_seconds: 86400, // Daily
        retention_days: config.broker.audit_log_retention_days.unwrap_or(90),
    };
    utils::background_tasks::start_audit_log_cleanup_task(dal.clone(), audit_cleanup_config);

    // Start agent-events cleanup task (BROKKR-T-0228). Gated on a positive
    // retention window: 0 or unset disables eviction (historical behavior).
    let agent_events_retention_days = config.broker.agent_events_retention_days.unwrap_or(30);
    if agent_events_retention_days > 0 {
        let agent_events_cleanup_config = utils::background_tasks::AgentEventsCleanupConfig {
            interval_seconds: 3600, // Every hour
            retention_days: agent_events_retention_days,
        };
        utils::background_tasks::start_agent_events_cleanup_task(
            dal.clone(),
            agent_events_cleanup_config,
        );
    } else {
        info!(
            "Agent-events retention disabled (agent_events_retention_days = 0); skipping eviction task"
        );
    }

    // Start ConfigMap watcher for Kubernetes hot-reload (if running in K8s)
    if let Some(watcher_config) = utils::config_watcher::ConfigWatcherConfig::from_environment() {
        utils::config_watcher::start_config_watcher(reloadable_config.clone(), watcher_config);
    }

    // Configure API routes
    info!("Configuring API routes");
    let app = api::configure_api_routes(dal.clone(), &config.cors, Some(reloadable_config))
        .with_state(dal);

    // Set up the server address
    let addr = "0.0.0.0:3000";
    info!("Starting server on {}", addr);
    let listener = tokio::net::TcpListener::bind(addr).await?;

    // Set up shutdown signal handler
    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    tokio::spawn(async move {
        signal::ctrl_c().await.expect("Failed to listen for ctrl+c");
        shutdown_tx.send(()).ok();
    });

    // Start the server with graceful shutdown
    info!("Brokkr Broker is now running");
    axum::serve(listener, app)
        .with_graceful_shutdown(utils::shutdown(shutdown_rx))
        .await?;

    Ok(())
}

fn rotate_​admin

pub

fn rotate_admin(
    config: &Settings) -> Result<(,
) -> Result<(), Box<dyn std::error::Error>>

Function to rotate the admin key

This function generates a new admin key and updates it in the database.

The connection comes from connection_pool_from_settings rather than a bare PgConnection::establish, so database.schema is applied to it. With a bare connection no search_path was ever set and the rotation silently hit the public schema — failing to find the configured schema’s admin_role row, or rewriting public’s (BROKKR-T-0297). Output mirrors the sibling rotate commands, which print the new credential directly (rotate agent, rotate generator), and generate-pak, which prints both halves. This one used to print nothing at all and leave the minted PAK in a file nothing in the tree reads — and, worse, to report success on the branch that mints nothing (BROKKR-T-0317).

Source

pub fn rotate_admin(config: &Settings) -> Result<(), Box<dyn std::error::Error>> {
    info!("Rotating admin key");

    // Create database connection (search_path applied on checkout)
    let pool = migrated_pool_from_settings(config, 1)?;
    let mut conn = pool.get()?;

    let outcome = utils::upsert_admin(&mut conn, config)?;

    match outcome {
        utils::AdminPakOutcome::Minted { pak, hash } => {
            info!("Admin key rotated successfully");
            println!("Admin PAK rotated. A new credential was minted:");
            println!();
            println!("  PAK (secret — send as `Authorization: Bearer <PAK>`; store securely):");
            println!("    {pak}");
            println!();
            println!("  PAK hash (set as BROKKR__BROKER__PAK_HASH / the chart's broker.pakHash):");
            println!("    {hash}");
            println!();
            println!("The PAK is shown once and cannot be recovered from the hash. It was also");
            println!("written to /tmp/brokkr-keys/key.txt, which is deleted on graceful shutdown.");
            println!();
            println!("Set the hash in this broker's configuration now: it is currently unset, so");
            println!("the stored hash and the configured one disagree, and a later `rotate admin`");
            println!("would overwrite this credential with whatever the config then holds.");
        }
        utils::AdminPakOutcome::ReappliedConfigured { hash } => {
            // Deliberately not an error: re-applying the configured hash is the
            // supported way to commit a hash you minted yourself. It is only
            // misleading when the caller expected a new credential.
            info!("Admin hash re-applied from configuration; no new PAK was minted");
            println!("No new admin PAK was minted.");
            println!();
            println!("`broker.pak_hash` is set, so the configured hash was validated and");
            println!("re-applied to the admin role and admin generator:");
            println!();
            println!("    {hash}");
            println!();
            println!("Any PAK matching that hash keeps working; nothing was rotated or revoked.");
            println!("This is the correct path when committing a hash you minted yourself.");
            println!();
            println!("To actually replace the credential, either:");
            println!("  * run `brokkr-broker generate-pak`, set the printed hash as");
            println!("    BROKKR__BROKER__PAK_HASH (or the chart's broker.pakHash /");
            println!("    broker.pakHashExistingSecret), then re-run `rotate admin`; or");
            println!("  * clear the setting and let the broker mint one:");
            println!("      BROKKR__BROKER__PAK_HASH=\"\" brokkr-broker rotate admin");
        }
    }

    Ok(())
}

fn generate_​pak

pub

fn generate_pak(
    config: &Settings) -> Result<(,
) -> Result<(), Box<dyn std::error::Error>>

Mints a fresh PAK + hash pair offline, without touching the database or writing a key file.

This is the day-zero bootstrap path: an operator runs it before the broker has ever started, captures the hash, sets BROKKR__BROKER__PAK_HASH, and keeps the PAK as the admin credential. On first startup upsert_admin then takes the configured-hash branch and stores it directly — no PAK is generated server-side and no key file is dropped in /tmp/brokkr-keys/.

The hash is a plain SHA-256 of the key’s long token (no secret, no salt), so it is reproducible from the PAK but the PAK cannot be recovered from it. We only need the PAK controller (whose prefix/length come from config) to mint a key the broker will recognize; no database connection is opened.

Source

pub fn generate_pak(config: &Settings) -> Result<(), Box<dyn std::error::Error>> {
    info!("Minting admin PAK + hash pair (offline)");

    // Initialize the controller from config if it has not been already. This is
    // idempotent (OnceCell) and keeps the command self-contained rather than
    // relying on the caller having initialized it first.
    pak::create_pak_controller(Some(config)).map_err(|e| e.to_string())?;

    let (pak, pak_hash) = pak::create_pak()?;

    println!("Minted admin PAK (offline — nothing was written to the database):");
    println!();
    println!("  PAK (secret — send as `Authorization: Bearer <PAK>`; store securely):");
    println!("    {pak}");
    println!();
    println!("  PAK hash (set as BROKKR__BROKER__PAK_HASH before first startup):");
    println!("    {pak_hash}");
    println!();
    println!("Day-zero flow:");
    println!("  1. export BROKKR__BROKER__PAK_HASH={pak_hash}");
    println!("  2. Start the broker. First startup stores this hash on the admin role");
    println!("     and the admin generator; no key file is written to /tmp/brokkr-keys/.");
    println!("  3. Authenticate with the PAK above. It cannot be recovered from the hash,");
    println!("     so the hash is safe to keep in config while the PAK stays secret.");

    Ok(())
}

fn rotate_​agent_​key

pub

fn rotate_agent_key(
    config: &Settings,
    uuid: Uuid,
) -> Result<String, Box<dyn std::error::Error>>

Source

pub fn rotate_agent_key(
    config: &Settings,
    uuid: Uuid,
) -> Result<String, Box<dyn std::error::Error>> {
    info!("Rotating agent key");

    let pool = migrated_pool_from_settings(config, 1)?;
    let dal = DAL::new(pool.clone());

    let agent = dal.agents().get(uuid)?.ok_or("Agent not found")?;
    let (new_pak, new_pak_hash) = utils::pak::create_pak()?;
    dal.agents().update_pak_hash(agent.id, new_pak_hash)?;

    info!("Agent key rotated successfully for agent: {}", agent.name);
    audit_cli_pak_event(
        &dal,
        ACTION_PAK_ROTATED,
        RESOURCE_TYPE_AGENT,
        agent.id,
        &agent.name,
    );
    println!("Agent PAK rotated successfully:");
    println!("ID: {}", agent.id);
    println!("Name: {}", agent.name);
    println!("New PAK: {}", new_pak);
    Ok(new_pak)
}

fn rotate_​generator_​key

pub

fn rotate_generator_key(
    config: &Settings,
    uuid: Uuid,
) -> Result<String, Box<dyn std::error::Error>>

Source

pub fn rotate_generator_key(
    config: &Settings,
    uuid: Uuid,
) -> Result<String, Box<dyn std::error::Error>> {
    info!("Rotating generator key");

    let pool = migrated_pool_from_settings(config, 1)?;
    let dal = DAL::new(pool.clone());

    let generator = dal.generators().get(uuid)?.ok_or("Generator not found")?;

    let (new_pak, new_pak_hash) = utils::pak::create_pak()?;
    dal.generators()
        .update_pak_hash(generator.id, new_pak_hash)?;

    info!(
        "Generator key rotated successfully for generator: {}",
        generator.name
    );
    audit_cli_pak_event(
        &dal,
        ACTION_PAK_ROTATED,
        RESOURCE_TYPE_GENERATOR,
        generator.id,
        &generator.name,
    );
    println!("Generator PAK rotated successfully:");
    println!("ID: {}", generator.id);
    println!("Name: {}", generator.name);
    println!("New PAK: {}", new_pak);
    Ok(new_pak)
}

fn create_​agent

pub

fn create_agent(
    config: &Settings,
    name: String,
    cluster_name: String,
    generator_ids: Vec<uuid::Uuid>,
) -> Result<(), Box<dyn std::error::Error>>

Creates an agent and its initial PAK.

Mirrors POST /api/v1/agents (BROKKR-T-0289): the agent is always registered with the system generator, and additionally with any generators named in generator_ids, which are validated before anything is written. Registration is the consent boundary — an agent with no registrations receives no stacks from label or annotation matching (BROKKR-T-0287).

Source

pub fn create_agent(
    config: &Settings,
    name: String,
    cluster_name: String,
    generator_ids: Vec<uuid::Uuid>,
) -> Result<(), Box<dyn std::error::Error>> {
    info!("Creating new agent: {}", name);

    // Use pool size 2 because agent creation emits webhook events
    // which require a second connection while the first is still held
    let pool = migrated_pool_from_settings(config, 2)?;
    let dal = DAL::new(pool.clone());

    // Validate every requested generator before creating anything, so a typo
    // doesn't leave a half-registered agent behind.
    for gid in &generator_ids {
        if dal.generators().get(*gid)?.is_none() {
            return Err(format!("generator {} does not exist", gid).into());
        }
    }

    let new_agent = NewAgent::new(name, cluster_name)
        .map_err(|e| format!("Failed to create NewAgent: {}", e))?;

    let (pak, pak_hash) = pak::create_pak()?;

    let agent = dal.agents().create(&new_agent)?;
    dal.agents().update_pak_hash(agent.id, pak_hash)?;
    audit_cli_pak_event(
        &dal,
        ACTION_PAK_CREATED,
        RESOURCE_TYPE_PAK,
        agent.id,
        &agent.name,
    );

    // Register with the system generator (always), matching the API path.
    let mut registered: Vec<String> = Vec::new();
    if let Some(system_id) = dal.generators().get_system_generator_id()? {
        dal.agent_generator_registrations()
            .create(agent.id, system_id)?;
        registered.push("system (fleet scope)".to_string());
    } else {
        warn!(
            "System generator not found; agent {} created without fleet-scope registration. \
             Start the broker once to provision it, then register the agent.",
            agent.id
        );
    }

    // Register with any additional generators requested.
    for gid in &generator_ids {
        dal.agent_generator_registrations().create(agent.id, *gid)?;
        registered.push(gid.to_string());
    }

    info!("Successfully created agent with ID: {}", agent.id);
    println!("Agent created successfully:");
    println!("ID: {}", agent.id);
    println!("Name: {}", agent.name);
    println!("Cluster: {}", agent.cluster_name);
    println!("Initial PAK: {}", pak);
    println!("Registered with: {}", registered.join(", "));

    Ok(())
}

fn create_​generator

pub

fn create_generator(
    config: &Settings,
    name: String,
    description: Option<String>,
) -> Result<(), Box<dyn std::error::Error>>

Source

pub fn create_generator(
    config: &Settings,
    name: String,
    description: Option<String>,
) -> Result<(), Box<dyn std::error::Error>> {
    info!("Creating new generator: {}", name);

    let pool = migrated_pool_from_settings(config, 1)?;
    let dal = DAL::new(pool.clone());

    let new_generator = NewGenerator::new(name, description)
        .map_err(|e| format!("Failed to create NewGenerator: {}", e))?;

    let (pak, pak_hash) = pak::create_pak()?;

    let generator = dal.generators().create(&new_generator)?;
    dal.generators().update_pak_hash(generator.id, pak_hash)?;
    audit_cli_pak_event(
        &dal,
        ACTION_PAK_CREATED,
        RESOURCE_TYPE_PAK,
        generator.id,
        &generator.name,
    );

    info!("Successfully created generator with ID: {}", generator.id);
    println!("Generator created successfully:");
    println!("ID: {}", generator.id);
    println!("Name: {}", generator.name);
    println!("Initial PAK: {}", pak);

    Ok(())
}

Constants

Name	Type	Value	Description
MIGRATIONS	EmbeddedMigrations	embed_migrations!("../brokkr-models/migrations")