diff --git a/Cargo.lock b/Cargo.lock index 19e8924c..4c6c3d4e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1433,6 +1433,20 @@ dependencies = [ "wiremock", ] +[[package]] +name = "mergify-events" +version = "0.0.0" +dependencies = [ + "chrono", + "mergify-core", + "serde", + "serde_json", + "thiserror", + "tokio", + "url", + "wiremock", +] + [[package]] name = "mergify-freeze" version = "0.0.0" diff --git a/crates/mergify-core/src/http.rs b/crates/mergify-core/src/http.rs index e0afc5d0..82d96934 100644 --- a/crates/mergify-core/src/http.rs +++ b/crates/mergify-core/src/http.rs @@ -67,6 +67,19 @@ pub enum DeleteOutcome { NotFound, } +/// One page of a cursor-paginated Mergify list endpoint. +/// +/// `next_cursor` is the opaque cursor of the following page, +/// extracted from the response's RFC 5988 `Link` header +/// (`rel="next"`); `None` on the last page. The caller re-issues its +/// own query with `("cursor", …)` appended — only the cursor is +/// taken from the header, never the rest of the echoed URL, so a +/// caller's query cannot be silently rewritten by the server. +pub struct Page { + pub body: T, + pub next_cursor: Option, +} + /// Caller hook to remap a terminal non-2xx HTTP status to a domain /// error before the default flavor mapping. Receives the status as a /// `u16` (command crates never import `reqwest`) and the rendered @@ -190,6 +203,28 @@ impl Client { self.decode_json(resp).await } + /// GET one page of a cursor-paginated endpoint: like + /// [`Self::get_with_query`], but also return the next page's + /// cursor from the response's `Link` header (see [`Page`]). + pub async fn get_page( + &self, + path: &str, + query: &[(&str, &str)], + ) -> Result, CliError> { + let mut url = self.join(path)?; + if !query.is_empty() { + url.query_pairs_mut().extend_pairs(query.iter().copied()); + } + let resp = self.execute_request(self.inner.get(url)).await?; + let next_cursor = resp + .headers() + .get(reqwest::header::LINK) + .and_then(|value| value.to_str().ok()) + .and_then(next_cursor_from_link); + let body = self.decode_json(resp).await?; + Ok(Page { body, next_cursor }) + } + /// POST `body` as JSON to `path` and deserialize the JSON /// response as `T`. pub async fn post( @@ -522,6 +557,37 @@ fn is_transient(e: &reqwest::Error) -> bool { e.is_timeout() || e.is_connect() } +/// Extract the next page's `cursor` from an RFC 5988 `Link` header +/// value (`; rel="next", ; rel="last", …`). +/// +/// Only the `cursor` query parameter of the `rel="next"` target is +/// returned — the pagination contract is "same query, new cursor", +/// so the caller keeps building its own request rather than blindly +/// following a server-echoed URL. `None` when there is no `next` +/// link or its URL carries no cursor (both mean "last page"). +fn next_cursor_from_link(header: &str) -> Option { + for part in header.split(',') { + let mut segments = part.split(';'); + let target = segments.next()?.trim(); + let is_next = segments.any(|param| { + let param = param.trim(); + param + .strip_prefix("rel=") + .is_some_and(|rel| rel.trim_matches('"') == "next") + }); + if !is_next { + continue; + } + let target = target.strip_prefix('<')?.strip_suffix('>')?; + let url = Url::parse(target).ok()?; + return url + .query_pairs() + .find(|(key, _)| key == "cursor") + .map(|(_, value)| value.into_owned()); + } + None +} + /// The wait hinted by a rejection's rate-limit headers, if any. /// Prefers `Retry-After` (delta-seconds, as GitHub sends); falls back /// to `X-RateLimit-Reset` (epoch seconds) when `X-RateLimit-Remaining` @@ -1236,6 +1302,76 @@ mod tests { ); } + #[test] + fn next_cursor_from_link_finds_the_next_rel() { + let header = concat!( + "; rel=\"first\", ", + "; rel=\"next\", ", + "; rel=\"last\"", + ); + assert_eq!(next_cursor_from_link(header), Some("abc123".to_string())); + } + + #[test] + fn next_cursor_from_link_accepts_unquoted_rel() { + let header = "; rel=next"; + assert_eq!(next_cursor_from_link(header), Some("zzz".to_string())); + } + + #[test] + fn next_cursor_from_link_is_none_without_a_next_rel() { + let header = "; rel=\"first\""; + assert_eq!(next_cursor_from_link(header), None); + } + + #[test] + fn next_cursor_from_link_is_none_when_the_next_url_has_no_cursor() { + let header = "; rel=\"next\""; + assert_eq!(next_cursor_from_link(header), None); + } + + #[tokio::test] + async fn get_page_returns_body_and_next_cursor() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/paged")) + .respond_with( + ResponseTemplate::new(200) + .insert_header( + "link", + "; rel=\"next\"", + ) + .set_body_json(Foo { bar: 1 }), + ) + .expect(1) + .mount(&server) + .await; + + let client = fast_client(&server, ApiFlavor::Mergify); + let page: Page = client + .get_page("/paged", &[("per_page", "10")]) + .await + .unwrap(); + assert_eq!(page.body, Foo { bar: 1 }); + assert_eq!(page.next_cursor, Some("next-cursor".to_string())); + } + + #[tokio::test] + async fn get_page_has_no_cursor_on_the_last_page() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/paged")) + .respond_with(ResponseTemplate::new(200).set_body_json(Foo { bar: 2 })) + .expect(1) + .mount(&server) + .await; + + let client = fast_client(&server, ApiFlavor::Mergify); + let page: Page = client.get_page("/paged", &[]).await.unwrap(); + assert_eq!(page.body, Foo { bar: 2 }); + assert_eq!(page.next_cursor, None); + } + #[tokio::test] async fn get_if_exists_returns_some_on_2xx() { let server = MockServer::start().await; diff --git a/crates/mergify-core/src/lib.rs b/crates/mergify-core/src/lib.rs index 53b1d573..bf72da6c 100644 --- a/crates/mergify-core/src/lib.rs +++ b/crates/mergify-core/src/lib.rs @@ -27,7 +27,7 @@ pub mod pull_request; pub use command_context::CommandContext; pub use error::CliError; pub use exit_code::ExitCode; -pub use http::{ApiFlavor, Client as HttpClient, DeleteOutcome, RetryPolicy}; +pub use http::{ApiFlavor, Client as HttpClient, DeleteOutcome, Page, RetryPolicy}; pub use output::{Output, OutputMode, StdioOutput}; /// Compile-time version string taken from the crate package metadata diff --git a/crates/mergify-events/Cargo.toml b/crates/mergify-events/Cargo.toml new file mode 100644 index 00000000..446de43f --- /dev/null +++ b/crates/mergify-events/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "mergify-events" +version = "0.0.0" +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +authors.workspace = true +description = "Shared client for the Mergify activity log (`/logs`) and the `mergify events` command." +publish = false + +[dependencies] +mergify-core = { path = "../mergify-core" } +chrono = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true } +url = { workspace = true } +wiremock = { workspace = true } + +[lints] +workspace = true diff --git a/crates/mergify-events/src/client.rs b/crates/mergify-events/src/client.rs new file mode 100644 index 00000000..844d02fd --- /dev/null +++ b/crates/mergify-events/src/client.rs @@ -0,0 +1,445 @@ +//! The `/logs` fetch: explicit window, cursor pagination followed to +//! completion, newest-first guarantee. + +use mergify_core::CliError; +use mergify_core::http::Client; +use serde::Deserialize; + +use crate::event::Event; +use crate::window::Window; + +/// The API's page-size ceiling. +const PER_PAGE_MAX: usize = 100; + +/// A filtered query over a repository's activity log. +pub struct Query { + /// Only events belonging to this pull request; `None` queries the + /// whole repository. + pub pull_request: Option, + /// Only these event types (repeated `event_type` keys, OR + /// semantics); empty selects all. Values pass through verbatim — + /// the server owns the vocabulary, so a type this CLI has never + /// heard of still works against a newer engine. + pub event_types: Vec, + /// The explicit time range — never optional, see [`Window`]. + pub window: Window, + /// Stop after this many events. `None` follows pagination until + /// the window is exhausted. + pub limit: Option, +} + +#[derive(Deserialize)] +struct EventsResponse { + #[serde(default)] + events: Vec, +} + +/// Fetch every event matching `query`, newest first. +/// +/// Both window bounds are always sent (the API's silent +/// `received_to - 1 day` default is the trap this crate exists to +/// bury), pagination cursors are followed to completion unless +/// [`Query::limit`] stops earlier, and the result is sorted +/// newest-first — a guarantee of this function, not an assumption +/// about the server. Events with no parseable `received_at` sort +/// last, in server order. +/// +/// # Errors +/// +/// Propagates the API failure ([`CliError::MergifyApi`] for HTTP +/// errors). What that means is the caller's call — `queue show` +/// treats it as "could not determine", not as a command failure. +pub async fn fetch( + client: &Client, + repository: &str, + query: &Query, +) -> Result, CliError> { + let path = format!("/v1/repos/{repository}/logs"); + let from = query.window.from().to_rfc3339(); + let to = query.window.to().to_rfc3339(); + let pull_request = query.pull_request.map(|n| n.to_string()); + let per_page = query + .limit + .map_or(PER_PAGE_MAX, |limit| limit.clamp(1, PER_PAGE_MAX)) + .to_string(); + + let mut base: Vec<(&str, &str)> = Vec::new(); + if let Some(pull_request) = &pull_request { + base.push(("pull_request", pull_request)); + } + for event_type in &query.event_types { + base.push(("event_type", event_type)); + } + base.push(("received_from", &from)); + base.push(("received_to", &to)); + base.push(("per_page", &per_page)); + + let mut raw_events: Vec = Vec::new(); + let mut cursor: Option = None; + loop { + let mut pairs = base.clone(); + if let Some(cursor) = &cursor { + pairs.push(("cursor", cursor)); + } + let page = client.get_page::(&path, &pairs).await?; + raw_events.extend(page.body.events); + if query.limit.is_some_and(|limit| raw_events.len() >= limit) { + break; + } + match page.next_cursor { + // A server bug echoing the cursor we just used would + // otherwise loop forever. + Some(next) if Some(&next) != cursor.as_ref() => cursor = Some(next), + _ => break, + } + } + + let mut events: Vec = raw_events.into_iter().map(Event::from_raw).collect(); + // Stable sort: ties (and timestamp-less events, which compare + // smallest and thus sink to the end) keep the server's order. + events.sort_by_key(|event| std::cmp::Reverse(event.received_at_utc())); + if let Some(limit) = query.limit { + events.truncate(limit); + } + Ok(events) +} + +#[cfg(test)] +mod tests { + use chrono::DateTime; + use chrono::TimeDelta; + use chrono::Utc; + use mergify_core::http::ApiFlavor; + use serde_json::json; + use url::Url; + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::ResponseTemplate; + use wiremock::matchers::method; + use wiremock::matchers::path; + use wiremock::matchers::query_param; + use wiremock::matchers::query_param_is_missing; + + use super::*; + + fn at(iso: &str) -> DateTime { + DateTime::parse_from_rfc3339(iso) + .unwrap() + .with_timezone(&Utc) + } + + fn client(server: &MockServer) -> Client { + Client::new(Url::parse(&server.uri()).unwrap(), "t", ApiFlavor::Mergify).unwrap() + } + + fn event(id: u64, received_at: &str) -> serde_json::Value { + json!({ + "id": id, + "type": "action.queue.leave", + "received_at": received_at, + "metadata": {}, + }) + } + + fn page_body(events: &[serde_json::Value]) -> serde_json::Value { + json!({"size": events.len(), "per_page": 100, "events": events}) + } + + #[tokio::test] + async fn fetch_sends_both_window_bounds_and_every_filter() { + // The explicit window is the whole point: without + // `received_from` the API silently narrows to 1 day, and + // `received_to` rides along so the range is what the caller + // stated, not what the server's clock says. + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .respond_with(ResponseTemplate::new(200).set_body_json(page_body(&[]))) + .expect(1) + .mount(&server) + .await; + + let now = at("2026-07-30T00:00:00Z"); + let query = Query { + pull_request: Some(1700), + event_types: vec![ + "action.queue.leave".to_string(), + "command.queue".to_string(), + ], + window: Window::last(TimeDelta::days(7), now).unwrap(), + limit: None, + }; + fetch(&client(&server), "owner/repo", &query).await.unwrap(); + + let received = server.received_requests().await.unwrap(); + let pairs: Vec<(String, String)> = received[0] + .url + .query_pairs() + .map(|(k, v)| (k.into_owned(), v.into_owned())) + .collect(); + assert!( + pairs.contains(&("pull_request".into(), "1700".into())), + "got: {pairs:?}", + ); + // Repeated keys, OR semantics — one `event_type` per value. + assert!( + pairs.contains(&("event_type".into(), "action.queue.leave".into())), + "got: {pairs:?}", + ); + assert!( + pairs.contains(&("event_type".into(), "command.queue".into())), + "got: {pairs:?}", + ); + let from = pairs + .iter() + .find(|(k, _)| k == "received_from") + .map(|(_, v)| v.clone()) + .expect("received_from must be explicit, never the API's 1-day default"); + assert_eq!(at(&from), at("2026-07-23T00:00:00Z")); + let to = pairs + .iter() + .find(|(k, _)| k == "received_to") + .map(|(_, v)| v.clone()) + .expect("received_to must be explicit too"); + assert_eq!(at(&to), now); + assert!( + pairs.contains(&("per_page".into(), "100".into())), + "got: {pairs:?}", + ); + } + + #[tokio::test] + async fn fetch_omits_the_pull_request_filter_for_a_repo_wide_query() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .and(query_param_is_missing("pull_request")) + .and(query_param_is_missing("event_type")) + .respond_with(ResponseTemplate::new(200).set_body_json(page_body(&[]))) + .expect(1) + .mount(&server) + .await; + + let query = Query { + pull_request: None, + event_types: vec![], + window: Window::retained(at("2026-07-30T00:00:00Z")), + limit: None, + }; + let events = fetch(&client(&server), "owner/repo", &query).await.unwrap(); + assert!(events.is_empty()); + } + + #[tokio::test] + async fn fetch_follows_pagination_to_completion() { + // Three pages chained by Link-header cursors: the client must + // walk all of them, not silently return the first. + let server = MockServer::start().await; + let link = |cursor: &str| { + format!("; rel=\"next\"") + }; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .and(query_param_is_missing("cursor")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("link", link("c2").as_str()) + .set_body_json(page_body(&[event(3, "2026-07-29T12:00:00Z")])), + ) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .and(query_param("cursor", "c2")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("link", link("c3").as_str()) + .set_body_json(page_body(&[event(2, "2026-07-29T11:00:00Z")])), + ) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .and(query_param("cursor", "c3")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(page_body(&[event(1, "2026-07-29T10:00:00Z")])), + ) + .expect(1) + .mount(&server) + .await; + + let query = Query { + pull_request: None, + event_types: vec![], + window: Window::retained(at("2026-07-30T00:00:00Z")), + limit: None, + }; + let events = fetch(&client(&server), "owner/repo", &query).await.unwrap(); + let ids: Vec = events + .iter() + .map(|e| e.raw["id"].as_u64().unwrap()) + .collect(); + assert_eq!(ids, vec![3, 2, 1]); + } + + #[tokio::test] + async fn fetch_stops_at_the_limit_without_following_further_pages() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .and(query_param_is_missing("cursor")) + .and(query_param("per_page", "2")) + .respond_with( + ResponseTemplate::new(200) + .insert_header( + "link", + "; rel=\"next\"", + ) + .set_body_json(page_body(&[ + event(2, "2026-07-29T12:00:00Z"), + event(1, "2026-07-29T11:00:00Z"), + ])), + ) + .expect(1) + .mount(&server) + .await; + // No mock for cursor=more: requesting it would 404 and fail + // the fetch, so success proves the limit stopped pagination. + + let query = Query { + pull_request: None, + event_types: vec![], + window: Window::retained(at("2026-07-30T00:00:00Z")), + limit: Some(2), + }; + let events = fetch(&client(&server), "owner/repo", &query).await.unwrap(); + assert_eq!(events.len(), 2); + } + + #[tokio::test] + async fn fetch_guarantees_newest_first_even_when_the_server_does_not() { + // Ordering is this crate's promise. A server page in the + // wrong order gets re-sorted; a timestamp-less event sinks to + // the end instead of poisoning the sort. + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .respond_with(ResponseTemplate::new(200).set_body_json(page_body(&[ + event(1, "2026-07-29T10:00:00Z"), + json!({"id": 99, "type": "mystery"}), + event(3, "2026-07-29T12:00:00Z"), + event(2, "2026-07-29T11:00:00Z"), + ]))) + .expect(1) + .mount(&server) + .await; + + let query = Query { + pull_request: None, + event_types: vec![], + window: Window::retained(at("2026-07-30T00:00:00Z")), + limit: None, + }; + let events = fetch(&client(&server), "owner/repo", &query).await.unwrap(); + let ids: Vec = events + .iter() + .map(|e| e.raw["id"].as_u64().unwrap()) + .collect(); + assert_eq!(ids, vec![3, 2, 1, 99]); + } + + #[tokio::test] + async fn fetch_survives_a_server_echoing_the_same_cursor() { + // A `next` link pointing at the page we just asked for must + // terminate, not loop forever. + let server = MockServer::start().await; + let echo = "; rel=\"next\""; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .and(query_param_is_missing("cursor")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("link", echo) + .set_body_json(page_body(&[event(2, "2026-07-29T12:00:00Z")])), + ) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .and(query_param("cursor", "stuck")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("link", echo) + .set_body_json(page_body(&[event(1, "2026-07-29T11:00:00Z")])), + ) + .expect(1) + .mount(&server) + .await; + + let query = Query { + pull_request: None, + event_types: vec![], + window: Window::retained(at("2026-07-30T00:00:00Z")), + limit: None, + }; + let events = fetch(&client(&server), "owner/repo", &query).await.unwrap(); + assert_eq!(events.len(), 2); + } + + #[tokio::test] + async fn fetch_keeps_unknown_fields_verbatim() { + let server = MockServer::start().await; + let raw = json!({ + "id": 7, + "type": "action.queue.leave", + "received_at": "2026-07-29T12:00:00Z", + "field_from_2027": {"nested": [1, 2, 3]}, + }); + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .respond_with( + ResponseTemplate::new(200).set_body_json(page_body(std::slice::from_ref(&raw))), + ) + .expect(1) + .mount(&server) + .await; + + let query = Query { + pull_request: None, + event_types: vec![], + window: Window::retained(at("2026-07-30T00:00:00Z")), + limit: None, + }; + let events = fetch(&client(&server), "owner/repo", &query).await.unwrap(); + assert_eq!(events[0].raw, raw); + } + + #[tokio::test] + async fn fetch_propagates_the_api_failure() { + // A 403 (token can read the queue but not the whole activity + // log) surfaces as the typed Mergify API error; the caller + // decides whether that is fatal. + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v1/repos/owner/repo/logs")) + .respond_with(ResponseTemplate::new(403).set_body_json(json!({"detail": "nope"}))) + .expect(1) + .mount(&server) + .await; + + let query = Query { + pull_request: None, + event_types: vec![], + window: Window::retained(at("2026-07-30T00:00:00Z")), + limit: None, + }; + let err = fetch(&client(&server), "owner/repo", &query) + .await + .unwrap_err(); + assert!(matches!(err, CliError::MergifyApi(_)), "got: {err:?}"); + } +} diff --git a/crates/mergify-events/src/event.rs b/crates/mergify-events/src/event.rs new file mode 100644 index 00000000..9b17c46a --- /dev/null +++ b/crates/mergify-events/src/event.rs @@ -0,0 +1,153 @@ +//! One activity-log event: the raw API object plus the decoded +//! envelope every event type shares. + +use chrono::DateTime; +use chrono::Utc; +use serde::Deserialize; + +/// An event from the activity log. +/// +/// `raw` is the API's own object, untouched — the schema is +/// Mergify's contract, not this CLI's, so unknown fields must +/// survive the round-trip (`--json` republishes them, and a newer +/// engine cannot break an older CLI). The envelope accessors decode +/// only the fields every event type carries. +#[derive(Debug)] +pub struct Event { + pub raw: serde_json::Value, + envelope: Envelope, +} + +/// The common envelope. Every field is optional: an event shape this +/// CLI has never seen must still render, not crash. +#[derive(Deserialize, Default, Debug)] +struct Envelope { + #[serde(default, rename = "type")] + event_type: Option, + #[serde(default)] + received_at: Option, + #[serde(default)] + trigger: Option, + #[serde(default)] + pull_request: Option, +} + +impl Event { + /// Wrap a raw API event. Infallible by design: a payload whose + /// envelope does not decode (wrong types, not an object) is still + /// an event — its accessors read as absent and `raw` keeps + /// everything. + #[must_use] + pub fn from_raw(raw: serde_json::Value) -> Self { + let envelope = Envelope::deserialize(&raw).unwrap_or_default(); + Self { raw, envelope } + } + + /// The event type (e.g. `action.queue.leave`), verbatim from the + /// API — new engine types pass through unrecognized. + #[must_use] + pub fn event_type(&self) -> Option<&str> { + non_empty(self.envelope.event_type.as_deref()) + } + + /// When the engine recorded the event, as the API sent it. + #[must_use] + pub fn received_at(&self) -> Option<&str> { + non_empty(self.envelope.received_at.as_deref()) + } + + /// [`Self::received_at`] parsed, for sorting and rendering. + /// `None` when absent or unparseable — degrading beats crashing + /// on a timestamp format change. + #[must_use] + pub fn received_at_utc(&self) -> Option> { + self.received_at() + .and_then(|ts| DateTime::parse_from_rfc3339(ts).ok()) + .map(|ts| ts.with_timezone(&Utc)) + } + + /// What caused the event (e.g. `merge queue internal`, a command + /// author). + #[must_use] + pub fn trigger(&self) -> Option<&str> { + non_empty(self.envelope.trigger.as_deref()) + } + + /// The pull request the event belongs to, when it has one. + #[must_use] + pub const fn pull_request(&self) -> Option { + self.envelope.pull_request + } + + /// The event's type-specific `metadata` object; `Null` when the + /// payload has none. Callers decode the slice of it they + /// understand. + #[must_use] + pub fn metadata(&self) -> &serde_json::Value { + self.raw.get("metadata").unwrap_or(&serde_json::Value::Null) + } +} + +fn non_empty(value: Option<&str>) -> Option<&str> { + value.filter(|s| !s.is_empty()) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn from_raw_decodes_the_shared_envelope() { + let event = Event::from_raw(json!({ + "id": 1, + "type": "action.queue.leave", + "received_at": "2026-07-20T23:25:11.263987Z", + "trigger": "merge queue internal", + "pull_request": 1700, + "metadata": {"merged": false}, + })); + assert_eq!(event.event_type(), Some("action.queue.leave")); + assert_eq!(event.received_at(), Some("2026-07-20T23:25:11.263987Z")); + assert_eq!(event.trigger(), Some("merge queue internal")); + assert_eq!(event.pull_request(), Some(1700)); + assert_eq!(event.metadata()["merged"], json!(false)); + } + + #[test] + fn from_raw_keeps_unknown_fields_verbatim() { + let raw = json!({ + "type": "something.from.2027", + "future_field": {"nested": true}, + }); + let event = Event::from_raw(raw.clone()); + assert_eq!(event.raw, raw); + assert_eq!(event.event_type(), Some("something.from.2027")); + } + + #[test] + fn a_malformed_envelope_still_yields_an_event() { + // Wrong types must not crash — accessors read as absent and + // the raw payload survives for `--json`. + let raw = json!({"type": 42, "received_at": ["not", "a", "date"]}); + let event = Event::from_raw(raw.clone()); + assert_eq!(event.event_type(), None); + assert_eq!(event.received_at(), None); + assert_eq!(event.raw, raw); + } + + #[test] + fn received_at_utc_parses_and_degrades() { + let parsed = Event::from_raw(json!({"received_at": "2026-07-20T23:25:11Z"})); + assert!(parsed.received_at_utc().is_some()); + let garbage = Event::from_raw(json!({"received_at": "not-a-date"})); + assert!(garbage.received_at_utc().is_none()); + } + + #[test] + fn metadata_is_null_when_absent() { + let event = Event::from_raw(json!({"type": "action.label"})); + assert!(event.metadata().is_null()); + } +} diff --git a/crates/mergify-events/src/lib.rs b/crates/mergify-events/src/lib.rs new file mode 100644 index 00000000..e12cb721 --- /dev/null +++ b/crates/mergify-events/src/lib.rs @@ -0,0 +1,42 @@ +//! Shared client for the Mergify activity log +//! (`GET /v1/repos///logs`). +//! +//! The activity log carries every event type the engine records — +//! the `action.queue.*` family, the workflow actions, `ci_insights.*`, +//! `command.*`, … — behind one filtered, paginated endpoint. Its +//! contract has sharp edges that every hand-rolled consumer gets +//! subtly wrong, so this crate owns them once: +//! +//! - **The window is always explicit.** The API defaults +//! `received_from` to `received_to - 1 day`, which makes a pull +//! request dequeued last week look identical to one never queued. +//! A [`Window`] carries both bounds and every request sends them; +//! the silent 1-day default is unreachable from here. +//! - **The 93-day cap is a typed error.** The API rejects a range +//! wider than [`MAX_SPAN_DAYS`] with a raw 422 while retention is +//! [`RETENTION_DAYS`] days. [`Window`] refuses to construct such a +//! range ([`WindowError`]), and "everything retained" is spelled +//! [`Window::retained`] — so the 422 cannot happen and the widest +//! useful window has a name. +//! - **Pagination is followed to completion.** The endpoint pages +//! with opaque cursors in RFC 5988 `Link` headers; [`fetch`] +//! follows them until the window is exhausted (or an explicit +//! [`Query::limit`] is reached) rather than silently returning the +//! first page. +//! - **Ordering is guaranteed newest-first.** The API serves events +//! newest-first; [`fetch`] enforces it after collecting, so it is a +//! promise of this crate rather than an observation about today's +//! server. +//! - **Unknown fields pass through verbatim.** An [`Event`] keeps the +//! API's raw object untouched next to a decoded envelope of the +//! fields every event shares — a newer engine cannot break an older +//! CLI, and `--json` consumers see Mergify's contract, not this +//! crate's. + +pub mod client; +pub mod event; +pub mod window; + +pub use client::{Query, fetch}; +pub use event::Event; +pub use window::{MAX_SPAN_DAYS, RETENTION_DAYS, Window, WindowError}; diff --git a/crates/mergify-events/src/window.rs b/crates/mergify-events/src/window.rs new file mode 100644 index 00000000..bd67312f --- /dev/null +++ b/crates/mergify-events/src/window.rs @@ -0,0 +1,180 @@ +//! Explicit time range for an activity-log query. +//! +//! `/logs` has two window traps and this type is where both die: +//! the silent `received_to - 1 day` default (a [`Window`] always +//! carries both bounds, so callers cannot forget to send them) and +//! the 422 on a span over 93 days (construction refuses it, so the +//! raw HTTP error is unreachable). + +use chrono::DateTime; +use chrono::TimeDelta; +use chrono::Utc; + +/// How many days of events the API retains. A query reaching further +/// back returns nothing extra — [`Window::retained`] is the widest +/// window worth asking for. +pub const RETENTION_DAYS: i64 = 90; + +/// The widest `received_from`/`received_to` span the API accepts; +/// anything wider is rejected with a 422 (`'received_from' and +/// 'received_to' cannot span more than 93 days`). +pub const MAX_SPAN_DAYS: i64 = 93; + +/// An explicit `received_from`/`received_to` range, guaranteed to be +/// positive and within the API's span cap. +#[derive(Copy, Clone, Debug)] +pub struct Window { + from: DateTime, + to: DateTime, +} + +/// Why a [`Window`] could not be constructed. Surfacing this at +/// construction is the whole point: the caller gets a typed, +/// explainable refusal instead of the API's raw 422. +#[derive(thiserror::Error, Debug, PartialEq, Eq)] +pub enum WindowError { + #[error( + "a {requested_days}-day window is wider than the API accepts \ + (over {max} days); the activity log retains {retention} days \ + — use {retention} days or less", + max = MAX_SPAN_DAYS, + retention = RETENTION_DAYS + )] + SpanTooWide { requested_days: i64 }, + + #[error("the window must cover a positive length of time")] + NotPositive, +} + +impl Window { + /// The whole retained history, ending at `now`. This is + /// "everything the API can still answer for", spelled without the + /// caller knowing the retention number. + #[must_use] + pub fn retained(now: DateTime) -> Self { + Self { + from: now - TimeDelta::days(RETENTION_DAYS), + to: now, + } + } + + /// The last `span` of time, ending at `now`. + /// + /// # Errors + /// + /// [`WindowError::SpanTooWide`] when `span` exceeds + /// [`MAX_SPAN_DAYS`] (the API would 422); + /// [`WindowError::NotPositive`] when it is zero or negative. + pub fn last(span: TimeDelta, now: DateTime) -> Result { + if span <= TimeDelta::zero() { + return Err(WindowError::NotPositive); + } + if span > TimeDelta::days(MAX_SPAN_DAYS) { + return Err(WindowError::SpanTooWide { + requested_days: ceil_days(span), + }); + } + Ok(Self { + from: now - span, + to: now, + }) + } + + #[must_use] + pub const fn from(&self) -> DateTime { + self.from + } + + #[must_use] + pub const fn to(&self) -> DateTime { + self.to + } +} + +/// `span` in days, rounded **up**. Truncating (`num_days`) makes a +/// 93-day-and-one-hour ask report itself as "a 93-day window is wider +/// than the API accepts (over 93 days)" — a refusal that contradicts +/// its own reason. A partial day is a day the caller asked for. +fn ceil_days(span: TimeDelta) -> i64 { + let whole = span.num_days(); + if span > TimeDelta::days(whole) { + whole + 1 + } else { + whole + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn at(iso: &str) -> DateTime { + DateTime::parse_from_rfc3339(iso) + .unwrap() + .with_timezone(&Utc) + } + + #[test] + fn retained_covers_the_whole_retention() { + let now = at("2026-07-30T00:00:00Z"); + let window = Window::retained(now); + assert_eq!(window.to(), now); + assert_eq!(window.from(), at("2026-05-01T00:00:00Z")); + } + + #[test] + fn last_builds_an_explicit_range_ending_now() { + let now = at("2026-07-30T12:00:00Z"); + let window = Window::last(TimeDelta::hours(24), now).unwrap(); + assert_eq!(window.from(), at("2026-07-29T12:00:00Z")); + assert_eq!(window.to(), now); + } + + #[test] + fn last_refuses_a_span_the_api_would_422_on() { + // The typed replacement for the raw `422: 'received_from' and + // 'received_to' cannot span more than 93 days`. + let now = at("2026-07-30T00:00:00Z"); + let err = Window::last(TimeDelta::days(94), now).unwrap_err(); + assert_eq!(err, WindowError::SpanTooWide { requested_days: 94 }); + // The message must hand the user the fix, not just the refusal. + assert!(err.to_string().contains("retains 90 days"), "got: {err}"); + } + + #[test] + fn a_part_day_over_the_cap_rounds_the_message_up() { + // `--since 2233h` is 93 days and one hour. Reported with a + // truncating day count it reads "a 93-day window is wider than + // the API accepts (over 93 days)". + let now = at("2026-07-30T00:00:00Z"); + let span = TimeDelta::days(MAX_SPAN_DAYS) + TimeDelta::hours(1); + let err = Window::last(span, now).unwrap_err(); + assert_eq!(err, WindowError::SpanTooWide { requested_days: 94 }); + // Sub-second precision counts too: a whole day plus a nanosecond + // is still more than the caller may have. + let sliver = TimeDelta::days(MAX_SPAN_DAYS) + TimeDelta::nanoseconds(1); + assert_eq!( + Window::last(sliver, now).unwrap_err(), + WindowError::SpanTooWide { requested_days: 94 }, + ); + } + + #[test] + fn last_accepts_the_span_cap_itself() { + let now = at("2026-07-30T00:00:00Z"); + assert!(Window::last(TimeDelta::days(MAX_SPAN_DAYS), now).is_ok()); + } + + #[test] + fn last_refuses_an_empty_or_negative_span() { + let now = at("2026-07-30T00:00:00Z"); + assert_eq!( + Window::last(TimeDelta::zero(), now).unwrap_err(), + WindowError::NotPositive, + ); + assert_eq!( + Window::last(TimeDelta::hours(-1), now).unwrap_err(), + WindowError::NotPositive, + ); + } +}