brokkr-broker::dal::stacks

Rust

Data Access Layer for Stack operations.

This module provides functionality to interact with the database for Stack-related operations, including creating, retrieving, updating, and deleting stacks, as well as filtering stacks based on various criteria.

Structs

struct StacksDAL<'a>

pub

Data Access Layer for Stack operations.

Fields

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

Methods

create

pub

fn create(
    &self,
    new_stack: &NewStack,
) -> Result<Stack, diesel::result::Error>

Creates a new stack in the database.

Parameters:

Name	Type	Description
new_stack	&NewStack	A reference to the NewStack struct containing the stack details.

Returns:

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

Source

    pub fn create(&self, new_stack: &NewStack) -> Result<Stack, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        let stack: Stack = diesel::insert_into(stacks::table)
            .values(new_stack)
            .get_result(conn)?;

        // Emit stack.created event
        let event_data = serde_json::json!({
            "stack_id": stack.id,
            "name": stack.name,
            "description": stack.description,
            "created_at": stack.created_at,
        });
        event_bus::emit_event(self.dal, &BrokkrEvent::new(EVENT_STACK_CREATED, event_data));

        Ok(stack)
    }

get

pub

fn get(
    &self,
    stack_uuids: Vec<Uuid>,
) -> Result<Vec<Stack>, diesel::result::Error>

Retrieves non-deleted stacks by their UUIDs.

Parameters:

Name	Type	Description
stack_uuids	Vec<Uuid>	A Vec of UUIDs of the stacks to retrieve.

Returns:

Returns a Result containing a Vec of found non-deleted stacks, or a diesel::result::Error on failure.

Source

    pub fn get(&self, stack_uuids: Vec<Uuid>) -> Result<Vec<Stack>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        stacks::table
            .filter(stacks::id.eq_any(stack_uuids))
            .filter(stacks::deleted_at.is_null())
            .load::<Stack>(conn)
    }

get_​including_​deleted

pub

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

Retrieves a stack by its UUID, including deleted stacks.

Parameters:

Name	Type	Description
stack_uuid	Uuid	The UUID of the stack to retrieve.

Returns:

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

Source

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

list

pub

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

Lists all non-deleted stacks from the database.

Returns:

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

Source

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

list_​for_​generator

pub

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

Lists all non-deleted stacks owned by a specific generator.

Parameters:

Name	Type	Description
generator_id	Uuid	The UUID of the generator whose stacks to list.

Returns:

Returns a Result containing a Vec of non-deleted Stacks where generator_id matches, or a diesel::result::Error on failure.

Source

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

list_​all

pub

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

Lists all stacks from the database, including deleted ones.

Returns:

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

Source

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

update

pub

fn update(
    &self,
    stack_uuid: Uuid,
    updated_stack: &Stack,
) -> Result<Stack, diesel::result::Error>

Updates an existing stack in the database.

Parameters:

Name	Type	Description
stack_uuid	Uuid	The UUID of the stack to update.
updated_stack	&Stack	A reference to the Stack struct containing the updated details.

Returns:

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

Source

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

soft_​delete

pub

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

Soft deletes a stack by setting its deleted_at timestamp to the current time.

Parameters:

Name	Type	Description
stack_uuid	Uuid	The UUID of the stack 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, stack_uuid: Uuid) -> Result<usize, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        let rows = diesel::update(stacks::table.filter(stacks::id.eq(stack_uuid)))
            .set(stacks::deleted_at.eq(Utc::now()))
            .execute(conn)?;

        if rows > 0 {
            // Emit stack.deleted event
            let event_data = serde_json::json!({
                "stack_id": stack_uuid,
                "deleted_at": Utc::now(),
            });
            event_bus::emit_event(self.dal, &BrokkrEvent::new(EVENT_STACK_DELETED, event_data));
        }

        Ok(rows)
    }

hard_​delete

pub

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

Hard deletes a stack from the database.

Parameters:

Name	Type	Description
stack_uuid	Uuid	The UUID of the stack 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, stack_uuid: Uuid) -> Result<usize, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;
        diesel::delete(stacks::table.filter(stacks::id.eq(stack_uuid))).execute(conn)
    }

filter_​by_​labels

pub

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

Filters stacks by labels.

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 Stacks on success, or a diesel::result::Error on failure.

Source

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

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

                for label in &labels {
                    let subquery = stack_labels::table
                        .filter(stack_labels::stack_id.eq(stacks::id))
                        .filter(stack_labels::label.eq(label));
                    query = query.filter(diesel::dsl::exists(subquery));
                }

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

filter_​by_​annotations

pub

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

Filters stacks by annotations.

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 Stacks 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<Stack>, diesel::result::Error> {
        let conn = &mut self.dal.conn()?;

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

                for (key, value) in annotations {
                    let matching_stacks: Vec<Stack> = stacks::table
                        .inner_join(stack_annotations::table)
                        .filter(stacks::deleted_at.is_null())
                        .filter(stack_annotations::key.eq(key))
                        .filter(stack_annotations::value.eq(value))
                        .select(stacks::all_columns)
                        .load(conn)?;

                    all_matching_stacks.extend(matching_stacks);
                }

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

                // Use SQL EXISTS subqueries for efficient AND filtering (single query)
                let mut query = stacks::table
                    .filter(stacks::deleted_at.is_null())
                    .into_boxed();

                for (key, value) in &annotations {
                    let subquery = stack_annotations::table
                        .filter(stack_annotations::stack_id.eq(stacks::id))
                        .filter(stack_annotations::key.eq(key))
                        .filter(stack_annotations::value.eq(value));
                    query = query.filter(diesel::dsl::exists(subquery));
                }

                query
                    .select(stacks::all_columns)
                    .distinct()
                    .load::<Stack>(conn)
            }
        }
    }

get_​associated_​stacks

pub

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

Retrieves all stacks associated with a specific agent based on its labels, annotations, and targets.

This method uses OR logic when matching labels and annotations, meaning a stack will be included if it matches any of the agent’s labels or annotations.

Registration is the consent boundary (BROKKR-T-0287): label and annotation matches only associate stacks whose owning generator the agent is registered with. A generator declares which labels it pushes to; an agent’s registrations declare which generators it accepts stacks from, so matching selects within consented generators rather than creating responsibility across them. Explicit targets are exempt — they are already gated at creation time (403 agent_not_registered).

Parameters:

Name	Type	Description
agent_id	Uuid	The UUID of the agent.

Returns:

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

Source

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

        // Get agent labels
        let labels: Vec<String> = agent_labels::table
            .filter(agent_labels::agent_id.eq(agent_id))
            .select(agent_labels::label)
            .load::<String>(conn)?;

        // Get agent annotations
        let annotations: Vec<(String, String)> = agent_annotations::table
            .filter(agent_annotations::agent_id.eq(agent_id))
            .select((agent_annotations::key, agent_annotations::value))
            .load::<(String, String)>(conn)?;

        // Get agent targets
        let targets: Vec<Uuid> = agent_targets::table
            .filter(agent_targets::agent_id.eq(agent_id))
            .select(agent_targets::stack_id)
            .load::<Uuid>(conn)?;

        // Generators this agent has consented to receive stacks from.
        let registered_generators: HashSet<Uuid> = agent_generator_registrations::table
            .filter(agent_generator_registrations::agent_id.eq(agent_id))
            .select(agent_generator_registrations::generator_id)
            .load::<Uuid>(conn)?
            .into_iter()
            .collect();

        let mut associated_stacks = HashSet::new();

        // Get stacks matching labels, restricted to consented generators.
        if !labels.is_empty() && !registered_generators.is_empty() {
            let label_stacks = self.filter_by_labels(labels, FilterType::Or)?;

            associated_stacks.extend(
                label_stacks
                    .into_iter()
                    .filter(|s| registered_generators.contains(&s.generator_id)),
            );
        }

        // Get stacks matching annotations, restricted to consented generators.
        if !annotations.is_empty() && !registered_generators.is_empty() {
            let annotation_stacks = self.filter_by_annotations(annotations, FilterType::Or)?;

            associated_stacks.extend(
                annotation_stacks
                    .into_iter()
                    .filter(|s| registered_generators.contains(&s.generator_id)),
            );
        }

        // Get stacks matching targets
        if !targets.is_empty() {
            let target_stacks: Vec<Stack> = stacks::table
                .filter(stacks::id.eq_any(targets))
                .filter(stacks::deleted_at.is_null())
                .load::<Stack>(conn)?;

            associated_stacks.extend(target_stacks);
        }

        // Convert HashSet to Vec, sort, and return
        let mut result: Vec<Stack> = associated_stacks.into_iter().collect();
        result.sort_by(|a, b| a.id.cmp(&b.id));

        Ok(result)
    }