| 1 | |
| 2 | |
| 3 | |
| 4 | |
| 5 | |
| 6 | |
| 7 | use forge_core::webhooks::{Webhook, WebhookEvent}; |
| 8 | use forge_core::{Comment, Issue, Label, PullRequest, Repo, Review, User}; |
| 9 | use forge_git::browse::CommitInfo; |
| 10 | use serde_json::{Map, Value, json}; |
| 11 | |
| 12 | use super::payload; |
| 13 | use crate::internal::is_zero_oid; |
| 14 | use crate::state::AppState; |
| 15 | |
| 16 | |
| 17 | const MAX_PUSH_COMMITS: usize = 500; |
| 18 | |
| 19 | const MAX_NEW_REF_COMMITS: usize = 50; |
| 20 | |
| 21 | |
| 22 | #[derive(Clone)] |
| 23 | pub struct Sender { |
| 24 | pub id: String, |
| 25 | pub username: String, |
| 26 | } |
| 27 | |
| 28 | impl Sender { |
| 29 | pub fn of(user: &User) -> Self { |
| 30 | Self { |
| 31 | id: user.id.clone(), |
| 32 | username: user.username.clone(), |
| 33 | } |
| 34 | } |
| 35 | |
| 36 | pub fn system() -> Self { |
| 37 | Self { |
| 38 | id: String::new(), |
| 39 | username: String::new(), |
| 40 | } |
| 41 | } |
| 42 | } |
| 43 | |
| 44 | |
| 45 | pub async fn lookup_sender(state: &AppState, actor_id: &str) -> Sender { |
| 46 | if actor_id.is_empty() { |
| 47 | return Sender::system(); |
| 48 | } |
| 49 | match state.store.user_by_id(actor_id).await { |
| 50 | Ok(Some(user)) => Sender::of(&user), |
| 51 | Ok(None) => Sender::system(), |
| 52 | Err(e) => { |
| 53 | tracing::error!(error = %e, actor_id, "webhooks: sender lookup failed"); |
| 54 | Sender::system() |
| 55 | } |
| 56 | } |
| 57 | } |
| 58 | |
| 59 | |
| 60 | |
| 61 | pub async fn emit( |
| 62 | state: &AppState, |
| 63 | repo: &Repo, |
| 64 | sender: &Sender, |
| 65 | event: WebhookEvent, |
| 66 | action: &str, |
| 67 | payload: Value, |
| 68 | ) { |
| 69 | let hooks = subscribed(state, repo, event).await; |
| 70 | if hooks.is_empty() { |
| 71 | return; |
| 72 | } |
| 73 | let body = envelope(state, repo, sender, action, payload).await; |
| 74 | enqueue(state, &hooks, event, action, &body).await; |
| 75 | } |
| 76 | |
| 77 | |
| 78 | |
| 79 | |
| 80 | pub async fn emit_ref_update( |
| 81 | state: &AppState, |
| 82 | repo: &Repo, |
| 83 | sender: &Sender, |
| 84 | refname: &str, |
| 85 | old: &str, |
| 86 | new: &str, |
| 87 | ) { |
| 88 | let Some((ref_type, name)) = split_ref(refname) else { |
| 89 | return; |
| 90 | }; |
| 91 | if !subscribed(state, repo, WebhookEvent::Push).await.is_empty() { |
| 92 | let (commits, total) = if is_zero_oid(new) { |
| 93 | (Vec::new(), 0) |
| 94 | } else { |
| 95 | new_commits(state, repo, old, new).await |
| 96 | }; |
| 97 | let payload = payload::push( |
| 98 | &state.config.base_url, |
| 99 | repo, |
| 100 | refname, |
| 101 | old, |
| 102 | new, |
| 103 | &commits, |
| 104 | total, |
| 105 | ); |
| 106 | emit(state, repo, sender, WebhookEvent::Push, "", payload).await; |
| 107 | } |
| 108 | if is_zero_oid(old) { |
| 109 | let payload = payload::ref_change(ref_type, name); |
| 110 | emit(state, repo, sender, WebhookEvent::Create, "", payload).await; |
| 111 | } |
| 112 | if is_zero_oid(new) { |
| 113 | let payload = payload::ref_change(ref_type, name); |
| 114 | emit(state, repo, sender, WebhookEvent::Delete, "", payload).await; |
| 115 | } |
| 116 | } |
| 117 | |
| 118 | |
| 119 | pub async fn emit_issue( |
| 120 | state: &AppState, |
| 121 | repo: &Repo, |
| 122 | sender: &Sender, |
| 123 | action: &str, |
| 124 | issue: &Issue, |
| 125 | ) { |
| 126 | let payload = payload::issues(&state.config.base_url, repo, issue); |
| 127 | emit(state, repo, sender, WebhookEvent::Issues, action, payload).await; |
| 128 | } |
| 129 | |
| 130 | |
| 131 | pub async fn emit_pull( |
| 132 | state: &AppState, |
| 133 | repo: &Repo, |
| 134 | sender: &Sender, |
| 135 | action: &str, |
| 136 | pr: &PullRequest, |
| 137 | ) { |
| 138 | let payload = payload::pull_request(&state.config.base_url, repo, pr); |
| 139 | emit( |
| 140 | state, |
| 141 | repo, |
| 142 | sender, |
| 143 | WebhookEvent::PullRequest, |
| 144 | action, |
| 145 | payload, |
| 146 | ) |
| 147 | .await; |
| 148 | } |
| 149 | |
| 150 | |
| 151 | |
| 152 | |
| 153 | pub async fn emit_item_action( |
| 154 | state: &AppState, |
| 155 | repo: &Repo, |
| 156 | sender: &Sender, |
| 157 | pull: bool, |
| 158 | number: i64, |
| 159 | action: &str, |
| 160 | ) { |
| 161 | if pull { |
| 162 | match state.store.pull_by_number(&repo.id, number).await { |
| 163 | Ok(Some(pr)) => emit_pull(state, repo, sender, action, &pr).await, |
| 164 | Ok(None) => {} |
| 165 | Err(e) => { |
| 166 | tracing::error!(error = %e, repo_id = %repo.id, number, "webhooks: loading pull request failed"); |
| 167 | } |
| 168 | } |
| 169 | } else { |
| 170 | match state.store.issue_by_number(&repo.id, number).await { |
| 171 | Ok(Some(issue)) => emit_issue(state, repo, sender, action, &issue).await, |
| 172 | Ok(None) => {} |
| 173 | Err(e) => { |
| 174 | tracing::error!(error = %e, repo_id = %repo.id, number, "webhooks: loading issue failed"); |
| 175 | } |
| 176 | } |
| 177 | } |
| 178 | } |
| 179 | |
| 180 | |
| 181 | |
| 182 | pub async fn emit_label( |
| 183 | state: &AppState, |
| 184 | repo: &Repo, |
| 185 | sender: &Sender, |
| 186 | action: &str, |
| 187 | pull: bool, |
| 188 | issue: &Issue, |
| 189 | label: &Label, |
| 190 | ) { |
| 191 | let field = ("label", json!({"name": label.name, "color": label.color})); |
| 192 | emit_item_field(state, repo, sender, action, pull, issue, field).await; |
| 193 | } |
| 194 | |
| 195 | |
| 196 | |
| 197 | pub async fn emit_assignee( |
| 198 | state: &AppState, |
| 199 | repo: &Repo, |
| 200 | sender: &Sender, |
| 201 | action: &str, |
| 202 | pull: bool, |
| 203 | issue: &Issue, |
| 204 | username: &str, |
| 205 | ) { |
| 206 | let field = ("assignee", json!({"username": username})); |
| 207 | emit_item_field(state, repo, sender, action, pull, issue, field).await; |
| 208 | } |
| 209 | |
| 210 | |
| 211 | |
| 212 | pub async fn emit_label_changes( |
| 213 | state: &AppState, |
| 214 | repo: &Repo, |
| 215 | sender: &Sender, |
| 216 | pull: bool, |
| 217 | issue: &Issue, |
| 218 | before: &[Label], |
| 219 | after: &[Label], |
| 220 | ) { |
| 221 | let has = |list: &[Label], name: &str| list.iter().any(|label| label.name == name); |
| 222 | for label in after.iter().filter(|label| !has(before, &label.name)) { |
| 223 | emit_label(state, repo, sender, "labeled", pull, issue, label).await; |
| 224 | } |
| 225 | for label in before.iter().filter(|label| !has(after, &label.name)) { |
| 226 | emit_label(state, repo, sender, "unlabeled", pull, issue, label).await; |
| 227 | } |
| 228 | } |
| 229 | |
| 230 | |
| 231 | pub async fn emit_assignee_changes( |
| 232 | state: &AppState, |
| 233 | repo: &Repo, |
| 234 | sender: &Sender, |
| 235 | pull: bool, |
| 236 | issue: &Issue, |
| 237 | before: &[String], |
| 238 | after: &[String], |
| 239 | ) { |
| 240 | for name in after.iter().filter(|name| !before.contains(name)) { |
| 241 | emit_assignee(state, repo, sender, "assigned", pull, issue, name).await; |
| 242 | } |
| 243 | for name in before.iter().filter(|name| !after.contains(name)) { |
| 244 | emit_assignee(state, repo, sender, "unassigned", pull, issue, name).await; |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | |
| 249 | |
| 250 | pub async fn emit_milestone_change( |
| 251 | state: &AppState, |
| 252 | repo: &Repo, |
| 253 | sender: &Sender, |
| 254 | pull: bool, |
| 255 | issue: &Issue, |
| 256 | before: Option<&str>, |
| 257 | after: Option<&str>, |
| 258 | ) { |
| 259 | if before == after { |
| 260 | return; |
| 261 | } |
| 262 | if let Some(title) = before { |
| 263 | let field = ("milestone", json!({"title": title})); |
| 264 | emit_item_field(state, repo, sender, "demilestoned", pull, issue, field).await; |
| 265 | } |
| 266 | if let Some(title) = after { |
| 267 | let field = ("milestone", json!({"title": title})); |
| 268 | emit_item_field(state, repo, sender, "milestoned", pull, issue, field).await; |
| 269 | } |
| 270 | } |
| 271 | |
| 272 | |
| 273 | pub async fn emit_comment( |
| 274 | state: &AppState, |
| 275 | repo: &Repo, |
| 276 | sender: &Sender, |
| 277 | action: &str, |
| 278 | pull: bool, |
| 279 | issue: &Issue, |
| 280 | comment: &Comment, |
| 281 | ) { |
| 282 | let payload = payload::issue_comment(&state.config.base_url, repo, issue, pull, comment); |
| 283 | emit( |
| 284 | state, |
| 285 | repo, |
| 286 | sender, |
| 287 | WebhookEvent::IssueComment, |
| 288 | action, |
| 289 | payload, |
| 290 | ) |
| 291 | .await; |
| 292 | } |
| 293 | |
| 294 | |
| 295 | pub async fn emit_review( |
| 296 | state: &AppState, |
| 297 | repo: &Repo, |
| 298 | sender: &Sender, |
| 299 | pr: &PullRequest, |
| 300 | review: &Review, |
| 301 | ) { |
| 302 | let payload = payload::pull_request_review(&state.config.base_url, repo, pr, review); |
| 303 | emit( |
| 304 | state, |
| 305 | repo, |
| 306 | sender, |
| 307 | WebhookEvent::PullRequestReview, |
| 308 | "submitted", |
| 309 | payload, |
| 310 | ) |
| 311 | .await; |
| 312 | } |
| 313 | |
| 314 | |
| 315 | |
| 316 | |
| 317 | pub async fn emit_ping(state: &AppState, hook: &Webhook) -> forge_core::Result<()> { |
| 318 | let repo = state |
| 319 | .store |
| 320 | .repo_by_id(&hook.repo_id) |
| 321 | .await? |
| 322 | .ok_or(forge_core::Error::NotFound)?; |
| 323 | let body = envelope(state, &repo, &Sender::system(), "", payload::ping(&hook.id)).await; |
| 324 | if enqueue( |
| 325 | state, |
| 326 | std::slice::from_ref(hook), |
| 327 | WebhookEvent::Ping, |
| 328 | "", |
| 329 | &body, |
| 330 | ) |
| 331 | .await |
| 332 | == 0 |
| 333 | { |
| 334 | return Err(forge_core::Error::Database( |
| 335 | "could not queue the ping delivery".into(), |
| 336 | )); |
| 337 | } |
| 338 | Ok(()) |
| 339 | } |
| 340 | |
| 341 | |
| 342 | |
| 343 | |
| 344 | async fn emit_item_field( |
| 345 | state: &AppState, |
| 346 | repo: &Repo, |
| 347 | sender: &Sender, |
| 348 | action: &str, |
| 349 | pull: bool, |
| 350 | issue: &Issue, |
| 351 | (key, field): (&str, Value), |
| 352 | ) { |
| 353 | let (event, mut payload) = if pull { |
| 354 | match state.store.pull_by_number(&repo.id, issue.number).await { |
| 355 | Ok(Some(pr)) => ( |
| 356 | WebhookEvent::PullRequest, |
| 357 | payload::pull_request(&state.config.base_url, repo, &pr), |
| 358 | ), |
| 359 | Ok(None) => return, |
| 360 | Err(e) => { |
| 361 | tracing::error!(error = %e, repo_id = %repo.id, number = issue.number, "webhooks: loading pull request failed"); |
| 362 | return; |
| 363 | } |
| 364 | } |
| 365 | } else { |
| 366 | ( |
| 367 | WebhookEvent::Issues, |
| 368 | payload::issues(&state.config.base_url, repo, issue), |
| 369 | ) |
| 370 | }; |
| 371 | if let Value::Object(map) = &mut payload { |
| 372 | map.insert(key.to_owned(), field); |
| 373 | } |
| 374 | emit(state, repo, sender, event, action, payload).await; |
| 375 | } |
| 376 | |
| 377 | |
| 378 | async fn subscribed(state: &AppState, repo: &Repo, event: WebhookEvent) -> Vec<Webhook> { |
| 379 | match state.store.hooks_for_event(&repo.id, event).await { |
| 380 | Ok(hooks) => hooks, |
| 381 | Err(e) => { |
| 382 | tracing::error!(error = %e, repo_id = %repo.id, event = event.as_str(), "webhooks: hook lookup failed"); |
| 383 | Vec::new() |
| 384 | } |
| 385 | } |
| 386 | } |
| 387 | |
| 388 | |
| 389 | |
| 390 | async fn owner_kind(state: &AppState, repo: &Repo) -> &'static str { |
| 391 | match state.store.user_by_id(&repo.owner_id).await { |
| 392 | Ok(Some(owner)) if owner.is_org() => "org", |
| 393 | _ => "user", |
| 394 | } |
| 395 | } |
| 396 | |
| 397 | |
| 398 | async fn envelope( |
| 399 | state: &AppState, |
| 400 | repo: &Repo, |
| 401 | sender: &Sender, |
| 402 | action: &str, |
| 403 | payload: Value, |
| 404 | ) -> String { |
| 405 | let mut map = match payload { |
| 406 | Value::Object(map) => map, |
| 407 | _ => Map::new(), |
| 408 | }; |
| 409 | if !action.is_empty() { |
| 410 | map.insert("action".to_owned(), json!(action)); |
| 411 | } |
| 412 | let kind = owner_kind(state, repo).await; |
| 413 | map.insert( |
| 414 | "repository".to_owned(), |
| 415 | payload::repository(&state.config.base_url, repo, kind), |
| 416 | ); |
| 417 | map.insert( |
| 418 | "sender".to_owned(), |
| 419 | payload::sender(&sender.id, &sender.username), |
| 420 | ); |
| 421 | Value::Object(map).to_string() |
| 422 | } |
| 423 | |
| 424 | |
| 425 | |
| 426 | async fn enqueue( |
| 427 | state: &AppState, |
| 428 | hooks: &[Webhook], |
| 429 | event: WebhookEvent, |
| 430 | action: &str, |
| 431 | body: &str, |
| 432 | ) -> usize { |
| 433 | |
| 434 | let headers = json!({ |
| 435 | "Content-Type": "application/json", |
| 436 | "X-Forge-Event": event.as_str(), |
| 437 | }) |
| 438 | .to_string(); |
| 439 | let mut queued = 0; |
| 440 | for hook in hooks { |
| 441 | match state |
| 442 | .store |
| 443 | .enqueue_delivery(&hook.id, event, action, &headers, body) |
| 444 | .await |
| 445 | { |
| 446 | Ok(_) => { |
| 447 | queued += 1; |
| 448 | state.webhook_waker.notify_one(); |
| 449 | } |
| 450 | Err(e) => { |
| 451 | tracing::error!(error = %e, hook_id = %hook.id, event = event.as_str(), "webhooks: enqueue failed"); |
| 452 | } |
| 453 | } |
| 454 | } |
| 455 | queued |
| 456 | } |
| 457 | |
| 458 | |
| 459 | fn split_ref(refname: &str) -> Option<(&'static str, &str)> { |
| 460 | if let Some(name) = refname.strip_prefix("refs/heads/") { |
| 461 | Some(("branch", name)) |
| 462 | } else { |
| 463 | refname.strip_prefix("refs/tags/").map(|name| ("tag", name)) |
| 464 | } |
| 465 | } |
| 466 | |
| 467 | |
| 468 | |
| 469 | async fn new_commits( |
| 470 | state: &AppState, |
| 471 | repo: &Repo, |
| 472 | old: &str, |
| 473 | new: &str, |
| 474 | ) -> (Vec<CommitInfo>, usize) { |
| 475 | let dir = match forge_git::repo::repo_path(&state.git.data_dir, &repo.id) { |
| 476 | Ok(dir) => dir, |
| 477 | Err(e) => { |
| 478 | tracing::error!(error = %e, repo_id = %repo.id, "webhooks: bad repo path for push commits"); |
| 479 | return (Vec::new(), 0); |
| 480 | } |
| 481 | }; |
| 482 | let old = (!is_zero_oid(old)).then_some(old); |
| 483 | let max = if old.is_some() { |
| 484 | MAX_PUSH_COMMITS |
| 485 | } else { |
| 486 | MAX_NEW_REF_COMMITS |
| 487 | }; |
| 488 | match forge_git::history::new_commits(&state.git, &dir, old, new, max).await { |
| 489 | Ok(commits) => { |
| 490 | let total = commits.len(); |
| 491 | (commits, total) |
| 492 | } |
| 493 | Err(e) => { |
| 494 | tracing::error!(error = %e, repo_id = %repo.id, "webhooks: listing push commits failed"); |
| 495 | (Vec::new(), 0) |
| 496 | } |
| 497 | } |
| 498 | } |
| 499 | |
| 500 | #[cfg(test)] |
| 501 | mod tests { |
| 502 | use super::*; |
| 503 | use crate::test_support::{KEY, test_state}; |
| 504 | use forge_core::Visibility; |
| 505 | |
| 506 | |
| 507 | |
| 508 | async fn fixture(dir: &std::path::Path) -> (AppState, Repo, [String; 3]) { |
| 509 | let state = test_state(dir).await; |
| 510 | let owner = state |
| 511 | .store |
| 512 | .create_user("alice", None, "unused", false) |
| 513 | .await |
| 514 | .expect("user"); |
| 515 | let repo = state |
| 516 | .store |
| 517 | .create_repo(&owner.id, "proj", "", Visibility::Public, "main") |
| 518 | .await |
| 519 | .expect("repo"); |
| 520 | let sealed = forge_core::secret::seal(&KEY, b"s").expect("seal"); |
| 521 | let mut ids = Vec::new(); |
| 522 | for (events, active) in [ |
| 523 | (vec![WebhookEvent::Push], true), |
| 524 | (vec![WebhookEvent::Issues], true), |
| 525 | (vec![WebhookEvent::Push], false), |
| 526 | ] { |
| 527 | let hook = state |
| 528 | .store |
| 529 | .create_webhook( |
| 530 | &repo.id, |
| 531 | &owner.id, |
| 532 | "https://hooks.example/h", |
| 533 | sealed.as_bytes(), |
| 534 | &events, |
| 535 | active, |
| 536 | ) |
| 537 | .await |
| 538 | .expect("hook"); |
| 539 | ids.push(hook.id); |
| 540 | } |
| 541 | let ids: [String; 3] = ids.try_into().expect("three hooks"); |
| 542 | (state, repo, ids) |
| 543 | } |
| 544 | |
| 545 | #[tokio::test] |
| 546 | async fn emit_queues_only_for_subscribed_active_hooks() { |
| 547 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 548 | let (state, repo, [push_hook, issues_hook, inactive_push]) = fixture(tmp.path()).await; |
| 549 | |
| 550 | emit( |
| 551 | &state, |
| 552 | &repo, |
| 553 | &Sender::system(), |
| 554 | WebhookEvent::Push, |
| 555 | "", |
| 556 | json!({"ref": "refs/heads/main"}), |
| 557 | ) |
| 558 | .await; |
| 559 | |
| 560 | assert_eq!( |
| 561 | state |
| 562 | .store |
| 563 | .list_deliveries(&push_hook, 10) |
| 564 | .await |
| 565 | .expect("list") |
| 566 | .len(), |
| 567 | 1 |
| 568 | ); |
| 569 | assert!( |
| 570 | state |
| 571 | .store |
| 572 | .list_deliveries(&issues_hook, 10) |
| 573 | .await |
| 574 | .expect("list") |
| 575 | .is_empty() |
| 576 | ); |
| 577 | assert!( |
| 578 | state |
| 579 | .store |
| 580 | .list_deliveries(&inactive_push, 10) |
| 581 | .await |
| 582 | .expect("list") |
| 583 | .is_empty() |
| 584 | ); |
| 585 | } |
| 586 | |
| 587 | #[tokio::test] |
| 588 | async fn emit_body_has_repository_sender_and_action_only_when_given() { |
| 589 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 590 | let (state, repo, [_, issues_hook, _]) = fixture(tmp.path()).await; |
| 591 | let sender = Sender { |
| 592 | id: "u1".into(), |
| 593 | username: "alice".into(), |
| 594 | }; |
| 595 | |
| 596 | emit( |
| 597 | &state, |
| 598 | &repo, |
| 599 | &sender, |
| 600 | WebhookEvent::Issues, |
| 601 | "opened", |
| 602 | json!({"issue": {"number": 3}}), |
| 603 | ) |
| 604 | .await; |
| 605 | |
| 606 | let rows = state |
| 607 | .store |
| 608 | .list_deliveries(&issues_hook, 10) |
| 609 | .await |
| 610 | .expect("list"); |
| 611 | let row = rows.first().expect("one delivery"); |
| 612 | assert_eq!(row.action, "opened"); |
| 613 | let body: Value = serde_json::from_str(&row.request_body).expect("json"); |
| 614 | assert_eq!(body["action"], "opened"); |
| 615 | assert_eq!(body["issue"]["number"], 3); |
| 616 | assert_eq!(body["repository"]["full_name"], "alice/proj"); |
| 617 | assert_eq!(body["sender"], json!({"id": "u1", "username": "alice"})); |
| 618 | let headers: Value = serde_json::from_str(&row.request_headers).expect("json"); |
| 619 | assert_eq!(headers["X-Forge-Event"], "issues"); |
| 620 | assert!(headers.get("X-Forge-Signature-256").is_none()); |
| 621 | } |
| 622 | |
| 623 | #[tokio::test] |
| 624 | async fn system_changes_send_null_sender_and_no_action() { |
| 625 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 626 | let (state, repo, [push_hook, _, _]) = fixture(tmp.path()).await; |
| 627 | emit( |
| 628 | &state, |
| 629 | &repo, |
| 630 | &Sender::system(), |
| 631 | WebhookEvent::Push, |
| 632 | "", |
| 633 | json!({}), |
| 634 | ) |
| 635 | .await; |
| 636 | let rows = state |
| 637 | .store |
| 638 | .list_deliveries(&push_hook, 10) |
| 639 | .await |
| 640 | .expect("list"); |
| 641 | let body: Value = serde_json::from_str(&rows[0].request_body).expect("json"); |
| 642 | assert!(body["sender"].is_null()); |
| 643 | assert!(body.get("action").is_none()); |
| 644 | } |
| 645 | |
| 646 | #[tokio::test] |
| 647 | async fn ping_reaches_its_hook_even_without_the_subscription() { |
| 648 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 649 | let (state, _repo, [_, issues_hook, _]) = fixture(tmp.path()).await; |
| 650 | let hook = state |
| 651 | .store |
| 652 | .webhook(&issues_hook) |
| 653 | .await |
| 654 | .expect("lookup") |
| 655 | .expect("hook"); |
| 656 | |
| 657 | emit_ping(&state, &hook).await.expect("ping queued"); |
| 658 | |
| 659 | let rows = state |
| 660 | .store |
| 661 | .list_deliveries(&issues_hook, 10) |
| 662 | .await |
| 663 | .expect("list"); |
| 664 | assert_eq!(rows.len(), 1); |
| 665 | assert_eq!(rows[0].event, WebhookEvent::Ping); |
| 666 | let body: Value = serde_json::from_str(&rows[0].request_body).expect("json"); |
| 667 | assert_eq!(body["hook_id"], issues_hook.as_str()); |
| 668 | } |
| 669 | |
| 670 | |
| 671 | async fn hook_for(state: &AppState, repo: &Repo, events: &[WebhookEvent]) -> String { |
| 672 | let sealed = forge_core::secret::seal(&KEY, b"s").expect("seal"); |
| 673 | state |
| 674 | .store |
| 675 | .create_webhook( |
| 676 | &repo.id, |
| 677 | &repo.owner_id, |
| 678 | "https://hooks.example/h", |
| 679 | sealed.as_bytes(), |
| 680 | events, |
| 681 | true, |
| 682 | ) |
| 683 | .await |
| 684 | .expect("hook") |
| 685 | .id |
| 686 | } |
| 687 | |
| 688 | |
| 689 | async fn bodies(state: &AppState, hook: &str) -> Vec<(WebhookEvent, String, Value)> { |
| 690 | state |
| 691 | .store |
| 692 | .list_deliveries(hook, 50) |
| 693 | .await |
| 694 | .expect("list") |
| 695 | .into_iter() |
| 696 | .map(|d| { |
| 697 | let body = serde_json::from_str(&d.request_body).expect("json"); |
| 698 | (d.event, d.action, body) |
| 699 | }) |
| 700 | .collect() |
| 701 | } |
| 702 | |
| 703 | #[tokio::test] |
| 704 | async fn pull_request_label_change_queues_pull_request_labeled() { |
| 705 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 706 | let state = test_state(tmp.path()).await; |
| 707 | let owner = state |
| 708 | .store |
| 709 | .create_user("alice", None, "unused", false) |
| 710 | .await |
| 711 | .expect("user"); |
| 712 | let repo = state |
| 713 | .store |
| 714 | .create_repo(&owner.id, "proj", "", Visibility::Public, "main") |
| 715 | .await |
| 716 | .expect("repo"); |
| 717 | let zeros = "1".repeat(40); |
| 718 | let pr = state |
| 719 | .store |
| 720 | .create_pull( |
| 721 | &repo.id, &owner.id, "Title", "", "main", &repo.id, "feat", &zeros, &zeros, |
| 722 | ) |
| 723 | .await |
| 724 | .expect("pr"); |
| 725 | let hook = hook_for( |
| 726 | &state, |
| 727 | &repo, |
| 728 | &[WebhookEvent::PullRequest, WebhookEvent::Issues], |
| 729 | ) |
| 730 | .await; |
| 731 | let label = Label { |
| 732 | id: "l1".into(), |
| 733 | repo_id: repo.id.clone(), |
| 734 | name: "bug".into(), |
| 735 | color: "#ff0000".into(), |
| 736 | description: String::new(), |
| 737 | }; |
| 738 | |
| 739 | emit_label_changes( |
| 740 | &state, |
| 741 | &repo, |
| 742 | &Sender::of(&owner), |
| 743 | true, |
| 744 | &pr.issue, |
| 745 | &[], |
| 746 | std::slice::from_ref(&label), |
| 747 | ) |
| 748 | .await; |
| 749 | |
| 750 | let rows = bodies(&state, &hook).await; |
| 751 | assert_eq!(rows.len(), 1); |
| 752 | let (event, action, body) = &rows[0]; |
| 753 | assert_eq!(*event, WebhookEvent::PullRequest); |
| 754 | assert_eq!(action, "labeled"); |
| 755 | assert_eq!(body["action"], "labeled"); |
| 756 | assert_eq!(body["label"], json!({"name": "bug", "color": "#ff0000"})); |
| 757 | assert_eq!(body["number"], pr.issue.number); |
| 758 | assert_eq!(body["pull_request"]["title"], "Title"); |
| 759 | } |
| 760 | |
| 761 | #[tokio::test] |
| 762 | async fn issue_milestone_set_and_clear_queue_milestoned_and_demilestoned() { |
| 763 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 764 | let state = test_state(tmp.path()).await; |
| 765 | let owner = state |
| 766 | .store |
| 767 | .create_user("alice", None, "unused", false) |
| 768 | .await |
| 769 | .expect("user"); |
| 770 | let repo = state |
| 771 | .store |
| 772 | .create_repo(&owner.id, "proj", "", Visibility::Public, "main") |
| 773 | .await |
| 774 | .expect("repo"); |
| 775 | let issue = state |
| 776 | .store |
| 777 | .create_issue(&repo.id, &owner.id, "Broken", "") |
| 778 | .await |
| 779 | .expect("issue"); |
| 780 | let hook = hook_for(&state, &repo, &[WebhookEvent::Issues]).await; |
| 781 | let sender = Sender::of(&owner); |
| 782 | |
| 783 | emit_milestone_change(&state, &repo, &sender, false, &issue, None, Some("v1")).await; |
| 784 | emit_milestone_change(&state, &repo, &sender, false, &issue, Some("v1"), None).await; |
| 785 | |
| 786 | emit_milestone_change( |
| 787 | &state, |
| 788 | &repo, |
| 789 | &sender, |
| 790 | false, |
| 791 | &issue, |
| 792 | Some("v1"), |
| 793 | Some("v1"), |
| 794 | ) |
| 795 | .await; |
| 796 | |
| 797 | let rows = bodies(&state, &hook).await; |
| 798 | assert_eq!(rows.len(), 2); |
| 799 | |
| 800 | let (event, action, body) = &rows[0]; |
| 801 | assert_eq!(*event, WebhookEvent::Issues); |
| 802 | assert_eq!(action, "demilestoned"); |
| 803 | assert_eq!(body["milestone"], json!({"title": "v1"})); |
| 804 | assert_eq!(rows[1].1, "milestoned"); |
| 805 | assert_eq!(rows[1].2["milestone"], json!({"title": "v1"})); |
| 806 | assert_eq!(rows[1].2["issue"]["number"], issue.number); |
| 807 | } |
| 808 | } |