diff --git a/Cargo.toml b/Cargo.toml index ceb436405..890f38565 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -80,6 +80,7 @@ hex = "0.4.3" sha1 = "0.10" sha2 = "0.10.9" rsa = "0.9.10" +hmac = "0.12" idgenerator = "2.0.0" config = "0.15.19" diff --git a/ceres/src/api_service/mono_api_service.rs b/ceres/src/api_service/mono_api_service.rs index 6df7d2b12..e66f0ac9d 100644 --- a/ceres/src/api_service/mono_api_service.rs +++ b/ceres/src/api_service/mono_api_service.rs @@ -537,6 +537,9 @@ impl ApiHandler for MonoApiService { let editor = OneditCodeEdit::from( repo_path, + MEGA_BRANCH_NAME + .strip_prefix("refs/heads/") + .unwrap_or(MEGA_BRANCH_NAME), &src_commit.id.to_string(), self, self.storage.mono_storage(), @@ -648,6 +651,9 @@ impl ApiHandler for MonoApiService { let editor = OneditCodeEdit::from( &repo_path_str, + MEGA_BRANCH_NAME + .strip_prefix("refs/heads/") + .unwrap_or(MEGA_BRANCH_NAME), &src_commit.id.to_string(), self, self.storage.mono_storage(), @@ -3125,12 +3131,22 @@ impl MonoApiService { .get_main_ref(&normalized_path) .await? .ok_or_else(|| MegaError::NotFound(format!("Path not found: {}", normalized_path)))?; + let base_branch = refs + .ref_name + .strip_prefix("refs/heads/") + .unwrap_or(refs.ref_name.as_str()) + .to_string(); // Use canonical path from mega_refs as the single source of truth for repository path let canonical_path = refs.path.clone(); let response = self .storage .buck_service - .create_session(username, &canonical_path, refs.ref_commit_hash) + .create_session( + username, + &canonical_path, + &base_branch, + refs.ref_commit_hash, + ) .await?; Ok(response) diff --git a/ceres/src/code_edit/model.rs b/ceres/src/code_edit/model.rs index 5a334b05b..43d6cb2ac 100644 --- a/ceres/src/code_edit/model.rs +++ b/ceres/src/code_edit/model.rs @@ -5,7 +5,7 @@ use callisto::{entity_ext::generate_link, mega_cl, mega_refs, sea_orm_active_enu use common::errors::MegaError; use git_internal::internal::object::commit::Commit; use jupiter::{ - service::reviewer_service::ReviewerService, + service::{reviewer_service::ReviewerService, webhook_service::WebhookEvent}, storage::{Storage, mono_storage::MonoStorage}, utils::converter::FromMegaModel, }; @@ -147,6 +147,7 @@ where DR: Director, { pub repo_path: String, + pub base_branch: String, pub from_hash: String, formator: FMT, clref_visitor: VT, @@ -202,6 +203,7 @@ impl< #[allow(clippy::too_many_arguments)] pub fn new( repo_path: &str, + base_branch: &str, from_hash: &str, formator: FMT, clref_visitor: VT, @@ -212,6 +214,7 @@ impl< ) -> Self { Self { repo_path: repo_path.to_string(), + base_branch: base_branch.to_string(), from_hash: from_hash.to_string(), formator, clref_visitor, @@ -279,6 +282,7 @@ impl< repo_path, &cl_link, &dst_commit.format_message(), + &self.base_branch, from_hash, to_hash, username, @@ -303,6 +307,9 @@ impl< ConvTypeEnum::Comment, ) .await?; + storage + .webhook_service + .dispatch(WebhookEvent::ClCreated, &cl); Ok(cl) } diff --git a/ceres/src/code_edit/on_edit.rs b/ceres/src/code_edit/on_edit.rs index e28ece593..324362533 100644 --- a/ceres/src/code_edit/on_edit.rs +++ b/ceres/src/code_edit/on_edit.rs @@ -96,12 +96,14 @@ pub(crate) type OneditCodeEdit = model::CodeEditService< impl OneditCodeEdit { pub fn from( repo_path: &str, + base_branch: &str, from_hash: &str, handler: &MonoApiService, mono_storage: MonoStorage, ) -> Self { Self::new( repo_path, + base_branch, from_hash, OneditFormator {}, OneditVisitor { mono_storage }, diff --git a/ceres/src/code_edit/on_push.rs b/ceres/src/code_edit/on_push.rs index 7e627444b..c031021a6 100644 --- a/ceres/src/code_edit/on_push.rs +++ b/ceres/src/code_edit/on_push.rs @@ -65,9 +65,15 @@ pub(crate) type OnpushCodeEdit = model::CodeEditService< // impl<'a> model::CodeEditService> { impl OnpushCodeEdit { - pub fn from(repo_path: &str, from_hash: &str, handler: &MonoApiService) -> Self { + pub fn from( + repo_path: &str, + base_branch: &str, + from_hash: &str, + handler: &MonoApiService, + ) -> Self { Self::new( repo_path, + base_branch, from_hash, OnpushFormator {}, OnpushVisitor {}, diff --git a/ceres/src/pack/monorepo.rs b/ceres/src/pack/monorepo.rs index e76fbde8b..ab7f2aa8c 100644 --- a/ceres/src/pack/monorepo.rs +++ b/ceres/src/pack/monorepo.rs @@ -50,6 +50,7 @@ pub struct MonoRepo { pub storage: Storage, pub git_object_cache: Arc, pub path: PathBuf, + pub base_branch: String, pub from_hash: String, pub to_hash: String, // current_commit only exists when an unpack operation occurs. @@ -155,6 +156,7 @@ impl RepoHandler for MonoRepo { let mono_api_service = self.into(); let editor = OnpushCodeEdit::from( self.path.to_str().unwrap(), + &self.base_branch, &self.from_hash, &mono_api_service, ); diff --git a/ceres/src/protocol/mod.rs b/ceres/src/protocol/mod.rs index 224119d5e..1cf30cf32 100644 --- a/ceres/src/protocol/mod.rs +++ b/ceres/src/protocol/mod.rs @@ -201,6 +201,7 @@ impl SmartProtocol { git_object_cache: state.git_object_cache.clone(), storage: state.storage.clone(), path: self.path.clone(), + base_branch: "main".to_string(), from_hash: String::new(), to_hash: String::new(), current_commit: Arc::new(RwLock::new(None)), @@ -215,6 +216,11 @@ impl SmartProtocol { { res.from_hash = command.old_id.clone(); res.to_hash = command.new_id.clone(); + res.base_branch = command + .ref_name + .strip_prefix("refs/heads/") + .unwrap_or(command.ref_name.as_str()) + .to_string(); } Ok(Arc::new(res)) } diff --git a/jupiter/Cargo.toml b/jupiter/Cargo.toml index b4161978d..5ec7e78e1 100644 --- a/jupiter/Cargo.toml +++ b/jupiter/Cargo.toml @@ -34,7 +34,7 @@ futures = { workspace = true } serde_json = { workspace = true } idgenerator = { workspace = true } serde = { workspace = true } -tokio = { workspace = true, features = ["macros", "rt-multi-thread", "io-std"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "io-std", "net"] } uuid = { workspace = true } tempfile = { workspace = true } indexmap = { workspace = true } @@ -48,9 +48,13 @@ redis = { workspace = true, features = [ rustls = { workspace = true } sha1 = { workspace = true } sha2 = { workspace = true } +hmac = { workspace = true } hex = { workspace = true } rand = { workspace = true } rsa = { workspace = true } +reqwest = { workspace = true } +base64 = { workspace = true } +ring = "0.17.14" [dev-dependencies] redis-test = { workspace = true, features = ["aio"] } diff --git a/jupiter/callisto/src/entity_ext/mega_cl.rs b/jupiter/callisto/src/entity_ext/mega_cl.rs index f23d3322a..934439301 100644 --- a/jupiter/callisto/src/entity_ext/mega_cl.rs +++ b/jupiter/callisto/src/entity_ext/mega_cl.rs @@ -52,13 +52,14 @@ impl Related for Entity { } impl mega_cl::Model { - pub fn new( - path: String, - title: String, - link: String, - from_hash: String, - to_hash: String, - username: String, + pub fn new( + path: String, + title: String, + link: String, + base_branch: String, + from_hash: String, + to_hash: String, + username: String, ) -> Self { let now = chrono::Utc::now().naive_utc(); Self { @@ -68,21 +69,23 @@ impl mega_cl::Model { status: MergeStatusEnum::Open, created_at: now, updated_at: now, - merge_date: None, - path, - from_hash, - to_hash, - username, + merge_date: None, + path, + base_branch, + from_hash, + to_hash, + username, } } /// Create a new CL with Draft status - pub fn new_draft( - path: String, - title: String, - link: String, - from_hash: String, - username: String, + pub fn new_draft( + path: String, + title: String, + link: String, + base_branch: String, + from_hash: String, + username: String, ) -> Self { let now = chrono::Utc::now().naive_utc(); Self { @@ -92,11 +95,12 @@ impl mega_cl::Model { status: MergeStatusEnum::Draft, created_at: now, updated_at: now, - merge_date: None, - path, - from_hash, - to_hash: String::new(), - username, + merge_date: None, + path, + base_branch, + from_hash, + to_hash: String::new(), + username, } } } diff --git a/jupiter/callisto/src/mega_cl.rs b/jupiter/callisto/src/mega_cl.rs index 803fca9e5..d2b6654ad 100644 --- a/jupiter/callisto/src/mega_cl.rs +++ b/jupiter/callisto/src/mega_cl.rs @@ -17,6 +17,7 @@ pub struct Model { pub status: MergeStatusEnum, #[sea_orm(column_type = "Text")] pub path: String, + pub base_branch: String, pub from_hash: String, pub to_hash: String, pub created_at: DateTime, diff --git a/jupiter/callisto/src/mega_webhook.rs b/jupiter/callisto/src/mega_webhook.rs new file mode 100644 index 000000000..83e708d5e --- /dev/null +++ b/jupiter/callisto/src/mega_webhook.rs @@ -0,0 +1,44 @@ +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "mega_webhook")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub id: i64, + pub target_url: String, + pub secret: String, + #[sea_orm(column_type = "Text")] + pub event_types: String, + pub path_filter: Option, + pub active: bool, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +#[derive(Copy, Clone, Debug, EnumIter)] +pub enum Relation { + WebhookEventTypes, +} + +impl RelationTrait for Relation { + fn def(&self) -> RelationDef { + match self { + Self::WebhookEventTypes => { + Entity::has_many(super::mega_webhook_event_type::Entity).into() + } + } + } +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::WebhookEventTypes.def() + } + + fn via() -> Option { + None + } +} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/callisto/src/mega_webhook_delivery.rs b/jupiter/callisto/src/mega_webhook_delivery.rs new file mode 100644 index 000000000..973d9caaf --- /dev/null +++ b/jupiter/callisto/src/mega_webhook_delivery.rs @@ -0,0 +1,28 @@ +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +use super::sea_orm_active_enums::WebhookEventTypeEnum; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "mega_webhook_delivery")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub id: i64, + pub webhook_id: i64, + pub event_type: WebhookEventTypeEnum, + #[sea_orm(column_type = "Text")] + pub payload: String, + pub response_status: Option, + #[sea_orm(column_type = "Text", nullable)] + pub response_body: Option, + pub success: bool, + pub attempt: i32, + #[sea_orm(column_type = "Text", nullable)] + pub error_message: Option, + pub created_at: DateTime, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/callisto/src/mega_webhook_event_type.rs b/jupiter/callisto/src/mega_webhook_event_type.rs new file mode 100644 index 000000000..59fdcfa00 --- /dev/null +++ b/jupiter/callisto/src/mega_webhook_event_type.rs @@ -0,0 +1,33 @@ +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; + +use super::sea_orm_active_enums::WebhookEventTypeEnum; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "mega_webhook_event_type")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub webhook_id: i64, + #[sea_orm(primary_key, auto_increment = false)] + pub event_type: WebhookEventTypeEnum, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::mega_webhook::Entity", + from = "Column::WebhookId", + to = "super::mega_webhook::Column::Id", + on_update = "NoAction", + on_delete = "Cascade" + )] + MegaWebhook, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::MegaWebhook.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/callisto/src/mod.rs b/jupiter/callisto/src/mod.rs index 7f62aa0ad..57c655469 100644 --- a/jupiter/callisto/src/mod.rs +++ b/jupiter/callisto/src/mod.rs @@ -51,6 +51,9 @@ pub mod mega_refs; pub mod mega_resource_permission; pub mod mega_tag; pub mod mega_tree; +pub mod mega_webhook; +pub mod mega_webhook_delivery; +pub mod mega_webhook_event_type; pub mod merge_queue; pub mod notes; pub mod orion_tasks; diff --git a/jupiter/callisto/src/prelude.rs b/jupiter/callisto/src/prelude.rs index f5ffc6c24..7149aee72 100644 --- a/jupiter/callisto/src/prelude.rs +++ b/jupiter/callisto/src/prelude.rs @@ -23,7 +23,9 @@ pub use super::{ mega_conversation::Entity as MegaConversation, mega_group::Entity as MegaGroup, mega_group_member::Entity as MegaGroupMember, mega_issue::Entity as MegaIssue, mega_refs::Entity as MegaRefs, mega_resource_permission::Entity as MegaResourcePermission, - mega_tag::Entity as MegaTag, mega_tree::Entity as MegaTree, merge_queue::Entity as MergeQueue, + mega_tag::Entity as MegaTag, mega_tree::Entity as MegaTree, + mega_webhook::Entity as MegaWebhook, mega_webhook_delivery::Entity as MegaWebhookDelivery, + mega_webhook_event_type::Entity as MegaWebhookEventType, merge_queue::Entity as MergeQueue, notes::Entity as Notes, orion_tasks::Entity as OrionTasks, path_check_configs::Entity as PathCheckConfigs, reactions::Entity as Reactions, ssh_keys::Entity as SshKeys, target_build_status::Entity as TargetBuildStatus, diff --git a/jupiter/callisto/src/sea_orm_active_enums.rs b/jupiter/callisto/src/sea_orm_active_enums.rs index 994b044b4..d40a0664b 100644 --- a/jupiter/callisto/src/sea_orm_active_enums.rs +++ b/jupiter/callisto/src/sea_orm_active_enums.rs @@ -289,3 +289,23 @@ pub enum ThreadStatusEnum { #[sea_orm(string_value = "open")] Open, } +#[derive( + Debug, Clone, Copy, PartialEq, Eq, Hash, EnumIter, DeriveActiveEnum, Serialize, Deserialize, +)] +#[sea_orm(rs_type = "String", db_type = "String(StringLen::None)")] +pub enum WebhookEventTypeEnum { + #[sea_orm(string_value = "cl.created")] + ClCreated, + #[sea_orm(string_value = "cl.updated")] + ClUpdated, + #[sea_orm(string_value = "cl.merged")] + ClMerged, + #[sea_orm(string_value = "cl.closed")] + ClClosed, + #[sea_orm(string_value = "cl.reopened")] + ClReopened, + #[sea_orm(string_value = "cl.comment.created")] + ClCommentCreated, + #[sea_orm(string_value = "*")] + All, +} diff --git a/jupiter/src/migration/m20260308_191753_create_webhook.rs b/jupiter/src/migration/m20260308_191753_create_webhook.rs new file mode 100644 index 000000000..9ca5f0816 --- /dev/null +++ b/jupiter/src/migration/m20260308_191753_create_webhook.rs @@ -0,0 +1,106 @@ +use sea_orm_migration::{prelude::*, schema::*}; + +use crate::migration::pk_bigint; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .create_table( + Table::create() + .table(MegaWebhook::Table) + .if_not_exists() + .col(pk_bigint(MegaWebhook::Id)) + .col(string(MegaWebhook::TargetUrl)) + .col(string(MegaWebhook::Secret)) + .col(text(MegaWebhook::EventTypes)) + .col(string_null(MegaWebhook::PathFilter)) + .col(boolean(MegaWebhook::Active)) + .col(date_time(MegaWebhook::CreatedAt)) + .col(date_time(MegaWebhook::UpdatedAt)) + .to_owned(), + ) + .await?; + + manager + .create_table( + Table::create() + .table(MegaWebhookDelivery::Table) + .if_not_exists() + .col(pk_bigint(MegaWebhookDelivery::Id)) + .col(big_integer(MegaWebhookDelivery::WebhookId)) + .col(string(MegaWebhookDelivery::EventType)) + .col(text(MegaWebhookDelivery::Payload)) + .col(integer_null(MegaWebhookDelivery::ResponseStatus)) + .col(text_null(MegaWebhookDelivery::ResponseBody)) + .col(boolean(MegaWebhookDelivery::Success)) + .col(integer(MegaWebhookDelivery::Attempt)) + .col(text_null(MegaWebhookDelivery::ErrorMessage)) + .col(date_time(MegaWebhookDelivery::CreatedAt)) + .foreign_key( + ForeignKey::create() + .name("fk_webhook_delivery_webhook_id") + .from(MegaWebhookDelivery::Table, MegaWebhookDelivery::WebhookId) + .to(MegaWebhook::Table, MegaWebhook::Id) + .on_delete(ForeignKeyAction::Cascade), + ) + .to_owned(), + ) + .await?; + + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_webhook_delivery_webhook_id") + .table(MegaWebhookDelivery::Table) + .col(MegaWebhookDelivery::WebhookId) + .to_owned(), + ) + .await?; + + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(MegaWebhookDelivery::Table).to_owned()) + .await?; + manager + .drop_table(Table::drop().table(MegaWebhook::Table).to_owned()) + .await?; + + Ok(()) + } +} + +#[derive(DeriveIden)] +enum MegaWebhook { + Table, + Id, + TargetUrl, + Secret, + EventTypes, + PathFilter, + Active, + CreatedAt, + UpdatedAt, +} + +#[derive(DeriveIden)] +enum MegaWebhookDelivery { + Table, + Id, + WebhookId, + EventType, + Payload, + ResponseStatus, + ResponseBody, + Success, + Attempt, + ErrorMessage, + CreatedAt, +} diff --git a/jupiter/src/migration/m20260308_220000_add_base_branch_to_mega_cl.rs b/jupiter/src/migration/m20260308_220000_add_base_branch_to_mega_cl.rs new file mode 100644 index 000000000..55950f5fa --- /dev/null +++ b/jupiter/src/migration/m20260308_220000_add_base_branch_to_mega_cl.rs @@ -0,0 +1,33 @@ +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .alter_table( + Table::alter() + .table(MegaCl::Table) + .add_column_if_not_exists( + ColumnDef::new(MegaCl::BaseBranch) + .string() + .not_null() + .default("main"), + ) + .to_owned(), + ) + .await + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + Ok(()) + } +} + +#[derive(DeriveIden)] +enum MegaCl { + Table, + BaseBranch, +} diff --git a/jupiter/src/migration/m20260308_230000_normalize_webhook_event_types.rs b/jupiter/src/migration/m20260308_230000_normalize_webhook_event_types.rs new file mode 100644 index 000000000..40a8a26a8 --- /dev/null +++ b/jupiter/src/migration/m20260308_230000_normalize_webhook_event_types.rs @@ -0,0 +1,104 @@ +use std::collections::HashSet; + +use sea_orm::{FromQueryResult, Statement}; +use sea_orm_migration::{ + prelude::*, + schema::{big_integer, string}, +}; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[derive(Debug, FromQueryResult)] +struct WebhookRow { + id: i64, + event_types: String, +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .create_table( + Table::create() + .table(MegaWebhookEventType::Table) + .if_not_exists() + .col(big_integer(MegaWebhookEventType::WebhookId)) + .col(string(MegaWebhookEventType::EventType)) + .primary_key( + Index::create() + .col(MegaWebhookEventType::WebhookId) + .col(MegaWebhookEventType::EventType), + ) + .foreign_key( + ForeignKey::create() + .name("fk_webhook_event_type_webhook_id") + .from(MegaWebhookEventType::Table, MegaWebhookEventType::WebhookId) + .to(MegaWebhook::Table, MegaWebhook::Id) + .on_delete(ForeignKeyAction::Cascade), + ) + .to_owned(), + ) + .await?; + + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_webhook_event_type_event_type") + .table(MegaWebhookEventType::Table) + .col(MegaWebhookEventType::EventType) + .to_owned(), + ) + .await?; + + let conn = manager.get_connection(); + let webhooks: Vec = WebhookRow::find_by_statement(Statement::from_string( + manager.get_database_backend(), + "SELECT id, event_types FROM mega_webhook".to_string(), + )) + .all(conn) + .await?; + + for webhook in webhooks { + let parsed: Vec = + serde_json::from_str(&webhook.event_types).unwrap_or_default(); + let mut dedup = HashSet::new(); + for event_type in parsed { + if event_type.is_empty() || !dedup.insert(event_type.clone()) { + continue; + } + let insert = Query::insert() + .into_table(MegaWebhookEventType::Table) + .columns([ + MegaWebhookEventType::WebhookId, + MegaWebhookEventType::EventType, + ]) + .values_panic([webhook.id.into(), event_type.into()]) + .to_owned(); + manager.exec_stmt(insert).await?; + } + } + + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(MegaWebhookEventType::Table).to_owned()) + .await + } +} + +#[derive(DeriveIden)] +enum MegaWebhook { + Table, + Id, +} + +#[derive(DeriveIden)] +enum MegaWebhookEventType { + Table, + WebhookId, + EventType, +} diff --git a/jupiter/src/migration/mod.rs b/jupiter/src/migration/mod.rs index 0edd5c233..a41238311 100644 --- a/jupiter/src/migration/mod.rs +++ b/jupiter/src/migration/mod.rs @@ -91,6 +91,9 @@ mod m20260228_100254_change_build_target_and_add_index_for_build_event_start_at; mod m20260302_082846_add_cla_sign_status; mod m20260304_013434_seed_cla_sign_check_config; mod m20260306_121829_create_bots_related_table; +mod m20260308_191753_create_webhook; +mod m20260308_220000_add_base_branch_to_mega_cl; +mod m20260308_230000_normalize_webhook_event_types; /// Creates a primary key column definition with big integer type. /// @@ -172,6 +175,9 @@ impl MigratorTrait for Migrator { Box::new(m20260302_082846_add_cla_sign_status::Migration), Box::new(m20260304_013434_seed_cla_sign_check_config::Migration), Box::new(m20260306_121829_create_bots_related_table::Migration), + Box::new(m20260308_191753_create_webhook::Migration), + Box::new(m20260308_220000_add_base_branch_to_mega_cl::Migration), + Box::new(m20260308_230000_normalize_webhook_event_types::Migration), ] } } diff --git a/jupiter/src/service/buck_service.rs b/jupiter/src/service/buck_service.rs index a8201fbe4..c9d9dcfef 100644 --- a/jupiter/src/service/buck_service.rs +++ b/jupiter/src/service/buck_service.rs @@ -334,6 +334,7 @@ impl BuckService { &self, username: &str, path: &str, + base_branch: &str, from_hash: String, ) -> Result { // Generate session_id @@ -344,7 +345,14 @@ impl BuckService { // Pre-create Draft CL self.cl_service - .create_draft_cl(path, &session_id, "Pending upload", &from_hash, username) + .create_draft_cl( + path, + &session_id, + "Pending upload", + base_branch, + &from_hash, + username, + ) .await?; // Create session record diff --git a/jupiter/src/service/cl_service.rs b/jupiter/src/service/cl_service.rs index f15f06f0f..f8fe696e7 100644 --- a/jupiter/src/service/cl_service.rs +++ b/jupiter/src/service/cl_service.rs @@ -83,11 +83,12 @@ impl CLService { path: &str, link: &str, title: &str, + base_branch: &str, from_hash: &str, username: &str, ) -> Result { self.cl_storage - .new_cl_draft(path, link, title, from_hash, username) + .new_cl_draft(path, link, title, base_branch, from_hash, username) .await } } diff --git a/jupiter/src/service/mod.rs b/jupiter/src/service/mod.rs index 09eef9224..e149f256d 100644 --- a/jupiter/src/service/mod.rs +++ b/jupiter/src/service/mod.rs @@ -9,3 +9,4 @@ pub mod lfs_service; pub mod merge_queue_service; pub mod mono_service; pub mod reviewer_service; +pub mod webhook_service; diff --git a/jupiter/src/service/webhook_service.rs b/jupiter/src/service/webhook_service.rs new file mode 100644 index 000000000..ec507672b --- /dev/null +++ b/jupiter/src/service/webhook_service.rs @@ -0,0 +1,639 @@ +use std::{ + net::{IpAddr, Ipv4Addr, Ipv6Addr}, + time::Duration, +}; + +use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD}; +use callisto::{ + mega_cl, + sea_orm_active_enums::{MergeStatusEnum, WebhookEventTypeEnum}, +}; +use chrono::Utc; +use hmac::{Hmac, Mac}; +use idgenerator::IdInstance; +use reqwest::redirect::Policy; +use ring::{ + aead::{self, Aad, LessSafeKey, Nonce, UnboundKey}, + rand::{SecureRandom, SystemRandom}, +}; +use sea_orm::ActiveEnum; +use serde::Serialize; +use sha2::Sha256; +use tokio::net::lookup_host; +use url::{Host, Url}; + +use crate::storage::webhook_storage::WebhookStorage; + +const WEBHOOK_SECRET_ENC_PREFIX: &str = "enc:v1:"; +const WEBHOOK_SECRET_ENC_KEY_ENV: &str = "MEGA_WEBHOOK_SECRET_ENC_KEY"; +const WEBHOOK_SECRET_KEY_LEN: usize = 32; +const WEBHOOK_SECRET_NONCE_LEN: usize = 12; +const WEBHOOK_DELIVERY_MAX_ATTEMPTS: i32 = 3; +const WEBHOOK_DELIVERY_BACKOFF_BASE_SECS: u64 = 2; + +pub type WebhookEvent = WebhookEventTypeEnum; + +#[derive(Debug, Clone, Serialize)] +pub struct WebhookPayload { + pub mega_version: String, + pub event: String, + pub timestamp: i64, + pub cl: ClPayload, + pub repository: RepositoryPayload, +} + +#[derive(Debug, Clone, Serialize)] +pub struct RepositoryPayload { + pub path: String, + pub name: String, +} + +#[derive(Debug, Clone, Serialize)] +pub struct AuthorPayload { + pub name: String, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ClPayload { + pub id: i64, + pub link: String, + pub title: String, + pub author: AuthorPayload, + pub status: String, + pub path: String, + pub base_branch: String, + pub head_commit: String, +} + +impl From<&mega_cl::Model> for ClPayload { + fn from(model: &mega_cl::Model) -> Self { + Self { + id: model.id, + link: model.link.clone(), + title: model.title.clone(), + author: AuthorPayload { + name: model.username.clone(), + }, + status: merge_status_to_str(&model.status).to_string(), + path: model.path.clone(), + base_branch: model.base_branch.clone(), + head_commit: model.to_hash.clone(), + } + } +} + +#[derive(Clone)] +pub struct WebhookService { + storage: WebhookStorage, + client: reqwest::Client, +} + +impl WebhookService { + pub fn new(storage: WebhookStorage) -> Result { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .redirect(Policy::none()) + .build() + .map_err(|e| { + common::errors::MegaError::Other(format!("failed to build reqwest client: {e}")) + })?; + Ok(Self { storage, client }) + } + + pub fn mock(storage: WebhookStorage) -> Self { + Self { + storage, + client: reqwest::Client::new(), + } + } + + pub fn dispatch(&self, event_type: WebhookEvent, cl_model: &mega_cl::Model) { + let svc = self.clone(); + let cl_payload = ClPayload::from(cl_model); + let path = cl_model.path.clone(); + + tokio::spawn(async move { + if let Err(e) = svc.dispatch_inner(event_type, cl_payload, &path).await { + tracing::error!("webhook dispatch error: {e}"); + } + }); + } + + async fn dispatch_inner( + &self, + event_type: WebhookEvent, + cl_payload: ClPayload, + path: &str, + ) -> Result<(), common::errors::MegaError> { + let event_type_str = event_type.to_value(); + let webhooks = self + .storage + .find_matching_webhooks(event_type, path) + .await?; + + for webhook in webhooks { + let repo_name = extract_repo_name(path).to_string(); + let payload = WebhookPayload { + mega_version: env!("CARGO_PKG_VERSION").to_string(), + event: event_type_str.clone(), + timestamp: Utc::now().timestamp(), + cl: cl_payload.clone(), + repository: RepositoryPayload { + path: path.to_string(), + name: repo_name, + }, + }; + + let payload_json = serde_json::to_string(&payload)?; + + let mut last_failure_message = None; + let mut delivered_successfully = false; + for attempt in 1..=WEBHOOK_DELIVERY_MAX_ATTEMPTS { + let should_retry = match self + .deliver( + &webhook.target_url, + &webhook.secret, + &event_type_str, + &payload_json, + ) + .await + { + Ok((status, body)) => { + let success = (200..300).contains(&status); + let error_message = if success { + None + } else { + Some(format!("webhook endpoint returned HTTP {status}")) + }; + let delivery = callisto::mega_webhook_delivery::Model { + id: IdInstance::next_id(), + webhook_id: webhook.id, + event_type, + payload: payload_json.clone(), + response_status: Some(status as i32), + response_body: Some(body), + success, + attempt, + error_message: error_message.clone(), + created_at: Utc::now().naive_utc(), + }; + if let Err(e) = self.storage.save_delivery(delivery).await { + tracing::warn!("failed to save webhook delivery record: {e}"); + } + if success { + delivered_successfully = true; + last_failure_message = None; + break; + } + last_failure_message = error_message; + true + } + Err(e) => { + let error_message = e.to_string(); + let delivery = callisto::mega_webhook_delivery::Model { + id: IdInstance::next_id(), + webhook_id: webhook.id, + event_type, + payload: payload_json.clone(), + response_status: None, + response_body: None, + success: false, + attempt, + error_message: Some(error_message.clone()), + created_at: Utc::now().naive_utc(), + }; + if let Err(save_err) = self.storage.save_delivery(delivery).await { + tracing::warn!("failed to save webhook delivery record: {save_err}"); + } + last_failure_message = Some(error_message); + true + } + }; + + if should_retry && attempt < WEBHOOK_DELIVERY_MAX_ATTEMPTS { + let backoff = + Duration::from_secs(WEBHOOK_DELIVERY_BACKOFF_BASE_SECS.pow(attempt as u32)); + tokio::time::sleep(backoff).await; + } + } + + if !delivered_successfully && let Some(error_message) = last_failure_message { + tracing::warn!( + "webhook delivery failed after {} attempts for webhook_id={}: {}", + WEBHOOK_DELIVERY_MAX_ATTEMPTS, + webhook.id, + error_message + ); + } + } + + Ok(()) + } + + async fn deliver( + &self, + url: &str, + secret: &str, + event_type: &str, + payload: &str, + ) -> Result<(u16, String), common::errors::MegaError> { + validate_webhook_target_url_for_delivery(url).await?; + let signing_secret = decrypt_webhook_secret(secret)?; + let signature = compute_hmac_signature(&signing_secret, payload); + + let resp = self + .client + .post(url) + .header("Content-Type", "application/json") + .header("X-Mega-Event", event_type) + .header("X-Mega-Signature", format!("sha256={signature}")) + .body(payload.to_string()) + .send() + .await + .map_err(|e| common::errors::MegaError::Other(e.to_string()))?; + + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + Ok((status, body)) + } +} + +fn merge_status_to_str(status: &MergeStatusEnum) -> &'static str { + match status { + MergeStatusEnum::Open => "open", + MergeStatusEnum::Merged => "merged", + MergeStatusEnum::Closed => "closed", + MergeStatusEnum::Draft => "draft", + } +} + +fn extract_repo_name(path: &str) -> &str { + path.trim_matches('/') + .rsplit('/') + .find(|segment| !segment.is_empty()) + .unwrap_or("unknown") +} + +fn compute_hmac_signature(secret: &str, payload: &str) -> String { + type HmacSha256 = Hmac; + let mut mac = + HmacSha256::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size"); + mac.update(payload.as_bytes()); + hex::encode(mac.finalize().into_bytes()) +} + +pub fn validate_webhook_target_url(target_url: &str) -> Result<(), common::errors::MegaError> { + let url = parse_webhook_url(target_url)?; + validate_webhook_url_static(&url) +} + +pub fn encrypt_webhook_secret(secret: &str) -> Result { + if secret.is_empty() { + return Err(common::errors::MegaError::Other( + "webhook secret cannot be empty".to_string(), + )); + } + + let key_bytes = load_webhook_secret_key_from_env()?; + encrypt_webhook_secret_with_key(secret, &key_bytes) +} + +pub fn decrypt_webhook_secret(secret: &str) -> Result { + if !secret.starts_with(WEBHOOK_SECRET_ENC_PREFIX) { + // Backward compatibility for existing plaintext secrets. + return Ok(secret.to_string()); + } + + let key_bytes = load_webhook_secret_key_from_env()?; + decrypt_webhook_secret_with_key(secret, &key_bytes) +} + +async fn validate_webhook_target_url_for_delivery( + target_url: &str, +) -> Result<(), common::errors::MegaError> { + let url = parse_webhook_url(target_url)?; + validate_webhook_url_static(&url)?; + validate_resolved_ips(&url).await +} + +fn parse_webhook_url(target_url: &str) -> Result { + Url::parse(target_url) + .map_err(|e| common::errors::MegaError::Other(format!("invalid webhook target_url: {e}"))) +} + +fn load_webhook_secret_key_from_env() +-> Result<[u8; WEBHOOK_SECRET_KEY_LEN], common::errors::MegaError> { + let encoded_key = std::env::var(WEBHOOK_SECRET_ENC_KEY_ENV).map_err(|_| { + common::errors::MegaError::Other(format!( + "{WEBHOOK_SECRET_ENC_KEY_ENV} is not set for webhook secret encryption" + )) + })?; + + let decoded_key = BASE64_STANDARD.decode(encoded_key).map_err(|_| { + common::errors::MegaError::Other(format!( + "{WEBHOOK_SECRET_ENC_KEY_ENV} must be a valid base64-encoded 32-byte key" + )) + })?; + + let key: [u8; WEBHOOK_SECRET_KEY_LEN] = decoded_key.try_into().map_err(|_| { + common::errors::MegaError::Other(format!( + "{WEBHOOK_SECRET_ENC_KEY_ENV} must decode to exactly {} bytes", + WEBHOOK_SECRET_KEY_LEN + )) + })?; + Ok(key) +} + +fn encrypt_webhook_secret_with_key( + secret: &str, + key_bytes: &[u8; WEBHOOK_SECRET_KEY_LEN], +) -> Result { + let unbound_key = UnboundKey::new(&aead::AES_256_GCM, key_bytes).map_err(|_| { + common::errors::MegaError::Other("invalid webhook encryption key".to_string()) + })?; + let key = LessSafeKey::new(unbound_key); + + let mut nonce_bytes = [0u8; WEBHOOK_SECRET_NONCE_LEN]; + SystemRandom::new().fill(&mut nonce_bytes).map_err(|_| { + common::errors::MegaError::Other("failed to generate webhook secret nonce".to_string()) + })?; + + let mut in_out = secret.as_bytes().to_vec(); + key.seal_in_place_append_tag( + Nonce::assume_unique_for_key(nonce_bytes), + Aad::empty(), + &mut in_out, + ) + .map_err(|_| { + common::errors::MegaError::Other("failed to encrypt webhook secret".to_string()) + })?; + + let mut payload = Vec::with_capacity(WEBHOOK_SECRET_NONCE_LEN + in_out.len()); + payload.extend_from_slice(&nonce_bytes); + payload.extend_from_slice(&in_out); + + Ok(format!( + "{WEBHOOK_SECRET_ENC_PREFIX}{}", + BASE64_STANDARD.encode(payload) + )) +} + +fn decrypt_webhook_secret_with_key( + encrypted_secret: &str, + key_bytes: &[u8; WEBHOOK_SECRET_KEY_LEN], +) -> Result { + let encoded_payload = encrypted_secret + .strip_prefix(WEBHOOK_SECRET_ENC_PREFIX) + .ok_or_else(|| { + common::errors::MegaError::Other("invalid encrypted webhook secret prefix".to_string()) + })?; + + let payload = BASE64_STANDARD.decode(encoded_payload).map_err(|_| { + common::errors::MegaError::Other("invalid encrypted webhook secret format".to_string()) + })?; + + if payload.len() <= WEBHOOK_SECRET_NONCE_LEN { + return Err(common::errors::MegaError::Other( + "invalid encrypted webhook secret payload".to_string(), + )); + } + + let mut nonce_bytes = [0u8; WEBHOOK_SECRET_NONCE_LEN]; + nonce_bytes.copy_from_slice(&payload[..WEBHOOK_SECRET_NONCE_LEN]); + let mut cipher_and_tag = payload[WEBHOOK_SECRET_NONCE_LEN..].to_vec(); + + let unbound_key = UnboundKey::new(&aead::AES_256_GCM, key_bytes).map_err(|_| { + common::errors::MegaError::Other("invalid webhook encryption key".to_string()) + })?; + let key = LessSafeKey::new(unbound_key); + + let plaintext = key + .open_in_place( + Nonce::assume_unique_for_key(nonce_bytes), + Aad::empty(), + &mut cipher_and_tag, + ) + .map_err(|_| { + common::errors::MegaError::Other("failed to decrypt webhook secret".to_string()) + })?; + + String::from_utf8(plaintext.to_vec()).map_err(|_| { + common::errors::MegaError::Other("decrypted webhook secret is not valid UTF-8".to_string()) + }) +} + +fn validate_webhook_url_static(url: &Url) -> Result<(), common::errors::MegaError> { + if url.scheme() != "https" { + return Err(common::errors::MegaError::Other( + "webhook target_url must use https".to_string(), + )); + } + + let host = url.host().ok_or_else(|| { + common::errors::MegaError::Other("webhook target_url must include a host".to_string()) + })?; + + match host { + Host::Domain(domain) => { + let domain_lc = domain.to_ascii_lowercase(); + if domain_lc == "localhost" + || domain_lc.ends_with(".localhost") + || domain_lc.ends_with(".local") + { + return Err(common::errors::MegaError::Other( + "webhook target_url host is not allowed".to_string(), + )); + } + } + Host::Ipv4(ip) => { + if is_forbidden_ip(IpAddr::V4(ip)) { + return Err(common::errors::MegaError::Other( + "webhook target_url IP is not allowed".to_string(), + )); + } + } + Host::Ipv6(ip) => { + if is_forbidden_ip(IpAddr::V6(ip)) { + return Err(common::errors::MegaError::Other( + "webhook target_url IP is not allowed".to_string(), + )); + } + } + } + + Ok(()) +} + +async fn validate_resolved_ips(url: &Url) -> Result<(), common::errors::MegaError> { + if matches!(url.host(), Some(Host::Ipv4(_) | Host::Ipv6(_))) { + return Ok(()); + } + + let host = url.host_str().ok_or_else(|| { + common::errors::MegaError::Other("webhook target_url must include a host".to_string()) + })?; + let port = url.port_or_known_default().ok_or_else(|| { + common::errors::MegaError::Other("webhook target_url must include a valid port".to_string()) + })?; + + let mut resolved_any = false; + let addrs = lookup_host((host, port)).await.map_err(|e| { + common::errors::MegaError::Other(format!("failed to resolve webhook host: {e}")) + })?; + + for addr in addrs { + resolved_any = true; + if is_forbidden_ip(addr.ip()) { + return Err(common::errors::MegaError::Other( + "webhook target_url resolves to a restricted IP".to_string(), + )); + } + } + + if !resolved_any { + return Err(common::errors::MegaError::Other( + "webhook target_url host did not resolve to any address".to_string(), + )); + } + + Ok(()) +} + +fn is_forbidden_ip(ip: IpAddr) -> bool { + match ip { + IpAddr::V4(v4) => is_forbidden_ipv4(v4), + IpAddr::V6(v6) => is_forbidden_ipv6(v6), + } +} + +fn is_forbidden_ipv4(ip: Ipv4Addr) -> bool { + let [a, b, _, _] = ip.octets(); + ip.is_private() + || ip.is_loopback() + || ip.is_link_local() + || ip.is_multicast() + || ip.is_broadcast() + || ip.is_unspecified() + || a == 0 + || (a == 100 && (64..=127).contains(&b)) + || (a == 198 && (18..=19).contains(&b)) +} + +fn is_forbidden_ipv6(ip: Ipv6Addr) -> bool { + ip.is_loopback() + || ip.is_unspecified() + || ip.is_unicast_link_local() + || ip.is_unique_local() + || ip.is_multicast() +} + +#[cfg(test)] +mod tests { + use callisto::sea_orm_active_enums::MergeStatusEnum; + use chrono::DateTime; + + use super::*; + + #[test] + fn test_validate_webhook_target_url_accepts_public_https_domain() { + let result = validate_webhook_target_url("https://example.com/webhook"); + assert!(result.is_ok()); + } + + #[test] + fn test_validate_webhook_target_url_rejects_non_https() { + let result = validate_webhook_target_url("http://example.com/webhook"); + assert!(result.is_err()); + } + + #[test] + fn test_validate_webhook_target_url_rejects_loopback_ip() { + let result = validate_webhook_target_url("https://127.0.0.1:8443/webhook"); + assert!(result.is_err()); + } + + #[test] + fn test_validate_webhook_target_url_rejects_private_ipv4() { + let result = validate_webhook_target_url("https://192.168.1.2/webhook"); + assert!(result.is_err()); + } + + #[test] + fn test_validate_webhook_target_url_rejects_localhost_domain() { + let result = validate_webhook_target_url("https://localhost/webhook"); + assert!(result.is_err()); + } + + #[test] + fn test_validate_webhook_target_url_rejects_loopback_ipv6() { + let result = validate_webhook_target_url("https://[::1]/webhook"); + assert!(result.is_err()); + } + + #[test] + fn test_validate_webhook_target_url_rejects_link_local_ipv6() { + let result = validate_webhook_target_url("https://[fe80::1]/webhook"); + assert!(result.is_err()); + } + + #[test] + fn test_encrypt_decrypt_webhook_secret_round_trip() { + let key = [7u8; 32]; + let secret = "test-webhook-secret"; + + let encrypted = encrypt_webhook_secret_with_key(secret, &key).expect("encrypt failed"); + assert!(encrypted.starts_with("enc:v1:")); + + let decrypted = decrypt_webhook_secret_with_key(&encrypted, &key).expect("decrypt failed"); + assert_eq!(decrypted, secret); + } + + #[test] + fn test_decrypt_webhook_secret_plaintext_is_backward_compatible() { + let plaintext = "legacy-plaintext-secret"; + let decrypted = decrypt_webhook_secret(plaintext).expect("decrypt failed"); + assert_eq!(decrypted, plaintext); + } + + #[test] + fn test_extract_repo_name_normal_path() { + assert_eq!(extract_repo_name("/mono/repo"), "repo"); + } + + #[test] + fn test_extract_repo_name_trailing_slash() { + assert_eq!(extract_repo_name("/mono/repo/"), "repo"); + } + + #[test] + fn test_extract_repo_name_root_path_fallback() { + assert_eq!(extract_repo_name("/"), "unknown"); + } + + #[test] + fn test_extract_repo_name_empty_path_fallback() { + assert_eq!(extract_repo_name(""), "unknown"); + } + + #[test] + fn test_cl_payload_from_model_uses_contract_fields() { + let model = callisto::mega_cl::Model { + id: 1, + link: "CL123".to_string(), + title: "Test".to_string(), + merge_date: None, + status: MergeStatusEnum::Open, + path: "/repo/path".to_string(), + base_branch: "main".to_string(), + from_hash: "abc".to_string(), + to_hash: "def".to_string(), + created_at: DateTime::from_timestamp(0, 0).unwrap().naive_utc(), + updated_at: DateTime::from_timestamp(0, 0).unwrap().naive_utc(), + username: "alice".to_string(), + }; + + let payload = ClPayload::from(&model); + assert_eq!(payload.status, "open"); + assert_eq!(payload.path, "/repo/path"); + } +} diff --git a/jupiter/src/storage/cl_storage.rs b/jupiter/src/storage/cl_storage.rs index 4dbd3a917..810e428de 100644 --- a/jupiter/src/storage/cl_storage.rs +++ b/jupiter/src/storage/cl_storage.rs @@ -227,25 +227,29 @@ impl ClStorage { Ok(()) } + #[allow(clippy::too_many_arguments)] pub async fn new_cl( &self, path: &str, link: &str, title: &str, + base_branch: &str, from_hash: &str, to_hash: &str, username: &str, ) -> Result { - self.new_cl_model(path, link, title, from_hash, to_hash, username) + self.new_cl_model(path, link, title, base_branch, from_hash, to_hash, username) .await .map(|res| res.link) } + #[allow(clippy::too_many_arguments)] pub async fn new_cl_model( &self, path: &str, link: &str, title: &str, + base_branch: &str, from_hash: &str, to_hash: &str, username: &str, @@ -254,6 +258,7 @@ impl ClStorage { path.to_owned(), title.to_owned(), link.to_owned(), + base_branch.to_owned(), from_hash.to_owned(), to_hash.to_owned(), username.to_owned(), @@ -271,6 +276,7 @@ impl ClStorage { path: &str, link: &str, title: &str, + base_branch: &str, from_hash: &str, username: &str, ) -> Result { @@ -278,6 +284,7 @@ impl ClStorage { path.to_owned(), title.to_owned(), link.to_owned(), + base_branch.to_owned(), from_hash.to_owned(), username.to_owned(), ); diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index e08223d2f..f1f6c8d4c 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -22,6 +22,7 @@ pub mod note_storage; pub mod stg_common; pub mod user_storage; pub mod vault_storage; +pub mod webhook_storage; use std::sync::{Arc, LazyLock, Weak}; @@ -35,6 +36,7 @@ use crate::{ code_review_service::CodeReviewService, git_service::GitService, import_service::ImportService, issue_service::IssueService, lfs_service::LfsService, merge_queue_service::MergeQueueService, mono_service::MonoService, + webhook_service::WebhookService, }, storage::{ base_storage::{BaseStorage, StorageConnector}, @@ -60,6 +62,7 @@ use crate::{ note_storage::NoteStorage, user_storage::UserStorage, vault_storage::VaultStorage, + webhook_storage::WebhookStorage, }, }; @@ -86,6 +89,7 @@ pub struct AppService { pub code_review_thread_storage: CodeReviewThreadStorage, pub build_trigger_storage: BuildTriggerStorage, pub bots_storage: BotsStorage, + pub webhook_storage: WebhookStorage, } impl AppService { @@ -116,6 +120,7 @@ impl AppService { code_review_thread_storage: CodeReviewThreadStorage { base: mock.clone() }, build_trigger_storage: BuildTriggerStorage { base: mock.clone() }, bots_storage: BotsStorage { base: mock.clone() }, + webhook_storage: WebhookStorage { base: mock.clone() }, }) } } @@ -134,6 +139,7 @@ pub struct Storage { pub lfs_service: LfsService, pub config: Weak, pub code_review_service: CodeReviewService, + pub webhook_service: WebhookService, } impl Storage { @@ -177,6 +183,7 @@ impl Storage { let code_review_thread_storage = CodeReviewThreadStorage { base: base.clone() }; let build_trigger_storage = BuildTriggerStorage { base: base.clone() }; let bots_storage = BotsStorage { base: base.clone() }; + let webhook_storage = WebhookStorage { base: base.clone() }; let git_service = GitService { obj_storage: ObjectStorageFactory::build( @@ -236,6 +243,7 @@ impl Storage { code_review_thread_storage, build_trigger_storage, bots_storage, + webhook_storage: webhook_storage.clone(), }; let merge_queue_service = MergeQueueService::new(base.clone()); let buck_service = BuckService::new( @@ -248,6 +256,8 @@ impl Storage { ) .expect("failed to create BuckService"); + let webhook_service = WebhookService::new(webhook_storage.clone())?; + Ok(Storage { app_service: app_service.into(), cla_service: ClaService::new(base.clone()), @@ -261,6 +271,7 @@ impl Storage { import_service, lfs_service, code_review_service: CodeReviewService::new(base.clone()), + webhook_service, }) } @@ -417,13 +428,21 @@ impl Storage { self.app_service.build_trigger_storage.clone() } + pub fn webhook_storage(&self) -> WebhookStorage { + self.app_service.webhook_storage.clone() + } + pub fn mock() -> Self { // During test time, we don't need a AppContext, // Put config in a leaked static variable thus the weak reference will always be valid. static CONFIG: LazyLock> = LazyLock::new(|| Config::mock().into()); + let app_service = AppService::mock(); + let webhook_service = WebhookService::mock(app_service.webhook_storage.clone()); + Storage { - app_service: AppService::mock(), + app_service, + // app_service: AppService::mock(), cla_service: ClaService::mock(), issue_service: IssueService::mock(), cl_service: CLService::mock(), @@ -435,6 +454,7 @@ impl Storage { import_service: ImportService::mock(), lfs_service: LfsService::mock(), code_review_service: CodeReviewService::mock(), + webhook_service, } } } diff --git a/jupiter/src/storage/webhook_storage.rs b/jupiter/src/storage/webhook_storage.rs new file mode 100644 index 000000000..3fcdb86c5 --- /dev/null +++ b/jupiter/src/storage/webhook_storage.rs @@ -0,0 +1,396 @@ +use std::ops::Deref; + +use api_model::common::Pagination; +use callisto::{ + mega_webhook, mega_webhook_delivery, mega_webhook_event_type, + sea_orm_active_enums::WebhookEventTypeEnum, +}; +use common::errors::MegaError; +use sea_orm::{ + ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, JoinType, PaginatorTrait, + QueryFilter, QueryOrder, QuerySelect, RelationTrait, Set, +}; + +use crate::storage::base_storage::{BaseStorage, StorageConnector}; + +#[derive(Clone)] +pub struct WebhookWithEventTypes { + pub webhook: mega_webhook::Model, + pub event_types: Vec, +} + +#[derive(Clone)] +pub struct WebhookStorage { + pub base: BaseStorage, +} + +impl Deref for WebhookStorage { + type Target = BaseStorage; + fn deref(&self) -> &Self::Target { + &self.base + } +} + +impl WebhookStorage { + pub async fn create_webhook( + &self, + mut model: mega_webhook::Model, + event_types: Vec, + ) -> Result { + let normalized_event_types = normalize_event_types(event_types); + model.event_types = serde_json::to_string(&normalized_event_types)?; + let res = model + .into_active_model() + .insert(self.get_connection()) + .await?; + self.replace_event_types(res.id, &normalized_event_types) + .await?; + + Ok(WebhookWithEventTypes { + webhook: res, + event_types: normalized_event_types, + }) + } + + pub async fn list_webhooks( + &self, + page: Pagination, + ) -> Result<(Vec, u64), MegaError> { + let paginator = mega_webhook::Entity::find() + .order_by_desc(mega_webhook::Column::CreatedAt) + .paginate(self.get_connection(), page.per_page); + let total = paginator.num_items().await?; + let webhooks = paginator.fetch_page(page.page.saturating_sub(1)).await?; + let event_types_map = self.event_types_map(&webhooks).await?; + let items = webhooks + .into_iter() + .map(|webhook| { + let event_types = event_types_map + .get(&webhook.id) + .cloned() + .unwrap_or_default(); + WebhookWithEventTypes { + webhook, + event_types, + } + }) + .collect(); + Ok((items, total)) + } + + pub async fn get_webhook(&self, id: i64) -> Result, MegaError> { + let model = mega_webhook::Entity::find_by_id(id) + .one(self.get_connection()) + .await?; + Ok(model) + } + + pub async fn delete_webhook(&self, id: i64) -> Result<(), MegaError> { + let delete_result = mega_webhook::Entity::delete_by_id(id) + .exec(self.get_connection()) + .await?; + + if delete_result.rows_affected == 0 { + return Err(MegaError::NotFound(format!( + "Webhook with id `{id}` not found" + ))); + } + + Ok(()) + } + + pub async fn find_matching_webhooks( + &self, + event_type: WebhookEventTypeEnum, + path: &str, + ) -> Result, MegaError> { + let candidates = mega_webhook::Entity::find() + .join( + JoinType::InnerJoin, + mega_webhook::Relation::WebhookEventTypes.def(), + ) + .filter(mega_webhook::Column::Active.eq(true)) + .filter( + mega_webhook_event_type::Column::EventType + .is_in([event_type, WebhookEventTypeEnum::All]), + ) + .distinct() + .all(self.get_connection()) + .await?; + + let matching = candidates + .into_iter() + .filter(|w| { + if let Some(ref filter) = w.path_filter { + path.starts_with(filter.as_str()) + } else { + true + } + }) + .collect(); + + Ok(matching) + } + + pub async fn save_delivery( + &self, + model: mega_webhook_delivery::Model, + ) -> Result<(), MegaError> { + model + .into_active_model() + .insert(self.get_connection()) + .await?; + Ok(()) + } + + async fn replace_event_types( + &self, + webhook_id: i64, + event_types: &[WebhookEventTypeEnum], + ) -> Result<(), MegaError> { + mega_webhook_event_type::Entity::delete_many() + .filter(mega_webhook_event_type::Column::WebhookId.eq(webhook_id)) + .exec(self.get_connection()) + .await?; + + for event_type in event_types { + mega_webhook_event_type::ActiveModel { + webhook_id: Set(webhook_id), + event_type: Set(*event_type), + } + .insert(self.get_connection()) + .await?; + } + Ok(()) + } + + async fn event_types_map( + &self, + webhooks: &[mega_webhook::Model], + ) -> Result>, MegaError> { + let mut map = std::collections::HashMap::new(); + if webhooks.is_empty() { + return Ok(map); + } + + let ids: Vec = webhooks.iter().map(|w| w.id).collect(); + let rows = mega_webhook_event_type::Entity::find() + .filter(mega_webhook_event_type::Column::WebhookId.is_in(ids)) + .all(self.get_connection()) + .await?; + + for row in rows { + map.entry(row.webhook_id) + .or_insert_with(Vec::new) + .push(row.event_type); + } + + Ok(map) + } +} + +fn normalize_event_types(event_types: Vec) -> Vec { + let mut dedup = std::collections::HashSet::new(); + let mut normalized = Vec::new(); + for event in event_types { + if dedup.insert(event) { + normalized.push(event); + } + } + normalized +} + +#[cfg(test)] +mod tests { + use api_model::common::Pagination; + use chrono::Utc; + use idgenerator::IdInstance; + use tempfile::TempDir; + + use super::*; + use crate::{ + migration::apply_migrations, + tests::{test_db_connection, test_storage}, + }; + + #[test] + fn test_normalize_event_types_dedupes() { + let normalized = normalize_event_types(vec![ + WebhookEventTypeEnum::ClCreated, + WebhookEventTypeEnum::ClCreated, + WebhookEventTypeEnum::All, + WebhookEventTypeEnum::All, + ]); + assert_eq!( + normalized, + vec![WebhookEventTypeEnum::ClCreated, WebhookEventTypeEnum::All] + ); + } + + #[tokio::test] + async fn test_find_matching_webhooks_uses_normalized_event_types() { + let temp_dir = TempDir::new().expect("failed to create temporary directory"); + let conn = test_db_connection(temp_dir.path()).await; + apply_migrations(&conn, true).await.unwrap(); + let storage = WebhookStorage { + base: BaseStorage::new(std::sync::Arc::new(conn)), + }; + + let now = Utc::now().naive_utc(); + let created = storage + .create_webhook( + mega_webhook::Model { + id: IdInstance::next_id(), + target_url: "https://example.com/a".to_string(), + secret: "s1".to_string(), + event_types: "[]".to_string(), + path_filter: Some("/repo".to_string()), + active: true, + created_at: now, + updated_at: now, + }, + vec![ + WebhookEventTypeEnum::ClCreated, + WebhookEventTypeEnum::All, + WebhookEventTypeEnum::ClCreated, + ], + ) + .await + .unwrap(); + + storage + .create_webhook( + mega_webhook::Model { + id: IdInstance::next_id(), + target_url: "https://example.com/b".to_string(), + secret: "s2".to_string(), + event_types: "[]".to_string(), + path_filter: Some("/repo".to_string()), + active: true, + created_at: now, + updated_at: now, + }, + vec![WebhookEventTypeEnum::ClUpdated], + ) + .await + .unwrap(); + + let (items, total) = storage + .list_webhooks(Pagination { + page: 1, + per_page: 20, + }) + .await + .unwrap(); + assert_eq!(total, 2); + let created_item = items + .into_iter() + .find(|item| item.webhook.id == created.webhook.id) + .unwrap(); + let mut event_types = created_item.event_types; + event_types.sort_by_key(|e| format!("{e:?}")); + assert_eq!( + event_types, + vec![WebhookEventTypeEnum::All, WebhookEventTypeEnum::ClCreated] + ); + + let created_matches = storage + .find_matching_webhooks(WebhookEventTypeEnum::ClCreated, "/repo/app") + .await + .unwrap(); + assert_eq!(created_matches.len(), 1); + assert_eq!(created_matches[0].id, created.webhook.id); + + let updated_matches = storage + .find_matching_webhooks(WebhookEventTypeEnum::ClUpdated, "/repo/app") + .await + .unwrap(); + assert_eq!(updated_matches.len(), 2); + + let outside_path_matches = storage + .find_matching_webhooks(WebhookEventTypeEnum::ClUpdated, "/outside/path") + .await + .unwrap(); + assert!(outside_path_matches.is_empty()); + } + + #[tokio::test] + async fn test_create_list_delete_webhook_lifecycle() { + let temp_dir = TempDir::new().expect("failed to create temporary directory"); + let storage = test_storage(temp_dir.path()).await; + let webhook_storage = storage.webhook_storage(); + + let now = Utc::now().naive_utc(); + let webhook_a = webhook_storage + .create_webhook( + mega_webhook::Model { + id: IdInstance::next_id(), + target_url: "https://example.com/a".to_string(), + secret: "s1".to_string(), + event_types: "[]".to_string(), + path_filter: Some("/repo".to_string()), + active: true, + created_at: now, + updated_at: now, + }, + vec![WebhookEventTypeEnum::ClCreated], + ) + .await + .unwrap(); + let webhook_b = webhook_storage + .create_webhook( + mega_webhook::Model { + id: IdInstance::next_id(), + target_url: "https://example.com/b".to_string(), + secret: "s2".to_string(), + event_types: "[]".to_string(), + path_filter: Some("/repo".to_string()), + active: true, + created_at: now, + updated_at: now, + }, + vec![WebhookEventTypeEnum::ClUpdated], + ) + .await + .unwrap(); + + let (before_delete, total_before_delete) = webhook_storage + .list_webhooks(Pagination { + page: 1, + per_page: 20, + }) + .await + .unwrap(); + assert_eq!(total_before_delete, 2); + assert_eq!(before_delete.len(), 2); + + webhook_storage + .delete_webhook(webhook_a.webhook.id) + .await + .unwrap(); + assert!( + webhook_storage + .get_webhook(webhook_a.webhook.id) + .await + .unwrap() + .is_none() + ); + + let (after_delete, total_after_delete) = webhook_storage + .list_webhooks(Pagination { + page: 1, + per_page: 20, + }) + .await + .unwrap(); + assert_eq!(total_after_delete, 1); + assert_eq!(after_delete.len(), 1); + assert_eq!(after_delete[0].webhook.id, webhook_b.webhook.id); + + let err = webhook_storage + .delete_webhook(webhook_a.webhook.id) + .await + .unwrap_err(); + assert!(matches!(err, MegaError::NotFound(_))); + } +} diff --git a/jupiter/src/tests.rs b/jupiter/src/tests.rs index cdbf75c0b..4fbb09d97 100644 --- a/jupiter/src/tests.rs +++ b/jupiter/src/tests.rs @@ -14,6 +14,7 @@ use crate::{ code_review_service::CodeReviewService, git_service::GitService, import_service::ImportService, issue_service::IssueService, lfs_service::LfsService, merge_queue_service::MergeQueueService, mono_service::MonoService, + webhook_service::WebhookService, }, storage::{ AppService, Storage, @@ -39,6 +40,7 @@ use crate::{ note_storage::NoteStorage, user_storage::UserStorage, vault_storage::VaultStorage, + webhook_storage::WebhookStorage, }, }; @@ -86,10 +88,13 @@ pub async fn test_storage(temp_dir: impl AsRef) -> Storage { code_review_thread_storage: CodeReviewThreadStorage { base: base.clone() }, build_trigger_storage: BuildTriggerStorage { base: base.clone() }, bots_storage: BotsStorage { base: base.clone() }, + webhook_storage: WebhookStorage { base: base.clone() }, }; apply_migrations(&connection, true).await.unwrap(); + let webhook_service = WebhookService::mock(svc.webhook_storage.clone()); + Storage { app_service: Arc::new(svc), cla_service: ClaService::new(base.clone()), @@ -103,5 +108,6 @@ pub async fn test_storage(temp_dir: impl AsRef) -> Storage { import_service: ImportService::mock(), lfs_service: LfsService::mock(), code_review_service: CodeReviewService::mock(), + webhook_service, } } diff --git a/mono/Cargo.toml b/mono/Cargo.toml index 6a4b7abde..c8e2678b4 100644 --- a/mono/Cargo.toml +++ b/mono/Cargo.toml @@ -72,6 +72,7 @@ utoipa-axum = { workspace = true } utoipa-swagger-ui = { workspace = true, features = ["axum"] } once_cell = { workspace = true } sea-orm = { workspace = true } +idgenerator = { workspace = true } [target.'cfg(not(windows))'.dependencies] diff --git a/mono/src/api/api_router.rs b/mono/src/api/api_router.rs index 5addfbd6a..d7dc3b3ac 100644 --- a/mono/src/api/api_router.rs +++ b/mono/src/api/api_router.rs @@ -19,7 +19,7 @@ use crate::{ admin_router, buck_router, build_trigger_router, cl_router, code_review_router, commit_router, conv_router, dynamic_sidebar_router, gpg_router, group_router, issue_router, label_router, merge_queue_router, permission_router, preview_router, - repo_router, reviewer_router, tag_router, user_router, + repo_router, reviewer_router, tag_router, user_router, webhook_router, }, }, server::http_server::SYSTEM_COMMON, @@ -50,6 +50,7 @@ pub fn routers() -> OpenApiRouter { .merge(permission_router::routers()) .merge(code_review_router::routers()) .merge(build_trigger_router::routers()) + .merge(webhook_router::routers()) } /// Health Check diff --git a/mono/src/api/mod.rs b/mono/src/api/mod.rs index 25ca1a526..6e4d816d5 100644 --- a/mono/src/api/mod.rs +++ b/mono/src/api/mod.rs @@ -14,10 +14,14 @@ use ceres::{ protocol::repo::Repo, }; use common::errors::ProtocolError; -use jupiter::storage::{ - Storage, cl_storage::ClStorage, conversation_storage::ConversationStorage, - dynamic_sidebar_storage::DynamicSidebarStorage, gpg_storage::GpgStorage, - issue_storage::IssueStorage, note_storage::NoteStorage, user_storage::UserStorage, +use jupiter::{ + service::webhook_service::WebhookService, + storage::{ + Storage, cl_storage::ClStorage, conversation_storage::ConversationStorage, + dynamic_sidebar_storage::DynamicSidebarStorage, gpg_storage::GpgStorage, + issue_storage::IssueStorage, note_storage::NoteStorage, user_storage::UserStorage, + webhook_storage::WebhookStorage, + }, }; use oauth2::{ Client, EndpointNotSet, EndpointSet, StandardRevocableToken, @@ -145,6 +149,14 @@ impl MonoApiServiceState { self.storage.note_storage() } + fn webhook_stg(&self) -> WebhookStorage { + self.storage.webhook_storage() + } + + fn webhook_svc(&self) -> WebhookService { + self.storage.webhook_service.clone() + } + fn dynamic_sidebar_stg(&self) -> DynamicSidebarStorage { self.storage.dynamic_sidebar_storage() } diff --git a/mono/src/api/router/cl_router.rs b/mono/src/api/router/cl_router.rs index ac3ba3d38..106f6fbb4 100644 --- a/mono/src/api/router/cl_router.rs +++ b/mono/src/api/router/cl_router.rs @@ -14,7 +14,7 @@ use ceres::model::{ label::LabelUpdatePayload, }; use common::errors::MegaError; -use jupiter::service::cl_service::CLService; +use jupiter::service::{cl_service::CLService, webhook_service::WebhookEvent}; use utoipa_axum::{router::OpenApiRouter, routes}; use crate::{ @@ -73,7 +73,7 @@ async fn reopen_cl( if model.status == MergeStatusEnum::Closed { let link = model.link.clone(); - state.cl_stg().reopen_cl(model).await?; + state.cl_stg().reopen_cl(model.clone()).await?; state .conv_stg() .add_conversation( @@ -84,6 +84,14 @@ async fn reopen_cl( ) .await .unwrap(); + let updated_model = state + .cl_stg() + .get_cl(&link) + .await? + .ok_or(MegaError::Other("Not Found".to_string()))?; + state + .webhook_svc() + .dispatch(WebhookEvent::ClReopened, &updated_model); } Ok(Json(CommonResult::success(None))) } @@ -110,7 +118,7 @@ async fn close_cl( if matches!(model.status, MergeStatusEnum::Open | MergeStatusEnum::Draft) { let link = model.link.clone(); - state.cl_stg().close_cl(model).await?; + state.cl_stg().close_cl(model.clone()).await?; state .conv_stg() .add_conversation( @@ -120,6 +128,14 @@ async fn close_cl( ConvTypeEnum::Closed, ) .await?; + let updated_model = state + .cl_stg() + .get_cl(&link) + .await? + .ok_or(MegaError::Other("Not Found".to_string()))?; + state + .webhook_svc() + .dispatch(WebhookEvent::ClClosed, &updated_model); } Ok(Json(CommonResult::success(None))) } @@ -151,7 +167,18 @@ async fn merge( } if model.status == MergeStatusEnum::Open { - state.monorepo().merge_cl(&user.username, model).await?; + state + .monorepo() + .merge_cl(&user.username, model.clone()) + .await?; + let updated_model = state + .cl_stg() + .get_cl(&link) + .await? + .ok_or(MegaError::Other("Not Found".to_string()))?; + state + .webhook_svc() + .dispatch(WebhookEvent::ClMerged, &updated_model); } Ok(Json(CommonResult::success(None))) } @@ -185,7 +212,18 @@ async fn merge_no_auth( // No authentication required - using default system user let default_username = "system"; - state.monorepo().merge_cl(default_username, model).await?; + state + .monorepo() + .merge_cl(default_username, model.clone()) + .await?; + let updated_model = state + .cl_stg() + .get_cl(&link) + .await? + .ok_or(MegaError::Other("CL Not Found".to_string()))?; + state + .webhook_svc() + .dispatch(WebhookEvent::ClMerged, &updated_model); Ok(Json(CommonResult::success(Some( "Merge completed successfully".to_string(), @@ -372,6 +410,11 @@ async fn update_branch( .monorepo() .update_branch(&user.username, &link) .await?; + if let Ok(Some(cl_model)) = state.cl_stg().get_cl(&link).await { + state + .webhook_svc() + .dispatch(WebhookEvent::ClUpdated, &cl_model); + } Ok(Json(CommonResult::success(Some(new_head)))) } @@ -455,6 +498,13 @@ async fn save_comment( conv_type, ) .await?; + + if let Ok(Some(cl_model)) = state.cl_stg().get_cl(&link).await { + state + .webhook_svc() + .dispatch(WebhookEvent::ClCommentCreated, &cl_model); + } + api_common::comment::check_comment_ref(user, state, &payload.content, &link).await } @@ -478,6 +528,11 @@ async fn edit_title( Json(payload): Json, ) -> Result>, ApiError> { state.cl_stg().edit_title(&link, &payload.content).await?; + if let Ok(Some(cl_model)) = state.cl_stg().get_cl(&link).await { + state + .webhook_svc() + .dispatch(WebhookEvent::ClUpdated, &cl_model); + } Ok(Json(CommonResult::success(None))) } @@ -552,7 +607,10 @@ async fn update_cl_status( // Only allow Draft ↔ Open transitions match (&model.status, &new_status) { (MergeStatusEnum::Draft, MergeStatusEnum::Open) => { - state.cl_stg().update_cl_status(model, new_status).await?; + state + .cl_stg() + .update_cl_status(model.clone(), new_status.clone()) + .await?; state .conv_stg() .add_conversation( @@ -562,9 +620,20 @@ async fn update_cl_status( ConvTypeEnum::Review, ) .await?; + let updated_model = state + .cl_stg() + .get_cl(&link) + .await? + .ok_or(MegaError::Other("Not Found".to_string()))?; + state + .webhook_svc() + .dispatch(WebhookEvent::ClCreated, &updated_model); } (MergeStatusEnum::Open, MergeStatusEnum::Draft) => { - state.cl_stg().update_cl_status(model, new_status).await?; + state + .cl_stg() + .update_cl_status(model.clone(), new_status.clone()) + .await?; state .conv_stg() .add_conversation( @@ -574,6 +643,14 @@ async fn update_cl_status( ConvTypeEnum::Draft, ) .await?; + let updated_model = state + .cl_stg() + .get_cl(&link) + .await? + .ok_or(MegaError::Other("Not Found".to_string()))?; + state + .webhook_svc() + .dispatch(WebhookEvent::ClUpdated, &updated_model); } _ => { return Err(ApiError::from(MegaError::Other( diff --git a/mono/src/api/router/mod.rs b/mono/src/api/router/mod.rs index db46a5a5b..37fa97611 100644 --- a/mono/src/api/router/mod.rs +++ b/mono/src/api/router/mod.rs @@ -18,3 +18,4 @@ pub mod repo_router; pub mod reviewer_router; pub mod tag_router; pub mod user_router; +pub mod webhook_router; diff --git a/mono/src/api/router/webhook_router.rs b/mono/src/api/router/webhook_router.rs new file mode 100644 index 000000000..347f76dac --- /dev/null +++ b/mono/src/api/router/webhook_router.rs @@ -0,0 +1,193 @@ +use api_model::common::{CommonPage, CommonResult, Pagination}; +use axum::{ + Json, + extract::{Path, Query, State}, +}; +use callisto::sea_orm_active_enums::WebhookEventTypeEnum; +use chrono::Utc; +use idgenerator::IdInstance; +use jupiter::{ + service::webhook_service::{encrypt_webhook_secret, validate_webhook_target_url}, + storage::webhook_storage::WebhookWithEventTypes, +}; +use sea_orm::ActiveEnum; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; +use utoipa_axum::{router::OpenApiRouter, routes}; + +use crate::{ + api::{MonoApiServiceState, error::ApiError}, + server::http_server::WEBHOOK_TAG, +}; + +pub fn routers() -> OpenApiRouter { + OpenApiRouter::new() + .routes(routes!(create_webhook)) + .routes(routes!(list_webhooks)) + .routes(routes!(delete_webhook)) +} + +#[derive(Debug, Deserialize, ToSchema)] +pub struct CreateWebhookRequest { + pub target_url: String, + pub secret: String, + /// Event types: "cl.created", "cl.updated", "cl.merged", "cl.closed", "cl.reopened", "cl.comment.created", "*" + pub event_types: Vec, + pub path_filter: Option, + pub active: Option, +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct WebhookResponse { + pub id: i64, + pub target_url: String, + pub event_types: Vec, + pub path_filter: Option, + pub active: bool, + pub created_at: String, + pub updated_at: String, +} + +#[derive(Debug, Deserialize)] +pub struct ListWebhooksQuery { + pub page: Option, + pub per_page: Option, +} + +impl From for WebhookResponse { + fn from(value: WebhookWithEventTypes) -> Self { + let m = value.webhook; + Self { + id: m.id, + target_url: m.target_url, + event_types: value + .event_types + .into_iter() + .map(|e| e.to_value()) + .collect(), + path_filter: m.path_filter, + active: m.active, + created_at: m.created_at.to_string(), + updated_at: m.updated_at.to_string(), + } + } +} + +fn parse_event_types(raw: Vec) -> Result, ApiError> { + raw.into_iter() + .map(|s| { + WebhookEventTypeEnum::try_from_value(&s) + .map_err(|_| ApiError::bad_request(anyhow::anyhow!("invalid event type: {s}"))) + }) + .collect() +} + +/// Create a webhook +#[utoipa::path( + post, + path = "/webhooks", + request_body = CreateWebhookRequest, + responses( + (status = 200, body = CommonResult, content_type = "application/json") + ), + tag = WEBHOOK_TAG +)] +async fn create_webhook( + state: State, + Json(payload): Json, +) -> Result>, ApiError> { + validate_webhook_target_url(&payload.target_url).map_err(ApiError::bad_request)?; + if payload.secret.is_empty() { + return Err(ApiError::bad_request(anyhow::anyhow!( + "webhook secret cannot be empty" + ))); + } + let encrypted_secret = encrypt_webhook_secret(&payload.secret).map_err(ApiError::internal)?; + let event_types = parse_event_types(payload.event_types)?; + + let now = Utc::now().naive_utc(); + let model = callisto::mega_webhook::Model { + id: IdInstance::next_id(), + target_url: payload.target_url, + secret: encrypted_secret, + event_types: serde_json::to_string(&event_types).unwrap_or_else(|_| "[]".to_string()), + path_filter: payload.path_filter, + active: payload.active.unwrap_or(true), + created_at: now, + updated_at: now, + }; + + let created = state + .webhook_stg() + .create_webhook(model, event_types) + .await?; + Ok(Json(CommonResult::success(Some(created.into())))) +} + +/// List webhooks +#[utoipa::path( + get, + path = "/webhooks", + params( + ("page" = Option, Query, description = "Page number, starts from 1. Default: 1"), + ("per_page" = Option, Query, description = "Items per page. Default: 20") + ), + responses( + (status = 200, body = CommonResult>, content_type = "application/json") + ), + tag = WEBHOOK_TAG +)] +async fn list_webhooks( + state: State, + Query(query): Query, +) -> Result>>, ApiError> { + let pagination = build_webhook_pagination(query)?; + let (webhooks, total) = state.webhook_stg().list_webhooks(pagination).await?; + let items: Vec = webhooks.into_iter().map(|w| w.into()).collect(); + Ok(Json(CommonResult::success(Some(CommonPage { + total, + items, + })))) +} + +/// Delete a webhook +#[utoipa::path( + delete, + params(("id", description = "Webhook ID")), + path = "/webhooks/{id}", + responses( + (status = 200, body = CommonResult, content_type = "application/json"), + (status = 404, description = "Webhook not found") + ), + tag = WEBHOOK_TAG +)] +async fn delete_webhook( + state: State, + Path(id): Path, +) -> Result>, ApiError> { + state.webhook_stg().delete_webhook(id).await?; + Ok(Json(CommonResult::success(None))) +} + +fn build_webhook_pagination(query: ListWebhooksQuery) -> Result { + let mut pagination = Pagination::default(); + if let Some(page) = query.page { + pagination.page = page; + } + if let Some(per_page) = query.per_page { + pagination.per_page = per_page; + } + + if pagination.page == 0 { + return Err(ApiError::bad_request(anyhow::anyhow!( + "page must be greater than or equal to 1" + ))); + } + if pagination.per_page == 0 { + return Err(ApiError::bad_request(anyhow::anyhow!( + "per_page must be greater than or equal to 1" + ))); + } + + Ok(pagination) +} diff --git a/mono/src/server/http_server.rs b/mono/src/server/http_server.rs index 6bf1b337c..9c5e54776 100644 --- a/mono/src/server/http_server.rs +++ b/mono/src/server/http_server.rs @@ -428,6 +428,7 @@ pub const LFS_TAG: &str = "Git LFS"; pub const CODE_REVIEW_TAG: &str = "Code Review"; pub const BUILD_TRIGGER_TAG: &str = "Build Trigger"; pub const GROUP_PERMISSION_TAG: &str = "Group Permission Management"; +pub const WEBHOOK_TAG: &str = "Webhook"; #[derive(OpenApi)] #[openapi()] struct ApiDoc; diff --git a/scripts/webhook_receiver.py b/scripts/webhook_receiver.py new file mode 100644 index 000000000..5a5969674 --- /dev/null +++ b/scripts/webhook_receiver.py @@ -0,0 +1,15 @@ +from http.server import HTTPServer, BaseHTTPRequestHandler +import json + +class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get('Content-Length', 0)) + body = self.rfile.read(length) + print(f'\n--- Webhook Received ---') + print(f'X-Mega-Event: {self.headers.get("X-Mega-Event")}') + print(f'X-Mega-Signature: {self.headers.get("X-Mega-Signature")}') + print(json.dumps(json.loads(body), indent=2)) + self.send_response(200) + self.end_headers() + +HTTPServer(('0.0.0.0', 9001), Handler).serve_forever() diff --git a/webhook-test.txt b/webhook-test.txt new file mode 100644 index 000000000..1c4f9af91 --- /dev/null +++ b/webhook-test.txt @@ -0,0 +1 @@ +test_1