121 lines
3.7 KiB
Rust
121 lines
3.7 KiB
Rust
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)
|
|
}
|
|
}
|