194c44c
use models::chat::room_webhooks;
use rand::Rng;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::messages::SendMessageCmd;
use crate::services::service::ChatService;
use crate::share::authz::can_manage_space;
use crate::share::lookup::{is_room_member, lookup_room, require_room_readable};
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct WebhookDto {
pub id: Uuid,
pub room_id: Uuid,
pub name: String,
pub disabled: bool,
pub last_used_at: Option<OffsetDateTime>,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct WebhookSecretDto {
pub id: Uuid,
pub room_id: Uuid,
pub name: String,
pub token: String,
pub disabled: bool,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateWebhookCmd {
pub name: String,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateWebhookCmd {
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub disabled: Option<bool>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct WebhookPostCmd {
#[serde(default)]
pub content: String,
#[serde(default)]
pub nonce: Option<String>,
#[serde(default)]
pub thread_id: Option<Uuid>,
}
fn webhook_token() -> String {
const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
let mut rng = rand::thread_rng();
(0..48)
.map(|_| CHARSET[rng.gen_range(0..CHARSET.len())] as char)
.collect()
}
pub fn token_hash(token: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(token.as_bytes());
base64::Engine::encode(
&base64::engine::general_purpose::URL_SAFE_NO_PAD,
hasher.finalize(),
)
}
const HOOK_WINDOW_SECS: u64 = 60;
const HOOK_MAX_PER_WINDOW: u32 = 30;
impl From<room_webhooks::Model> for WebhookDto {
fn from(w: room_webhooks::Model) -> Self {
Self {
id: w.id,
room_id: w.room_id,
name: w.name,
disabled: w.disabled,
last_used_at: w.last_used_at,
created_at: w.created_at,
}
}
}
impl ChatService {
pub async fn create_webhook(
&self,
caller: i64,
room_id: Uuid,
cmd: CreateWebhookCmd,
) -> Result<WebhookSecretDto, ChatError> {
if !self
.state
.redis
.quota_allow(
&format!("quota:webhook_create:{caller}"),
self.config().quota.webhook_create_per_day,
86_400,
)
.await
{
return Err(ChatError::RateLimited(
"webhook creation quota exceeded, retry tomorrow".to_string(),
));
}
let ctx = require_room_readable(self, room_id, caller).await?;
match &ctx.space {
Some(space) => {
if !can_manage_space(self, space, caller).await? {
return Err(ChatError::Forbidden);
}
}
None => return Err(ChatError::Forbidden),
}
let name = cmd.name.trim().to_string();
if name.is_empty() || name.chars().count() > self.config().chat.max_webhook_name_len {
return Err(ChatError::BadRequest("invalid webhook name".to_string()));
}
let now = OffsetDateTime::now_utc();
let id = Uuid::now_v7();
let token = webhook_token();
let hash = token_hash(&token);
room_webhooks::ActiveModel {
id: Set(id),
room_id: Set(room_id),
user_id: Set(caller),
name: Set(name.clone()),
token_hash: Set(hash),
disabled: Set(false),
last_used_at: Set(None),
created_at: Set(now),
}
.insert(self.db().write())
.await?;
Ok(WebhookSecretDto {
id,
room_id,
name,
token,
disabled: false,
created_at: now,
})
}
pub async fn list_webhooks(
&self,
caller: i64,
room_id: Uuid,
) -> Result<Vec<WebhookDto>, ChatError> {
let ctx = require_room_readable(self, room_id, caller).await?;
match &ctx.space {
Some(space) => {
if !can_manage_space(self, space, caller).await? {
return Err(ChatError::Forbidden);
}
}
None => return Err(ChatError::Forbidden),
}
let rows = room_webhooks::Entity::find()
.filter(room_webhooks::Column::RoomId.eq(room_id))
.all(self.db().read())
.await?;
Ok(rows.into_iter().map(WebhookDto::from).collect())
}
pub async fn update_webhook(
&self,
caller: i64,
room_id: Uuid,
webhook_id: Uuid,
cmd: UpdateWebhookCmd,
) -> Result<WebhookDto, ChatError> {
let ctx = require_room_readable(self, room_id, caller).await?;
match &ctx.space {
Some(space) => {
if !can_manage_space(self, space, caller).await? {
return Err(ChatError::Forbidden);
}
}
None => {
if !is_room_member(self, room_id, caller).await? {
return Err(ChatError::Forbidden);
}
}
}
let row = room_webhooks::Entity::find_by_id(webhook_id)
.one(self.db().write())
.await?
.filter(|w| w.room_id == room_id)
.ok_or(ChatError::NotFound)?;
let mut am: room_webhooks::ActiveModel = row.into();
if let Some(name) = cmd.name {
let name = name.trim();
if name.is_empty() || name.chars().count() > self.config().chat.max_webhook_name_len {
return Err(ChatError::BadRequest("invalid webhook name".to_string()));
}
am.name = Set(name.to_string());
}
if let Some(disabled) = cmd.disabled {
am.disabled = Set(disabled);
}
let updated = am.update(self.db().write()).await?;
Ok(WebhookDto::from(updated))
}
pub async fn delete_webhook(
&self,
caller: i64,
room_id: Uuid,
webhook_id: Uuid,
) -> Result<(), ChatError> {
let ctx = require_room_readable(self, room_id, caller).await?;
match &ctx.space {
Some(space) => {
if !can_manage_space(self, space, caller).await? {
return Err(ChatError::Forbidden);
}
}
None => return Err(ChatError::Forbidden),
}
let deleted = room_webhooks::Entity::delete_many()
.filter(room_webhooks::Column::Id.eq(webhook_id))
.filter(room_webhooks::Column::RoomId.eq(room_id))
.exec(self.db().write())
.await?;
if deleted.rows_affected == 0 {
return Err(ChatError::NotFound);
}
Ok(())
}
pub async fn post_via_webhook(
&self,
token: &str,
cmd: WebhookPostCmd,
) -> Result<serde_json::Value, ChatError> {
let hash = token_hash(token);
if !self
.state
.redis
.quota_allow(
&format!("chat:webhook:{hash}"),
HOOK_MAX_PER_WINDOW as i64,
HOOK_WINDOW_SECS,
)
.await
{
return Err(ChatError::RateLimited(
"webhook rate limit exceeded, retry in 60s".to_string(),
));
}
let webhook = room_webhooks::Entity::find()
.filter(room_webhooks::Column::TokenHash.eq(&hash))
.one(self.db().write())
.await?
.ok_or(ChatError::Unauthorized)?;
if webhook.disabled {
return Err(ChatError::Forbidden);
}
let room = lookup_room(self, webhook.room_id).await?;
if room.space_id.is_none() {
return Err(ChatError::Forbidden);
}
let now = OffsetDateTime::now_utc();
let mut am: room_webhooks::ActiveModel = webhook.clone().into();
am.last_used_at = Set(Some(now));
am.update(self.db().write()).await?;
if let Some(thread_id) = cmd.thread_id {
let thread = models::chat::threads::Entity::find_by_id(thread_id)
.one(self.db().read())
.await?
.filter(|t| t.room_id == webhook.room_id)
.ok_or(ChatError::NotFound)?;
if thread.archived || thread.locked {
return Err(ChatError::Forbidden);
}
}
let message = self
.post_webhook_message(
webhook.room_id,
&webhook,
SendMessageCmd {
content: cmd.content,
reply_to_id: None,
thread_id: cmd.thread_id,
thread_broadcast: false,
attachments: Vec::new(),
nonce: cmd.nonce,
urgent: false,
forwarded_from: None,
},
)
.await?;
Ok(serde_json::to_value(message)?)
}
async fn post_webhook_message(
&self,
room_id: Uuid,
webhook: &room_webhooks::Model,
cmd: SendMessageCmd,
) -> Result<crate::services::messages::MessageDto, ChatError> {
use models::chat::messages::{self as chat_messages, AuthorType, ChatMessageType};
use models::room::rooms;
use sea_orm::TransactionTrait;
use sea_orm::sea_query::Expr;
let content = cmd.content.trim().to_string();
if content.is_empty() {
return Err(ChatError::BadRequest("content is empty".to_string()));
}
if content.chars().count() > self.config().chat.max_message_len {
return Err(ChatError::BadRequest("message too long".to_string()));
}
if let Some(nonce) = &cmd.nonce {
let existing = chat_messages::Entity::find()
.filter(chat_messages::Column::RoomId.eq(room_id))
.filter(chat_messages::Column::AuthorId.eq(webhook.user_id))
.filter(chat_messages::Column::Nonce.eq(nonce.as_str()))
.one(self.db().read())
.await?;
if let Some(existing) = existing {
let items = self
.build_message_dto(vec![existing], webhook.user_id)
.await?;
return Ok(items.into_iter().next().unwrap());
}
}
let now = OffsetDateTime::now_utc();
let message_id = Uuid::now_v7();
let content_clone = content.clone();
let webhook_id = webhook.id;
let owner = webhook.user_id;
let thread_id = cmd.thread_id;
let seq = self
.db()
.primary()
.transaction::<_, i64, ChatError>(move |txn| {
Box::pin(async move {
let seq = crate::share::notify::next_room_seq(txn, room_id).await?;
chat_messages::ActiveModel {
id: Set(message_id),
room_id: Set(room_id),
author_id: Set(owner),
author_type: Set(AuthorType::Webhook),
webhook_id: Set(Some(webhook_id)),
thread_id: Set(thread_id),
thread_broadcast: Set(false),
reply_to_id: Set(None),
forwarded_from_id: Set(None),
message_type: Set(ChatMessageType::Default),
content: Set(content_clone),
mentions: Set(serde_json::json!([])),
mention_everyone: Set(false),
mention_group_ids: Set(serde_json::json!([])),
attachments: Set(serde_json::json!([])),
embeds: Set(serde_json::json!([])),
flags: Set(0),
nonce: Set(cmd.nonce),
pinned: Set(false),
pinned_by: Set(None),
pinned_at: Set(None),
edited_at: Set(None),
deleted_at: Set(None),
seq: Set(seq),
created_at: Set(now),
}
.insert(txn)
.await?;
if let Some(tid) = thread_id
&& let Some(t) = models::chat::threads::Entity::find_by_id(tid)
.one(txn)
.await?
{
crate::services::threads::maintain_thread_on_reply(
txn, &t, message_id, owner, now,
)
.await?;
}
rooms::Entity::update_many()
.filter(rooms::Column::Id.eq(room_id))
.col_expr(rooms::Column::LastMessageId, Expr::value(message_id))
.col_expr(rooms::Column::UpdatedAt, Expr::value(now))
.exec(txn)
.await?;
Ok(seq)
})
})
.await?;
self.queue_search_message(message_id).await;
let row = chat_messages::Entity::find_by_id(message_id)
.one(self.db().read())
.await?
.ok_or(ChatError::Internal("message vanished".into()))?;
let items = self.build_message_dto(vec![row], owner).await?;
let dto = items.into_iter().next().unwrap();
let value = serde_json::to_value(&dto)?;
crate::share::notify::publish_message_event(self, room_id, seq, value).await;
crate::services::unfurl::spawn_unfurl(self, room_id, message_id, dto.content.clone());
Ok(dto)
}
}