shoutdotdev/src/messages.rs

use anyhow::Context as _;
use chrono::{DateTime, TimeDelta, Utc};
use diesel::{
    dsl::{auto_type, AsSelect},
    pg::Pg,
    prelude::*,
};
use uuid::Uuid;

use crate::{
    channels::Channel,
    governors::{Governor, GovernorError},
    projects::{self, Project},
    schema::messages,
    teams::{self, Team},
};

pub use crate::schema::messages::{dsl, table};

const TEAM_GOVERNOR_DEFAULT_WINDOW_SIZE_SEC: i64 = 300;
const TEAM_GOVERNOR_DEFAULT_MAX_COUNT: i32 = 50;

/// A "/say" message queued for sending
#[derive(Associations, Clone, Debug, Identifiable, Queryable, Selectable)]
#[diesel(table_name = messages)]
#[diesel(belongs_to(Channel))]
pub struct Message {
    pub id: Uuid,
    pub project_id: Uuid,
    pub channel_id: Uuid,
    pub created_at: DateTime<Utc>,
    pub sent_at: Option<DateTime<Utc>>,
    pub failed_at: Option<DateTime<Utc>>,
    pub message: String,
}

impl Message {
    #[auto_type(no_type_alias)]
    pub fn all() -> _ {
        let select: AsSelect<Message, Pg> = Message::as_select();
        table.select(select)
    }

    #[auto_type(no_type_alias)]
    pub fn with_id<'a>(id: &'a Uuid) -> _ {
        dsl::id.eq(id)
    }

    #[auto_type(no_type_alias)]
    pub fn with_channel<'a>(channel_id: &'a Uuid) -> _ {
        dsl::channel_id.eq(channel_id)
    }

    #[auto_type(no_type_alias)]
    pub fn is_pending() -> _ {
        dsl::sent_at.is_null().and(dsl::failed_at.is_null())
    }
}

#[derive(Clone, Debug, Insertable)]
#[diesel(table_name = messages)]
struct InsertableMessage<'a> {
    id: Uuid,
    project_id: Uuid,
    channel_id: Uuid,
    message: &'a str,
}

#[derive(Clone, Debug)]
pub struct Broadcast {
    pub project_id: Uuid,
    pub message: String,
}

impl Broadcast {
    pub fn queue_messages(&self, db_conn: &mut PgConnection) -> Result<Vec<Uuid>, GovernorError> {
        let _guard = tracing::debug_span!(
            "Broadcast::queue_messages()",
            project = self.project_id.hyphenated().to_string()
        )
        .entered();
        let (project, team): (Project, Team) = projects::table
            .inner_join(teams::table)
            .filter(Project::with_id(&self.project_id))
            .first(db_conn)
            .context("failed to load project and team")?;
        let selected_channels = project
            .selected_channels()
            .load(db_conn)
            .context("failed to load selected channels")?;
        Governor::lazy_getter()
            .with_team_id(team.id)
            .with_default_window_size(TimeDelta::seconds(TEAM_GOVERNOR_DEFAULT_WINDOW_SIZE_SEC))
            .with_default_max_count(TEAM_GOVERNOR_DEFAULT_MAX_COUNT)
            .build()
            .context("failed to build governor lazy getter")?
            .load(db_conn)
            .context("failed to lazily get governor")?
            .create_entries(selected_channels.len(), db_conn)?;
        let mut message_ids: Vec<Uuid> = Vec::with_capacity(selected_channels.len());
        for channel in selected_channels {
            message_ids.push(
                diesel::insert_into(messages::table)
                    .values(InsertableMessage {
                        id: Uuid::now_v7(),
                        project_id: project.id,
                        channel_id: channel.id,
                        message: &self.message,
                    })
                    .returning(dsl::id)
                    .get_result(db_conn)
                    .context("failed to queue message")
                    .inspect_err(|_| {
                        tracing::error!("error queuing message for channel {}", channel.id);
                    })?,
            );
        }
        tracing::debug!("queued {} messages", message_ids.len());
        Ok(message_ids)
    }
}