brokkr-broker::dal::agents

Rust

Structs

struct AgentFilter

pub

Struct for filtering agents based on various criteria.

Fields

Name	Type	Description
labels	Vec<String>
annotations	Vec<(String, String)>
agent_targets	Vec<Uuid>
filter_type	FilterType

struct AgentsDAL<'a>

pub

Data Access Layer for Agent operations.

Fields

Name	Type	Description
dal	&'a DAL	Reference to the main DAL instance.

Methods

create

pub

fn create(
    &self,
    new_agent: &NewAgent,
) -> Result<Agent, diesel::result::Error>

Creates a new agent in the database.

Parameters:

Name	Type	Description
new_agent	&NewAgent	A NewAgent struct containing the agent’s details.

Returns:

Returns a Result containing the created Agent on success, or a diesel::result::Error on failure.

Examples:

use brokkr_broker::dal::DAL;
use brokkr_models::models::agents::NewAgent;

let dal = DAL::new(pool);
let new_agent = NewAgent::new("Agent Name", "Cluster Name").unwrap();
let created_agent = dal.agents().create(&new_agent).expect("Failed to create agent");

Source

    pub fn create(&self, new_agent: &NewAgent) -> Result<Agent, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        let agent: Agent = diesel::insert_into(agents::table)
            .values(new_agent)
            .get_result(conn)?;

        // Emit agent.registered event
        let event_data = serde_json::json!({
            "agent_id": agent.id,
            "name": agent.name,
            "cluster_name": agent.cluster_name,
            "status": agent.status,
            "created_at": agent.created_at,
        });
        event_bus::emit_event(
            self.dal,
            &BrokkrEvent::new(EVENT_AGENT_REGISTERED, event_data),
        );

        Ok(agent)
    }

get

pub

fn get(
    &self,
    agent_uuid: Uuid,
) -> Result<Option<Agent>, diesel::result::Error>

Retrieves a non-deleted agent by its UUID.

Parameters:

Name	Type	Description
agent_uuid	Uuid	The UUID of the agent to retrieve.

Returns:

Returns a Result containing an Option<Agent> if found (and not deleted), or a diesel::result::Error on failure.

Source

    pub fn get(&self, agent_uuid: Uuid) -> Result<Option<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        agents::table
            .filter(agents::id.eq(agent_uuid))
            .filter(agents::deleted_at.is_null())
            .first(conn)
            .optional()
    }

get_​including_​deleted

pub

fn get_including_deleted(
    &self,
    agent_uuid: Uuid,
) -> Result<Option<Agent>, diesel::result::Error>

Retrieves an agent by its UUID, including deleted agents.

Parameters:

Name	Type	Description
agent_uuid	Uuid	The UUID of the agent to retrieve.

Returns:

Returns a Result containing an Option<Agent> if found (including deleted agents), or a diesel::result::Error on failure.

Source

    pub fn get_including_deleted(
        &self,
        agent_uuid: Uuid,
    ) -> Result<Option<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        agents::table
            .filter(agents::id.eq(agent_uuid))
            .first(conn)
            .optional()
    }

list

pub

fn list(&self) -> Result<Vec<Agent>, diesel::result::Error>

Lists all non-deleted agents from the database.

Returns:

Returns a Result containing a Vec of all non-deleted Agents on success, or a diesel::result::Error on failure.

Source

    pub fn list(&self) -> Result<Vec<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        agents::table
            .filter(agents::deleted_at.is_null())
            .load::<Agent>(conn)
    }

list_​for_​generator

pub

fn list_for_generator(
    &self,
    generator_id: Uuid,
) -> Result<Vec<Agent>, diesel::result::Error>

Lists the non-deleted agents registered with generator_id.

This is the tenant-scoped counterpart to Self::list (BROKKR-T-0315): a generator sees the agents that have explicitly registered with it, and nothing else. Soft-delete handling deliberately matches list, so the admin and generator paths through GET /agents cannot disagree about which agents exist.

Note that every agent is registered with the system generator, so calling this with the system generator’s id returns the whole fleet. Callers on the API path must reject system generators first; see require_tenant_generator in api/v1/agents.rs.

Source

    pub fn list_for_generator(
        &self,
        generator_id: Uuid,
    ) -> Result<Vec<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        agents::table
            .inner_join(agent_generator_registrations::table)
            .filter(agent_generator_registrations::generator_id.eq(generator_id))
            .filter(agents::deleted_at.is_null())
            .select(agents::all_columns)
            .load::<Agent>(conn)
    }

get_​names_​by_​ids

pub

fn get_names_by_ids(
    &self,
    ids: &[Uuid],
) -> Result<Vec<(Uuid, String)>, diesel::result::Error>

Resolves agent names for a set of IDs in one query (audit-log actor enrichment, BROKKR-T-0271). Includes soft-deleted agents so historical audit entries still resolve.

Source

    pub fn get_names_by_ids(
        &self,
        ids: &[Uuid],
    ) -> Result<Vec<(Uuid, String)>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        agents::table
            .filter(agents::id.eq_any(ids))
            .select((agents::id, agents::name))
            .load::<(Uuid, String)>(conn)
    }

list_​all

pub

fn list_all(&self) -> Result<Vec<Agent>, diesel::result::Error>

Lists all agents from the database, including deleted ones.

Returns:

Returns a Result containing a Vec of all Agents (including deleted ones) on success, or a diesel::result::Error on failure.

Source

    pub fn list_all(&self) -> Result<Vec<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        agents::table.load::<Agent>(conn)
    }

update

pub

fn update(
    &self,
    agent_uuid: Uuid,
    updated_agent: &Agent,
) -> Result<Agent, diesel::result::Error>

Updates an existing agent in the database.

Parameters:

Name	Type	Description
agent_uuid	Uuid	The UUID of the agent to update.
updated_agent	&Agent	A reference to the Agent struct containing the updated details.

Returns:

Returns a Result containing the updated Agent on success, or a diesel::result::Error on failure.

Source

    pub fn update(
        &self,
        agent_uuid: Uuid,
        updated_agent: &Agent,
    ) -> Result<Agent, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        diesel::update(agents::table.filter(agents::id.eq(agent_uuid)))
            .set(updated_agent)
            .get_result(conn)
    }

soft_​delete

pub

fn soft_delete(
    &self,
    agent_uuid: Uuid,
) -> Result<usize, diesel::result::Error>

Soft deletes an agent by setting its deleted_at timestamp to the current time.

Parameters:

Name	Type	Description
agent_uuid	Uuid	The UUID of the agent to soft delete.

Returns:

Returns a Result containing the number of affected rows (0 or 1) on success, or a diesel::result::Error on failure.

Source

    pub fn soft_delete(&self, agent_uuid: Uuid) -> Result<usize, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        let rows = diesel::update(agents::table.filter(agents::id.eq(agent_uuid)))
            .set(agents::deleted_at.eq(Utc::now()))
            .execute(conn)?;

        if rows > 0 {
            // Emit agent.deregistered event
            let event_data = serde_json::json!({
                "agent_id": agent_uuid,
                "deleted_at": Utc::now(),
            });
            event_bus::emit_event(
                self.dal,
                &BrokkrEvent::new(EVENT_AGENT_DEREGISTERED, event_data),
            );
        }

        Ok(rows)
    }

hard_​delete

pub

fn hard_delete(
    &self,
    agent_uuid: Uuid,
) -> Result<usize, diesel::result::Error>

Hard deletes an agent from the database.

Parameters:

Name	Type	Description
agent_uuid	Uuid	The UUID of the agent to hard delete.

Returns:

Returns a Result containing the number of affected rows (0 or 1) on success, or a diesel::result::Error on failure.

Source

    pub fn hard_delete(&self, agent_uuid: Uuid) -> Result<usize, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        diesel::delete(agents::table.filter(agents::id.eq(agent_uuid))).execute(conn)
    }

filter_​by_​labels

pub

fn filter_by_labels(
    &self,
    labels: Vec<String>,
    filter_type: FilterType,
) -> Result<Vec<Agent>, diesel::result::Error>

Filters agents by labels.

SQL Queries

For FilterType::And:

SELECT DISTINCT agents.*
FROM agents
WHERE agents.deleted_at IS NULL
  AND EXISTS (SELECT 1 FROM agent_labels WHERE agent_labels.agent_id = agents.id AND agent_labels.label = ?)
  AND EXISTS (SELECT 1 FROM agent_labels WHERE agent_labels.agent_id = agents.id AND agent_labels.label = ?)
  -- ... (repeated for each label)

For FilterType::Or:

SELECT DISTINCT agents.*
FROM agents
INNER JOIN agent_labels ON agents.id = agent_labels.agent_id
WHERE agents.deleted_at IS NULL
  AND agent_labels.label IN (?, ?, ...)

Parameters:

Name	Type	Description
labels	Vec<String>	A vector of label strings to filter by.
filter_type	FilterType	Specifies whether to use AND or OR logic for multiple labels.

Returns:

Returns a Result containing a Vec of matching Agents on success, or a diesel::result::Error on failure.

Source

    pub fn filter_by_labels(
        &self,
        labels: Vec<String>,
        filter_type: FilterType,
    ) -> Result<Vec<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;

        match filter_type {
            FilterType::And => {
                let mut query = agents::table
                    .filter(agents::deleted_at.is_null())
                    .into_boxed();

                for label in &labels {
                    let subquery = agent_labels::table
                        .filter(agent_labels::agent_id.eq(agents::id))
                        .filter(agent_labels::label.eq(label));
                    query = query.filter(diesel::dsl::exists(subquery));
                }

                query
                    .select(agents::all_columns)
                    .distinct()
                    .load::<Agent>(conn)
            }
            FilterType::Or => agents::table
                .inner_join(agent_labels::table)
                .filter(agents::deleted_at.is_null())
                .filter(agent_labels::label.eq_any(labels))
                .select(agents::all_columns)
                .distinct()
                .load::<Agent>(conn),
        }
    }

filter_​by_​annotations

pub

fn filter_by_annotations(
    &self,
    annotations: Vec<(String, String)>,
    filter_type: FilterType,
) -> Result<Vec<Agent>, diesel::result::Error>

Filters agents by annotations.

SQL Generated (Roughly Equivalent)

For FilterType::Or:

SELECT DISTINCT agents.*
FROM agents
INNER JOIN agent_annotations ON agents.id = agent_annotations.agent_id
WHERE agents.deleted_at IS NULL
  AND ((agent_annotations.key = ? AND agent_annotations.value = ?)
    OR (agent_annotations.key = ? AND agent_annotations.value = ?)
    OR ...)

For FilterType::And:

SELECT DISTINCT agents.*
FROM agents
WHERE agents.deleted_at IS NULL
  AND EXISTS (SELECT 1 FROM agent_annotations WHERE agent_annotations.agent_id = agents.id AND agent_annotations.key = ? AND agent_annotations.value = ?)
  AND EXISTS (SELECT 1 FROM agent_annotations WHERE agent_annotations.agent_id = agents.id AND agent_annotations.key = ? AND agent_annotations.value = ?)
  -- ... (repeated for each annotation pair)

Note: The actual implementation uses Rust code to perform the filtering, which may not directly translate to a single SQL query.

Parameters:

Name	Type	Description
annotations	Vec<(String, String)>	A vector of (key, value) pairs to filter by.
filter_type	FilterType	Specifies whether to use AND or OR logic for multiple annotations.

Returns:

Returns a Result containing a Vec of matching Agents on success, or a diesel::result::Error on failure.

Source

    pub fn filter_by_annotations(
        &self,
        annotations: Vec<(String, String)>,
        filter_type: FilterType,
    ) -> Result<Vec<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;

        match filter_type {
            FilterType::Or => {
                let mut all_matching_agents = HashSet::new();

                for (key, value) in annotations {
                    let matching_agents: Vec<Agent> = agents::table
                        .inner_join(agent_annotations::table)
                        .filter(agents::deleted_at.is_null())
                        .filter(agent_annotations::key.eq(key))
                        .filter(agent_annotations::value.eq(value))
                        .select(agents::all_columns)
                        .load(conn)?;

                    all_matching_agents.extend(matching_agents);
                }

                Ok(all_matching_agents.into_iter().collect())
            }
            FilterType::And => {
                if annotations.is_empty() {
                    return Ok(Vec::new());
                }

                let mut all_matching_agents: Option<HashSet<Agent>> = None;

                for (key, value) in annotations {
                    let matching_agents: HashSet<Agent> = agents::table
                        .inner_join(agent_annotations::table)
                        .filter(agents::deleted_at.is_null())
                        .filter(agent_annotations::key.eq(key))
                        .filter(agent_annotations::value.eq(value))
                        .select(agents::all_columns)
                        .load(conn)?
                        .into_iter()
                        .collect();

                    all_matching_agents = match all_matching_agents {
                        Some(agents) => {
                            Some(agents.intersection(&matching_agents).cloned().collect())
                        }
                        None => Some(matching_agents),
                    };

                    if let Some(ref agents) = all_matching_agents
                        && agents.is_empty()
                    {
                        break;
                    }
                }

                Ok(
                    all_matching_agents
                        .map_or_else(Vec::new, |agents| agents.into_iter().collect()),
                )
            }
        }
    }

get_​agent_​by_​target_​id

pub

fn get_agent_by_target_id(
    &self,
    agent_target_id: Uuid,
) -> Result<Option<Agent>, diesel::result::Error>

Retrieves an agent by its target ID.

Parameters:

Name	Type	Description
agent_target_id	Uuid	The UUID of the agent target to retrieve.

Returns:

Returns a Result containing an Option<Agent> if found, or a diesel::result::Error on failure.

Source

    pub fn get_agent_by_target_id(
        &self,
        agent_target_id: Uuid,
    ) -> Result<Option<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        agents::table
            .inner_join(agent_targets::table)
            .filter(agents::deleted_at.is_null())
            .filter(agent_targets::stack_id.eq(agent_target_id))
            .select(agents::all_columns)
            .first(conn)
            .optional()
    }

get_​agent_​details

pub

fn get_agent_details(
    &self,
    agent_id: Uuid,
) -> Result<
    (Vec<AgentLabel>, Vec<AgentTarget>, Vec<AgentAnnotation>),
    diesel::result::Error,
>

Retrieves labels, targets, and annotations associated with a specific agent.

Parameters:

Name	Type	Description
agent_id	Uuid	The UUID of the agent to retrieve details for.

Returns:

Returns a Result containing a tuple of (Vec<AgentLabel>, Vec<AgentTarget>, Vec<AgentAnnotation>) on success, or a diesel::result::Error on failure.

Source

    pub fn get_agent_details(
        &self,
        agent_id: Uuid,
    ) -> Result<(Vec<AgentLabel>, Vec<AgentTarget>, Vec<AgentAnnotation>), diesel::result::Error>
    {
        let conn = &mut self.dal.conn()?;

        let labels = agent_labels::table
            .filter(agent_labels::agent_id.eq(agent_id))
            .load::<AgentLabel>(conn)?;

        let targets = agent_targets::table
            .filter(agent_targets::agent_id.eq(agent_id))
            .load::<AgentTarget>(conn)?;

        let annotations = agent_annotations::table
            .filter(agent_annotations::agent_id.eq(agent_id))
            .load::<AgentAnnotation>(conn)?;

        Ok((labels, targets, annotations))
    }

record_​heartbeat

pub

fn record_heartbeat(
    &self) -> Result<(,
    agent_id: Uuid) -> Result<(,
) -> Result<(), diesel::result::Error>

Records a heartbeat for the specified agent.

Parameters:

Name	Type	Description
agent_id	Uuid	The UUID of the agent to record the heartbeat for.

Returns:

Returns a Result<()> on success, or a diesel::result::Error on failure.

Source

    pub fn record_heartbeat(&self, agent_id: Uuid) -> Result<(), diesel::result::Error> {
        let conn = &mut self.dal.conn()?;

        diesel::update(agents::table.filter(agents::id.eq(agent_id)))
            .set(agents::last_heartbeat.eq(diesel::dsl::now))
            .execute(conn)?;

        Ok(())
    }

record_​k8s_​connectivity

pub

fn record_k8s_connectivity(
    &self,
    agent_id: Uuid,
    reachable: bool,
    latency_ms: Option<i32>,
) -> Result<(), diesel::result::Error>

Stores the latest agent-reported Kubernetes connectivity snapshot (BROKKR-T-0227).

Sets k8s_reachable, k8s_api_latency_ms and stamps k8s_reported_at with the server-side ingestion time so readers can judge freshness. The broker trusts the agent-reported values as-is (it cannot compute them itself); latency_ms is optional because an agent may know reachability without a usable latency measurement.

Parameters:

Name	Type	Description
agent_id	Uuid	The UUID of the reporting agent.
reachable	bool	Whether the agent can reach its own Kubernetes API.
latency_ms	Option<i32>	Optional measured probe latency in milliseconds.

Returns:

Returns a Result<()> on success, or a diesel::result::Error on failure.

Source

    pub fn record_k8s_connectivity(
        &self,
        agent_id: Uuid,
        reachable: bool,
        latency_ms: Option<i32>,
    ) -> Result<(), diesel::result::Error> {
        let conn = &mut self.dal.conn()?;

        diesel::update(agents::table.filter(agents::id.eq(agent_id)))
            .set((
                agents::k8s_reachable.eq(Some(reachable)),
                agents::k8s_api_latency_ms.eq(latency_ms),
                agents::k8s_reported_at.eq(Some(Utc::now())),
            ))
            .execute(conn)?;

        Ok(())
    }

update_​pak_​hash

pub

fn update_pak_hash(
    &self,
    agent_uuid: Uuid,
    new_pak_hash: String,
) -> Result<Agent, diesel::result::Error>

Updates the pak_hash for an agent.

Parameters:

Name	Type	Description
agent_uuid	Uuid	The UUID of the agent to update.
new_pak_hash	String	The new pak_hash value.

Returns:

Returns a Result containing the updated Agent on success, or a diesel::result::Error on failure.

Source

    pub fn update_pak_hash(
        &self,
        agent_uuid: Uuid,
        new_pak_hash: String,
    ) -> Result<Agent, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        diesel::update(agents::table.filter(agents::id.eq(agent_uuid)))
            .set(agents::pak_hash.eq(new_pak_hash))
            .get_result(conn)
    }

get_​by_​name_​and_​cluster_​name

pub

fn get_by_name_and_cluster_name(
    &self,
    name: String,
    cluster_name: String,
) -> Result<Option<Agent>, diesel::result::Error>

Retrieves an agent by its name and cluster name.

Parameters:

Name	Type	Description
name	String	The name of the agent to retrieve.
cluster_name	String	The name of the cluster the agent belongs to.

Returns:

Returns a Result containing an Option<Agent> if found, or a diesel::result::Error on failure.

Source

    pub fn get_by_name_and_cluster_name(
        &self,
        name: String,
        cluster_name: String,
    ) -> Result<Option<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        let query = agents::table
            .filter(agents::name.eq(&name))
            .filter(agents::cluster_name.eq(&cluster_name))
            .filter(agents::deleted_at.is_null());

        query.first(conn).optional()
    }

get_​by_​pak_​hash

pub

fn get_by_pak_hash(
    &self,
    pak_hash: &str,
) -> Result<Option<Agent>, diesel::result::Error>

Retrieves an agent by its PAK hash.

This method uses the indexed pak_hash column for O(1) lookups instead of scanning the entire table.

Parameters:

Name	Type	Description
pak_hash	&str	The hashed PAK to search for.

Returns:

Returns a Result containing an Option<Agent> if found, or a diesel::result::Error on failure.

Source

    pub fn get_by_pak_hash(&self, pak_hash: &str) -> Result<Option<Agent>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        agents::table
            .filter(agents::pak_hash.eq(pak_hash))
            .filter(agents::deleted_at.is_null())
            .first(conn)
            .optional()
    }