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, pub sent_at: Option>, pub failed_at: Option>, pub message: String, } impl Message { #[auto_type(no_type_alias)] pub fn all() -> _ { let select: AsSelect = 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, 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 = 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) } }